diff --git a/.env.example b/.env.example index b6c6d72b8..cb3fd0dfb 100644 --- a/.env.example +++ b/.env.example @@ -105,6 +105,8 @@ STORE_SESSION_RESPONSE_BODY=true # 是否在 Redis 中存储会话响应 # - true：存储（SSE/JSON），用于调试/定位问题（Redis 临时缓存） # - false：不存储响应体（注意：不影响本次请求处理；仅影响后续查看 response body） # 说明：该开关不影响内部统计读取响应体（tokens/费用统计、SSE 假 200 检测仍会进行） +SESSION_RESPONSE_BODY_MAX_BYTES=5242880 # 单份会话响应体 Redis 存储上限（默认 5 MiB，范围 64 KiB-64 MiB） + # 超限正文不落 Redis；before/after snapshot 的 headers/meta 仍保留 # Dashboard 配置 DASHBOARD_LOGS_POLL_INTERVAL_MS=5000 # 日志页自动刷新轮询间隔（毫秒，默认 5000，范围 250-60000） diff --git a/CHANGELOG.md b/CHANGELOG.md index d5474c884..aa6aca9c7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,17 @@ --- +## Unreleased + +### 修复 + +- 修复 Replay owner 在客户端断线后保留完整流正文和 300 秒传输资源导致的内存失控：限制 Redis + write-behind backlog，Replay 失效后按断线起点恢复 60 秒 drain，并为 Redis session response body + 增加默认 5 MiB 的可配置存储上限，避免大 SSE 正文及 before/after 快照放大内存和持久化压力； + 三份 response body 的物理存储去重由 #1415 跟踪 (#1408) + +--- + ## v0.8.7 (2026-06-14) ### 新增 diff --git a/docs/troubleshooting/issue-1408-replay-oom.md b/docs/troubleshooting/issue-1408-replay-oom.md new file mode 100644 index 000000000..57f9b03ff --- /dev/null +++ b/docs/troubleshooting/issue-1408-replay-oom.md @@ -0,0 +1,285 @@ +# Issue #1408 Replay 断线流内存失控根因报告 + +## 调查结论 + +Issue #1408 的 Node 内存失控触发链已经在本地完成机制级复现和因子隔离。该链路能够在受控 +环境中稳定触发同类 V8 heap OOM，并与生产环境的 Replay 配置、客户端断线和 Redis 压力现象 +高度重合。 + +已经证明的主触发链由两个同时存在的生命周期缺陷组成： + +1. `ReplaySpool` 在活跃响应期间把每个流块解码成字符串，同时保存在本地 + `parts[]`、待写 `pending`/`writeChain` batch 和 Redis LIST 中。本地 `parts[]` 会一直保留 + 整条响应，直到上游出现终态或 Replay 被禁用。 +2. Replay owner 的客户端断开后，`ProxyResponseHandler` 把普通 60 秒 drain 窗口改成 + `REPLAY_MAX_DETACHED_MS`，默认 300 秒。spool 后续失效时，这个已选定的窗口不会降级， + 因而失去 Replay 价值的上游流仍可能被保留到 300 秒。 + +这两点使断线流的 JS 字符串、ArrayBuffer、Undici Response、后台任务和 socket 在同一个 +300 秒窗口内按请求波次叠加。持续到来的断线请求不需要形成永久引用泄漏，也可以在最早一批 +请求进入超时回收之前耗尽 V8 heap 或容器内存。 + +生产故障发生时没有 heap snapshot，因此本报告证明的是“存在一条足以解释并复现 #1408 的 +决定性机制”，不是从生产进程对象图中证明了唯一根因。shadow/hedge 请求复制等其他 v0.9.x +内存放大路径仍可能在实际流量中共同贡献。 + +## 适用版本与代码边界 + +- 有效复现版本：`v0.9.2`，提交 `ccbad37f266e3e69d57a4427e2f27cf288796e63` +- 修复目标分支：`dev`，调查时提交 `3fe3225c9f6397d22db27193199e2a8fef4a05f7` +- `/v1/responses` 真实 Codex 转换路径 +- `ENABLE_REQUEST_REPLAY=true` +- `REPLAY_MAX_PAYLOAD_BYTES=8 MiB` +- `REPLAY_MAX_DETACHED_MS=300000` +- `STREAM_GATE_MODE=enforce` + +PR #1405 已补充 queued batch/abort 清理与 request copy-on-write，但 `dev` 中仍保留整条 +`ReplaySpool.parts[]`，也没有处理 spool 失效后的 drain 窗口降级。因此 #1405 降低了部分 +异常路径的保留风险，但没有覆盖本报告复现的主触发链。 + +## 本地复现夹具 + +隔离环境使用独立 PostgreSQL、Redis、mock upstream 和两个 v0.9.2 app 容器。app 容器限制为 +1 GiB，避免实验影响扩散到其他进程。 + +mock upstream 对每个请求发送有效的 `response.output_text.delta` SSE 帧，累计约 7.5 MiB 后 +保持连接打开且不发送终态。客户端确认 mock 收到请求后约 250 ms 主动断开。 + +可重复运行的夹具已纳入仓库： + +```text +tests/load/issue-1408-replay-oom/mock-upstream.cjs +tests/load/issue-1408-replay-oom/drive-disconnect-waves.cjs +tests/load/issue-1408-replay-oom/memory-probe.cjs +tests/load/issue-1408-replay-oom/sample-container.sh +tests/load/issue-1408-replay-oom/run-wave.sh +tests/load/issue-1408-replay-oom/start-mock-container.sh +tests/load/issue-1408-replay-oom/README.md +``` + +本次调查的原始采样与 fatal report 保留在本机： + +```text +/private/tmp/cch1408-wave-on-samples.out +/private/tmp/cch1408-wave-off-samples.out +/private/tmp/cch1408-wave-64k-samples.out +/private/tmp/cch1408-wave-fixed2-samples.out +/private/tmp/cch1408-wave-fixed2-repeat2-samples.out +/private/tmp/cch1408-on-reports/report.20260811.072324.1.0.001.json +``` + +## 证据一：Replay 把断线 drain 从 60 秒延长到 300 秒 + +固定 8 个断线请求时： + +| 场景 | 25 秒 | 60 秒 | 300 秒后 | +| --- | --- | --- | --- | +| Replay off | external 79.62 MiB，ArrayBuffer 75.65 MiB，21 sockets | 8/8 timeout 开始释放 | external 4.73 MiB，ArrayBuffer 0.77 MiB，13 sockets | +| Replay on | external 81.10 MiB，ArrayBuffer 77.13 MiB，21 sockets | 0/8 timeout，继续保持 | 8/8 timeout 后释放 | + +Replay-on 在终态清理后的最终状态为： + +```text +external 4.67 MiB +arrayBuffers 0.70 MiB +TCP sockets 13 +Async tasks 0 +``` + +这说明单波请求最终会释放，但 300 秒窗口允许多个请求波次在释放前持续叠加。 + +## 证据二：40 个活跃 Replay 流复现 V8 heap OOM + +以 10 秒间隔发送 5 波、每波 8 个不同 Replay 请求。所有请求均在客户端断开后保持上游悬挂。 + +| 活跃任务 | heapUsed | external | ArrayBuffer | RSS | +| ---: | ---: | ---: | ---: | ---: | +| 8 | 193.78 MiB | 92.27 MiB | 88.30 MiB | 383.68 MiB | +| 16 | 258.36 MiB | 168.32 MiB | 164.35 MiB | 520.14 MiB | +| 24 | 315.20 MiB | 240.16 MiB | 236.19 MiB | 664.54 MiB | +| 32 | 380.65 MiB | 314.59 MiB | 310.62 MiB | 811.62 MiB | +| 40 | 382.60 MiB | 311.58 MiB | 307.61 MiB | 793.50 MiB | + +随后 Node 进程直接输出： + +```text +FATAL ERROR: Reached heap limit Allocation failed - JavaScript heap out of memory +``` + +容器退出码为 `133`。Node fatal report 记录： + +```text +javascriptHeap.usedMemory 437,947,120 bytes +javascriptHeap.memoryLimit 562,036,736 bytes +javascriptHeap.externalMemory 369,633,015 bytes +javascriptHeap.heapSpaces.old_space.used 408,208,280 bytes +resourceUsage.rss 904,556,544 bytes +resourceUsage.maxRss 929,734,656 bytes +``` + +这与 #1408 的生产现象属于同一种 Node 内存增长并最终退出的故障类别。本地容器有 1 GiB 限制， +所以在约 0.9 GiB RSS 时提前终止；生产容器没有内存限制，允许同一类请求波次继续累积到更高 +RSS。生产环境约 30 GiB 的绝对值不能仅由该缩小夹具外推。 + +## 证据三：64 KiB spool 上限隔离出 `parts[]` heap 主因 + +第三组实验保持 Replay 和 300 秒 drain 不变，只把 `REPLAY_MAX_PAYLOAD_BYTES` 降到 64 KiB。 +每个 spool 在首个大块后立即 `payload_too_large`，清空本地 `parts[]`，但上游流仍按 300 秒 +窗口 drain。 + +相同 8 波、共 64 个活跃断线请求的结果： + +```text +active stream tasks 64 +heapUsed 133.46 MiB +external 586.72 MiB +arrayBuffers 582.75 MiB +RSS 886.67 MiB +container running +``` + +正常 8 MiB spool 在 40 个活跃流时 `heapUsed` 已到 382.60 MiB 并触发 V8 fatal OOM；64 KiB +spool 在 64 个活跃流时 `heapUsed` 仍约 133 MiB。唯一关键变量是 spool 本地正文是否继续保留。 +因此 `ReplaySpool.parts[]` 是本次 V8 heap OOM 的决定性持有对象，300 秒 detached drain 是并发 +驻留时间放大器。 + +## 证据四：Session Response Body 三份复制放大 Redis + +第一版 Node 修复移除 `parts[]` 并缩短失效 spool 的 drain 后，Node 已不再触发 V8 OOM，但隔离 +Redis 在约 300 秒 RDB 保存点仍退出 `137`。检查 Redis key 和配置后确认： + +- `STORE_SESSION_RESPONSE_BODY=true`； +- 本夹具约 7.5 MiB 的 SSE 小于 `STREAM_STATS_MAX_BUFFER_BYTES=10 MiB`，会形成完整统计快照； +- 同一正文写入 legacy/request response、before snapshot、after snapshot 三份 Redis key； +- 64 个请求约产生 `3 x 64 x 7.5 MiB = 1.4 GiB` 的正文值。 + +第一版修复 Redis 的实际峰值为： + +```text +used_memory_human 1.50G +used_memory_peak_human 1.51G +OOMKilled=true +ExitCode=137 +``` + +旧版应用 fatal OOM 发生在 Redis 退出之前，因此 Redis 退出不是 Node fatal 的起因；它是独立的 +伴随放大器。RDB/AOF fork 与写入压力会进一步放大整机内存和 I/O 压力，这与 #1408 中 Redis +`bio_aof` 阻塞、healthcheck 超时的后续现象一致。 + +## 代码持有链 + +当前路径可以简化为： + +```text +upstream Uint8Array + -> ResponseHandler.observeChunk() + -> BoundedStreamTextAccumulator / stream transport buffers + -> ReplaySpool.observe() + -> TextDecoder string + -> pending[] + -> queued writeChain batch + -> parts[] retained until terminal + -> Redis LIST copy + +client disconnect + -> responsePump.startDrain() + -> replay owner selects 300s timeout + -> each new request wave adds another retained response + -> V8 old_space reaches heap limit before oldest wave expires + +stream finalization + -> storeSessionResponse() + -> before response snapshot + -> after response snapshot + -> three Redis values retain the same multi-MiB SSE body until SESSION_TTL + -> RDB/AOF persistence amplifies Redis memory and I/O pressure +``` + +spool 因 payload 超限、Redis 异常或 owner lease 丢失而失效时，会清理自身正文，但 +`ProxyResponseHandler` 已经选择的 300 秒 timer 仍继续运行，所以失效后的 Response、ArrayBuffer、 +socket 和后台任务仍会保留。这解释了 8 MiB 超限样本在 spool 清空后依然保持约 77 MiB +ArrayBuffer 到 300 秒的现象。 + +## 已实现修复 + +修复在 `dev` 提交 `3fe3225c9f6397d22db27193199e2a8fef4a05f7` 的工作树上完成，包含： + +1. 删除 `ReplaySpool.parts[]`，活跃正文只长期保存在 Redis fenced chunks 中。 +2. 完成时从 Redis 回读 chunks，校验数量后重建 durable payload；大 payload 的回读、拼接和 PG + 持久化全局串行，避免多个终态同时形成 heap 峰值。 +3. Redis write-behind backlog 上限固定为 1 MiB。超过时以 + `write_backlog_too_large` fail-open 关闭当前 spool，并立即清空 pending/queued batch。 +4. spool disable、halt 或 abort 通过一次性 `onInactive` 通知 response handler；若客户端已断开， + drain 从 Replay 300 秒降回普通 60 秒，并从实际断线时刻计算剩余时间。 +5. 新增 `SESSION_RESPONSE_BODY_MAX_BYTES`，默认 5 MiB、范围 64 KiB 到 64 MiB。legacy response + 和 before/after snapshot 都按 UTF-8 字节限制；超限时删除同 key 的旧正文，但继续保存 + headers/meta。三份 response body 的物理存储去重由 #1415 跟踪。 +6. 保留 Replay fenced owner、终态计费屏障、live attach、PG durable winner 和冲突处理语义。 + +## 修复负载回归 + +修复镜像保持相同 PostgreSQL、provider、key、mock、Replay 配置和 1 GiB app cgroup，连续运行 +两组完整 64 请求波次，另有一组 8 请求预检，共 136 个断线请求。 + +第一组关键点： + +| 活跃任务 | heapUsed | external | ArrayBuffer | RSS | +| ---: | ---: | ---: | ---: | ---: | +| 40 | 87.97 MiB | 371.74 MiB | 367.77 MiB | 594.50 MiB | +| 48 | 87.86 MiB | 442.92 MiB | 438.95 MiB | 679.63 MiB | + +第二组在复用同一进程和 allocator 状态后，48 个活跃任务时 `heapUsed=92.01 MiB`；整个波次 +最高观测 `heapUsed=155.31 MiB`、`RSS=905.63 MiB`，随后 64 个任务全部清理，`heapUsed` 回到 +约 95 MiB。对比旧版 40 个任务时 `heapUsed=382.60 MiB` 并 fatal OOM，Node heap 持有链已经 +被切断。继续静默等待 GC 后，进程为 `heapUsed=83.38 MiB`、`external=4.64 MiB`、 +`ArrayBuffer=0.67 MiB`、13 sockets，证明第二轮峰值没有形成阶梯式引用累积。 + +累计运行日志： + +```text +Client abort drain window exceeded 136 +write_backlog_too_large 136 +oversized session body skipped 408 +FATAL ERROR / heap out of memory 0 +remaining async tasks 0 +``` + +Redis 在两轮后： + +```text +used_memory_human 2.87M +used_memory_peak_human 5.34M +rdb_saves 3 +rdb_last_bgsave_status ok +rdb_last_cow_size 1138688 +container running, oom=false, exit=0 +``` + +上述负载回归显式使用 1 MiB session body 边界，证明它消除了原先约 1.50 GiB 的三份正文驻留， +并已跨过 `save 300 100` 的 RDB fork 点。当前产品默认值为 5 MiB；1 MiB 到 5 MiB 正文仍可能 +形成三份 Redis value，该放大边界及 5 MiB 负载/RDB 验证由 #1415 跟踪，不属于上述实验已证明的范围。 + +## 测试与证据边界 + +focused 回归共 5 个文件、104 个测试，覆盖： + +- 64 KiB flush、1 MiB backlog 包含边界和超限清理； +- Redis/PG 阻塞、abort/disable/halt 竞态、幂等和 active spool 配额释放； +- Redis chunks 缺失、durable 冲突、并发完成串行化和 UTF-8 截断尾部； +- spool 在断线前/后失效，以及活跃 spool 保持完整 300 秒窗口； +- session body 默认值、64 KiB/64 MiB 配置边界、UTF-8 字节边界、旧值删除； +- 超限 snapshot 只删除 body，headers/meta 继续保留。 + +最终 checkout 已通过 `bun run lint:fix`、`bun run lint`、`bun run typecheck` 和宿主机 +`bun run build`。Biome 仅提示配置 schema URL 为 2.5.6、CLI 为 2.5.7，没有 lint 错误或自动改动。 + +最终全量 `bun run test` 仍只有一个失败。唯一失败是既有 `language-switcher` sessionStorage +console 断言，隔离复跑仍为相同失败，与本次代理、Replay 和 session 存储路径无关。全量并行运行 +另报告一次 `price-list-ui-requirements` worker teardown console RPC rejection；该文件隔离复跑 4/4 +通过，因此记录为测试 harness 并行 teardown 噪声，不计入本次修复通过项。 + +## 调查状态 + +本地机制复现、修复、重复负载和 Redis 持久化边界均已完成。结论是“已证明并修复一条足以复现 +`#1408` 的决定性 Replay/断线流内存失控链，同时消除了 Session Response Body 的 Redis 放大器”； +生产 #1408 的唯一对象级根因仍受限于故障现场没有 heap snapshot。 diff --git a/src/app/v1/_lib/proxy/replay/replay-spool.ts b/src/app/v1/_lib/proxy/replay/replay-spool.ts index 11817432e..4582a06d4 100644 --- a/src/app/v1/_lib/proxy/replay/replay-spool.ts +++ b/src/app/v1/_lib/proxy/replay/replay-spool.ts @@ -27,10 +27,25 @@ import { const FLUSH_INTERVAL_MS = 100; const FLUSH_BYTES_THRESHOLD = 64 * 1024; +const MAX_QUEUED_WRITE_BYTES = 1024 * 1024; const OWNER_HEARTBEAT_INTERVAL_MS = 15_000; const PRE_SPOOL_ABORT_WAIT_MS = 100; let activeSpoolCount = 0; +let durablePersistenceChain: Promise = Promise.resolve(); + +function serializeDurablePersistence(operation: () => Promise): Promise { + const result = durablePersistenceChain.then(operation); + durablePersistenceChain = result.then( + () => undefined, + () => undefined + ); + return result; +} + +export interface ReplaySpoolOptions { + onInactive?: () => void; +} export function getActiveReplaySpoolCount(): number { return activeSpoolCount; @@ -39,10 +54,10 @@ export function getActiveReplaySpoolCount(): number { 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 queuedWriteBytes = 0; private totalBytes = 0; private chunkCount = 0; private disabled = false; @@ -58,7 +73,8 @@ export class ReplaySpool { private readonly ownerToken: string, private readonly statusCode: number, private readonly headers: Record, - private readonly delivery: ReplayDelivery = "stream" + private readonly delivery: ReplayDelivery = "stream", + private readonly options: ReplaySpoolOptions = {} ) { activeSpoolCount++; this.startOwnerHeartbeat(); @@ -79,11 +95,10 @@ export class ReplaySpool { this.disable("payload_too_large"); return; } + this.pendingBytes += chunk.byteLength; const text = this.decoder.decode(chunk, { stream: true }); if (text.length === 0) return; this.pending.push(text); - this.parts.push(text); - this.pendingBytes += chunk.byteLength; if (this.pendingBytes >= FLUSH_BYTES_THRESHOLD) { this.scheduleFlush(0); @@ -119,9 +134,10 @@ export class ReplaySpool { private enqueueFlush(): void { const batch = this.pending; if (batch.length === 0) return; + const batchBytes = this.pendingBytes; this.pending = []; this.pendingBytes = 0; - this.queuedBatches.add(batch); + if (!this.reserveQueuedBatch(batch, batchBytes)) return; // 续接体自带 try/catch：链永不 rejected；每个 await 之后复查 disabled， // 防止与 disable/halt 竞态时在清理之后又写回 owning meta this.writeChain = this.writeChain.then(async () => { @@ -154,10 +170,22 @@ export class ReplaySpool { this.disable("flush_error"); } finally { this.queuedBatches.delete(batch); + this.queuedWriteBytes = Math.max(0, this.queuedWriteBytes - batchBytes); } }); } + private reserveQueuedBatch(batch: string[], batchBytes: number): boolean { + if (this.queuedWriteBytes + batchBytes > MAX_QUEUED_WRITE_BYTES) { + batch.length = 0; + this.disable("write_backlog_too_large"); + return false; + } + this.queuedWriteBytes += batchBytes; + this.queuedBatches.add(batch); + return true; + } + private buildMeta(status: ReplayMeta["status"], extra?: Partial): ReplayMeta { return { status, @@ -236,12 +264,15 @@ export class ReplaySpool { const tail = this.decoder.decode(); if (tail.length > 0) { this.pending.push(tail); - this.parts.push(tail); } const batch = this.pending; + const batchBytes = this.pendingBytes; this.pending = []; this.pendingBytes = 0; - this.queuedBatches.add(batch); + if (!this.reserveQueuedBatch(batch, batchBytes)) { + await this.writeChain; + return; + } this.writeChain = this.writeChain.then(async () => { let pgPersisted = false; @@ -263,21 +294,27 @@ 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, - verifier: this.identity.verifier, - scopeTag: this.identity.scopeTag, - keyId: this.identity.keyId, - userId: this.identity.userId, - format: this.identity.format, - model: this.identity.model, - statusCode: this.statusCode, - headers: this.headers, - payload, - byteSize: this.totalBytes, - sourceMessageRequestId: messageRequestId, + // Redis 是活跃正文的唯一长期副本。大 payload 的读取、拼接和 PG 写入串行执行， + // 避免多个流同秒终态时在 V8 heap 中并发重建完整响应。 + const persistResult = await serializeDurablePersistence(async () => { + const chunks = await this.store.readChunks(this.identity.replayId, 0); + if (!chunks || chunks.length !== this.chunkCount) { + throw new Error("replay chunks unavailable before durable persistence"); + } + return this.store.persistCompleted({ + replayId: this.identity.replayId, + verifier: this.identity.verifier, + scopeTag: this.identity.scopeTag, + keyId: this.identity.keyId, + userId: this.identity.userId, + format: this.identity.format, + model: this.identity.model, + statusCode: this.statusCode, + headers: this.headers, + payload: chunks.join(""), + byteSize: this.totalBytes, + sourceMessageRequestId: messageRequestId, + }); }); pgPersisted = true; const completed = await this.store.completeOwned( @@ -322,7 +359,7 @@ export class ReplaySpool { .catch(() => false); } finally { this.queuedBatches.delete(batch); - this.clearPayload(); + this.queuedWriteBytes = Math.max(0, this.queuedWriteBytes - batchBytes); this.release(); } }); @@ -338,10 +375,10 @@ export class ReplaySpool { if (this.terminal) return; this.terminal = true; this.aborting = true; + this.notifyInactive(); this.clearTimer(); this.pending = []; this.pendingBytes = 0; - this.clearPayload(); this.clearQueuedBatches(); this.abortPromise = this.writeChain.then(async () => { try { @@ -375,9 +412,9 @@ export class ReplaySpool { private teardown(reason: string, deleteEntry: boolean): void { if (this.disabled) return; this.disabled = true; + this.notifyInactive(); this.clearTimer(); this.pending = []; - this.parts.length = 0; this.pendingBytes = 0; this.clearQueuedBatches(); // 清理顺着 writeChain 串行：与 in-flight append 竞态时绝不出现「删除后又写回」 @@ -435,19 +472,24 @@ export class ReplaySpool { 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(); + this.queuedWriteBytes = 0; + } + + private inactiveNotified = false; + + private notifyInactive(): void { + if (this.inactiveNotified) return; + this.inactiveNotified = true; + try { + this.options.onInactive?.(); + } catch (error) { + logger.debug("[ReplaySpool] inactive callback failed", { + error: error instanceof Error ? error.message : String(error), + }); + } } } @@ -515,7 +557,8 @@ export function releaseReplayOwnership(session: ProxySession): void { export function createReplaySpoolIfOwner( session: ProxySession, response: Response, - delivery: ReplayDelivery = "stream" + delivery: ReplayDelivery = "stream", + options: ReplaySpoolOptions = {} ): ReplaySpool | null { const replayState = session.replayState; if (replayState?.role !== "owner") return null; @@ -550,7 +593,8 @@ export function createReplaySpoolIfOwner( replayState.ownerToken, response.status, headers, - delivery + delivery, + options ); spool.bootstrap(); return spool; diff --git a/src/app/v1/_lib/proxy/response-handler.ts b/src/app/v1/_lib/proxy/response-handler.ts index 81e78c2cf..de8ad5d62 100644 --- a/src/app/v1/_lib/proxy/response-handler.ts +++ b/src/app/v1/_lib/proxy/response-handler.ts @@ -4147,6 +4147,7 @@ export class ProxyResponseHandler { // 提升 idleTimeoutId 到外部作用域，以便客户端断开时能清除 let idleTimeoutId: NodeJS.Timeout | null = null; let clientAbortDrainTimeoutId: NodeJS.Timeout | null = null; + let clientAbortDrainStartedAt: number | null = null; const streamTextAccumulator = new BoundedStreamTextAccumulator(); let lastStreamTextSnapshot: BoundedStreamTextSnapshot | null = null; const getCollectedChunkCount = () => @@ -4157,6 +4158,45 @@ export class ProxyResponseHandler { clientAbortDrainTimeoutId = null; } }; + const expireClientAbortDrain = () => { + clientAbortDrainTimeoutId = null; + clientAbortDrainStartedAt = null; + logger.info("ResponseHandler: Client abort drain window exceeded", { + taskId, + providerId: provider.id, + messageId: messageContext.id, + clientAbortDrainTimeoutMs, + }); + + try { + const sessionWithController = session as typeof session & { + responseController?: AbortController; + }; + sessionWithController.responseController?.abort(new Error("client_abort_drain_timeout")); + } catch (e) { + logger.warn("ResponseHandler: Failed to abort upstream after client drain timeout", { + taskId, + providerId: provider.id, + error: e, + }); + } + + const drainTimeoutError = new Error("client_abort_drain_timeout"); + abortController.abort(drainTimeoutError); + responsePump?.cancelSource(drainTimeoutError); + }; + const scheduleClientAbortDrainTimeout = (delayMs: number) => { + clearClientAbortDrainTimer(); + clientAbortDrainTimeoutId = setTimeout(expireClientAbortDrain, Math.max(0, delayMs)); + clientAbortDrainTimeoutId.unref?.(); + }; + const capInactiveReplayDrainWindow = () => { + if (clientAbortDrainTimeoutMs <= CLIENT_ABORT_DRAIN_MAX_MS) return; + clientAbortDrainTimeoutMs = CLIENT_ABORT_DRAIN_MAX_MS; + if (clientAbortDrainStartedAt === null) return; + const elapsedMs = Date.now() - clientAbortDrainStartedAt; + scheduleClientAbortDrainTimeout(CLIENT_ABORT_DRAIN_MAX_MS - elapsedMs); + }; const clearIdleTimer = () => { if (idleTimeoutId) { clearTimeout(idleTimeoutId); @@ -4237,31 +4277,8 @@ export class ProxyResponseHandler { if (!idleTimeoutId) { startIdleTimer(); } - clientAbortDrainTimeoutId = setTimeout(() => { - logger.info("ResponseHandler: Client abort drain window exceeded", { - taskId, - providerId: provider.id, - messageId: messageContext.id, - clientAbortDrainTimeoutMs, - }); - - try { - const sessionWithController = session as typeof session & { - responseController?: AbortController; - }; - sessionWithController.responseController?.abort(new Error("client_abort_drain_timeout")); - } catch (e) { - logger.warn("ResponseHandler: Failed to abort upstream after client drain timeout", { - taskId, - providerId: provider.id, - error: e, - }); - } - - const drainTimeoutError = new Error("client_abort_drain_timeout"); - abortController.abort(drainTimeoutError); - responsePump?.cancelSource(drainTimeoutError); - }, clientAbortDrainTimeoutMs); + clientAbortDrainStartedAt = Date.now(); + scheduleClientAbortDrainTimeout(clientAbortDrainTimeoutMs); }; // 统计/结算只保留有界的“头 + 尾”文本快照，避免长流式响应把进程堆撑满。 @@ -4823,7 +4840,9 @@ export class ProxyResponseHandler { // F2 owner spool：guard 阶段已抢到 owner 租约的请求，把客户端可见字节 // write-behind 喂入 Redis 热层，供并发/断线的相同请求 attach 跟尾。 - const replaySpool = createReplaySpoolIfOwner(session, response); + const replaySpool = createReplaySpoolIfOwner(session, response, "stream", { + onInactive: capInactiveReplayDrainWindow, + }); if (replaySpool) { try { clientAbortDrainTimeoutMs = getEnvConfig().REPLAY_MAX_DETACHED_MS; diff --git a/src/lib/config/env.schema.ts b/src/lib/config/env.schema.ts index 7193f9549..2cba1650e 100644 --- a/src/lib/config/env.schema.ts +++ b/src/lib/config/env.schema.ts @@ -157,6 +157,13 @@ export const EnvSchema = z.object({ // - 该开关只影响“写入 Redis 的响应体内容”，不影响内部统计逻辑读取响应体（例如 tokens/费用统计、SSE 结束后的假 200 检测）。 // - message 内容是否脱敏仍由 STORE_SESSION_MESSAGES 控制。 STORE_SESSION_RESPONSE_BODY: z.string().default("true").transform(booleanTransform), + // 单份会话响应正文写入 Redis 的字节上限；旧 response 与 before/after snapshot 都受此边界约束。 + SESSION_RESPONSE_BODY_MAX_BYTES: z.coerce + .number() + .int() + .min(64 * 1024) + .max(64 * 1024 * 1024) + .default(5 * 1024 * 1024), DEBUG_MODE: z.string().default("false").transform(booleanTransform), LOG_LEVEL: z.enum(["fatal", "error", "warn", "info", "debug", "trace"]).default("info"), TZ: z.string().default("Asia/Shanghai"), diff --git a/src/lib/session-manager-detail-snapshots.test.ts b/src/lib/session-manager-detail-snapshots.test.ts index a35ae9cd1..323e847f7 100644 --- a/src/lib/session-manager-detail-snapshots.test.ts +++ b/src/lib/session-manager-detail-snapshots.test.ts @@ -45,6 +45,7 @@ const redisMock = { return Promise.resolve("OK"); }), get: vi.fn((key: string) => Promise.resolve(redisStore.get(key) ?? null)), + del: vi.fn((key: string) => Promise.resolve(redisStore.delete(key) ? 1 : 0)), set: vi.fn().mockResolvedValue("OK"), expire: vi.fn().mockResolvedValue(1), incr: vi.fn().mockResolvedValue(1), @@ -58,11 +59,13 @@ vi.mock("@/lib/redis", () => ({ let mockStoreMessages = false; let mockStoreSessionResponseBody = true; +let mockSessionResponseBodyMaxBytes = 1024 * 1024; vi.mock("@/lib/config/env.schema", () => ({ getEnvConfig: () => ({ STORE_SESSION_MESSAGES: mockStoreMessages, STORE_SESSION_RESPONSE_BODY: mockStoreSessionResponseBody, + SESSION_RESPONSE_BODY_MAX_BYTES: mockSessionResponseBodyMaxBytes, SESSION_TTL: 300, }), })); @@ -76,6 +79,7 @@ describe("SessionManager detail snapshots", () => { redisMock.status = "ready"; mockStoreMessages = false; mockStoreSessionResponseBody = true; + mockSessionResponseBodyMaxBytes = 1024 * 1024; }); it("atomically persists the request sequence while expiring its owner marker", async () => { @@ -392,6 +396,65 @@ describe("SessionManager detail snapshots", () => { }); }); + it("skips only an oversized response body while preserving snapshot headers and meta", async () => { + mockSessionResponseBodyMaxBytes = 4; + + await SessionManager.storeSessionResponsePhaseSnapshot( + "sess_oversized_response", + "after", + { + body: "12345", + headers: new Headers({ "content-type": "text/event-stream" }), + meta: { upstreamUrl: null, statusCode: 200 }, + }, + 1 + ); + + expect( + await SessionManager.getSessionResponsePhaseSnapshot("sess_oversized_response", "after", 1) + ).toEqual({ + body: null, + headers: { "content-type": "text/event-stream" }, + meta: { upstreamUrl: null, statusCode: 200 }, + }); + expect(loggerMock.warn).toHaveBeenCalledWith( + "SessionManager: Skipped oversized session response body", + { context: "snapshot:after", byteSize: 5, maxBytes: 4 } + ); + }); + + it("removes a previous snapshot body when its replacement exceeds the limit", async () => { + mockSessionResponseBodyMaxBytes = 4; + + await SessionManager.storeSessionResponsePhaseSnapshot( + "sess_replaced_response", + "after", + { body: "1234" }, + 1 + ); + await SessionManager.storeSessionResponsePhaseSnapshot( + "sess_replaced_response", + "after", + { + body: "12345", + headers: new Headers({ "content-type": "text/event-stream" }), + meta: { upstreamUrl: null, statusCode: 200 }, + }, + 1 + ); + + expect( + await SessionManager.getSessionResponsePhaseSnapshot("sess_replaced_response", "after", 1) + ).toEqual({ + body: null, + headers: { "content-type": "text/event-stream" }, + meta: { upstreamUrl: null, statusCode: 200 }, + }); + expect(redisMock.del).toHaveBeenCalledWith( + "session:sess_replaced_response:req:1:snapshot:response:after:body" + ); + }); + it("treats empty headers as missing instead of an empty record", async () => { await SessionManager.storeSessionRequestPhaseSnapshot( "sess_empty_headers", diff --git a/src/lib/session-manager.ts b/src/lib/session-manager.ts index 4128f8e90..6b5983420 100644 --- a/src/lib/session-manager.ts +++ b/src/lib/session-manager.ts @@ -1,5 +1,6 @@ import "server-only"; +import { Buffer } from "node:buffer"; import crypto from "node:crypto"; import { extractCodexSessionId } from "@/app/v1/_lib/codex/session-extractor"; import { sanitizeHeaders, sanitizeUrl } from "@/app/v1/_lib/proxy/errors"; @@ -58,6 +59,24 @@ import { SessionTracker } from "./session-tracker"; const RESERVED_INTERNAL_HEADER_SET = new Set( RESERVED_INTERNAL_HEADERS.map((header) => header.toLowerCase()) ); +const DEFAULT_SESSION_RESPONSE_BODY_MAX_BYTES = 5 * 1024 * 1024; + +function canStoreSessionResponseBody(value: string, context: string): boolean { + const configuredMaxBytes = getEnvConfig().SESSION_RESPONSE_BODY_MAX_BYTES; + const maxBytes = + Number.isSafeInteger(configuredMaxBytes) && configuredMaxBytes > 0 + ? configuredMaxBytes + : DEFAULT_SESSION_RESPONSE_BODY_MAX_BYTES; + const byteSize = Buffer.byteLength(value, "utf8"); + if (byteSize <= maxBytes) return true; + + logger.warn("SessionManager: Skipped oversized session response body", { + context, + byteSize, + maxBytes, + }); + return false; +} function isReservedInternalHeader(name: string): boolean { const lowerName = name.toLowerCase(); @@ -2036,7 +2055,7 @@ export class SessionManager { * 存储 session 响应体（临时存储，5分钟过期） * * 存储行为受 STORE_SESSION_RESPONSE_BODY 控制： - * - true (默认)：存储响应体到 Redis 临时缓存 + * - true (默认)：在 SESSION_RESPONSE_BODY_MAX_BYTES 上限内存储响应体到 Redis 临时缓存 * - false：不存储（注意：不影响本次请求处理与统计，仅影响后续查看 response body） * * 存储策略（脱敏/原样）受 STORE_SESSION_MESSAGES 控制： @@ -2061,6 +2080,17 @@ export class SessionManager { if (redis?.status !== "ready") return; try { + // 新格式：session:{sessionId}:req:{sequence}:response（独立存储每个请求） + // 旧格式：session:{sessionId}:response（向后兼容） + const sequence = normalizeRequestSequence(requestSequence); + const key = sequence + ? `session:${sessionId}:req:${sequence}:response` + : `session:${sessionId}:response`; + if (typeof response === "string" && !canStoreSessionResponseBody(response, "response")) { + await redis.del(key); + return; + } + let responseString: string; if (SessionManager.STORE_MESSAGES) { @@ -2082,12 +2112,11 @@ export class SessionManager { } } - // 新格式：session:{sessionId}:req:{sequence}:response（独立存储每个请求） - // 旧格式：session:{sessionId}:response（向后兼容） - const sequence = normalizeRequestSequence(requestSequence); - const key = sequence - ? `session:${sessionId}:req:${sequence}:response` - : `session:${sessionId}:response`; + if (!canStoreSessionResponseBody(responseString, "response")) { + await redis.del(key); + return; + } + if (sequence) { await SessionManager.refreshSessionRequestOwner(redis, sessionId, sequence, keyId); } @@ -2670,8 +2699,24 @@ export class SessionManager { // 与旧平铺 response 字段保持同一隐私/存储契约：关闭时跳过任何 response body phase 落盘。 } else { let bodyToStore = snapshot.body ?? null; + let bodyExceededLimit = false; + const bodyKey = buildSessionDetailSnapshotKey( + sessionId, + sequence, + "response", + phase, + "body" + ); + + if ( + typeof bodyToStore === "string" && + !canStoreSessionResponseBody(bodyToStore, `snapshot:${phase}`) + ) { + bodyToStore = null; + bodyExceededLimit = true; + } - if (!SessionManager.STORE_MESSAGES) { + if (bodyToStore !== null && !SessionManager.STORE_MESSAGES) { if (typeof bodyToStore === "string") { try { bodyToStore = JSON.stringify( @@ -2687,14 +2732,18 @@ export class SessionManager { bodyToStore = JSON.stringify(bodyToStore); } - if (bodyToStore !== null) { - writes.push( - redis.setex( - buildSessionDetailSnapshotKey(sessionId, sequence, "response", phase, "body"), - SessionManager.SESSION_TTL, - bodyToStore - ) - ); + if ( + bodyToStore !== null && + canStoreSessionResponseBody(bodyToStore, `snapshot:${phase}`) + ) { + writes.push(redis.setex(bodyKey, SessionManager.SESSION_TTL, bodyToStore)); + } else if (bodyToStore !== null) { + bodyExceededLimit = true; + } + + if (bodyExceededLimit) { + // 同一 request/phase 可能被重写；超限时删除旧正文，避免读取到上一版小响应。 + writes.push(redis.del(bodyKey)); } } } diff --git a/tests/load/issue-1408-replay-oom/README.md b/tests/load/issue-1408-replay-oom/README.md new file mode 100644 index 000000000..1e794fb45 --- /dev/null +++ b/tests/load/issue-1408-replay-oom/README.md @@ -0,0 +1,96 @@ +# Issue 1408 Replay OOM Load Fixture + +This fixture reproduces the client-disconnect workload used to isolate issue #1408. It keeps the +mock upstream response open after sending a bounded SSE payload, disconnects each client only after +the mock confirms receipt, and samples Node, socket, task, timeout, and Redis state. + +The fixture is intentionally separate from the regular Vitest suite because it needs a configured +CC Hub instance, PostgreSQL, Redis, a Provider pointing at the mock, an API key, and Docker metrics. + +## Files + +- `mock-upstream.cjs`: sends `response.output_text.delta` frames, then remains open without a + terminal event. `CCH_MOCK_MIB` controls the payload size from 0.0625 to 64 MiB per request. +- `drive-disconnect-waves.cjs`: sends distinct `/v1/responses` Replay requests in waves, waits for + the mock receipt count, then disconnects the clients after `CCH_ABORT_DELAY_MS`. +- `memory-probe.cjs`: preload hook that emits RSS, V8 heap, external, ArrayBuffer, and active resource + counts as JSON. +- `sample-container.sh`: samples app logs and optional Redis state into a result file. +- `run-wave.sh`: runs the driver and sampler together. +- `start-mock-container.sh`: starts the mock on an existing Docker network without replacing an + existing container. + +## Prerequisites + +1. Build or select the CC Hub image/revision under test. +2. Start PostgreSQL and create a test database containing a Provider and API key for the fixture. +3. Create a Docker network shared by the app, Redis, and mock. +4. Configure the Provider base URL as `http://MOCK_CONTAINER:3001` and route model `gpt-5.6` to it. +5. Start the app with the probe preloaded. For a container, mount `memory-probe.cjs` read-only and + set `NODE_OPTIONS=--require=/fixture/memory-probe.cjs`. + +Do not point this fixture at a production Provider. The mock deliberately leaves every upstream +response open until the app or fixture closes it. + +## Start The Mock + +```bash +tests/load/issue-1408-replay-oom/start-mock-container.sh \ + cch1408-mock cch1408-network 31409 8 +``` + +The command prints both URLs: + +```text +stats=http://127.0.0.1:31409/stats +provider=http://cch1408-mock:3001 +``` + +## Run A Wave Test + +Store the test API key in a protected file outside the repository, then run: + +```bash +export CCH_API_KEY_FILE=/path/to/test-api-key + +tests/load/issue-1408-replay-oom/run-wave.sh \ + http://127.0.0.1:31415 \ + http://127.0.0.1:31409/stats \ + issue1408-fixed \ + cch1408-app \ + issue1408-fixed.samples.txt \ + cch1408-redis +``` + +Defaults match the investigation workload: + +```text +CCH_WAVES=8 +CCH_REQUESTS_PER_WAVE=8 +CCH_WAVE_INTERVAL_MS=10000 +CCH_ABORT_DELAY_MS=250 +CCH_SAMPLES=18 +CCH_SAMPLE_INTERVAL_SECONDS=10 +CCH_MOCK_MIB=8 +``` + +Use a unique scenario prefix for every run. Replay identity includes the scenario, wave, and request +index, so a unique prefix prevents a previous durable Replay entry from turning the workload into a +cache hit. Scenario prefixes accept 1 to 64 ASCII letters, digits, underscores, and hyphens. The +driver destroys all requests if mock receipt confirmation fails, so a failed run does not leave its +own upstream streams active. + +## Acceptance Signals + +For the fixed revision under the default workload: + +- Node heap remains bounded while external and ArrayBuffer memory follow active stream count. +- `write_backlog_too_large` makes an inactive Replay spool fall back to the 60-second drain window. +- Every disconnected task reaches `Client abort drain window exceeded` and the active task count + returns to zero. +- After a quiet GC period, external and ArrayBuffer memory return near the pre-wave baseline. +- Redis remains running across its configured RDB save window and does not retain three copies of + each multi-MiB response body. + +The historical measurements and the exact evidence boundary are documented in +`docs/troubleshooting/issue-1408-replay-oom.md`. diff --git a/tests/load/issue-1408-replay-oom/drive-disconnect-waves.cjs b/tests/load/issue-1408-replay-oom/drive-disconnect-waves.cjs new file mode 100644 index 000000000..8fff27546 --- /dev/null +++ b/tests/load/issue-1408-replay-oom/drive-disconnect-waves.cjs @@ -0,0 +1,216 @@ +"use strict"; + +const fs = require("node:fs"); +const http = require("node:http"); +const https = require("node:https"); + +const [appUrl, mockStatsUrl, scenarioPrefix, wavesArg, perWaveArg, intervalArg] = + process.argv.slice(2); + +if (!appUrl || !mockStatsUrl || !scenarioPrefix) { + throw new Error( + "usage: drive-disconnect-waves.cjs APP_URL MOCK_STATS_URL SCENARIO_PREFIX " + + "[WAVES] [REQUESTS_PER_WAVE] [INTERVAL_MS]" + ); +} + +const parsedAppUrl = parseHttpUrl(appUrl, "APP_URL"); +const parsedMockStatsUrl = parseHttpUrl(mockStatsUrl, "MOCK_STATS_URL"); +const normalizedScenarioPrefix = parseScenarioPrefix(scenarioPrefix); +const waves = parseBoundedInteger(wavesArg || "8", "WAVES", 1, 255); +const perWave = parseBoundedInteger(perWaveArg || "8", "REQUESTS_PER_WAVE", 1, 255); +const intervalMs = parseBoundedInteger(intervalArg || "10000", "INTERVAL_MS", 0, 3600000); +const abortDelayMs = parseBoundedInteger( + process.env.CCH_ABORT_DELAY_MS || "250", + "CCH_ABORT_DELAY_MS", + 0, + 60000 +); +const mockReceiptTimeoutMs = parseBoundedInteger( + process.env.CCH_MOCK_RECEIPT_TIMEOUT_MS || "30000", + "CCH_MOCK_RECEIPT_TIMEOUT_MS", + 1, + 3600000 +); +const model = (process.env.CCH_REQUEST_MODEL || "gpt-5.6").trim(); +if (!model || model.length > 256) { + throw new Error("CCH_REQUEST_MODEL must contain between 1 and 256 characters"); +} +const key = readApiKey(); + +function parseHttpUrl(raw, name) { + const url = new URL(raw); + if (url.protocol !== "http:" && url.protocol !== "https:") { + throw new Error(`${name} must use http or https`); + } + return url; +} + +function parseScenarioPrefix(raw) { + if (!/^[a-z0-9_-]{1,64}$/i.test(raw)) { + throw new Error("SCENARIO_PREFIX must match [a-z0-9_-] and contain 1 to 64 characters"); + } + return raw; +} + +function parseBoundedInteger(raw, name, minimum, maximum) { + const value = Number(raw); + if (!Number.isInteger(value) || value < minimum || value > maximum) { + throw new Error(`${name} must be an integer between ${minimum} and ${maximum}`); + } + return value; +} + +function readApiKey() { + const direct = process.env.CCH_API_KEY?.trim(); + if (direct) return direct; + const keyFile = process.env.CCH_API_KEY_FILE; + if (keyFile) { + const value = fs.readFileSync(keyFile, "utf8").trim(); + if (value) return value; + } + throw new Error("set CCH_API_KEY or CCH_API_KEY_FILE before running the fixture"); +} + +function transportFor(url) { + return url.protocol === "https:" ? https : http; +} + +function sleep(delayMs) { + return new Promise((resolve) => setTimeout(resolve, delayMs)); +} + +function getJson(rawUrl) { + return new Promise((resolve, reject) => { + const url = rawUrl instanceof URL ? rawUrl : new URL(rawUrl); + const request = transportFor(url).get(url, (response) => { + const chunks = []; + response.once("aborted", () => reject(new Error(`GET ${url} response aborted`))); + response.once("error", reject); + response.on("data", (chunk) => chunks.push(chunk)); + response.on("end", () => { + if ((response.statusCode || 500) >= 400) { + reject(new Error(`GET ${url} returned ${response.statusCode}`)); + return; + } + try { + resolve(JSON.parse(Buffer.concat(chunks).toString("utf8"))); + } catch (error) { + reject(error); + } + }); + }); + request.setTimeout(mockReceiptTimeoutMs, () => request.destroy(new Error("stats timeout"))); + request.on("error", reject); + }); +} + +async function waitForMock(scenario, target) { + const deadline = Date.now() + mockReceiptTimeoutMs; + while (Date.now() < deadline) { + const stats = await getJson(mockStatsUrl); + if ((stats.counts?.[scenario] || 0) >= target) return stats.counts[scenario]; + await sleep(50); + } + throw new Error(`mock receipt timeout for ${scenario}: target=${target}`); +} + +function hashScenario(value) { + return [...value].reduce((hash, char) => (hash * 33 + char.charCodeAt(0)) & 0xff, 0); +} + +function startRequest(scenario, scenarioHash, wave, index) { + const body = JSON.stringify({ + model, + input: [ + { + role: "user", + content: [ + { + type: "input_text", + text: `CCH_SCENARIO_${scenario} wave-${wave} request-${index}`, + }, + ], + }, + ], + stream: true, + prompt_cache_key: `cch1408-${scenario}-${wave}-${index}`, + }); + const url = new URL("/v1/responses", parsedAppUrl); + const suffix = `${scenarioHash.toString(16).padStart(2, "0")}${(wave + 1) + .toString(16) + .padStart(2, "0")}${index.toString(16).padStart(2, "0")}000000`; + const handle = { request: null, response: null }; + const request = transportFor(url).request( + url, + { + method: "POST", + headers: { + authorization: `Bearer ${key}`, + "content-type": "application/json", + "content-length": Buffer.byteLength(body), + session_id: `019c1408-0000-7000-8000-${suffix}`, + }, + }, + (response) => { + handle.response = response; + response.on("data", () => {}); + response.on("error", () => {}); + } + ); + handle.request = request; + request.on("error", () => {}); + request.end(body); + return handle; +} + +function abortRequests(handles) { + for (const handle of handles) { + handle.response?.destroy(); + handle.request?.destroy(); + } +} + +async function main() { + const scenarioHash = hashScenario(normalizedScenarioPrefix); + for (let wave = 0; wave < waves; wave += 1) { + const scenario = `${normalizedScenarioPrefix}-${wave}`; + const before = (await getJson(parsedMockStatsUrl)).counts?.[scenario] || 0; + const handles = []; + for (let index = 0; index < perWave; index += 1) { + handles.push(startRequest(scenario, scenarioHash, wave, index)); + } + + let received; + let confirmedAt; + let abortedAt; + try { + received = await waitForMock(scenario, before + perWave); + confirmedAt = Date.now(); + await sleep(abortDelayMs); + abortedAt = Date.now(); + } finally { + abortRequests(handles); + } + + process.stdout.write( + `${JSON.stringify({ + wave, + scenario, + perWave, + mockBefore: before, + mockReceived: received, + confirmedAt, + abortedAt, + abortDelayMs: abortedAt - confirmedAt, + })}\n` + ); + + if (wave + 1 < waves) await sleep(intervalMs); + } +} + +main().catch((error) => { + process.stderr.write(`${error instanceof Error ? error.stack : String(error)}\n`); + process.exitCode = 1; +}); diff --git a/tests/load/issue-1408-replay-oom/memory-probe.cjs b/tests/load/issue-1408-replay-oom/memory-probe.cjs new file mode 100644 index 000000000..4a6fa680a --- /dev/null +++ b/tests/load/issue-1408-replay-oom/memory-probe.cjs @@ -0,0 +1,41 @@ +"use strict"; + +const v8 = require("node:v8"); + +const rawInterval = process.env.CCH_MEMORY_PROBE_INTERVAL_MS || "1000"; +const intervalMs = Number(rawInterval); +if (!Number.isInteger(intervalMs) || intervalMs < 10 || intervalMs > 60000) { + throw new Error("CCH_MEMORY_PROBE_INTERVAL_MS must be an integer between 10 and 60000"); +} + +function toMiB(value) { + return Number((value / 1048576).toFixed(2)); +} + +function sample() { + const memory = process.memoryUsage(); + const resources = + typeof process.getActiveResourcesInfo === "function" ? process.getActiveResourcesInfo() : []; + const resourceCounts = {}; + for (const name of resources) { + resourceCounts[name] = (resourceCounts[name] || 0) + 1; + } + + process.stdout.write( + `${JSON.stringify({ + cchMemoryProbe: true, + ts: Date.now(), + rssMiB: toMiB(memory.rss), + heapUsedMiB: toMiB(memory.heapUsed), + heapTotalMiB: toMiB(memory.heapTotal), + externalMiB: toMiB(memory.external), + arrayBuffersMiB: toMiB(memory.arrayBuffers), + mallocedMiB: toMiB(v8.getHeapStatistics().malloced_memory), + resources: resourceCounts, + })}\n` + ); +} + +const timer = setInterval(sample, intervalMs); +timer.unref(); +sample(); diff --git a/tests/load/issue-1408-replay-oom/mock-upstream.cjs b/tests/load/issue-1408-replay-oom/mock-upstream.cjs new file mode 100644 index 000000000..d3cbd6598 --- /dev/null +++ b/tests/load/issue-1408-replay-oom/mock-upstream.cjs @@ -0,0 +1,158 @@ +"use strict"; + +const http = require("node:http"); + +const host = process.env.CCH_MOCK_HOST || "0.0.0.0"; +const port = parseBoundedInteger(process.env.CCH_MOCK_PORT || "3001", "CCH_MOCK_PORT", 0, 65535); +const totalMiB = parseBoundedNumber(process.env.CCH_MOCK_MIB || "8", "CCH_MOCK_MIB", 0.0625, 64); +const maxRequestBytes = parseBoundedInteger( + process.env.CCH_MOCK_MAX_REQUEST_BYTES || String(1024 * 1024), + "CCH_MOCK_MAX_REQUEST_BYTES", + 1, + 16 * 1024 * 1024 +); +const chunkText = "x".repeat(64 * 1024); +const framesPerMiB = 16; +const counts = new Map(); + +function parseBoundedNumber(raw, name, minimum, maximum) { + const value = Number(raw); + if (!Number.isFinite(value) || value < minimum || value > maximum) { + throw new Error(`${name} must be a number between ${minimum} and ${maximum}`); + } + return value; +} + +function parseBoundedInteger(raw, name, minimum, maximum) { + const value = Number(raw); + if (!Number.isInteger(value) || value < minimum || value > maximum) { + throw new Error(`${name} must be an integer between ${minimum} and ${maximum}`); + } + return value; +} + +function readBody(req) { + return new Promise((resolve, reject) => { + const chunks = []; + let receivedBytes = 0; + let tooLarge = false; + req.on("data", (chunk) => { + if (tooLarge) return; + receivedBytes += chunk.byteLength; + if (receivedBytes > maxRequestBytes) { + tooLarge = true; + chunks.length = 0; + reject(new Error("request body too large")); + return; + } + chunks.push(chunk); + }); + req.on("end", () => { + if (!tooLarge) resolve(Buffer.concat(chunks).toString("utf8")); + }); + req.on("error", (error) => { + if (!tooLarge) reject(error); + }); + }); +} + +function scenarioFrom(raw) { + const match = raw.match(/CCH_SCENARIO_([a-z0-9_-]+)/i); + return match ? match[1] : "unknown"; +} + +function writeJson(res, statusCode, value) { + res.writeHead(statusCode, { "content-type": "application/json" }); + res.end(JSON.stringify(value)); +} + +const server = http.createServer(async (req, res) => { + if (req.method === "GET" && (req.url === "/health" || req.url === "/stats")) { + writeJson(res, 200, { + counts: Object.fromEntries(counts), + totalMiB, + }); + return; + } + + if (req.method === "POST" && req.url === "/reset") { + counts.clear(); + writeJson(res, 200, { reset: true }); + return; + } + + if (req.method !== "POST" || req.url !== "/v1/responses") { + writeJson(res, 404, { error: "not found" }); + return; + } + + let raw; + try { + raw = await readBody(req); + } catch (error) { + if (!res.headersSent && !res.destroyed) { + writeJson(res, 413, { error: error instanceof Error ? error.message : String(error) }); + } + return; + } + + const scenario = scenarioFrom(raw); + counts.set(scenario, (counts.get(scenario) || 0) + 1); + + process.stdout.write( + `${JSON.stringify({ + event: "request", + path: req.url, + requestBytes: Buffer.byteLength(raw), + scenario, + ordinal: counts.get(scenario), + totalMiB, + })}\n` + ); + + res.on("error", () => {}); + res.writeHead(200, { + "content-type": "text/event-stream", + "cache-control": "no-cache", + connection: "keep-alive", + }); + + const totalFrames = Math.max(1, Math.ceil(totalMiB * framesPerMiB)); + let sent = 0; + const writeNext = () => { + if (res.destroyed || sent >= totalFrames) return; + sent += 1; + const event = `data: ${JSON.stringify({ + type: "response.output_text.delta", + delta: chunkText, + })}\n\n`; + if (!res.write(event)) { + res.once("drain", writeNext); + } else { + setImmediate(writeNext); + } + }; + writeNext(); +}); + +const sockets = new Set(); +server.on("connection", (socket) => { + sockets.add(socket); + socket.on("close", () => sockets.delete(socket)); +}); + +function shutdown() { + server.close(() => process.exit(0)); + for (const socket of sockets) socket.destroy(); +} + +process.on("SIGINT", shutdown); +process.on("SIGTERM", shutdown); + +server.listen(port, host, () => { + const address = server.address(); + const listeningPort = typeof address === "object" && address ? address.port : port; + process.stdout.write( + `${JSON.stringify({ event: "listening", host, port: listeningPort, totalMiB })}\n` + ); +}); diff --git a/tests/load/issue-1408-replay-oom/run-wave.sh b/tests/load/issue-1408-replay-oom/run-wave.sh new file mode 100755 index 000000000..564fdda6e --- /dev/null +++ b/tests/load/issue-1408-replay-oom/run-wave.sh @@ -0,0 +1,51 @@ +#!/bin/sh +set -eu + +if [ "$#" -lt 5 ] || [ "$#" -gt 6 ]; then + printf '%s\n' \ + "usage: run-wave.sh APP_URL MOCK_STATS_URL SCENARIO_PREFIX APP_CONTAINER OUTPUT [REDIS_CONTAINER]" >&2 + exit 2 +fi + +script_dir=$(CDPATH= cd -- "$(dirname -- "$0")" && pwd) +app_url="$1" +mock_stats_url="$2" +scenario_prefix="$3" +app_container="$4" +output="$5" +redis_container="${6:-}" + +waves="${CCH_WAVES:-8}" +requests_per_wave="${CCH_REQUESTS_PER_WAVE:-8}" +wave_interval_ms="${CCH_WAVE_INTERVAL_MS:-10000}" +samples="${CCH_SAMPLES:-18}" +sample_interval_seconds="${CCH_SAMPLE_INTERVAL_SECONDS:-10}" +node_bin="${NODE_BIN:-node}" + +sampler_pid="" +cleanup() { + if [ -n "$sampler_pid" ]; then + kill "$sampler_pid" 2>/dev/null || true + fi +} +trap cleanup EXIT INT TERM + +"$script_dir/sample-container.sh" \ + "$app_container" \ + "$output" \ + "$samples" \ + "$sample_interval_seconds" \ + "$redis_container" & +sampler_pid=$! + +"$node_bin" "$script_dir/drive-disconnect-waves.cjs" \ + "$app_url" \ + "$mock_stats_url" \ + "$scenario_prefix" \ + "$waves" \ + "$requests_per_wave" \ + "$wave_interval_ms" + +wait "$sampler_pid" +sampler_pid="" +trap - EXIT INT TERM diff --git a/tests/load/issue-1408-replay-oom/sample-container.sh b/tests/load/issue-1408-replay-oom/sample-container.sh new file mode 100755 index 000000000..ba202e560 --- /dev/null +++ b/tests/load/issue-1408-replay-oom/sample-container.sh @@ -0,0 +1,50 @@ +#!/bin/sh +set -eu + +if [ "$#" -lt 2 ] || [ "$#" -gt 5 ]; then + printf '%s\n' \ + "usage: sample-container.sh APP_CONTAINER OUTPUT [SAMPLES] [INTERVAL_SECONDS] [REDIS_CONTAINER]" >&2 + exit 2 +fi + +app="$1" +output="$2" +samples="${3:-18}" +interval="${4:-10}" +redis="${5:-}" + +case "$samples" in + *[!0-9]* | 0) printf '%s\n' "SAMPLES must be a positive integer" >&2; exit 2 ;; +esac +case "$interval" in + *[!0-9]*) printf '%s\n' "INTERVAL_SECONDS must be a non-negative integer" >&2; exit 2 ;; +esac + +start_epoch=$(date +%s) +: >"$output" +i=0 +while [ "$i" -lt "$samples" ]; do + epoch=$(date +%s) + state=$(docker inspect -f '{{.State.Status}} oom={{.State.OOMKilled}} exit={{.State.ExitCode}}' "$app" 2>/dev/null || true) + logs=$(docker logs --since "$start_epoch" "$app" 2>&1 || true) + memory=$(printf '%s\n' "$logs" | grep '"cchMemoryProbe":true' | tail -n 1 || true) + timeouts=$(printf '%s\n' "$logs" | grep -c 'Client abort drain window exceeded' || true) + backlogs=$(printf '%s\n' "$logs" | grep -c 'write_backlog_too_large' || true) + body_skips=$(printf '%s\n' "$logs" | grep -c 'Skipped oversized session response body' || true) + active=$(printf '%s\n' "$logs" | grep -E 'activeTasks|remainingTasks' | tail -n 1 || true) + + redis_state="" + redis_memory="" + if [ -n "$redis" ]; then + redis_state=$(docker inspect -f '{{.State.Status}} oom={{.State.OOMKilled}} exit={{.State.ExitCode}}' "$redis" 2>/dev/null || true) + redis_memory=$(docker exec "$redis" redis-cli --raw INFO memory 2>/dev/null | + grep -E '^(used_memory_human|used_memory_peak_human):' | + tr '\n' ',' || true) + fi + + printf '%s\n' \ + "sample=$i epoch=$epoch app=[$state] timeouts=$timeouts backlogs=$backlogs bodySkips=$body_skips memory=$memory lastTask=$active redis=[$redis_state] redisMemory=[$redis_memory]" \ + >>"$output" + i=$((i + 1)) + [ "$i" -ge "$samples" ] || sleep "$interval" +done diff --git a/tests/load/issue-1408-replay-oom/start-mock-container.sh b/tests/load/issue-1408-replay-oom/start-mock-container.sh new file mode 100755 index 000000000..3c9c0869e --- /dev/null +++ b/tests/load/issue-1408-replay-oom/start-mock-container.sh @@ -0,0 +1,54 @@ +#!/bin/sh +set -eu + +if [ "$#" -lt 3 ] || [ "$#" -gt 5 ]; then + printf '%s\n' \ + "usage: start-mock-container.sh CONTAINER NETWORK HOST_PORT [PAYLOAD_MIB] [NODE_IMAGE]" >&2 + exit 2 +fi + +container="$1" +network="$2" +host_port="$3" +payload_mib="${4:-8}" +node_image="${5:-node:22-alpine}" +script_dir=$(CDPATH= cd -- "$(dirname -- "$0")" && pwd) + +if docker container inspect "$container" >/dev/null 2>&1; then + printf '%s\n' "container already exists: $container" >&2 + exit 1 +fi + +docker run -d \ + --name "$container" \ + --network "$network" \ + -e CCH_MOCK_PORT=3001 \ + -e CCH_MOCK_MIB="$payload_mib" \ + -p "127.0.0.1:$host_port:3001" \ + -v "$script_dir/mock-upstream.cjs:/fixture/mock-upstream.cjs:ro" \ + "$node_image" \ + node /fixture/mock-upstream.cjs >/dev/null + +cleanup_container() { + docker rm -f "$container" >/dev/null 2>&1 || true +} + +trap cleanup_container 0 +trap 'exit 130' 2 +trap 'exit 143' 15 + +attempt=0 +while [ "$attempt" -lt 60 ]; do + if curl --connect-timeout 2 --max-time 5 -fsS \ + "http://127.0.0.1:$host_port/health" >/dev/null 2>&1; then + trap - 0 2 15 + printf '%s\n' \ + "ready container=$container stats=http://127.0.0.1:$host_port/stats provider=http://$container:3001" + exit 0 + fi + attempt=$((attempt + 1)) + sleep 1 +done + +docker logs --tail 100 "$container" >&2 || true +exit 1 diff --git a/tests/unit/lib/env-store-session-response-body.test.ts b/tests/unit/lib/env-store-session-response-body.test.ts index bcb25c45c..da16baf13 100644 --- a/tests/unit/lib/env-store-session-response-body.test.ts +++ b/tests/unit/lib/env-store-session-response-body.test.ts @@ -3,6 +3,7 @@ import { EnvSchema } from "@/lib/config/env.schema"; describe("EnvSchema - STORE_SESSION_RESPONSE_BODY", () => { const originalEnv = process.env.STORE_SESSION_RESPONSE_BODY; + const originalMaxBytes = process.env.SESSION_RESPONSE_BODY_MAX_BYTES; afterEach(() => { if (originalEnv === undefined) { @@ -10,6 +11,11 @@ describe("EnvSchema - STORE_SESSION_RESPONSE_BODY", () => { } else { process.env.STORE_SESSION_RESPONSE_BODY = originalEnv; } + if (originalMaxBytes === undefined) { + delete process.env.SESSION_RESPONSE_BODY_MAX_BYTES; + } else { + process.env.SESSION_RESPONSE_BODY_MAX_BYTES = originalMaxBytes; + } }); it("should default to true when not set", () => { @@ -41,4 +47,26 @@ describe("EnvSchema - STORE_SESSION_RESPONSE_BODY", () => { const result = EnvSchema.parse(process.env); expect(result.STORE_SESSION_RESPONSE_BODY).toBe(true); }); + + it("defaults the response body limit to 5 MiB", () => { + delete process.env.SESSION_RESPONSE_BODY_MAX_BYTES; + const result = EnvSchema.parse(process.env); + expect(result.SESSION_RESPONSE_BODY_MAX_BYTES).toBe(5 * 1024 * 1024); + }); + + it("accepts the inclusive 64 KiB and 64 MiB response body limit boundaries", () => { + process.env.SESSION_RESPONSE_BODY_MAX_BYTES = String(64 * 1024); + expect(EnvSchema.parse(process.env).SESSION_RESPONSE_BODY_MAX_BYTES).toBe(64 * 1024); + + process.env.SESSION_RESPONSE_BODY_MAX_BYTES = String(64 * 1024 * 1024); + expect(EnvSchema.parse(process.env).SESSION_RESPONSE_BODY_MAX_BYTES).toBe(64 * 1024 * 1024); + }); + + it("rejects response body limits outside the configured boundaries", () => { + process.env.SESSION_RESPONSE_BODY_MAX_BYTES = String(64 * 1024 - 1); + expect(() => EnvSchema.parse(process.env)).toThrow(); + + process.env.SESSION_RESPONSE_BODY_MAX_BYTES = String(64 * 1024 * 1024 + 1); + expect(() => EnvSchema.parse(process.env)).toThrow(); + }); }); diff --git a/tests/unit/lib/session-manager-redaction.test.ts b/tests/unit/lib/session-manager-redaction.test.ts index 3cc3c24f3..e3fb9ffb3 100644 --- a/tests/unit/lib/session-manager-redaction.test.ts +++ b/tests/unit/lib/session-manager-redaction.test.ts @@ -31,6 +31,7 @@ const redisMock = { status: "ready", setex: vi.fn().mockResolvedValue("OK"), get: vi.fn(), + del: vi.fn().mockResolvedValue(1), set: vi.fn().mockResolvedValue("OK"), expire: vi.fn().mockResolvedValue(1), incr: vi.fn().mockResolvedValue(1), @@ -49,10 +50,12 @@ vi.mock("@/lib/redis", () => ({ // Mock config - we'll control STORE_SESSION_MESSAGES dynamically let mockStoreMessages = false; let mockStoreSessionResponseBody = true; +let mockSessionResponseBodyMaxBytes = 1024 * 1024; vi.mock("@/lib/config/env.schema", () => ({ getEnvConfig: () => ({ STORE_SESSION_MESSAGES: mockStoreMessages, STORE_SESSION_RESPONSE_BODY: mockStoreSessionResponseBody, + SESSION_RESPONSE_BODY_MAX_BYTES: mockSessionResponseBodyMaxBytes, SESSION_TTL: 300, }), })); @@ -65,11 +68,13 @@ describe("SessionManager - Redaction based on STORE_SESSION_MESSAGES", () => { vi.clearAllMocks(); mockStoreMessages = false; // default: redact mockStoreSessionResponseBody = true; // default: store response body + mockSessionResponseBodyMaxBytes = 1024 * 1024; }); afterEach(() => { mockStoreMessages = false; mockStoreSessionResponseBody = true; + mockSessionResponseBodyMaxBytes = 1024 * 1024; }); describe("storeSessionMessages", () => { @@ -215,6 +220,37 @@ describe("SessionManager - Redaction based on STORE_SESSION_MESSAGES", () => { expect(value).toBe(nonJsonResponse); }); + it("should enforce the response body limit using UTF-8 bytes at the exact boundary", async () => { + mockSessionResponseBodyMaxBytes = 4; + + await SessionManager.storeSessionResponse("sess_utf8_exact", "中a", 1); + expect(redisMock.setex).toHaveBeenCalledWith( + "session:sess_utf8_exact:req:1:response", + 300, + "中a" + ); + + vi.clearAllMocks(); + await SessionManager.storeSessionResponse("sess_utf8_over", "中ab", 1); + + expect(redisMock.setex).not.toHaveBeenCalled(); + expect(loggerMock.warn).toHaveBeenCalledWith( + "SessionManager: Skipped oversized session response body", + { context: "response", byteSize: 5, maxBytes: 4 } + ); + expect(redisMock.del).toHaveBeenCalledWith("session:sess_utf8_over:req:1:response"); + }); + + it("should remove a previously stored response when the replacement exceeds the limit", async () => { + mockSessionResponseBodyMaxBytes = 4; + + await SessionManager.storeSessionResponse("sess_replace", "1234", 1); + await SessionManager.storeSessionResponse("sess_replace", "12345", 1); + + expect(redisMock.setex).toHaveBeenCalledTimes(1); + expect(redisMock.del).toHaveBeenCalledWith("session:sess_replace:req:1:response"); + }); + it("should handle OpenAI choices format when STORE_SESSION_MESSAGES=false", async () => { mockStoreMessages = false; const openaiResponse = { diff --git a/tests/unit/proxy/issue-1408-load-fixture.test.ts b/tests/unit/proxy/issue-1408-load-fixture.test.ts new file mode 100644 index 000000000..55a8478fd --- /dev/null +++ b/tests/unit/proxy/issue-1408-load-fixture.test.ts @@ -0,0 +1,520 @@ +import { + execFileSync, + spawn, + spawnSync, + type ChildProcessWithoutNullStreams, +} from "node:child_process"; +import { existsSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import http from "node:http"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { afterEach, describe, expect, it } from "vitest"; + +const fixtureDir = path.join(process.cwd(), "tests/load/issue-1408-replay-oom"); +const nodeScripts = ["mock-upstream.cjs", "drive-disconnect-waves.cjs", "memory-probe.cjs"]; +const shellScripts = ["sample-container.sh", "run-wave.sh", "start-mock-container.sh"]; +const children = new Set(); +const posixIt = it.skipIf(process.platform === "win32"); +const mockEnvironmentKeys = [ + "CCH_MOCK_HOST", + "CCH_MOCK_PORT", + "CCH_MOCK_MIB", + "CCH_MOCK_MAX_REQUEST_BYTES", +] as const; + +function createMockEnvironment(overrides: Record = {}): NodeJS.ProcessEnv { + const environment = { ...process.env }; + for (const key of mockEnvironmentKeys) delete environment[key]; + return { + ...environment, + CCH_MOCK_HOST: "127.0.0.1", + CCH_MOCK_PORT: "0", + CCH_MOCK_MIB: "0.0625", + ...overrides, + }; +} + +function getJson(url: URL): Promise> { + return new Promise((resolve, reject) => { + const request = http.get(url, (response) => { + const chunks: Buffer[] = []; + response.on("data", (chunk: Buffer) => chunks.push(chunk)); + response.on("end", () => { + try { + resolve(JSON.parse(Buffer.concat(chunks).toString("utf8")) as Record); + } catch (error) { + reject(error); + } + }); + }); + request.on("error", reject); + }); +} + +function requestJson( + url: URL, + method: string, + body = "" +): Promise<{ statusCode: number; body: Record }> { + return new Promise((resolve, reject) => { + const request = http.request( + url, + { + method, + headers: body + ? { + "content-type": "application/json", + "content-length": Buffer.byteLength(body), + } + : undefined, + }, + (response) => { + const chunks: Buffer[] = []; + response.on("data", (chunk: Buffer) => chunks.push(chunk)); + response.on("end", () => { + try { + resolve({ + statusCode: response.statusCode ?? 0, + body: JSON.parse(Buffer.concat(chunks).toString("utf8")) as Record, + }); + } catch (error) { + reject(error); + } + }); + } + ); + request.on("error", reject); + request.end(body); + }); +} + +async function startMock( + overrides: Record = {} +): Promise<{ child: ChildProcessWithoutNullStreams; baseUrl: URL }> { + const child = spawn(process.execPath, [path.join(fixtureDir, "mock-upstream.cjs")], { + env: createMockEnvironment(overrides), + stdio: ["ignore", "pipe", "pipe"], + }); + children.add(child); + + const listening = await waitForJsonLine(child, (value) => value.event === "listening"); + expect(listening.port).toEqual(expect.any(Number)); + return { child, baseUrl: new URL(`http://127.0.0.1:${listening.port}`) }; +} + +function waitForJsonLine( + child: ChildProcessWithoutNullStreams, + predicate: (value: Record) => boolean +): Promise> { + return new Promise((resolve, reject) => { + let buffered = ""; + const timer = setTimeout(() => reject(new Error("fixture output timeout")), 5000); + child.stdout.on("data", (chunk: Buffer) => { + buffered += chunk.toString("utf8"); + const lines = buffered.split("\n"); + buffered = lines.pop() ?? ""; + for (const line of lines) { + if (!line.trim()) continue; + const value = JSON.parse(line) as Record; + if (predicate(value)) { + clearTimeout(timer); + resolve(value); + return; + } + } + }); + child.once("error", (error) => { + clearTimeout(timer); + reject(error); + }); + child.once("exit", (code) => { + clearTimeout(timer); + reject(new Error(`fixture exited before readiness: ${code}`)); + }); + }); +} + +function postAndAbort(url: URL, body: string): Promise { + return new Promise((resolve, reject) => { + const request = http.request( + url, + { + method: "POST", + headers: { + "content-type": "application/json", + "content-length": Buffer.byteLength(body), + }, + }, + (response) => { + response.once("data", () => { + response.destroy(); + resolve(); + }); + response.on("error", () => resolve()); + } + ); + request.on("error", reject); + request.end(body); + }); +} + +function waitForExit( + child: ChildProcessWithoutNullStreams +): Promise<{ code: number | null; stderr: string }> { + return new Promise((resolve, reject) => { + let stderr = ""; + child.stderr.on("data", (chunk: Buffer) => { + stderr += chunk.toString("utf8"); + }); + child.once("error", reject); + child.once("exit", (code) => resolve({ code, stderr })); + }); +} + +function installFakeContainerCommands(directory: string): void { + writeFileSync( + path.join(directory, "docker"), + `#!/bin/sh +set -eu +printf '%s\\n' "$*" >> "$CCH_FAKE_DOCKER_LOG" +case "$1" in + container) + [ "$2" = "inspect" ] || exit 2 + exit "\${CCH_FAKE_CONTAINER_INSPECT_STATUS:-1}" + ;; + inspect) + exit "\${CCH_FAKE_IMAGE_INSPECT_STATUS:-0}" + ;; + run) + : > "$CCH_FAKE_CONTAINER_STATE" + ;; + logs) + ;; + rm) + rm -f "$CCH_FAKE_CONTAINER_STATE" + ;; + *) + exit 2 + ;; +esac +`, + { mode: 0o755 } + ); + writeFileSync( + path.join(directory, "curl"), + `#!/bin/sh +printf '%s\\n' "$*" >> "$CCH_FAKE_CURL_LOG" +exit "\${CCH_FAKE_CURL_STATUS:-1}" +`, + { mode: 0o755 } + ); + writeFileSync( + path.join(directory, "sleep"), + `#!/bin/sh +case "\${CCH_FAKE_SLEEP_MODE:-success}" in + signal-int) + kill -INT "$PPID" + ;; + signal-term) + kill -TERM "$PPID" + ;; + fail) + exit 7 + ;; +esac +`, + { mode: 0o755 } + ); +} + +function runStartMockWithFakeCommands( + directory: string, + overrides: Record +): ReturnType { + return spawnSync( + "sh", + [ + path.join(fixtureDir, "start-mock-container.sh"), + "fixture-container", + "fixture-network", + "31409", + ], + { + encoding: "utf8", + env: { + ...process.env, + PATH: `${directory}:${process.env.PATH ?? ""}`, + CCH_FAKE_CONTAINER_STATE: path.join(directory, "container-state"), + CCH_FAKE_CURL_LOG: path.join(directory, "curl.log"), + CCH_FAKE_DOCKER_LOG: path.join(directory, "docker.log"), + ...overrides, + }, + } + ); +} + +afterEach(async () => { + const exits = [...children].map( + (child) => + new Promise((resolve) => { + if (child.exitCode !== null || child.signalCode !== null) { + resolve(); + return; + } + const forceTimer = setTimeout(() => { + if (child.exitCode === null && child.signalCode === null) child.kill("SIGKILL"); + }, 1000); + child.once("exit", () => { + clearTimeout(forceTimer); + resolve(); + }); + child.kill("SIGTERM"); + }) + ); + children.clear(); + await Promise.all(exits); +}); + +describe("issue #1408 load fixture", () => { + it("keeps repository scripts syntactically valid and independent from temporary paths", () => { + for (const filename of nodeScripts) { + const file = path.join(fixtureDir, filename); + execFileSync(process.execPath, ["--check", file]); + expect(readFileSync(file, "utf8")).not.toContain("/private/tmp"); + } + + for (const filename of shellScripts) { + const file = path.join(fixtureDir, filename); + if (process.platform !== "win32") execFileSync("sh", ["-n", file]); + expect(readFileSync(file, "utf8")).not.toContain("/private/tmp"); + } + + const startMock = readFileSync(path.join(fixtureDir, "start-mock-container.sh"), "utf8"); + expect(startMock).toContain('docker container inspect "$container"'); + }); + + posixIt("checks container existence without treating a same-named image as a collision", () => { + const directory = mkdtempSync(path.join(tmpdir(), "cch1408-container-fixture-")); + try { + installFakeContainerCommands(directory); + const result = runStartMockWithFakeCommands(directory, { CCH_FAKE_CURL_STATUS: "0" }); + + expect(result.status).toBe(0); + expect(result.stdout).toContain("ready container=fixture-container"); + expect(existsSync(path.join(directory, "container-state"))).toBe(true); + expect(readFileSync(path.join(directory, "docker.log"), "utf8")).toMatch( + /^container inspect fixture-container$/m + ); + } finally { + rmSync(directory, { recursive: true, force: true }); + } + }); + + posixIt.each(["signal-int", "signal-term", "fail"])( + "removes the new container when readiness exits via %s", + (sleepMode) => { + const directory = mkdtempSync(path.join(tmpdir(), "cch1408-container-fixture-")); + try { + installFakeContainerCommands(directory); + const result = runStartMockWithFakeCommands(directory, { + CCH_FAKE_CURL_STATUS: "1", + CCH_FAKE_SLEEP_MODE: sleepMode, + }); + + expect(result.status).not.toBe(0); + expect(existsSync(path.join(directory, "container-state"))).toBe(false); + expect(readFileSync(path.join(directory, "docker.log"), "utf8")).toMatch( + /^rm -f fixture-container$/m + ); + } finally { + rmSync(directory, { recursive: true, force: true }); + } + } + ); + + posixIt("bounds stalled health probes and cleans up after the retry budget", () => { + const directory = mkdtempSync(path.join(tmpdir(), "cch1408-container-fixture-")); + try { + installFakeContainerCommands(directory); + const result = runStartMockWithFakeCommands(directory, { + CCH_FAKE_CURL_STATUS: "28", + }); + + expect(result.status).not.toBe(0); + expect(existsSync(path.join(directory, "container-state"))).toBe(false); + const probes = readFileSync(path.join(directory, "curl.log"), "utf8").trim().split("\n"); + expect(probes).toHaveLength(60); + expect(new Set(probes)).toEqual( + new Set(["--connect-timeout 2 --max-time 5 -fsS http://127.0.0.1:31409/health"]) + ); + } finally { + rmSync(directory, { recursive: true, force: true }); + } + }); + + it("emits a valid hanging Responses SSE stream and records the scenario count", async () => { + const { baseUrl } = await startMock(); + await postAndAbort( + new URL("/v1/responses", baseUrl), + JSON.stringify({ input: "CCH_SCENARIO_contract-fixture", stream: true }) + ); + + const stats = await getJson(new URL("/stats", baseUrl)); + expect(stats).toMatchObject({ + counts: { "contract-fixture": 1 }, + totalMiB: 0.0625, + }); + }); + + it("ignores inherited mock configuration when starting the fixture", async () => { + const previous = process.env.CCH_MOCK_MAX_REQUEST_BYTES; + process.env.CCH_MOCK_MAX_REQUEST_BYTES = "1.5"; + try { + const { baseUrl } = await startMock(); + await expect(getJson(new URL("/stats", baseUrl))).resolves.toMatchObject({ counts: {} }); + } finally { + if (previous === undefined) delete process.env.CCH_MOCK_MAX_REQUEST_BYTES; + else process.env.CCH_MOCK_MAX_REQUEST_BYTES = previous; + } + }); + + it("resets counters, rejects unknown routes, and bounds request bodies", async () => { + const { baseUrl } = await startMock({ CCH_MOCK_MAX_REQUEST_BYTES: "64" }); + + const missing = await requestJson(new URL("/unknown", baseUrl), "GET"); + expect(missing).toEqual({ statusCode: 404, body: { error: "not found" } }); + + const oversized = await requestJson( + new URL("/v1/responses", baseUrl), + "POST", + JSON.stringify({ input: "x".repeat(128), stream: true }) + ); + expect(oversized).toEqual({ + statusCode: 413, + body: { error: "request body too large" }, + }); + + await postAndAbort( + new URL("/v1/responses", baseUrl), + JSON.stringify({ input: "CCH_SCENARIO_reset-me", stream: true }) + ); + await expect(getJson(new URL("/stats", baseUrl))).resolves.toMatchObject({ + counts: { "reset-me": 1 }, + }); + + const reset = await requestJson(new URL("/reset", baseUrl), "POST"); + expect(reset).toEqual({ statusCode: 200, body: { reset: true } }); + await expect(getJson(new URL("/stats", baseUrl))).resolves.toMatchObject({ counts: {} }); + }); + + it.each([ + ["payload below one frame", { CCH_MOCK_MIB: "0.01" }, "CCH_MOCK_MIB"], + ["payload above the fixture cap", { CCH_MOCK_MIB: "65" }, "CCH_MOCK_MIB"], + ["fractional request limit", { CCH_MOCK_MAX_REQUEST_BYTES: "1.5" }, "CCH_MOCK_MAX"], + ])("rejects invalid mock configuration: %s", (_name, overrides, expected) => { + const result = spawnSync(process.execPath, [path.join(fixtureDir, "mock-upstream.cjs")], { + encoding: "utf8", + env: createMockEnvironment(overrides), + }); + + expect(result.status).not.toBe(0); + expect(result.stderr).toContain(expected); + }); + + it("prints a structured memory sample when preloaded", () => { + const output = execFileSync(process.execPath, [path.join(fixtureDir, "memory-probe.cjs")], { + encoding: "utf8", + }); + const sample = JSON.parse(output.trim()) as Record; + expect(sample).toMatchObject({ cchMemoryProbe: true }); + expect(sample.rssMiB).toEqual(expect.any(Number)); + expect(sample.heapUsedMiB).toEqual(expect.any(Number)); + expect(sample.externalMiB).toEqual(expect.any(Number)); + expect(sample.arrayBuffersMiB).toEqual(expect.any(Number)); + expect(sample.resources).toEqual(expect.any(Object)); + }); + + it("requires the API key through an explicit environment boundary", () => { + const result = spawnSync( + process.execPath, + [ + path.join(fixtureDir, "drive-disconnect-waves.cjs"), + "http://127.0.0.1:1", + "http://127.0.0.1:2/stats", + "missing-key", + "1", + "1", + "0", + ], + { + encoding: "utf8", + env: { ...process.env, CCH_API_KEY: "", CCH_API_KEY_FILE: "" }, + } + ); + + expect(result.status).not.toBe(0); + expect(result.stderr).toContain("set CCH_API_KEY or CCH_API_KEY_FILE"); + }); + + it("fails when the mock stats response is interrupted after headers", async () => { + const server = http.createServer((_request, response) => { + response.writeHead(200, { "content-type": "application/json" }); + response.write('{"counts":'); + setTimeout(() => response.destroy(), 10); + }); + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(0, "127.0.0.1", resolve); + }); + + try { + const address = server.address(); + if (!address || typeof address === "string") throw new Error("fixture server has no port"); + const child = spawn( + process.execPath, + [ + path.join(fixtureDir, "drive-disconnect-waves.cjs"), + "http://127.0.0.1:1", + `http://127.0.0.1:${address.port}/stats`, + "interrupted-stats", + "1", + "1", + "0", + ], + { + env: { ...process.env, CCH_API_KEY: "fixture-key", CCH_API_KEY_FILE: "" }, + stdio: ["ignore", "pipe", "pipe"], + } + ); + children.add(child); + + const result = await waitForExit(child); + expect(result.code).not.toBe(0); + expect(result.stderr).toContain("response aborted"); + } finally { + await new Promise((resolve) => server.close(() => resolve())); + } + }); + + it.each([ + ["unsupported URL protocol", ["ftp://127.0.0.1", "http://127.0.0.1/stats", "valid"], "APP_URL"], + [ + "invalid scenario characters", + ["http://127.0.0.1", "http://127.0.0.1/stats", "bad:value"], + "SCENARIO_PREFIX", + ], + ["zero waves", ["http://127.0.0.1", "http://127.0.0.1/stats", "valid", "0"], "WAVES"], + ])("rejects invalid driver input: %s", (_name, args, expected) => { + const result = spawnSync( + process.execPath, + [path.join(fixtureDir, "drive-disconnect-waves.cjs"), ...args], + { + encoding: "utf8", + env: { ...process.env, CCH_API_KEY: "fixture-key", CCH_API_KEY_FILE: "" }, + } + ); + + expect(result.status).not.toBe(0); + expect(result.stderr).toContain(expected); + }); +}); diff --git a/tests/unit/proxy/replay-spool.test.ts b/tests/unit/proxy/replay-spool.test.ts index e366083d2..f07e67408 100644 --- a/tests/unit/proxy/replay-spool.test.ts +++ b/tests/unit/proxy/replay-spool.test.ts @@ -29,7 +29,7 @@ const envControl = vi.hoisted(() => ({ const storeControl = vi.hoisted(() => { const order: string[] = []; - let ownedChunkCount = 0; + const ownedChunksByReplayId = new Map(); const store = { appendChunks: vi.fn(async (_replayId: string, values: string[]) => { order.push(`append:${values.join("|")}`); @@ -41,16 +41,22 @@ const storeControl = vi.hoisted(() => { }), writeOwned: vi.fn( async ( - _replayId: string, + replayId: string, _ownerToken: string, _meta: { status: string }, values: string[] = [] ) => { order.push(`write:${values.join("|")}`); - ownedChunkCount += values.length; - return ownedChunkCount; + const ownedChunks = ownedChunksByReplayId.get(replayId) ?? []; + ownedChunks.push(...values); + ownedChunksByReplayId.set(replayId, ownedChunks); + return ownedChunks.length; } ), + readChunks: vi.fn(async (replayId: string, fromIndex: number) => { + order.push("read"); + return (ownedChunksByReplayId.get(replayId) ?? []).slice(fromIndex); + }), completeOwned: vi.fn( async (_replayId: string, _ownerToken: string, meta: { status: string }) => { order.push(`meta:${meta.status}`); @@ -89,8 +95,8 @@ const storeControl = vi.hoisted(() => { return { order, store, - resetOwnedChunkCount: () => { - ownedChunkCount = 0; + resetOwnedChunks: () => { + ownedChunksByReplayId.clear(); }, }; }); @@ -156,13 +162,6 @@ 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" }, @@ -181,7 +180,7 @@ beforeEach(() => { envControl.maxPayloadBytes = 8 * 1024 * 1024; envControl.maxConcurrentSpools = 64; storeControl.order.length = 0; - storeControl.resetOwnedChunkCount(); + storeControl.resetOwnedChunks(); storeControl.store.abortOwned.mockImplementation( async (_replayId: string, _ownerToken: string, meta: { status: string }) => { storeControl.order.push(`meta:${meta.status}`); @@ -191,6 +190,11 @@ beforeEach(() => { } ); storeControl.store.completeOwned.mockClear(); + storeControl.store.persistCompleted.mockReset(); + storeControl.store.persistCompleted.mockImplementation(async () => { + storeControl.order.push("persist"); + return "persisted" as const; + }); storeControl.store.releaseOwner.mockImplementation(async () => { storeControl.order.push("release"); }); @@ -256,6 +260,28 @@ describe("ReplaySpool：write-behind 批量冲刷", () => { await spool.abort("test_cleanup"); }); + it("跨 chunk UTF-8 序列按全部输入字节触发冲刷阈值", async () => { + const spool = makeSpool(); + const continuation = new Uint8Array(1 + (64 * 1024 - 3)); + continuation[0] = 0xad; + continuation.fill(0x78, 1); + + spool.observe(new Uint8Array([0xe4, 0xb8])); + expect((spool as unknown as { pendingBytes: number }).pendingBytes).toBe(2); + + spool.observe(continuation); + expect((spool as unknown as { queuedWriteBytes: number }).queuedWriteBytes).toBe(64 * 1024); + + await drainWriteChain(spool); + expect(storeControl.store.writeOwned).toHaveBeenCalledWith( + identity.replayId, + "owner-token", + expect.objectContaining({ byteSize: 64 * 1024 }), + [`中${"x".repeat(64 * 1024 - 3)}`] + ); + await spool.abort("test_cleanup"); + }); + it("没有响应 chunk 时仍按 15 秒间隔续租 owner", async () => { const spool = makeSpool(); @@ -324,6 +350,49 @@ describe("ReplaySpool：write-behind 批量冲刷", () => { expect(spool.isTerminal).toBe(true); expect(getActiveReplaySpoolCount()).toBe(0); }); + + it("Redis write backlog 超过 1 MiB 时放弃 spool 并清空排队 batch", async () => { + let resolveFirstWrite!: (value: number) => void; + const onInactive = vi.fn(); + storeControl.store.writeOwned.mockImplementationOnce( + () => + new Promise((resolve) => { + resolveFirstWrite = resolve; + }) + ); + const spool = new ReplaySpool( + identity, + "owner-token", + 200, + { "content-type": "text/event-stream" }, + "stream", + { onInactive } + ); + + for (let index = 0; index < 16; index += 1) { + spool.observe(encoder.encode("x".repeat(64 * 1024))); + await vi.advanceTimersByTimeAsync(0); + } + + expect(spool.isTerminal).toBe(false); + expect((spool as unknown as { queuedWriteBytes: number }).queuedWriteBytes).toBe(1024 * 1024); + + spool.observe(encoder.encode("x".repeat(64 * 1024))); + await vi.advanceTimersByTimeAsync(0); + + expect(spool.isTerminal).toBe(true); + expect(onInactive).toHaveBeenCalledTimes(1); + expect((spool as unknown as { queuedBatches: Set }).queuedBatches.size).toBe(0); + expect((spool as unknown as { queuedWriteBytes: number }).queuedWriteBytes).toBe(0); + + resolveFirstWrite(1); + await drainWriteChain(spool); + expect(storeControl.store.abortOwned).toHaveBeenCalledWith( + identity.replayId, + "owner-token", + expect.objectContaining({ status: "aborted", abortReason: "write_backlog_too_large" }) + ); + }); }); describe("ReplaySpool：续租丢失 halt", () => { @@ -403,6 +472,7 @@ describe("ReplaySpool：completeAfterBilling 终态屏障", () => { expect(storeControl.order).toEqual([ "write:data: hello \n\n|data: world\n\n", + "read", "persist", "meta:completed", "release", @@ -496,6 +566,41 @@ describe("ReplaySpool：completeAfterBilling 终态屏障", () => { expect(getActiveReplaySpoolCount()).toBe(0); }); + it("阻塞写入加完成尾批超过 1 MiB 时放弃 spool 并等待 fenced cleanup", async () => { + let resolveFirstWrite!: (value: number) => void; + storeControl.store.writeOwned.mockImplementationOnce( + () => + new Promise((resolve) => { + resolveFirstWrite = resolve; + }) + ); + const spool = makeSpool(); + for (let index = 0; index < 16; index += 1) { + spool.observe(encoder.encode("x".repeat(64 * 1024))); + await vi.advanceTimersByTimeAsync(0); + } + expect((spool as unknown as { queuedWriteBytes: number }).queuedWriteBytes).toBe(1024 * 1024); + + spool.observe(encoder.encode("tail")); + const completion = spool.completeAfterBilling(5); + await vi.advanceTimersByTimeAsync(0); + + expect(spool.isTerminal).toBe(true); + expect((spool as unknown as { pending: string[] }).pending).toEqual([]); + expect((spool as unknown as { queuedBatches: Set }).queuedBatches.size).toBe(0); + expect((spool as unknown as { queuedWriteBytes: number }).queuedWriteBytes).toBe(0); + expect(storeControl.store.persistCompleted).not.toHaveBeenCalled(); + + resolveFirstWrite(1); + await completion; + expect(storeControl.store.abortOwned).toHaveBeenCalledWith( + identity.replayId, + "owner-token", + expect.objectContaining({ status: "aborted", abortReason: "write_backlog_too_large" }) + ); + expect(getActiveReplaySpoolCount()).toBe(0); + }); + it("persist 成功但 completed 翻转失败：日志标记 pgPersisted=true，热层封死为 aborted", async () => { storeControl.store.completeOwned.mockResolvedValueOnce(false); const spool = makeSpool(); @@ -540,17 +645,10 @@ describe("ReplaySpool：completeAfterBilling 终态屏障", () => { expect(storeControl.store.completeOwned).toHaveBeenCalledTimes(1); }); - it("Redis 阻塞期间不提前复制 payload，PG 阻塞期间释放 parts", async () => { + it("活跃 spool 不保留整流 parts，完成时从 Redis chunks 组装 durable payload", 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) => { @@ -560,41 +658,79 @@ describe("ReplaySpool：completeAfterBilling 终态屏障", () => { const spool = makeSpool(); for (let index = 0; index < 64; index += 1) { spool.observe(encoder.encode(chunk)); + await vi.advanceTimersByTimeAsync(0); + await drainWriteChain(spool); } - expect(retainedAsciiPartBytes(spool)).toBe(payloadBytes); + expect("parts" in (spool as object)).toBe(false); 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.readChunks).toHaveBeenCalledWith(identity.replayId, 0); 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 () => { + it("并发完成时串行重建和持久化完整 payload，限制瞬时堆峰值", async () => { + const secondIdentity: ReplayIdentity = { + ...identity, + replayId: "1123456789abcdef0123456789abcdef", + verifier: "eedcba9876543210fedcba9876543210", + }; + let resolveFirstPersist!: (value: "persisted") => void; + storeControl.store.persistCompleted.mockImplementation((row: { replayId: string }) => { + if (row.replayId === identity.replayId) { + return new Promise<"persisted">((resolve) => { + resolveFirstPersist = resolve; + }); + } + return Promise.resolve("persisted" as const); + }); + const first = makeSpool(); + const second = new ReplaySpool( + secondIdentity, + "second-owner-token", + 200, + { "content-type": "text/event-stream" }, + "stream" + ); + first.observe(encoder.encode("data: first\n\n")); + second.observe(encoder.encode("data: second\n\n")); + + const firstCompletion = first.completeAfterBilling(10); + const secondCompletion = second.completeAfterBilling(11); + await vi.advanceTimersByTimeAsync(0); + + expect(storeControl.store.persistCompleted).toHaveBeenCalledTimes(1); + expect(storeControl.store.persistCompleted).toHaveBeenCalledWith( + expect.objectContaining({ replayId: identity.replayId, payload: "data: first\n\n" }) + ); + + resolveFirstPersist("persisted"); + await firstCompletion; + await vi.advanceTimersByTimeAsync(0); + await secondCompletion; + + expect(storeControl.store.persistCompleted).toHaveBeenCalledTimes(2); + expect(storeControl.store.persistCompleted).toHaveBeenLastCalledWith( + expect.objectContaining({ + replayId: secondIdentity.replayId, + payload: "data: second\n\n", + }) + ); + }); + + it("Redis chunks 缺失时封死热层并释放 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"); - }, - }; + storeControl.store.readChunks.mockResolvedValueOnce(null); await spool.completeAfterBilling(10); @@ -610,6 +746,36 @@ describe("ReplaySpool：completeAfterBilling 终态屏障", () => { expect(storeControl.store.renewOwnerLease).not.toHaveBeenCalled(); }); + it("inactive 回调异常时仍完成 spool 清理并释放并发配额", async () => { + const onInactive = vi.fn(() => { + throw new Error("callback failed"); + }); + const spool = new ReplaySpool( + identity, + "owner-token", + 200, + { "content-type": "text/event-stream" }, + "stream", + { onInactive } + ); + envControl.maxPayloadBytes = 4; + + spool.observe(encoder.encode("12345678")); + await drainWriteChain(spool); + + expect(onInactive).toHaveBeenCalledOnce(); + expect(storeControl.store.abortOwned).toHaveBeenCalledWith( + identity.replayId, + "owner-token", + expect.objectContaining({ abortReason: "payload_too_large" }) + ); + expect(logger.debug).toHaveBeenCalledWith( + "[ReplaySpool] inactive callback failed", + expect.objectContaining({ error: "callback failed" }) + ); + expect(getActiveReplaySpoolCount()).toBe(0); + }); + it("跨 chunk 截断的 UTF-8 序列在 complete 时冲刷解码尾部", async () => { const spool = makeSpool(); // "中" (0xE4 0xB8 0xAD) 只送前两字节：observe 阶段解码挂起，complete 时 flush 出替换字符 @@ -656,13 +822,14 @@ describe("ReplaySpool：abort 终态", () => { expect(getActiveReplaySpoolCount()).toBe(0); }); - it("abort 立即释放已累积的 payload", async () => { + it("abort 立即清空 pending 与 queued batch", 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([]); + expect((spool as unknown as { pending: string[] }).pending).toEqual([]); + expect((spool as unknown as { queuedBatches: Set }).queuedBatches.size).toBe(0); }); it("Redis flush 阻塞时 abort 立即释放 batch，并在 fenced cleanup 后释放并发配额", async () => { @@ -798,10 +965,22 @@ describe("ReplaySpool：isTerminal", () => { expect(aborted.isTerminal).toBe(true); envControl.maxPayloadBytes = 4; - const oversized = makeSpool(); + const onInactive = vi.fn(); + const oversized = new ReplaySpool( + identity, + "owner-token", + 200, + { "content-type": "text/event-stream" }, + "stream", + { onInactive } + ); oversized.observe(encoder.encode("12345678")); expect(oversized.isTerminal).toBe(true); + expect(onInactive).toHaveBeenCalledTimes(1); await drainWriteChain(oversized); + + await oversized.abort("late_abort"); + expect(onInactive).toHaveBeenCalledTimes(1); }); }); diff --git a/tests/unit/proxy/response-handler-stream-terminal.test.ts b/tests/unit/proxy/response-handler-stream-terminal.test.ts index 8486af410..3718a28c3 100644 --- a/tests/unit/proxy/response-handler-stream-terminal.test.ts +++ b/tests/unit/proxy/response-handler-stream-terminal.test.ts @@ -19,6 +19,7 @@ const mocks = vi.hoisted(() => ({ replayObserve: vi.fn(), replayComplete: vi.fn(async () => {}), replayAbort: vi.fn(async () => {}), + replayInactive: null as (() => void) | null, })); vi.mock("@/app/v1/_lib/proxy/response-fixer", () => ({ @@ -60,15 +61,21 @@ vi.mock("@/lib/proxy-status-tracker", () => ({ })); vi.mock("@/app/v1/_lib/proxy/replay/replay-spool", () => ({ abortReplayOwnership: vi.fn(async () => undefined), - createReplaySpoolIfOwner: (session: ProxySession) => - session.replayState?.role === "owner" - ? { - abort: mocks.replayAbort, - completeAfterBilling: mocks.replayComplete, - isTerminal: false, - observe: mocks.replayObserve, - } - : null, + createReplaySpoolIfOwner: ( + session: ProxySession, + _response: Response, + _delivery: string, + options: { onInactive?: () => void } = {} + ) => { + if (session.replayState?.role !== "owner") return null; + mocks.replayInactive = options.onInactive ?? null; + return { + abort: mocks.replayAbort, + completeAfterBilling: mocks.replayComplete, + isTerminal: false, + observe: mocks.replayObserve, + }; + }, releaseReplayOwnership: vi.fn(), })); vi.mock("@/repository/message", () => ({ @@ -222,9 +229,27 @@ function sseResponse(body: BodyInit, status = 200): Response { return new Response(body, { status, headers: { "content-type": "text/event-stream" } }); } +function setReplayOwner(session: ProxySession, suffix: string): void { + session.replayState = { + role: "owner", + ownerToken: `owner-token-${suffix}`, + identity: { + replayId: `replay-${suffix}`, + verifier: "verifier", + scopeTag: "scope-tag", + keyId: KEY.id, + userId: USER.id, + format: "claude", + model: "claude-test", + endpoint: "/v1/messages", + }, + }; +} + describe("ProxyResponseHandler.dispatch stream terminal behavior", () => { beforeEach(() => { mocks.tasks.length = 0; + mocks.replayInactive = null; vi.clearAllMocks(); mocks.durable.mockImplementation(async (_id, _details, options) => { await options?.onCommitted?.(); @@ -277,6 +302,131 @@ describe("ProxyResponseHandler.dispatch stream terminal behavior", () => { expect(releaseAgent).toHaveBeenCalledOnce(); }); + it("caps a detached Replay drain at 60 seconds after the spool becomes inactive", async () => { + vi.useFakeTimers(); + const previousReplayDetachedMs = process.env.REPLAY_MAX_DETACHED_MS; + process.env.REPLAY_MAX_DETACHED_MS = "300000"; + try { + const cancelSource = vi.fn(); + const responseController = new AbortController(); + const source = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('data: {"partial":true}\n\n')); + }, + cancel: cancelSource, + }); + const { session } = await createSession({ responseController }); + setReplayOwner(session, "detached"); + + const returned = await ProxyResponseHandler.dispatch(session, sseResponse(source)); + const reader = returned.body?.getReader(); + await reader?.read(); + await reader?.cancel(new Error("client disconnected")); + + expect(mocks.replayInactive).toEqual(expect.any(Function)); + await vi.advanceTimersByTimeAsync(59_999); + expect(responseController.signal.aborted).toBe(false); + + mocks.replayInactive?.(); + await vi.advanceTimersByTimeAsync(1); + + expect(responseController.signal.aborted).toBe(true); + expect(responseController.signal.reason).toEqual( + expect.objectContaining({ message: "client_abort_drain_timeout" }) + ); + expect(cancelSource).toHaveBeenCalledOnce(); + await settleTasks(); + } finally { + if (previousReplayDetachedMs === undefined) { + delete process.env.REPLAY_MAX_DETACHED_MS; + } else { + process.env.REPLAY_MAX_DETACHED_MS = previousReplayDetachedMs; + } + vi.useRealTimers(); + } + }); + + it("keeps the configured 300-second drain while the Replay spool remains active", async () => { + vi.useFakeTimers(); + const previousReplayDetachedMs = process.env.REPLAY_MAX_DETACHED_MS; + process.env.REPLAY_MAX_DETACHED_MS = "300000"; + try { + const cancelSource = vi.fn(); + const responseController = new AbortController(); + const source = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('data: {"partial":true}\n\n')); + }, + cancel: cancelSource, + }); + const { session } = await createSession({ responseController }); + setReplayOwner(session, "active"); + + const returned = await ProxyResponseHandler.dispatch(session, sseResponse(source)); + const reader = returned.body?.getReader(); + await reader?.read(); + await reader?.cancel(new Error("client disconnected")); + + await vi.advanceTimersByTimeAsync(60_000); + expect(responseController.signal.aborted).toBe(false); + await vi.advanceTimersByTimeAsync(239_999); + expect(responseController.signal.aborted).toBe(false); + + await vi.advanceTimersByTimeAsync(1); + + expect(responseController.signal.aborted).toBe(true); + expect(cancelSource).toHaveBeenCalledOnce(); + await settleTasks(); + } finally { + if (previousReplayDetachedMs === undefined) { + delete process.env.REPLAY_MAX_DETACHED_MS; + } else { + process.env.REPLAY_MAX_DETACHED_MS = previousReplayDetachedMs; + } + vi.useRealTimers(); + } + }); + + it("uses the 60-second drain when Replay becomes inactive before client detach", async () => { + vi.useFakeTimers(); + const previousReplayDetachedMs = process.env.REPLAY_MAX_DETACHED_MS; + process.env.REPLAY_MAX_DETACHED_MS = "300000"; + try { + const cancelSource = vi.fn(); + const responseController = new AbortController(); + const source = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('data: {"partial":true}\n\n')); + }, + cancel: cancelSource, + }); + const { session } = await createSession({ responseController }); + setReplayOwner(session, "inactive-before-detach"); + + const returned = await ProxyResponseHandler.dispatch(session, sseResponse(source)); + const reader = returned.body?.getReader(); + await reader?.read(); + expect(mocks.replayInactive).toEqual(expect.any(Function)); + mocks.replayInactive?.(); + await reader?.cancel(new Error("client disconnected")); + + await vi.advanceTimersByTimeAsync(59_999); + expect(responseController.signal.aborted).toBe(false); + await vi.advanceTimersByTimeAsync(1); + + expect(responseController.signal.aborted).toBe(true); + expect(cancelSource).toHaveBeenCalledOnce(); + await settleTasks(); + } finally { + if (previousReplayDetachedMs === undefined) { + delete process.env.REPLAY_MAX_DETACHED_MS; + } else { + process.env.REPLAY_MAX_DETACHED_MS = previousReplayDetachedMs; + } + vi.useRealTimers(); + } + }); + it("persists a response-controller timeout as 502 and cancels the source", async () => { const cancelSource = vi.fn(); const responseController = new AbortController();