Skip to content

Commit c1d3132

Browse files
authored
quic: mark drain promise handled
Fixes: #64290 Signed-off-by: James M Snell <jasnell@gmail.com> Assisted-by: Opencode/Opus PR-URL: #65319 Reviewed-By: Stephen Belanger <admin@stephenbelanger.com>
1 parent a8f9a7f commit c1d3132

2 files changed

Lines changed: 80 additions & 0 deletions

File tree

lib/internal/quic/quic.js

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2184,6 +2184,7 @@ class QuicStream {
21842184
errored = true;
21852185
error = reason;
21862186
if (drainWakeup != null) {
2187+
markPromiseAsHandled(drainWakeup.promise);
21872188
drainWakeup.reject(error);
21882189
drainWakeup = null;
21892190
}
@@ -2366,6 +2367,7 @@ class QuicStream {
23662367
getQuicSessionState(stream.#inner.session).internalErrorCode;
23672368
handle.resetStream(code);
23682369
if (drainWakeup != null) {
2370+
markPromiseAsHandled(drainWakeup.promise);
23692371
drainWakeup.reject(error);
23702372
drainWakeup = null;
23712373
}
Lines changed: 78 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,78 @@
1+
// Flags: --experimental-quic --experimental-stream-iter --no-warnings
2+
3+
// Regression test for https://github.com/nodejs/node/issues/64290
4+
// When a stream writer has a pending drain promise and the remote peer
5+
// resets the stream, the rejected drain promise must NOT surface as an
6+
// unhandled rejection.
7+
8+
import { hasQuic, skip, mustCall, mustNotCall } from '../common/index.mjs';
9+
import assert from 'node:assert';
10+
import { setImmediate as tick } from 'node:timers/promises';
11+
12+
if (!hasQuic) {
13+
skip('QUIC is not enabled');
14+
}
15+
16+
const { listen, connect } = await import('../common/quic.mjs');
17+
const { drainableProtocol } = await import('stream/iter');
18+
19+
// The test fails if any unhandled rejection fires.
20+
process.on('unhandledRejection',
21+
mustNotCall('unexpected unhandled rejection'));
22+
23+
const serverStreamReady = Promise.withResolvers();
24+
25+
const serverEndpoint = await listen(mustCall((serverSession) => {
26+
serverSession.onstream = mustCall((stream) => {
27+
serverStreamReady.resolve({ stream, session: serverSession });
28+
});
29+
}));
30+
31+
const clientSession = await connect(serverEndpoint.address);
32+
await clientSession.opened;
33+
34+
const stream = await clientSession.createBidirectionalStream();
35+
const writer = stream.writer;
36+
37+
// Write a small initial chunk so the server materializes the stream.
38+
writer.writeSync(new Uint8Array([1]));
39+
40+
const { stream: serverStream, session: serverSession } =
41+
await serverStreamReady.promise;
42+
43+
// Fill the write buffer to create backpressure. After this,
44+
// writeDesiredSize should be <= 0 and canWrite should be false.
45+
const chunk = new Uint8Array(64 * 1024);
46+
while (writer.canWrite) {
47+
if (!writer.writeSync(chunk)) break;
48+
}
49+
50+
// Create a drain wakeup via the drainable protocol. This simulates
51+
// what the stream/iter infrastructure does when checking for
52+
// backpressure. We deliberately do NOT await the returned promise —
53+
// that is the whole point of the test.
54+
const drainPromise = writer[drainableProtocol]();
55+
assert.ok(drainPromise instanceof Promise,
56+
'expected a drain promise (buffer should be full)');
57+
58+
// Suppress the expected rejection on both sides' closed promises so
59+
// they do not interfere with the unhandledRejection check.
60+
const clientClosed = stream.closed.catch(() => {});
61+
const serverClosed = serverStream.closed.catch(() => {});
62+
63+
// Have the server send STOP_SENDING. This triggers kStopSending on
64+
// the client writer, which rejects the unobserved drain promise.
65+
// Without the fix this surfaces as an unhandled rejection.
66+
serverStream.stopSending(1n);
67+
serverStream.writer.endSync();
68+
69+
// Give the event loop time to process the frame and fire any
70+
// unhandled-rejection events.
71+
await tick();
72+
await tick();
73+
74+
// Clean up.
75+
await Promise.all([clientClosed, serverClosed]);
76+
serverSession.close();
77+
await clientSession.close();
78+
await serverEndpoint.close();

0 commit comments

Comments
 (0)