From a5f51d1cf7e2f094a72b5eeb4378ea724c19b0aa Mon Sep 17 00:00:00 2001 From: ding113 Date: Sat, 8 Aug 2026 14:45:43 +0800 Subject: [PATCH 1/2] fix(proxy): prevent shared mutation and replay memory retention Rectifiers and cache TTL overrides mutated shared nested arrays in place, leaking attempt-specific edits across concurrent shadow sessions and the original request. Switch to copy-on-write: filtered arrays are assigned through the top-level message object, TTL override rebuilds only changed message entries, and shadow sessions shallow-copy the request message while sharing the readonly buffer instead of deep-cloning multi-MB request bodies per shadow. ReplaySpool retained full payload while Redis or PG writes blocked and abort/disable raced with in-flight flushes. Payload is now snapshotted and cleared before persistence; abort immediately releases accumulated parts and queued batches, deduplicates concurrent calls through a shared barrier, and fences store cleanup through the writeChain so concurrency quota is freed only after cleanup completes. Streaming detection now reads the stream flag directly from the outgoing message instead of re-parsing the serialized body. --- .../v1/_lib/proxy/billing-header-rectifier.ts | 8 +- src/app/v1/_lib/proxy/forwarder.ts | 30 +-- src/app/v1/_lib/proxy/replay/replay-spool.ts | 88 +++++--- .../proxy/billing-header-rectifier.test.ts | 15 +- tests/unit/proxy/cache-ttl-override.test.ts | 29 +-- .../proxy-forwarder-hedge-first-byte.test.ts | 28 +++ tests/unit/proxy/replay-spool.test.ts | 202 +++++++++++++++++- 7 files changed, 335 insertions(+), 65 deletions(-) diff --git a/src/app/v1/_lib/proxy/billing-header-rectifier.ts b/src/app/v1/_lib/proxy/billing-header-rectifier.ts index f52693dc5..e12a97270 100644 --- a/src/app/v1/_lib/proxy/billing-header-rectifier.ts +++ b/src/app/v1/_lib/proxy/billing-header-rectifier.ts @@ -20,7 +20,7 @@ const BILLING_HEADER_PATTERN = /^\s*x-anthropic-billing-header\s*:/i; /** * Remove x-anthropic-billing-header text blocks from the request system prompt. - * Mutates the message object in place (matches existing rectifier conventions). + * Writes changes back through the top-level message object without mutating shared nested arrays. */ export function rectifyBillingHeader( message: Record @@ -62,11 +62,7 @@ export function rectifyBillingHeader( } if (extractedValues.length > 0) { - // Mutate in place: replace system array contents - system.length = 0; - for (const item of filtered) { - system.push(item); - } + message.system = filtered; return { applied: true, removedCount: extractedValues.length, extractedValues }; } diff --git a/src/app/v1/_lib/proxy/forwarder.ts b/src/app/v1/_lib/proxy/forwarder.ts index a0e8a8701..84bc26942 100644 --- a/src/app/v1/_lib/proxy/forwarder.ts +++ b/src/app/v1/_lib/proxy/forwarder.ts @@ -543,17 +543,23 @@ export function applyCacheTtlOverrideToMessage( // messages[].content[] const messages = message.messages; if (Array.isArray(messages)) { - for (const msg of messages) { + let nextMessages: unknown[] | null = null; + for (let index = 0; index < messages.length; index += 1) { + const msg = messages[index]; if (!msg || typeof msg !== "object") continue; const msgObj = msg as Record; const content = msgObj.content; if (!Array.isArray(content)) continue; const result = applyTtlToContentBlocks(content, ttl); if (result.applied) { - msgObj.content = result.blocks; + nextMessages ??= [...messages]; + nextMessages[index] = { ...msgObj, content: result.blocks }; applied = true; } } + if (nextMessages) { + message.messages = nextMessages; + } } return applied; @@ -1166,7 +1172,11 @@ async function tryApplyReactiveRectifier(params: { } const requestDetailsBeforeRectify = buildRequestDetails(requestSession); - const rectified = descriptor.rectify(requestSession.request.message as Record); + const mutableMessage = structuredClone( + requestSession.request.message as Record + ); + requestSession.request.message = mutableMessage; + const rectified = descriptor.rectify(mutableMessage); addSpecialSettingForPersistence( requestSession, @@ -3325,13 +3335,7 @@ export class ProxyForwarder { const bodyString = JSON.stringify(messageToSend); requestBody = bodyString; session.forwardedRequestBody = bodyString; - - try { - const parsed = JSON.parse(bodyString); - isStreaming = parsed.stream === true; - } catch { - isStreaming = false; - } + isStreaming = messageToSend.stream === true; if (process.env.NODE_ENV === "development") { logger.trace("ProxyForwarder: Forwarding request", { @@ -7770,8 +7774,10 @@ export class ProxyForwarder { shadowState.request = { ...session.request, - message: structuredClone(session.request.message), - buffer: session.request.buffer ? session.request.buffer.slice(0) : undefined, + // attempt 改写采用顶层 copy-on-write；发送前的私有参数过滤会生成独立深拷贝。 + message: { ...session.request.message }, + // 原始请求字节只读；shadow 共享底层 buffer，任何改写都必须整体替换属性。 + buffer: session.request.buffer, imageRequestMetadata: cloneOpenAIImageRequestMetadata(session.request.imageRequestMetadata), }; shadow.requestUrl = new URL(session.requestUrl.toString()); diff --git a/src/app/v1/_lib/proxy/replay/replay-spool.ts b/src/app/v1/_lib/proxy/replay/replay-spool.ts index 3691494c3..11817432e 100644 --- a/src/app/v1/_lib/proxy/replay/replay-spool.ts +++ b/src/app/v1/_lib/proxy/replay/replay-spool.ts @@ -40,6 +40,7 @@ export class ReplaySpool { private readonly store = getReplayStore(); private readonly decoder = new TextDecoder("utf-8"); private readonly parts: string[] = []; + private readonly queuedBatches = new Set(); private pending: string[] = []; private pendingBytes = 0; private totalBytes = 0; @@ -120,11 +121,12 @@ export class ReplaySpool { if (batch.length === 0) return; this.pending = []; this.pendingBytes = 0; + this.queuedBatches.add(batch); // 续接体自带 try/catch：链永不 rejected；每个 await 之后复查 disabled， // 防止与 disable/halt 竞态时在清理之后又写回 owning meta this.writeChain = this.writeChain.then(async () => { try { - if (this.disabled) return; + if (this.disabled || this.aborting) return; const expectedChunkCount = this.chunkCount + batch.length; const appended = await this.store.writeOwned( this.identity.replayId, @@ -132,7 +134,7 @@ export class ReplaySpool { this.buildMeta("owning", { chunkCount: expectedChunkCount }), batch ); - if (this.disabled) return; + if (this.disabled || this.aborting) return; if (appended === null) { // Redis 不可用：本次 replay 放弃（热层写是原子的，不会留下半批数据） this.disable("redis_unavailable"); @@ -145,10 +147,13 @@ export class ReplaySpool { this.chunkCount = appended; this.metaWritten = true; } catch (error) { + if (this.aborting) return; logger.debug("[ReplaySpool] flush failed, disabling spool", { error: error instanceof Error ? error.message : String(error), }); this.disable("flush_error"); + } finally { + this.queuedBatches.delete(batch); } }); } @@ -172,15 +177,15 @@ export class ReplaySpool { private startOwnerHeartbeat(): void { this.ownerHeartbeatTimer = setInterval(() => { - if (this.disabled || this.released || this.ownerHeartbeatInFlight) return; + if (this.disabled || this.aborting || this.released || this.ownerHeartbeatInFlight) return; this.ownerHeartbeatInFlight = true; void this.store .renewOwnerLease(this.identity.replayId, this.ownerToken) .then((leaseHeld) => { - if (!leaseHeld && !this.released) this.halt("owner_lease_lost"); + if (!leaseHeld && !this.aborting && !this.released) this.halt("owner_lease_lost"); }) .catch(() => { - if (!this.released) this.halt("owner_lease_lost"); + if (!this.aborting && !this.released) this.halt("owner_lease_lost"); }) .finally(() => { this.ownerHeartbeatInFlight = false; @@ -193,12 +198,13 @@ export class ReplaySpool { bootstrap(): void { this.writeChain = this.writeChain.then(async () => { try { - if (this.disabled || this.metaWritten) return; + if (this.disabled || this.aborting || this.metaWritten) return; const chunkCount = await this.store.writeOwned( this.identity.replayId, this.ownerToken, this.buildMeta("owning") ); + if (this.disabled || this.aborting) return; if (chunkCount === null) { this.disable("redis_unavailable"); return; @@ -210,6 +216,7 @@ export class ReplaySpool { this.chunkCount = chunkCount; this.metaWritten = true; } catch (error) { + if (this.aborting) return; logger.debug("[ReplaySpool] bootstrap failed, disabling spool", { error: error instanceof Error ? error.message : String(error), }); @@ -234,11 +241,12 @@ export class ReplaySpool { const batch = this.pending; this.pending = []; this.pendingBytes = 0; + this.queuedBatches.add(batch); this.writeChain = this.writeChain.then(async () => { let pgPersisted = false; try { - if (this.disabled) return; + if (this.disabled || this.aborting) return; const expectedChunkCount = this.chunkCount + batch.length; const appended = await this.store.writeOwned( this.identity.replayId, @@ -255,6 +263,7 @@ export class ReplaySpool { } this.chunkCount = appended; this.metaWritten = true; + const payload = this.takePayload(); // 先写 PG（持久 payload），再翻 Redis meta 为 completed（热层可服务） const persistResult = await this.store.persistCompleted({ replayId: this.identity.replayId, @@ -266,7 +275,7 @@ export class ReplaySpool { model: this.identity.model, statusCode: this.statusCode, headers: this.headers, - payload: this.parts.join(""), + payload, byteSize: this.totalBytes, sourceMessageRequestId: messageRequestId, }); @@ -312,6 +321,8 @@ export class ReplaySpool { ) .catch(() => false); } finally { + this.queuedBatches.delete(batch); + this.clearPayload(); this.release(); } }); @@ -320,12 +331,19 @@ export class ReplaySpool { /** 终态失败：meta 置 aborted + 删块；已 aborted 的条目绝不被重放命中。 */ async abort(reason: string): Promise { + if (this.abortPromise) { + await this.abortPromise; + return; + } if (this.terminal) return; this.terminal = true; + this.aborting = true; this.clearTimer(); this.pending = []; this.pendingBytes = 0; - this.writeChain = this.writeChain.then(async () => { + this.clearPayload(); + this.clearQueuedBatches(); + this.abortPromise = this.writeChain.then(async () => { try { // 已失效（disable 已清理 / halt 已让渡所有权）：不得再写 meta 覆盖新 owner if (this.disabled) return; @@ -340,7 +358,8 @@ export class ReplaySpool { this.release(); } }); - await this.writeChain; + this.writeChain = this.abortPromise; + await this.abortPromise; } /** 失效并删除条目（payload 超限 / Redis 不可用 / 冲刷异常等本 spool 自身的失败）。 */ @@ -360,31 +379,37 @@ export class ReplaySpool { this.pending = []; this.parts.length = 0; this.pendingBytes = 0; + this.clearQueuedBatches(); // 清理顺着 writeChain 串行：与 in-flight append 竞态时绝不出现「删除后又写回」 this.writeChain = this.writeChain.then(async () => { - if (deleteEntry) { - await this.store - .abortOwned( - this.identity.replayId, - this.ownerToken, - this.buildMeta("aborted", { abortReason: reason }) - ) - .catch(() => false); - } else { - // compare-delete 只删自己的 token：所有权已失时为安全 no-op - await this.store - .releaseOwner(this.identity.replayId, this.ownerToken) - .catch(() => undefined); + try { + if (deleteEntry) { + await this.store + .abortOwned( + this.identity.replayId, + this.ownerToken, + this.buildMeta("aborted", { abortReason: reason }) + ) + .catch(() => false); + } else { + // compare-delete 只删自己的 token：所有权已失时为安全 no-op + await this.store + .releaseOwner(this.identity.replayId, this.ownerToken) + .catch(() => undefined); + } + } finally { + this.release(); } }); logger.debug("[ReplaySpool] spool disabled", { replayId: this.identity.replayId.slice(0, 12), reason, }); - this.release(); } private released = false; + private aborting = false; + private abortPromise: Promise | null = null; private release(): void { if (this.released) return; @@ -409,6 +434,21 @@ export class ReplaySpool { this.clearFlushTimer(); this.clearOwnerHeartbeat(); } + + private takePayload(): string { + const payload = this.parts.join(""); + this.clearPayload(); + return payload; + } + + private clearPayload(): void { + this.parts.length = 0; + } + + private clearQueuedBatches(): void { + for (const batch of this.queuedBatches) batch.length = 0; + this.queuedBatches.clear(); + } } /** Forwarder 在 spool 创建前终止时，以 owner token 原子封死 Replay 条目。 */ diff --git a/tests/unit/proxy/billing-header-rectifier.test.ts b/tests/unit/proxy/billing-header-rectifier.test.ts index c9213ca1d..e4a973eb9 100644 --- a/tests/unit/proxy/billing-header-rectifier.test.ts +++ b/tests/unit/proxy/billing-header-rectifier.test.ts @@ -56,13 +56,12 @@ describe("rectifyBillingHeader", () => { }); test("system array with billing header mixed with real prompts - only removes billing header blocks", () => { - const message: Record = { - system: [ - { type: "text", text: "You are a helpful assistant." }, - { type: "text", text: "x-anthropic-billing-header: cc_version=2.1.36; cch=1;" }, - { type: "text", text: "Follow instructions carefully." }, - ], - }; + const originalSystem = [ + { type: "text", text: "You are a helpful assistant." }, + { type: "text", text: "x-anthropic-billing-header: cc_version=2.1.36; cch=1;" }, + { type: "text", text: "Follow instructions carefully." }, + ]; + const message: Record = { system: originalSystem }; const result = rectifyBillingHeader(message); @@ -72,6 +71,8 @@ describe("rectifyBillingHeader", () => { { type: "text", text: "You are a helpful assistant." }, { type: "text", text: "Follow instructions carefully." }, ]); + expect(message.system).not.toBe(originalSystem); + expect(originalSystem).toHaveLength(3); }); test("system as plain string that IS a billing header - deletes system field", () => { diff --git a/tests/unit/proxy/cache-ttl-override.test.ts b/tests/unit/proxy/cache-ttl-override.test.ts index 5a38e9e4a..3403bc3e5 100644 --- a/tests/unit/proxy/cache-ttl-override.test.ts +++ b/tests/unit/proxy/cache-ttl-override.test.ts @@ -42,20 +42,19 @@ describe("applyCacheTtlOverrideToMessage", () => { }); it("rewrites ttl on messages[].content[] ephemeral blocks (existing behavior)", () => { - const message: Record = { - messages: [ - { - role: "user", - content: [ - { - type: "text", - text: "hello", - cache_control: { type: "ephemeral" }, - }, - ], - }, - ], - }; + const originalMessages = [ + { + role: "user", + content: [ + { + type: "text", + text: "hello", + cache_control: { type: "ephemeral" }, + }, + ], + }, + ]; + const message: Record = { messages: originalMessages }; const applied = applyCacheTtlOverrideToMessage(message, "1h"); @@ -67,6 +66,8 @@ describe("applyCacheTtlOverrideToMessage", () => { type: "ephemeral", ttl: "1h", }); + expect(message.messages).not.toBe(originalMessages); + expect(originalMessages[0].content[0].cache_control).toEqual({ type: "ephemeral" }); }); it("rewrites both system and messages breakpoints in a single pass", () => { diff --git a/tests/unit/proxy/proxy-forwarder-hedge-first-byte.test.ts b/tests/unit/proxy/proxy-forwarder-hedge-first-byte.test.ts index 8d2d8948f..02cab8a5f 100644 --- a/tests/unit/proxy/proxy-forwarder-hedge-first-byte.test.ts +++ b/tests/unit/proxy/proxy-forwarder-hedge-first-byte.test.ts @@ -674,6 +674,34 @@ describe("ProxyForwarder - first-byte hedge scheduling", () => { expect(sessionState.currentModelRedirect.redirect.redirectedModel).toBe(fireworksRedirect); }); + test("shadow sessions share readonly request data while isolating top-level attempt state", () => { + const session = createSession(); + const requestBuffer = new ArrayBuffer(4 * 1024 * 1024); + session.request.buffer = requestBuffer; + + const createShadow = ( + ProxyForwarder as unknown as { + createStreamingShadowSession: (session: ProxySession, provider: Provider) => ProxySession; + } + ).createStreamingShadowSession; + const shadows = Array.from({ length: 4 }, (_, index) => + createShadow(session, createProvider({ id: index + 10, name: `p${index + 10}` })) + ); + + expect(shadows.every((shadow) => shadow.request.buffer === requestBuffer)).toBe(true); + expect(new Set(shadows.map((shadow) => shadow.request.buffer)).size).toBe(1); + + shadows[0].request.message.model = "shadow-only"; + + expect(session.request.message.model).not.toBe("shadow-only"); + expect(shadows[1].request.message.model).not.toBe("shadow-only"); + expect(shadows[0].request.message.messages).toBe(session.request.message.messages); + + shadows[0].request.buffer = new ArrayBuffer(16); + expect(session.request.buffer).toBe(requestBuffer); + expect(shadows[1].request.buffer).toBe(requestBuffer); + }); + test("switching to provider without redirect should clear stale redirect snapshot", () => { const requestedModel = "claude-haiku-4-5-20251001"; const fireworksRedirect = "accounts/fireworks/routers/kimi-k2p5-turbo"; diff --git a/tests/unit/proxy/replay-spool.test.ts b/tests/unit/proxy/replay-spool.test.ts index f3988350e..e366083d2 100644 --- a/tests/unit/proxy/replay-spool.test.ts +++ b/tests/unit/proxy/replay-spool.test.ts @@ -156,6 +156,13 @@ async function drainWriteChain(spool: ReplaySpool): Promise { await (spool as unknown as { writeChain: Promise }).writeChain; } +function retainedAsciiPartBytes(spool: ReplaySpool): number { + return (spool as unknown as { parts: string[] }).parts.reduce( + (total, part) => total + part.length, + 0 + ); +} + function makeOwnerSession(): ProxySession { return { replayState: { identity, ownerToken: "owner-token", role: "owner" }, @@ -365,9 +372,11 @@ describe("ReplaySpool：超尺寸自失效", () => { spool.observe(encoder.encode("x".repeat(32))); - // 计数同步归还；存储清理顺着 writeChain 串行执行（避免与 in-flight append 竞态） - expect(getActiveReplaySpoolCount()).toBe(0); + // 存储清理顺着 writeChain 串行执行；清理完成前继续占用 quota， + // 避免慢 Redis 下不断创建新 spool 绕过并发内存上限。 + expect(getActiveReplaySpoolCount()).toBe(1); await drainWriteChain(spool); + expect(getActiveReplaySpoolCount()).toBe(0); expect(storeControl.store.abortOwned).toHaveBeenCalledWith( identity.replayId, "owner-token", @@ -531,6 +540,76 @@ describe("ReplaySpool：completeAfterBilling 终态屏障", () => { expect(storeControl.store.completeOwned).toHaveBeenCalledTimes(1); }); + it("Redis 阻塞期间不提前复制 payload，PG 阻塞期间释放 parts", async () => { + const payloadBytes = 4 * 1024 * 1024; + const chunk = "x".repeat(64 * 1024); + let resolveRedis!: (value: number) => void; + let resolvePersist!: (value: "persisted") => void; + storeControl.store.writeOwned.mockImplementationOnce( + () => + new Promise((resolve) => { + resolveRedis = resolve; + }) + ); + storeControl.store.persistCompleted.mockImplementationOnce( + () => + new Promise<"persisted">((resolve) => { + resolvePersist = resolve; + }) + ); + const spool = makeSpool(); + for (let index = 0; index < 64; index += 1) { + spool.observe(encoder.encode(chunk)); + } + expect(retainedAsciiPartBytes(spool)).toBe(payloadBytes); + + const completion = spool.completeAfterBilling(10); + await vi.advanceTimersByTimeAsync(0); + + expect(storeControl.store.writeOwned).toHaveBeenCalledTimes(1); + expect(storeControl.store.persistCompleted).not.toHaveBeenCalled(); + expect(retainedAsciiPartBytes(spool)).toBe(payloadBytes); + + resolveRedis(1); + await vi.advanceTimersByTimeAsync(0); + + expect(storeControl.store.persistCompleted).toHaveBeenCalledTimes(1); + expect(storeControl.store.persistCompleted).toHaveBeenCalledWith( + expect.objectContaining({ + payload: "x".repeat(payloadBytes), + byteSize: payloadBytes, + }) + ); + expect(retainedAsciiPartBytes(spool)).toBe(0); + + resolvePersist("persisted"); + await completion; + }); + + it("payload 组装失败时封死热层并释放 heartbeat 与并发配额", async () => { + const spool = makeSpool(); + spool.observe(encoder.encode("data: partial\n\n")); + const parts = (spool as unknown as { parts: unknown[] }).parts; + parts[0] = { + toString: () => { + throw new Error("payload assembly failed"); + }, + }; + + await spool.completeAfterBilling(10); + + expect(storeControl.store.persistCompleted).not.toHaveBeenCalled(); + expect(storeControl.store.abortOwned).toHaveBeenCalledWith( + identity.replayId, + "owner-token", + expect.objectContaining({ status: "aborted", abortReason: "complete_failed" }) + ); + expect(getActiveReplaySpoolCount()).toBe(0); + + await vi.advanceTimersByTimeAsync(15_000); + expect(storeControl.store.renewOwnerLease).not.toHaveBeenCalled(); + }); + it("跨 chunk 截断的 UTF-8 序列在 complete 时冲刷解码尾部", async () => { const spool = makeSpool(); // "中" (0xE4 0xB8 0xAD) 只送前两字节：observe 阶段解码挂起，complete 时 flush 出替换字符 @@ -577,6 +656,125 @@ describe("ReplaySpool：abort 终态", () => { expect(getActiveReplaySpoolCount()).toBe(0); }); + it("abort 立即释放已累积的 payload", async () => { + const spool = makeSpool(); + spool.observe(encoder.encode("data: partial\n\n")); + + await spool.abort("upstream_error"); + + expect((spool as unknown as { parts: string[] }).parts).toEqual([]); + }); + + it("Redis flush 阻塞时 abort 立即释放 batch，并在 fenced cleanup 后释放并发配额", async () => { + let resolveRedis!: (value: number) => void; + storeControl.store.writeOwned.mockImplementationOnce( + (...args: unknown[]) => + new Promise((resolve) => { + resolveRedis = resolve; + void args; + }) + ); + const spool = makeSpool(); + spool.observe(encoder.encode("x".repeat(64 * 1024))); + await vi.advanceTimersByTimeAsync(0); + spool.observe(encoder.encode("y".repeat(64 * 1024))); + await vi.advanceTimersByTimeAsync(0); + + expect(storeControl.store.writeOwned).toHaveBeenCalledTimes(1); + const batch = storeControl.store.writeOwned.mock.calls[0][3] as string[]; + expect(batch.length).toBeGreaterThan(0); + const queuedBatches = (spool as unknown as { queuedBatches: Set }).queuedBatches; + expect(queuedBatches.size).toBe(2); + + let abortSettled = false; + const abortPromise = spool.abort("client_disconnect").then(() => { + abortSettled = true; + }); + await vi.advanceTimersByTimeAsync(0); + + expect(batch).toEqual([]); + expect([...queuedBatches].every((queuedBatch) => queuedBatch.length === 0)).toBe(true); + expect(abortSettled).toBe(false); + expect(getActiveReplaySpoolCount()).toBe(1); + expect(storeControl.store.abortOwned).not.toHaveBeenCalled(); + + resolveRedis(1); + await abortPromise; + expect(abortSettled).toBe(true); + expect(getActiveReplaySpoolCount()).toBe(0); + expect(storeControl.store.abortOwned).toHaveBeenCalledWith( + identity.replayId, + "owner-token", + expect.objectContaining({ status: "aborted", abortReason: "client_disconnect" }) + ); + }); + + it("bootstrap 阻塞时 abort 等待真正的 fenced cleanup 后再释放并发配额", async () => { + let resolveBootstrap!: (value: null) => void; + let resolveCleanup!: (value: boolean) => void; + storeControl.store.writeOwned.mockImplementationOnce( + () => + new Promise((resolve) => { + resolveBootstrap = resolve; + }) + ); + storeControl.store.abortOwned.mockImplementationOnce( + () => + new Promise((resolve) => { + resolveCleanup = resolve; + }) + ); + const spool = makeSpool(); + spool.bootstrap(); + await vi.advanceTimersByTimeAsync(0); + + let abortSettled = false; + const abortPromise = spool.abort("client_disconnect").then(() => { + abortSettled = true; + }); + await vi.advanceTimersByTimeAsync(0); + + resolveBootstrap(null); + await vi.advanceTimersByTimeAsync(0); + + expect(storeControl.store.abortOwned).toHaveBeenCalledTimes(1); + expect(abortSettled).toBe(false); + expect(getActiveReplaySpoolCount()).toBe(1); + + resolveCleanup(true); + await abortPromise; + expect(abortSettled).toBe(true); + expect(getActiveReplaySpoolCount()).toBe(0); + }); + + it("并发重复 abort 都等待同一个 fenced cleanup barrier", async () => { + let resolveRedis!: (value: number) => void; + storeControl.store.writeOwned.mockImplementationOnce( + () => + new Promise((resolve) => { + resolveRedis = resolve; + }) + ); + const spool = makeSpool(); + spool.observe(encoder.encode("x".repeat(64 * 1024))); + await vi.advanceTimersByTimeAsync(0); + + const firstAbort = spool.abort("client_disconnect"); + let secondAbortSettled = false; + const secondAbort = spool.abort("client_disconnect").then(() => { + secondAbortSettled = true; + }); + await vi.advanceTimersByTimeAsync(0); + + const secondSettledBeforeCleanup = secondAbortSettled; + resolveRedis(1); + await Promise.all([firstAbort, secondAbort]); + + expect(secondSettledBeforeCleanup).toBe(false); + expect(getActiveReplaySpoolCount()).toBe(0); + expect(storeControl.store.abortOwned).toHaveBeenCalledTimes(1); + }); + it("abort 后 observe 与 complete 均无副作用", async () => { const spool = makeSpool(); await spool.abort("client_disconnect"); From 81c1c11ecafc75849284ce9c03cf9393b5e5425c Mon Sep 17 00:00:00 2001 From: "claude[bot]" <41898282+claude[bot]@users.noreply.github.com> Date: Sat, 8 Aug 2026 07:31:45 +0000 Subject: [PATCH 2/2] fix(types): resolve tsgo type inference failures for zod 4 body parsing tsgo (TypeScript native preview) could not infer the generic output type from JsonBodySchema's structural safeParse signature because zod 4 uses this-type polymorphism (core.output) instead of a plain generic. This left body.data as unknown across all v1 API handlers, producing 36 TS18046/TS2698/TS2345 errors. Switch parseHonoJsonBody/parseJsonBody/parseJson to infer the schema type directly () and extract the output via z.output, which tsgo resolves correctly. Add explicit parameter annotations to four zod .transform()/.refine() callbacks where tsgo also failed to infer the this-dependent input type. CI Run: https://github.com/ding113/claude-code-hub/actions/runs/31245382874 Co-Authored-By: Claude Opus 4.6 --- src/app/api/v1/resources/providers/handlers.ts | 10 ++++------ src/lib/api/v1/_shared/request-body.ts | 16 ++++++---------- src/lib/api/v1/schemas/audit-logs.ts | 4 +++- src/lib/api/v1/schemas/me.ts | 2 +- src/lib/api/v1/schemas/system-config.ts | 2 +- src/lib/api/v1/schemas/usage-logs.ts | 2 +- 6 files changed, 16 insertions(+), 20 deletions(-) diff --git a/src/app/api/v1/resources/providers/handlers.ts b/src/app/api/v1/resources/providers/handlers.ts index 93398344c..3129765f5 100644 --- a/src/app/api/v1/resources/providers/handlers.ts +++ b/src/app/api/v1/resources/providers/handlers.ts @@ -1,5 +1,4 @@ import type { Context } from "hono"; -import type { ZodError } from "zod"; import { z } from "zod"; import type { ActionResult } from "@/actions/types"; import { hasLegacyRedactedWritePlaceholders } from "@/lib/api/legacy-action-sanitizers"; @@ -773,11 +772,10 @@ function providerNotFound(c: Context): Response { }); } -type JsonBodySchema = { - safeParse: (value: unknown) => { success: true; data: T } | { success: false; error: ZodError }; -}; - -async function parseJson(c: Context, schema: JsonBodySchema): Promise { +async function parseJson( + c: Context, + schema: S +): Promise | Response> { const body = await parseHonoJsonBody(c, schema); if (!body.ok) return body.response; return body.data; diff --git a/src/lib/api/v1/_shared/request-body.ts b/src/lib/api/v1/_shared/request-body.ts index 60094b359..9013d5a93 100644 --- a/src/lib/api/v1/_shared/request-body.ts +++ b/src/lib/api/v1/_shared/request-body.ts @@ -3,10 +3,6 @@ import { createProblemResponse, normalizeZodPath } from "./error-envelope"; export type ParsedBodyResult = { ok: true; data: T } | { ok: false; response: Response }; -type JsonBodySchema = { - safeParse: (value: unknown) => { success: true; data: T } | { success: false; error: z.ZodError }; -}; - type ParseJsonBodyOptions = { validationErrorCode?: (error: z.ZodError) => string | undefined; }; @@ -20,10 +16,10 @@ type HonoJsonRequest = { }; }; -export async function parseJsonBody( +export async function parseJsonBody( request: Request, - schema: JsonBodySchema -): Promise> { + schema: S +): Promise>> { const contentType = request.headers.get("content-type") ?? ""; if (!contentType.toLowerCase().includes("application/json")) { return { @@ -73,11 +69,11 @@ export async function parseJsonBody( return { ok: true, data: parsed.data }; } -export async function parseHonoJsonBody( +export async function parseHonoJsonBody( c: HonoJsonRequest, - schema: JsonBodySchema, + schema: S, options?: ParseJsonBodyOptions -): Promise> { +): Promise>> { const contentType = c.req.header("content-type") ?? c.req.header("Content-Type") ?? diff --git a/src/lib/api/v1/schemas/audit-logs.ts b/src/lib/api/v1/schemas/audit-logs.ts index 5814d7386..14635b1a9 100644 --- a/src/lib/api/v1/schemas/audit-logs.ts +++ b/src/lib/api/v1/schemas/audit-logs.ts @@ -23,7 +23,9 @@ export const AuditLogListQuerySchema = z.object({ success: z .enum(["true", "false"]) .optional() - .transform((val) => (val === undefined ? undefined : val === "true")) + .transform((val: "true" | "false" | undefined) => + val === undefined ? undefined : val === "true" + ) .describe("Optional success filter."), from: IsoDateTimeStringSchema.optional().describe("Optional inclusive start time."), to: IsoDateTimeStringSchema.optional().describe("Optional inclusive end time."), diff --git a/src/lib/api/v1/schemas/me.ts b/src/lib/api/v1/schemas/me.ts index 4880d1240..39b3856a2 100644 --- a/src/lib/api/v1/schemas/me.ts +++ b/src/lib/api/v1/schemas/me.ts @@ -3,7 +3,7 @@ import { z } from "@hono/zod-openapi"; const NumberQuerySchema = z.coerce.number().optional(); const BooleanQuerySchema = z .union([z.literal("true"), z.literal("false"), z.boolean()]) - .transform((value) => value === true || value === "true") + .transform((value: "true" | "false" | boolean) => value === true || value === "true") .optional(); export const MeUsageLogsQuerySchema = z.object({ diff --git a/src/lib/api/v1/schemas/system-config.ts b/src/lib/api/v1/schemas/system-config.ts index 5dd0253a6..944e4cab2 100644 --- a/src/lib/api/v1/schemas/system-config.ts +++ b/src/lib/api/v1/schemas/system-config.ts @@ -28,7 +28,7 @@ const CodexPriorityBillingSourceSchema = z const TimeZoneSchema = z .string() .refine( - (value) => { + (value: string) => { try { new Intl.DateTimeFormat("en-US", { timeZone: value }); return true; diff --git a/src/lib/api/v1/schemas/usage-logs.ts b/src/lib/api/v1/schemas/usage-logs.ts index 2b01bc88a..6a63e3f67 100644 --- a/src/lib/api/v1/schemas/usage-logs.ts +++ b/src/lib/api/v1/schemas/usage-logs.ts @@ -3,7 +3,7 @@ import { z } from "@hono/zod-openapi"; const NumberQuerySchema = z.coerce.number().optional(); const BooleanQuerySchema = z .union([z.literal("true"), z.literal("false"), z.boolean()]) - .transform((value) => value === true || value === "true") + .transform((value: "true" | "false" | boolean) => value === true || value === "true") .optional(); export const UsageLogsQuerySchema = z.object({