Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 12 additions & 1 deletion src/js/internal/quic/quic.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2609,7 +2609,18 @@ class QuicStream {
const inner = this.#inner;

if (inner.pendingTrailers) {
Comment thread
robobun marked this conversation as resolved.
this.sendTrailers(inner.pendingTrailers);
// Node v26.10.0: https://github.com/nodejs/node/blob/v26.10.0/lib/internal/quic/quic.js#L2794-L2807
let sent;
try {
sent = this.sendTrailers(inner.pendingTrailers);
} catch (error) {
this.destroy(error);
Comment thread
robobun marked this conversation as resolved.
return;
}
if (!sent) {
this.destroy(ERR_QUIC_STREAM_ABORTED("Failed to submit trailing headers"));
return;
}
inner.pendingTrailers = undefined;
} else {
const { onwanttrailers } = inner;
Expand Down
139 changes: 139 additions & 0 deletions test/js/node/quic/quic-stream.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -284,6 +284,145 @@ describe("HTTP/3 header blocks that follow another header block", () => {
});
});

// A stream sends `pendingTrailers` itself when the body is out, so the caller
// has no answer to read. Node v26.10.0 destroys the stream when the block
// cannot be sent (nodejs/node#65863).
describe("pendingTrailers", () => {
const encoder = new TextEncoder();
const request = { ":method": "POST", ":scheme": "https", ":authority": "localhost" };
// Longer than a test: an idle timeout must not be what ends the session.
const transportParams = { maxIdleTimeout: 60 };

// One response with a body and `trailers` as its pendingTrailers. Gives how
// `closed` settles for the server stream, and what the client gets.
async function respondWith(trailers: Record<string, string>) {
let closed = "open";
const settled = Promise.withResolvers<string>();
const settle = (how: string) => {
closed = how;
settled.resolve(how);
};
const server = await listen(
async serverSession => {
serverSession.onstream = (stream: any) => {
stream.closed.catch(() => {});
};
await serverSession.closed.catch(() => {});
},
{
sni: { "*": { keys: [key], certs: [cert] } },
transportParams,
onheaders(this: any, headers: Record<string, any>) {
if (headers[":path"] !== "/trailers") {
this.sendHeaders({ ":status": "200" }, { terminal: true });
return;
}
this.closed.then(
() => settle("resolved"),
(err: any) => settle(err.code),
);
this.sendHeaders({ ":status": "200" });
this.pendingTrailers = trailers;
this.writer.writeSync(encoder.encode("body"));
this.writer.endSync();
},
// A stream is asked for its trailers only when this is set.
onwanttrailers() {},
},
);
let client: any;
// Not a graceful close. That waits for every stream, and a stream stays
// open when its trailers are never sent.
const dispose = async () => {
client?.destroy();
await server.destroy();
};
try {
client = await connect(server.address!, {
servername: "localhost",
verifyPeer: "manual",
transportParams,
} as SessionOptions);
await client.opened;

const clientTrailers = Promise.withResolvers<Record<string, string>>();
const stream = await client.createBidirectionalStream({
headers: { ...request, ":path": "/trailers" },
onheaders() {},
ontrailers: clientTrailers.resolve,
});
// A stream gets its trailers before it closes.
stream.closed.then(
() => clientTrailers.reject(new Error("the stream closed with no trailers")),
clientTrailers.reject,
);
clientTrailers.promise.catch(() => {});
const arrived = Promise.withResolvers<void>();
const body = (async () => {
let text = "";
try {
for await (const batch of stream) {
for (const chunk of batch) text += Buffer.from(chunk).toString("latin1");
if (text === "body") arrived.resolve();
}
return text;
} finally {
arrived.resolve();
}
})();
body.catch(() => {});
await arrived.promise;

// The server asks the stream for its trailers when the body leaves, which
// is before it gets this request. So the answer comes after that.
const answered = Promise.withResolvers<void>();
const later = await client.createBidirectionalStream({
headers: { ...request, ":path": "/" },
onheaders() {
answered.resolve();
},
});
later.closed.then(() => answered.reject(new Error("the stream closed with no answer")), answered.reject);
await answered.promise;
return {
closed,
settled: settled.promise,
body,
trailers: clientTrailers.promise,
[Symbol.asyncDispose]: dispose,
};
} catch (err) {
await dispose();
throw err;
}
}

test("a block that is sent reaches the client, and the stream closes with no error", async () => {
await using response = await respondWith({ "x-checksum": "abc123" });
expect({
trailers: { ...(await response.trailers) },
body: await response.body,
closed: await response.settled,
}).toEqual({ trailers: { "x-checksum": "abc123" }, body: "body", closed: "resolved" });
});

test("a block that sendTrailers() throws for destroys the stream with the error", async () => {
await using response = await respondWith({ connection: "close" });
expect(response.closed).toBe("ERR_HTTP2_INVALID_CONNECTION_HEADERS");
// destroy() still ends a response that the stream wrote and ended.
expect(await response.body).toBe("body");
});

test("a block that sendTrailers() answers false for destroys the stream", async () => {
// Latin-1 on the wire makes a NUL of U+0100. The block then has more
// pairs than it declares, and the native side refuses it.
await using response = await respondWith({ "x-bad": "a\u0100b" });
expect(response.closed).toBe("ERR_QUIC_STREAM_ABORTED");
// The client's read ends. The native side decides how.
await response.body.catch(() => {});
});
Comment thread
robobun marked this conversation as resolved.
});

describe("verifyClient", () => {
test("a server requiring a client certificate never surfaces streams from a client that presented none", async () => {
let announcedStreams = 0;
Expand Down
Loading