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/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/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({ 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");