diff --git a/docs-site/src/content/docs/guides/claude-code.md b/docs-site/src/content/docs/guides/claude-code.md index 5b688af5fbf..4fa06efa41f 100644 --- a/docs-site/src/content/docs/guides/claude-code.md +++ b/docs-site/src/content/docs/guides/claude-code.md @@ -316,6 +316,11 @@ applies to it as well. Desktop and CLI catalog rewrites share bounded row and me if adding routed models would exceed a limit, OpenCodex returns the original Anthropic catalog unchanged rather than publishing a partial list. +The picker relay limits unfinished uploads separately from its overall request capacity. An upload +closes after 30 seconds without body progress or five minutes total; finish or retry the upload +rather than keeping it open. These deadlines stop when the request body completes and do not +limit a long-lived response or SSE subscription. Excess uploads receive an empty 503 response. + ### Use opencodex models from the Desktop Code tab (first-party bindings) In first-party mode the Code tab's model picker belongs to claude.ai: its rows (Opus 5.5, diff --git a/docs-site/src/content/docs/ko/guides/claude-code.md b/docs-site/src/content/docs/ko/guides/claude-code.md index 378e672bd99..3d0838ceb96 100644 --- a/docs-site/src/content/docs/ko/guides/claude-code.md +++ b/docs-site/src/content/docs/ko/guides/claude-code.md @@ -193,6 +193,11 @@ Picker 모드를 끄거나 Desktop을 완전히 다시 시작할 때까지 Deskt Picker 모드는 1P의 일부이므로 [1P 계정 위험](#1p직접-선택)도 그대로 적용돼요. +피커 중계는 미완료 업로드 수를 전체 요청 수와 별도로 제한합니다. 본문 전송이 30초간 +진행되지 않거나 총 5분이 지나면 해당 업로드가 종료됩니다. 전송을 완료하거나 다시 +시도하세요. 본문 전송이 끝나면 제한 타이머가 해제되므로 긴 응답이나 SSE 구독의 +유지시간에는 적용되지 않습니다. 업로드 상한을 넘는 요청에는 본문 없는 503이 반환됩니다. + ### Code 탭에서 opencodex 모델 쓰기 (1P 바인딩) 1P 모드에서 Code 탭의 모델 선택기는 claude.ai가 채워요. Opus 5.5, Sonnet 5, Haiku 4.5와 diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index cb8d63e4681..477009925ac 100644 --- a/scripts/test-layout/layout.json +++ b/scripts/test-layout/layout.json @@ -2,6 +2,7 @@ "version": 1, "root": "tests", "explicit": { + "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", diff --git a/src/claude/intercept/picker-listener.ts b/src/claude/intercept/picker-listener.ts index d4dad33ffb9..53e49dff3ba 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"; @@ -34,6 +34,10 @@ export interface PickerListenerOptions { maxEncodedBytes?: number; /** Test seam: concurrent upstream requests; production uses PICKER_MAX_ACTIVE_UPSTREAMS. */ maxActiveUpstreams?: number; + /** Test seams: production uses the fixed upload concurrency and deadline limits below. */ + maxActiveUploads?: number; + uploadIdleTimeoutMs?: number; + uploadTimeoutMs?: number; log?: (line: string) => void; } export interface PickerListenerHandle { port: number; close(): Promise } @@ -47,6 +51,10 @@ export const PICKER_MAX_HEADER_BYTES = 64 * 1024; * Desktop uses (a handful of SSE subscriptions plus bursts of ordinary calls). */ export const PICKER_MAX_ACTIVE_UPSTREAMS = 256; +/** Reserve half the aggregate capacity for requests whose upload is already complete. */ +export const PICKER_MAX_ACTIVE_UPLOADS = 128; +export const PICKER_UPLOAD_IDLE_TIMEOUT_MS = 30_000; +export const PICKER_UPLOAD_TIMEOUT_MS = 300_000; /** Advertised per HTTP/2 connection; Chromium queues further requests rather than failing them. */ export const PICKER_MAX_CONCURRENT_STREAMS = 100; /** A connection that has not sent a complete ClientHello by then is dropped. */ @@ -152,7 +160,11 @@ export async function startPickerListener(options: PickerListenerOptions): Promi const upstream = options.upstream ?? { host: "claude.ai", port: 443, servername: "claude.ai" }; const cap = options.maxEncodedBytes ?? BOOTSTRAP_MAX_ENCODED_BYTES; const maxActiveUpstreams = options.maxActiveUpstreams ?? PICKER_MAX_ACTIVE_UPSTREAMS; + const maxActiveUploads = options.maxActiveUploads ?? PICKER_MAX_ACTIVE_UPLOADS; + const uploadIdleMs = options.uploadIdleTimeoutMs ?? PICKER_UPLOAD_IDLE_TIMEOUT_MS; + const uploadMs = options.uploadTimeoutMs ?? PICKER_UPLOAD_TIMEOUT_MS; let activeUpstreams = 0; + let activeUploads = 0; const upgrades = new Set(); const spliced = new Set(); const h2Sockets = new Set(); @@ -184,14 +196,33 @@ 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 = () => { + if (h2) { + const stream = (req as Http2ServerRequest).stream; + if (!stream.closed && !stream.destroyed) stream.close(h2Constants.NGHTTP2_CANCEL); + } else if (!req.destroyed) 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.once("error", () => res.destroy()); + responseWritable.once("finish", closeInput); + } + res.writeHead(status, { "Content-Length": "0", ...(!h2 && uploading ? { Connection: "close" } : {}) }); res.end(); options.log?.(`picker request refused ${status}`); }; if (h2 && !isRelayableH2Target(req.method, req.url)) { refuse(400); return; } - if (activeUpstreams >= maxActiveUpstreams) { refuse(503); return; } + if (activeUpstreams >= maxActiveUpstreams || (uploading && activeUploads >= maxActiveUploads)) { + refuse(503); return; + } const method = req.method ?? "GET"; let pathname: string; // A target like "//[" reads as an authority and throws; refuse it like any unrelayable target. @@ -215,6 +246,7 @@ export async function startPickerListener(options: PickerListenerOptions): Promi 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,6 +254,7 @@ 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[]) => { @@ -293,6 +326,46 @@ 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) { + activeUploads++; + let uploadFinished = false; + const finishUpload = () => { + if (uploadFinished) return; + uploadFinished = true; + activeUploads--; + clearTimeout(idleTimer); + clearTimeout(totalTimer); + req.off("data", progress); + }; + const expireUpload = (kind: "idle" | "deadline") => { + if (uploadFinished) return; + // Cancellation is stream-scoped on h2. Never impose a response/SSE lifetime limit. + clientGone = true; + req.unpipe(upReq); + upReq.destroy(); + closeInput(); + finishUpload(); + options.log?.(`picker upload expired ${kind}`); + }; + const idleTimer = setTimeout(() => expireUpload("idle"), uploadIdleMs); + const totalTimer = setTimeout(() => expireUpload("deadline"), uploadMs); + idleTimer.unref(); totalTimer.unref(); + const progress = (chunk: Buffer) => { if (chunk.length > 0) idleTimer.refresh(); }; + req.on("data", progress); + req.once("end", finishUpload); + req.once("close", finishUpload); + // Upstream close can precede draining a successful reply to a slow downstream. Keep + // upload deadlines armed until that writable finishes, then cancel only its unread input. + const stopUnfinishedUpload = () => { + if (!uploadFinished) { closeInput(); finishUpload(); } + }; + upReq.once("close", () => { + if (!upstreamResponded && !res.writableEnded) stopUnfinishedUpload(); + }); + responseWritable.once("finish", stopUnfinishedUpload); + 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 39d79a32da0..c83210c7360 100644 --- a/structure/clients/claude-desktop.md +++ b/structure/clients/claude-desktop.md @@ -195,10 +195,9 @@ Production always uses the configured adjacent ports. Lifecycle tests inject onl factory and bind the real handlers on kernel-assigned ports; this preserves request handling while avoiding the false reservation created by probing and closing a port pair before the ephemeral TLS listener starts. The injected factory does not change production port selection. -The User-Agent is a routing hint, not a trust boundary: a client that fakes it reaches only what -any local process already reaches (the `api.anthropic.com` intercept is on the Claude Code proxy -too; the `claude.ai` relay verifies upstream and adds no credential) and breaks only its own TLS, -because each terminator presents a certificate only its intended client trusts. `claude.ai:443` is +The User-Agent is a routing hint, not client authentication. Each terminator presents the +certificate intended for that client; the `claude.ai` relay verifies upstream and adds no credential. +Listener-wide concurrency and upload-lifetime bounds protect the shared relay capacity separately. `claude.ai:443` is intercepted only while the runtime's cached decision is armed: macOS, persisted resolved Desktop mode first-party, Desktop intent on, `claudeCode.intercept.picker !== false`, no disarm latch, listener up, and the current picker CA @@ -214,6 +213,21 @@ 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 a 256-request aggregate ceiling and allows at most 128 unfinished uploads +across both HTTP versions and all sessions, leaving headroom for bodyless requests under the aggregate ceiling. +Upload framing, not the method, selects that sub-budget. Each unfinished upload has a 30-second +idle deadline and a five-minute absolute deadline; progress refreshes only the former. Expiry +cancels its upstream and only its h2 stream (or its HTTP/1.1 connection). Completion clears +upload timers and frees upload admission without timing out a long response/SSE stream. +Early replies cancel unfinished input only after the downstream writable finishes; upstream +request closure alone does not prove a buffered reply was delivered. Upload deadlines remain +armed during that wait. Cancellation, failure and shutdown release upload capacity once. +Rejected uploads close after their empty reply's writable finishes, using the underlying h2 +stream's finish event rather than its compatibility response's stream-close-dependent finish. +These bounds do not authenticate clients or guarantee availability under sustained churn or +exhaustion by completed requests. `tests/claude-integration/claude-picker-upload.test.ts` covers +cross-session admission, both protocol deadlines, progress, headers-only refusals, byte-exact +early replies under backpressure, SSE admission reuse, failure and shutdown cleanup. 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..559c94707fb --- /dev/null +++ b/tests/claude-integration/claude-picker-upload.test.ts @@ -0,0 +1,412 @@ +import { expect, test } from "bun:test"; +import { 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, + limits: 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: () => [], ...limits, + upstream: { host: "127.0.0.1", port: address.port, servername: "claude.ai", ca: ca.certPem }, + log: line => logs.push(line), + }); + const clients: Array = []; + 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: false, 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(); + 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 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 status of [400, 503] as const) { + bounded(`h2 ${status} refusal closes a headers-only upload without DATA or END_STREAM`, async wait => { + let relayed = 0; + const f = await fixture((_req, res) => { relayed++; res.end("ordinary"); }, + { maxActiveUploads: status === 503 ? 0 : 1 }); + try { + const session = f.h2(); + const upload = session.request({ ":method": "POST", ":path": status === 400 ? "//[" : "/refused", + ":authority": "claude.ai" }, { endStream: false }); + const result = h2Result(upload); + const closed = observeClose(upload); + expect(await wait(result)).toEqual({ status, text: "" }); + await wait(closed); + expect(relayed).toBe(0); + expect(await wait(normalH2(session))).toEqual({ status: 200, text: "ordinary" }); + } finally { await wait(f.close()); } + }); +} + +for (const protocol of ["h1", "h2"] as const) { + bounded(`${protocol} body completion reuses the upload slot while its SSE response remains open`, 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: 3, maxActiveUploads: 1, uploadIdleTimeoutMs: 80, uploadTimeoutMs: 160 }); + 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); + const third = session.request({ ":method": "POST", ":path": "/excess", ":authority": "claude.ai" }, { endStream: false }); + const thirdClosed = observeClose(third); + expect((await wait(h2Result(third))).status).toBe(503); await wait(thirdClosed); + second.close(constants.NGHTTP2_CANCEL); await wait(secondClosed); await wait(upstreamClosed.promise); + await wait(new Promise(resolve => setTimeout(resolve, 200))); + 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" }); + expect(f.logs.some(line => line.startsWith("picker upload expired"))).toBe(false); + } 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"); + }, { maxActiveUploads: 1, uploadIdleTimeoutMs: 80, uploadTimeoutMs: 160 }); + 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 = observeClose(upload); + 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); + 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" }); + await wait(new Promise(resolve => setTimeout(resolve, 200))); + expect(await wait(normalH2(session))).toEqual({ status: 200, text: "ordinary" }); + expect(f.logs.some(line => line.startsWith("picker upload expired"))).toBe(false); + } finally { await wait(f.close()); } + }); + + bounded(`${protocol} shutdown closes unfinished input and upstream with no later expiry`, 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(); + }, { uploadIdleTimeoutMs: 80, uploadTimeoutMs: 160 }); + try { + const upload = protocol === "h2" + ? f.h2().request({ ":method": "POST", ":path": "/held", ":authority": "claude.ai" }) + : f.h1("/held", "POST", { "Content-Length": "100" }); + const closed = observeClose(upload); upload.write("x"); await wait(arrived.promise); + await wait(f.relay.close()); await wait(closed); await wait(upstreamClosed.promise); + await wait(new Promise(resolve => setTimeout(resolve, 200))); + expect(f.logs.some(line => line.startsWith("picker upload expired"))).toBe(false); + } 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 sent = Promise.withResolvers(); + const f = await fixture((req, res) => { + req.on("error", () => res.destroy()); + if (req.url === "/early-large") { + res.writeHead(413, { "Content-Length": String(body.length) }); + res.end(body, () => sent.resolve()); + } else res.end("ordinary"); + }, { maxActiveUploads: 1, uploadIdleTimeoutMs: 2_000, uploadTimeoutMs: 3_000 }); + let resumeTimer: ReturnType | undefined; + 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": "100" }); + const closed = observeClose(upload); + 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(); + resumeTimer = setTimeout(() => response.resume(), 150); + }; + 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("x"); + const received = await wait(result.promise); + await wait(sent.promise); + expect(received.status).toBe(413); + expect(received.body.length).toBe(body.length); + expect(received.body.equals(body)).toBe(true); + await wait(closed); + expect(await wait(normalH2(session))).toEqual({ status: 200, text: "ordinary" }); + expect(f.logs.some(line => line.startsWith("picker upload expired"))).toBe(false); + } finally { clearTimeout(resumeTimer); await wait(f.close()); } + }); +} + +bounded("h2 a stalled early reply retains upload deadlines and leaves sibling streams healthy", async wait => { + const upstreamClosed = Promise.withResolvers(); + const f = await fixture((req, res) => { + req.on("error", () => res.destroy()); req.resume(); + if (req.url === "/early-stalled") { + req.socket.once("close", () => upstreamClosed.resolve()); + res.writeHead(413, { "Content-Length": String(8 * 1024 * 1024) }); + res.end(Buffer.alloc(8 * 1024 * 1024, 1)); + } else res.end("ordinary"); + }, { maxActiveUploads: 1, uploadIdleTimeoutMs: 150, uploadTimeoutMs: 300 }); + try { + const session = f.h2(); + const stream = session.request({ ":method": "POST", ":path": "/early-stalled", ":authority": "claude.ai" }); + const closed = observeClose(stream); + stream.once("aborted", () => stream.resume()); stream.pause(); stream.write("x"); + await wait(closed); await wait(upstreamClosed.promise); + expect(f.logs).toContain("picker upload expired idle"); + expect(await wait(normalH2(session))).toEqual({ status: 200, text: "ordinary" }); + await wait(new Promise(resolve => setTimeout(resolve, 200))); + expect(f.logs).not.toContain("picker upload expired deadline"); + } finally { await wait(f.close()); } +}); + +bounded("unfinished uploads across h2 sessions leave capacity for ordinary h1 and h2 requests", async wait => { + let uploads = 0; + const ready = Promise.withResolvers(); + const f = await fixture((req, res) => { + if (req.method === "POST") { req.resume(); if (++uploads === 2) ready.resolve(); } + else res.end("ordinary"); + }, { maxActiveUpstreams: 4, maxActiveUploads: 2 }); + try { + const sessions = [f.h2(), f.h2(), f.h2()]; + for (const session of sessions.slice(0, 2)) { + const stream = session.request({ ":method": "POST", ":path": "/upload", ":authority": "claude.ai" }); + observeClose(stream); stream.write("x"); + } + await wait(ready.promise); + const excess = sessions[2]!.request({ ":method": "POST", ":path": "/excess", ":authority": "claude.ai" }); + const refusal = h2Result(excess); + const closed = observeClose(excess); + excess.write("x"); + expect((await wait(refusal)).status).toBe(503); + await wait(closed); + const excessH1 = f.h1("/excess", "POST", { "Content-Length": "100" }); + const h1Refusal = h1Result(excessH1); excessH1.write("x"); + expect((await wait(h1Refusal)).status).toBe(503); + const get = f.h1("/ordinary"); const result = h1Result(get); get.end(); + expect(await wait(result)).toEqual({ status: 200, text: "ordinary" }); + expect(await wait(normalH2(sessions[0]!))).toEqual({ status: 200, text: "ordinary" }); + expect(uploads).toBe(2); + } finally { await wait(f.close()); } +}); + +for (const protocol of ["h1", "h2"] as const) { + bounded(`${protocol} idle upload expires, closes upstream, and restores capacity`, async wait => { + const ready = Promise.withResolvers(); + const upstreamClosed = Promise.withResolvers(); + const f = await fixture((req, res) => { + if (req.url === "/upload") { + req.on("error", () => res.destroy()); req.resume(); + req.once("close", () => upstreamClosed.resolve()); ready.resolve(); + } else res.end("ordinary"); + }, { maxActiveUpstreams: 2, maxActiveUploads: 1, uploadIdleTimeoutMs: 100, uploadTimeoutMs: 2_000 }); + try { + const session = f.h2(); + const upload = protocol === "h2" + ? session.request({ ":method": "POST", ":path": "/upload", ":authority": "claude.ai" }) + : f.h1("/upload", "POST", { "Content-Length": "100" }); + const closed = observeClose(upload); upload.write("x"); + await wait(ready.promise); await wait(closed); await wait(upstreamClosed.promise); + expect(f.logs).toContain("picker upload expired idle"); + expect(await wait(normalH2(session))).toEqual({ status: 200, text: "ordinary" }); + } finally { await wait(f.close()); } + }); + + bounded(`${protocol} completed upload does not time out a long-lived response`, async wait => { + let finishTimer: ReturnType | undefined; + const f = await fixture((req, res) => { + req.resume(); req.once("end", () => { + res.writeHead(200, { "Content-Type": "text/event-stream" }); res.write("data: first\n\n"); + finishTimer = setTimeout(() => res.end("data: last\n\n"), 300); + }); + }, { uploadIdleTimeoutMs: 80, uploadTimeoutMs: 160 }); + try { + if (protocol === "h2") { + const stream = f.h2().request({ ":method": "POST", ":path": "/events", ":authority": "claude.ai" }); + const result = h2Result(stream); stream.end("complete"); + expect(await wait(result)).toEqual({ status: 200, text: "data: first\n\ndata: last\n\n" }); + } else { + const req = f.h1("/events", "POST", { "Content-Length": "8" }); + const result = h1Result(req); req.end("complete"); + expect(await wait(result)).toEqual({ status: 200, text: "data: first\n\ndata: last\n\n" }); + } + expect(f.logs.some(line => line.startsWith("picker upload expired"))).toBe(false); + } finally { clearTimeout(finishTimer); await wait(f.close()); } + }); +} + +bounded("h2 trickle progress cannot extend the absolute upload deadline", async wait => { + const ready = Promise.withResolvers(); + let chunks = 0; + const f = await fixture((req, res) => { + req.on("error", () => res.destroy()); + req.on("data", () => { chunks++; ready.resolve(); }); + }, { uploadIdleTimeoutMs: 200, uploadTimeoutMs: 450 }); + let tick: ReturnType | undefined; + try { + const stream = f.h2().request({ ":method": "POST", ":path": "/upload", ":authority": "claude.ai" }); + const closed = observeClose(stream); stream.write("x"); + await wait(ready.promise); + tick = setInterval(() => { if (!stream.closed && !stream.destroyed) stream.write("x"); }, 25); + await wait(closed); + expect(chunks).toBeGreaterThan(2); + expect(f.logs).toContain("picker upload expired deadline"); + expect(f.logs).not.toContain("picker upload expired idle"); + } finally { clearInterval(tick); await wait(f.close()); } +}); + +bounded("h2 cancellation frees the upload slot 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: 4, maxActiveUploads: 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); third.write("x"); + expect((await wait(result)).status).toBe(503); + expect(count).toBe(2); + } finally { await wait(f.close()); } +}); + +bounded("an h2 body-bearing GET with no DATA still has an upload deadline", async wait => { + const f = await fixture((req, res) => { req.on("error", () => res.destroy()); req.resume(); }, + { uploadIdleTimeoutMs: 100, uploadTimeoutMs: 2_000 }); + try { + const stream = f.h2().request({ ":method": "GET", ":path": "/held", ":authority": "claude.ai" }, { endStream: false }); + await wait(observeClose(stream)); + expect(f.logs).toContain("picker upload expired idle"); + } 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"); + }, { maxActiveUploads: 1, uploadIdleTimeoutMs: 2_000 }); + 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 = observeClose(upload); upload.write("x"); + expect(await wait(result)).toEqual({ status: 413, text: "early" }); + await wait(closed); + expect(await wait(normalH2(session))).toEqual({ status: 200, text: "ordinary" }); + expect(f.logs.some(line => line.startsWith("picker upload expired"))).toBe(false); + } finally { await wait(f.close()); } + }); +} diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index d993598182c..d826c0b1957 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",