diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index 05cae70aa51..eab21ece344 100644 --- a/scripts/test-layout/layout.json +++ b/scripts/test-layout/layout.json @@ -2,7 +2,7 @@ "version": 1, "root": "tests", "explicit": { - "messages-native-scope.test.ts": "claude-integration", + "claude-picker-upload.test.ts": "claude-integration", "messages-native-scope.test.ts": "claude-integration", "cli-log-view-filter.test.ts": "cli", "cli-usage-model-search.test.ts": "cli", "cli-companion-usage.test.ts": "cli", "cli-account-key-quota.test.ts": "cli", "cli-observe-snapshot.test.ts": "cli", "cli-access-data-plane.test.ts": "cli", "cli-access-key-guard.test.ts": "cli", "cli-access-rename.test.ts": "cli", "cli-access-audio.test.ts": "cli", "cli-access-audio-live.test.ts": "cli", "cli-log-follow.test.ts": "cli", "cli-injection-follow.test.ts": "cli", @@ -370,15 +370,9 @@ "claude-desktop-policy.test.ts": "claude-integration", "claude-cli-picker.test.ts": "claude-integration", "claude-cli-picker-surface.test.ts": "claude-integration", - "claude-picker-bootstrap.test.ts": "claude-integration", - "claude-picker-ca.test.ts": "claude-integration", - "claude-picker-ca-store.test.ts": "claude-integration", - "claude-picker-listener.test.ts": "claude-integration", - "claude-picker-models.test.ts": "claude-integration", - "claude-picker-recovery.test.ts": "claude-integration", - "claude-picker-runtime.test.ts": "claude-integration", - "claude-picker-startup.test.ts": "claude-integration", - "claude-picker-trust.test.ts": "claude-integration", + "claude-picker-bootstrap.test.ts": "claude-integration", "claude-picker-ca.test.ts": "claude-integration", "claude-picker-ca-store.test.ts": "claude-integration", + "claude-picker-listener.test.ts": "claude-integration", "claude-picker-models.test.ts": "claude-integration", "claude-picker-recovery.test.ts": "claude-integration", + "claude-picker-runtime.test.ts": "claude-integration", "claude-picker-startup.test.ts": "claude-integration", "claude-picker-trust.test.ts": "claude-integration", "claude-desktop-remote-hub.test.ts": "claude-integration", "claude-dotenv-provenance-transport.test.ts": "claude-integration", "claude-estimate-projection.test.ts": "claude-integration", diff --git a/src/claude/intercept/picker-listener.ts b/src/claude/intercept/picker-listener.ts index d4dad33ffb9..23e8f5e1c8d 100644 --- a/src/claude/intercept/picker-listener.ts +++ b/src/claude/intercept/picker-listener.ts @@ -11,7 +11,7 @@ * HTTP/1.1 server, unchanged. Upstream is one HTTP/1.1 request per client request on both paths. * Only the bounded bootstrap response is held; other bodies and upgraded sockets relay as streams. */ -import { createSecureServer } from "node:http2"; +import { constants as h2Constants, createSecureServer } from "node:http2"; import type { Http2ServerRequest, Http2ServerResponse, ServerHttp2Session } from "node:http2"; import { createServer, request as httpsRequest } from "node:https"; import type { IncomingMessage, ServerResponse } from "node:http"; @@ -184,9 +184,35 @@ export async function startPickerListener(options: PickerListenerOptions): Promi // reason phrase differs, and sendHead branches on it. const res = relayRes as ServerResponse; const h2 = req.httpVersionMajor === 2; + // Bun's h2 compat response emits finish only after both stream halves close. + const responseWritable = h2 ? (req as Http2ServerRequest).stream : res; + // Framing, not method, determines whether a peer can keep sending a body. A bodyless h2 + // request carries END_STREAM on its headers; HTTP/1.1 needs length or chunked framing. + const uploading = h2 ? !(req as Http2ServerRequest).stream.endAfterHeaders + : req.headers["transfer-encoding"] !== undefined || Number(req.headers["content-length"] ?? 0) > 0; + const closeInput = (completeReply = false) => { + if (h2) { + const stream = (req as Http2ServerRequest).stream; + if (!stream.closed && !stream.destroyed) { + stream.close(completeReply ? h2Constants.NGHTTP2_NO_ERROR : h2Constants.NGHTTP2_CANCEL); + } + } else if (!req.destroyed) { + if (completeReply) { + // finish is not peer receipt: destroy can truncate bytes still queued in TLS/TCP. + // Discard further input and flush a graceful FIN without waiting for body completion. + req.resume(); + req.socket.end(); + } else req.destroy(); + } + }; // Refusals answer with an empty response and a fixed log line that carries no request data. const refuse = (status: 400 | 503) => { - res.writeHead(status, { "Content-Length": "0" }); + // Do not leave a refused, still-uploading stream behind after delivering its empty reply. + if (uploading && !req.complete) { + req.once("error", () => res.destroy()); + responseWritable.once("finish", () => closeInput(true)); + } + res.writeHead(status, { "Content-Length": "0", ...(!h2 && uploading && !req.complete ? { Connection: "close" } : {}) }); res.end(); options.log?.(`picker request refused ${status}`); }; @@ -209,12 +235,16 @@ export async function startPickerListener(options: PickerListenerOptions): Promi const fail = () => { if (clientGone) return; if (res.headersSent) res.destroy(); - else { res.writeHead(502, { "Content-Length": "0" }); res.end(); log(502); } + else { + res.writeHead(502, { "Content-Length": "0", ...(!h2 && uploading && !req.complete ? { Connection: "close" } : {}) }); + res.end(); log(502); + } }; const omit = bootstrap ? new Set(["accept-encoding"]) : new Set(); const headers = upstreamRequestHeaders(req, omit); if (bootstrap) headers.push("Accept-Encoding", narrowBootstrapAcceptEncoding()); let upReq: ReturnType; + let upstreamResponded = false; try { // The HTTP/1.1 client validates the method, path and header values synchronously. upReq = httpsRequest({ @@ -222,13 +252,17 @@ export async function startPickerListener(options: PickerListenerOptions): Promi ca: upstream.ca, rejectUnauthorized: true, agent: false, method, path: req.url, headers, maxHeaderSize: PICKER_MAX_HEADER_BYTES, }, upRes => { + upstreamResponded = true; const status = upRes.statusCode ?? 502; const originalHeaders = filteredHeaders(upRes.rawHeaders); const sendHead = (raw: string[]) => { if (res.headersSent) return; // HTTP/2 has no reason phrase; its compat writeHead takes the same flat raw header array. if (h2) (relayRes as Http2ServerResponse).writeHead(status, raw as unknown as Record); - else res.writeHead(status, upRes.statusMessage, raw); + else { + if (uploading && !req.complete) raw.push("Connection", "close"); + res.writeHead(status, upRes.statusMessage, raw); + } log(status); }; upRes.on("error", fail); @@ -293,6 +327,39 @@ export async function startPickerListener(options: PickerListenerOptions): Promi // Bun's compat response skips its close event for a HEAD reset before end(); the stream's own // close always fires. Destroying an upstream request that already completed is a no-op. if (h2) (req as Http2ServerRequest).stream.once("close", onClientClose); + if (uploading) { + let uploadFinished = false; + const finishUpload = () => { + if (uploadFinished) return; + uploadFinished = true; + req.off("end", finishUpload); + req.off("close", finishUpload); + upReq.off("close", onUpstreamClose); + responseWritable.off("finish", onResponseFinish); + res.off("close", stopUnfinishedUpload); + if (h2) (req as Http2ServerRequest).stream.off("close", finishUpload); + }; + // Upstream close can precede draining a successful reply to a slow downstream. + // Stop relaying input only once that reply's writable finishes or closes. + const stopUnfinishedUpload = (completeReply = false) => { + if (!uploadFinished) { + req.unpipe(upReq); + closeInput(completeReply); + finishUpload(); + upReq.destroy(); + } + }; + const onResponseFinish = () => stopUnfinishedUpload(true); + const onUpstreamClose = () => { + if (!upstreamResponded && !res.writableEnded) stopUnfinishedUpload(); + }; + req.once("end", finishUpload); + req.once("close", finishUpload); + upReq.once("close", onUpstreamClose); + responseWritable.once("finish", onResponseFinish); + res.once("close", stopUnfinishedUpload); + if (h2) (req as Http2ServerRequest).stream.once("close", finishUpload); + } req.pipe(upReq); }; server.on("request", relay); diff --git a/structure/clients/claude-desktop.md b/structure/clients/claude-desktop.md index fb864035d15..aa94003a72c 100644 --- a/structure/clients/claude-desktop.md +++ b/structure/clients/claude-desktop.md @@ -214,6 +214,26 @@ HTTP/1.1 request per client request. Incoming requests and ordinary upstream res a 64 KiB header allowance for browser session cookies; Bun enforces the HTTP/2 inbound bound natively, counting name + value + 32 bytes per field and rejecting an oversized stream with `RST_STREAM ENHANCE_YOUR_CALM` before the request handler runs. +The relay retains its 256-request aggregate ceiling across both HTTP versions and all sessions. +Upload framing, not the method, determines whether input cleanup is needed: HTTP/2 headers +without END_STREAM, or HTTP/1.1 transfer encoding or a positive Content-Length. +Rejected unfinished uploads close after their empty reply's writable finishes; HTTP/2 uses the +underlying stream's finish event rather than the compatibility response's finish event, which +waits for both stream halves to close. HTTP/1.1 refusals advertise `Connection: close` only while input is unfinished. +Early replies stop relaying unfinished input only after the downstream writable finishes or the +response closes; upstream request closure alone does not prove a buffered reply was delivered. +After a complete response, HTTP/2 closes only that stream with `RST_STREAM NO_ERROR` (RFC 9113 +ยง8.1), including a complete generated 502. Without a complete downstream response, cleanup uses +`CANCEL` for a premature downstream close or an upstream close without a reply. +HTTP/1.1 refusals, generated 502s and early replies advertise `Connection: close` only while input is unfinished; after the writable finishes, +the listener unpipes the upstream upload, reads and discards remaining input, and ends the socket +gracefully so queued response bytes flush. Errors and premature downstream closes still destroy +the request. Draining adds no upload deadline or byte cap and does not wait for request-body +completion before ending the response side. Body completion removes the input-cleanup listeners, +leaving long-lived responses and SSE subscriptions independent of upload cleanup. Completed HTTP/1.1 uploads retain keep-alive on generated 502s and can reuse the same client socket. +`tests/claude-integration/claude-picker-upload.test.ts` covers headers-only refusals on both +protocols, byte-exact early replies under backpressure, completed uploads with SSE, cancellation, +failure and shutdown cleanup, and healthy sibling streams. The picker CA (`picker-ca.ts`) carries critical name constraints permitting only `claude.ai` and excluding every IPv4 and IPv6 address. Its exportable signing identity is protected by the OS credential store and scoped to the canonical config directory; diff --git a/tests/claude-integration/claude-picker-upload.test.ts b/tests/claude-integration/claude-picker-upload.test.ts new file mode 100644 index 00000000000..18c4ea0eec5 --- /dev/null +++ b/tests/claude-integration/claude-picker-upload.test.ts @@ -0,0 +1,369 @@ +import { expect, test } from "bun:test"; +import { Agent, createServer, request } from "node:https"; +import { connect, constants, type ClientHttp2Session, type ClientHttp2Stream } from "node:http2"; +import type { ClientRequest, IncomingMessage, ServerResponse } from "node:http"; +import type { Duplex } from "node:stream"; +import { createLocalInterceptCa, issueLocalInterceptLeaf } from "../../src/claude/intercept/local-ca"; +import { startPickerListener, type PickerListenerOptions } from "../../src/claude/intercept/picker-listener"; + +async function fixture(handler: (req: IncomingMessage, res: ServerResponse) => void, + options: Partial = {}) { + const ca = createLocalInterceptCa(); + const leaf = issueLocalInterceptLeaf(ca, ["claude.ai"]); + const upstream = createServer({ cert: leaf.certPem, key: leaf.keyPem }, handler); + const sockets = new Set(); + for (const event of ["connection", "secureConnection"] as const) upstream.on(event, socket => { + sockets.add(socket); socket.once("close", () => sockets.delete(socket)); + }); + await new Promise(resolve => upstream.listen(0, "127.0.0.1", resolve)); + const address = upstream.address(); + if (!address || typeof address === "string") throw new Error("missing fixture port"); + const logs: string[] = []; + const relay = await startPickerListener({ leaf, models: () => [], ...options, + upstream: { host: "127.0.0.1", port: address.port, servername: "claude.ai", ca: ca.certPem }, + log: line => logs.push(line), + }); + const clients: Array = []; + const h1Agent = new Agent({ keepAlive: true }); + const h2 = () => { + const session = connect(`https://127.0.0.1:${relay.port}`, { ca: ca.certPem, servername: "claude.ai" }); + session.on("error", () => session.destroy()); clients.push(session); return session; + }; + const h1 = (path: string, method = "GET", headers: Record = {}) => { + const req = request({ host: "127.0.0.1", port: relay.port, servername: "claude.ai", ca: ca.certPem, + rejectUnauthorized: true, agent: h1Agent, path, method, headers: { Host: "claude.ai", ...headers } }); + clients.push(req); return req; + }; + return { logs, h2, h1, relay, + async close() { + const closed = clients.map(client => client.destroyed ? Promise.resolve() : observeClose(client)); + await relay.close(); + await Promise.all(closed); + for (const client of clients) client.destroy(); + h1Agent.destroy(); + for (const socket of sockets) socket.destroy(); + await new Promise(resolve => upstream.close(() => resolve())); + }, + }; +} + +function bounded(name: string, run: (wait: (p: PromiseLike) => Promise) => Promise) { + test(name, async () => { + const deadline = Promise.withResolvers(); + const timer = setTimeout(() => deadline.reject(new Error("picker upload test deadline")), 4_000); + void deadline.promise.catch(() => {}); + const wait = (p: PromiseLike) => Promise.race([p, deadline.promise]); + try { await wait(run(wait)); } finally { clearTimeout(timer); } + }, 5_000); +} + +function h2Result(stream: ClientHttp2Stream) { + const result = Promise.withResolvers<{ status: number; text: string }>(); + let status = 0; + const chunks: Buffer[] = []; + stream.on("response", headers => { status = Number(headers[":status"]); }); + stream.on("data", chunk => chunks.push(Buffer.from(chunk))); + stream.once("end", () => result.resolve({ status, text: Buffer.concat(chunks).toString() })); + stream.once("error", result.reject); + void result.promise.catch(() => {}); + return result.promise; +} +function h1Result(req: ClientRequest) { + const result = Promise.withResolvers<{ status: number; text: string }>(); + req.once("response", res => { + const chunks: Buffer[] = []; + res.on("data", chunk => chunks.push(Buffer.from(chunk))); + res.once("end", () => result.resolve({ status: res.statusCode!, text: Buffer.concat(chunks).toString() })); + res.once("error", result.reject); + }); + req.once("error", result.reject); + void result.promise.catch(() => {}); + return result.promise; +} +function observeClose(stream: ClientRequest | ClientHttp2Stream | ClientHttp2Session) { + const closed = Promise.withResolvers(); + stream.once("close", () => closed.resolve()); + stream.on("error", () => stream.destroy()); + return closed.promise; +} +function observeSocketClose(req: ClientRequest) { + const closed = Promise.withResolvers(); + req.once("socket", socket => { socket.once("close", () => closed.resolve()); }); + req.on("error", () => req.destroy()); + return closed.promise; +} +function normalH2(session: ClientHttp2Session, path = "/ok") { + const req = session.request({ ":method": "GET", ":path": path, ":authority": "claude.ai" }); + const result = h2Result(req); req.end(); return result; +} + +for (const protocol of ["h1", "h2"] as const) { + for (const status of [400, 503] as const) { + for (const input of ["headers-only", "buffered"] as const) { + bounded(`${protocol} ${status} refusal closes a ${input} upload without body completion`, async wait => { + const held = Promise.withResolvers(); + const left = Promise.withResolvers(); + let relayed = 0; + const f = await fixture((req, res) => { + req.on("error", () => res.destroy()); req.resume(); + if (req.url === "/held") { + req.once("close", () => left.resolve()); held.resolve(); + } else { relayed++; res.end("ordinary"); } + }, { maxActiveUpstreams: 1 }); + try { + const session = f.h2(); + let blocker: ClientHttp2Stream | undefined; + let blockerClosed: Promise | undefined; + if (status === 503) { + blocker = session.request({ ":method": "POST", ":path": "/held", ":authority": "claude.ai" }); + blockerClosed = observeClose(blocker); blocker.write("x"); await wait(held.promise); + } + const path = status === 400 ? "//[" : "/refused"; + const upload = protocol === "h2" + ? session.request({ ":method": "POST", ":path": path, ":authority": "claude.ai" }, { endStream: false }) + : f.h1(path, "POST", { "Content-Length": String(32 * 1024 * 1024) }); + const result = protocol === "h2" ? h2Result(upload as ClientHttp2Stream) : h1Result(upload as ClientRequest); + const closed = protocol === "h2" ? observeClose(upload) : observeSocketClose(upload as ClientRequest); + let connection: string | undefined; + if (protocol === "h1") { + (upload as ClientRequest).once("response", res => { connection = res.headers.connection; }); + if (input === "headers-only") (upload as ClientRequest).flushHeaders(); + } + if (input === "buffered") upload.write(Buffer.alloc(4 * 1024 * 1024, 0x61)); + expect(await wait(result)).toEqual({ status, text: "" }); + if (protocol === "h1") expect(connection).toBe("close"); + await wait(closed); + if (protocol === "h2") expect((upload as ClientHttp2Stream).rstCode).toBe(constants.NGHTTP2_NO_ERROR); + expect(relayed).toBe(0); + if (blocker) { + blocker.close(constants.NGHTTP2_CANCEL); await wait(blockerClosed!); await wait(left.promise); + } + expect(await wait(normalH2(session))).toEqual({ status: 200, text: "ordinary" }); + } finally { await wait(f.close()); } + }); + } + } +} + +bounded("h1 completed upload keeps the same connection alive after an upstream failure", async wait => { + const body = "complete"; + let received = ""; + let complete = false; + const f = await fixture((req, res) => { + req.on("error", () => res.destroy()); + if (req.url === "/failed") { + req.on("data", chunk => { received += chunk.toString(); }); + req.once("end", () => { complete = req.complete; req.socket.destroy(); }); + } else { req.resume(); res.end("ordinary"); } + }, { maxActiveUpstreams: 1 }); + try { + const upload = f.h1("/failed", "POST", { "Content-Length": String(body.length) }); + const firstSocket = Promise.withResolvers(); + upload.once("socket", firstSocket.resolve); + let connection: string | undefined; + upload.once("response", res => { connection = res.headers.connection; }); + const result = h1Result(upload); upload.end(body); + expect(await wait(result)).toEqual({ status: 502, text: "" }); + expect(received).toBe(body); + expect(complete).toBe(true); + expect(connection).toBe("keep-alive"); + const next = f.h1("/next"); + const nextSocket = Promise.withResolvers(); + next.once("socket", nextSocket.resolve); + const nextResult = h1Result(next); next.end(); + expect(await wait(nextResult)).toEqual({ status: 200, text: "ordinary" }); + expect(await wait(nextSocket.promise)).toBe(await wait(firstSocket.promise)); + } finally { await wait(f.close()); } +}); + +for (const protocol of ["h1", "h2"] as const) { + bounded(`${protocol} completed upload preserves SSE while a sibling upload is cancelled`, async wait => { + const events = Promise.withResolvers(); + const admitted = Promise.withResolvers(); + const upstreamClosed = Promise.withResolvers(); + let sse: ServerResponse | undefined; + const f = await fixture((req, res) => { + req.on("error", () => res.destroy()); req.resume(); + if (req.url === "/events") req.once("end", () => { + sse = res; res.writeHead(200, { "Content-Type": "text/event-stream" }); + res.write("data: first\n\n"); events.resolve(); + }); + else if (req.url === "/held") { + req.once("data", () => admitted.resolve()); + req.once("close", () => upstreamClosed.resolve()); + } else res.end("ordinary"); + }, { maxActiveUpstreams: 2 }); + try { + const session = f.h2(); + const first = protocol === "h2" + ? session.request({ ":method": "POST", ":path": "/events", ":authority": "claude.ai" }) + : f.h1("/events", "POST", { "Content-Length": "8" }); + const result = protocol === "h2" ? h2Result(first as ClientHttp2Stream) : h1Result(first as ClientRequest); + first.end("complete"); await wait(events.promise); + const second = session.request({ ":method": "POST", ":path": "/held", ":authority": "claude.ai" }); + const secondClosed = observeClose(second); second.write("x"); await wait(admitted.promise); + expect(sse!.writableEnded).toBe(false); + second.close(constants.NGHTTP2_CANCEL); await wait(secondClosed); await wait(upstreamClosed.promise); + expect(second.rstCode).toBe(constants.NGHTTP2_CANCEL); + expect(sse!.writableEnded).toBe(false); + expect(await wait(normalH2(session))).toEqual({ status: 200, text: "ordinary" }); + sse!.end("data: last\n\n"); + expect(await wait(result)).toEqual({ status: 200, text: "data: first\n\ndata: last\n\n" }); + } finally { await wait(f.close()); } + }); + + bounded(`${protocol} upstream failure delivers 502 and releases an unfinished upload`, async wait => { + const f = await fixture((req, res) => { + req.on("error", () => res.destroy()); req.resume(); + if (req.url === "/failed") req.socket.destroy(); else res.end("ordinary"); + }, { maxActiveUpstreams: 1 }); + try { + const session = f.h2(); + const upload = protocol === "h2" + ? session.request({ ":method": "POST", ":path": "/failed", ":authority": "claude.ai" }) + : f.h1("/failed", "POST", { "Content-Length": "100" }); + const closed = protocol === "h2" ? observeClose(upload) : observeSocketClose(upload as ClientRequest); + const result = protocol === "h2" ? h2Result(upload as ClientHttp2Stream) : h1Result(upload as ClientRequest); + upload.write("x"); + expect(await wait(result)).toEqual({ status: 502, text: "" }); await wait(closed); + if (protocol === "h2") expect((upload as ClientHttp2Stream).rstCode).toBe(constants.NGHTTP2_NO_ERROR); + const next = session.request({ ":method": "POST", ":path": "/next", ":authority": "claude.ai" }); + const nextResult = h2Result(next); next.end("complete"); + expect(await wait(nextResult)).toEqual({ status: 200, text: "ordinary" }); + expect(await wait(normalH2(session))).toEqual({ status: 200, text: "ordinary" }); + } finally { await wait(f.close()); } + }); + + bounded(`${protocol} shutdown closes unfinished input and upstream`, async wait => { + const arrived = Promise.withResolvers(); + const upstreamClosed = Promise.withResolvers(); + const f = await fixture((req, res) => { + req.on("error", () => res.destroy()); req.resume(); + req.once("close", () => upstreamClosed.resolve()); arrived.resolve(); + }); + try { + const upload = protocol === "h2" + ? f.h2().request({ ":method": "POST", ":path": "/held", ":authority": "claude.ai" }) + : f.h1("/held", "POST", { "Content-Length": "100" }); + const closed = protocol === "h2" ? observeClose(upload) : observeSocketClose(upload as ClientRequest); upload.write("x"); await wait(arrived.promise); + await wait(f.relay.close()); await wait(closed); await wait(upstreamClosed.promise); + } finally { await wait(f.close()); } + }); +} + +for (const protocol of ["h1", "h2"] as const) { + bounded(`${protocol} backpressure preserves every byte of a large early reply`, async wait => { + // Exceed the h2 receive window and the h1 socket buffers, including a non-text tail. + const body = Buffer.alloc(8 * 1024 * 1024); + for (let i = 0; i < body.length; i++) body[i] = i % 251; + const uploadBody = Buffer.alloc(4 * 1024 * 1024, 0x61); + const sent = Promise.withResolvers(); + let inputBytes = 0; + const backpressured = Promise.withResolvers(); + const f = await fixture((req, res) => { + req.on("error", () => res.destroy()); req.resume(); + if (req.url === "/early-large") { + // Consume the known prefix before replying, so the fixture itself has no queued input + // when its HTTP/1.1 connection closes. The client's request body is still unfinished. + req.on("data", chunk => { + inputBytes += chunk.length; + if (inputBytes !== uploadBody.length) return; + res.writeHead(413, { "Content-Length": String(body.length) }); + expect(res.write(body)).toBe(false); + expect(res.writableNeedDrain).toBe(true); + backpressured.resolve(); + res.end(() => sent.resolve()); + }); + } else res.end("ordinary"); + }); + const buffered = Promise.withResolvers<{ response: IncomingMessage | ClientHttp2Stream; resume(): void }>(); + try { + const session = f.h2(); + const upload = protocol === "h2" + ? session.request({ ":method": "POST", ":path": "/early-large", ":authority": "claude.ai" }) + : f.h1("/early-large", "POST", { "Content-Length": String(32 * 1024 * 1024) }); + const closed = protocol === "h2" ? observeClose(upload) : observeSocketClose(upload as ClientRequest); + const result = Promise.withResolvers<{ status: number; body: Buffer }>(); + void result.promise.catch(() => {}); + const collect = (response: IncomingMessage | ClientHttp2Stream, status: number) => { + const chunks: Buffer[] = []; + response.on("data", chunk => chunks.push(Buffer.from(chunk))); + response.once("end", () => result.resolve({ status, body: Buffer.concat(chunks) })); + response.once("error", result.reject); + response.pause(); + // Resume only when the paused reader has a full readable buffer, not after a delay. + const onReadable = () => { + if (response.readableLength < response.readableHighWaterMark) return; + buffered.resolve({ response, resume() { response.off("readable", onReadable); response.resume(); } }); + }; + response.on("readable", onReadable); + onReadable(); + }; + if (protocol === "h2") { + const stream = upload as ClientHttp2Stream; + stream.once("response", headers => collect(stream, Number(headers[":status"]))); + } else (upload as ClientRequest).once("response", res => collect(res, res.statusCode!)); + upload.once("error", result.reject); + upload.write(uploadBody); + await wait(backpressured.promise); + const { response, resume } = await wait(buffered.promise); + expect(response.readableLength).toBeGreaterThanOrEqual(response.readableHighWaterMark); + if (protocol === "h1") expect((response as IncomingMessage).headers.connection).toBe("close"); + resume(); + const received = await wait(result.promise); + await wait(sent.promise); + expect(inputBytes).toBe(uploadBody.length); + expect(received.status).toBe(413); + expect(received.body.length).toBe(body.length); + expect(received.body.equals(body)).toBe(true); + await wait(closed); + if (protocol === "h2") expect((upload as ClientHttp2Stream).rstCode).toBe(constants.NGHTTP2_NO_ERROR); + expect(await wait(normalH2(session))).toEqual({ status: 200, text: "ordinary" }); + } finally { await wait(f.close()); } + }); +} + +bounded("h2 cancellation releases aggregate admission once without weakening the next admission", async wait => { + const arrived = [Promise.withResolvers(), Promise.withResolvers()]; + const left = Promise.withResolvers(); + let count = 0; + const f = await fixture((req, res) => { + req.on("error", () => res.destroy()); req.resume(); + if (req.url === "/first") req.once("close", () => left.resolve()); + arrived[count++]?.resolve(); + }, { maxActiveUpstreams: 1 }); + try { + const session = f.h2(); + const first = session.request({ ":method": "POST", ":path": "/first", ":authority": "claude.ai" }); + const firstClosed = observeClose(first); first.write("x"); await wait(arrived[0]!.promise); + first.close(constants.NGHTTP2_CANCEL); await wait(firstClosed); await wait(left.promise); + const second = session.request({ ":method": "POST", ":path": "/second", ":authority": "claude.ai" }); + observeClose(second); second.write("x"); await wait(arrived[1]!.promise); + const third = session.request({ ":method": "POST", ":path": "/third", ":authority": "claude.ai" }); + const result = h2Result(third); const thirdClosed = observeClose(third); third.write("x"); + expect((await wait(result)).status).toBe(503); await wait(thirdClosed); + expect(count).toBe(2); + } finally { await wait(f.close()); } +}); + +for (const protocol of ["h1", "h2"] as const) { + bounded(`${protocol} early upstream response closes an unfinished upload without losing the reply`, async wait => { + const f = await fixture((req, res) => { + req.on("error", () => res.destroy()); + if (req.url === "/early") { res.writeHead(413, { "Content-Length": "5" }); res.end("early"); } + else res.end("ordinary"); + }); + try { + const session = f.h2(); + const upload = protocol === "h2" + ? session.request({ ":method": "POST", ":path": "/early", ":authority": "claude.ai" }) + : f.h1("/early", "POST", { "Content-Length": "100" }); + const result = protocol === "h2" ? h2Result(upload as ClientHttp2Stream) : h1Result(upload as ClientRequest); + const closed = protocol === "h2" ? observeClose(upload) : observeSocketClose(upload as ClientRequest); upload.write("x"); + expect(await wait(result)).toEqual({ status: 413, text: "early" }); + await wait(closed); + if (protocol === "h2") expect((upload as ClientHttp2Stream).rstCode).toBe(constants.NGHTTP2_NO_ERROR); + expect(await wait(normalH2(session))).toEqual({ status: 200, text: "ordinary" }); + } finally { await wait(f.close()); } + }); +} diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index b41a1beb973..d2539df9b66 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -1,4 +1,5 @@ { + "claude-picker-upload.test.ts": "claude-integration", "messages-native-scope.test.ts": "claude-integration", "cli-log-view-filter.test.ts": "cli", "cli-usage-model-search.test.ts": "cli",