diff --git a/src/quic/session.cc b/src/quic/session.cc index d83c1cbb89db..6841706da19c 100644 --- a/src/quic/session.cc +++ b/src/quic/session.cc @@ -2077,27 +2077,23 @@ void Session::SendPendingData() { closed = true; return; } - } else if (stream_data.id >= 0) { + } else if (ndatalen < 0 && nwrite >= 0 && stream_data.id >= 0) { Debug(this, "Session did not accept any bytes from stream %" PRIi64 " into packet", stream_data.id); + // ndatalen < 0 means this stream's frame (data or FIN) was left out of + // the packet, due to congestion/pacing (nwrite == 0) or because other + // frames filled it. Reschedule the stream so it is retried later. + if (stream_data.count > 0 || stream_data.fin) { + application().ResumeStream(stream_data.id); + } } // When nwrite is zero, it means we are congestion limited or it is - // just not our turn to send something. Re-schedule the stream if it - // had unsent data (payload or FIN) so the next timer-triggered - // SendPendingData retries it. Without this, a FIN-only send that - // hits nwrite=0 is lost forever — the stream already returned EOS - // from Pull and won't be re-scheduled by anyone else. - // We call Application::ResumeStream directly (not Session::ResumeStream) - // to avoid creating a SendPendingDataScope — we're already inside - // SendPendingData and re-entering would just hit nwrite=0 again. + // just not our turn to send something. if (nwrite == 0 && (stream_data.id >= 0 || !HasPendingDatagrams())) { Debug(this, "Congestion or not our turn to send"); - if (stream_data.id >= 0 && (stream_data.count > 0 || stream_data.fin)) { - application().ResumeStream(stream_data.id); - } return; } diff --git a/test/parallel/test-quic-stream-fin-packet-full.mjs b/test/parallel/test-quic-stream-fin-packet-full.mjs new file mode 100644 index 000000000000..2c0749fd3dc5 --- /dev/null +++ b/test/parallel/test-quic-stream-fin-packet-full.mjs @@ -0,0 +1,57 @@ +// Flags: --experimental-quic --experimental-stream-iter --no-warnings + +// Test: a FIN that does not fit in the current packet is still sent. +// +// For a fixed body the final data and the FIN go out as separate STREAM +// frames, packed into the same packet where possible. When the data leaves +// only a few bytes free, the FIN-only frame does not fit and the packet is +// sent without it. The stream must then be retried, or the peer never sees +// the end of the stream. +// +// Which body size leaves too little room depends on per-packet overhead, so +// this sweeps sizes around one packet's payload, one stream at a time. + +import { hasQuic, skip, mustCall } from '../common/index.mjs'; +import assert from 'node:assert'; + +if (!hasQuic) { + skip('QUIC is not enabled'); +} + +const { listen, connect } = await import('../common/quic.mjs'); +const { bytes } = await import('stream/iter'); + +const kMinSize = 1000; +const kMaxSize = 1300; +const kCount = kMaxSize - kMinSize + 1; + +const received = []; +let next = Promise.withResolvers(); + +const serverEndpoint = await listen(mustCall((serverSession) => { + serverSession.onstream = mustCall(async (stream) => { + const data = await bytes(stream); + received.push(data.byteLength); + next.resolve(); + }, kCount); +})); + +const clientSession = await connect(serverEndpoint.address); +await clientSession.opened; +const closed = clientSession.closed.then(() => 'closed', () => 'closed'); + +for (let size = kMinSize; size <= kMaxSize; size++) { + const stream = await clientSession.createUnidirectionalStream(); + stream.setBody(new Uint8Array(size)); + const result = await Promise.race([next.promise.then(() => 'ok'), closed]); + assert.strictEqual(result, 'ok', + `session closed before the FIN of a ${size}-byte ` + + 'stream arrived'); + next = Promise.withResolvers(); + assert.strictEqual(received.at(-1), size); +} + +assert.strictEqual(received.length, kCount); + +await clientSession.close(); +await serverEndpoint.close();