quic: Bug in flow control for bidistreams and sessions?

[martenrichter]

I wrote a test in order to trigger stalls caused by session flow control. (So exhaustion of the flow control window of the session).
I am not sure if I achieved it, but it stalls. So it might be something else.
I built a node on Windows, not on the current main, but on #63230, so that I am sure that it is another issue.

The test is based on portions from different other node.js tests (test-quic-stream-bidi-echo-concurrent.mjs):

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

// Test: bidirectional stream echo with binary.
// The client sends a message, the server reads it and echoes it back.
// Both directions of the bidi stream carry data and are properly FIN'd.
// Verifies that both client and server can read and write on the same stream.

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

const { strictEqual } = assert;

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

const { listen, connect } = await import('../common/quic.mjs');
const { bytes, drainableProtocol: dp } = await import('stream/iter');

const chunkSizes = [60000, 12, 1000000, 50000, 1600, 20000, 1000000, 30000, 0, 100]
const numChunks = chunkSizes.length
const byteLength = chunkSizes.reduce((accumulator, currentValue) => accumulator + currentValue, 0);


// Build a deterministic payload so we can verify integrity.
function buildChunk(index) {
  const chunk = new Uint8Array(chunkSizes[index]);
  // Fill with a pattern derived from the chunk index.
  const val = index & 0xff;
  for (let i = 0; i < chunkSizes[index]; i++) {
    chunk[i] = (val + i) & 0xff;
  }
  return chunk;
}

function checksum(data) {
  let sum = 0;
  for (let i = 0; i < data.byteLength; i++) {
    sum = (sum + data[i]) | 0;
  }
  return sum;
}

// Compute expected checksum.
let expectedChecksum = 0;
for (let i = 0; i < numChunks; i++) {
  const chunk = buildChunk(i);
  expectedChecksum = (expectedChecksum + checksum(chunk)) | 0;
}

const maxStreams = 20;
let pendStreams = maxStreams;
const done = Promise.withResolvers();

const serverEndpoint = await listen(mustCall((serverSession) => {
  serverSession.onstream = mustCall(async (stream) => {
    const writer = stream.writer
    let written = 0;
    for await (const chunks of stream) {
      for (const chunk of chunks) {
        written += chunk.byteLength;

        while (!writer.writeSync(chunk)) {
          // Flow controlled — wait for drain before retrying.
          const drainable = writer[dp]();
          if (drainable) await drainable;
        }
      }
    }
    writer.end();

    await stream.closed;
    pendStreams--;
    console.log('end pendstreams', pendStreams)
    if (pendStreams === 0) {
      serverSession.close();
      done.resolve();
    }
  });
}));

const clientSession = await connect(serverEndpoint.address);
await clientSession.opened;





const readFromStream = async (stream) => {
  const readChunks = [];
  for await (const chunks of stream) {
    readChunks.push(...chunks);
  }
  const receivedBytes = readChunks.reduce((accu, curVal) => accu + curVal.byteLength, 0);

  strictEqual(receivedBytes, byteLength);

  let receivedChecksum = 0;
  for (const chunk of readChunks) {
    receivedChecksum = (receivedChecksum + checksum(chunk)) | 0;
  }

  strictEqual(receivedChecksum, expectedChecksum);
}


const writeToStream = async (stream) => {
  const w = stream.writer
  for (let i = 0; i < numChunks; i++) {
    const chunk = buildChunk(i);
    while (!w.writeSync(chunk)) {
      // Flow controlled — wait for drain before retrying.
      const drainable = w[dp]();
      if (drainable) await drainable;
    }
  }
  w.endSync();
}

const streamClosed = []

await Promise.all(Array.from({ length: 20 }, async () => {
  const stream = await clientSession.createBidirectionalStream()
  streamClosed.push(stream.closed);
  await Promise.all([readFromStream(stream), writeToStream(stream)])
}))


await Promise.all([...streamClosed, done.promise]);
await clientSession.close();
await serverEndpoint.close();

The results are flaky and stall after different number of streams. For example:

end pendstreams 19
end pendstreams 18
end pendstreams 17
end pendstreams 16
end pendstreams 15
end pendstreams 14
end pendstreams 13
end pendstreams 12
end pendstreams 11
end pendstreams 10
end pendstreams 9  <--- stall
node:internal/quic/quic:2122  <--- session time out I presume
      handle.endWrite();
             ^

Error: Stream is no longer writable
    at endSync (node:internal/quic/quic:2122:14)
    at Object.end (node:internal/quic/quic:2146:17)
    at QuicSession.<anonymous> (file:///C:/PrivateDaten/Projekte/failscomponents/failslib/thirdparty/nodepatch/node/test/parallel/test-quic-stream-bidi-echo-concurrent.mjs:70:12) {
  code: 'ERR_INVALID_STATE'
}

@jasnell That would be another issue.....