nodejs / nodejs/node

quic: Bug in flow control for bidistreams and sessions?

Open
#63,304 2 comments 0 reactions 0 assignees View on GitHub

Nobody has claimed this yet.

never-stale quic
Dominant language
JavaScript
Stars
122k
Forks
37.3k
Avg merge
4d 2h
Merged PRs (30d)
283

Description

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 https://github.com/nodejs/node/pull/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.....

Contributor guide

Open the contributing guide

First steps

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. Open a pull request that references the issue number.

Research direction

Start with test/parallel/test-quic-stream-bidi-echo-concurrent.mjs and run the concurrent bidirectional-stream reproduction described in the issue. Trace the session and stream flow-control behavior around the reported stall and timeout. Done means the test reliably completes all 20 streams without stalling or producing the invalid writable-state error.

Written by the indexing model from the issue text.

Assessment

Tech stack
javascript, nodejs
Domain
networking, testing-qa
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Active
Clarity
Mostly clear
Newbie friendliness
45/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.