perf: optimize full-path concurrency and resource ownership#1341
Open
ding113 wants to merge 17 commits into
Open
perf: optimize full-path concurrency and resource ownership#1341ding113 wants to merge 17 commits into
ding113 wants to merge 17 commits into
Conversation
…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.
📝 WalkthroughWalkthrough本次变更重构了数据库连接池、代理流生命周期、消息 durable 写入、Redis 限流结算、异步任务关停和 WebSocket 背压处理，并补充大量单元及集成测试覆盖相关时序、错误和资源释放行为。 ChangesRuntime and persistence flows
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
+414
to
417
| (clientReq, clientRes) => { | ||
| currentInternalReq = clientReq; | ||
| if (clientRes) currentInternalRes = clientRes; | ||
| }, |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
关闭后不要重新注册迟到的内部响应。
客户端关闭可能先执行 abortCurrentInternalReq()，随后已排队的 HTTP response 回调再次写入 currentInternalRes。由于 closed 已为 true，后续不会再清理该响应，上游工作可能继续运行和计费。
建议修复
(clientReq, clientRes) => {
+ if (closed) {
+ try {
+ if (!clientReq.destroyed) clientReq.destroy();
+ } catch {}
+ try {
+ if (clientRes && !clientRes.destroyed) clientRes.destroy();
+ } catch {}
+ return;
+ }
currentInternalReq = clientReq;
if (clientRes) currentInternalRes = clientRes;
},📝 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
| (clientReq, clientRes) => { | |
| currentInternalReq = clientReq; | |
| if (clientRes) currentInternalRes = clientRes; | |
| }, | |
| (clientReq, clientRes) => { | |
| if (closed) { | |
| try { | |
| if (!clientReq.destroyed) clientReq.destroy(); | |
| } catch {} | |
| try { | |
| if (clientRes && !clientRes.destroyed) clientRes.destroy(); | |
| } catch {} | |
| return; | |
| } | |
| currentInternalReq = clientReq; | |
| if (clientRes) currentInternalRes = clientRes; | |
| }, |
🤖 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 `@server.js` around lines 414 - 417, 更新内部 HTTP response 回调，在写入
currentInternalReq 或 currentInternalRes 前检查
closed；客户端关闭后直接忽略已排队的迟到回调，避免重新注册响应并绕过 abortCurrentInternalReq
的清理流程。保留未关闭时现有的赋值行为。
Comment on lines
+1693
to
+1708
| if (messageContext) { | ||
| const duration = Date.now() - session.startTime; | ||
| await updateMessageRequestDuration(messageContext.id, duration); | ||
| await updateMessageRequestDetailsDurably(messageContext.id, { | ||
| statusCode: finalizedStatusCode, | ||
| ...errorDetails, | ||
| ttfbMs: session.ttfbMs ?? duration, | ||
| providerChain: session.getProviderChain(), | ||
| model: session.getCurrentModel() ?? undefined, | ||
| providerId: session.provider?.id, | ||
| context1mApplied: session.getContext1mApplied(), | ||
| swapCacheTtlApplied: session.provider?.swapCacheTtlBilling ?? false, | ||
| specialSettings: session.getSpecialSettings() ?? undefined, | ||
| }); | ||
| const tracker = ProxyStatusTracker.getInstance(); | ||
| tracker.endRequest(messageContext.user.id, messageContext.id); |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
为 Gemini 非流式错误终态补上 CAS fallback。
这里的 durable 写入一旦失败，控制流只会进入外层日志处理；不会调用 updateMessageRequestDetailsIfUnfinalized，tracker.endRequest 也不会执行。请复用 persistNonStreamTerminalDetails()，并在 finally 中结束 tracker。
🤖 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 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.
Comment on lines
+2444
to
+2455
| onClientCancel: (reason) => { | ||
| passthroughPump.startDrain(reason); | ||
| passthroughPump.cancelSource(reason); | ||
| }, | ||
| }); | ||
| const cleanupPassthroughClientAbortListener = bindClientAbortListener( | ||
| session.clientAbortSignal, | ||
| () => { | ||
| const reason = session.clientAbortSignal?.reason; | ||
| passthroughPump.startDrain(reason); | ||
| passthroughPump.cancelSource(reason); | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
进入后台 drain 后不要立即取消源流。
startDrain() 后立刻调用 cancelSource() 会马上 settle 并取消 reader，终态 usage 无法继续收集，客户端断开将被错误结算为未计费的 499。应仅由有界 drain、idle 或 response timeout 负责最终取消。
建议修复
onClientCancel: (reason) => {
passthroughPump.startDrain(reason);
- passthroughPump.cancelSource(reason);
},
...
const reason = session.clientAbortSignal?.reason;
passthroughPump.startDrain(reason);
- passthroughPump.cancelSource(reason);📝 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
| onClientCancel: (reason) => { | |
| passthroughPump.startDrain(reason); | |
| passthroughPump.cancelSource(reason); | |
| }, | |
| }); | |
| const cleanupPassthroughClientAbortListener = bindClientAbortListener( | |
| session.clientAbortSignal, | |
| () => { | |
| const reason = session.clientAbortSignal?.reason; | |
| passthroughPump.startDrain(reason); | |
| passthroughPump.cancelSource(reason); | |
| } | |
| onClientCancel: (reason) => { | |
| passthroughPump.startDrain(reason); | |
| }, | |
| }); | |
| const cleanupPassthroughClientAbortListener = bindClientAbortListener( | |
| session.clientAbortSignal, | |
| () => { | |
| const reason = session.clientAbortSignal?.reason; | |
| passthroughPump.startDrain(reason); | |
| } |
🤖 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 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.
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
+101
to
+119
| test.each([ | ||
| [400, "上游请求参数无效，请检查后重试"], | ||
| [401, "上游鉴权失败，请稍后重试"], | ||
| [429, "上游服务当前限流，请稍后重试"], | ||
| [503, "上游服务暂时不可用，请稍后重试"], | ||
| ])("maps status %i to a generic message when pass-through is disabled", (status, expected) => { | ||
| const error = new ProxyError("sensitive upstream failure", status, { | ||
| body: "sensitive upstream failure", | ||
| }); | ||
|
|
||
| const message = resolveFinalClientErrorMessage({ | ||
| error, | ||
| currentFallbackMessage: "sensitive upstream failure Upstream detail: relay-a", | ||
| settings: { passThroughUpstreamErrorMessage: false }, | ||
| override: null, | ||
| }); | ||
|
|
||
| expect(message).toBe(expected); | ||
| }); |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | 🏗️ Heavy lift
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== File outline ==\n'
ast-grep outline tests/unit/proxy/error-handler-client-message.test.ts --view expanded || true
printf '\n== Relevant snippets ==\n'
sed -n '1,220p' tests/unit/proxy/error-handler-client-message.test.ts | cat -n
printf '\n== Search for message resolver and i18n usage ==\n'
rg -n "resolveFinalClientErrorMessage|passThroughUpstreamErrorMessage|next-intl|useTranslations|getTranslations|locale" src tests -g '!**/dist/**' -g '!**/build/**' || true
printf '\n== Files that look relevant ==\n'
fd -a "error-handler-client-message" src tests || trueRepository: ding113/claude-code-hub
Length of output: 50380
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== test file focused slice ==\n'
sed -n '1,220p' tests/unit/proxy/error-handler-client-message.test.ts | cat -n
printf '\n== resolver implementation locations ==\n'
rg -n "function resolveFinalClientErrorMessage|const resolveFinalClientErrorMessage|resolveFinalClientErrorMessage\\(" src tests -g '!**/dist/**' -g '!**/build/**' | head -n 50
printf '\n== proxy error message implementation slice ==\n'
FILE="$(rg -l "resolveFinalClientErrorMessage" src tests -g '!**/dist/**' -g '!**/build/**' | head -n 1)"
if [ -n "$FILE" ]; then
sed -n '1,260p' "$FILE" | cat -n
fiRepository: ding113/claude-code-hub
Length of output: 16844
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== error-handler implementation slice ==\n'
sed -n '1,220p' src/app/v1/_lib/proxy/error-handler.ts | cat -n
printf '\n== search for related status messages and i18n keys ==\n'
rg -n "上游请求参数无效|上游鉴权失败|上游服务当前限流|上游服务暂时不可用|getTranslations|next-intl|resolveFinalClientErrorMessage" src messages tests -g '!**/dist/**' -g '!**/build/**' | head -n 200
printf '\n== locale message files around proxy/error related keys ==\n'
fd -a "messages" . || trueRepository: ding113/claude-code-hub
Length of output: 32942
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== locale usage in proxy error handling ==\n'
rg -n "getLocale|locale" src/app/v1/_lib/proxy/error-handler.ts src/app/v1/_lib/proxy/*.ts -g '!**/dist/**' -g '!**/build/**' || true
printf '\n== related client error message helper ==\n'
sed -n '1,220p' src/app/v1/_lib/proxy/client-error-message.ts | cat -n
printf '\n== proxy error-related message keys in messages ==\n'
rg -n "upstream|error|proxy|limit|鉴权|限流|不可用|参数无效" messages src/app/v1 -g '!**/dist/**' -g '!**/build/**' | head -n 200Repository: ding113/claude-code-hub
Length of output: 31888
不要把通用上游错误写死为简体中文。 src/app/v1/_lib/proxy/error-handler.ts:78-107 里的状态码兜底文案目前直接返回 zh-CN 字符串，tests/unit/proxy/error-handler-client-message.test.ts:101-136 也把这些文案锁死了；这里需要把 locale 传入错误解析并通过 next-intl 取翻译，再把测试改成按 5 个支持语言参数化。
🤖 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/error-handler-client-message.test.ts` around lines 101 -
119, 更新 error-handler.ts 中的错误解析流程，为 resolveFinalClientErrorMessage 传入 locale，并使用
next-intl 获取各状态码的通用上游错误翻译，避免写死简体中文。同步修改
error-handler-client-message.test.ts，将状态码断言按 5 个支持的语言参数化，并验证每种 locale 返回对应翻译。
Source: Coding guidelines
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.
Comment on lines
+180
to
+196
| it("writes timeout fallback details only while the request is unfinalized", async () => { | ||
| mockWriterDbUpdateReturning.mockResolvedValueOnce([{ id: 606 }]); | ||
| const { updateMessageRequestDetailsIfUnfinalized } = await import("@/repository/message"); | ||
|
|
||
| await updateMessageRequestDetailsIfUnfinalized(606, { | ||
| statusCode: 500, | ||
| errorMessage: "Error: stream_finalization_timeout", | ||
| }); | ||
|
|
||
| expect(mockGetMessageWriterDb).toHaveBeenCalledTimes(1); | ||
| expect(mockWriterDbUpdateWhere).toHaveBeenCalledTimes(1); | ||
| expect(mockWriterDbUpdateReturning).toHaveBeenCalledWith({ id: "id" }); | ||
| expect(mockDbUpdateSet).not.toHaveBeenCalled(); | ||
| const whereSql = sqlToString(mockWriterDbUpdateWhere.mock.calls[0]?.[0]).toLowerCase(); | ||
| expect(whereSql).toContain("statuscode"); | ||
| expect(whereSql).toContain("is null"); | ||
| }); |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
请验证终态 CAS 的 WHERE 条件同时包含目标消息 ID。
这些用例只检查 status_code IS NULL，或直接模拟空 RETURNING。如果 ID 谓词意外丢失，测试仍会通过，而 fallback 更新可能影响全部未终态请求。
tests/unit/repository/message-public-status-rollup.test.ts#L180-L196: 将条件渲染为 SQL 与参数，并断言包含 ID606。tests/unit/repository/message-public-status-rollup.test.ts#L198-L217: 同时断言异步路径包含 ID608和终态空值条件。tests/unit/repository/message-public-status-rollup.test.ts#L219-L232: 在模拟 CAS 失败前断言条件包含 ID607和status_code IS NULL。
📍 Affects 1 file
tests/unit/repository/message-public-status-rollup.test.ts#L180-L196(this comment)tests/unit/repository/message-public-status-rollup.test.ts#L198-L217tests/unit/repository/message-public-status-rollup.test.ts#L219-L232
🤖 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/repository/message-public-status-rollup.test.ts` around lines 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.
Comment on lines
+369
to
+390
| it("sync durable details remain direct and queue rollup after the DB write", async () => { | ||
| mockGetEnvConfig.mockReturnValue({ MESSAGE_REQUEST_WRITE_MODE: "sync" }); | ||
| mockDbSelectLimit.mockResolvedValueOnce([ | ||
| { | ||
| createdAt: new Date("2026-04-21T10:04:00.000Z"), | ||
| model: "gpt-4.1", | ||
| originalModel: "gpt-4.1", | ||
| durationMs: 100, | ||
| }, | ||
| ]); | ||
| const { updateMessageRequestDetailsDurably } = await import("@/repository/message"); | ||
|
|
||
| await updateMessageRequestDetailsDurably(809, { | ||
| statusCode: 200, | ||
| providerChain: [{ id: 1, name: "provider-a", groupTag: "openai" }], | ||
| }); | ||
| await flushMicrotasks(); | ||
|
|
||
| expect(mockDbUpdateSet).toHaveBeenCalledTimes(1); | ||
| expect(mockGetMessageWriterDb).not.toHaveBeenCalled(); | ||
| expect(mockEnqueueMessageRequestUpdateDurably).not.toHaveBeenCalled(); | ||
| expect(mockQueuePublicStatusRollupWrite).toHaveBeenCalledTimes(1); |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
请用延迟 DB Promise 验证 rollup 确实发生在写入完成之后。
当前用例只验证 DB 更新和 rollup 都被调用；若实现先发布 rollup、后等待数据库，测试仍会通过。请挂起 mockDbUpdateWhere，先断言未入队，再释放写入并断言入队。
🤖 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/repository/message-public-status-rollup.test.ts` around lines 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.
![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.
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 (
perf: optimize full-path concurrency and resource ownership) is a large-scale performance and reliability overhaul across 112 files (+18,038/−2,100 lines). It replaces eagerResponse.clone()/tee()draining with a demand-driven response pump, introduces a three-lane PostgreSQL pool with admission control, adds an O(log n) eviction heap for the write buffer, implements atomic Lua lease settlement, batches Redis cost pipelines, adds WebSocket outbound buffering, and hardens shutdown ordering.demand-driven-response-pump.ts): newReadableStream(HWM=0) with state machine (client-active → draining → finalizing → closed), 60s pending-chunk deadline, and cleanwasClientAbortedsignal for circuit-breaker logic.admitted-client.ts,db.ts):createAdmittedSqlClientProxy wrapsunsafe/beginwith an outstanding-query counter; three named pools (data/control/writer) with budget splitsplitPoolBudget(total).lease-service.ts): single Lua script covers all 12 lease dimensions with idempotency marker TTL=5 min;shutdownAlland DB close are now critical barriers with no timeout wrapping.Confidence Score: 4/5
Safe to merge with minor follow-up items; no data-loss or security-critical bugs found in the diff.
The PR is a well-structured performance overhaul. All five findings are P2 (non-blocking). The most significant concern is the tag-template-literal bypass of admission control, but it affects back-pressure rather than correctness. Shutdown-delay risk from ignored AbortSignal and silent cost-tracking drift are also P2. No P0/P1 issues were identified.
src/drizzle/admitted-client.ts (apply-trap gap), src/app/v1/_lib/proxy/response-handler.ts (AbortSignal threading), src/lib/rate-limit/service.ts (pipeline error propagation)
Important Files Changed
Sequence Diagram
%%{init: {'theme': 'neutral'}}%% sequenceDiagram participant C as Client participant RH as ResponseHandler participant Pump as DemandDrivenPump participant UP as UpstreamProvider participant WB as WriteBuffer participant DB as DB (writer lane) C->>RH: POST /v1/... RH->>Pump: createDemandDrivenResponsePump() RH->>UP: fetch(upstream) UP-->>Pump: ReadableStream chunks Pump-->>C: "demand-driven stream (HWM=0)" C->>Pump: pull() [demand signal] Pump->>UP: read chunk UP-->>Pump: chunk Pump-->>C: enqueue chunk Note over Pump: armPendingChunkDeadline (60s) alt Client cancels C->>Pump: cancel() Pump->>Pump: "startDrain() clientAborted=true" else 60s deadline Pump->>Pump: "startDrain() clientAborted=true" Pump->>UP: cancelSource() end Pump->>RH: completion (wasClientAborted) RH->>WB: updateMessageRequestDetailsDurably() WB->>DB: INSERT/UPDATE (writer pool) DB-->>WB: commit ack WB-->>RH: onCommittedCallback RH->>RH: schedulePostTerminalSideEffects()%%{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 C as Client participant RH as ResponseHandler participant Pump as DemandDrivenPump participant UP as UpstreamProvider participant WB as WriteBuffer participant DB as DB (writer lane) C->>RH: POST /v1/... RH->>Pump: createDemandDrivenResponsePump() RH->>UP: fetch(upstream) UP-->>Pump: ReadableStream chunks Pump-->>C: "demand-driven stream (HWM=0)" C->>Pump: pull() [demand signal] Pump->>UP: read chunk UP-->>Pump: chunk Pump-->>C: enqueue chunk Note over Pump: armPendingChunkDeadline (60s) alt Client cancels C->>Pump: cancel() Pump->>Pump: "startDrain() clientAborted=true" else 60s deadline Pump->>Pump: "startDrain() clientAborted=true" Pump->>UP: cancelSource() end Pump->>RH: completion (wasClientAborted) RH->>WB: updateMessageRequestDetailsDurably() WB->>DB: INSERT/UPDATE (writer pool) DB-->>WB: commit ack WB-->>RH: onCommittedCallback RH->>RH: schedulePostTerminalSideEffects()Prompt To Fix All With AI
Reviews (1): Last reviewed commit: "chore: format code (perf-full-path-bench..." | Re-trigger Greptile