perf: optimize full-path concurrency and resource ownership - #1341
Merged
Conversation
📝 WalkthroughWalkthrough本次变更重构数据库连接池、代理流式处理、终态持久化、Redis 计费结算、异步任务关停和 WebSocket 传输控制，并新增大量单元与集成测试覆盖相关竞态、背压、超时和恢复路径。 Changes运行时与基础设施
Estimated code review effort: 5 (Critical) | ~180 minutes Possibly related PRs
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
github-actions
Bot
added
enhancement
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Code Review
This pull request introduces significant improvements to the message request persistence layer, including a durable write API, refined database connection pool management with lane isolation, and a robust shutdown orchestration process. My review identified two issues: a premature publication of status rollups for non-durable writes in updateMessageRequestDetails, and a redundant call to publishCommit in updateMessageRequestDetailsDurably after the await on the durable write promise.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| if (shouldQueuePublicStatusRollup) { | ||
| queuePublicStatusRollupForFinalDetails(id, details); | ||
| } | ||
| publishCommittedMessageRequestDetails(id, details); |
Contributor
There was a problem hiding this comment.
This call to publishCommittedMessageRequestDetails is premature. enqueueMessageRequestUpdate is a fire-and-forget operation for the async writer, meaning the update is only buffered and not guaranteed to be committed to the database (it could be dropped on overflow or fail during the batch write). Publishing the commit to the public status rollup at this stage can lead to inconsistent data if the write never completes. This publication should only occur for writes that are confirmed to be durable.
| ...options, | ||
| onCommitted: publishCommit, | ||
| }); | ||
| publishCommit(details); |
Contributor
There was a problem hiding this comment.
This call to publishCommit(details) appears to be redundant. The onCommitted callback passed to enqueueMessageRequestUpdateDurably is designed to be executed once the write is confirmed as part of a successful batch. The await on the preceding line ensures this function doesn't proceed until the durable write promise settles. Calling publishCommit again here is unnecessary. While the idempotency guard within publishCommit prevents duplicate rollups, this line should be removed for clarity and correctness.
Contributor
🧪 测试结果
总体结果: ✅ 所有测试通过 |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 13
Note
Due to the large number of review comments, Critical, Major severity comments were prioritized as inline comments.
🟡 Minor comments (11)
tests/unit/proxy/proxy-forwarder-retry-limit.test.ts-328-340 (1)
328-340: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win不要 mock 掉本测试要验证的 admission 分类。
categorizeErrorAsync被直接固定为数值5后，构造的DbPoolAdmissionErrorcause 完全不影响结果；cause 链识别失效时该测试仍会通过。请调用真实分类器，并使用命名枚举而非序号。🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/unit/proxy/proxy-forwarder-retry-limit.test.ts` around lines 328 - 340, Update the test around categorizeErrorAsync to use the real error classifier instead of mocking it to numeric value 5, so the DbPoolAdmissionError cause chain is actually exercised. Assert or configure the expected classification with the named ErrorCategory enum member rather than a positional number, while preserving the existing wrappedError and doForward rejection setup.tests/unit/proxy/terminal-outcome-contract.test.ts-225-249 (1)
225-249: 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win确保失败路径也停止全局 message writer。
stopMessageRequestWriteBuffer()只在成功路径执行；前置断言失败时，模块级 writer 会残留并污染或挂起后续测试。请用finally先释放releaseCommit，再等待并停止 writer。🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/unit/proxy/terminal-outcome-contract.test.ts` around lines 225 - 249, 更新该测试的清理流程，将 releaseCommit.resolve()、等待 flushPromise 以及 stopMessageRequestWriteBuffer() 放入 finally 中，确保前置断言失败时也会释放提交并停止全局 message writer；保留现有成功路径断言和 handlePromise 响应验证。tests/unit/proxy/response-handler-gemini-terminal.test.ts-119-123 (1)
119-123: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win不要吞掉托管终态任务的拒绝。
当前 helper 会忽略所有 rejected settlement，因此取消、idle reset 和 durable fallback 测试可能在后台任务失败时仍通过。请像同批其他测试一样汇总并抛出拒绝原因。
建议修改
async function settleTasks(): Promise<void> { while (mocks.tasks.length > 0) { - await Promise.allSettled(mocks.tasks.splice(0, mocks.tasks.length)); + const settlements = await Promise.allSettled( + mocks.tasks.splice(0, mocks.tasks.length) + ); + const errors = settlements + .filter( + (result): result is PromiseRejectedResult => + result.status === "rejected" + ) + .map((result) => result.reason); + if (errors.length > 0) { + throw new AggregateError(errors, "Unexpected terminal task rejection"); + } } }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/unit/proxy/response-handler-gemini-terminal.test.ts` around lines 119 - 123, 更新 settleTasks，使每批 mocks.tasks 的托管任务拒绝不会被忽略：汇总 Promise.allSettled 的 rejected 结果，并在存在拒绝时抛出其原因；继续等待后续批次任务，保留当前清空并重复处理动态新增任务的行为。tests/integration/billing-model-source.test.ts-26-33 (1)
26-33: 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win清理前应观察被中止的异步任务。
当前
beforeEach在 abort 后立即清空 Promise 引用；遗留任务若随后拒绝或继续修改 mock，会产生未处理拒绝或跨用例污染。请在注册时附加拒绝观察，并在重置状态前等待已中止任务 settled。建议修改
asyncTasks.push(promise); asyncTaskControllers.set(promise, controller); + void promise.catch(() => undefined); return controller; @@ -beforeEach(() => { +beforeEach(async () => { vi.clearAllMocks(); asyncTaskAdmissionOpen = false; + const pendingTasks = asyncTasks.splice(0, asyncTasks.length); for (const controller of asyncTaskControllers.values()) { controller.abort(); } - asyncTasks.splice(0, asyncTasks.length); + await Promise.allSettled(pendingTasks); asyncTaskControllers.clear(); asyncTaskAdmissionOpen = true;Also applies to: 121-129
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/integration/billing-model-source.test.ts` around lines 26 - 33, 在异步任务注册逻辑中为每个 promise 附加拒绝观察，避免任务后续拒绝形成未处理拒绝；更新 beforeEach 的清理流程，在清空 asyncTasks 和 asyncTaskControllers 前等待所有已中止任务 settled，确保遗留任务不会跨用例继续运行或修改 mock。tests/unit/lib/redis/client.test.ts-37-39 (1)
37-39: 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win恢复原始环境变量值，而不是无条件删除。
如果测试进程启动时已配置
REDIS_COMMAND_TIMEOUT_MS，当前清理会永久移除它并影响后续测试。请在修改前保存原值，并在afterEach中按原状态恢复。🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/unit/lib/redis/client.test.ts` around lines 37 - 39, 在测试环境变量清理逻辑中保存 REDIS_COMMAND_TIMEOUT_MS 的初始值，并更新 afterEach，使其根据初始状态恢复原值；若初始未设置则删除该变量，避免影响后续测试。tests/unit/proxy/response-handler-lease-decrement.test.ts-584-593 (1)
584-593: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win不要将缺失的 5h 重置模式固化为正确结果。
该用例声称验证实体与窗口参数，但三个
"5h"值均断言为undefined。请为测试 provider、user、key 设置明确的"fixed"或"rolling"模式并验证传播，否则 5h 结算参数回归仍会通过。🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/unit/proxy/response-handler-lease-decrement.test.ts` around lines 584 - 593, Update the test around RateLimitService.settleLeaseBudgets to assign explicit “5h” reset modes for the provider, user, and key fixtures, using fixed or rolling values as appropriate, and assert those same modes are propagated in the settlement entities instead of undefined. Keep the existing daily mode and other settlement assertions unchanged.tests/unit/proxy/response-handler-lease-decrement.test.ts-19-40 (1)
19-40: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win统一三处
AsyncTaskManager.registermock 的预中止行为。生产实现不会为已中止的 controller 调用 factory；三处 mock 均会继续启动任务，可能掩盖取消后的副作用。
tests/unit/proxy/response-handler-lease-decrement.test.ts#L19-L40：在任务入队前检查controller.signal.aborted。tests/unit/proxy/response-handler-hedge-loser-priority.test.ts#L39-L50：在执行 factory 前直接返回已中止的 controller。tests/unit/proxy/response-handler-non200.test.ts#L22-L38：在创建 Promise 前补充相同检查。🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/unit/proxy/response-handler-lease-decrement.test.ts` around lines 19 - 40, Update all three AsyncTaskManager.register mocks to match production cancellation behavior: in tests/unit/proxy/response-handler-lease-decrement.test.ts (lines 19-40), check controller.signal.aborted before enqueueing the task and return without invoking factory; in tests/unit/proxy/response-handler-hedge-loser-priority.test.ts (lines 39-50), return the already-aborted controller before executing factory; and in tests/unit/proxy/response-handler-non200.test.ts (lines 22-38), add the same aborted-signal check before creating the promise.tests/unit/lib/rate-limit/lease-service.test.ts-1141-1153 (1)
1141-1153: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win请实际验证全部租约读取完成后才开始写入。
当前比较只证明
SETEX位于pendingWrites循环内；即使整个写入循环被移到租约验证循环之前，测试仍会通过。请同时断言读取循环、结果编码和写入循环的先后顺序。建议修改
+ const readLoop = script.indexOf("for keyIndex = 2, `#KEYS` do"); + const encodeResults = script.indexOf("local encodedOk, encoded = pcall"); + const writeLoop = script.indexOf("for writeIndex = 1, `#pendingWrites` do"); + + expect(readLoop).toBeLessThan(encodeResults); + expect(encodeResults).toBeLessThan(writeLoop); - expect(script.indexOf("for writeIndex = 1, `#pendingWrites` do")).toBeLessThan( + expect(writeLoop).toBeLessThan( script.indexOf('redis.call("SETEX", pendingWrite[1]') );🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/unit/lib/rate-limit/lease-service.test.ts` around lines 1141 - 1153, Update the test for LeaseService.settleLeaseBudgets to assert the Lua script ordering: the lease-reading/validation loop must complete before settlement result encoding, and encoding must complete before the pendingWrites write loop. Extend the existing script.indexOf checks around the visible pendingWrites and SETEX markers without changing the production implementation.src/repository/message.ts-494-496 (1)
494-496: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win移除代码注释中的 emoji。
建议修改
- model?: string; // ⭐ 新增：支持更新重定向后的模型名称 + model?: string; // 支持更新重定向后的模型名称 actualResponseModel?: string | null; // 上游响应实际返回的模型名(audit 用途，不影响计费) - providerId?: number; // ⭐ 新增：支持更新最终供应商ID（重试切换后） + providerId?: number; // 支持更新最终供应商 ID（重试切换后）As per coding guidelines, “Never use emoji characters in any code, comments, or string literals”.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/repository/message.ts` around lines 494 - 496, Remove the emoji characters from the comments on the model and providerId properties in the relevant message type, while preserving the existing comment text and declarations.Source: Coding guidelines
src/app/v1/_lib/proxy/node-stream-to-web.test.ts-133-157 (1)
133-157: 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win确保
uncaughtException监听器在失败路径也被移除。这些测试若在
removeListener()前失败，会遗留全局异常处理器，进而污染或掩盖后续测试。请将每段注册后的测试逻辑包在try/finally中，并在finally删除监听器。Also applies to: 300-333, 362-384
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/app/v1/_lib/proxy/node-stream-to-web.test.ts` around lines 133 - 157, Update the tests registering process uncaughtException listeners, including the cases around the delayed destroy, other failure-path test block, and the third referenced test block, to wrap their assertions and awaits in try/finally. Move each corresponding process.removeListener call into finally so listeners are always removed even when the test fails, while preserving the existing assertions.tests/unit/server-response-write-backpressure.test.ts-114-114 (1)
114-114: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win在
afterEach中恢复真实计时器。vi.restoreAllMocks()不会退出 fake-timer 模式；这个文件里有vi.useFakeTimers()，但没有对应的vi.useRealTimers()，后续用例可能继承假时钟。建议在清理里补上vi.useRealTimers()。🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/unit/server-response-write-backpressure.test.ts` at line 114, 在该测试文件的 afterEach 清理中补充 vi.useRealTimers()，与现有的 vi.restoreAllMocks() 一起执行，确保 vi.useFakeTimers() 创建的假计时器在每个用例后被恢复为真实计时器。
🧹 Nitpick comments (1)
src/app/v1/_lib/proxy/demand-driven-response-pump.test.ts (1)
2-5: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win改用
@/路径别名。建议修改
} from "./demand-driven-response-pump"; +} from "`@/app/v1/_lib/proxy/demand-driven-response-pump`";As per coding guidelines,
**/*.{ts,tsx,js,jsx}: Use path alias@/to map to ./src/ for imports.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/app/v1/_lib/proxy/demand-driven-response-pump.test.ts` around lines 2 - 5, Update the imports in the demand-driven response pump test to use the configured `@/` path alias mapped to src instead of the relative "./demand-driven-response-pump" path, while preserving the imported symbols.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@server.js`:
- Around line 414-417: 更新内部 HTTP response 回调，在写入 currentInternalReq 或
currentInternalRes 前检查 closed；客户端关闭后直接忽略已排队的迟到回调，避免重新注册响应并绕过
abortCurrentInternalReq 的清理流程。保留未关闭时现有的赋值行为。
In `@src/app/v1/_lib/proxy/response-handler.ts`:
- Around line 1693-1708: Update the non-stream terminal-details persistence flow
around messageContext to reuse persistNonStreamTerminalDetails() as the CAS
fallback when updateMessageRequestDetailsDurably() fails. Ensure
tracker.endRequest(messageContext.user.id, messageContext.id) runs from a
finally block so it executes on both successful and failed persistence paths.
- Around line 5246-5255: 修改 Error cause 处理逻辑，避免在 errorCause 中原样序列化任意 cause。围绕
ErrnoException.cause 仅提取允许的 name、code 及脱敏后的 message，过滤请求头、令牌、URL
和用户数据等敏感内容，再执行现有长度截断；保留不可安全提取时的安全降级行为。
- Around line 2444-2455: Remove the immediate
passthroughPump.cancelSource(reason) call from the bindClientAbortListener
callback after startDrain; client abort should only enter background draining.
Preserve the existing cancellation in onClientCancel if required, and let
bounded drain, idle, or response-timeout handling perform the final source
cancellation so terminal usage can be collected.
In `@src/drizzle/db.ts`:
- Around line 7-8: Update the imports in the db module to use the configured `@/`
path alias instead of relative "./" paths, targeting the existing
admitted-client and schema modules while preserving their imported symbols.
In `@src/lib/redis/lua-scripts.ts`:
- Around line 273-290: Require a non-empty, stable request ID for rolling cost
tracking and construct the Redis sorted-set member from that ID so concurrent
requests cannot overwrite one another. Update TRACK_COST_ROLLING_WINDOW in
src/lib/redis/lua-scripts.ts (lines 273-290) to reject empty request_id and use
it for an idempotent member; require and validate options.requestId in the
rolling paths at src/lib/rate-limit/service.ts (lines 977-995 and 1577-1593).
In `@src/repository/message.ts`:
- Around line 20-25: Replace the relative import in src/repository/message.ts
lines 20-25 with the `@/repository/message-write-buffer` alias, and replace the
relative import in src/lib/rate-limit/service.ts lines 94-99 with the
`@/lib/rate-limit/lease-service` alias. Preserve all imported symbols and
behavior.
- Around line 505-513: 更新 updateMessageRequestDetails 的普通 async 分支，移除
enqueueMessageRequestUpdate 后立即调用 publishCommittedMessageRequestDetails 的逻辑；仅在队列
flush 成功并确认 PostgreSQL 已提交后发布终态，或将携带终态字段的调用改为使用现有 durable API，确保未持久化的更新不会被
rollup 消费。
In `@tests/integration/db-pool-isolation-postgres.test.ts`:
- Around line 8-14: Ensure tests restore every modified process environment
variable: in tests/integration/db-pool-isolation-postgres.test.ts, move DSN and
DB_POOL_MAX setup into the non-skipped lifecycle, preserve their prior values,
and restore or delete them appropriately after the tests; in
tests/unit/lib/redis/client.test.ts, preserve REDIS_COMMAND_TIMEOUT_MS before
modification and restore it afterward, deleting it only when it was originally
unset.
In `@tests/unit/proxy/error-handler-client-message.test.ts`:
- Around line 101-119: 更新 error-handler.ts 中的错误解析流程，为
resolveFinalClientErrorMessage 传入 locale，并使用 next-intl
获取各状态码的通用上游错误翻译，避免写死简体中文。同步修改 error-handler-client-message.test.ts，将状态码断言按 5
个支持的语言参数化，并验证每种 locale 返回对应翻译。
In `@tests/unit/proxy/response-handler-client-abort-drain.test.ts`:
- Around line 23-24: Remove the duplicated inserted lines in
response-handler-client-abort-drain.test.ts, including the repeated
asyncTasks/registeredTasks declarations and the duplicate content at the
referenced later locations. Preserve the original declarations, parameters, and
valid expressions so the test compiles without duplicate declarations or invalid
syntax.
In `@tests/unit/repository/message-public-status-rollup.test.ts`:
- Around line 369-390: Update the test around updateMessageRequestDetailsDurably
to defer mockDbUpdateWhere with a pending Promise, then assert
mockQueuePublicStatusRollupWrite has not been called before resolving that
Promise. Release the deferred database write, flush microtasks, and assert the
rollup is queued afterward while preserving the existing sync-mode assertions.
- Around line 180-196: Strengthen the CAS WHERE-condition assertions in
updateMessageRequestDetailsIfUnfinalized tests: at
tests/unit/repository/message-public-status-rollup.test.ts:180-196, render the
SQL and parameters and assert the condition includes target ID 606; at :198-217,
assert ID 608 and the terminal-null condition on the async path; at :219-232,
assert ID 607 and status_code IS NULL before simulating CAS failure.
---
Minor comments:
In `@src/app/v1/_lib/proxy/node-stream-to-web.test.ts`:
- Around line 133-157: Update the tests registering process uncaughtException
listeners, including the cases around the delayed destroy, other failure-path
test block, and the third referenced test block, to wrap their assertions and
awaits in try/finally. Move each corresponding process.removeListener call into
finally so listeners are always removed even when the test fails, while
preserving the existing assertions.
In `@src/repository/message.ts`:
- Around line 494-496: Remove the emoji characters from the comments on the
model and providerId properties in the relevant message type, while preserving
the existing comment text and declarations.
In `@tests/integration/billing-model-source.test.ts`:
- Around line 26-33: 在异步任务注册逻辑中为每个 promise 附加拒绝观察，避免任务后续拒绝形成未处理拒绝；更新 beforeEach
的清理流程，在清空 asyncTasks 和 asyncTaskControllers 前等待所有已中止任务
settled，确保遗留任务不会跨用例继续运行或修改 mock。
In `@tests/unit/lib/rate-limit/lease-service.test.ts`:
- Around line 1141-1153: Update the test for LeaseService.settleLeaseBudgets to
assert the Lua script ordering: the lease-reading/validation loop must complete
before settlement result encoding, and encoding must complete before the
pendingWrites write loop. Extend the existing script.indexOf checks around the
visible pendingWrites and SETEX markers without changing the production
implementation.
In `@tests/unit/lib/redis/client.test.ts`:
- Around line 37-39: 在测试环境变量清理逻辑中保存 REDIS_COMMAND_TIMEOUT_MS 的初始值，并更新
afterEach，使其根据初始状态恢复原值；若初始未设置则删除该变量，避免影响后续测试。
In `@tests/unit/proxy/proxy-forwarder-retry-limit.test.ts`:
- Around line 328-340: Update the test around categorizeErrorAsync to use the
real error classifier instead of mocking it to numeric value 5, so the
DbPoolAdmissionError cause chain is actually exercised. Assert or configure the
expected classification with the named ErrorCategory enum member rather than a
positional number, while preserving the existing wrappedError and doForward
rejection setup.
In `@tests/unit/proxy/response-handler-gemini-terminal.test.ts`:
- Around line 119-123: 更新 settleTasks，使每批 mocks.tasks 的托管任务拒绝不会被忽略：汇总
Promise.allSettled 的 rejected 结果，并在存在拒绝时抛出其原因；继续等待后续批次任务，保留当前清空并重复处理动态新增任务的行为。
In `@tests/unit/proxy/response-handler-lease-decrement.test.ts`:
- Around line 584-593: Update the test around
RateLimitService.settleLeaseBudgets to assign explicit “5h” reset modes for the
provider, user, and key fixtures, using fixed or rolling values as appropriate,
and assert those same modes are propagated in the settlement entities instead of
undefined. Keep the existing daily mode and other settlement assertions
unchanged.
- Around line 19-40: Update all three AsyncTaskManager.register mocks to match
production cancellation behavior: in
tests/unit/proxy/response-handler-lease-decrement.test.ts (lines 19-40), check
controller.signal.aborted before enqueueing the task and return without invoking
factory; in tests/unit/proxy/response-handler-hedge-loser-priority.test.ts
(lines 39-50), return the already-aborted controller before executing factory;
and in tests/unit/proxy/response-handler-non200.test.ts (lines 22-38), add the
same aborted-signal check before creating the promise.
In `@tests/unit/proxy/terminal-outcome-contract.test.ts`:
- Around line 225-249: 更新该测试的清理流程，将 releaseCommit.resolve()、等待 flushPromise 以及
stopMessageRequestWriteBuffer() 放入 finally 中，确保前置断言失败时也会释放提交并停止全局 message
writer；保留现有成功路径断言和 handlePromise 响应验证。
In `@tests/unit/server-response-write-backpressure.test.ts`:
- Line 114: 在该测试文件的 afterEach 清理中补充 vi.useRealTimers()，与现有的 vi.restoreAllMocks()
一起执行，确保 vi.useFakeTimers() 创建的假计时器在每个用例后被恢复为真实计时器。
---
Nitpick comments:
In `@src/app/v1/_lib/proxy/demand-driven-response-pump.test.ts`:
- Around line 2-5: Update the imports in the demand-driven response pump test to
use the configured `@/` path alias mapped to src instead of the relative
"./demand-driven-response-pump" path, while preserving the imported symbols.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: ef7158f8-7620-4220-ae59-aefafb11d89d
📒 Files selected for processing (112)
.env.exampledeploy/k8s/app/deployment.yamlserver.jssrc/app/v1/[...route]/route.tssrc/app/v1/_lib/proxy-handler.tssrc/app/v1/_lib/proxy/demand-driven-response-pump.test.tssrc/app/v1/_lib/proxy/demand-driven-response-pump.tssrc/app/v1/_lib/proxy/error-handler.tssrc/app/v1/_lib/proxy/errors.tssrc/app/v1/_lib/proxy/forwarder.tssrc/app/v1/_lib/proxy/node-stream-to-web.test.tssrc/app/v1/_lib/proxy/node-stream-to-web.tssrc/app/v1/_lib/proxy/response-handler.tssrc/app/v1beta/[...route]/route.tssrc/drizzle/admitted-client.tssrc/drizzle/db.tssrc/lib/async-task-manager.tssrc/lib/config/env.schema.tssrc/lib/langfuse/trace-proxy-request.tssrc/lib/lifecycle/shutdown.tssrc/lib/price-sync/cloud-price-updater.tssrc/lib/provider-testing/test-service.test.tssrc/lib/rate-limit/lease-service.tssrc/lib/rate-limit/service.tssrc/lib/redis/client.tssrc/lib/redis/lua-scripts.tssrc/lib/utils/upstream-error-detection.test.tssrc/repository/message-write-buffer.tssrc/repository/message.tstests/configs/integration.config.tstests/integration/billing-model-source.test.tstests/integration/db-pool-isolation-postgres.test.tstests/integration/db-pool-slow-close-postgres.test.tstests/integration/lease-settlement-redis.test.tstests/integration/message-write-buffer-recovery-postgres.test.tstests/integration/proxy-hedge-lifecycle.test.tstests/integration/rolling-cost-redis.test.tstests/unit/actions/providers-patch-contract.test.tstests/unit/api/actions/legacy-deprecation.test.tstests/unit/api/v1/status-code-map.test.tstests/unit/dashboard/user-insights-page.test.tsxtests/unit/drizzle/db-admission.test.tstests/unit/drizzle/db-pool-config.test.tstests/unit/drizzle/db-scope.test.tstests/unit/drizzle/db-shutdown.test.tstests/unit/i18n/key-created-copy.test.tstests/unit/instrumentation-crash-handler.test.tstests/unit/langfuse/langfuse-trace.test.tstests/unit/lib/async-task-manager-edge-runtime.test.tstests/unit/lib/provider-allowed-model-schema.test.tstests/unit/lib/provider-model-redirect-schema.test.tstests/unit/lib/rate-limit/lease-service.test.tstests/unit/lib/rate-limit/rolling-window-5h.test.tstests/unit/lib/rate-limit/rolling-window-cache-warm.test.tstests/unit/lib/rate-limit/service-extra.test.tstests/unit/lib/redis/client.test.tstests/unit/lib/shutdown.test.tstests/unit/lib/upstream-error-detection-status.test.tstests/unit/price-sync/cloud-price-updater.test.tstests/unit/proxy/client-abort-vs-upstream-499.test.tstests/unit/proxy/client-detector.test.tstests/unit/proxy/codex-provider-overrides.test.tstests/unit/proxy/connected-non-reader-lifetime.test.tstests/unit/proxy/endpoint-family-catalog.test.tstests/unit/proxy/endpoint-family-provider-routing.test.tstests/unit/proxy/endpoint-path-normalization.test.tstests/unit/proxy/error-handler-client-message.test.tstests/unit/proxy/error-handler-durable-persistence.test.tstests/unit/proxy/error-handler-langfuse-trace.test.tstests/unit/proxy/error-handler-overrides.test.tstests/unit/proxy/error-handler-terminal-status.test.tstests/unit/proxy/fake-streaming-response-validator.test.tstests/unit/proxy/fake-streaming-response.test.tstests/unit/proxy/fake-streaming-stream-intent.test.tstests/unit/proxy/pricing-no-price.test.tstests/unit/proxy/provider-selector-cross-type-model.test.tstests/unit/proxy/proxy-forwarder-endpoint-audit.test.tstests/unit/proxy/proxy-forwarder-hedge-first-byte.test.tstests/unit/proxy/proxy-forwarder-retry-limit.test.tstests/unit/proxy/proxy-handler-concurrency-ownership.test.tstests/unit/proxy/proxy-handler-public-errors.test.tstests/unit/proxy/proxy-handler-public-success.test.tstests/unit/proxy/response-handler-abort-listener-cleanup.test.tstests/unit/proxy/response-handler-bill-non-success.test.tstests/unit/proxy/response-handler-client-abort-drain.test.tstests/unit/proxy/response-handler-endpoint-circuit-isolation.test.tstests/unit/proxy/response-handler-exported-finalizers.test.tstests/unit/proxy/response-handler-gemini-stream-passthrough-timeouts.test.tstests/unit/proxy/response-handler-gemini-terminal.test.tstests/unit/proxy/response-handler-hedge-loser-priority.test.tstests/unit/proxy/response-handler-lease-decrement.test.tstests/unit/proxy/response-handler-non200.test.tstests/unit/proxy/response-handler-nonstream-terminal.test.tstests/unit/proxy/response-handler-public-dispatch.test.tstests/unit/proxy/response-handler-stream-terminal.test.tstests/unit/proxy/session.test.tstests/unit/proxy/terminal-outcome-contract.test.tstests/unit/repository/message-aggregate-multiple-session-stats.test.tstests/unit/repository/message-aggregate-session-stats.test.tstests/unit/repository/message-public-readback.test.tstests/unit/repository/message-public-status-rollup.test.tstests/unit/repository/message-query-test-support.tstests/unit/repository/message-session-readback.test.tstests/unit/repository/message-session-request-query.test.tstests/unit/repository/message-terminal-cas-durable.test.tstests/unit/repository/message-terminal-cost-accounting.test.tstests/unit/repository/message-terminal-public-status-seam.test.tstests/unit/repository/message-terminal-write-apis.test.tstests/unit/repository/message-usage-logs-query.test.tstests/unit/repository/message-write-buffer.test.tstests/unit/server-response-write-backpressure.test.tstests/unit/server-shutdown.test.ts
Comment on lines
+5246
to
+5255
| let errorCause: string | undefined; | ||
| if (error instanceof Error && (error as NodeJS.ErrnoException).cause) { | ||
| try { | ||
| const cause = (error as NodeJS.ErrnoException).cause; | ||
| errorCause = JSON.stringify(cause, Object.getOwnPropertyNames(cause as object)); | ||
| } catch { | ||
| errorCause = String((error as NodeJS.ErrnoException).cause); | ||
| } | ||
| if (errorCause && errorCause.length > maxErrorCauseLength) { | ||
| errorCause = `${errorCause.substring(0, maxErrorCauseLength)}...[truncated]`; |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
不要原样持久化任意 Error.cause。
cause 可能包含请求头、认证令牌、完整 URL 或用户数据；长度截断并不能脱敏。请仅保留允许字段（如 name、code 和脱敏后的 message），避免敏感信息进入数据库。
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/app/v1/_lib/proxy/response-handler.ts` around lines 5246 - 5255, 修改 Error
cause 处理逻辑，避免在 errorCause 中原样序列化任意 cause。围绕 ErrnoException.cause 仅提取允许的
name、code 及脱敏后的 message，过滤请求头、令牌、URL 和用户数据等敏感内容，再执行现有长度截断；保留不可安全提取时的安全降级行为。
Comment on lines
+7
to
+8
| import { createAdmittedSqlClient } from "./admitted-client"; | ||
| import * as schema from "./schema"; |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick win
请对 src 内部模块使用 @/ 路径别名。
这两个相对导入不符合仓库的 TypeScript 导入约定。
建议修改
-import { createAdmittedSqlClient } from "./admitted-client";
-import * as schema from "./schema";
+import { createAdmittedSqlClient } from "`@/drizzle/admitted-client`";
+import * as schema from "`@/drizzle/schema`";As per coding guidelines，TypeScript 导入必须使用 @/ 路径别名映射到 ./src/。
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| import { createAdmittedSqlClient } from "./admitted-client"; | |
| import * as schema from "./schema"; | |
| import { createAdmittedSqlClient } from "`@/drizzle/admitted-client`"; | |
| import * as schema from "`@/drizzle/schema`"; |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/drizzle/db.ts` around lines 7 - 8, Update the imports in the db module to
use the configured `@/` path alias instead of relative "./" paths, targeting the
existing admitted-client and schema modules while preserving their imported
symbols.
Source: Coding guidelines
Comment on lines
+8
to
+14
| if (!process.env.DSN && process.env.DATABASE_URL) { | ||
| process.env.DSN = process.env.DATABASE_URL; | ||
| } | ||
|
|
||
| const previousPoolMax = process.env.DB_POOL_MAX; | ||
| process.env.DB_POOL_MAX = "6"; | ||
| vi.resetModules(); |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
请保证测试完整恢复其修改的进程环境。
两个测试都可能覆盖调用方预先配置的环境变量，导致同一 Vitest worker 中的后续测试使用错误配置。
tests/integration/db-pool-isolation-postgres.test.ts#L8-L14：仅在非跳过的生命周期内设置DSN和DB_POOL_MAX，并在结束时恢复两者。tests/unit/lib/redis/client.test.ts#L37-L39：保存测试前的REDIS_COMMAND_TIMEOUT_MS，结束后恢复原值；仅当原值不存在时才删除。
📍 Affects 2 files
tests/integration/db-pool-isolation-postgres.test.ts#L8-L14(this comment)tests/unit/lib/redis/client.test.ts#L37-L39
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/integration/db-pool-isolation-postgres.test.ts` around lines 8 - 14,
Ensure tests restore every modified process environment variable: in
tests/integration/db-pool-isolation-postgres.test.ts, move DSN and DB_POOL_MAX
setup into the non-skipped lifecycle, preserve their prior values, and restore
or delete them appropriately after the tests; in
tests/unit/lib/redis/client.test.ts, preserve REDIS_COMMAND_TIMEOUT_MS before
modification and restore it afterward, deleting it only when it was originally
unset.
Comment on lines
23
to
+24
| const asyncTasks: Promise<void>[] = []; | ||
| const registeredTasks: Array<{ taskType: string; promise: Promise<void> }> = []; |
There was a problem hiding this comment.
🎯 Functional Correctness | 🔴 Critical | ⚡ Quick win
删除重复插入的代码行。
Line 24、Line 39、Line 1382 和 Line 2133 的重复内容会分别造成重复声明、重复参数或无效表达式，导致 TypeScript 编译失败。
建议修复
const registeredTasks: Array<{ taskType: string; promise: Promise<void> }> = [];
-const registeredTasks: Array<{ taskType: string; promise: Promise<void> }> = [];
options?: string | { abortController?: AbortController; taskType?: string }
-options?: string | { abortController?: AbortController; taskType?: string }
let resolveBinding!: (result: { updated: boolean; reason: string }) => void;
-let resolveBinding!: (result: { updated: boolean; reason: string }) => void;
const calls = (
updateMessageRequestDetailsDurably as unknown as { mock: { calls: unknown[][] } }
- updateMessageRequestDetailsDurably as unknown as { mock: { calls: unknown[][] } }
).mock.calls;Also applies to: 35-39, 1378-1383, 2131-2134
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/unit/proxy/response-handler-client-abort-drain.test.ts` around lines 23
- 24, Remove the duplicated inserted lines in
response-handler-client-abort-drain.test.ts, including the repeated
asyncTasks/registeredTasks declarations and the duplicate content at the
referenced later locations. Preserve the original declarations, parameters, and
valid expressions so the test compiles without duplicate declarations or invalid
syntax.
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 2e2b059bf4
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| onChunk: (value) => observePassthroughChunk(value), | ||
| onClientCancel: (reason) => { | ||
| passthroughPump.startDrain(reason); | ||
| passthroughPump.cancelSource(reason); |
There was a problem hiding this comment.
When a Gemini passthrough client stops reading the returned stream, this cancelSource() call immediately cancels the only upstream reader right after startDrain(). That defeats the new bounded drain behavior used by the normal streaming path, so a stream that would have produced terminal usage/status within the drain window is finalized from partial content as a client abort instead of being billed and recorded accurately. Let the pump drain on downstream cancel and only cancel on the explicit drain timeout/idle timeout paths.
Useful? React with 👍 / 👎.
| ? parsed.error | ||
| : { code: `http_${res.statusCode}`, message: text.slice(0, 512) }, | ||
| }, | ||
| { response: res, onSuccess: settleResponse, onFailure: settleAndClose } |
There was a problem hiding this comment.
For WebSocket /v1/responses turns that get a non-SSE JSON response, this now waits for the WebSocket send callback before marking the internal response settled. http.IncomingMessage emits close after a normal end, so if the WS callback is delayed by the new outbound queue, the close handler below still sees responseSettled === false and sends internal_response_closed/closes the socket even though the JSON body was complete. Mark the HTTP response complete after parsing end (or ignore close after end) while keeping the WS send queued.
Useful? React with 👍 / 👎.
Comment on lines
+606
to
+610
| const rawResult = (await redis.eval( | ||
| LeaseService.SETTLE_LEASE_BUDGETS_LUA_SCRIPT, | ||
| 1 + leaseKeys.length, | ||
| markerKey, | ||
| ...leaseKeys, |
There was a problem hiding this comment.
In Redis Cluster deployments, this EVAL now passes the marker plus twelve lease keys such as lease:key:..., lease:user:..., and lease:provider:... without a shared hash tag, so the script is rejected as a cross-slot multi-key command. The catch path turns that into fail_open, which means successful requests stop decrementing cached lease budgets until the next DB refresh and can exceed quota slices; put all settlement keys in the same hash slot or avoid a multi-key script for cluster setups.
Useful? React with 👍 / 👎.
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
Comment on lines
+144
to
+178
| logger.error("ResponseHandler: Durable non-stream terminal persistence failed", { | ||
| taskId: options.taskId, | ||
| messageId: options.messageRequestId, | ||
| statusCode: options.details.statusCode, | ||
| error: primaryError, | ||
| }); | ||
| } | ||
|
|
||
| if (provider.firstByteTimeoutStreamingMs > 0) { | ||
| return Math.max(provider.firstByteTimeoutStreamingMs, provider.streamingIdleTimeoutMs); | ||
| try { | ||
| await updateMessageRequestDetailsIfUnfinalized( | ||
| options.messageRequestId, | ||
| completeTerminalDetails | ||
| ); | ||
| } catch (fallbackError) { | ||
| logger.error("ResponseHandler: Conditional non-stream terminal fallback failed", { | ||
| taskId: options.taskId, | ||
| messageId: options.messageRequestId, | ||
| statusCode: options.details.statusCode, | ||
| fallbackError, | ||
| }); | ||
| throw markNonStreamTerminalPersistenceError(fallbackError); | ||
| } | ||
| } | ||
|
|
||
| return Number.POSITIVE_INFINITY; | ||
| function raceWithTimeout<T>(promise: Promise<T>, timeoutMs: number, message: string): Promise<T> { | ||
| let timeoutId: NodeJS.Timeout | null = null; | ||
| const timeoutPromise = new Promise<never>((_resolve, reject) => { | ||
| timeoutId = setTimeout(() => reject(new Error(message)), timeoutMs); | ||
| timeoutId.unref?.(); | ||
| }); | ||
|
|
||
| return Promise.race([promise, timeoutPromise]).finally(() => { | ||
| if (timeoutId) { | ||
| clearTimeout(timeoutId); | ||
| } |
There was a problem hiding this comment.
The factory passed to AsyncTaskManager.register accepts an AbortSignal parameter but the implementation never threads it into the work it schedules. When the pod receives SIGTERM, shutdownAllAsyncTasks() fires the signal but the in-flight finalization work continues uninterrupted — delaying pod shutdown by up to STREAM_FINALIZATION_MAX_MS = 120_000 ms regardless of the shutdown deadline.
Prompt To Fix With AI
This is a comment left during a code review.
Path: src/app/v1/_lib/proxy/response-handler.ts
Line: 144-178
Comment:
**AbortSignal ignored in post-terminal task factory**
The factory passed to `AsyncTaskManager.register` accepts an `AbortSignal` parameter but the implementation never threads it into the work it schedules. When the pod receives SIGTERM, `shutdownAllAsyncTasks()` fires the signal but the in-flight finalization work continues uninterrupted — delaying pod shutdown by up to `STREAM_FINALIZATION_MAX_MS = 120_000` ms regardless of the shutdown deadline.
How can I resolve this? If you propose a fix, please make it concise.
Comment on lines
+195
to
+260
| export function createAdmittedSqlClient<TClient extends object>( | ||
| client: TClient, | ||
| options: AdmittedClientOptions | ||
| ): TClient { | ||
| const rawClient = client as unknown as UnsafeAndBeginClient; | ||
| const originalUnsafe = rawClient.unsafe.bind(client); | ||
| const originalBegin = rawClient.begin.bind(client); | ||
| let outstanding = 0; | ||
|
|
||
| const acquire = () => { | ||
| if (outstanding >= options.maxOutstanding) { | ||
| throw new DbPoolAdmissionError(options.pool, options.maxOutstanding); | ||
| } | ||
| outstanding += 1; | ||
| let released = false; | ||
| return () => { | ||
| if (released) return; | ||
| released = true; | ||
| outstanding -= 1; | ||
| }; | ||
| }; | ||
|
|
||
| const admittedUnsafe = (...args: unknown[]) => { | ||
| const release = acquire(); | ||
| try { | ||
| return wrapPendingQuery(originalUnsafe(...args), release); | ||
| } catch (error) { | ||
| release(); | ||
| throw error; | ||
| } | ||
| }; | ||
|
|
||
| const admittedBegin = (...args: unknown[]) => { | ||
| const release = acquire(); | ||
| let result: unknown; | ||
| try { | ||
| result = originalBegin(...args); | ||
| } catch (error) { | ||
| release(); | ||
| throw error; | ||
| } | ||
|
|
||
| if (!isPromiseLike(result)) { | ||
| release(); | ||
| return result; | ||
| } | ||
| return Promise.resolve(result).then( | ||
| (value) => { | ||
| release(); | ||
| return value; | ||
| }, | ||
| (error) => { | ||
| release(); | ||
| throw error; | ||
| } | ||
| ); | ||
| }; | ||
|
|
||
| return new Proxy(client, { | ||
| get(target, property, receiver) { | ||
| if (property === "unsafe") return admittedUnsafe; | ||
| if (property === "begin") return admittedBegin; | ||
| return Reflect.get(target, property, receiver); | ||
| }, | ||
| }); | ||
| } |
There was a problem hiding this comment.
The createAdmittedSqlClient Proxy intercepts named properties (unsafe, begin) but has no apply trap. When Drizzle calls the client as a tag template literal (e.g. client`SELECT 1`), the call goes directly to the underlying postgres.js client, completely bypassing the outstanding-query counter. Under pool pressure this silently removes the back-pressure guarantee for the most common query path.
Prompt To Fix With AI
This is a comment left during a code review.
Path: src/drizzle/admitted-client.ts
Line: 195-260
Comment:
**Tag template literal calls bypass admission control**
The `createAdmittedSqlClient` Proxy intercepts named properties (`unsafe`, `begin`) but has no `apply` trap. When Drizzle calls the client as a tag template literal (e.g. `` client`SELECT 1` ``), the call goes directly to the underlying `postgres.js` client, completely bypassing the outstanding-query counter. Under pool pressure this silently removes the back-pressure guarantee for the most common query path.
How can I resolve this? If you propose a fix, please make it concise.
Comment on lines
+124
to
+158
| const startDrain = (_reason?: unknown) => { | ||
| if (settled || state === "finalizing" || state === "closed") return; | ||
| if (state === "draining") { | ||
| scheduleDrain(); | ||
| return; | ||
| } | ||
| clientAborted = true; | ||
| state = "draining"; | ||
| try { | ||
| clientController?.error( | ||
| _reason == null ? new Error("Client disconnected") : toError(_reason) | ||
| ); | ||
| } catch (controllerError) { | ||
| if (!(controllerError instanceof TypeError)) throw controllerError; | ||
| // The ReadableStream cancel algorithm may have already detached the controller. | ||
| } | ||
| scheduleDrain(); | ||
| }; | ||
|
|
||
| const cancelSource = (reason?: unknown) => { | ||
| if (settled) return; | ||
| const normalized = reason == null ? new Error("Source cancelled") : toError(reason); | ||
| settle(false, normalized, normalized); | ||
| }; | ||
|
|
||
| const armPendingChunkDeadline = () => { | ||
| clearPendingChunkDeadline(); | ||
| pendingChunkDeadlineId = setTimeout(() => { | ||
| const error = new DOMException( | ||
| `Client response body was not consumed within ${PENDING_CHUNK_DEADLINE_MS}ms`, | ||
| "AbortError" | ||
| ); | ||
| startDrain(error); | ||
| cancelSource(error); | ||
| }, PENDING_CHUNK_DEADLINE_MS); |
There was a problem hiding this comment.
clientAborted conflates client cancel and server-side drain
startDrain sets clientAborted = true unconditionally. It is called both when the client cancels and when armPendingChunkDeadline fires after 60s of unconsumed chunk. In the deadline case the provider delivered data that the client simply has not consumed yet — setting clientAborted = true here causes wasClientAborted() to report a client fault, which may suppress an accurate circuit-breaker update blaming the provider.
Prompt To Fix With AI
This is a comment left during a code review.
Path: src/app/v1/_lib/proxy/demand-driven-response-pump.ts
Line: 124-158
Comment:
**`clientAborted` conflates client cancel and server-side drain**
`startDrain` sets `clientAborted = true` unconditionally. It is called both when the client cancels and when `armPendingChunkDeadline` fires after 60s of unconsumed chunk. In the deadline case the provider delivered data that the client simply has not consumed yet — setting `clientAborted = true` here causes `wasClientAborted()` to report a client fault, which may suppress an accurate circuit-breaker update blaming the provider.
How can I resolve this? If you propose a fix, please make it concise.
Comment on lines
+220
to
+250
| TRACK_COST_ROLLING_WINDOW, | ||
| 1, | ||
| key, | ||
| cost.toString(), | ||
| now.toString(), | ||
| windowMs.toString(), | ||
| requestId, | ||
| ttlSeconds.toString() | ||
| ); | ||
| } | ||
|
|
||
| private static logCostPipelineErrors( | ||
| results: Array<[Error | null, unknown]> | null, | ||
| operation: "trackCost" | "trackUserDailyCost" | ||
| ): void { | ||
| if (!results) { | ||
| logger.error("[RateLimit] Cost pipeline returned null", { operation }); | ||
| return; | ||
| } | ||
|
|
||
| for (let commandIndex = 0; commandIndex < results.length; commandIndex += 1) { | ||
| const error = results[commandIndex]?.[0]; | ||
| if (!error) continue; | ||
|
|
||
| logger.error("[RateLimit] Cost pipeline command failed", { | ||
| operation, | ||
| commandIndex, | ||
| error: error.message, | ||
| }); | ||
| } | ||
| } |
There was a problem hiding this comment.
logCostPipelineErrors logs individual command failures at error level but returns void. The trackCost caller discards the result, so any Redis pipeline failure causes silent cost-tracking drift — counters are neither decremented nor retried, leading to incorrect billing or rate-limit state with no observable error in the call stack.
Prompt To Fix With AI
This is a comment left during a code review.
Path: src/lib/rate-limit/service.ts
Line: 220-250
Comment:
**Pipeline errors are logged but not propagated**
`logCostPipelineErrors` logs individual command failures at `error` level but returns `void`. The `trackCost` caller discards the result, so any Redis pipeline failure causes silent cost-tracking drift — counters are neither decremented nor retried, leading to incorrect billing or rate-limit state with no observable error in the call stack.
How can I resolve this? If you propose a fix, please make it concise.
Comment on lines
321
to
390
| } | ||
| } | ||
|
|
||
| /** | ||
| * 取消并等待 shutdown 时仍在飞的全部任务 settled。 | ||
| * | ||
| * task 的 finally 可能在等待期间注册尾部任务，因此循环到 pending 集合为空；并发 shutdown | ||
| * 调用共享同一个 Promise，避免重复取消或提前返回。 | ||
| */ | ||
| shutdownAll(): Promise<void> { | ||
| if (this.shutdownPromise) { | ||
| return this.shutdownPromise; | ||
| } | ||
|
|
||
| let resolveShutdown!: () => void; | ||
| let rejectShutdown!: (reason?: unknown) => void; | ||
| const shutdownPromise = new Promise<void>((resolve, reject) => { | ||
| resolveShutdown = resolve; | ||
| rejectShutdown = reject; | ||
| }); | ||
| this.shutdownPromise = shutdownPromise; | ||
| this.lifecycleState = "draining"; | ||
|
|
||
| // 先发布共享 Promise，再同步开始 abort；这样既保留既有同步取消语义， | ||
| // 同步 abort listener 重入时也会复用同一次 shutdown。 | ||
| void (async () => { | ||
| if (this.cleanupInterval) { | ||
| clearInterval(this.cleanupInterval); | ||
| this.cleanupInterval = null; | ||
| } | ||
|
|
||
| while (true) { | ||
| if (this.pendingTasks.size === 0) { | ||
| this.lifecycleState = "closed"; | ||
| return; | ||
| } | ||
|
|
||
| const activeTasks = Array.from(this.pendingTasks); | ||
| logger.info("[AsyncTaskManager] Cancelling and joining active tasks", { | ||
| count: activeTasks.length, | ||
| }); | ||
|
|
||
| for (const taskInfo of activeTasks) { | ||
| if (!taskInfo.abortController.signal.aborted) { | ||
| taskInfo.abortController.abort(); | ||
| } | ||
| } | ||
|
|
||
| await Promise.allSettled(activeTasks.map((taskInfo) => taskInfo.promise)); | ||
|
|
||
| for (const taskInfo of activeTasks) { | ||
| this.cleanup(taskInfo.taskId, taskInfo); | ||
| } | ||
| } | ||
| })().then(resolveShutdown, (error) => { | ||
| this.lifecycleState = "closed"; | ||
| rejectShutdown(error); | ||
| }); | ||
|
|
||
| return shutdownPromise; | ||
| } | ||
|
|
||
| /** | ||
| * 获取当前活跃任务数 | ||
| */ | ||
| getActiveTaskCount(): number { | ||
| return this.tasks.size; | ||
| return this.pendingTasks.size; | ||
| } | ||
|
|
||
| /** |
There was a problem hiding this comment.
allSettled cleanup calls in shutdownAll loop are always no-ops
After Promise.allSettled, the loop iterates pendingTasks and calls pendingTasks.delete(task) for each settled task. But tasks self-delete from pendingTasks inside their own .finally() chain before the allSettled resolves, so by the time the loop runs, pendingTasks is already empty. The explicit cleanup is dead code and may create confusion about the actual cleanup path.
Prompt To Fix With AI
This is a comment left during a code review.
Path: src/lib/async-task-manager.ts
Line: 321-390
Comment:
**Post-`allSettled` cleanup calls in `shutdownAll` loop are always no-ops**
After `Promise.allSettled`, the loop iterates `pendingTasks` and calls `pendingTasks.delete(task)` for each settled task. But tasks self-delete from `pendingTasks` inside their own `.finally()` chain before the `allSettled` resolves, so by the time the loop runs, `pendingTasks` is already empty. The explicit cleanup is dead code and may create confusion about the actual cleanup path.
How can I resolve this? If you propose a fix, please make it concise.Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
| options.response.pause(); | ||
| state.pressuredResponses.add(options.response); | ||
| } | ||
| if (!state.active || state.pendingBytes + bytes > MAX_PENDING_OUTBOUND_BYTES) { |
Contributor
There was a problem hiding this comment.
[HIGH] [LOGIC-BUG] safeSend() turns the 1 MiB backlog guard into a hard per-frame size limit
Why this is a problem: state.pendingBytes is 0 on the first send, so this branch rejects any single payload larger than MAX_PENDING_OUTBOUND_BYTES before ws.send() is even attempted. That means a legitimate non-stream response.completed frame bigger than 1 MiB now always closes the socket as outbound_backpressure, even when there is no queued backlog yet. I reproduced this by sending a >1 MiB JSON response through forwardToInternalHttp(): it immediately logged ws_send_failed and closed with 1011/outbound_backpressure.
Suggested fix:
const wouldOverflowQueuedBacklog =
(state.inFlight || state.pending.length > 0) &&
state.pendingBytes + bytes > MAX_PENDING_OUTBOUND_BYTES;
if (!state.active || wouldOverflowQueuedBacklog) {
failOutboundSends(ws, state, {
reason: "outbound_backpressure",
onFailure: options.onFailure,
});
return false;
}If you still want to cap single-frame payloads, that needs a separate explicit frame-size check so large terminal JSON responses do not get rejected by the backlog guard.
| commitPromise = Promise.reject(error); | ||
| } | ||
|
|
||
| await raceWithTimeout( |
Contributor
There was a problem hiding this comment.
[HIGH] [LOGIC-BUG] The timeout wrapper drops post-terminal-side-effects from shutdown tracking while the real side effect is still running
Why this is a problem: raceWithTimeout(...).catch(...) only waits for the 120s race, not for options.commit() itself. Once the timeout fires, this registered async task resolves and shutdownAllAsyncTasks() will stop waiting for it, but the original commitPromise keeps running in the background. These commits mutate session bindings and circuit state, so shutdown can now proceed to stopMessageRequestWriteBuffer() / closeDbPools() / closeRedis() while the side effect is still doing I/O against those resources.
Suggested fix:
const warningTimer = setTimeout(() => {
logger.warn("[ResponseHandler] Post-terminal side effects did not complete", {
taskId: options.taskId,
providerId: options.providerId,
sessionId: options.sessionId,
});
}, STREAM_FINALIZATION_MAX_MS);
warningTimer.unref?.();
try {
await commitPromise;
} finally {
clearTimeout(warningTimer);
}If these side effects really need a hard deadline, thread an AbortSignal into options.commit() and only resolve the registered task after that cancellation path has actually settled.
Contributor
There was a problem hiding this comment.
Code Review Summary
This XL PR rewires several critical paths at once: WebSocket transport ownership, proxy finalization, shutdown orchestration, DB pool partitioning, and rate-limit settlement. I found two high-priority regressions in the new transport/finalization logic that should be addressed before merge.
PR Size: XL
- Lines changed: 21441
- Files changed: 112
- Split suggestion: Separate the WebSocket transport / response-finalization changes (
server.js,src/app/v1/_lib/proxy/*, related shutdown tests) from the persistence / infrastructure changes (src/drizzle/*,src/repository/message*,src/lib/rate-limit/*,src/lib/redis/*). - Split suggestion: Land the DB pool + durable writer work independently from the lease-settlement + Redis timeout work so failures can be isolated more easily.
Issues Found
| Category | Critical | High | Medium | Low |
|---|---|---|---|---|
| Logic/Bugs | 0 | 1 | 0 | 0 |
| Security | 0 | 0 | 0 | 0 |
| Error Handling | 0 | 1 | 0 | 0 |
| Types | 0 | 0 | 0 | 0 |
| Comments/Docs | 0 | 0 | 0 | 0 |
| Tests | 0 | 0 | 0 | 0 |
| Simplification | 0 | 0 | 0 | 0 |
Critical Issues (Must Fix)
- None.
High Priority Issues (Should Fix)
server.js:208— the new backlog guard now rejects any single outbound WebSocket payload over 1 MiB before the firstws.send()attempt, which breaks large non-streamresponse.completedframes by closing the socket asoutbound_backpressure.src/app/v1/_lib/proxy/response-handler.ts:216— the timeout wrapper resolves the registeredpost-terminal-side-effectstask after 120s even if the real side effect is still running, so shutdown can close the writer/DB/Redis while that I/O is still in flight.
Review Coverage
- Logic and correctness
- Security (OWASP Top 10)
- Error handling
- Type safety
- Documentation accuracy
- Test coverage
- Code clarity
Automated review by Codex AI
Contributor
There was a problem hiding this comment.
Review Posted
- Reviewed PR
#1341, applied thesize/XLlabel, and submitted the summary review plus 2 inline comments.
Key Findings
server.js:208— the new outbound backpressure check now rejects any single WebSocket frame over 1 MiB before the firstws.send(), which breaks large non-streamresponse.completedpayloads.src/app/v1/_lib/proxy/response-handler.ts:216—post-terminal-side-effectstime out of AsyncTaskManager tracking after 120s even if the real side effect is still running, so shutdown can close DB/Redis while that work is still in flight.
If you want, I can also pull the exact posted comment text into a compact handoff note for the author.
…ssion control Replace the single postgres.js pool with three isolated lanes (data, control, writer) that share a configurable total connection budget. Each lane gets a per-pool outstanding-operation admission wrapper that fast-fails with DB_POOL_ADMISSION_EXCEEDED before saturating the underlying driver queue, preventing unbounded memory growth under load. Route handlers now wrap with withDataDbScope so request-path queries use the data lane via AsyncLocalStorage, while control-plane and writer traffic remain isolated. Add statement_timeout and lock_timeout at the connection level so slow SQL cannot block the streaming finalization deadline. Introduce LOCAL_OVERLOAD error category so admission rejections are never retried, failovered, or counted against Provider/endpoint circuits.
Introduce createDemandDrivenResponsePump to replace the TransformStream + tee() pipeline with a single high-water-mark-0 ReadableStream that primes exactly one upstream chunk and never reads ahead past unconsumed downstream demand. This eliminates unbounded buffering for slow clients and removes the generic streaming stale watchdog. Rewrite nodeStreamToWebStreamSafe to pause/resume the Node source on controller.desiredSize backpressure, handle pre-existing errored/destroyed/closed states before listener registration, and guard delayed asynchronous destroy(error) events so they cannot surface as uncaught exceptions after the Web stream has settled. Defer all circuit-breaker, session-binding, and Codex cache side effects into post-terminal tasks so a slow Redis call cannot turn an already-completed billable request into a fallback 500 or block lease settlement.
…ssage requests Introduce enqueueMessageRequestUpdateDurably which returns a Promise that resolves only after the batch SQL commits, using RETURNING id and a status_code IS NULL fence to guarantee exactly-once terminal persistence. If the durable write fails or the row is already finalized, a conditional fallback (updateMessageRequestDetailsIfUnfinalized) prevents overwriting an existing terminal status. The write buffer now uses a dedicated writer-lane DB handle, an evictable min-heap for priority-aware overflow dropping, and aggregated overflow logging. Public-status rollup callbacks fire only after the batch commit acknowledgement, preventing premature or duplicate rollup emission when a timeout fallback races with a late primary write.
… leases atomically Consolidate all rolling-window and fixed-window cost writes into one Redis pipeline per trackCost / trackUserDailyCost call, eliminating sequential round-trips. The unified TRACK_COST_ROLLING_WINDOW script is now write-only (cleanup + append + TTL repair) and no longer scans the full ZSET on every successful request. Add settleLeaseBudgets: a single Lua invocation that validates all twelve lease keys, applies decrements, and writes an idempotency marker in one atomic step. Replace twelve fire-and-forget decrementLeaseBudget calls with one settlement, preventing duplicate deductions on ioredis reconnect. Configure commandTimeout, socketTimeout, and autoResendUnfulfilledCommands on the shared Redis client so a TCP blackhole cannot grow the command queue or replay timed-out writes after reconnection.
…uiescence AsyncTaskManager.register now accepts a factory invoked after admission rather than a pre-started Promise, enabling shutdown to abort and join all pending generations—including tail tasks registered by abort listeners. shutdownAll returns a shared Promise that loops until the pending set is empty. runApplicationCleanup treats async-task settlement, writer flush, and DB pool close as non-detachable critical barriers: per-step timeouts become soft warnings, and failures propagate to server.js which exits non-zero. The hard-exit watchdog timer is now referenced so the process stays alive long enough to report a truthful exit status.
Replace the raw headersToRecord helper with redactHeaders so authorization, cookie, x-api-key, and set-cookie values are masked before being written to Langfuse generation metadata, preventing secret leakage in observability exports.
Verify that buildRedisOptionsForUrl sets commandTimeout, socketTimeout, and autoResendUnfulfilledCommands on both redis:// and rediss:// URLs, and that REDIS_COMMAND_TIMEOUT_MS overrides the defaults. Guards the rate-limit performance commit against silent regressions in Redis client hardening.
…callback The public-status rollup test for timed-out durable waiters previously typed onCommitted as () => void and invoked it with no arguments, mismatching the real DurableMessageRequestUpdateOptions contract which receives the committed MessageRequestUpdatePatch. Align the test mock and invocation so the delayed-commit path exercises the actual callback signature. Also fix import ordering in message.ts to satisfy the formatter.
Replace fire-and-forget safeSend with a per-WebSocket outbound queue that caps pending bytes at 1 MiB, serializes sends behind a single in-flight callback, and pauses the upstream SSE response until the client drains. Late callbacks from a closed socket are invalidated by generation counter so they cannot deliver stale frames. Guard the internal HTTP request body write with a 30 s drain timeout and destroy both the request and response when the client vanishes mid-payload. Error and close paths now send a structured error frame before initiating the WebSocket close handshake so clients always receive a terminal event.
Capture the sessionId at increment time into a dedicated variable so the finally block decrements exactly the session it acquired, even if ProxySession.fromContext or the guard pipeline mutates session state before forwarding begins. Previously the finally block re-read session.sessionId, which could be null or different by the time the handler reached its cleanup path, leaking a concurrency slot. Add unit tests covering early guard rejection, successful forwarding, post-session error translation, and pre-session decode failures.
Replace the cloned-response reader pattern with a single demand-driven response pump that feeds the client stream and observes chunks for stats collection. The client now receives a new Response wrapping the pump stream instead of the original upstream Response, eliminating the unbounded clone that held a second copy of every streaming byte in memory. The pump arms a 60 s deadline on each unconsumed lookahead chunk so a connected-but-non-reading client cannot pin the upstream connection indefinitely. Client cancellation propagates to the source even when the cancel observer throws, and reentrant cancel calls preserve the first hard-cancel owner. Response handler terminal paths now use durable persistence with conditional fallback and bounded failure deadlines so a hanging database cannot stall the streaming task indefinitely.
…king Switch error-handler from separate duration and details writes to a single updateMessageRequestDetailsDurably call that includes durationMs, ensuring the Langfuse trace, persistence commit, and status-tracker endRequest fire in deterministic order. The trace is emitted first, the durable write is awaited, and only then is the status tracker released — so a database failure surfaces as a rejected handler rather than a silently orphaned trace. Add comprehensive unit tests for client-safe message sanitisation, override application, terminal status mapping, durable persistence ordering, and the end-to-end terminal outcome contract.
…ansport controller Replace combineAbortSignals polyfill with a dedicated AbortController whose abort is driven by lightweight bindClientAbortListener registrations on the response and client signals. The client-signal listener is cleaned up as soon as the forwarder obtains the upstream response, so a client disconnect after headers no longer aborts the in-flight transport — only the response controller retains that authority through stream consumption. Add an integration test exercising real loopback transports through the hedge lifecycle: winner settlement fences loser timers, each launched transport releases its agent exactly once, database overload is classified before fanout, and hedge winner/loser billing fires exactly once per request.
Add a PostgreSQL integration test that exercises the async message write buffer under row-level locks: mixed durable/ordinary batches settle only after commit, timed-out primaries yield to fallback ownership, retired pending patches are excluded from reinserted generations, a saturated 5000-entry queue evicts non-terminal patches while retaining terminal priority, and winner-cost retries respect the bounded cadence with authoritative loser-cost accumulation.
Align test assertions with the production call site, which now requires a timezone argument to resolve date presets deterministically.
…, and readback Add a shared drizzle query tracing harness and comprehensive unit tests for the message repository: terminal detail writes with durable acknowledgement and CAS-guarded unfinalized claims, winner and hedge-loser cost accounting with idempotent retries, session stats aggregation with provider/model/cache-TTL grouping, paged session request queries with adjacent-sequence lookup, usage-log filtering with ledger fallback, and public readback projections for single-request, session, and audit lookups.
Diagnostic reports now exclude the entire process environment via --report-exclude-env in both Dockerfiles. Crash handlers wrap database errors through findSafeDatabaseError before writing report files or stderr fallbacks, preventing DrizzleQueryError SQL text and bound parameters from reaching any public boundary. The proxy error handler returns a generic 503/500 for database failures without loading system settings or awaiting durable persistence. Langfuse traces and session metadata strip x-cch-* reserved internal headers. clearSessionProvider uses a Redis CAS compare so late cleanup from a prior provider cannot delete a binding already moved to a new one.
The first durable write claimant owns the terminal patch; later contenders observe its acknowledgement but cannot overwrite the committed outcome. Request duration is merged into the same CAS patch so overflow or process exit cannot leave a permanently active record. Post-terminal side effects (provider circuit mutation, session binding clear, Codex cache binding) fire from an onCommitted callback after the SQL write is acknowledged rather than speculatively before it. server.js WebSocket turns track per-turn request, response, and settle lifecycle: sendErrorAndClose aborts the active internal request before queuing a fatal frame, and terminal events wait for their WS send acknowledgement before releasing turn ownership.
Background schedulers (probe, public-status, log-cleanup) now expose their in-flight promise so shutdown awaits true quiescence instead of racing a fixed timeout. Bull queues are joined before DB pool and Redis closure; failures are collected into an AggregateError so critical cleanup still runs. Ledger backfill registers as a cancellable AsyncTaskManager job with abort-signal checkpoints between batches. The lease settlement Lua script consumes the cached slice when a request exceeds the remaining budget, preventing repeated overshoot within the refresh window.
ding113
force-pushed
the
perf/full-path-benchmark-optimization
branch
from
July 17, 2026 06:26
8351fd9 to
048caed
Compare
github-project-automation
Bot
moved this from Backlog
to Done
in Claude Code Hub Roadmap
This was referenced Jul 17, 2026
There was a problem hiding this comment.
Actionable comments posted: 9
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/lib/session-manager.ts (1)
79-90: 🔒 Security & Privacy | 🟠 Major | ⚡ Quick win统一脱敏对象形式的 header snapshot。
当前只有
Headers输入会掩码authorization、cookie 等敏感值；对象输入和旧 JSON 记录只过滤内部头，因此可能把原始凭据写入 Redis或重新返回给调用方。两个路径都应复用headersToSanitizedObject。建议修改
- return record; + return headersToSanitizedObject(new Headers(record)); @@ - const normalized = Object.fromEntries( - Object.entries(headers).filter( - ([key, value]) => typeof value === "string" && !isReservedInternalHeader(key) - ) - ); + const normalized = headersToSanitizedObject(new Headers(headers));Also applies to: 130-145
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/lib/session-manager.ts` around lines 79 - 90, 统一 parseHeaderRecord 与对象输入路径的脱敏逻辑：不要直接保留对象或旧 JSON 记录中的 header 值，而应先转换为 Headers，再复用 headersToSanitizedObject，确保 authorization、cookie 等敏感字段一致被掩码，同时继续过滤内部 header，并让 Redis 写入和返回调用方的快照都使用该统一结果。
🧹 Nitpick comments (2)
src/app/v1/_lib/proxy/response-handler.ts (1)
53-56: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win使用
@/路径别名导入 demand-driven pump。Line 53-56 新增了相对路径导入，不符合项目统一的源码导入约定。
建议修改
import { createDemandDrivenResponsePump, type DemandDrivenResponsePump, -} from "./demand-driven-response-pump"; +} from "`@/app/v1/_lib/proxy/demand-driven-response-pump`";As per coding guidelines，TypeScript 导入必须使用映射到
./src/的@/路径别名。🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/app/v1/_lib/proxy/response-handler.ts` around lines 53 - 56, Update the demand-driven response pump import in response-handler.ts to use the project’s `@/` alias mapped to ./src/ instead of the relative "./demand-driven-response-pump" path, preserving the imported symbols createDemandDrivenResponsePump and DemandDrivenResponsePump.Source: Coding guidelines
src/repository/message.ts (1)
492-496: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win移除 TypeScript 注释中的 emoji。
这两处装饰字符违反仓库规范。
建议修改
- model?: string; // ⭐ 新增：支持更新重定向后的模型名称 + model?: string; // 支持更新重定向后的模型名称 actualResponseModel?: string | null; // 上游响应实际返回的模型名(audit 用途，不影响计费) - providerId?: number; // ⭐ 新增：支持更新最终供应商ID（重试切换后） + providerId?: number; // 支持更新最终供应商ID（重试切换后）As per coding guidelines，“Never use emoji characters in any code, comments, or string literals”。
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/repository/message.ts` around lines 492 - 496, Remove the emoji decoration characters from the comments on the model and providerId fields in the relevant message type, while preserving the comments’ descriptive text and field behavior.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/app/v1/_lib/proxy/error-handler.ts`:
- Around line 298-307: Update the terminal persistence flow in ProxyErrorHandler
so both database-error and normal branches attempt the durable write via
logErrorToDatabase, while treating any persistence failure as best-effort and
preserving the original error response. Wrap the write attempt in failure
handling and always call endRequestTracking(session) from a finally block,
covering the related flow around the alternate lines as well.
- Around line 253-264: Update the databaseError fallback in the error-handler
flow to use a five-language static message map keyed by the request locale
instead of hardcoded English. Preserve the existing
getLocale/getErrorMessageServer path, and when locale resolution or translation
fails, select the matching supported-language message with a safe default
locale.
In `@src/app/v1/_lib/proxy/response-handler.ts`:
- Around line 328-341: 更新 runPostTerminalSideEffects，保留 Promise.allSettled
以等待所有后置副作用完成，但不要丢弃其结果；检查 rejected 结果并将失败向外传播，使 session 清理或 circuit
更新失败能够触发外层错误日志，同时维持 signal.aborted 时直接返回的行为。
In `@src/instrumentation.ts`:
- Around line 452-468: Move the __CCH_STOP_BACKGROUND_QUEUES__ hook assignment
before either background queue is initialized, then initialize the cleanup and
notification queues sequentially. Ensure a failure during notification queue
startup still leaves the previously installed hook available to stop the cleanup
queue.
- Around line 362-372: 调整 src/instrumentation.ts 的 362-372 和 533-543
行：在启动生命周期中等待对应模块导入完成后，立即同步调用 AsyncTaskManager.register 注册回填任务，不要仅创建未被等待的 .then()
链。生产环境和开发环境均应保持相同的注册时序，并保留现有 startup-ledger-backfill 配置及 backfillUsageLedger 调用。
In `@src/lib/lifecycle/shutdown.ts`:
- Around line 142-163: Make queue shutdown a resource-ownership barrier: in the
background-queue stop flow around awaitWithWarning, propagate any stop failure
immediately and prevent writer, database, and Redis cleanup from continuing
while queues may remain active. Update the shutdown error propagation at
src/lib/lifecycle/shutdown.ts lines 266-268 accordingly, and adjust
tests/unit/lib/shutdown.test.ts lines 265-309 to assert dependent resources are
not closed after an unproven queue shutdown.
In `@src/repository/message.ts`:
- Around line 586-593: Update the terminal compare-and-set condition in the
onlyIfUnfinalized path of the message update flow to also require
messageRequest.deletedAt to be null, matching the batch durable path. Preserve
the existing id and statusCode predicates so soft-deleted records are excluded
from updates and subsequent publication.
- Around line 510-516: 在终态分支中更新 message request details 时，返回
updateMessageRequestDetailsDurably(id, details) 的实际布尔结果，而不是无条件返回 true；保留 await
以确保调用完成，并维持非终态 metadata 的现有处理路径。
In `@tests/unit/proxy/response-handler-gemini-terminal.test.ts`:
- Around line 10-16: Update the mocks in the hoisted `mocks` object so `details`
and `conditional` both preserve the `Promise<boolean>` return contract of
`updateMessageRequestDetails` and `updateMessageRequestDetailsIfUnfinalized`;
adjust the conditional mock’s pending implementation to resolve `true` rather
than `void`.
---
Outside diff comments:
In `@src/lib/session-manager.ts`:
- Around line 79-90: 统一 parseHeaderRecord 与对象输入路径的脱敏逻辑：不要直接保留对象或旧 JSON 记录中的
header 值，而应先转换为 Headers，再复用 headersToSanitizedObject，确保 authorization、cookie
等敏感字段一致被掩码，同时继续过滤内部 header，并让 Redis 写入和返回调用方的快照都使用该统一结果。
---
Nitpick comments:
In `@src/app/v1/_lib/proxy/response-handler.ts`:
- Around line 53-56: Update the demand-driven response pump import in
response-handler.ts to use the project’s `@/` alias mapped to ./src/ instead of
the relative "./demand-driven-response-pump" path, preserving the imported
symbols createDemandDrivenResponsePump and DemandDrivenResponsePump.
In `@src/repository/message.ts`:
- Around line 492-496: Remove the emoji decoration characters from the comments
on the model and providerId fields in the relevant message type, while
preserving the comments’ descriptive text and field behavior.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 04569eef-cd66-4818-98f0-a60a8910a25a
📒 Files selected for processing (110)
.env.exampleDockerfiledeploy/Dockerfiledeploy/k8s/app/deployment.yamlserver.jssrc/app/v1/[...route]/route.tssrc/app/v1/_lib/proxy-handler.tssrc/app/v1/_lib/proxy/demand-driven-response-pump.test.tssrc/app/v1/_lib/proxy/demand-driven-response-pump.tssrc/app/v1/_lib/proxy/error-handler.tssrc/app/v1/_lib/proxy/errors.tssrc/app/v1/_lib/proxy/forwarder.tssrc/app/v1/_lib/proxy/node-stream-to-web.test.tssrc/app/v1/_lib/proxy/node-stream-to-web.tssrc/app/v1/_lib/proxy/provider-selector.tssrc/app/v1/_lib/proxy/response-handler.tssrc/app/v1beta/[...route]/route.tssrc/drizzle/admitted-client.tssrc/drizzle/db.tssrc/instrumentation.tssrc/lib/async-task-manager.tssrc/lib/config/env.schema.tssrc/lib/langfuse/emit-proxy-trace.tssrc/lib/langfuse/trace-proxy-request.tssrc/lib/ledger-backfill/service.tssrc/lib/lifecycle/shutdown.tssrc/lib/price-sync/cloud-price-updater.tssrc/lib/provider-endpoints/probe-log-cleanup.tssrc/lib/provider-endpoints/probe-scheduler.tssrc/lib/public-status/scheduler.tssrc/lib/rate-limit/lease-service.tssrc/lib/rate-limit/service.tssrc/lib/redis/client.tssrc/lib/redis/lua-scripts.tssrc/lib/session-manager.tssrc/repository/message-write-buffer.tssrc/repository/message.tstests/configs/integration.config.tstests/integration/billing-model-source.test.tstests/integration/db-pool-isolation-postgres.test.tstests/integration/db-pool-slow-close-postgres.test.tstests/integration/lease-settlement-redis.test.tstests/integration/message-write-buffer-recovery-postgres.test.tstests/integration/proxy-hedge-lifecycle.test.tstests/integration/rolling-cost-redis.test.tstests/unit/dashboard/user-insights-page.test.tsxtests/unit/deploy-dockerfile-contract.test.tstests/unit/drizzle/db-admission.test.tstests/unit/drizzle/db-pool-config.test.tstests/unit/drizzle/db-scope.test.tstests/unit/drizzle/db-shutdown.test.tstests/unit/instrumentation-crash-handler.test.tstests/unit/langfuse/emit-proxy-trace.test.tstests/unit/langfuse/langfuse-trace.test.tstests/unit/lib/async-task-manager-edge-runtime.test.tstests/unit/lib/rate-limit/lease-service.test.tstests/unit/lib/rate-limit/rolling-window-5h.test.tstests/unit/lib/rate-limit/rolling-window-cache-warm.test.tstests/unit/lib/rate-limit/service-extra.test.tstests/unit/lib/redis/client.test.tstests/unit/lib/session-manager-helpers.test.tstests/unit/lib/session-manager-terminate-session.test.tstests/unit/lib/shutdown.test.tstests/unit/price-sync/cloud-price-updater.test.tstests/unit/proxy/build-request-details-redaction.test.tstests/unit/proxy/client-abort-vs-upstream-499.test.tstests/unit/proxy/connected-non-reader-lifetime.test.tstests/unit/proxy/error-handler-client-message.test.tstests/unit/proxy/error-handler-durable-persistence.test.tstests/unit/proxy/error-handler-langfuse-trace.test.tstests/unit/proxy/error-handler-overrides.test.tstests/unit/proxy/error-handler-terminal-status.test.tstests/unit/proxy/pricing-no-price.test.tstests/unit/proxy/provider-selector-cross-type-model.test.tstests/unit/proxy/provider-selector-model-mismatch-binding.test.tstests/unit/proxy/proxy-forwarder-hedge-first-byte.test.tstests/unit/proxy/proxy-forwarder-retry-limit.test.tstests/unit/proxy/proxy-handler-concurrency-ownership.test.tstests/unit/proxy/proxy-handler-public-errors.test.tstests/unit/proxy/proxy-handler-public-success.test.tstests/unit/proxy/response-handler-abort-listener-cleanup.test.tstests/unit/proxy/response-handler-bill-non-success.test.tstests/unit/proxy/response-handler-client-abort-drain.test.tstests/unit/proxy/response-handler-endpoint-circuit-isolation.test.tstests/unit/proxy/response-handler-exported-finalizers.test.tstests/unit/proxy/response-handler-gemini-stream-passthrough-timeouts.test.tstests/unit/proxy/response-handler-gemini-terminal.test.tstests/unit/proxy/response-handler-hedge-loser-priority.test.tstests/unit/proxy/response-handler-lease-decrement.test.tstests/unit/proxy/response-handler-non200.test.tstests/unit/proxy/response-handler-nonstream-terminal.test.tstests/unit/proxy/response-handler-public-dispatch.test.tstests/unit/proxy/response-handler-stream-terminal.test.tstests/unit/proxy/terminal-outcome-contract.test.tstests/unit/repository/message-aggregate-multiple-session-stats.test.tstests/unit/repository/message-aggregate-session-stats.test.tstests/unit/repository/message-public-readback.test.tstests/unit/repository/message-public-status-rollup.test.tstests/unit/repository/message-query-test-support.tstests/unit/repository/message-session-readback.test.tstests/unit/repository/message-session-request-query.test.tstests/unit/repository/message-terminal-cas-durable.test.tstests/unit/repository/message-terminal-cost-accounting.test.tstests/unit/repository/message-terminal-public-status-seam.test.tstests/unit/repository/message-terminal-write-apis.test.tstests/unit/repository/message-usage-logs-query.test.tstests/unit/repository/message-write-buffer.test.tstests/unit/server-response-write-backpressure.test.tstests/unit/server-shutdown.test.tstests/unit/usage-ledger/backfill.test.ts
🚧 Files skipped from review as they are similar to previous changes (66)
- deploy/k8s/app/deployment.yaml
- tests/unit/proxy/client-abort-vs-upstream-499.test.ts
- src/app/v1beta/[...route]/route.ts
- tests/unit/lib/rate-limit/rolling-window-cache-warm.test.ts
- tests/unit/dashboard/user-insights-page.test.tsx
- tests/configs/integration.config.ts
- tests/unit/proxy/response-handler-bill-non-success.test.ts
- src/app/v1/[...route]/route.ts
- src/app/v1/_lib/proxy-handler.ts
- tests/unit/proxy/response-handler-non200.test.ts
- src/lib/redis/client.ts
- tests/unit/repository/message-public-readback.test.ts
- src/lib/price-sync/cloud-price-updater.ts
- tests/unit/proxy/error-handler-terminal-status.test.ts
- tests/unit/drizzle/db-scope.test.ts
- tests/unit/repository/message-terminal-cost-accounting.test.ts
- tests/unit/repository/message-query-test-support.ts
- tests/unit/proxy/pricing-no-price.test.ts
- tests/unit/proxy/proxy-handler-concurrency-ownership.test.ts
- tests/unit/proxy/response-handler-abort-listener-cleanup.test.ts
- tests/unit/proxy/response-handler-exported-finalizers.test.ts
- tests/unit/repository/message-terminal-public-status-seam.test.ts
- tests/integration/proxy-hedge-lifecycle.test.ts
- tests/unit/repository/message-aggregate-multiple-session-stats.test.ts
- tests/unit/langfuse/langfuse-trace.test.ts
- src/app/v1/_lib/proxy/node-stream-to-web.test.ts
- tests/unit/proxy/error-handler-overrides.test.ts
- tests/integration/rolling-cost-redis.test.ts
- src/app/v1/_lib/proxy/node-stream-to-web.ts
- tests/unit/proxy/proxy-handler-public-success.test.ts
- tests/unit/proxy/response-handler-public-dispatch.test.ts
- tests/unit/lib/rate-limit/lease-service.test.ts
- src/lib/config/env.schema.ts
- tests/unit/repository/message-aggregate-session-stats.test.ts
- tests/unit/repository/message-terminal-write-apis.test.ts
- tests/unit/proxy/proxy-handler-public-errors.test.ts
- tests/integration/db-pool-slow-close-postgres.test.ts
- tests/integration/db-pool-isolation-postgres.test.ts
- tests/unit/repository/message-usage-logs-query.test.ts
- tests/unit/proxy/connected-non-reader-lifetime.test.ts
- src/lib/redis/lua-scripts.ts
- tests/unit/proxy/response-handler-hedge-loser-priority.test.ts
- tests/unit/repository/message-public-status-rollup.test.ts
- tests/unit/lib/redis/client.test.ts
- tests/unit/proxy/error-handler-langfuse-trace.test.ts
- tests/unit/drizzle/db-shutdown.test.ts
- tests/unit/lib/rate-limit/service-extra.test.ts
- tests/unit/proxy/proxy-forwarder-hedge-first-byte.test.ts
- src/lib/rate-limit/service.ts
- src/app/v1/_lib/proxy/demand-driven-response-pump.ts
- tests/unit/repository/message-write-buffer.test.ts
- tests/unit/proxy/response-handler-endpoint-circuit-isolation.test.ts
- src/lib/async-task-manager.ts
- tests/integration/lease-settlement-redis.test.ts
- tests/unit/lib/async-task-manager-edge-runtime.test.ts
- tests/unit/lib/rate-limit/rolling-window-5h.test.ts
- tests/integration/billing-model-source.test.ts
- tests/unit/server-shutdown.test.ts
- tests/unit/proxy/response-handler-lease-decrement.test.ts
- tests/unit/drizzle/db-pool-config.test.ts
- src/drizzle/db.ts
- tests/unit/proxy/response-handler-gemini-stream-passthrough-timeouts.test.ts
- src/app/v1/_lib/proxy/demand-driven-response-pump.test.ts
- src/app/v1/_lib/proxy/forwarder.ts
- tests/unit/proxy/response-handler-client-abort-drain.test.ts
- tests/unit/proxy/error-handler-durable-persistence.test.ts
Comment on lines
+253
to
+264
| if (databaseError) { | ||
| // Drizzle wraps the admission cause with SQL text and bound parameters. | ||
| // Never let that wrapper cross the public, log, or observability boundary. | ||
| try { | ||
| const { getLocale } = await import("next-intl/server"); | ||
| clientErrorMessage = await getErrorMessageServer( | ||
| await getLocale(), | ||
| ERROR_CODES.DATABASE_ERROR | ||
| ); | ||
| } catch { | ||
| clientErrorMessage = "An error occurred"; | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
数据库错误兜底仍然绕过 i18n。
Line 263 的硬编码英文会让非英语用户在 locale 解析失败时收到错误语言。请提供基于请求 locale 的五语言静态兜底，而不是直接返回英文。
依据编码规范：“All user-facing strings must use i18n ... Never hardcode display text”。
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/app/v1/_lib/proxy/error-handler.ts` around lines 253 - 264, Update the
databaseError fallback in the error-handler flow to use a five-language static
message map keyed by the request locale instead of hardcoded English. Preserve
the existing getLocale/getErrorMessageServer path, and when locale resolution or
translation fails, select the matching supported-language message with a safe
default locale.
Source: Coding guidelines
Comment on lines
+298
to
+307
| if (databaseError) { | ||
| ProxyErrorHandler.endRequestTracking(session); | ||
| } else { | ||
| await ProxyErrorHandler.logErrorToDatabase( | ||
| session, | ||
| logErrorMessage, | ||
| finalResponse.status, | ||
| null | ||
| ); | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
让终态持久化成为 best-effort，并始终结束请求追踪。
数据库错误分支完全跳过 durable patch；普通分支若写入失败，则会在 Line 675 前抛出。两种情况都可能留下永久 active 记录，后者还会用二次持久化异常替换原错误响应。应统一尝试 durable 写入、捕获失败，并在 finally 中结束 tracking。
建议修改
- if (databaseError) {
- ProxyErrorHandler.endRequestTracking(session);
- } else {
+ try {
await ProxyErrorHandler.logErrorToDatabase(
session,
logErrorMessage,
finalResponse.status,
- null
+ null,
);
+ } catch (persistenceError) {
+ const safeError = findSafeDatabaseError(persistenceError);
+ logger.warn("ProxyErrorHandler: failed to persist terminal error", {
+ error: safeError?.message ?? "Terminal error persistence failed",
+ databaseCode: safeError?.code,
+ });
+ } finally {
+ ProxyErrorHandler.endRequestTracking(session);
}
@@
- ProxyErrorHandler.endRequestTracking(session);Also applies to: 663-675
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/app/v1/_lib/proxy/error-handler.ts` around lines 298 - 307, Update the
terminal persistence flow in ProxyErrorHandler so both database-error and normal
branches attempt the durable write via logErrorToDatabase, while treating any
persistence failure as best-effort and preserving the original error response.
Wrap the write attempt in failure handling and always call
endRequestTracking(session) from a finally block, covering the related flow
around the alternate lines as well.
Comment on lines
+328
to
+341
| async function runPostTerminalSideEffects( | ||
| effects: ReadonlyArray<() => Promise<void>>, | ||
| signal: AbortSignal | ||
| ): Promise<void> { | ||
| if (signal.aborted) return; | ||
| await Promise.allSettled( | ||
| effects.map((effect) => { | ||
| try { | ||
| return effect(); | ||
| } catch (error) { | ||
| return Promise.reject(error); | ||
| } | ||
| }) | ||
| ); |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
不要静默吞掉后置副作用失败。
Line 333 的 Promise.allSettled() 结果被直接丢弃，因此 session 清理或 circuit 更新失败时，外层错误日志也不会触发，状态可能长期不一致。
建议修改
- await Promise.allSettled(
+ const results = await Promise.allSettled(
effects.map((effect) => {
try {
return effect();
} catch (error) {
return Promise.reject(error);
}
})
);
+ const failures = results
+ .filter((result): result is PromiseRejectedResult => result.status === "rejected")
+ .map((result) => result.reason);
+ if (failures.length > 0) {
+ throw new AggregateError(failures, "Post-terminal side effects failed");
+ }📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| async function runPostTerminalSideEffects( | |
| effects: ReadonlyArray<() => Promise<void>>, | |
| signal: AbortSignal | |
| ): Promise<void> { | |
| if (signal.aborted) return; | |
| await Promise.allSettled( | |
| effects.map((effect) => { | |
| try { | |
| return effect(); | |
| } catch (error) { | |
| return Promise.reject(error); | |
| } | |
| }) | |
| ); | |
| async function runPostTerminalSideEffects( | |
| effects: ReadonlyArray<() => Promise<void>>, | |
| signal: AbortSignal | |
| ): Promise<void> { | |
| if (signal.aborted) return; | |
| const results = await Promise.allSettled( | |
| effects.map((effect) => { | |
| try { | |
| return effect(); | |
| } catch (error) { | |
| return Promise.reject(error); | |
| } | |
| }) | |
| ); | |
| const failures = results | |
| .filter((result): result is PromiseRejectedResult => result.status === "rejected") | |
| .map((result) => result.reason); | |
| if (failures.length > 0) { | |
| throw new AggregateError(failures, "Post-terminal side effects failed"); | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/app/v1/_lib/proxy/response-handler.ts` around lines 328 - 341, 更新
runPostTerminalSideEffects，保留 Promise.allSettled 以等待所有后置副作用完成，但不要丢弃其结果；检查
rejected 结果并将失败向外传播，使 session 清理或 circuit 更新失败能够触发外层错误日志，同时维持 signal.aborted
时直接返回的行为。
Comment on lines
+362
to
+372
| Promise.all([import("@/lib/async-task-manager"), import("@/lib/ledger-backfill")]) | ||
| .then(([{ AsyncTaskManager }, { backfillUsageLedger }]) => { | ||
| AsyncTaskManager.register( | ||
| "startup-ledger-backfill", | ||
| async (signal) => { | ||
| const result = await backfillUsageLedger(signal); | ||
| logger.info("[Instrumentation] Ledger backfill complete", result); | ||
| }, | ||
| { taskType: "startup-ledger-backfill", staleTimeoutMs: Number.POSITIVE_INFINITY } | ||
| ); | ||
| }) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
请在启动生命周期内完成回填任务注册。 两处 .then() 都留下了关停先于注册完成的窗口，可能导致数据库池关闭后才启动回填。
src/instrumentation.ts#L362-L372：先等待模块导入，再同步注册生产环境回填任务。src/instrumentation.ts#L533-L543：对开发环境采用相同的注册时序。
📍 Affects 1 file
src/instrumentation.ts#L362-L372(this comment)src/instrumentation.ts#L533-L543
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/instrumentation.ts` around lines 362 - 372, 调整 src/instrumentation.ts 的
362-372 和 533-543 行：在启动生命周期中等待对应模块导入完成后，立即同步调用 AsyncTaskManager.register
注册回填任务，不要仅创建未被等待的 .then() 链。生产环境和开发环境均应保持相同的注册时序，并保留现有 startup-ledger-backfill
配置及 backfillUsageLedger 调用。
Comment on lines
+452
to
+468
| ( | ||
| globalThis as typeof globalThis & { | ||
| __CCH_STOP_BACKGROUND_QUEUES__?: () => Promise<void>; | ||
| } | ||
| ).__CCH_STOP_BACKGROUND_QUEUES__ = async () => { | ||
| const [{ stopCleanupQueue }, { stopNotificationQueue }] = await Promise.all([ | ||
| import("@/lib/log-cleanup/cleanup-queue"), | ||
| import("@/lib/notification/notification-queue"), | ||
| ]); | ||
| const results = await Promise.allSettled([stopCleanupQueue(), stopNotificationQueue()]); | ||
| const failures = results.flatMap((result) => | ||
| result.status === "rejected" ? [result.reason] : [] | ||
| ); | ||
| if (failures.length > 0) { | ||
| throw new AggregateError(failures, "Failed to stop background queues"); | ||
| } | ||
| }; |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
请在启动队列前安装停止钩子。
若清理队列启动成功但通知队列启动失败，此赋值不会执行，已创建的清理队列便无法被关停流程关闭。应先安装钩子，再依次初始化队列。
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/instrumentation.ts` around lines 452 - 468, Move the
__CCH_STOP_BACKGROUND_QUEUES__ hook assignment before either background queue is
initialized, then initialize the cleanup and notification queues sequentially.
Ensure a failure during notification queue startup still leaves the previously
installed hook available to stop the cleanup queue.
Comment on lines
+142
to
+163
| // 5. Bull queues own Redis connections and may still ACK jobs or emit DB work. | ||
| // Join them before closing either backing resource. | ||
| try { | ||
| await awaitWithWarning( | ||
| (async () => { | ||
| const stopQueues = ( | ||
| globalThis as typeof globalThis & { | ||
| __CCH_STOP_BACKGROUND_QUEUES__?: () => Promise<void>; | ||
| } | ||
| ).__CCH_STOP_BACKGROUND_QUEUES__; | ||
| if (stopQueues) await stopQueues(); | ||
| })(), | ||
| stepMs, | ||
| "stopBackgroundQueues" | ||
| ); | ||
| } catch (error) { | ||
| const queueError = error instanceof Error ? error : new Error(String(error)); | ||
| deferredErrors.push(queueError); | ||
| logger.error("[Shutdown] background queues failed to stop; continuing critical cleanup", { | ||
| error: queueError.message, | ||
| }); | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
队列停止失败后仍关闭其依赖，破坏了资源所有权屏障。
src/lib/lifecycle/shutdown.ts#L142-L163：只有在所有队列 worker 已完成 join 后，才能继续 writer、数据库及 Redis 回收。src/lib/lifecycle/shutdown.ts#L266-L268：不要等依赖关闭后才传播无法证明静默的队列错误。tests/unit/lib/shutdown.test.ts#L265-L309：调整测试，禁止在队列仍可能活跃时关闭其依赖资源。
📍 Affects 2 files
src/lib/lifecycle/shutdown.ts#L142-L163(this comment)src/lib/lifecycle/shutdown.ts#L266-L268tests/unit/lib/shutdown.test.ts#L265-L309
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/lib/lifecycle/shutdown.ts` around lines 142 - 163, Make queue shutdown a
resource-ownership barrier: in the background-queue stop flow around
awaitWithWarning, propagate any stop failure immediately and prevent writer,
database, and Redis cleanup from continuing while queues may remain active.
Update the shutdown error propagation at src/lib/lifecycle/shutdown.ts lines
266-268 accordingly, and adjust tests/unit/lib/shutdown.test.ts lines 265-309 to
assert dependent resources are not closed after an unproven queue shutdown.
Comment on lines
+510
to
516
| if (getEnvConfig().MESSAGE_REQUEST_WRITE_MODE === "async" && !options.onlyIfUnfinalized) { | ||
| // 终态 patch 必须观察 SQL commit 后再发布 public-status rollup。 | ||
| // 非终态 metadata 仍保持轻量 enqueue，但不能伪称已提交。 | ||
| if (details.statusCode !== undefined) { | ||
| await updateMessageRequestDetailsDurably(id, details); | ||
| return true; | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick win
返回耐久写入的实际提交结果。
另一终态写入已取得所有权时，updateMessageRequestDetailsDurably() 会返回 false；当前代码却无条件返回 true，破坏了新布尔返回值的 CAS 语义。
建议修改
if (details.statusCode !== undefined) {
- await updateMessageRequestDetailsDurably(id, details);
- return true;
+ return updateMessageRequestDetailsDurably(id, details);
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| if (getEnvConfig().MESSAGE_REQUEST_WRITE_MODE === "async" && !options.onlyIfUnfinalized) { | |
| // 终态 patch 必须观察 SQL commit 后再发布 public-status rollup。 | |
| // 非终态 metadata 仍保持轻量 enqueue，但不能伪称已提交。 | |
| if (details.statusCode !== undefined) { | |
| await updateMessageRequestDetailsDurably(id, details); | |
| return true; | |
| } | |
| if (getEnvConfig().MESSAGE_REQUEST_WRITE_MODE === "async" && !options.onlyIfUnfinalized) { | |
| // 终态 patch 必须观察 SQL commit 后再发布 public-status rollup。 | |
| // 非终态 metadata 仍保持轻量 enqueue，但不能伪称已提交。 | |
| if (details.statusCode !== undefined) { | |
| return updateMessageRequestDetailsDurably(id, details); | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/repository/message.ts` around lines 510 - 516, 在终态分支中更新 message request
details 时，返回 updateMessageRequestDetailsDurably(id, details) 的实际布尔结果，而不是无条件返回
true；保留 await 以确保调用完成，并维持非终态 metadata 的现有处理路径。
Comment on lines
+586
to
+593
| if (options.onlyIfUnfinalized) { | ||
| const terminalDb = | ||
| getEnvConfig().MESSAGE_REQUEST_WRITE_MODE === "async" ? getMessageWriterDb() : db; | ||
| const updated = await terminalDb | ||
| .update(messageRequest) | ||
| .set(updateData) | ||
| .where(and(eq(messageRequest.id, id), isNull(messageRequest.statusCode))) | ||
| .returning({ id: messageRequest.id }); |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
终态 CAS 必须排除已软删除记录。
批量耐久路径包含 deleted_at IS NULL，但此回退路径仅检查 status_code。因此软删除且尚未终态化的记录仍可被更新，并触发提交观察者和公共状态发布。
建议修改
.update(messageRequest)
.set(updateData)
- .where(and(eq(messageRequest.id, id), isNull(messageRequest.statusCode)))
+ .where(
+ and(
+ eq(messageRequest.id, id),
+ isNull(messageRequest.statusCode),
+ isNull(messageRequest.deletedAt)
+ )
+ )
.returning({ id: messageRequest.id });📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| if (options.onlyIfUnfinalized) { | |
| const terminalDb = | |
| getEnvConfig().MESSAGE_REQUEST_WRITE_MODE === "async" ? getMessageWriterDb() : db; | |
| const updated = await terminalDb | |
| .update(messageRequest) | |
| .set(updateData) | |
| .where(and(eq(messageRequest.id, id), isNull(messageRequest.statusCode))) | |
| .returning({ id: messageRequest.id }); | |
| if (options.onlyIfUnfinalized) { | |
| const terminalDb = | |
| getEnvConfig().MESSAGE_REQUEST_WRITE_MODE === "async" ? getMessageWriterDb() : db; | |
| const updated = await terminalDb | |
| .update(messageRequest) | |
| .set(updateData) | |
| .where( | |
| and( | |
| eq(messageRequest.id, id), | |
| isNull(messageRequest.statusCode), | |
| isNull(messageRequest.deletedAt) | |
| ) | |
| ) | |
| .returning({ id: messageRequest.id }); |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/repository/message.ts` around lines 586 - 593, Update the terminal
compare-and-set condition in the onlyIfUnfinalized path of the message update
flow to also require messageRequest.deletedAt to be null, matching the batch
durable path. Preserve the existing id and statusCode predicates so soft-deleted
records are excluded from updates and subsequent publication.
Comment on lines
+10
to
+16
| const mocks = vi.hoisted(() => ({ | ||
| conditional: | ||
| vi.fn<(id: number, details: object, options?: TerminalWriterOptions) => Promise<boolean>>(), | ||
| details: vi.fn<(id: number, details: object) => Promise<void>>(), | ||
| durable: | ||
| vi.fn<(id: number, details: object, options?: TerminalWriterOptions) => Promise<boolean>>(), | ||
| tasks: Array.from<Promise<void>>([]), |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -eu
ast-grep outline tests/unit/proxy/response-handler-gemini-terminal.test.ts \
--items all --match 'mocks|beforeEach'
rg -n -C2 \
'Promise<void>|mockResolvedValue\(undefined\)' \
tests/unit/proxy/response-handler-gemini-terminal.test.ts
fd -a -t f '^tsconfig.*\.json$' . \
--exec sh -c 'echo "--- $1"; cat "$1"' _ {}Repository: ding113/claude-code-hub
Length of output: 2728
🏁 Script executed:
#!/bin/bash
set -eu
echo "== test file context =="
sed -n '1,280p' tests/unit/proxy/response-handler-gemini-terminal.test.ts | cat -n
echo
echo "== production symbols =="
rg -n -C3 'conditional|details|durable|TerminalWriterOptions|ProxyResponseHandler' src tests -g '!**/dist/**' -g '!**/.next/**'Repository: ding113/claude-code-hub
Length of output: 50380
🏁 Script executed:
#!/bin/bash
set -eu
echo "== find relevant interfaces/usages =="
rg -n -C4 'details\s*:\s*vi\.fn|conditional\s*:\s*vi\.fn|durable\s*:\s*vi\.fn|Promise<boolean>|Promise<void>' \
tests src -g '!**/dist/**' -g '!**/.next/**'
echo
echo "== locate terminal writer / proxy handler definitions =="
fd -a -t f 'response-handler|terminal' src testsRepository: ding113/claude-code-hub
Length of output: 50380
🏁 Script executed:
#!/bin/bash
set -eu
echo "== search for the likely methods =="
rg -n -C3 '\bconditional\b|\bdetails\b|\bdurable\b' src tests | sed -n '1,220p'
echo
echo "== search for writer interfaces =="
rg -n -C4 'interface .*Writer|type .*Writer|TerminalWriterOptions|onCommitted' src testsRepository: ding113/claude-code-hub
Length of output: 50380
🏁 Script executed:
#!/bin/bash
set -eu
sed -n '1,260p' tests/unit/proxy/response-handler-gemini-terminal.test.ts | cat -nRepository: ding113/claude-code-hub
Length of output: 10470
🏁 Script executed:
#!/bin/bash
set -eu
sed -n '560,710p' src/repository/message.ts | cat -nRepository: ding113/claude-code-hub
Length of output: 6423
让这两个 mock 保持 Promise<boolean> 契约。 updateMessageRequestDetails / updateMessageRequestDetailsIfUnfinalized 都返回 Promise<boolean>，这里的 details 仍是 Promise<void>，conditional 的挂起实现也应返回 Promise<boolean> 并 resolve true，否则测试会偏离当前持久化路径的返回值。
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/unit/proxy/response-handler-gemini-terminal.test.ts` around lines 10 -
16, Update the mocks in the hoisted `mocks` object so `details` and
`conditional` both preserve the `Promise<boolean>` return contract of
`updateMessageRequestDetails` and `updateMessageRequestDetailsIfUnfinalized`;
adjust the conditional mock’s pending implementation to resolve `true` rather
than `void`.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 048caedb62
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
Comment on lines
+364
to
+367
| AsyncTaskManager.register( | ||
| "startup-ledger-backfill", | ||
| async (signal) => { | ||
| const result = await backfillUsageLedger(signal); |
There was a problem hiding this comment.
In production startup, a pod that receives SIGTERM while this registered backfill is inside its 10k-row tx.execute batch will have shutdownAllAsyncTasks() abort and then join the task, but backfillUsageLedger only observes the signal between batches and cannot cancel the running SQL. The DB statement timeout defaults to 90s while the server hard-exit watchdog defaults to 28s, so large backfills can turn normal rolling shutdowns into process.exit(1) before cleanup finishes; either keep this backfill detached or make the batch query actually abortable before joining it.
Useful? React with 👍 / 👎.
| let errorCategory = await categorizeErrorAsync(error); | ||
| lastErrorCategory = errorCategory; | ||
| const statusCode = error instanceof ProxyError ? error.statusCode : undefined; | ||
| const databaseError = findSafeDatabaseError(error); |
There was a problem hiding this comment.
For hedged streaming attempts, a non-admission Drizzle query failure is recognized here as a safe database error, but errorCategory is not promoted the way the non-hedged path does. When the classifier leaves that local DB failure as SYSTEM_ERROR, the next block records endpoint failure and the attempt proceeds through provider failover, so a local database outage can penalize healthy provider endpoints and launch unnecessary alternatives; handle any databaseError as local overload before endpoint accounting.
Useful? React with 👍 / 👎.
Comment on lines
+298
to
+299
| if (databaseError) { | ||
| ProxyErrorHandler.endRequestTracking(session); |
There was a problem hiding this comment.
When a request has already created a message_request and a local DB error comes from the data/control pool, this branch only ends in-memory tracking and skips the durable terminal write entirely. Since the writer pool is isolated for terminal updates, those 503/500 responses can leave status_code and duration_ms NULL permanently, making active/history views and rollups wrong; try the writer-backed terminal persistence and swallow only if that also fails.
Useful? React with 👍 / 👎.
13 tasks
This was referenced Jul 20, 2026
10 tasks
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
This PR is the accumulated full-path concurrency, persistence, and resource-ownership candidate based on
dev@6fcb827b. It contains 17 commits across 112 files (+18,684/-2,757), including one GitHub Actions formatting-only commit. Local benchmark infrastructure and raw artifacts remain outside the repository.The branch is ready for code review and CI: build and formatter checks passed on the semantic head; the GitHub Actions formatting-only commit has no whitespace-insensitive semantic diff; and typecheck, the full Vitest suite, focused tests, and whitespace checks pass on the final PR head. Repository-wide lint currently reports pre-existing optional-chain errors in untouched
src/actions/webhook-targets.ts. This PR does not claim that the final post-change gateway benchmark or two-hour production soak is complete; the benchmark qualification limits and failed runner attempts are documented below.Changes by objective
Performance and concurrency
DB_POOL_MAX=20resolves to15/4/1; Kubernetes uses a total budget of24rather than the previous per-pod30.Response.clone()/tee()draining with a demand-driven single-reader pump, bounded lookahead, backpressure propagation, explicit drain/cancel ownership, and connected non-reader deadlines.ZADD+EXPIRE; batch Key/Provider/User counters into one explicit pipeline.Stability, correctness, and resource ownership
status_code IS NULLfallback CAS and publish public rollups from the committed merged patch.Security, observability, and test hardening
.env.exampleand deployment configuration.Benchmark and experiment record
Run census
claude-rep1-5-off-vs-on.jsonandcodex-rep1-5-off-vs-on.jsonare the final selected five-pair historical comparisons. Earlier one/two-pair files are superseded.Direct fixture capacity and scheduler calibration
The original CAR run scheduled only 17,403 requests in 10 seconds and had five transport errors. The cumulative-offer/deadline scheduler was fixed before the corrected run.
Historical high-concurrency mode diagnostics
These 20 cases use the unmodified baseline SHA, five ABBA-balanced repetitions per protocol and mode, 30 s warmup, 120 s measurement, 30 s cooldown, and seeds 20260713-20260717. They show where work amplification existed, but they are not a post-change candidate comparison and the old analyzer used unpaired-mean rather than valid paired-median confidence intervals.
Mode-off dropped 1,111 Claude offers in one repetition and 2,084 Codex offers across five repetitions; mode-on dropped zero. The two-minute windows are insufficient for leak/no-leak claims.
Module-level before/after evidence
DB_POOL_ADMISSION_EXCEEDED.Mixed 30-minute qualification: failed, not promoted
Runtime/integration attempts
cause.code=55P03versus top-levelcode). Cleanup also failed on container-owned files. The failed attempt remains immutable.docker compose stopremained active for roughly 579.77 s after both containers exited; the 600 s outer deadline sent SIGTERM. Cleanup then recorded 21 failures, five missing manifest pairs, five unreadable paths, a partial freeze, and one retained quarantine. No T10 task container/network/volume/listener/process/lock remains; ambient benchmark PostgreSQL/Redis remained healthy.Validation
bun run build: PASS; 187 static pages generated; 11 existing Turbopack/Edge Runtime warnings.bun run lint: the semantic head passed; on the final formatted PR head it reports pre-existing optional-chain errors in untouchedsrc/actions/webhook-targets.ts.bun run lint:fix: PASS; no fixes applied.bun run typecheck: PASS.bun run test: PASS on the final committed tree.as anywith a realProxySessionfixture.git diff --check origin/dev...HEAD: PASS.Readiness and residual risk
Ready now
Not claimed
Review guidance
Suggested order:
Greptile Summary
This PR introduces a comprehensive set of concurrency, persistence, and resource-ownership improvements across the full request pipeline. The changes are split into 17 logical commits covering DB lane pooling, demand-driven response streaming, Redis batching, durable terminal persistence, and shutdown coordination.
applytrap for the tag-template-literal path, and adds statement/lock timeouts per connection.Response.clone()/tee()draining with a pull-based pump that tracks backpressure, bounded lookahead, drain/cancel ownership, and amarkClientAborted=falsefast path for the chunk-deadline timeout—fixing the previousclientAbortedconflation issue.enqueueDurably) with CAS fencing (status_code IS NULL), a 120 s timeout fallback, andonCommittedcallbacks triggered only after the SQL batch is confirmed.shutdownAllinAsyncTaskManagernow waits for all in-flight task generations to settle before pool close;stopMessageRequestWriteBufferis a critical barrier beforecloseDbPools; cleanup failures propagate non-zero to the hard watchdog.Confidence Score: 5/5
The PR is safe to merge; the correctness and resource-ownership invariants across the new durable-persistence and DB-lane paths are sound, and the changes build, type-check, lint, and pass the full test suite.
The three issues flagged in previous review rounds (tag-template-literal admission bypass, clientAborted conflation on chunk deadline, and AbortSignal unused in task factory) are all addressed. The two new observations are minor: a .then(() => false) that should also suppress the rejection case to avoid spurious error logs, and an unbounded while(true) shutdown loop that relies on the external hard watchdog. Neither affects correctness or data integrity under normal operation.
src/repository/message-write-buffer.ts (second-contender rejection path) and src/lib/async-task-manager.ts (shutdown loop iteration bound) warrant a second look.
Important Files Changed
Sequence Diagram
%%{init: {'theme': 'neutral'}}%% sequenceDiagram participant Client participant ResponseHandler participant DemandPump participant WriterBuffer participant DB Client->>ResponseHandler: upstream stream arrives ResponseHandler->>DemandPump: createDemandDrivenResponsePump(source) loop pull-based backpressure Client->>DemandPump: pull() DemandPump->>DemandPump: ensureRead() → pendingChunk DemandPump-->>Client: enqueue(chunk) end Note over DemandPump: source.done=true → finishNormally() DemandPump-->>ResponseHandler: "completion{streamEndedNormally, clientAborted}" ResponseHandler->>ResponseHandler: finalizeDeferredStreaming (sync) ResponseHandler->>DB: updateMessageRequestDetailsDurably (durable CAS) DB-->>WriterBuffer: batch flush + RETURNING id WriterBuffer->>WriterBuffer: notifyDurableCommit → onCommitted callback WriterBuffer-->>ResponseHandler: ack resolved ResponseHandler->>ResponseHandler: schedulePostTerminalSideEffects (circuit + session) alt Client disconnects mid-stream Client->>DemandPump: "cancel(reason) → startDrain(reason, markClientAborted=true)" DemandPump->>DemandPump: "state=draining, drain remaining source" DemandPump-->>ResponseHandler: "completion{clientAborted=true}" end alt Chunk deadline fires (60 s unconsumed) DemandPump->>DemandPump: "startDrain(error, markClientAborted=false)" Note over DemandPump: clientAborted stays false → circuit-breaker not blamed on client end%%{init: {'theme': 'base', 'themeVariables': {"darkMode": true, "background": "#0d1117", "primaryColor": "#21262d", "primaryTextColor": "#e6edf3", "primaryBorderColor": "#8b949e", "lineColor": "#8b949e", "textColor": "#e6edf3", "edgeLabelBackground": "#161b22", "actorBkg": "#21262d", "actorBorder": "#8b949e", "actorTextColor": "#e6edf3", "actorLineColor": "#8b949e", "signalColor": "#8b949e", "signalTextColor": "#e6edf3", "noteBkgColor": "#373320", "noteBorderColor": "#d4a72c", "noteTextColor": "#f0e6c0", "labelBoxBkgColor": "#21262d", "labelBoxBorderColor": "#8b949e", "labelTextColor": "#e6edf3", "loopTextColor": "#e6edf3", "activationBkgColor": "#30363d", "activationBorderColor": "#8b949e"}}}%% sequenceDiagram participant Client participant ResponseHandler participant DemandPump participant WriterBuffer participant DB Client->>ResponseHandler: upstream stream arrives ResponseHandler->>DemandPump: createDemandDrivenResponsePump(source) loop pull-based backpressure Client->>DemandPump: pull() DemandPump->>DemandPump: ensureRead() → pendingChunk DemandPump-->>Client: enqueue(chunk) end Note over DemandPump: source.done=true → finishNormally() DemandPump-->>ResponseHandler: "completion{streamEndedNormally, clientAborted}" ResponseHandler->>ResponseHandler: finalizeDeferredStreaming (sync) ResponseHandler->>DB: updateMessageRequestDetailsDurably (durable CAS) DB-->>WriterBuffer: batch flush + RETURNING id WriterBuffer->>WriterBuffer: notifyDurableCommit → onCommitted callback WriterBuffer-->>ResponseHandler: ack resolved ResponseHandler->>ResponseHandler: schedulePostTerminalSideEffects (circuit + session) alt Client disconnects mid-stream Client->>DemandPump: "cancel(reason) → startDrain(reason, markClientAborted=true)" DemandPump->>DemandPump: "state=draining, drain remaining source" DemandPump-->>ResponseHandler: "completion{clientAborted=true}" end alt Chunk deadline fires (60 s unconsumed) DemandPump->>DemandPump: "startDrain(error, markClientAborted=false)" Note over DemandPump: clientAborted stays false → circuit-breaker not blamed on client endReviews (2): Last reviewed commit: "fix(shutdown): await scheduler quiescenc..." | Re-trigger Greptile