quic: http/3 resumeStream does not work

[martenrichter]

I am still trying to implement webtransport in node.js.
Though I made some progress (and I hope to be able to share it soon).
I have currently a problem in my test harness (I am sitting the second day on it).
Basically, what it does is to connect a node.js http/3 (webtransport client, but does not matter).

First, the clients opens a bidi stream, the server echos 2x 3 bytes and it works fine. (Before or in parallel the server opened a server initiated bidi stream, and the clients also try to send 2 x 3bytes.
However, these are never collected by the application send loop.

I narrowed it down to:

node/src/quic/http3.cc

Line 357 in 2adaeee

void ResumeStream(stream_id id) override {

the ResumeStreamfunction.
It currently looks like:

void ResumeStream(stream_id id) override {
    nghttp3_conn_resume_stream(*this, id);
    Application::ResumeStream(id);
  }

Currently, it calls:

node/src/quic/application.h

Line 127 in 2adaeee

virtual void ResumeStream(stream_id id) {}

which is:

 virtual void ResumeStream(stream_id id) {}

I think the idea was to call the ResumeStream in the implementation:

node/src/quic/application.cc

Line 847 in 2adaeee

void ResumeStream(stream_id id) override { ScheduleStream(id); }

 void ResumeStream(stream_id id) override { ScheduleStream(id); }

where ScheduleStream is a private method of the implementation.
Or do I miss something and nghttp3_conn_resume_stream(*this, id); should trigger some callbacks?
@jasnell how should it be fixed? I mean I can just copy over the ScheduleStream code over to the ResumeStream of the http3 class and see if it works? What I think you had add something in mind with the hidden implementation objects?

I am also not sure if I can provide a test for this, as I am not sure how to hit the correct timings.

[martenrichter]

OK, I copied it over and obviously, it should be nghttp3_conn_resume_stream(*this, id); that handles the awakening....,

[martenrichter]

Well, I tracked it further down, nghttp3_conn_resume_stream(*this, id) does not resume that stream, as the new data did not arrive at the stream->frq or stream->outq ringbuffer, so the ringbuffer has a length of zero. So the dataqueue append, should have triggered data to be pushed to nghttp3 ...... So this is missing.

[martenrichter]

Ok, next clue ResumeStream(...) has a SendPendingDataScope so data is pushed to the http3 stream after nghttp3_conn_resume_stream is called, This is the right order for the outgoing quic stream, but not for the stream data, which must go through nghttp3 beforehand.
So I could make a manual call to SendPendingData, though I think this may cause issues with object lifetimes.

[martenrichter]

@jasnell
I was able to generate a test, where I hope it simulates the bug. At least it simulates a bug:

// Flags: --experimental-quic --experimental-stream-iter --no-warnings

// Test: Streaming of a few bytes from client to server
// It is build so that the ResumeStream mechanism of http3 classes
// is tested

import { hasQuic, skip, mustCall } from '../common/index.mjs';
import assert from 'node:assert';
import * as fixtures from '../common/fixtures.mjs';

const { strictEqual } = assert;
const { readKey } = fixtures;

if (!hasQuic) {
  skip('QUIC is not enabled');
}

const { listen, connect } = await import('node:quic');
const { createPrivateKey } = await import('node:crypto');
const { drainableProtocol: dp } = await import('stream/iter');

const key = createPrivateKey(readKey('agent1-key.pem'));
const cert = readKey('agent1-cert.pem');

const serverDone = Promise.withResolvers();

const streamOpened = Promise.withResolvers();

let result = 0;

const serverEndpoint = await listen(mustCall(async (ss) => {
  ss.onstream = mustCall(async (stream) => {
    console.log('ONSTREAM mark 1')
    streamOpened.resolve();
     console.log('ONSTREAM mark 2')

    for await (const chunks of stream) {
      console.log('ONSTREAM chunks', chunks)
      for (const chunk of chunks) {
        for (let i = 0; i < chunk.length; i++) {
          result += chunk[i];
        }
      }
      if (result === 60 + 61 + 62) {
        serverDone.resolve();
      }
    }
     console.log('ONSTREAM mark 3')
    stream.writer.endSync();
     console.log('ONSTREAM mark 4')
    await stream.closed;
    console.log('ONSTREAM mark 5')
    ss.close();
  });
  await ss.closed;
}), {
  sni: { '*': { keys: [key], certs: [cert] } },
});

const clientSession = await connect(serverEndpoint.address, {
  servername: 'localhost',
  verifyPeer: 'manual',
});
await clientSession.opened;

// Create a stream with headers
const stream = await clientSession.createBidirectionalStream({
  headers: {
    ':method': 'GET',
    ':path': '/count',
    ':scheme': 'https',
    ':authority': 'localhost',
  },
});

const w = stream.writer;

await streamOpened.promise; // wait that the stream arrives at the server
console.log('WRITEBYTES bf')
// write three bytes
while (!w.writeSync(new Uint8Array([60, 61, 62]))) {
  const drainable = w[dp]();
  if (drainable) await drainable;
}
console.log('WRITEBYTES aft')

await serverDone.promise;
w.endSync();

await stream.closed;


// Close everything cleanly.
await clientSession.close();
await serverEndpoint.close();
strictEqual(result,  60 + 61 + 62);

You see that it actually does not do its job currently.

I have to pause for chasing this bug until may be the weekend after the next weekend. (Or a bit next weekend). I will be happy if anyone has ideas how to fix it.

[martenrichter]

Update of the test, I did not consider, then when sending headers without body the stream is automatically terminal (btw. I fall for this repeatedly):

// Flags: --experimental-quic --experimental-stream-iter --no-warnings

// Test: Streaming of a few bytes from client to server
// It is build so that the ResumeStream mechanism of http3 classes
// is tested

import { hasQuic, skip, mustCall } from '../common/index.mjs';
import assert from 'node:assert';
import * as fixtures from '../common/fixtures.mjs';

const { strictEqual } = assert;
const { readKey } = fixtures;

if (!hasQuic) {
  skip('QUIC is not enabled');
}

const { listen, connect } = await import('node:quic');
const { createPrivateKey } = await import('node:crypto');
const { drainableProtocol: dp } = await import('stream/iter');

const key = createPrivateKey(readKey('agent1-key.pem'));
const cert = readKey('agent1-cert.pem');

const serverDone = Promise.withResolvers();

const streamOpened = Promise.withResolvers();

let result = 0;

const serverEndpoint = await listen(mustCall(async (ss) => {
  ss.onstream = mustCall(async (stream) => {
    console.log('ONSTREAM mark 1')
    streamOpened.resolve();
     console.log('ONSTREAM mark 2')

    for await (const chunks of stream) {
      console.log('ONSTREAM chunks', chunks)
      for (const chunk of chunks) {
        for (let i = 0; i < chunk.length; i++) {
          result += chunk[i];
        }
      }
      if (result === 60 + 61 + 62) {
        serverDone.resolve();
      }
    }
     console.log('ONSTREAM mark 3')
    stream.writer.endSync();
     console.log('ONSTREAM mark 4')
    await stream.closed;
    console.log('ONSTREAM mark 5')
    ss.close();
  });
  await ss.closed;
}), {
  sni: { '*': { keys: [key], certs: [cert] } },
});

const clientSession = await connect(serverEndpoint.address, {
  servername: 'localhost',
  verifyPeer: 'manual',
});
await clientSession.opened;

// Create a stream with headers
const stream = await clientSession.createBidirectionalStream({
   incremental: true,
});

await stream.sendHeaders(
   {
    ':method': 'GET',
    ':path': '/count',
    ':scheme': 'https',
    ':authority': 'localhost',
  }
);

const w = stream.writer;

await streamOpened.promise; // wait that the stream arrives at the server
console.log('WRITEBYTES bf')
// write three bytes
while (!w.writeSync(new Uint8Array([60, 61, 62]))) {
  const drainable = w[dp]();
  if (drainable) await drainable;
}
console.log('WRITEBYTES aft')

await serverDone.promise;
w.endSync();

await stream.closed;


// Close everything cleanly.
await clientSession.close();
await serverEndpoint.close();
strictEqual(result,  60 + 61 + 62);

but now the test works, so it is the wrong test for the problem.

[martenrichter]

Ok, I did not understand correctly to handle server side initiated WT stream. This was no issue related to normal http/3 and quic.