diff --git a/.env.example b/.env.example index cb3fd0dfb..0d6137a7c 100644 --- a/.env.example +++ b/.env.example @@ -105,7 +105,14 @@ 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） +SESSION_RESPONSE_BODY_DEDUP_ENABLED=false # response body 单 key 去重 writer（默认：false） + # 滚动部署必须先保持 false 部署新版 reader；确认所有实例升级后再设为 true + # true 时 legacy/before/after 共享同一 request-scoped Redis Hash 和 TTL + # requestSequence 有效时，false/true writer 以原子 generation marker 协调回滚与 retry，避免读取旧正文 + # request 开始时还会记录正文 generation；full termination 后，迟到的旧 request writer 不得重建正文 + # 无 requestSequence 时走兼容 legacy/before/after key，不执行 generation-marker Lua +SESSION_RESPONSE_BODY_MAX_BYTES=5242880 # 会话响应体 Redis 存储上限（默认 5 MiB，范围 64 KiB-64 MiB） + # 去重关闭时限制每份正文；去重开启时限制所有唯一正文的 UTF-8 总字节数 # 超限正文不落 Redis；before/after snapshot 的 headers/meta 仍保留 # Dashboard 配置 diff --git a/docs/research/issue-1415-session-response-body-dedup.md b/docs/research/issue-1415-session-response-body-dedup.md new file mode 100644 index 000000000..26ef18b6f --- /dev/null +++ b/docs/research/issue-1415-session-response-body-dedup.md @@ -0,0 +1,247 @@ +# Issue #1415 Session Response Body Redis 去重设计 + +## 结论 + +Issue [#1415](https://github.com/ding113/claude-code-hub/issues/1415) 需要解决的是同一 +`request sequence` 内的物理复制, 不需要跨 session 或跨租户共享正文. 最小且完整的方案是: + +1. 先按 `STORE_SESSION_MESSAGES` 生成最终可存储正文, 再做精确字符串去重和 UTF-8 字节计数. +2. 将 legacy, before, after 三个逻辑视图存入一个 request-scoped Redis Hash. +3. 用 Lua 原子替换 Hash 和三个旧正文 key 组成的完整 generation. +4. 新 reader 只在 Hash 不存在或明确标记为 `layout=legacy` 时读取旧 key; dedup Hash 存在但 + 正文缺失时不得回退旧值. +5. 使用 reader-first 两阶段发布. 第一阶段保持旧 writer, 第二阶段才启用去重 writer. + +该方案把共享域限制在一个 request, 不引入跨请求 refcount, GC, TTL 延长或内容相等性泄漏. + +## 已确认的现状 + +修复前, `ProxyResponseHandler` 的 4 个终态路径分别写入 legacy response, before body 和 +after body. `SessionManager.storeSessionResponse()` 与 +`SessionManager.storeSessionResponsePhaseSnapshot()` 对每份正文独立执行 `SETEX`. + +读取同样分成 3 条路径: + +- `SessionManager.getSessionResponse()` 读取 scoped legacy response. +- `SessionManager.getSessionResponsePhaseSnapshot(..., "before")` 读取 before body. +- `SessionManager.getSessionResponsePhaseSnapshot(..., "after")` 读取 after body. + +Dashboard action 会同时读取这 3 个视图. headers 和 meta 使用独立 key, 因此正文缺失不应影响 +状态码, URL 和阶段诊断信息. 这些事实可从以下一手源码核对: + +- `src/app/v1/_lib/proxy/response-handler.ts` +- `src/actions/active-sessions.ts` +- `src/actions/session-response.ts` +- `src/lib/session-manager.ts` + +Issue #1408 的机制复现记录了 64 个约 7.5 MiB SSE response 在旧布局下形成约 1.4 GiB 原始正文, +并在 RDB 窗口把 Redis peak 推到约 1.5 GiB. 原始调查见 +`docs/troubleshooting/issue-1408-replay-oom.md`. + +## 约束冲突与取舍 + +### 当前旧 reader 无法直接读取引用 + +当前旧 reader 对三个 key 执行普通 `GET`. Redis 不会让多个 key 自动共享一个任意字符串 value. +如果新 writer 把旧 key 改写为引用描述符, 旧 reader 会把描述符当正文; 如果新 writer 省略重复 +key, 旧 reader 会返回 `null`. + +因此, 以下三个目标不能在单阶段滚动中同时成立: + +- 当前旧 reader 无需升级即可读取新布局. +- 新 writer 只保存一份物理正文. +- 不改变三个逻辑视图的返回值. + +继续双写三份旧正文虽然兼容旧 reader, 但会直接保留 #1415 要消除的放大器. 本实现采用 +reader-first 发布, 将可回滚下限提升到已经理解新布局的版本. + +### 三份不同正文不可能无损装入 5 MiB 总预算 + +任意三份互不相同且各为 5 MiB 的正文至少需要 15 MiB 原始值. 不存在能对任意输入保证压缩到 +5 MiB 的无损布局. 因此 `SESSION_RESPONSE_BODY_MAX_BYTES` 在去重模式中的契约是: + +- 去重后的唯一正文总字节数不超预算时, 三个视图分别可读. +- 唯一正文总字节数超预算时, 写入 authoritative empty marker, 不保留部分正文. +- headers 和 meta 继续独立保存. + +authoritative marker 防止更新或重试超限后错误回退到上一 generation 的旧正文. + +## 数据布局 + +每个 `(sessionId, requestSequence)` 使用一个 Hash: + +```text +session:{sessionId}:req:{sequence}:response-bodies:v1 +``` + +每次 writer 还会把 bundle key 登记到按正文 TTL 剪枝的 session-level ZSET: + +```text +session:{sessionId}:response-body-bundles:v1 +``` + +请求取得 sequence 时, 还会把当前 session-level response body generation 复制到 request-scoped +generation key. full `terminateSession()` 通过一个 Lua 命令推进 generation, 删除索引中登记的 +bundle 及其可能对应的三份 legacy value, 并清理无 sequence 兼容路径固定使用的正文 key. 因此旧 +request 的迟到 writer 无法在 termination 后重建正文. provider-scoped termination 不推进该 +generation, 也不删除共享 Session artifact. + +字段如下: + +```text +schema=1 +layout=dedup +total_bytes=<去重后唯一正文 UTF-8 总字节数> +over_budget=0|1 +present:legacy=1 +present:before=1 +present:after=1 +ref:legacy=0 +ref:before=0 +ref:after=1 +body:0=<最终可存储正文> +body:1=<另一份不同正文> +``` + +`present:*` 区分"该视图原本存在但正文因总预算缺失"与"调用方没有提供该视图". +`ref:*` 只指向同一 Hash 内的 `body:*`, 不使用正文 hash, 因此没有 hash collision 或跨 request +内容相等性 side channel. + +## 写入与读取协议 + +`SessionManager.storeSessionResponseBodySet()` 接收完整的 legacy/before/after 集合. 生产终态路径 +不再发起三个独立 body writer, 从而避免异步到达顺序把不同 retry generation 混在一起. + +dedup 写入 Lua 同时声明 Hash, 三个旧正文 key 和 session-level bundle index: + +1. 校验 request owner 和 request/session response body generation 一致. +2. 原子删除上一 generation 的 Hash 和旧正文 key. +3. 写入固定字段, present/ref 和唯一 body. +4. 对整个 Hash 和 generation marker 执行 `EXPIRE`. +5. 按 Redis `TIME` 的毫秒过期分数剪枝 ZSET, 登记当前 bundle, 并刷新 index TTL. + +flag 关闭时, writer 在同一脚本内写三个旧正文 key, 并将 Hash 写为小型 +`schema=1, layout=legacy` generation marker. 新 reader 看到该 marker 后读取旧 key; 完全不理解 +Hash 的旧 reader 仍直接读取旧 key. flag 开启和关闭的 writer 使用同一组 `KEYS`, 因此 retry, +回滚和异构 writer 并发时, Redis 只会暴露最后一个完整执行的 generation. + +读 Lua 同时声明 Hash 和目标 view 的 legacy key, 在同一 Redis 命令中读取 +layout/present/ref/body 或旧正文. 因此 writer 切换 generation 时, reader 不会在看到 legacy marker +后再读到已被下一 generation 删除的旧 key. dedup Hash 与内部引用同时过期, 不会产生独立 +blob/ref 的悬空或 orphan 状态. 旧正文清理不再是 best-effort 后置命令, 因此 dedup generation +成功时不会同时残留三份旧正文. + +实际代理路径始终携带 `keyId` 和有效 request sequence, 因而启用 owner/generation fence. 缺失 +sequence 的无 owner 兼容调用仍按旧 key 语义写入; full termination 会原子删除其 legacy response +和 sequence 1 的 before/after body. 携带 `keyId` 却缺失有效 sequence 的调用 fail closed, 避免 +绕过 termination fence. + +当前 Redis client 使用 standalone `ioredis`, 不是 Redis Cluster. 跨 bundle, index 和三个旧 key 的 +原子脚本依赖这一现有部署契约; 若未来引入 Redis Cluster, 需要先统一 key hash tag 再迁移. + +## 隐私与预算顺序 + +正文处理顺序固定为: + +```text +原始 view + -> STORE_SESSION_MESSAGES redaction + -> 最终存储字符串 + -> 精确字符串去重 + -> Buffer.byteLength(..., "utf8") 聚合计数 + -> Redis Hash +``` + +原始敏感正文不会参与共享 key 派生, 也不会进入全局或 tenant-wide content hash. +`STORE_SESSION_RESPONSE_BODY=false` 在任何 Redis body 写入前直接返回. + +脱敏能力范围保持既有契约: 可解析 JSON 通过 `redactResponseBody()`, 非 JSON/SSE 仍按原样存储. +Issue #1415 不改变该诊断语义; request-scoped Hash 不跨 session, sequence 或租户共享物理 value, 因而不会 +把原样 SSE 扩散到新的共享域. + +## 发布流程 + +新增 `SESSION_RESPONSE_BODY_DEDUP_ENABLED`, 默认 `false`. + +1. Release A: 所有实例部署新 reader, flag 保持 `false`, writer 原子写旧 key 和 + `layout=legacy` marker. +2. 确认所有运行实例和回滚版本都至少为 Release A. +3. Release B: 设置 flag 为 `true`, writer 切换到单 Hash 布局. +4. 至少等待一个 `SESSION_TTL` 后, 才能考虑移除旧 key fallback. + +Release B 可以回滚到 Release A: false writer 会原子取代已有 dedup generation, 新 reader 随即 +读取本次 legacy generation. 不能回滚到完全不理解 Hash 的 Release A 之前版本. 这是物理去重和 +旧 reader 语义之间的结构性边界, 不是通过额外双写可以消除的实现细节. + +## 验证矩阵 + +自动化测试覆盖: + +- 三视图全同, 三种两两相同, 三者不同. +- 脱敏后相同与脱敏后不同. +- UTF-8 精确边界, 重复正文只计一次, 唯一正文聚合超限. +- authoritative marker 不回退 stale legacy key. +- v1 key TTL 窗口读取兼容. +- 同 sequence 覆盖, dedup/legacy 切换和异构 writer 并发 generation 原子性. +- 单 Hash TTL, ref/body 同时过期, 无悬空引用. +- rollout flag 关闭时继续旧布局. +- `STORE_SESSION_RESPONSE_BODY=false` 完全跳过正文. + +负载验收复用 `tests/load/issue-1408-replay-oom/`, 使用精确 5,242,880-byte SSE response 和 +8 x 8 request waves. `inspect-redis.cjs` 按 manifest 对每个 request 记录 `HSTRLEN`, refs, +`total_bytes`, TTL, 旧 key, `used_memory_peak`, RDB 状态和 Redis 容器 OOM/exit 状态. active +artifact 验证正文预算, expired artifact 按同一 manifest 自动验证 Hash 与旧正文 key 均已清理. + +## 2026-08-12 完整 5 MiB 负载验收 + +正式存储验收使用隔离的 `cch1415-*` Docker daemon, PostgreSQL, Redis 7.4.10 和本仓库构建的 +应用镜像. mock 使用 `CCH_MOCK_RESPONSE_MODE=complete`, 将 `response.completed` 终态事件计入 +每个 response 精确 5,242,880-byte SSE wire body. driver 使用 `CCH_REQUEST_MODE=complete` 等待 +每个客户端 response 结束. 这与 #1408 的默认 `disconnect` 挂起流复现互补, 不能将两者的 Node +内存结论混为一谈. + +工作负载为 8 x 8 requests, `SESSION_TTL=300`, +`SESSION_RESPONSE_BODY_DEDUP_ENABLED=true`, `SESSION_RESPONSE_BODY_MAX_BYTES=5242880`. +所有 64 个 response 完成后等待 70 秒, 触发并等待 Redis `BGSAVE`, 随后在 TTL 窗口内检查 active +artifact, 最后按相同 manifest 轮询 expiry. 运行前后宿主均为 `d_state=0`; IO PSI `avg10` 均为 +`some=0.00`, `full=0.00`. + +| 检查项 | 实际结果 | +| --- | --- | +| 已完成 response / bundles | 64 / 64 | +| 原始 body 预算 | 335,544,320 bytes = 64 x 5,242,880 bytes | +| `totalRawBodyBytes` / `totalDeclaredBytes` | 335,544,320 / 335,544,320 bytes | +| body fields / identical three-view refs | 64 / 64 | +| 旧 response body keys / dangling refs | 0 / 0 | +| Redis `used_memory` / `used_memory_peak` at active check | 412,133,448 / 437,018,776 bytes | +| Redis BGSAVE | `rdb_saves=1`, `rdb_last_bgsave_status=ok`, `rdb_last_cow_size=2,330,624` bytes | +| Redis container | running, `OOMKilled=false`, `ExitCode=0` | +| App container peak / final cgroup memory | 417,484,800 / 245,481,472 bytes | +| Redis container peak / final cgroup memory | 376,094,720 / 55,201,792 bytes | +| Expired artifact | 0 manifest bundles, 0 legacy response body keys, passed | + +`reports/issue-1415-20260812a.samples.txt.redis.json` and +`reports/issue-1415-20260812a.samples.txt.redis-expired.json` are machine-local artifacts for this +run. The implementation does not commit generated multi-MiB load artifacts; the table above records +the acceptance evidence necessary to reproduce and review the result. + +## 未采用的方案 + +- 在三个旧 key 中写 ref marker: 旧 reader 会把 marker 当正文. +- 同时写新 Hash 和三份旧正文: 不满足内存上界. +- tenant-wide 或 global content-addressed blob: 引入 refcount, TTL/GC 和跨请求泄漏面. +- 三个独立 writer 增量维护 refcount: retry 和异步乱序可生成混合 generation. +- 独立 blob/ref keys: 需要额外事务和 GC 才能避免 TTL 错位与悬空引用. +- 压缩或 delta: 无法对任意不同正文保证固定总上界, 并显著增加读取迁移复杂度. + +## 一手来源 + +- [GitHub Issue #1415](https://github.com/ding113/claude-code-hub/issues/1415) +- [GitHub Issue #1408](https://github.com/ding113/claude-code-hub/issues/1408) +- [GitHub PR #1414](https://github.com/ding113/claude-code-hub/pull/1414) +- `src/lib/session-manager.ts` +- `src/app/v1/_lib/proxy/response-handler.ts` +- `src/app/v1/_lib/proxy/warmup-guard.ts` +- `src/actions/active-sessions.ts` +- `src/actions/session-response.ts` +- `tests/load/issue-1408-replay-oom/` diff --git a/docs/troubleshooting/issue-1408-replay-oom.md b/docs/troubleshooting/issue-1408-replay-oom.md index 57f9b03ff..776422bfa 100644 --- a/docs/troubleshooting/issue-1408-replay-oom.md +++ b/docs/troubleshooting/issue-1408-replay-oom.md @@ -259,6 +259,40 @@ container running, oom=false, exit=0 并已跨过 `save 300 100` 的 RDB fork 点。当前产品默认值为 5 MiB；1 MiB 到 5 MiB 正文仍可能 形成三份 Redis value，该放大边界及 5 MiB 负载/RDB 验证由 #1415 跟踪，不属于上述实验已证明的范围。 +## Issue #1415 5 MiB 去重后续验收 + +Issue #1415 已将同一 `(sessionId, requestSequence)` 的 legacy, before 和 after response body +改为一个 request-scoped Redis Hash 中的 `body:*` 字段和 view refs. 这项后续验收使用同一仓库的 +fixture, 但选择 complete-response 模式而非本报告用于复现 Replay 生命周期的 disconnect 模式: + +```text +CCH_MOCK_RESPONSE_BYTES=5242880 +CCH_MOCK_RESPONSE_MODE=complete +CCH_REQUEST_MODE=complete +CCH_WAVES=8 +CCH_REQUESTS_PER_WAVE=8 +SESSION_RESPONSE_BODY_DEDUP_ENABLED=true +SESSION_RESPONSE_BODY_MAX_BYTES=5242880 +SESSION_TTL=300 +``` + +mock 将 terminal `response.completed` 也计入每个精确 5 MiB SSE body, driver 等待所有 64 个 +response 完成. 运行在隔离 Redis 7.4.10 上通过 BGSAVE 和 TTL cleanup 验证: + +| 项目 | 结果 | +| --- | --- | +| 64 request response body 原始总量 | 335,544,320 bytes, 等于 `64 x 5,242,880` | +| session body bundles / `body:*` fields / identical refs | 64 / 64 / 64 | +| stale legacy response body keys / dangling refs | 0 / 0 | +| Redis `used_memory_peak` | 437,018,776 bytes | +| RDB / Redis process | `rdb_last_bgsave_status=ok`; running; `OOMKilled=false`; `ExitCode=0` | +| TTL cleanup | manifest 的 64 个 bundle 和所有 legacy response body key 均为 0 | + +运行前后宿主观测均为 `d_state=0`, IO PSI `avg10 some=0.00` 和 `full=0.00`. 该结果证明 5 MiB +完整 response 的 session body Redis 原始值预算不再随三个视图线性放大. 它不替代本报告关于 +client-disconnect Replay drain 的机制结论, 因为该验收刻意让 response 正常终止以覆盖 session +response body 持久化路径。 + ## 测试与证据边界 focused 回归共 5 个文件、104 个测试，覆盖： diff --git a/src/app/v1/_lib/proxy/response-handler.ts b/src/app/v1/_lib/proxy/response-handler.ts index de8ad5d62..d06c75c65 100644 --- a/src/app/v1/_lib/proxy/response-handler.ts +++ b/src/app/v1/_lib/proxy/response-handler.ts @@ -2491,7 +2491,7 @@ export class ProxyResponseHandler { let finalResponse = response; let finalResponseBodyForSnapshot: string | null = null; let responseTransformFailed = false; - const persistNonStreamAfterSnapshot = (targetResponse: Response, body: string) => { + const persistNonStreamAfterSnapshot = (targetResponse: Response) => { if (!session.sessionId || !session.shouldPersistSessionDebugArtifacts()) { return; } @@ -2500,7 +2500,6 @@ export class ProxyResponseHandler { session.sessionId, "after", { - body, headers: targetResponse.headers, meta: { upstreamUrl: null, @@ -2560,35 +2559,13 @@ export class ProxyResponseHandler { // 存储响应体到 Redis（5分钟过期） if (session.sessionId && session.shouldPersistSessionDebugArtifacts()) { const beforeBody = (await consumeBeforeResponseBodySnapshot(session)) ?? responseText; - void SessionManager.storeSessionResponse( + void SessionManager.storeSessionResponseBodySet( session.sessionId, - responseText, + { legacy: responseText, before: beforeBody, after: responseText }, session.requestSequence, getSessionRequestOwnerKeyId(session) ).catch((err) => { - logger.error("[ResponseHandler] Failed to store response:", err); - }); - - const responseBeforeSnapshotTask = SessionManager.storeSessionResponsePhaseSnapshot?.( - session.sessionId, - "before", - { body: beforeBody }, - session.requestSequence, - getSessionRequestOwnerKeyId(session) - ); - responseBeforeSnapshotTask?.catch((err) => { - logger.error("[ResponseHandler] Failed to store response before snapshot:", err); - }); - - const responseAfterSnapshotTask = SessionManager.storeSessionResponsePhaseSnapshot?.( - session.sessionId, - "after", - { body: responseText }, - session.requestSequence, - getSessionRequestOwnerKeyId(session) - ); - responseAfterSnapshotTask?.catch((err) => { - logger.error("[ResponseHandler] Failed to store response after snapshot:", err); + logger.error("[ResponseHandler] Failed to store response body set:", err); }); } @@ -3137,29 +3114,22 @@ export class ProxyResponseHandler { // 存储响应体到 Redis（5分钟过期） if (session.sessionId && session.shouldPersistSessionDebugArtifacts()) { const beforeBody = (await consumeBeforeResponseBodySnapshot(session)) ?? responseText; - void SessionManager.storeSessionResponse( + void SessionManager.storeSessionResponseBodySet( session.sessionId, - responseText, + { + legacy: responseText, + before: beforeBody, + after: clientVisibleResponseText, + }, session.requestSequence, getSessionRequestOwnerKeyId(session) ).catch((err) => { - logger.error("[ResponseHandler] Failed to store response:", err); - }); - - const responseBeforeSnapshotTask = SessionManager.storeSessionResponsePhaseSnapshot?.( - session.sessionId, - "before", - { body: beforeBody }, - session.requestSequence, - getSessionRequestOwnerKeyId(session) - ); - responseBeforeSnapshotTask?.catch((err) => { - logger.error("[ResponseHandler] Failed to store response before snapshot:", err); + logger.error("[ResponseHandler] Failed to store response body set:", err); }); // after 快照复用本任务已经读取到的响应文本，避免再启动一个未受 // AsyncTaskManager 管理的 clone().text() 读取分支。 - persistNonStreamAfterSnapshot(finalResponse, clientVisibleResponseText); + persistNonStreamAfterSnapshot(finalResponse); } if (billableUsageMetrics && messageContext) { @@ -3847,35 +3817,13 @@ export class ProxyResponseHandler { !streamSnapshot?.truncated && session.shouldPersistSessionDebugArtifacts() ) { - void SessionManager.storeSessionResponse( + void SessionManager.storeSessionResponseBodySet( session.sessionId, - allContent, + { legacy: allContent, before: allContent, after: allContent }, session.requestSequence, getSessionRequestOwnerKeyId(session) ).catch((err) => { - logger.error("[ResponseHandler] Failed to store stream passthrough response:", err); - }); - - const responseBeforeSnapshotTask = SessionManager.storeSessionResponsePhaseSnapshot?.( - session.sessionId, - "before", - { body: allContent }, - session.requestSequence, - getSessionRequestOwnerKeyId(session) - ); - responseBeforeSnapshotTask?.catch((err) => { - logger.error("[ResponseHandler] Failed to store response before snapshot:", err); - }); - - const responseAfterSnapshotTask = SessionManager.storeSessionResponsePhaseSnapshot?.( - session.sessionId, - "after", - { body: allContent }, - session.requestSequence, - getSessionRequestOwnerKeyId(session) - ); - responseAfterSnapshotTask?.catch((err) => { - logger.error("[ResponseHandler] Failed to store response after snapshot:", err); + logger.error("[ResponseHandler] Failed to store response body set:", err); }); } else if (session.sessionId && streamSnapshot?.truncated) { logger.warn("[ResponseHandler] Skip storing passthrough response: body too large", { @@ -4423,35 +4371,13 @@ export class ProxyResponseHandler { !streamSnapshot?.truncated ) { const beforeBody = allContent; - void SessionManager.storeSessionResponse( + void SessionManager.storeSessionResponseBodySet( session.sessionId, - allContent, + { legacy: allContent, before: beforeBody, after: allContent }, session.requestSequence, getSessionRequestOwnerKeyId(session) ).catch((err) => { - logger.error("[ResponseHandler] Failed to store response:", err); - }); - - const responseAfterSnapshotTask = SessionManager.storeSessionResponsePhaseSnapshot?.( - session.sessionId, - "after", - { body: allContent }, - session.requestSequence, - getSessionRequestOwnerKeyId(session) - ); - responseAfterSnapshotTask?.catch((err) => { - logger.error("[ResponseHandler] Failed to store response after snapshot:", err); - }); - - const responseBeforeSnapshotTask = SessionManager.storeSessionResponsePhaseSnapshot?.( - session.sessionId, - "before", - { body: beforeBody }, - session.requestSequence, - getSessionRequestOwnerKeyId(session) - ); - responseBeforeSnapshotTask?.catch((err) => { - logger.error("[ResponseHandler] Failed to store response before snapshot:", err); + logger.error("[ResponseHandler] Failed to store response body set:", err); }); } else if (session.sessionId && streamSnapshot?.truncated) { discardBeforeResponseBodySnapshot(session); diff --git a/src/app/v1/_lib/proxy/warmup-guard.ts b/src/app/v1/_lib/proxy/warmup-guard.ts index 208fa4aa8..622473b13 100644 --- a/src/app/v1/_lib/proxy/warmup-guard.ts +++ b/src/app/v1/_lib/proxy/warmup-guard.ts @@ -48,7 +48,12 @@ export class ProxyWarmupGuard { if (session.sessionId && session.shouldPersistSessionDebugArtifacts()) { const seq = session.getRequestSequence(); await Promise.allSettled([ - SessionManager.storeSessionResponse(session.sessionId, responseText, seq, authState.key.id), + SessionManager.storeSessionResponseBodySet( + session.sessionId, + { legacy: responseText }, + seq, + authState.key.id + ), SessionManager.storeSessionResponseHeaders( session.sessionId, responseHeaders, diff --git a/src/lib/config/env.schema.ts b/src/lib/config/env.schema.ts index 2cba1650e..2eb8a6097 100644 --- a/src/lib/config/env.schema.ts +++ b/src/lib/config/env.schema.ts @@ -157,7 +157,9 @@ 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 都受此边界约束。 + // 两阶段发布开关。false 时保持旧 key 写入，所有实例升级后再启用单 key 去重布局。 + SESSION_RESPONSE_BODY_DEDUP_ENABLED: z.string().default("false").transform(booleanTransform), + // 会话响应正文写入 Redis 的字节上限。旧布局按单份限制，去重布局按唯一正文总字节限制。 SESSION_RESPONSE_BODY_MAX_BYTES: z.coerce .number() .int() diff --git a/src/lib/session-manager-detail-snapshots.test.ts b/src/lib/session-manager-detail-snapshots.test.ts index 323e847f7..a659bc9cd 100644 --- a/src/lib/session-manager-detail-snapshots.test.ts +++ b/src/lib/session-manager-detail-snapshots.test.ts @@ -49,7 +49,12 @@ const redisMock = { set: vi.fn().mockResolvedValue("OK"), expire: vi.fn().mockResolvedValue(1), incr: vi.fn().mockResolvedValue(1), - eval: vi.fn().mockResolvedValue(1), + eval: vi.fn((script: string, keyCount: number, ...rawArgs: Array) => { + if (!script.includes("cch:session-response-bundle:read:v1")) return Promise.resolve(1); + const keys = rawArgs.slice(0, keyCount).map(String); + const body = redisStore.get(keys[1]); + return Promise.resolve([0, body === undefined ? 0 : 1, body ?? null]); + }), pipeline: vi.fn(() => redisPipeline), }; @@ -89,8 +94,9 @@ describe("SessionManager detail snapshots", () => { expect(redisMock.eval).toHaveBeenCalledWith( expect.stringContaining("redis.call('PERSIST', KEYS[1])"), - 1, + 2, "session:sess_owner:seq", + "session:sess_owner:response-body-generation:v1", "session:sess_owner:req:", "300", "42" diff --git a/src/lib/session-manager-response-body-dedup.test.ts b/src/lib/session-manager-response-body-dedup.test.ts new file mode 100644 index 000000000..18ced9821 --- /dev/null +++ b/src/lib/session-manager-response-body-dedup.test.ts @@ -0,0 +1,543 @@ +import { Buffer } from "node:buffer"; +import { beforeEach, describe, expect, it, vi } from "vitest"; + +vi.mock("server-only", () => ({})); + +const loggerMock = { + trace: vi.fn(), + debug: vi.fn(), + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), +}; + +vi.mock("@/lib/logger", () => ({ logger: loggerMock })); +vi.mock("@/app/v1/_lib/proxy/errors", () => ({ + sanitizeHeaders: vi.fn(() => ""), + sanitizeUrl: vi.fn((url: unknown) => String(url)), +})); + +const stringStore = new Map(); +const hashStore = new Map>(); +const sortedSetStore = new Map>(); +const ttlStore = new Map(); +let afterBundleRead: (() => void) | null = null; + +function responseBundle(key: string): Map | undefined { + return hashStore.get(key); +} + +function bodyFields(key: string): string[] { + return [...(responseBundle(key)?.entries() ?? [])] + .filter(([field]) => field.startsWith("body:")) + .map(([, value]) => value); +} + +const redisMock = { + status: "ready", + setex: vi.fn((key: string, ttl: number, value: string) => { + stringStore.set(key, value); + ttlStore.set(key, ttl); + return Promise.resolve("OK"); + }), + get: vi.fn((key: string) => Promise.resolve(stringStore.get(key) ?? null)), + del: vi.fn((...keys: string[]) => { + let removed = 0; + for (const key of keys) { + removed += Number(stringStore.delete(key)); + removed += Number(hashStore.delete(key)); + removed += Number(sortedSetStore.delete(key)); + ttlStore.delete(key); + } + return Promise.resolve(removed); + }), + set: vi.fn().mockResolvedValue("OK"), + expire: vi.fn().mockResolvedValue(1), + incr: vi.fn().mockResolvedValue(1), + pipeline: vi.fn(() => ({ + setex: vi.fn().mockReturnThis(), + hset: vi.fn().mockReturnThis(), + expire: vi.fn().mockReturnThis(), + del: vi.fn().mockReturnThis(), + exec: vi.fn().mockResolvedValue([]), + })), + eval: vi.fn(async (script: string, keyCount: number, ...rawArgs: Array) => { + const keys = rawArgs.slice(0, keyCount).map(String); + const args = rawArgs.slice(keyCount); + const key = keys[0]; + if (script.includes("cch:session-response-bundle:write:v1")) { + const [ + ttl, + expectedKeyId, + totalBytes, + overBudget, + legacyPresent, + beforePresent, + afterPresent, + legacyRef, + beforeRef, + afterRef, + ...bodies + ] = args.map(String); + const currentGeneration = stringStore.get(keys[5]); + const requestGeneration = stringStore.get(keys[6]); + if ( + expectedKeyId && + (stringStore.get(keys[7]) !== expectedKeyId || + currentGeneration === undefined || + requestGeneration === undefined || + currentGeneration !== requestGeneration) + ) { + return 0; + } + for (const currentKey of keys.slice(0, 4)) { + stringStore.delete(currentKey); + hashStore.delete(currentKey); + ttlStore.delete(currentKey); + } + const value = new Map([ + ["schema", "1"], + ["layout", "dedup"], + ["total_bytes", totalBytes], + ["over_budget", overBudget], + ]); + for (const [view, present, ref] of [ + ["legacy", legacyPresent, legacyRef], + ["before", beforePresent, beforeRef], + ["after", afterPresent, afterRef], + ] as const) { + if (present === "1") value.set(`present:${view}`, "1"); + if (ref) value.set(`ref:${view}`, ref); + } + bodies.forEach((body, index) => value.set(`body:${index}`, body)); + hashStore.set(key, value); + ttlStore.set(key, Number(ttl)); + const bundleIndex = sortedSetStore.get(keys[4]) ?? new Set(); + bundleIndex.add(key); + sortedSetStore.set(keys[4], bundleIndex); + ttlStore.set(keys[4], Number(ttl)); + return 1; + } + + if (script.includes("cch:session-response-bundle:write-legacy:v1")) { + const [ttl, expectedKeyId, ...viewArgs] = args.map(String); + const currentGeneration = stringStore.get(keys[5]); + const requestGeneration = stringStore.get(keys[6]); + if ( + expectedKeyId && + (stringStore.get(keys[7]) !== expectedKeyId || + currentGeneration === undefined || + requestGeneration === undefined || + currentGeneration !== requestGeneration) + ) { + return 0; + } + for (const currentKey of keys.slice(0, 4)) { + stringStore.delete(currentKey); + hashStore.delete(currentKey); + ttlStore.delete(currentKey); + } + + const value = new Map([ + ["schema", "1"], + ["layout", "legacy"], + ]); + for (let index = 0; index < 3; index += 1) { + const view = ["legacy", "before", "after"][index]; + const offset = index * 3; + if (viewArgs[offset] === "1") value.set(`present:${view}`, "1"); + if (viewArgs[offset + 1] === "1") { + stringStore.set(keys[index + 1], viewArgs[offset + 2]); + ttlStore.set(keys[index + 1], Number(ttl)); + } + } + hashStore.set(key, value); + ttlStore.set(key, Number(ttl)); + const bundleIndex = sortedSetStore.get(keys[4]) ?? new Set(); + bundleIndex.add(key); + sortedSetStore.set(keys[4], bundleIndex); + ttlStore.set(keys[4], Number(ttl)); + return 1; + } + + if (script.includes("cch:session-response-bundle:read:v1")) { + const value = hashStore.get(key); + const legacyBody = stringStore.get(keys[1]); + let result: [number, number, string | null]; + if (!value) { + result = [0, legacyBody === undefined ? 0 : 1, legacyBody ?? null]; + } else { + const present = value.get(`present:${String(args[0])}`) === "1"; + if (value.get("layout") === "legacy") { + result = [1, present ? 1 : 0, legacyBody ?? null]; + } else { + const ref = value.get(`ref:${String(args[0])}`); + result = [ + 1, + present ? 1 : 0, + ref === undefined ? null : (value.get(`body:${ref}`) ?? null), + ]; + } + } + const callback = afterBundleRead; + afterBundleRead = null; + callback?.(); + return result; + } + + throw new Error("unexpected Redis script"); + }), +}; + +vi.mock("@/lib/redis", () => ({ getRedisClient: () => redisMock })); + +let mockStoreMessages = true; +let mockStoreSessionResponseBody = true; +let mockResponseBodyDedupEnabled = true; +let mockSessionResponseBodyMaxBytes = 1024; + +vi.mock("@/lib/config/env.schema", () => ({ + getEnvConfig: () => ({ + STORE_SESSION_MESSAGES: mockStoreMessages, + STORE_SESSION_RESPONSE_BODY: mockStoreSessionResponseBody, + SESSION_RESPONSE_BODY_DEDUP_ENABLED: mockResponseBodyDedupEnabled, + SESSION_RESPONSE_BODY_MAX_BYTES: mockSessionResponseBodyMaxBytes, + SESSION_TTL: 300, + }), +})); + +const { SessionManager } = await import("@/lib/session-manager"); + +describe("SessionManager response body deduplication", () => { + beforeEach(() => { + vi.clearAllMocks(); + stringStore.clear(); + hashStore.clear(); + sortedSetStore.clear(); + ttlStore.clear(); + redisMock.status = "ready"; + mockStoreMessages = true; + mockStoreSessionResponseBody = true; + mockResponseBodyDedupEnabled = true; + mockSessionResponseBodyMaxBytes = 1024; + afterBundleRead = null; + }); + + it.each([ + ["all views are identical", { legacy: "same", before: "same", after: "same" }, 1], + ["legacy and before are identical", { legacy: "same", before: "same", after: "after" }, 2], + ["legacy and after are identical", { legacy: "same", before: "before", after: "same" }, 2], + ["before and after are identical", { legacy: "legacy", before: "same", after: "same" }, 2], + ["all views are different", { legacy: "legacy", before: "before", after: "after" }, 3], + ])("stores one physical bundle when %s", async (_name, bodies, expectedUniqueBodies) => { + await SessionManager.storeSessionResponseBodySet("sess_views", bodies, 1); + + const key = "session:sess_views:req:1:response-bodies:v1"; + expect(hashStore.size).toBe(1); + expect(sortedSetStore.get("session:sess_views:response-body-bundles:v1")).toEqual( + new Set([key]) + ); + expect(bodyFields(key)).toHaveLength(expectedUniqueBodies); + expect(stringStore.size).toBe(0); + await expect(SessionManager.getSessionResponse("sess_views", 1)).resolves.toBe(bodies.legacy); + await expect( + SessionManager.getSessionResponsePhaseSnapshot("sess_views", "before", 1) + ).resolves.toEqual({ + body: bodies.before, + headers: null, + meta: { upstreamUrl: null, statusCode: null }, + }); + await expect( + SessionManager.getSessionResponsePhaseSnapshot("sess_views", "after", 1) + ).resolves.toEqual({ + body: bodies.after, + headers: null, + meta: { upstreamUrl: null, statusCode: null }, + }); + }); + + it("deduplicates bodies only after redaction", async () => { + mockStoreMessages = false; + const legacy = JSON.stringify({ id: "same", content: [{ type: "text", text: "secret-a" }] }); + const before = JSON.stringify({ id: "same", content: [{ type: "text", text: "secret-b" }] }); + const after = JSON.stringify({ + id: "different", + content: [{ type: "text", text: "secret-c" }], + }); + + await SessionManager.storeSessionResponseBodySet("sess_redacted", { legacy, before, after }, 1); + + const bodies = bodyFields("session:sess_redacted:req:1:response-bodies:v1"); + expect(bodies).toHaveLength(2); + expect(bodies.some((body) => body.includes("secret-"))).toBe(false); + await expect(SessionManager.getSessionResponse("sess_redacted", 1)).resolves.toBe( + JSON.stringify({ id: "same", content: [{ type: "text", text: "[REDACTED]" }] }) + ); + }); + + it("keeps bodies distinct when their post-redaction values differ", async () => { + mockStoreMessages = false; + + await SessionManager.storeSessionResponseBodySet( + "sess_redacted_distinct", + { + legacy: JSON.stringify({ id: "one", content: [{ type: "text", text: "secret-a" }] }), + before: JSON.stringify({ id: "two", content: [{ type: "text", text: "secret-b" }] }), + after: JSON.stringify({ id: "three", content: [{ type: "text", text: "secret-c" }] }), + }, + 1 + ); + + expect(bodyFields("session:sess_redacted_distinct:req:1:response-bodies:v1")).toHaveLength(3); + }); + + it("does not share response body values across request-scoped bundles", async () => { + mockStoreMessages = false; + const body = JSON.stringify({ id: "same", content: [{ type: "text", text: "secret" }] }); + + await SessionManager.storeSessionResponseBodySet( + "sess_scope_one", + { legacy: body, before: body, after: body }, + 1 + ); + await SessionManager.storeSessionResponseBodySet( + "sess_scope_two", + { legacy: body, before: body, after: body }, + 1 + ); + + expect([...hashStore.keys()].sort()).toEqual([ + "session:sess_scope_one:req:1:response-bodies:v1", + "session:sess_scope_two:req:1:response-bodies:v1", + ]); + expect(bodyFields("session:sess_scope_one:req:1:response-bodies:v1")).toEqual([ + JSON.stringify({ id: "same", content: [{ type: "text", text: "[REDACTED]" }] }), + ]); + expect(bodyFields("session:sess_scope_two:req:1:response-bodies:v1")).toHaveLength(1); + }); + + it("applies the aggregate limit to distinct UTF-8 bytes", async () => { + mockSessionResponseBodyMaxBytes = 4; + + await SessionManager.storeSessionResponseBodySet( + "sess_utf8_exact", + { legacy: "中a", before: "中a", after: "中a" }, + 1 + ); + expect(bodyFields("session:sess_utf8_exact:req:1:response-bodies:v1")).toEqual(["中a"]); + + await SessionManager.storeSessionResponseBodySet( + "sess_utf8_over", + { legacy: "中ab", before: "中ab", after: "中ab" }, + 1 + ); + const overBudget = responseBundle("session:sess_utf8_over:req:1:response-bodies:v1"); + expect(overBudget?.get("over_budget")).toBe("1"); + expect(bodyFields("session:sess_utf8_over:req:1:response-bodies:v1")).toEqual([]); + await expect(SessionManager.getSessionResponse("sess_utf8_over", 1)).resolves.toBeNull(); + expect(loggerMock.warn).toHaveBeenCalledWith( + "SessionManager: Skipped response body bundle over aggregate limit", + { sessionId: "sess_utf8_over", requestSequence: 1, byteSize: 5, maxBytes: 4 } + ); + }); + + it("reads legacy keys when no response bundle exists", async () => { + stringStore.set("session:sess_legacy:req:1:response", "legacy response"); + stringStore.set("session:sess_legacy:req:1:snapshot:response:before:body", "legacy before"); + stringStore.set("session:sess_legacy:req:1:snapshot:response:after:body", "legacy after"); + + await expect(SessionManager.getSessionResponse("sess_legacy", 1)).resolves.toBe( + "legacy response" + ); + await expect( + SessionManager.getSessionResponsePhaseSnapshot("sess_legacy", "before", 1) + ).resolves.toMatchObject({ body: "legacy before" }); + await expect( + SessionManager.getSessionResponsePhaseSnapshot("sess_legacy", "after", 1) + ).resolves.toMatchObject({ body: "legacy after" }); + }); + + it("treats an over-budget bundle as authoritative over stale legacy keys", async () => { + mockSessionResponseBodyMaxBytes = 4; + stringStore.set("session:sess_authoritative:req:1:response", "stale"); + stringStore.set("session:sess_authoritative:req:1:snapshot:response:after:body", "stale"); + + await SessionManager.storeSessionResponseBodySet( + "sess_authoritative", + { legacy: "12345", before: "12345", after: "12345" }, + 1 + ); + + await expect(SessionManager.getSessionResponse("sess_authoritative", 1)).resolves.toBeNull(); + await expect( + SessionManager.getSessionResponsePhaseSnapshot("sess_authoritative", "after", 1) + ).resolves.toEqual({ + body: null, + headers: null, + meta: { upstreamUrl: null, statusCode: null }, + }); + }); + + it("atomically replaces a prior generation and refreshes bundle/index TTLs", async () => { + await SessionManager.storeSessionResponseBodySet( + "sess_retry", + { legacy: "first", before: "first", after: "first" }, + 1 + ); + await SessionManager.storeSessionResponseBodySet( + "sess_retry", + { legacy: "second", before: "before", after: "second" }, + 1 + ); + + const key = "session:sess_retry:req:1:response-bodies:v1"; + expect(bodyFields(key).sort()).toEqual(["before", "second"]); + expect(bodyFields(key)).not.toContain("first"); + expect(ttlStore).toEqual( + new Map([ + [key, 300], + ["session:sess_retry:response-body-bundles:v1", 300], + ]) + ); + await expect(SessionManager.getSessionResponse("sess_retry", 1)).resolves.toBe("second"); + await expect( + SessionManager.getSessionResponsePhaseSnapshot("sess_retry", "before", 1) + ).resolves.toMatchObject({ body: "before" }); + }); + + it("keeps legacy writes during the reader-first rollout phase", async () => { + mockResponseBodyDedupEnabled = false; + + await SessionManager.storeSessionResponseBodySet( + "sess_rollout", + { legacy: "same", before: "same", after: "same" }, + 1 + ); + + expect(responseBundle("session:sess_rollout:req:1:response-bodies:v1")).toEqual( + new Map([ + ["schema", "1"], + ["layout", "legacy"], + ["present:legacy", "1"], + ["present:before", "1"], + ["present:after", "1"], + ]) + ); + expect(stringStore.get("session:sess_rollout:req:1:response")).toBe("same"); + expect(stringStore.get("session:sess_rollout:req:1:snapshot:response:before:body")).toBe( + "same" + ); + expect(stringStore.get("session:sess_rollout:req:1:snapshot:response:after:body")).toBe("same"); + }); + + it("atomically switches between dedup and legacy writer generations", async () => { + await SessionManager.storeSessionResponseBodySet( + "sess_mixed_rollout", + { legacy: "dedup-first", before: "dedup-first", after: "dedup-first" }, + 1 + ); + + mockResponseBodyDedupEnabled = false; + await SessionManager.storeSessionResponseBodySet( + "sess_mixed_rollout", + { legacy: "legacy-second", before: "legacy-before", after: "legacy-second" }, + 1 + ); + + const key = "session:sess_mixed_rollout:req:1:response-bodies:v1"; + expect(responseBundle(key)?.get("layout")).toBe("legacy"); + expect(bodyFields(key)).toEqual([]); + await expect(SessionManager.getSessionResponse("sess_mixed_rollout", 1)).resolves.toBe( + "legacy-second" + ); + await expect( + SessionManager.getSessionResponsePhaseSnapshot("sess_mixed_rollout", "before", 1) + ).resolves.toMatchObject({ body: "legacy-before" }); + + mockResponseBodyDedupEnabled = true; + await SessionManager.storeSessionResponseBodySet( + "sess_mixed_rollout", + { legacy: "dedup-third", before: "dedup-third", after: "dedup-third" }, + 1 + ); + + expect(responseBundle(key)?.get("layout")).toBe("dedup"); + expect(bodyFields(key)).toEqual(["dedup-third"]); + expect(stringStore.size).toBe(0); + await expect(SessionManager.getSessionResponse("sess_mixed_rollout", 1)).resolves.toBe( + "dedup-third" + ); + }); + + it("returns one atomic legacy generation when a writer switches after the read", async () => { + mockResponseBodyDedupEnabled = false; + await SessionManager.storeSessionResponseBodySet( + "sess_atomic_read", + { legacy: "legacy generation", before: "legacy before", after: "legacy generation" }, + 1 + ); + + afterBundleRead = () => { + stringStore.clear(); + hashStore.set( + "session:sess_atomic_read:req:1:response-bodies:v1", + new Map([ + ["schema", "1"], + ["layout", "dedup"], + ["present:legacy", "1"], + ["ref:legacy", "0"], + ["body:0", "dedup generation"], + ]) + ); + }; + + await expect(SessionManager.getSessionResponse("sess_atomic_read", 1)).resolves.toBe( + "legacy generation" + ); + expect(stringStore.size).toBe(0); + }); + + it("does not create legacy or bundled bodies when response storage is disabled", async () => { + mockStoreSessionResponseBody = false; + + await SessionManager.storeSessionResponseBodySet( + "sess_disabled", + { legacy: "same", before: "same", after: "same" }, + 1 + ); + + expect(hashStore.size).toBe(0); + expect(stringStore.size).toBe(0); + expect(redisMock.eval).not.toHaveBeenCalled(); + }); + + it("rejects a late response writer from a terminated request generation", async () => { + stringStore.set("session:sess_fenced:response-body-generation:v1", "generation-new"); + stringStore.set("session:sess_fenced:req:1:response-body-generation:v1", "generation-old"); + + await SessionManager.storeSessionResponseBodySet( + "sess_fenced", + { legacy: "late", before: "late", after: "late" }, + 1, + 42 + ); + + expect(responseBundle("session:sess_fenced:req:1:response-bodies:v1")).toBeUndefined(); + expect(stringStore.get("session:sess_fenced:req:1:owner")).toBe("42"); + }); + + it("counts only unique body bytes against the aggregate budget", async () => { + mockSessionResponseBodyMaxBytes = 8; + + await SessionManager.storeSessionResponseBodySet( + "sess_budget", + { legacy: "1234", before: "1234", after: "5678" }, + 1 + ); + + const key = "session:sess_budget:req:1:response-bodies:v1"; + expect(bodyFields(key).reduce((sum, body) => sum + Buffer.byteLength(body), 0)).toBe(8); + expect(responseBundle(key)?.get("total_bytes")).toBe("8"); + expect(responseBundle(key)?.get("over_budget")).toBe("0"); + }); +}); diff --git a/src/lib/session-manager.ts b/src/lib/session-manager.ts index 6b5983420..116bd8afc 100644 --- a/src/lib/session-manager.ts +++ b/src/lib/session-manager.ts @@ -60,13 +60,176 @@ const RESERVED_INTERNAL_HEADER_SET = new Set( RESERVED_INTERNAL_HEADERS.map((header) => header.toLowerCase()) ); const DEFAULT_SESSION_RESPONSE_BODY_MAX_BYTES = 5 * 1024 * 1024; +const SESSION_RESPONSE_BODY_VIEWS = ["legacy", "before", "after"] as const; +const WRITE_SESSION_RESPONSE_BODY_BUNDLE_LUA = ` +-- cch:session-response-bundle:write:v1 +local expected_key_id = ARGV[2] +if expected_key_id ~= "" then + if redis.call("GET", KEYS[8]) ~= expected_key_id then + return 0 + end + local current_generation = redis.call("GET", KEYS[6]) + local request_generation = redis.call("GET", KEYS[7]) + if not current_generation or not request_generation or current_generation ~= request_generation then + return 0 + end +end + +redis.call("DEL", KEYS[1], KEYS[2], KEYS[3], KEYS[4]) +redis.call( + "HSET", + KEYS[1], + "schema", "1", + "layout", "dedup", + "total_bytes", ARGV[3], + "over_budget", ARGV[4] +) + +local views = { "legacy", "before", "after" } +for index, view in ipairs(views) do + if ARGV[index + 4] == "1" then + redis.call("HSET", KEYS[1], "present:" .. view, "1") + end + local ref = ARGV[index + 7] + if ref ~= "" then + redis.call("HSET", KEYS[1], "ref:" .. view, ref) + end +end + +for index = 11, #ARGV do + redis.call("HSET", KEYS[1], "body:" .. (index - 11), ARGV[index]) +end + +redis.call("EXPIRE", KEYS[1], ARGV[1]) +if expected_key_id ~= "" then + redis.call("EXPIRE", KEYS[6], ARGV[1]) + redis.call("EXPIRE", KEYS[7], ARGV[1]) +end +local redis_time = redis.call("TIME") +local now_ms = (tonumber(redis_time[1]) * 1000) + math.floor(tonumber(redis_time[2]) / 1000) +local expires_at_ms = now_ms + (tonumber(ARGV[1]) * 1000) +redis.call("ZREMRANGEBYSCORE", KEYS[5], "-inf", now_ms) +redis.call("ZADD", KEYS[5], expires_at_ms, KEYS[1]) +redis.call("EXPIRE", KEYS[5], ARGV[1]) +return 1 +`; +const WRITE_LEGACY_SESSION_RESPONSE_BODY_SET_LUA = ` +-- cch:session-response-bundle:write-legacy:v1 +local expected_key_id = ARGV[2] +if expected_key_id ~= "" then + if redis.call("GET", KEYS[8]) ~= expected_key_id then + return 0 + end + local current_generation = redis.call("GET", KEYS[6]) + local request_generation = redis.call("GET", KEYS[7]) + if not current_generation or not request_generation or current_generation ~= request_generation then + return 0 + end +end + +redis.call("DEL", KEYS[1], KEYS[2], KEYS[3], KEYS[4]) +redis.call("HSET", KEYS[1], "schema", "1", "layout", "legacy") + +local views = { "legacy", "before", "after" } +for index, view in ipairs(views) do + local offset = 3 + ((index - 1) * 3) + if ARGV[offset] == "1" then + redis.call("HSET", KEYS[1], "present:" .. view, "1") + end + if ARGV[offset + 1] == "1" then + redis.call("SETEX", KEYS[index + 1], ARGV[1], ARGV[offset + 2]) + end +end + +redis.call("EXPIRE", KEYS[1], ARGV[1]) +if expected_key_id ~= "" then + redis.call("EXPIRE", KEYS[6], ARGV[1]) + redis.call("EXPIRE", KEYS[7], ARGV[1]) +end +local redis_time = redis.call("TIME") +local now_ms = (tonumber(redis_time[1]) * 1000) + math.floor(tonumber(redis_time[2]) / 1000) +local expires_at_ms = now_ms + (tonumber(ARGV[1]) * 1000) +redis.call("ZREMRANGEBYSCORE", KEYS[5], "-inf", now_ms) +redis.call("ZADD", KEYS[5], expires_at_ms, KEYS[1]) +redis.call("EXPIRE", KEYS[5], ARGV[1]) +return 1 +`; +const DELETE_SESSION_RESPONSE_BODY_BUNDLES_LUA = ` +-- cch:session-response-bundle:delete-session:v1 +local bundle_keys = redis.call("ZRANGE", KEYS[1], 0, -1) +local deleted = redis.call("DEL", KEYS[1]) +redis.call("SETEX", KEYS[2], ARGV[1], ARGV[2]) +deleted = deleted + redis.call("DEL", KEYS[3], KEYS[4], KEYS[5]) + +local bundle_suffix = "response-bodies:v1" +for _, bundle_key in ipairs(bundle_keys) do + if string.sub(bundle_key, -string.len(bundle_suffix)) == bundle_suffix then + local request_prefix = string.sub(bundle_key, 1, string.len(bundle_key) - string.len(bundle_suffix)) + deleted = deleted + redis.call( + "DEL", + bundle_key, + request_prefix .. "response", + request_prefix .. "snapshot:response:before:body", + request_prefix .. "snapshot:response:after:body", + request_prefix .. "response-body-generation:v1" + ) + else + deleted = deleted + redis.call("DEL", bundle_key) + end +end +return deleted +`; +const READ_SESSION_RESPONSE_BODY_BUNDLE_LUA = ` +-- cch:session-response-bundle:read:v1 +if redis.call("EXISTS", KEYS[1]) == 0 then + local legacy = redis.call("GET", KEYS[2]) + return { 0, legacy and 1 or 0, legacy or false } +end + +local view = ARGV[1] +local present = redis.call("HGET", KEYS[1], "present:" .. view) +if redis.call("HGET", KEYS[1], "layout") == "legacy" then + return { 1, present and 1 or 0, redis.call("GET", KEYS[2]) or false } +end + +local ref = redis.call("HGET", KEYS[1], "ref:" .. view) +if not ref then + return { 1, present and 1 or 0, false } +end +return { 1, present and 1 or 0, redis.call("HGET", KEYS[1], "body:" .. ref) or false } +`; + +type SessionResponseBodyView = (typeof SESSION_RESPONSE_BODY_VIEWS)[number]; +export type SessionResponseBodySetInput = Partial>; + +type PreparedSessionResponseBodyBundle = { + bodies: string[]; + byteSize: number; + overBudget: boolean; + present: Record; + refs: Record; +}; -function canStoreSessionResponseBody(value: string, context: string): boolean { +type SessionResponseBodyBundleRead = { + body: string | null; + exists: boolean; + present: boolean; +}; + +type PreparedLegacySessionResponseBodySet = { + bodies: Record; + present: Record; +}; + +function getSessionResponseBodyMaxBytes(): number { const configuredMaxBytes = getEnvConfig().SESSION_RESPONSE_BODY_MAX_BYTES; - const maxBytes = - Number.isSafeInteger(configuredMaxBytes) && configuredMaxBytes > 0 - ? configuredMaxBytes - : DEFAULT_SESSION_RESPONSE_BODY_MAX_BYTES; + return Number.isSafeInteger(configuredMaxBytes) && configuredMaxBytes > 0 + ? configuredMaxBytes + : DEFAULT_SESSION_RESPONSE_BODY_MAX_BYTES; +} + +function canStoreSessionResponseBody(value: string, context: string): boolean { + const maxBytes = getSessionResponseBodyMaxBytes(); const byteSize = Buffer.byteLength(value, "utf8"); if (byteSize <= maxBytes) return true; @@ -78,6 +241,125 @@ function canStoreSessionResponseBody(value: string, context: string): boolean { return false; } +function normalizeSessionResponseBody(value: string | object, storeMessages: boolean): string { + if (storeMessages) return typeof value === "string" ? value : JSON.stringify(value); + if (typeof value === "object") return JSON.stringify(redactResponseBody(value)); + + try { + return JSON.stringify(redactResponseBody(JSON.parse(value) as unknown)); + } catch { + return value; + } +} + +function buildSessionResponseBodyBundleKey(sessionId: string, sequence: number): string { + return `session:${sessionId}:req:${sequence}:response-bodies:v1`; +} + +function buildSessionResponseBodyBundleIndexKey(sessionId: string): string { + return `session:${sessionId}:response-body-bundles:v1`; +} + +function buildSessionResponseBodyGenerationKey(sessionId: string): string { + return `session:${sessionId}:response-body-generation:v1`; +} + +function buildSessionRequestResponseBodyGenerationKey(sessionId: string, sequence: number): string { + return `session:${sessionId}:req:${sequence}:response-body-generation:v1`; +} + +function buildLegacySessionResponseBodyViewKey( + sessionId: string, + sequence: number, + view: SessionResponseBodyView +): string { + if (view === "legacy") return `session:${sessionId}:req:${sequence}:response`; + return buildSessionDetailSnapshotKey(sessionId, sequence, "response", view, "body"); +} + +function prepareSessionResponseBodyBundle( + input: SessionResponseBodySetInput, + storeMessages: boolean +): PreparedSessionResponseBodyBundle { + const bodies: string[] = []; + const bodyIndexes = new Map(); + const refs = { legacy: "", before: "", after: "" }; + const present = { legacy: false, before: false, after: false }; + let byteSize = 0; + + for (const view of SESSION_RESPONSE_BODY_VIEWS) { + const value = input[view]; + if (value === undefined || value === null) continue; + present[view] = true; + const normalized = normalizeSessionResponseBody(value, storeMessages); + let ref = bodyIndexes.get(normalized); + if (ref === undefined) { + ref = String(bodies.length); + bodyIndexes.set(normalized, ref); + bodies.push(normalized); + byteSize += Buffer.byteLength(normalized, "utf8"); + } + refs[view] = ref; + } + + const overBudget = byteSize > getSessionResponseBodyMaxBytes(); + return { + bodies: overBudget ? [] : bodies, + byteSize, + overBudget, + present, + refs: overBudget ? { legacy: "", before: "", after: "" } : refs, + }; +} + +function prepareLegacySessionResponseBodySet( + input: SessionResponseBodySetInput, + storeMessages: boolean +): PreparedLegacySessionResponseBodySet { + const bodies: Record = { + legacy: null, + before: null, + after: null, + }; + const present = { legacy: false, before: false, after: false }; + + for (const view of SESSION_RESPONSE_BODY_VIEWS) { + const value = input[view]; + if (value === undefined || value === null) continue; + present[view] = true; + const normalized = normalizeSessionResponseBody(value, storeMessages); + if (canStoreSessionResponseBody(normalized, `response:${view}`)) { + bodies[view] = normalized; + } + } + + return { bodies, present }; +} + +async function readSessionResponseBodyBundleView( + redis: NonNullable>, + sessionId: string, + sequence: number, + view: SessionResponseBodyView +): Promise { + const result = (await redis.eval( + READ_SESSION_RESPONSE_BODY_BUNDLE_LUA, + 2, + buildSessionResponseBodyBundleKey(sessionId, sequence), + buildLegacySessionResponseBodyViewKey(sessionId, sequence, view), + view + )) as unknown; + if (!Array.isArray(result) || result.length < 3) { + throw new Error("invalid session response body bundle read result"); + } + + return { + exists: Number(result[0]) === 1, + present: Number(result[1]) === 1, + body: typeof result[2] === "string" ? result[2] : null, + }; +} + function isReservedInternalHeader(name: string): boolean { const lowerName = name.toLowerCase(); return lowerName.startsWith("x-cch-") || RESERVED_INTERNAL_HEADER_SET.has(lowerName); @@ -401,12 +683,22 @@ export class SessionManager { ` local sequence = redis.call('INCR', KEYS[1]) redis.call('PERSIST', KEYS[1]) + local generation = redis.call('GET', KEYS[2]) + if not generation then + generation = '0' + redis.call('SETEX', KEYS[2], ARGV[2], generation) + else + redis.call('EXPIRE', KEYS[2], ARGV[2]) + end local ownerKey = ARGV[1] .. sequence .. ':owner' + local requestGenerationKey = ARGV[1] .. sequence .. ':response-body-generation:v1' redis.call('SETEX', ownerKey, ARGV[2], ARGV[3]) + redis.call('SETEX', requestGenerationKey, ARGV[2], generation) return sequence `, - 1, + 2, key, + buildSessionResponseBodyGenerationKey(sessionId), `session:${sessionId}:req:`, String(SessionManager.SESSION_TTL), String(keyId) @@ -2134,6 +2426,140 @@ export class SessionManager { } } + static async storeSessionResponseBodySet( + sessionId: string, + input: SessionResponseBodySetInput, + requestSequence?: number, + keyId?: number + ): Promise { + if (!getEnvConfig().STORE_SESSION_RESPONSE_BODY) return; + + const redis = getRedisClient(); + if (redis?.status !== "ready") return; + + const sequence = normalizeRequestSequence(requestSequence); + if (sequence === null) { + if (keyId !== undefined) { + logger.warn("SessionManager: Skipped response body set with invalid request sequence", { + sessionId, + requestSequence, + keyId, + }); + return; + } + const writes: Array> = []; + if (input.legacy !== undefined && input.legacy !== null) { + writes.push( + SessionManager.storeSessionResponse(sessionId, input.legacy, requestSequence, keyId) + ); + } + if (input.before !== undefined && input.before !== null) { + writes.push( + SessionManager.storeSessionResponsePhaseSnapshot( + sessionId, + "before", + { body: input.before }, + requestSequence, + keyId + ) + ); + } + if (input.after !== undefined && input.after !== null) { + writes.push( + SessionManager.storeSessionResponsePhaseSnapshot( + sessionId, + "after", + { body: input.after }, + requestSequence, + keyId + ) + ); + } + await Promise.all(writes); + return; + } + + try { + const bundleKey = buildSessionResponseBodyBundleKey(sessionId, sequence); + const bundleIndexKey = buildSessionResponseBodyBundleIndexKey(sessionId); + const generationKey = buildSessionResponseBodyGenerationKey(sessionId); + const requestGenerationKey = buildSessionRequestResponseBodyGenerationKey( + sessionId, + sequence + ); + const requestOwnerKey = `session:${sessionId}:req:${sequence}:owner`; + const legacyKeys = SESSION_RESPONSE_BODY_VIEWS.map((view) => + buildLegacySessionResponseBodyViewKey(sessionId, sequence, view) + ); + if (!getEnvConfig().SESSION_RESPONSE_BODY_DEDUP_ENABLED) { + const legacy = prepareLegacySessionResponseBodySet(input, SessionManager.STORE_MESSAGES); + await SessionManager.refreshSessionRequestOwner(redis, sessionId, sequence, keyId); + await redis.eval( + WRITE_LEGACY_SESSION_RESPONSE_BODY_SET_LUA, + 8, + bundleKey, + ...legacyKeys, + bundleIndexKey, + generationKey, + requestGenerationKey, + requestOwnerKey, + SessionManager.SESSION_TTL, + keyId ?? "", + legacy.present.legacy ? 1 : 0, + legacy.bodies.legacy === null ? 0 : 1, + legacy.bodies.legacy ?? "", + legacy.present.before ? 1 : 0, + legacy.bodies.before === null ? 0 : 1, + legacy.bodies.before ?? "", + legacy.present.after ? 1 : 0, + legacy.bodies.after === null ? 0 : 1, + legacy.bodies.after ?? "" + ); + return; + } + + const bundle = prepareSessionResponseBodyBundle(input, SessionManager.STORE_MESSAGES); + const maxBytes = getSessionResponseBodyMaxBytes(); + if (bundle.overBudget) { + logger.warn("SessionManager: Skipped response body bundle over aggregate limit", { + sessionId, + requestSequence: sequence, + byteSize: bundle.byteSize, + maxBytes, + }); + } + + await SessionManager.refreshSessionRequestOwner(redis, sessionId, sequence, keyId); + await redis.eval( + WRITE_SESSION_RESPONSE_BODY_BUNDLE_LUA, + 8, + bundleKey, + ...legacyKeys, + bundleIndexKey, + generationKey, + requestGenerationKey, + requestOwnerKey, + SessionManager.SESSION_TTL, + keyId ?? "", + bundle.byteSize, + bundle.overBudget ? 1 : 0, + bundle.present.legacy ? 1 : 0, + bundle.present.before ? 1 : 0, + bundle.present.after ? 1 : 0, + bundle.refs.legacy, + bundle.refs.before, + bundle.refs.after, + ...bundle.bodies + ); + } catch (error) { + logger.error("SessionManager: Failed to store response body bundle", { + error, + sessionId, + requestSequence, + }); + } + } + /** * 存储 session 完整请求体（客户端原始请求体，临时存储，5分钟过期） * @@ -2524,9 +2950,13 @@ export class SessionManager { const sequence = normalizeRequestSequence(requestSequence); if (sequence === null) return null; - const newKey = `session:${sessionId}:req:${sequence}:response`; - const response = await redis.get(newKey); - return response; + const bundled = await readSessionResponseBodyBundleView( + redis, + sessionId, + sequence, + "legacy" + ); + return bundled.body; } // 向后兼容：尝试旧格式 @@ -2796,13 +3226,19 @@ export class SessionManager { const sequence = normalizeRequestSequence(requestSequence); if (!sequence) return null; - const [bodyValue, headersValue, metaValue] = await Promise.all([ - redis.get(buildSessionDetailSnapshotKey(sessionId, sequence, "response", phase, "body")), + const bundled = await readSessionResponseBodyBundleView(redis, sessionId, sequence, phase); + const [headersValue, metaValue] = await Promise.all([ redis.get(buildSessionDetailSnapshotKey(sessionId, sequence, "response", phase, "headers")), redis.get(buildSessionDetailSnapshotKey(sessionId, sequence, "response", phase, "meta")), ]); + const bodyValue = bundled.body; - if (bodyValue === null && headersValue === null && metaValue === null) { + if ( + bodyValue === null && + headersValue === null && + metaValue === null && + (!bundled.exists || !bundled.present) + ) { return null; } @@ -3220,13 +3656,23 @@ export class SessionManager { const pipeline = redis.pipeline(); // Binding mirrors are mutated only by the tenant-authorized helpers above. + pipeline.eval( + DELETE_SESSION_RESPONSE_BODY_BUNDLES_LUA, + 5, + buildSessionResponseBodyBundleIndexKey(sessionId), + buildSessionResponseBodyGenerationKey(sessionId), + `session:${sessionId}:response`, + buildLegacySessionResponseBodyViewKey(sessionId, 1, "before"), + buildLegacySessionResponseBodyViewKey(sessionId, 1, "after"), + SessionManager.SESSION_TTL, + crypto.randomUUID() + ); pipeline.del(`session:${sessionId}:info`); pipeline.del(`session:${sessionId}:last_seen`); pipeline.del(`session:${sessionId}:concurrent_count`); // 可选：messages 和 response（如果启用了存储） pipeline.del(`session:${sessionId}:messages`); - pipeline.del(`session:${sessionId}:response`); // 3. 从 ZSET 中移除（始终尝试，即使查询失败） pipeline.zrem(getGlobalActiveSessionsKey(), sessionId); diff --git a/tests/configs/integration.config.mts b/tests/configs/integration.config.mts index 6351f7ca8..0dd5c4cd8 100644 --- a/tests/configs/integration.config.mts +++ b/tests/configs/integration.config.mts @@ -11,6 +11,7 @@ export default createTestRunnerConfig({ "tests/integration/rolling-cost-redis.test.ts", "tests/integration/lease-settlement-redis.test.ts", "tests/integration/session-binding-versioning-redis.test.ts", + "tests/integration/session-response-body-dedup-redis.test.ts", "tests/integration/db-pool-isolation-postgres.test.ts", "tests/integration/db-pool-slow-close-postgres.test.ts", "tests/integration/message-write-buffer-recovery-postgres.test.ts", diff --git a/tests/integration/billing-model-source.test.ts b/tests/integration/billing-model-source.test.ts index a4d223e13..f8576f970 100644 --- a/tests/integration/billing-model-source.test.ts +++ b/tests/integration/billing-model-source.test.ts @@ -67,6 +67,7 @@ vi.mock("@/repository/message", () => ({ updateMessageRequestCostWithBreakdown: vi.fn(), updateMessageRequestDetails: vi.fn(), updateMessageRequestDetailsDurably: vi.fn(), + updateMessageRequestDetailsIfUnfinalized: vi.fn(async () => ({ updated: true })), updateMessageRequestDuration: vi.fn(), })); @@ -75,6 +76,7 @@ vi.mock("@/lib/session-manager", () => ({ updateSessionUsage: vi.fn(), updateSessionProvider: vi.fn(), storeSessionResponse: vi.fn(), + storeSessionResponseBodySet: vi.fn(async () => undefined), extractCodexPromptCacheKey: vi.fn(), updateSessionWithCodexCacheKey: vi.fn(), }, diff --git a/tests/integration/session-response-body-dedup-redis.test.ts b/tests/integration/session-response-body-dedup-redis.test.ts new file mode 100644 index 000000000..658dfd6ea --- /dev/null +++ b/tests/integration/session-response-body-dedup-redis.test.ts @@ -0,0 +1,423 @@ +import { randomUUID } from "node:crypto"; +import Redis from "ioredis"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, test, vi } from "vitest"; + +const configState = vi.hoisted(() => ({ dedupEnabled: true })); + +vi.mock("@/lib/config/env.schema", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + getEnvConfig: () => ({ + ...actual.getEnvConfig(), + SESSION_RESPONSE_BODY_DEDUP_ENABLED: configState.dedupEnabled, + }), + }; +}); + +process.env.ENABLE_RATE_LIMIT = "true"; +process.env.SESSION_TTL = "2"; +process.env.STORE_SESSION_MESSAGES = "true"; +process.env.STORE_SESSION_RESPONSE_BODY = "true"; +process.env.SESSION_RESPONSE_BODY_DEDUP_ENABLED = "true"; + +const { SessionManager } = await import("@/lib/session-manager"); +const { closeRedis, getRedisClient } = await import("@/lib/redis/client"); + +const HAS_REDIS = Boolean(process.env.REDIS_URL); +const runWithRedis = describe.skipIf(!HAS_REDIS); +const TEST_PREFIX = `it-session-response-body-${Date.now()}-${randomUUID()}`; +const touchedKeys = new Set(); + +function bundleKey(sessionId: string, sequence = 1): string { + return `session:${sessionId}:req:${sequence}:response-bodies:v1`; +} + +function bundleIndexKey(sessionId: string): string { + return `session:${sessionId}:response-body-bundles:v1`; +} + +function legacyBodyKey( + sessionId: string, + view: "legacy" | "before" | "after", + sequence = 1 +): string { + if (view === "legacy") return `session:${sessionId}:req:${sequence}:response`; + return `session:${sessionId}:req:${sequence}:snapshot:response:${view}:body`; +} + +function generationKey(sessionId: string): string { + return `session:${sessionId}:response-body-generation:v1`; +} + +function requestGenerationKey(sessionId: string, sequence = 1): string { + return `session:${sessionId}:req:${sequence}:response-body-generation:v1`; +} + +async function waitForRedisReady() { + const client = getRedisClient({ allowWhenRateLimitDisabled: true }); + if (!client) throw new Error("Redis client unavailable for integration test"); + if (client.status !== "ready") { + await new Promise((resolve, reject) => { + const timeout = setTimeout(() => reject(new Error("Redis ready timeout")), 5_000); + client.once("ready", () => { + clearTimeout(timeout); + resolve(); + }); + }); + } + if (client.status !== "ready") throw new Error(`Redis not ready: ${client.status}`); + return client; +} + +runWithRedis("session response body deduplication Redis integration", () => { + let redis: Redis; + let sequence = 0; + + function nextSessionId(label: string): string { + sequence += 1; + const sessionId = `${TEST_PREFIX}:${label}:${sequence}`; + touchedKeys.add(bundleIndexKey(sessionId)); + return sessionId; + } + + async function cleanupTouchedKeys(): Promise { + for (const key of touchedKeys) await redis.del(key); + touchedKeys.clear(); + } + + beforeAll(async () => { + await waitForRedisReady(); + redis = new Redis(process.env.REDIS_URL!, { + lazyConnect: true, + enableOfflineQueue: false, + maxRetriesPerRequest: 1, + }); + await redis.connect(); + await expect(redis.ping()).resolves.toBe("PONG"); + }); + + beforeEach(() => { + configState.dedupEnabled = true; + }); + + afterEach(async () => { + await cleanupTouchedKeys(); + }); + + afterAll(async () => { + if (!redis) { + await closeRedis(); + return; + } + await cleanupTouchedKeys(); + if (redis.status !== "end") await redis.quit(); + await closeRedis(); + }); + + test("stores duplicate views once in a Redis Hash and resolves all three logical views", async () => { + const sessionId = nextSessionId("same"); + const key = bundleKey(sessionId); + const indexKey = bundleIndexKey(sessionId); + touchedKeys.add(key); + touchedKeys.add(indexKey); + await redis.zadd(indexKey, Date.now() - 1_000, "expired-bundle-entry"); + + await SessionManager.storeSessionResponseBodySet( + sessionId, + { legacy: "same body", before: "same body", after: "same body" }, + 1 + ); + + expect(await redis.type(key)).toBe("hash"); + const stored = await redis.hgetall(key); + expect(stored).toMatchObject({ + schema: "1", + total_bytes: String(Buffer.byteLength("same body")), + over_budget: "0", + "present:legacy": "1", + "present:before": "1", + "present:after": "1", + "ref:legacy": "0", + "ref:before": "0", + "ref:after": "0", + "body:0": "same body", + }); + expect(Object.keys(stored).filter((field) => field.startsWith("body:"))).toEqual(["body:0"]); + expect(await redis.ttl(key)).toBeGreaterThan(0); + expect(await redis.zrange(indexKey, 0, -1)).toEqual([key]); + expect(await redis.ttl(indexKey)).toBeGreaterThan(0); + for (const view of ["legacy", "before", "after"] as const) { + const legacyKey = legacyBodyKey(sessionId, view); + touchedKeys.add(legacyKey); + expect(await redis.exists(legacyKey)).toBe(0); + } + + await expect(SessionManager.getSessionResponse(sessionId, 1)).resolves.toBe("same body"); + await expect( + SessionManager.getSessionResponsePhaseSnapshot(sessionId, "before", 1) + ).resolves.toMatchObject({ body: "same body" }); + await expect( + SessionManager.getSessionResponsePhaseSnapshot(sessionId, "after", 1) + ).resolves.toMatchObject({ body: "same body" }); + }); + + test("atomically replaces a complete generation under concurrent retries", async () => { + const sessionId = nextSessionId("retry"); + const key = bundleKey(sessionId); + touchedKeys.add(key); + const first = { legacy: "first legacy", before: "first before", after: "first legacy" }; + const second = { legacy: "second legacy", before: "second before", after: "second legacy" }; + + await Promise.all([ + SessionManager.storeSessionResponseBodySet(sessionId, first, 1), + SessionManager.storeSessionResponseBodySet(sessionId, second, 1), + ]); + + const actual = { + legacy: await SessionManager.getSessionResponse(sessionId, 1), + before: (await SessionManager.getSessionResponsePhaseSnapshot(sessionId, "before", 1))?.body, + after: (await SessionManager.getSessionResponsePhaseSnapshot(sessionId, "after", 1))?.body, + }; + expect([first, second]).toContainEqual(actual); + + const stored = await redis.hgetall(key); + const rawBodies = Object.entries(stored) + .filter(([field]) => field.startsWith("body:")) + .map(([, value]) => value) + .sort(); + expect(rawBodies).toEqual( + [actual.legacy, actual.before].filter((value): value is string => value !== null).sort() + ); + }); + + test("atomically switches layouts across rollout and rollback writers", async () => { + const sessionId = nextSessionId("mixed-layout"); + const key = bundleKey(sessionId); + touchedKeys.add(key); + for (const view of ["legacy", "before", "after"] as const) { + touchedKeys.add(legacyBodyKey(sessionId, view)); + } + + await SessionManager.storeSessionResponseBodySet( + sessionId, + { legacy: "dedup first", before: "dedup first", after: "dedup first" }, + 1 + ); + + configState.dedupEnabled = false; + await SessionManager.storeSessionResponseBodySet( + sessionId, + { legacy: "legacy second", before: "legacy before", after: "legacy second" }, + 1 + ); + + expect(await redis.hget(key, "layout")).toBe("legacy"); + expect(await redis.hkeys(key)).not.toContain("body:0"); + await expect(SessionManager.getSessionResponse(sessionId, 1)).resolves.toBe("legacy second"); + await expect( + SessionManager.getSessionResponsePhaseSnapshot(sessionId, "before", 1) + ).resolves.toMatchObject({ body: "legacy before" }); + const legacyTtls = await Promise.all([ + redis.ttl(key), + ...(["legacy", "before", "after"] as const).map((view) => + redis.ttl(legacyBodyKey(sessionId, view)) + ), + ]); + expect(Math.max(...legacyTtls) - Math.min(...legacyTtls)).toBeLessThanOrEqual(1); + + configState.dedupEnabled = true; + await SessionManager.storeSessionResponseBodySet( + sessionId, + { legacy: "dedup third", before: "dedup third", after: "dedup third" }, + 1 + ); + + expect(await redis.hget(key, "layout")).toBe("dedup"); + expect(await redis.hget(key, "body:0")).toBe("dedup third"); + for (const view of ["legacy", "before", "after"] as const) { + expect(await redis.exists(legacyBodyKey(sessionId, view))).toBe(0); + } + await expect(SessionManager.getSessionResponse(sessionId, 1)).resolves.toBe("dedup third"); + }); + + test("returns one complete generation when legacy and dedup writers race", async () => { + const sessionId = nextSessionId("mixed-race"); + const key = bundleKey(sessionId); + touchedKeys.add(key); + for (const view of ["legacy", "before", "after"] as const) { + touchedKeys.add(legacyBodyKey(sessionId, view)); + } + + const legacyGeneration = { + legacy: "legacy generation", + before: "legacy before", + after: "legacy generation", + }; + const dedupGeneration = { + legacy: "dedup generation", + before: "dedup before", + after: "dedup generation", + }; + + configState.dedupEnabled = false; + const legacyWrite = SessionManager.storeSessionResponseBodySet(sessionId, legacyGeneration, 1); + configState.dedupEnabled = true; + const dedupWrite = SessionManager.storeSessionResponseBodySet(sessionId, dedupGeneration, 1); + await Promise.all([legacyWrite, dedupWrite]); + + const actual = { + legacy: await SessionManager.getSessionResponse(sessionId, 1), + before: (await SessionManager.getSessionResponsePhaseSnapshot(sessionId, "before", 1))?.body, + after: (await SessionManager.getSessionResponsePhaseSnapshot(sessionId, "after", 1))?.body, + }; + expect([legacyGeneration, dedupGeneration]).toContainEqual(actual); + + const layout = await redis.hget(key, "layout"); + const oldKeyCount = await Promise.all( + (["legacy", "before", "after"] as const).map((view) => + redis.exists(legacyBodyKey(sessionId, view)) + ) + ).then((values) => values.reduce((sum, value) => sum + value, 0)); + expect(["legacy", "dedup"]).toContain(layout); + expect(oldKeyCount).toBe(layout === "legacy" ? 3 : 0); + }); + + test("falls back to TTL-window legacy keys only when no v2 bundle exists", async () => { + const sessionId = nextSessionId("legacy"); + for (const [view, body] of [ + ["legacy", "legacy response"], + ["before", "legacy before"], + ["after", "legacy after"], + ] as const) { + const key = legacyBodyKey(sessionId, view); + touchedKeys.add(key); + await redis.setex(key, 2, body); + } + + await expect(SessionManager.getSessionResponse(sessionId, 1)).resolves.toBe("legacy response"); + await expect( + SessionManager.getSessionResponsePhaseSnapshot(sessionId, "before", 1) + ).resolves.toMatchObject({ body: "legacy before" }); + await expect( + SessionManager.getSessionResponsePhaseSnapshot(sessionId, "after", 1) + ).resolves.toMatchObject({ body: "legacy after" }); + }); + + test("expires refs and body values together with the request-scoped bundle", async () => { + const sessionId = nextSessionId("ttl"); + const key = bundleKey(sessionId); + const indexKey = bundleIndexKey(sessionId); + touchedKeys.add(key); + touchedKeys.add(indexKey); + + await SessionManager.storeSessionResponseBodySet( + sessionId, + { legacy: "ttl body", before: "ttl body", after: "ttl body" }, + 1 + ); + + expect(await redis.pttl(key)).toBeGreaterThan(0); + await new Promise((resolve) => setTimeout(resolve, 2_200)); + expect(await redis.exists(key)).toBe(0); + expect(await redis.exists(indexKey)).toBe(0); + await expect(SessionManager.getSessionResponse(sessionId, 1)).resolves.toBeNull(); + await expect( + SessionManager.getSessionResponsePhaseSnapshot(sessionId, "after", 1) + ).resolves.toBeNull(); + }); + + test("full session termination removes every indexed response body bundle", async () => { + const sessionId = nextSessionId("terminate"); + const firstKey = bundleKey(sessionId, 1); + const secondKey = bundleKey(sessionId, 2); + const indexKey = bundleIndexKey(sessionId); + touchedKeys.add(firstKey); + touchedKeys.add(secondKey); + touchedKeys.add(indexKey); + + await SessionManager.storeSessionResponseBodySet(sessionId, { legacy: "first" }, 1); + await SessionManager.storeSessionResponseBodySet(sessionId, { legacy: "second" }, 2); + expect(await redis.zrange(indexKey, 0, -1)).toEqual([firstKey, secondKey]); + + await expect(SessionManager.terminateSession(sessionId)).resolves.toBe(true); + + expect(await redis.exists(firstKey, secondKey, indexKey)).toBe(0); + }); + + test("full termination removes legacy-layout and sequence-fallback response bodies", async () => { + const sessionId = nextSessionId("terminate-legacy"); + const indexKey = bundleIndexKey(sessionId); + configState.dedupEnabled = false; + + for (const sequence of [1, 2]) { + touchedKeys.add(bundleKey(sessionId, sequence)); + for (const view of ["legacy", "before", "after"] as const) { + touchedKeys.add(legacyBodyKey(sessionId, view, sequence)); + } + await SessionManager.storeSessionResponseBodySet( + sessionId, + { legacy: `legacy-${sequence}`, before: `before-${sequence}`, after: `after-${sequence}` }, + sequence + ); + } + + const fallbackKeys = [ + `session:${sessionId}:response`, + legacyBodyKey(sessionId, "before"), + legacyBodyKey(sessionId, "after"), + ]; + fallbackKeys.forEach((key) => touchedKeys.add(key)); + await SessionManager.storeSessionResponseBodySet(sessionId, { + legacy: "fallback", + before: "fallback-before", + after: "fallback-after", + }); + + expect(await redis.zcard(indexKey)).toBe(2); + expect(await redis.exists(...fallbackKeys)).toBe(3); + await expect(SessionManager.terminateSession(sessionId)).resolves.toBe(true); + + const indexedPhysicalKeys = [1, 2].flatMap((sequence) => [ + bundleKey(sessionId, sequence), + ...(["legacy", "before", "after"] as const).map((view) => + legacyBodyKey(sessionId, view, sequence) + ), + ]); + expect(await redis.exists(indexKey, ...indexedPhysicalKeys, ...fallbackKeys)).toBe(0); + await expect(SessionManager.getSessionResponse(sessionId, 1)).resolves.toBeNull(); + await expect( + SessionManager.getSessionResponsePhaseSnapshot(sessionId, "after", 2) + ).resolves.toBeNull(); + }); + + test("full termination fences late writers while allowing a newly sequenced request", async () => { + const sessionId = nextSessionId("terminate-fence"); + const indexKey = bundleIndexKey(sessionId); + const responseGenerationKey = generationKey(sessionId); + const firstRequestGenerationKey = requestGenerationKey(sessionId); + const requestOwnerKey = `session:${sessionId}:req:1:owner`; + const sequenceKey = `session:${sessionId}:seq`; + [responseGenerationKey, firstRequestGenerationKey, requestOwnerKey, sequenceKey].forEach( + (key) => touchedKeys.add(key) + ); + + await redis.setex(responseGenerationKey, 2, "generation-before-termination"); + await redis.setex(firstRequestGenerationKey, 2, "generation-before-termination"); + await redis.setex(requestOwnerKey, 2, "42"); + await SessionManager.storeSessionResponseBodySet(sessionId, { legacy: "first" }, 1, 42); + expect(await redis.exists(bundleKey(sessionId))).toBe(1); + + await expect(SessionManager.terminateSession(sessionId)).resolves.toBe(true); + expect(await redis.exists(bundleKey(sessionId), firstRequestGenerationKey)).toBe(0); + + await SessionManager.storeSessionResponseBodySet(sessionId, { legacy: "late" }, 1, 42); + expect(await redis.exists(bundleKey(sessionId))).toBe(0); + + await expect(SessionManager.getNextRequestSequence(sessionId, 42)).resolves.toBe(1); + await SessionManager.storeSessionResponseBodySet(sessionId, { legacy: "new" }, 1, 42); + expect(await redis.hget(bundleKey(sessionId), "body:0")).toBe("new"); + expect(await redis.get(firstRequestGenerationKey)).toBe(await redis.get(responseGenerationKey)); + expect(await redis.zrange(indexKey, 0, -1)).toEqual([bundleKey(sessionId)]); + }); +}); diff --git a/tests/load/issue-1408-replay-oom/README.md b/tests/load/issue-1408-replay-oom/README.md index 1e794fb45..c16da06f0 100644 --- a/tests/load/issue-1408-replay-oom/README.md +++ b/tests/load/issue-1408-replay-oom/README.md @@ -9,14 +9,21 @@ CC Hub instance, PostgreSQL, Redis, a Provider pointing at the mock, an API key, ## 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. +- `mock-upstream.cjs`: sends an exact-size SSE body. `CCH_MOCK_RESPONSE_MODE=disconnect` (the + default) leaves the stream open after the delta frames for the #1408 abort reproduction; + `complete` appends a terminal `response.completed` event and closes the response for #1415 + storage acceptance. `CCH_MOCK_RESPONSE_BYTES` controls total wire bytes from 64 KiB to 64 MiB. - `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`. + the mock receipt count, then either disconnects after `CCH_ABORT_DELAY_MS` or waits for every + response to complete when `CCH_REQUEST_MODE=complete`, and writes exact session IDs to a manifest. - `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. +- `sample-container.sh`: samples app logs, cgroup current/peak memory, Redis memory, RDB state, and + response/Replay key counts into a result file. +- `inspect-redis.cjs`: validates each manifest response bundle with Redis `HSTRLEN`, refs, TTL, + stale legacy keys, RDB status, and container exit/OOM state. +- `run-wave.sh`: runs the driver and sampler together, optionally triggers `BGSAVE`, then writes + driver, manifest, and Redis evidence artifacts next to the sample output. - `start-mock-container.sh`: starts the mock on an existing Docker network without replacing an existing container. @@ -28,15 +35,18 @@ CC Hub instance, PostgreSQL, Redis, a Provider pointing at the mock, an API key, 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`. +6. For response body dedup validation, set `STORE_SESSION_RESPONSE_BODY=true`, + `SESSION_RESPONSE_BODY_DEDUP_ENABLED=true`, and + `SESSION_RESPONSE_BODY_MAX_BYTES=5242880`. Use an isolated Redis with RDB enabled. -Do not point this fixture at a production Provider. The mock deliberately leaves every upstream -response open until the app or fixture closes it. +Do not point this fixture at a production Provider. The default mock mode 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 + cch1408-mock cch1408-network 31409 5242880 ``` The command prints both URLs: @@ -46,12 +56,27 @@ stats=http://127.0.0.1:31409/stats provider=http://cch1408-mock:3001 ``` +For Issue #1415 completed-response storage acceptance, pass `complete` as the fifth argument: + +```bash +tests/load/issue-1408-replay-oom/start-mock-container.sh \ + cch1415-mock cch1415-network 31409 5242880 complete +``` + ## 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 +export CCH_MOCK_RESPONSE_BYTES=5242880 +export CCH_REQUEST_MODE=complete +export CCH_COMPLETION_TIMEOUT_MS=60000 +export CCH_TRIGGER_RDB_BGSAVE=true +export CCH_RDB_DELAY_SECONDS=70 +export CCH_SAMPLES=36 +export CCH_VERIFY_TTL_CLEANUP=true +export CCH_TTL_CLEANUP_TIMEOUT_SECONDS=360 tests/load/issue-1408-replay-oom/run-wave.sh \ http://127.0.0.1:31415 \ @@ -69,11 +94,38 @@ CCH_WAVES=8 CCH_REQUESTS_PER_WAVE=8 CCH_WAVE_INTERVAL_MS=10000 CCH_ABORT_DELAY_MS=250 +CCH_REQUEST_MODE=disconnect +CCH_COMPLETION_TIMEOUT_MS=60000 CCH_SAMPLES=18 CCH_SAMPLE_INTERVAL_SECONDS=10 -CCH_MOCK_MIB=8 +CCH_MOCK_RESPONSE_BYTES=5242880 +CCH_TRIGGER_RDB_BGSAVE=false +CCH_RDB_DELAY_SECONDS=70 +CCH_VERIFY_TTL_CLEANUP=false +CCH_TTL_CLEANUP_TIMEOUT_SECONDS=360 ``` +The output argument is the prefix for four artifacts, plus a fifth artifact when TTL cleanup +verification is enabled: + +```text +issue1408-fixed.samples.txt +issue1408-fixed.samples.txt.driver.jsonl +issue1408-fixed.samples.txt.sessions.json +issue1408-fixed.samples.txt.redis.json +issue1408-fixed.samples.txt.redis-expired.json +``` + +The active Redis artifact is captured immediately after the requested `BGSAVE` completes, while +the response bodies are still inside `SESSION_TTL`. Sampling then continues for the configured +window. With `CCH_VERIFY_TTL_CLEANUP=true`, the runner polls the same manifest until every bundle +and legacy response body key has expired, then writes the separate expired artifact. The timeout +must be long enough to cover the maximum remaining `SESSION_TTL` after sampling. + +`CCH_MOCK_MIB` remains a compatibility fallback when `CCH_MOCK_RESPONSE_BYTES` is unset. It is not +used for the 5 MiB acceptance run because the acceptance measures total wire bytes, including SSE +envelopes and the terminal event. + 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 @@ -89,8 +141,15 @@ For the fixed revision under the default workload: - 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 Redis evidence reports 64 bundles, one `body:*` field and three identical refs per bundle, + zero dangling refs, and zero old response body keys. +- `summary.totalRawBodyBytes` is no greater than `64 * 5242880`, and every bundle's raw `HSTRLEN` + equals its declared `total_bytes`. +- Redis remains running across `BGSAVE`, `rdb_last_bgsave_status=ok`, and both app and Redis report + `OOMKilled=false` with non-error exit state. +- The expired Redis evidence reports zero remaining manifest bundles and zero legacy response body + keys after `SESSION_TTL`. Replay meta may remain until its own TTL; owner and chunk counts are + reported separately and must not be conflated with response body cleanup. 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 index 8fff27546..db1c39e8a 100644 --- a/tests/load/issue-1408-replay-oom/drive-disconnect-waves.cjs +++ b/tests/load/issue-1408-replay-oom/drive-disconnect-waves.cjs @@ -26,6 +26,16 @@ const abortDelayMs = parseBoundedInteger( 0, 60000 ); +const requestMode = parseEnum(process.env.CCH_REQUEST_MODE || "disconnect", "CCH_REQUEST_MODE", [ + "disconnect", + "complete", +]); +const completionTimeoutMs = parseBoundedInteger( + process.env.CCH_COMPLETION_TIMEOUT_MS || "60000", + "CCH_COMPLETION_TIMEOUT_MS", + 1, + 3600000 +); const mockReceiptTimeoutMs = parseBoundedInteger( process.env.CCH_MOCK_RECEIPT_TIMEOUT_MS || "30000", "CCH_MOCK_RECEIPT_TIMEOUT_MS", @@ -37,6 +47,8 @@ if (!model || model.length > 256) { throw new Error("CCH_REQUEST_MODEL must contain between 1 and 256 characters"); } const key = readApiKey(); +const sessionManifestPath = process.env.CCH_SESSION_MANIFEST?.trim() || null; +const manifestSessionIds = []; function parseHttpUrl(raw, name) { const url = new URL(raw); @@ -61,6 +73,13 @@ function parseBoundedInteger(raw, name, minimum, maximum) { return value; } +function parseEnum(raw, name, values) { + if (!values.includes(raw)) { + throw new Error(`${name} must be one of: ${values.join(", ")}`); + } + return raw; +} + function readApiKey() { const direct = process.env.CCH_API_KEY?.trim(); if (direct) return direct; @@ -119,6 +138,28 @@ function hashScenario(value) { return [...value].reduce((hash, char) => (hash * 33 + char.charCodeAt(0)) & 0xff, 0); } +function writeSessionManifest(completedWaves) { + if (!sessionManifestPath) return; + const temporaryPath = `${sessionManifestPath}.tmp`; + fs.writeFileSync( + temporaryPath, + `${JSON.stringify( + { + scenarioPrefix: normalizedScenarioPrefix, + waves, + requestsPerWave: perWave, + completedWaves, + requestSequence: 1, + sessionIds: manifestSessionIds, + }, + null, + 2 + )}\n`, + { mode: 0o600 } + ); + fs.renameSync(temporaryPath, sessionManifestPath); +} + function startRequest(scenario, scenarioHash, wave, index) { const body = JSON.stringify({ model, @@ -140,7 +181,15 @@ function startRequest(scenario, scenarioHash, wave, index) { 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 sessionId = `019c1408-0000-7000-8000-${suffix}`; + let settleCompletion; + let rejectCompletion; + const completion = new Promise((resolve, reject) => { + settleCompletion = resolve; + rejectCompletion = reject; + }); + completion.catch(() => {}); + const handle = { request: null, response: null, sessionId, completion }; const request = transportFor(url).request( url, { @@ -149,17 +198,22 @@ function startRequest(scenario, scenarioHash, wave, index) { authorization: `Bearer ${key}`, "content-type": "application/json", "content-length": Buffer.byteLength(body), - session_id: `019c1408-0000-7000-8000-${suffix}`, + session_id: sessionId, }, }, (response) => { handle.response = response; + if ((response.statusCode || 500) >= 400) { + rejectCompletion(new Error(`POST ${url} returned ${response.statusCode}`)); + } response.on("data", () => {}); - response.on("error", () => {}); + response.once("end", settleCompletion); + response.once("error", rejectCompletion); + response.once("aborted", () => rejectCompletion(new Error(`POST ${url} response aborted`))); } ); handle.request = request; - request.on("error", () => {}); + request.once("error", rejectCompletion); request.end(body); return handle; } @@ -171,6 +225,23 @@ function abortRequests(handles) { } } +async function waitForCompletions(handles) { + let timeout; + try { + await Promise.race([ + Promise.all(handles.map((handle) => handle.completion)), + new Promise((_, reject) => { + timeout = setTimeout( + () => reject(new Error(`response completion timeout after ${completionTimeoutMs}ms`)), + completionTimeoutMs + ); + }), + ]); + } finally { + clearTimeout(timeout); + } +} + async function main() { const scenarioHash = hashScenario(normalizedScenarioPrefix); for (let wave = 0; wave < waves; wave += 1) { @@ -184,25 +255,42 @@ async function main() { let received; let confirmedAt; let abortedAt; + let completedAt; + let shouldAbort = requestMode === "disconnect"; try { received = await waitForMock(scenario, before + perWave); confirmedAt = Date.now(); - await sleep(abortDelayMs); - abortedAt = Date.now(); + if (requestMode === "complete") { + await waitForCompletions(handles); + completedAt = Date.now(); + } else { + await sleep(abortDelayMs); + abortedAt = Date.now(); + } + } catch (error) { + shouldAbort = true; + throw error; } finally { - abortRequests(handles); + if (shouldAbort) abortRequests(handles); } + const sessionIds = handles.map((handle) => handle.sessionId); + manifestSessionIds.push(...sessionIds); + writeSessionManifest(wave + 1); + process.stdout.write( `${JSON.stringify({ wave, scenario, + sessionIds, perWave, mockBefore: before, mockReceived: received, + requestMode, confirmedAt, + completedAt, abortedAt, - abortDelayMs: abortedAt - confirmedAt, + abortDelayMs: abortedAt === undefined ? null : abortedAt - confirmedAt, })}\n` ); diff --git a/tests/load/issue-1408-replay-oom/inspect-redis.cjs b/tests/load/issue-1408-replay-oom/inspect-redis.cjs new file mode 100644 index 000000000..ee20faff1 --- /dev/null +++ b/tests/load/issue-1408-replay-oom/inspect-redis.cjs @@ -0,0 +1,251 @@ +"use strict"; + +const fs = require("node:fs"); +const { spawnSync } = require("node:child_process"); + +const [redisContainer, manifestPath, expectedBytesArg, modeArg] = process.argv.slice(2); +if (!redisContainer || !manifestPath) { + throw new Error( + "usage: inspect-redis.cjs REDIS_CONTAINER SESSION_MANIFEST [EXPECTED_RESPONSE_BYTES] [active|expired]" + ); +} + +const mode = modeArg || "active"; +if (mode !== "active" && mode !== "expired") { + throw new Error("inspection mode must be active or expired"); +} + +const expectedResponseBytes = parsePositiveInteger( + expectedBytesArg || "5242880", + "EXPECTED_RESPONSE_BYTES" +); +const manifest = JSON.parse(fs.readFileSync(manifestPath, "utf8")); +if (!Array.isArray(manifest.sessionIds) || !Number.isInteger(manifest.requestSequence)) { + throw new Error("session manifest must contain sessionIds and requestSequence"); +} + +const inspectBundleLua = ` +local bundle = KEYS[1] +local old_key_count = redis.call("EXISTS", KEYS[2], KEYS[3], KEYS[4]) +if redis.call("EXISTS", bundle) == 0 then + return { 0, "", "", "", "", 0, 0, 0, "", "", "", "", "", "", 0, -2, old_key_count } +end + +local raw_body_bytes = 0 +local body_field_count = 0 +for _, field in ipairs(redis.call("HKEYS", bundle)) do + if string.sub(field, 1, 5) == "body:" then + body_field_count = body_field_count + 1 + raw_body_bytes = raw_body_bytes + redis.call("HSTRLEN", bundle, field) + end +end + +local dangling_refs = 0 +for _, view in ipairs({ "legacy", "before", "after" }) do + local ref = redis.call("HGET", bundle, "ref:" .. view) + if ref and redis.call("HEXISTS", bundle, "body:" .. ref) == 0 then + dangling_refs = dangling_refs + 1 + end +end + +return { + 1, + redis.call("HGET", bundle, "schema") or "", + redis.call("HGET", bundle, "layout") or "", + redis.call("HGET", bundle, "total_bytes") or "", + redis.call("HGET", bundle, "over_budget") or "", + body_field_count, + raw_body_bytes, + redis.call("HLEN", bundle), + redis.call("HGET", bundle, "present:legacy") or "", + redis.call("HGET", bundle, "present:before") or "", + redis.call("HGET", bundle, "present:after") or "", + redis.call("HGET", bundle, "ref:legacy") or "", + redis.call("HGET", bundle, "ref:before") or "", + redis.call("HGET", bundle, "ref:after") or "", + dangling_refs, + redis.call("PTTL", bundle), + old_key_count +} +`; + +function parsePositiveInteger(raw, name) { + const value = Number(raw); + if (!Number.isSafeInteger(value) || value <= 0) { + throw new Error(`${name} must be a positive integer`); + } + return value; +} + +function runDocker(args) { + const result = spawnSync("docker", args, { encoding: "utf8", maxBuffer: 16 * 1024 * 1024 }); + if (result.error) throw result.error; + if (result.status !== 0) { + throw new Error(`docker ${args[0]} failed: ${(result.stderr || result.stdout).trim()}`); + } + return result.stdout.trim(); +} + +function redisCli(args) { + return runDocker(["exec", redisContainer, "redis-cli", ...args]); +} + +function parseInfo(section) { + const result = {}; + for (const line of redisCli(["--raw", "INFO", section]).split(/\r?\n/)) { + if (!line || line.startsWith("#")) continue; + const separator = line.indexOf(":"); + if (separator === -1) continue; + result[line.slice(0, separator)] = line.slice(separator + 1); + } + return result; +} + +function inspectBundle(sessionId) { + const sequence = manifest.requestSequence; + const prefix = `session:${sessionId}:req:${sequence}`; + const keys = [ + `${prefix}:response-bodies:v1`, + `${prefix}:response`, + `${prefix}:snapshot:response:before:body`, + `${prefix}:snapshot:response:after:body`, + ]; + const raw = redisCli(["--json", "EVAL", inspectBundleLua, "4", ...keys]); + const reply = JSON.parse(raw); + if (!Array.isArray(reply) || reply.length !== 17) { + throw new Error(`unexpected bundle inspection reply for ${sessionId}`); + } + + return { + sessionId, + exists: Number(reply[0]) === 1, + schema: String(reply[1]), + layout: String(reply[2]), + declaredTotalBytes: Number(reply[3] || 0), + overBudget: String(reply[4]) === "1", + bodyFieldCount: Number(reply[5]), + rawBodyBytes: Number(reply[6]), + hashFieldCount: Number(reply[7]), + present: { + legacy: String(reply[8]) === "1", + before: String(reply[9]) === "1", + after: String(reply[10]) === "1", + }, + refs: { + legacy: String(reply[11]), + before: String(reply[12]), + after: String(reply[13]), + }, + danglingRefs: Number(reply[14]), + pttlMs: Number(reply[15]), + oldBodyKeyCount: Number(reply[16]), + }; +} + +const bundles = manifest.sessionIds.map(inspectBundle); +const memory = parseInfo("memory"); +const persistence = parseInfo("persistence"); +const containerState = JSON.parse( + runDocker(["inspect", "--format", "{{json .State}}", redisContainer]) +); +const existingBundles = bundles.filter((bundle) => bundle.exists); +const ttlValues = existingBundles.map((bundle) => bundle.pttlMs).filter((ttl) => ttl >= 0); +const totalRawBodyBytes = bundles.reduce((total, bundle) => total + bundle.rawBodyBytes, 0); +const totalDeclaredBytes = bundles.reduce((total, bundle) => total + bundle.declaredTotalBytes, 0); +const totalBodyFieldCount = bundles.reduce((total, bundle) => total + bundle.bodyFieldCount, 0); +const totalOldBodyKeyCount = bundles.reduce((total, bundle) => total + bundle.oldBodyKeyCount, 0); +const totalDanglingRefs = bundles.reduce((total, bundle) => total + bundle.danglingRefs, 0); +const identicalThreeViewRefs = bundles.filter( + (bundle) => + bundle.present.legacy && + bundle.present.before && + bundle.present.after && + bundle.refs.legacy !== "" && + bundle.refs.legacy === bundle.refs.before && + bundle.refs.legacy === bundle.refs.after +).length; +const budgetBytes = expectedResponseBytes * manifest.sessionIds.length; +const invariantFailures = []; + +if (bundles.length !== manifest.waves * manifest.requestsPerWave) { + invariantFailures.push("manifest request count does not match waves times requestsPerWave"); +} +if (mode === "active") { + if (existingBundles.length !== bundles.length) + invariantFailures.push("one or more bundles are missing"); + if (bundles.some((bundle) => bundle.schema !== "1")) invariantFailures.push("unexpected schema"); + if (bundles.some((bundle) => bundle.layout !== "dedup")) + invariantFailures.push("unexpected layout"); + if (bundles.some((bundle) => bundle.overBudget)) + invariantFailures.push("one or more bundles are over budget"); + if (totalBodyFieldCount !== bundles.length) + invariantFailures.push("expected one body field per request"); + if (identicalThreeViewRefs !== bundles.length) { + invariantFailures.push("legacy, before, and after do not share one body ref for every request"); + } + if (totalRawBodyBytes > budgetBytes) + invariantFailures.push("raw body bytes exceed request budget"); + if (bundles.some((bundle) => bundle.rawBodyBytes !== bundle.declaredTotalBytes)) { + invariantFailures.push("raw body bytes differ from declared total_bytes"); + } + if (totalOldBodyKeyCount !== 0) + invariantFailures.push("legacy body keys remain after bundle write"); + if (totalDanglingRefs !== 0) invariantFailures.push("one or more bundle refs are dangling"); +} else { + if (existingBundles.length !== 0) invariantFailures.push("one or more bundles remain after TTL"); + if (totalOldBodyKeyCount !== 0) invariantFailures.push("legacy body keys remain after TTL"); +} +if (containerState.OOMKilled) invariantFailures.push("Redis container was OOM-killed"); +if (containerState.Status !== "running") invariantFailures.push("Redis container is not running"); +if (persistence.rdb_last_bgsave_status !== "ok") invariantFailures.push("last RDB save failed"); + +process.stdout.write( + `${JSON.stringify( + { + mode, + manifest: { + scenarioPrefix: manifest.scenarioPrefix, + waves: manifest.waves, + requestsPerWave: manifest.requestsPerWave, + completedWaves: manifest.completedWaves, + requestSequence: manifest.requestSequence, + sessionCount: manifest.sessionIds.length, + }, + expectedResponseBytes, + budgetBytes, + redis: { + containerState, + memory: { + usedMemory: Number(memory.used_memory || 0), + usedMemoryPeak: Number(memory.used_memory_peak || 0), + }, + persistence: { + rdbSaves: Number(persistence.rdb_saves || 0), + rdbLastBgsaveStatus: persistence.rdb_last_bgsave_status || null, + rdbLastCowSize: Number(persistence.rdb_last_cow_size || 0), + rdbBgsaveInProgress: Number(persistence.rdb_bgsave_in_progress || 0), + rdbLastSaveTime: Number(persistence.rdb_last_save_time || 0), + }, + }, + summary: { + bundleCount: existingBundles.length, + missingBundleCount: bundles.length - existingBundles.length, + totalRawBodyBytes, + totalDeclaredBytes, + totalBodyFieldCount, + identicalThreeViewRefs, + totalOldBodyKeyCount, + totalDanglingRefs, + minPttlMs: ttlValues.length > 0 ? Math.min(...ttlValues) : null, + maxPttlMs: ttlValues.length > 0 ? Math.max(...ttlValues) : null, + invariantFailures, + passed: invariantFailures.length === 0, + }, + bundles, + }, + null, + 2 + )}\n` +); + +if (invariantFailures.length > 0) process.exitCode = 1; diff --git a/tests/load/issue-1408-replay-oom/mock-upstream.cjs b/tests/load/issue-1408-replay-oom/mock-upstream.cjs index d3cbd6598..d1c760680 100644 --- a/tests/load/issue-1408-replay-oom/mock-upstream.cjs +++ b/tests/load/issue-1408-replay-oom/mock-upstream.cjs @@ -4,16 +4,38 @@ 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 responseBytes = process.env.CCH_MOCK_RESPONSE_BYTES + ? parseBoundedInteger( + process.env.CCH_MOCK_RESPONSE_BYTES, + "CCH_MOCK_RESPONSE_BYTES", + 64 * 1024, + 64 * 1024 * 1024 + ) + : Math.round( + parseBoundedNumber(process.env.CCH_MOCK_MIB || "5", "CCH_MOCK_MIB", 0.0625, 64) * 1024 * 1024 + ); +const totalMiB = responseBytes / (1024 * 1024); +const responseMode = parseEnum( + process.env.CCH_MOCK_RESPONSE_MODE || "disconnect", + "CCH_MOCK_RESPONSE_MODE", + ["disconnect", "complete"] +); 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 maxDeltaBytes = 64 * 1024; const counts = new Map(); +const completedCounts = new Map(); +const emittedBytesByScenario = new Map(); + +const sseFramePrefix = 'data: {"type":"response.output_text.delta","delta":"'; +const sseFrameSuffix = '"}\n\n'; +const sseFrameOverheadBytes = Buffer.byteLength(sseFramePrefix + sseFrameSuffix, "utf8"); +const sseCompletedFrame = + 'event: response.completed\ndata: {"type":"response.completed","response":{"status":"completed","usage":{"input_tokens":1,"output_tokens":1}}}\n\n'; function parseBoundedNumber(raw, name, minimum, maximum) { const value = Number(raw); @@ -31,6 +53,44 @@ function parseBoundedInteger(raw, name, minimum, maximum) { return value; } +function parseEnum(raw, name, values) { + if (!values.includes(raw)) { + throw new Error(`${name} must be one of: ${values.join(", ")}`); + } + return raw; +} + +function buildSseFrames(targetBytes, mode = responseMode) { + const terminalFrame = mode === "complete" ? sseCompletedFrame : ""; + const terminalBytes = Buffer.byteLength(terminalFrame, "utf8"); + const deltaBytes = targetBytes - terminalBytes; + if (!Number.isSafeInteger(targetBytes) || deltaBytes < sseFrameOverheadBytes) { + throw new Error( + `targetBytes must be an integer of at least ${sseFrameOverheadBytes + terminalBytes}` + ); + } + + const frames = []; + const maxFrameBytes = sseFrameOverheadBytes + maxDeltaBytes; + let remainingBytes = deltaBytes; + while (remainingBytes > 0) { + const frameBytes = + remainingBytes <= maxFrameBytes + ? remainingBytes + : Math.min(maxFrameBytes, remainingBytes - sseFrameOverheadBytes); + const frame = `${sseFramePrefix}${"x".repeat(frameBytes - sseFrameOverheadBytes)}${sseFrameSuffix}`; + if (Buffer.byteLength(frame, "utf8") !== frameBytes) { + throw new Error("failed to construct an exact-size SSE frame"); + } + frames.push(frame); + remainingBytes -= frameBytes; + } + if (terminalFrame) frames.push(terminalFrame); + return frames; +} + +const responseFrames = buildSseFrames(responseBytes, responseMode); + function readBody(req) { return new Promise((resolve, reject) => { const chunks = []; @@ -70,6 +130,10 @@ const server = http.createServer(async (req, res) => { if (req.method === "GET" && (req.url === "/health" || req.url === "/stats")) { writeJson(res, 200, { counts: Object.fromEntries(counts), + completedCounts: Object.fromEntries(completedCounts), + emittedBytesByScenario: Object.fromEntries(emittedBytesByScenario), + responseBytes, + responseMode, totalMiB, }); return; @@ -77,6 +141,8 @@ const server = http.createServer(async (req, res) => { if (req.method === "POST" && req.url === "/reset") { counts.clear(); + completedCounts.clear(); + emittedBytesByScenario.clear(); writeJson(res, 200, { reset: true }); return; } @@ -106,6 +172,8 @@ const server = http.createServer(async (req, res) => { requestBytes: Buffer.byteLength(raw), scenario, ordinal: counts.get(scenario), + responseBytes, + responseMode, totalMiB, })}\n` ); @@ -117,20 +185,38 @@ const server = http.createServer(async (req, res) => { connection: "keep-alive", }); - const totalFrames = Math.max(1, Math.ceil(totalMiB * framesPerMiB)); let sent = 0; + let emittedBytes = 0; const writeNext = () => { - if (res.destroyed || sent >= totalFrames) return; + if (res.destroyed || sent >= responseFrames.length) return; + const event = responseFrames[sent]; 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); + emittedBytes += Buffer.byteLength(event, "utf8"); + if (sent === responseFrames.length) { + completedCounts.set(scenario, (completedCounts.get(scenario) || 0) + 1); + emittedBytesByScenario.set( + scenario, + (emittedBytesByScenario.get(scenario) || 0) + emittedBytes + ); + process.stdout.write( + `${JSON.stringify({ + event: "response_emitted", + scenario, + emittedBytes, + frames: responseFrames.length, + })}\n` + ); } + const responseFullyEmitted = sent === responseFrames.length; + const continueResponse = () => { + if (responseFullyEmitted && responseMode === "complete") { + res.end(); + return; + } + writeNext(); + }; + if (!res.write(event)) res.once("drain", continueResponse); + else setImmediate(continueResponse); }; writeNext(); }); @@ -146,13 +232,30 @@ function shutdown() { for (const socket of sockets) socket.destroy(); } -process.on("SIGINT", shutdown); -process.on("SIGTERM", shutdown); +if (require.main === module) { + 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` - ); -}); + 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, + responseBytes, + responseMode, + totalMiB, + })}\n` + ); + }); +} + +module.exports = { + buildSseFrames, + responseBytes, + responseMode, + sseCompletedFrame, + sseFrameOverheadBytes, +}; diff --git a/tests/load/issue-1408-replay-oom/run-wave.sh b/tests/load/issue-1408-replay-oom/run-wave.sh index 564fdda6e..f80a7334b 100755 --- a/tests/load/issue-1408-replay-oom/run-wave.sh +++ b/tests/load/issue-1408-replay-oom/run-wave.sh @@ -21,14 +21,46 @@ 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}" +session_manifest="${CCH_SESSION_MANIFEST:-${output}.sessions.json}" +driver_output="${CCH_DRIVER_OUTPUT:-${output}.driver.jsonl}" +redis_evidence_output="${CCH_REDIS_EVIDENCE_OUTPUT:-${output}.redis.json}" +redis_expired_output="${CCH_REDIS_EXPIRED_OUTPUT:-${output}.redis-expired.json}" +response_bytes="${CCH_MOCK_RESPONSE_BYTES:-5242880}" +trigger_rdb_bgsave="${CCH_TRIGGER_RDB_BGSAVE:-false}" +rdb_delay_seconds="${CCH_RDB_DELAY_SECONDS:-70}" +verify_ttl_cleanup="${CCH_VERIFY_TTL_CLEANUP:-false}" +ttl_cleanup_timeout_seconds="${CCH_TTL_CLEANUP_TIMEOUT_SECONDS:-360}" sampler_pid="" +ttl_stdout_tmp="" +ttl_stderr_tmp="" cleanup() { if [ -n "$sampler_pid" ]; then kill "$sampler_pid" 2>/dev/null || true fi + if [ -n "$ttl_stdout_tmp" ]; then + rm -f -- "$ttl_stdout_tmp" + fi + if [ -n "$ttl_stderr_tmp" ]; then + rm -f -- "$ttl_stderr_tmp" + fi } -trap cleanup EXIT INT TERM +trap cleanup EXIT +trap 'exit 130' INT +trap 'exit 143' TERM + +case "$rdb_delay_seconds" in + *[!0-9]*) + printf '%s\n' "CCH_RDB_DELAY_SECONDS must be a non-negative integer" >&2 + exit 2 + ;; +esac +case "$ttl_cleanup_timeout_seconds" in + *[!0-9]*) + printf '%s\n' "CCH_TTL_CLEANUP_TIMEOUT_SECONDS must be a non-negative integer" >&2 + exit 2 + ;; +esac "$script_dir/sample-container.sh" \ "$app_container" \ @@ -38,14 +70,88 @@ trap cleanup EXIT INT TERM "$redis_container" & sampler_pid=$! -"$node_bin" "$script_dir/drive-disconnect-waves.cjs" \ +CCH_SESSION_MANIFEST="$session_manifest" \ + "$node_bin" "$script_dir/drive-disconnect-waves.cjs" \ "$app_url" \ "$mock_stats_url" \ "$scenario_prefix" \ "$waves" \ "$requests_per_wave" \ - "$wave_interval_ms" + "$wave_interval_ms" >"$driver_output" +cat "$driver_output" + +case "$trigger_rdb_bgsave" in + true | 1) + if [ -z "$redis_container" ]; then + printf '%s\n' "CCH_TRIGGER_RDB_BGSAVE requires REDIS_CONTAINER" >&2 + exit 2 + fi + sleep "$rdb_delay_seconds" + docker exec "$redis_container" redis-cli BGSAVE + rdb_wait_started=$(date +%s) + while :; do + rdb_in_progress=$( + docker exec "$redis_container" redis-cli --raw INFO persistence | + awk -F: '$1 == "rdb_bgsave_in_progress" { gsub("\r", "", $2); print $2 }' + ) + [ "$rdb_in_progress" = "0" ] && break + if [ $(( $(date +%s) - rdb_wait_started )) -ge 120 ]; then + printf '%s\n' "timed out waiting for Redis BGSAVE" >&2 + exit 1 + fi + sleep 1 + done + ;; + false | 0) ;; + *) + printf '%s\n' "CCH_TRIGGER_RDB_BGSAVE must be true, false, 1, or 0" >&2 + exit 2 + ;; +esac + +if [ -n "$redis_container" ]; then + "$node_bin" "$script_dir/inspect-redis.cjs" \ + "$redis_container" \ + "$session_manifest" \ + "$response_bytes" >"$redis_evidence_output" + cat "$redis_evidence_output" +fi wait "$sampler_pid" sampler_pid="" + +case "$verify_ttl_cleanup" in + true | 1) + if [ -z "$redis_container" ]; then + printf '%s\n' "CCH_VERIFY_TTL_CLEANUP requires REDIS_CONTAINER" >&2 + exit 2 + fi + ttl_wait_started=$(date +%s) + ttl_stdout_tmp=$(mktemp "${redis_expired_output}.tmp.XXXXXX") + ttl_stderr_tmp=$(mktemp "${redis_expired_output}.stderr.tmp.XXXXXX") + while ! "$node_bin" "$script_dir/inspect-redis.cjs" \ + "$redis_container" \ + "$session_manifest" \ + "$response_bytes" \ + expired >"$ttl_stdout_tmp" 2>"$ttl_stderr_tmp"; do + if [ $(( $(date +%s) - ttl_wait_started )) -ge "$ttl_cleanup_timeout_seconds" ]; then + printf '%s\n' "timed out waiting for session response body TTL cleanup" >&2 + cat "$ttl_stdout_tmp" >&2 + cat "$ttl_stderr_tmp" >&2 + exit 1 + fi + sleep 5 + done + mv "$ttl_stdout_tmp" "$redis_expired_output" + ttl_stdout_tmp="" + rm -f -- "$ttl_stderr_tmp" + ttl_stderr_tmp="" + cat "$redis_expired_output" + ;; + false | 0) ;; + *) + printf '%s\n' "CCH_VERIFY_TTL_CLEANUP must be true, false, 1, or 0" >&2 + exit 2 + ;; +esac 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 index ba202e560..f372b81b1 100755 --- a/tests/load/issue-1408-replay-oom/sample-container.sh +++ b/tests/load/issue-1408-replay-oom/sample-container.sh @@ -13,6 +13,19 @@ samples="${3:-18}" interval="${4:-10}" redis="${5:-}" +container_cgroup_metric() { + container="$1" + metric="$2" + pid=$(docker inspect -f '{{.State.Pid}}' "$container" 2>/dev/null || true) + case "$pid" in + *[!0-9]* | "" | 0) return ;; + esac + cgroup_path=$(awk -F: '$1 == "0" { print $3 }' "/proc/$pid/cgroup" 2>/dev/null || true) + [ -n "$cgroup_path" ] || return + metric_path="/sys/fs/cgroup${cgroup_path}/${metric}" + [ -r "$metric_path" ] && tr -d '\n' <"$metric_path" +} + case "$samples" in *[!0-9]* | 0) printf '%s\n' "SAMPLES must be a positive integer" >&2; exit 2 ;; esac @@ -26,6 +39,8 @@ 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) + app_memory_current=$(container_cgroup_metric "$app" memory.current) + app_memory_peak=$(container_cgroup_metric "$app" memory.peak) 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) @@ -35,15 +50,31 @@ while [ "$i" -lt "$samples" ]; do redis_state="" redis_memory="" + redis_persistence="" + redis_memory_current="" + redis_memory_peak="" + redis_body_bundles="" + redis_replay_owners="" + redis_replay_meta="" + redis_replay_chunks="" if [ -n "$redis" ]; then redis_state=$(docker inspect -f '{{.State.Status}} oom={{.State.OOMKilled}} exit={{.State.ExitCode}}' "$redis" 2>/dev/null || true) + redis_memory_current=$(container_cgroup_metric "$redis" memory.current) + redis_memory_peak=$(container_cgroup_metric "$redis" memory.peak) redis_memory=$(docker exec "$redis" redis-cli --raw INFO memory 2>/dev/null | - grep -E '^(used_memory_human|used_memory_peak_human):' | + grep -E '^(used_memory|used_memory_peak):' | + tr '\n' ',' || true) + redis_persistence=$(docker exec "$redis" redis-cli --raw INFO persistence 2>/dev/null | + grep -E '^(rdb_bgsave_in_progress|rdb_last_bgsave_status|rdb_saves|rdb_last_cow_size|rdb_last_save_time):' | tr '\n' ',' || true) + redis_body_bundles=$(docker exec "$redis" redis-cli --scan --pattern 'session:*:response-bodies:v1' 2>/dev/null | wc -l | tr -d ' ') + redis_replay_owners=$(docker exec "$redis" redis-cli --scan --pattern 'cch:replay:owner:*' 2>/dev/null | wc -l | tr -d ' ') + redis_replay_meta=$(docker exec "$redis" redis-cli --scan --pattern 'cch:replay:meta:*' 2>/dev/null | wc -l | tr -d ' ') + redis_replay_chunks=$(docker exec "$redis" redis-cli --scan --pattern 'cch:replay:chunks:*' 2>/dev/null | wc -l | tr -d ' ') 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]" \ + "sample=$i epoch=$epoch app=[$state] appMemoryCurrent=$app_memory_current appMemoryPeak=$app_memory_peak timeouts=$timeouts backlogs=$backlogs bodySkips=$body_skips memory=$memory lastTask=$active redis=[$redis_state] redisMemoryCurrent=$redis_memory_current redisMemoryPeak=$redis_memory_peak redisMemory=[$redis_memory] redisPersistence=[$redis_persistence] responseBodyBundles=$redis_body_bundles replayOwners=$redis_replay_owners replayMeta=$redis_replay_meta replayChunks=$redis_replay_chunks" \ >>"$output" i=$((i + 1)) [ "$i" -ge "$samples" ] || sleep "$interval" diff --git a/tests/load/issue-1408-replay-oom/start-mock-container.sh b/tests/load/issue-1408-replay-oom/start-mock-container.sh index 3c9c0869e..22b319554 100755 --- a/tests/load/issue-1408-replay-oom/start-mock-container.sh +++ b/tests/load/issue-1408-replay-oom/start-mock-container.sh @@ -1,19 +1,38 @@ #!/bin/sh set -eu -if [ "$#" -lt 3 ] || [ "$#" -gt 5 ]; then +if [ "$#" -lt 3 ] || [ "$#" -gt 6 ]; then printf '%s\n' \ - "usage: start-mock-container.sh CONTAINER NETWORK HOST_PORT [PAYLOAD_MIB] [NODE_IMAGE]" >&2 + "usage: start-mock-container.sh CONTAINER NETWORK HOST_PORT [RESPONSE_BYTES] [RESPONSE_MODE] [NODE_IMAGE]" >&2 exit 2 fi container="$1" network="$2" host_port="$3" -payload_mib="${4:-8}" -node_image="${5:-node:22-alpine}" +response_bytes="${4:-5242880}" +response_mode="${5:-disconnect}" +node_image="${6:-node:22-alpine}" script_dir=$(CDPATH= cd -- "$(dirname -- "$0")" && pwd) +case "$response_bytes" in + *[!0-9]* | "") + printf '%s\n' "RESPONSE_BYTES must be an integer between 65536 and 67108864" >&2 + exit 2 + ;; +esac +if [ "$response_bytes" -lt 65536 ] || [ "$response_bytes" -gt 67108864 ]; then + printf '%s\n' "RESPONSE_BYTES must be an integer between 65536 and 67108864" >&2 + exit 2 +fi +case "$response_mode" in + disconnect | complete) ;; + *) + printf '%s\n' "RESPONSE_MODE must be disconnect or complete" >&2 + exit 2 + ;; +esac + if docker container inspect "$container" >/dev/null 2>&1; then printf '%s\n' "container already exists: $container" >&2 exit 1 @@ -23,7 +42,8 @@ docker run -d \ --name "$container" \ --network "$network" \ -e CCH_MOCK_PORT=3001 \ - -e CCH_MOCK_MIB="$payload_mib" \ + -e CCH_MOCK_RESPONSE_BYTES="$response_bytes" \ + -e CCH_MOCK_RESPONSE_MODE="$response_mode" \ -p "127.0.0.1:$host_port:3001" \ -v "$script_dir/mock-upstream.cjs:/fixture/mock-upstream.cjs:ro" \ "$node_image" \ 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 da16baf13..a5e4cbe33 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 originalDedupEnabled = process.env.SESSION_RESPONSE_BODY_DEDUP_ENABLED; const originalMaxBytes = process.env.SESSION_RESPONSE_BODY_MAX_BYTES; afterEach(() => { @@ -16,6 +17,11 @@ describe("EnvSchema - STORE_SESSION_RESPONSE_BODY", () => { } else { process.env.SESSION_RESPONSE_BODY_MAX_BYTES = originalMaxBytes; } + if (originalDedupEnabled === undefined) { + delete process.env.SESSION_RESPONSE_BODY_DEDUP_ENABLED; + } else { + process.env.SESSION_RESPONSE_BODY_DEDUP_ENABLED = originalDedupEnabled; + } }); it("should default to true when not set", () => { @@ -48,6 +54,21 @@ describe("EnvSchema - STORE_SESSION_RESPONSE_BODY", () => { expect(result.STORE_SESSION_RESPONSE_BODY).toBe(true); }); + it("defaults response body deduplication to the rollout-safe reader-only mode", () => { + delete process.env.SESSION_RESPONSE_BODY_DEDUP_ENABLED; + expect(EnvSchema.parse(process.env).SESSION_RESPONSE_BODY_DEDUP_ENABLED).toBe(false); + }); + + it.each([ + ["true", true], + ["1", true], + ["false", false], + ["0", false], + ])("parses SESSION_RESPONSE_BODY_DEDUP_ENABLED=%s", (value, expected) => { + process.env.SESSION_RESPONSE_BODY_DEDUP_ENABLED = value; + expect(EnvSchema.parse(process.env).SESSION_RESPONSE_BODY_DEDUP_ENABLED).toBe(expected); + }); + it("defaults the response body limit to 5 MiB", () => { delete process.env.SESSION_RESPONSE_BODY_MAX_BYTES; const result = EnvSchema.parse(process.env); diff --git a/tests/unit/lib/session-manager-terminate-session.test.ts b/tests/unit/lib/session-manager-terminate-session.test.ts index 237ac8c7d..3ae3aa017 100644 --- a/tests/unit/lib/session-manager-terminate-session.test.ts +++ b/tests/unit/lib/session-manager-terminate-session.test.ts @@ -40,6 +40,7 @@ describe("SessionManager.terminateSession", () => { pipelineRef = { del: vi.fn(() => pipelineRef), + eval: vi.fn(() => pipelineRef), zrem: vi.fn(() => pipelineRef), hdel: vi.fn(() => pipelineRef), exec: vi.fn(async () => [[null, 1]]), @@ -111,6 +112,25 @@ describe("SessionManager.terminateSession", () => { expect(pipelineRef.zrem).not.toHaveBeenCalledWith(getUserActiveSessionsKey(123), sessionId); }); + it("full termination atomically deletes all indexed response body bundles", async () => { + const sessionId = "sess_response_body_cleanup"; + const { SessionManager } = await import("@/lib/session-manager"); + + await expect(SessionManager.terminateSession(sessionId)).resolves.toBe(true); + + expect(pipelineRef.eval).toHaveBeenCalledWith( + expect.stringContaining("cch:session-response-bundle:delete-session:v1"), + 5, + `session:${sessionId}:response-body-bundles:v1`, + `session:${sessionId}:response-body-generation:v1`, + `session:${sessionId}:response`, + `session:${sessionId}:req:1:snapshot:response:before:body`, + `session:${sessionId}:req:1:snapshot:response:after:body`, + 300, + expect.any(String) + ); + }); + it("fails closed when the owner key lookup rejects during scoped termination", async () => { const sessionId = "sess_owner_lookup_failure"; redisClientRef.get.mockImplementation(async (key: string) => { @@ -260,6 +280,7 @@ describe("SessionManager.terminateSession", () => { expect(pipelineRef.zrem).toHaveBeenCalledWith("provider:42:active_sessions", sessionId); expect(pipelineRef.hdel).toHaveBeenCalledWith("provider:42:active_session_refs", sessionId); expect(pipelineRef.del).not.toHaveBeenCalled(); + expect(pipelineRef.eval).not.toHaveBeenCalled(); expect(pipelineRef.zrem).not.toHaveBeenCalledWith(getGlobalActiveSessionsKey(), sessionId); expect(pipelineRef.zrem).not.toHaveBeenCalledWith(getKeyActiveSessionsKey(7), sessionId); expect(pipelineRef.zrem).not.toHaveBeenCalledWith(getUserActiveSessionsKey(123), sessionId); @@ -293,6 +314,7 @@ describe("SessionManager.terminateSession", () => { await expect(SessionManager.terminateSession(sessionId)).resolves.toBe(false); expect(pipelineRef.exec).not.toHaveBeenCalled(); expect(pipelineRef.del).not.toHaveBeenCalled(); + expect(pipelineRef.eval).not.toHaveBeenCalled(); }); it("preserves shared Session state after scoped legacy termination linearizes on the old Provider", async () => { @@ -325,6 +347,7 @@ describe("SessionManager.terminateSession", () => { expect(pipelineRef.zrem).toHaveBeenCalledWith("provider:42:active_sessions", sessionId); expect(pipelineRef.hdel).toHaveBeenCalledWith("provider:42:active_session_refs", sessionId); expect(pipelineRef.del).not.toHaveBeenCalled(); + expect(pipelineRef.eval).not.toHaveBeenCalled(); expect(pipelineRef.zrem).not.toHaveBeenCalledWith(getGlobalActiveSessionsKey(), sessionId); expect(pipelineRef.zrem).not.toHaveBeenCalledWith(getKeyActiveSessionsKey(7), sessionId); expect(pipelineRef.zrem).not.toHaveBeenCalledWith(getUserActiveSessionsKey(123), sessionId); diff --git a/tests/unit/load/issue-1408-mock-upstream.test.ts b/tests/unit/load/issue-1408-mock-upstream.test.ts new file mode 100644 index 000000000..b54f226fd --- /dev/null +++ b/tests/unit/load/issue-1408-mock-upstream.test.ts @@ -0,0 +1,56 @@ +import { createRequire } from "node:module"; +import { describe, expect, it } from "vitest"; + +type MockFixtureModule = { + buildSseFrames: (targetBytes: number, mode?: "disconnect" | "complete") => string[]; + sseCompletedFrame: string; + sseFrameOverheadBytes: number; +}; + +const require = createRequire(import.meta.url); +const { buildSseFrames, sseCompletedFrame, sseFrameOverheadBytes } = + require("../../load/issue-1408-replay-oom/mock-upstream.cjs") as MockFixtureModule; + +function assertValidExactSseBody(targetBytes: number): void { + const frames = buildSseFrames(targetBytes); + expect(Buffer.byteLength(frames.join(""), "utf8")).toBe(targetBytes); + + for (const frame of frames) { + expect(frame.startsWith("data: ")).toBe(true); + expect(frame.endsWith("\n\n")).toBe(true); + expect(JSON.parse(frame.slice(6, -2))).toEqual({ + type: "response.output_text.delta", + delta: expect.any(String), + }); + } +} + +describe("issue #1408 mock upstream response sizing", () => { + it("builds a valid SSE body at the exact 5 MiB response limit", () => { + assertValidExactSseBody(5 * 1024 * 1024); + }); + + it("keeps the final frame valid around frame-size remainders", () => { + assertValidExactSseBody(sseFrameOverheadBytes); + assertValidExactSseBody(sseFrameOverheadBytes + 1); + assertValidExactSseBody(64 * 1024 + sseFrameOverheadBytes + 1); + }); + + it("rejects targets too small to contain one complete SSE frame", () => { + expect(() => buildSseFrames(sseFrameOverheadBytes - 1)).toThrow( + /targetBytes must be an integer/ + ); + }); + + it("builds an exact-size terminal SSE body for complete-response acceptance", () => { + const targetBytes = 5 * 1024 * 1024; + const frames = buildSseFrames(targetBytes, "complete"); + + expect(Buffer.byteLength(frames.join(""), "utf8")).toBe(targetBytes); + expect(frames.at(-1)).toBe(sseCompletedFrame); + expect(JSON.parse(frames.at(-1)?.split("data: ")[1].trim() ?? "{}")).toMatchObject({ + type: "response.completed", + response: { status: "completed" }, + }); + }); +}); diff --git a/tests/unit/proxy/issue-1408-load-fixture.test.ts b/tests/unit/proxy/issue-1408-load-fixture.test.ts index 55a8478fd..2c4851a26 100644 --- a/tests/unit/proxy/issue-1408-load-fixture.test.ts +++ b/tests/unit/proxy/issue-1408-load-fixture.test.ts @@ -4,7 +4,7 @@ import { spawnSync, type ChildProcessWithoutNullStreams, } from "node:child_process"; -import { existsSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import { existsSync, mkdtempSync, readFileSync, readdirSync, rmSync, writeFileSync } from "node:fs"; import http from "node:http"; import { tmpdir } from "node:os"; import path from "node:path"; @@ -15,18 +15,44 @@ const nodeScripts = ["mock-upstream.cjs", "drive-disconnect-waves.cjs", "memory- 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 = [ +const fixtureEnvironmentKeys = [ + "CCH_ABORT_DELAY_MS", + "CCH_API_KEY", + "CCH_API_KEY_FILE", + "CCH_COMPLETION_TIMEOUT_MS", + "CCH_DRIVER_OUTPUT", + "CCH_MOCK_RECEIPT_TIMEOUT_MS", "CCH_MOCK_HOST", "CCH_MOCK_PORT", "CCH_MOCK_MIB", "CCH_MOCK_MAX_REQUEST_BYTES", + "CCH_MOCK_RESPONSE_BYTES", + "CCH_MOCK_RESPONSE_MODE", + "CCH_RDB_DELAY_SECONDS", + "CCH_REDIS_EVIDENCE_OUTPUT", + "CCH_REDIS_EXPIRED_OUTPUT", + "CCH_REQUEST_MODE", + "CCH_REQUEST_MODEL", + "CCH_REQUESTS_PER_WAVE", + "CCH_SAMPLE_INTERVAL_SECONDS", + "CCH_SAMPLES", + "CCH_SESSION_MANIFEST", + "CCH_TRIGGER_RDB_BGSAVE", + "CCH_TTL_CLEANUP_TIMEOUT_SECONDS", + "CCH_VERIFY_TTL_CLEANUP", + "CCH_WAVE_INTERVAL_MS", + "CCH_WAVES", ] as const; -function createMockEnvironment(overrides: Record = {}): NodeJS.ProcessEnv { +function createFixtureEnvironment(overrides: Record = {}): NodeJS.ProcessEnv { const environment = { ...process.env }; - for (const key of mockEnvironmentKeys) delete environment[key]; + for (const key of fixtureEnvironmentKeys) delete environment[key]; + return { ...environment, ...overrides }; +} + +function createMockEnvironment(overrides: Record = {}): NodeJS.ProcessEnv { return { - ...environment, + ...createFixtureEnvironment(), CCH_MOCK_HOST: "127.0.0.1", CCH_MOCK_PORT: "0", CCH_MOCK_MIB: "0.0625", @@ -158,6 +184,30 @@ function postAndAbort(url: URL, body: string): Promise { }); } +function postAndConsume(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) => { + const chunks: Buffer[] = []; + response.on("data", (chunk: Buffer) => chunks.push(chunk)); + response.once("end", () => resolve(Buffer.concat(chunks))); + response.once("aborted", () => reject(new Error("response aborted"))); + response.once("error", reject); + } + ); + request.once("error", reject); + request.end(body); + }); +} + function waitForExit( child: ChildProcessWithoutNullStreams ): Promise<{ code: number | null; stderr: string }> { @@ -241,14 +291,13 @@ function runStartMockWithFakeCommands( ], { encoding: "utf8", - env: { - ...process.env, + env: createFixtureEnvironment({ 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, - }, + }), } ); } @@ -366,6 +415,29 @@ describe("issue #1408 load fixture", () => { }); }); + it("reuses exact-size frames across concurrent complete responses", async () => { + const responseBytes = 64 * 1024; + const { baseUrl } = await startMock({ + CCH_MOCK_RESPONSE_BYTES: String(responseBytes), + CCH_MOCK_RESPONSE_MODE: "complete", + }); + const body = JSON.stringify({ input: "CCH_SCENARIO_cached-frames", stream: true }); + + const responses = await Promise.all([ + postAndConsume(new URL("/v1/responses", baseUrl), body), + postAndConsume(new URL("/v1/responses", baseUrl), body), + ]); + + expect(responses.map((response) => response.byteLength)).toEqual([ + responseBytes, + responseBytes, + ]); + await expect(getJson(new URL("/stats", baseUrl))).resolves.toMatchObject({ + completedCounts: { "cached-frames": 2 }, + emittedBytesByScenario: { "cached-frames": responseBytes * 2 }, + }); + }); + 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"; @@ -448,7 +520,7 @@ describe("issue #1408 load fixture", () => { ], { encoding: "utf8", - env: { ...process.env, CCH_API_KEY: "", CCH_API_KEY_FILE: "" }, + env: createFixtureEnvironment({ CCH_API_KEY: "", CCH_API_KEY_FILE: "" }), } ); @@ -482,7 +554,10 @@ describe("issue #1408 load fixture", () => { "0", ], { - env: { ...process.env, CCH_API_KEY: "fixture-key", CCH_API_KEY_FILE: "" }, + env: createFixtureEnvironment({ + CCH_API_KEY: "fixture-key", + CCH_API_KEY_FILE: "", + }), stdio: ["ignore", "pipe", "pipe"], } ); @@ -496,6 +571,158 @@ describe("issue #1408 load fixture", () => { } }); + it("reports complete-mode request failures through the controlled receipt timeout", async () => { + const server = http.createServer((_request, response) => { + response.writeHead(200, { "content-type": "application/json" }); + response.end('{"counts":{}}'); + }); + 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`, + "early-request-failure", + "1", + "1", + "0", + ], + { + env: createFixtureEnvironment({ + CCH_API_KEY: "fixture-key", + CCH_API_KEY_FILE: "", + CCH_REQUEST_MODE: "complete", + CCH_MOCK_RECEIPT_TIMEOUT_MS: "100", + }), + stdio: ["ignore", "pipe", "pipe"], + } + ); + children.add(child); + + const result = await waitForExit(child); + expect(result.code).toBe(1); + expect(result.stderr).toContain("mock receipt timeout for early-request-failure-0"); + expect(result.stderr).not.toContain("UnhandledPromiseRejection"); + expect(result.stderr).not.toContain("ECONNREFUSED"); + } finally { + await new Promise((resolve) => server.close(() => resolve())); + } + }); + + posixIt.each([ + ["CCH_RDB_DELAY_SECONDS", "invalid"], + ["CCH_RDB_DELAY_SECONDS", "-1"], + ["CCH_RDB_DELAY_SECONDS", "1.5"], + ["CCH_TTL_CLEANUP_TIMEOUT_SECONDS", "invalid"], + ["CCH_TTL_CLEANUP_TIMEOUT_SECONDS", "-1"], + ["CCH_TTL_CLEANUP_TIMEOUT_SECONDS", "1.5"], + ])("rejects invalid run-wave integer configuration: %s=%s", (name, value) => { + const directory = mkdtempSync(path.join(tmpdir(), "cch1408-run-wave-config-")); + try { + const result = spawnSync( + "sh", + [ + path.join(fixtureDir, "run-wave.sh"), + "http://127.0.0.1:1", + "http://127.0.0.1:2/stats", + "invalid-config", + "app-container", + path.join(directory, "samples.log"), + ], + { + encoding: "utf8", + env: createFixtureEnvironment({ + CCH_RDB_DELAY_SECONDS: "0", + CCH_TTL_CLEANUP_TIMEOUT_SECONDS: "0", + [name]: value, + }), + } + ); + + expect(result.status).toBe(2); + expect(result.stderr).toContain(`${name} must be a non-negative integer`); + } finally { + rmSync(directory, { recursive: true, force: true }); + } + }); + + posixIt("preserves TTL inspection diagnostics and removes temporary files on timeout", () => { + const directory = mkdtempSync(path.join(tmpdir(), "cch1408-run-wave-ttl-")); + const output = path.join(directory, "samples.log"); + const manifest = path.join(directory, "sessions.json"); + const expiredOutput = path.join(directory, "redis-expired.json"); + const fakeNode = path.join(directory, "fake-node"); + try { + writeFileSync(path.join(directory, "docker"), "#!/bin/sh\nexit 0\n", { mode: 0o755 }); + writeFileSync( + fakeNode, + `#!/bin/sh +case "$1" in + *drive-disconnect-waves.cjs) + printf '%s\\n' '{"sessionIds":[],"requestSequence":1}' > "$CCH_SESSION_MANIFEST" + ;; + *inspect-redis.cjs) + if [ "\${5:-active}" = "expired" ]; then + printf '%s\\n' '{"passed":false}' + printf '%s\\n' 'fixture Redis inspection failed' >&2 + exit 1 + fi + printf '%s\\n' '{"passed":true}' + ;; + *) exit 2 ;; +esac +`, + { mode: 0o755 } + ); + + const result = spawnSync( + "sh", + [ + path.join(fixtureDir, "run-wave.sh"), + "http://127.0.0.1:1", + "http://127.0.0.1:2/stats", + "ttl-diagnostics", + "app-container", + output, + "redis-container", + ], + { + encoding: "utf8", + env: createFixtureEnvironment({ + PATH: `${directory}:${process.env.PATH ?? ""}`, + NODE_BIN: fakeNode, + CCH_SAMPLES: "1", + CCH_SAMPLE_INTERVAL_SECONDS: "0", + CCH_SESSION_MANIFEST: manifest, + CCH_REDIS_EXPIRED_OUTPUT: expiredOutput, + CCH_TRIGGER_RDB_BGSAVE: "false", + CCH_RDB_DELAY_SECONDS: "0", + CCH_VERIFY_TTL_CLEANUP: "true", + CCH_TTL_CLEANUP_TIMEOUT_SECONDS: "0", + CCH_DRIVER_OUTPUT: path.join(directory, "driver.jsonl"), + CCH_REDIS_EVIDENCE_OUTPUT: path.join(directory, "redis.json"), + }), + } + ); + + expect(result.status).toBe(1); + expect(result.stderr).toContain("timed out waiting for session response body TTL cleanup"); + expect(result.stderr).toContain('{"passed":false}'); + expect(result.stderr).toContain("fixture Redis inspection failed"); + expect(readdirSync(directory).filter((name) => name.includes(".tmp."))).toEqual([]); + } finally { + rmSync(directory, { recursive: true, force: true }); + } + }); + it.each([ ["unsupported URL protocol", ["ftp://127.0.0.1", "http://127.0.0.1/stats", "valid"], "APP_URL"], [ @@ -510,7 +737,10 @@ describe("issue #1408 load fixture", () => { [path.join(fixtureDir, "drive-disconnect-waves.cjs"), ...args], { encoding: "utf8", - env: { ...process.env, CCH_API_KEY: "fixture-key", CCH_API_KEY_FILE: "" }, + env: createFixtureEnvironment({ + CCH_API_KEY: "fixture-key", + CCH_API_KEY_FILE: "", + }), } ); diff --git a/tests/unit/proxy/pricing-no-price.test.ts b/tests/unit/proxy/pricing-no-price.test.ts index 78e2cd4f7..0d0f07910 100644 --- a/tests/unit/proxy/pricing-no-price.test.ts +++ b/tests/unit/proxy/pricing-no-price.test.ts @@ -45,6 +45,7 @@ vi.mock("@/lib/session-manager", () => ({ SessionManager: { updateSessionUsage: vi.fn(async () => {}), storeSessionResponse: vi.fn(async () => {}), + storeSessionResponseBodySet: vi.fn(async () => {}), extractCodexPromptCacheKey: vi.fn(), updateSessionWithCodexCacheKey: vi.fn(async () => {}), }, diff --git a/tests/unit/proxy/response-handler-abort-listener-cleanup.test.ts b/tests/unit/proxy/response-handler-abort-listener-cleanup.test.ts index 98b0288d7..8d2682925 100644 --- a/tests/unit/proxy/response-handler-abort-listener-cleanup.test.ts +++ b/tests/unit/proxy/response-handler-abort-listener-cleanup.test.ts @@ -81,6 +81,7 @@ vi.mock("@/lib/session-manager", () => ({ SessionManager: { clearSessionProvider: vi.fn(), storeSessionResponse: vi.fn(), + storeSessionResponseBodySet: vi.fn(async () => undefined), updateSessionUsage: vi.fn(), storeSessionRequestPhaseSnapshot: vi.fn(), storeSessionResponsePhaseSnapshot: vi.fn(), diff --git a/tests/unit/proxy/response-handler-client-abort-drain.test.ts b/tests/unit/proxy/response-handler-client-abort-drain.test.ts index e97233285..caf4026ae 100644 --- a/tests/unit/proxy/response-handler-client-abort-drain.test.ts +++ b/tests/unit/proxy/response-handler-client-abort-drain.test.ts @@ -142,6 +142,7 @@ vi.mock("@/lib/session-manager", () => ({ })), extractCodexPromptCacheKey: vi.fn(), storeSessionResponse: vi.fn(async () => undefined), + storeSessionResponseBodySet: vi.fn(async () => undefined), storeSessionRequestPhaseSnapshot: vi.fn(), storeSessionResponsePhaseSnapshot: vi.fn(), storeSessionRequestHeaders: vi.fn(), diff --git a/tests/unit/proxy/response-handler-endpoint-circuit-isolation.test.ts b/tests/unit/proxy/response-handler-endpoint-circuit-isolation.test.ts index 2fb9c50d5..6a99520ec 100644 --- a/tests/unit/proxy/response-handler-endpoint-circuit-isolation.test.ts +++ b/tests/unit/proxy/response-handler-endpoint-circuit-isolation.test.ts @@ -76,6 +76,7 @@ vi.mock("@/lib/session-manager", () => ({ SessionManager: { updateSessionUsage: vi.fn(), storeSessionResponse: vi.fn(), + storeSessionResponseBodySet: vi.fn(async () => undefined), clearSessionProvider: vi.fn(), clearVersionedSessionProvider: vi.fn(), compareAndSetSessionProvider: vi.fn(), diff --git a/tests/unit/proxy/response-handler-gemini-stream-passthrough-timeouts.test.ts b/tests/unit/proxy/response-handler-gemini-stream-passthrough-timeouts.test.ts index 2524cc5ce..c1a0df38c 100644 --- a/tests/unit/proxy/response-handler-gemini-stream-passthrough-timeouts.test.ts +++ b/tests/unit/proxy/response-handler-gemini-stream-passthrough-timeouts.test.ts @@ -130,6 +130,7 @@ vi.mock("@/repository/model-price", () => ({ vi.mock("@/lib/session-manager", () => ({ SessionManager: { storeSessionResponse: vi.fn(), + storeSessionResponseBodySet: vi.fn(async () => undefined), updateSessionUsage: vi.fn(async () => undefined), clearSessionProvider: vi.fn(), updateSessionBindingSmart: vi.fn(async () => ({ updated: false, reason: "test" })), diff --git a/tests/unit/proxy/response-handler-lease-decrement.test.ts b/tests/unit/proxy/response-handler-lease-decrement.test.ts index 0f4716200..ae6b094e3 100644 --- a/tests/unit/proxy/response-handler-lease-decrement.test.ts +++ b/tests/unit/proxy/response-handler-lease-decrement.test.ts @@ -81,6 +81,7 @@ vi.mock("@/lib/session-manager", () => ({ SessionManager: { updateSessionUsage: vi.fn(async () => undefined), storeSessionResponse: vi.fn(), + storeSessionResponseBodySet: vi.fn(async () => undefined), storeSessionResponsePhaseSnapshot: vi.fn(async () => undefined), extractCodexPromptCacheKey: vi.fn(), updateSessionWithCodexCacheKey: vi.fn(), @@ -641,12 +642,10 @@ describe("Lease Budget Decrement after trackCostToRedis", () => { .mock.calls.filter(([calledTaskId]) => calledTaskId === taskId); expect(touchCalls.length).toBeGreaterThanOrEqual(2); expect(cloneSpy).toHaveBeenCalledTimes(1); - expect(SessionManager.storeSessionResponsePhaseSnapshot).toHaveBeenCalledWith( + expect(SessionManager.storeSessionResponseBodySet).toHaveBeenCalledWith( session.sessionId, - "after", expect.objectContaining({ - body: expect.stringContaining('"type":"message"'), - meta: expect.objectContaining({ statusCode: 200 }), + after: expect.stringContaining('"type":"message"'), }), session.requestSequence, 456 diff --git a/tests/unit/proxy/response-handler-non200.test.ts b/tests/unit/proxy/response-handler-non200.test.ts index 666de4da3..431d75d34 100644 --- a/tests/unit/proxy/response-handler-non200.test.ts +++ b/tests/unit/proxy/response-handler-non200.test.ts @@ -76,6 +76,7 @@ vi.mock("@/lib/session-manager", () => ({ SessionManager: { updateSessionUsage: vi.fn(), storeSessionResponse: vi.fn(), + storeSessionResponseBodySet: vi.fn(async () => undefined), extractCodexPromptCacheKey: vi.fn(), updateSessionWithCodexCacheKey: vi.fn(), }, diff --git a/tests/unit/proxy/warmup-guard.test.ts b/tests/unit/proxy/warmup-guard.test.ts index a42df8ef6..78d461df8 100644 --- a/tests/unit/proxy/warmup-guard.test.ts +++ b/tests/unit/proxy/warmup-guard.test.ts @@ -6,7 +6,7 @@ const getCachedSystemSettingsMock = vi.fn(); const dbInsertValuesMock = vi.fn(); const dbInsertMock = vi.fn(() => ({ values: dbInsertValuesMock })); -const storeSessionResponseMock = vi.fn(); +const storeSessionResponseBodySetMock = vi.fn(); const storeSessionResponseHeadersMock = vi.fn(); const storeSessionUpstreamRequestMetaMock = vi.fn(); const storeSessionUpstreamResponseMetaMock = vi.fn(); @@ -26,7 +26,7 @@ vi.mock("@/drizzle/db", () => ({ vi.mock("@/lib/session-manager", () => ({ SessionManager: { - storeSessionResponse: storeSessionResponseMock, + storeSessionResponseBodySet: storeSessionResponseBodySetMock, storeSessionResponseHeaders: storeSessionResponseHeadersMock, storeSessionUpstreamRequestMeta: storeSessionUpstreamRequestMetaMock, storeSessionUpstreamResponseMeta: storeSessionUpstreamResponseMetaMock, @@ -82,7 +82,7 @@ beforeEach(() => { enableHighConcurrencyMode: false, }); dbInsertValuesMock.mockResolvedValue(undefined); - storeSessionResponseMock.mockResolvedValue(undefined); + storeSessionResponseBodySetMock.mockResolvedValue(undefined); storeSessionResponseHeadersMock.mockResolvedValue(undefined); storeSessionUpstreamRequestMetaMock.mockResolvedValue(undefined); storeSessionUpstreamResponseMetaMock.mockResolvedValue(undefined); @@ -141,10 +141,10 @@ describe("ProxyWarmupGuard.ensure", () => { }) ); - expect(storeSessionResponseMock).toHaveBeenCalledTimes(1); - expect(storeSessionResponseMock).toHaveBeenCalledWith( + expect(storeSessionResponseBodySetMock).toHaveBeenCalledTimes(1); + expect(storeSessionResponseBodySetMock).toHaveBeenCalledWith( "session_test", - expect.any(String), + { legacy: expect.any(String) }, 2, 456 ); @@ -209,7 +209,7 @@ describe("ProxyWarmupGuard.ensure", () => { expect(result).not.toBeNull(); expect(result?.status).toBe(200); - expect(storeSessionResponseMock).not.toHaveBeenCalled(); + expect(storeSessionResponseBodySetMock).not.toHaveBeenCalled(); expect(storeSessionResponseHeadersMock).not.toHaveBeenCalled(); expect(storeSessionUpstreamRequestMetaMock).not.toHaveBeenCalled(); expect(storeSessionUpstreamResponseMetaMock).not.toHaveBeenCalled();