feat(availability): 可用性监控 outbox + 1 分钟投影桶 - #1416
Conversation
Replace on-the-fly message_request scans with trigger/outbox-fed avail_bucket_1m/avail_current so availability APIs stay fast under load.
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (1)
🚧 Files skipped from review as they are similar to previous changes (1)
📝 WalkthroughWalkthrough新增可用性投影数据库结构、outbox 触发器和后台 worker。可用性查询改为读取分钟投影桶及当前状态表。应用启动和关闭流程接入 worker，并更新相关单元测试。 Changes可用性投影数据库与事件管线
Estimated code review effort: 4 (Complex) | ~60 minutes Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 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 |
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
| const currentRows = await db.execute(sql` | ||
| SELECT | ||
| ${messageRequest.providerId} AS "providerId", | ||
| COUNT(*) FILTER (WHERE ${buildAvailabilitySuccessOutcomeCondition( | ||
| buildRequestOutcomeSql( | ||
| messageRequest.blockedBy, | ||
| messageRequest.statusCode, | ||
| messageRequest.errorMessage, | ||
| messageRequest.providerChain | ||
| ) | ||
| )})::int AS "greenCount", | ||
| COUNT(*) FILTER (WHERE ${buildAvailabilityFailureOutcomeCondition( | ||
| buildRequestOutcomeSql( | ||
| messageRequest.blockedBy, | ||
| messageRequest.statusCode, | ||
| messageRequest.errorMessage, | ||
| messageRequest.providerChain | ||
| ) | ||
| )})::int AS "redCount", | ||
| MAX(${messageRequest.createdAt}) AS "lastRequestAt" | ||
| FROM ${messageRequest} | ||
| WHERE ${requestConditions} | ||
| GROUP BY ${messageRequest.providerId} | ||
| `; | ||
|
|
||
| const aggregateRows = Array.from( | ||
| await db.execute(aggregateQuery) | ||
| ) as AggregatedCurrentProviderStatusRow[]; | ||
| const providerStats = new Map< | ||
| number, | ||
| { | ||
| greenCount: number; | ||
| redCount: number; | ||
| lastRequestAt: string | null; | ||
| } | ||
| >(); | ||
|
|
||
| for (const provider of providerList) { | ||
| providerStats.set(provider.id, { | ||
| greenCount: 0, | ||
| redCount: 0, | ||
| lastRequestAt: null, | ||
| }); | ||
| c.provider_id AS "providerId", | ||
| c.state AS "state", | ||
| c.availability AS "availability", | ||
| c.request_count AS "requestCount", | ||
| c.last_request_at AS "lastRequestAt" | ||
| FROM avail_current c | ||
| WHERE c.provider_id IN (${sql.join( | ||
| providerList.map((p) => sql`${p.id}`), | ||
| sql`, ` | ||
| )}) | ||
| `); | ||
|
|
||
| const byId = new Map<number, CurrentRow>(); | ||
| for (const row of Array.from(currentRows as Iterable<CurrentRow>)) { | ||
| byId.set(Number(row.providerId), row); |
There was a problem hiding this comment.
When an enabled provider stops receiving requests, its existing avail_current row is accepted without a freshness check and is never recomputed, causing /api/availability/current to keep returning the old green or red status after the rolling 15-minute window should report unknown.
Knowledge Base Used: Provider Management
Prompt To Fix With AI
This is a comment left during a code review.
Path: src/lib/availability/availability-service.ts
Line: 487-503
Comment:
**Current status never expires**
When an enabled provider stops receiving requests, its existing `avail_current` row is accepted without a freshness check and is never recomputed, causing `/api/availability/current` to keep returning the old green or red status after the rolling 15-minute window should report unknown.
**Knowledge Base Used:** [Provider Management](https://app.greptile.com/ygxz/-/custom-context/knowledge-base/ding113/claude-code-hub/-/docs/provider-management.md)
---
For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.| AND bucket_start >= CAST(${startDate.toISOString()} AS timestamptz) | ||
| AND bucket_start <= CAST(${endDate.toISOString()} AS timestamptz) |
There was a problem hiding this comment.
When startTime or endTime is not minute-aligned, filtering only the truncated bucket_start omits the first partial minute and includes the entire final minute, causing counts and latency metrics to cover requests outside the exact requested range while dropping valid in-range requests.
Knowledge Base Used: Provider Management
Prompt To Fix With AI
This is a comment left during a code review.
Path: src/lib/availability/availability-service.ts
Line: 314-315
Comment:
**Partial-minute boundaries return wrong data**
When `startTime` or `endTime` is not minute-aligned, filtering only the truncated `bucket_start` omits the first partial minute and includes the entire final minute, causing counts and latency metrics to cover requests outside the exact requested range while dropping valid in-range requests.
**Knowledge Base Used:** [Provider Management](https://app.greptile.com/ygxz/-/custom-context/knowledge-base/ding113/claude-code-hub/-/docs/provider-management.md)
---
For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.| CASE | ||
| WHEN SUM(latency_cnt) > 0 THEN (SUM(latency_sum_ms)::double precision / SUM(latency_cnt)) | ||
| ELSE 0 | ||
| END AS "p50LatencyMs", | ||
| CASE | ||
| WHEN SUM(latency_cnt) > 0 THEN (SUM(latency_sum_ms)::double precision / SUM(latency_cnt)) | ||
| ELSE 0 | ||
| END AS "p95LatencyMs", | ||
| CASE | ||
| WHEN SUM(latency_cnt) > 0 THEN (SUM(latency_sum_ms)::double precision / SUM(latency_cnt)) | ||
| ELSE 0 | ||
| END AS "p99LatencyMs", |
There was a problem hiding this comment.
When a bucket contains requests with different latencies, p50LatencyMs, p95LatencyMs, and p99LatencyMs all use latency_sum_ms / latency_cnt, causing every percentile to equal the mean and under-reporting tail latency despite the unchanged response contract.
Knowledge Base Used: Provider Management
Prompt To Fix With AI
This is a comment left during a code review.
Path: src/lib/availability/availability-service.ts
Line: 296-307
Comment:
**Percentiles replaced with averages**
When a bucket contains requests with different latencies, `p50LatencyMs`, `p95LatencyMs`, and `p99LatencyMs` all use `latency_sum_ms / latency_cnt`, causing every percentile to equal the mean and under-reporting tail latency despite the unchanged response contract.
**Knowledge Base Used:** [Provider Management](https://app.greptile.com/ygxz/-/custom-context/knowledge-base/ding113/claude-code-hub/-/docs/provider-management.md)
---
For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.![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: e76345619b
ℹ️ 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".
| c.availability AS "availability", | ||
| c.request_count AS "requestCount", | ||
| c.last_request_at AS "lastRequestAt" | ||
| FROM avail_current c |
There was a problem hiding this comment.
When a provider has no new requests after its last projected event, avail_current is never recomputed or expired; this query accepts the stored row unconditionally, so /api/availability/current can keep returning green/red and a nonzero requestCount for hours, whereas the previous implementation recomputed a 15-minute window and would return unknown. Add a last_request_at/updated_at freshness check or recompute stale providers from avail_bucket_1m.
Useful? React with 👍 / 👎.
| FROM message_request mr | ||
| LEFT JOIN providers p ON p.id = mr.provider_id | ||
| WHERE mr.status_code IS NOT NULL | ||
| AND mr.created_at >= now() - interval '48 hours' |
There was a problem hiding this comment.
On the first deployment of this projection, existing requests older than 48 hours are never projected because the read path no longer scans message_request and the one-shot backfill_done guard prevents later catch-up. The dashboard/API still supports 7-day and 100-day ranges, so those ranges silently undercount after upgrade until enough new projected data accumulates; backfill the full supported range or fall back to raw rows for unprojected history.
Useful? React with 👍 / 👎.
| -- Availability projection: outbox + 1-minute buckets (read path no longer scans message_request) | ||
| CREATE EXTENSION IF NOT EXISTS pgcrypto;--> statement-breakpoint | ||
|
|
||
| CREATE TABLE IF NOT EXISTS "outbox_events" ( |
There was a problem hiding this comment.
This migration creates new tables, but they are not added to the schema used by drizzle.config.ts (src/drizzle/schema.ts), and no drizzle/meta/0120_snapshot.json was added; the next bun run db:generate will be based on a schema history that does not know about these projection tables and can drift or try to undo them. Please add the tables to src/drizzle/schema.ts and regenerate the migration.
AGENTS.md reference: AGENTS.md:L55-L58
Useful? React with 👍 / 👎.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 8
🧹 Nitpick comments (11)
drizzle/0120_availability_projection.sql (4)
128-133: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value触发器名称后缀
aiud与实际事件不匹配。后缀
aiud通常表示 after insert/update/delete。该触发器只注册了 INSERT 与 UPDATE，没有 DELETE。建议改名为message_request_outbox_aiu，避免误导后续维护者。🤖 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 `@drizzle/0120_availability_projection.sql` around lines 128 - 133, 将触发器名称从 message_request_outbox_aiud 统一改为 message_request_outbox_aiu，包括 DROP TRIGGER 和 CREATE TRIGGER 语句；保持现有 INSERT 与 UPDATE 事件配置不变。
23-26: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
outbox_processed表未被任何代码使用。
src/lib/availability/projection-worker.ts的幂等控制依赖outbox_events.published_at与proj_applied_requests，没有读写outbox_processed。该表创建后即为死结构，会造成 schema 噪音与后续维护困惑。建议删除该表定义，或补充使用它的代码。
♻️ 建议删除未使用的表
-CREATE TABLE IF NOT EXISTS "outbox_processed" ( - "event_id" uuid PRIMARY KEY, - "processed_at" timestamp with time zone DEFAULT now() NOT NULL -);--> statement-breakpoint -🤖 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 `@drizzle/0120_availability_projection.sql` around lines 23 - 26, 删除迁移文件中的 outbox_processed 表定义，保留现有 outbox_events.published_at 与 proj_applied_requests 的幂等控制逻辑不变；不要为该未使用表新增读写代码。
4-32: 🚀 Performance & Scalability | 🔵 Trivial | 🏗️ Heavy lift
outbox_events、proj_applied_requests与avail_bucket_1m缺少保留策略。三张表都只写不删。
outbox_events每条终态请求写入一行且永不清理，proj_applied_requests每个 request_id 保留一行，avail_bucket_1m按 provider 每分钟一行且无上限。在高流量部署下，这些表会持续增长，磁盘占用与 autovacuum 压力会上升。建议加入定期清理任务：删除
published_at早于 N 天的 outbox 行，删除超出可用性查询窗口（例如 30 天）的avail_bucket_1m行与对应的proj_applied_requests行。仓库中已有@/lib/log-cleanup/cleanup-queue，可复用同一调度机制。🤖 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 `@drizzle/0120_availability_projection.sql` around lines 4 - 32, 为 outbox_events、proj_applied_requests 和 avail_bucket_1m 增加定期保留清理任务：删除 published_at 早于配置保留天数的 outbox_events，并删除超出可用性查询窗口（默认例如 30 天）的 avail_bucket_1m 及对应 proj_applied_requests 记录。复用现有 `@/lib/log-cleanup/cleanup-queue` 调度机制，确保清理可重复执行且不影响保留窗口内的数据。
102-122: 🚀 Performance & Scalability | 🔵 Trivial | 💤 Low value移除未消费的
group_tag字段如果没有仓库外消费者依赖
group_tag，请从触发器和 backfill 的request_finalizedpayload 中移除该字段。当前projection-worker不读取它；虽然providers.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 `@drizzle/0120_availability_projection.sql` around lines 102 - 122, 移除 request_finalized payload 构造中的 group_tag 字段及其 providers 查询，并同步从相关 backfill 逻辑中删除该字段，确保触发器和 backfill 生成的 payload 保持一致。tests/unit/lib/availability-service.test.ts (2)
42-49: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win建议在断言辅助函数中固化百分位字段的当前语义。
expectProjectionBucketReadPath已断言不再出现percentile_cont。但没有任何用例验证p50LatencyMs、p95LatencyMs、p99LatencyMs现在等于avgLatencyMs。加入一条断言可以把这个有意的近似行为固定下来，后续实现真实百分位时测试会明确失败，提示同步更新契约。🤖 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/availability-service.test.ts` around lines 42 - 49, Update the expectProjectionBucketReadPath test helper to assert that p50LatencyMs, p95LatencyMs, and p99LatencyMs currently use the same value as avgLatencyMs. Add coverage in the relevant projection result assertions so a future switch to true percentile calculations fails until the expected contract is updated.
621-668: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win建议补充
avail_current存在但数据陈旧或计数为 0 的用例。当前用例覆盖了"
avail_current为空则回退"的路径。未覆盖的分支同样重要：
avail_current返回requestCount: 0的行。此时第 547 行的守卫生效，返回unknown，且不会触发桶回退。avail_current返回state: "yellow"的行，验证归一化为green或red。第二条尤其值得覆盖，因为
src/lib/availability/projection-worker.ts会写入yellow，而AvailabilityStatus类型不包含该值。🤖 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/availability-service.test.ts` around lines 621 - 668, 在 getCurrentProviderStatus 的测试中补充 avail_current 返回 requestCount 为 0 的用例，断言状态为 unknown 且不会触发 avail_bucket_1m 回退；另补充 state 为 yellow 的用例，验证其按现有归一化规则转换为 green 或 red，并覆盖 projection-worker 写入该状态的兼容行为。src/lib/availability/projection-tables.ts (1)
37-50: 🗄️ Data Integrity & Integration | 🔵 Trivial | 💤 Low value
availBucket1m未声明迁移中的idx_avail_bucket_1m_time索引。SQL 第 46-47 行创建了
bucket_start DESC索引，Drizzle 表定义中没有对应的index()声明。建议补充，保持 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/lib/availability/projection-tables.ts` around lines 37 - 50, 在 availBucket1m 的 Drizzle 表定义中补充与迁移对应的 idx_avail_bucket_1m_time 索引声明，使用 bucketStart 的降序排序，并保留现有主键定义不变。src/lib/availability/availability-service.ts (2)
558-566: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
yellow分支与else分支逻辑完全相同。第 562-563 行与第 564-565 行执行同一表达式。可以合并，减少重复。
♻️ 建议合并分支
const rawState = String(stats.state || "unknown"); - let status: AvailabilityStatus = "unknown"; - if (rawState === "green" || rawState === "red" || rawState === "unknown") { - status = rawState; - } else if (rawState === "yellow") { - status = toFiniteNumber(stats.availability) >= 0.5 ? "green" : "red"; - } else { - status = toFiniteNumber(stats.availability) >= 0.5 ? "green" : "red"; - } + // worker 会写入 'yellow'，读路径按可用率归一到 green/red。 + const status: AvailabilityStatus = + rawState === "green" || rawState === "red" || rawState === "unknown" + ? rawState + : toFiniteNumber(stats.availability) >= 0.5 + ? "green" + : "red";🤖 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/availability/availability-service.ts` around lines 558 - 566, 合并 availability 状态转换逻辑中 `rawState === "yellow"` 与后续 `else` 的重复分支，保留一次基于 `toFiniteNumber(stats.availability) >= 0.5` 返回 `"green"` 或 `"red"` 的判断，同时维持对 `"green"`、`"red"` 和 `"unknown"` 的现有处理。
316-316: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
GROUP BY provider_id, 2使用序数引用，较脆弱。序数
2指向SELECT列表中的date_bin(...)表达式。任何列顺序调整都会静默改变分组语义。建议改为重复写出date_bin(...)表达式，或用GROUP BY 1, 2，让意图一致且明确。🤖 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/availability/availability-service.ts` at line 316, 更新该查询的 GROUP BY 子句，避免仅使用脆弱的序数 2；重复 SELECT 中的 date_bin(...) 表达式，并继续按 provider_id 与该日期桶分组，确保列顺序调整不会改变分组语义。src/lib/availability/projection-worker.ts (2)
261-294: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
let current!的确定赋值断言可以避免。
current先声明再赋值，仅为了让finally中的第 286 行能够比较引用。可以用一个局部 token 对象或把 IIFE 提取为具名函数，去掉非空断言，降低阅读成本。🤖 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/availability/projection-worker.ts` around lines 261 - 294, Refactor runCycle to remove the definite-assignment assertion from current while preserving the finally check that clears s.currentPromise only for the active cycle. Use a local token or named-cycle structure to retain the reference identity without relying on let current!.
11-13: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick win轮询间隔与批量大小建议可配置。
TICK_MS = 200表示每秒发起 5 次 outbox 轮询查询。在空闲实例上这是持续的数据库负载，多实例部署会成倍放大。建议支持环境变量覆盖，并在连续空批后使用退避（例如从 200ms 逐步退到 2s，有事件时立即恢复）。
🤖 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/availability/projection-worker.ts` around lines 11 - 13, 使 projection worker 的轮询间隔和批量大小支持环境变量覆盖，更新 BATCH 与 TICK_MS 的配置读取并保留现有默认值。调整轮询逻辑，使连续返回空批次时逐步增加间隔，最高退避至约 2 秒；检测到事件后立即恢复默认轮询间隔。
🤖 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 `@drizzle/0120_availability_projection.sql`:
- Around line 87-100: 在触发器中调用 fn_compute_message_request_success_rate_outcome 的
EXCEPTION WHEN OTHERS 分支加入 RAISE WARNING，记录异常详情及相关请求上下文；保留 v_outcome := NULL 和后续
RETURN NEW 的现有流程。
In `@src/lib/availability/availability-service.ts`:
- Around line 296-307: Update the latency metrics projection around the
p50LatencyMs, p95LatencyMs, and p99LatencyMs aliases so their average-based
calculation is not exposed as true percentiles. Choose one explicit contract:
mark these fields as approximate in the API response or TimeBucketMetrics type,
remove them until histogram or t-digest data is available, or add
histogram-backed percentile calculation to avail_bucket_1m; do not retain
unchanged percentile semantics for SUM(latency_sum_ms) / SUM(latency_cnt).
In `@src/lib/availability/projection-tables.ts`:
- Around line 17-29: 使 Drizzle 表定义与迁移 SQL 保持一致：在
src/lib/availability/projection-tables.ts:17-29 的 outboxEvents 中为 eventId
增加唯一约束和随机默认值，并声明部分索引 idx_outbox_events_unpublished；在
drizzle/0120_availability_projection.sql:5 将 id 的 bigserial 改为与
generatedByDefaultAsIdentity() 对应的 bigint identity 定义；在
src/lib/availability/projection-tables.ts:37-50 的 availBucket1m 中声明迁移已有的
bucket_start DESC 索引 idx_avail_bucket_1m_time。
In `@src/lib/availability/projection-worker.ts`:
- Around line 222-255: Update the projection loop around the worker’s
avail_current upsert to periodically recompute every provider with an existing
avail_current row, not only touchedProviders, and extract the 5-minute freshness
window into a shared constant. In src/lib/availability/projection-worker.ts
lines 222-255, use that constant for recalculation; in
src/lib/availability/availability-service.ts lines 506-521, select
avail_current.updated_at, treat rows older than the shared window as missing,
and use the same constant for avail_bucket_1m fallback.
- Around line 134-220: 重构处理 claimed 的逐条数据库操作，改为集合式批处理以避免每条事件串行执行多个
tx.execute。先批量校验并使用 unnest 一次写入 proj_applied_requests，保留 request_id
冲突去重并获取新插入记录；再按 provider_id 与 bucket_start 预聚合成功、失败、排除及延迟指标，批量执行 avail_bucket_1m
的插入或冲突更新；最后使用事件 ID 集合一次性更新 outbox_events，并继续正确维护 touchedProviders 与 applied。
- Around line 45-98: Update bootstrapBackfill to run under the repository’s
withAdvisoryLock, covering both the backfill_done check and completion marker so
only one instance can enqueue data. Replace the single 48-hour INSERT ... SELECT
with bounded time-slice or ID-range batches, committing each batch independently
while preserving the existing filters, payload, and projection_meta completion
behavior.
- Around line 303-332: Update the startup IIFE around bootstrapBackfill so its
Promise is stored in SchedulerState, using a dedicated backfill-task field that
is cleared after completion. In stopAvailabilityProjectionWorker, await that
tracked backfill Promise in addition to s.currentPromise before marking the
worker stopped.
In `@src/lib/lifecycle/shutdown.ts`:
- Around line 141-152: 将 shutdown 中调用 stopAvailabilityProjectionWorker
的步骤改为关键屏障，确保其 currentPromise 即使超过 stepMs 也必须完成后才继续执行
closeDbPools；可保留超时计时与告警，但不能因超时而放行 pending 的停止操作。参照 message writer 的等待方式，并增加延迟停止
mock，验证数据库关闭不会早于 worker 停止完成。
---
Nitpick comments:
In `@drizzle/0120_availability_projection.sql`:
- Around line 128-133: 将触发器名称从 message_request_outbox_aiud 统一改为
message_request_outbox_aiu，包括 DROP TRIGGER 和 CREATE TRIGGER 语句；保持现有 INSERT 与
UPDATE 事件配置不变。
- Around line 23-26: 删除迁移文件中的 outbox_processed 表定义，保留现有
outbox_events.published_at 与 proj_applied_requests 的幂等控制逻辑不变；不要为该未使用表新增读写代码。
- Around line 4-32: 为 outbox_events、proj_applied_requests 和 avail_bucket_1m
增加定期保留清理任务：删除 published_at 早于配置保留天数的 outbox_events，并删除超出可用性查询窗口（默认例如 30 天）的
avail_bucket_1m 及对应 proj_applied_requests 记录。复用现有
`@/lib/log-cleanup/cleanup-queue` 调度机制，确保清理可重复执行且不影响保留窗口内的数据。
- Around line 102-122: 移除 request_finalized payload 构造中的 group_tag 字段及其
providers 查询，并同步从相关 backfill 逻辑中删除该字段，确保触发器和 backfill 生成的 payload 保持一致。
In `@src/lib/availability/availability-service.ts`:
- Around line 558-566: 合并 availability 状态转换逻辑中 `rawState === "yellow"` 与后续
`else` 的重复分支，保留一次基于 `toFiniteNumber(stats.availability) >= 0.5` 返回 `"green"` 或
`"red"` 的判断，同时维持对 `"green"`、`"red"` 和 `"unknown"` 的现有处理。
- Line 316: 更新该查询的 GROUP BY 子句，避免仅使用脆弱的序数 2；重复 SELECT 中的 date_bin(...) 表达式，并继续按
provider_id 与该日期桶分组，确保列顺序调整不会改变分组语义。
In `@src/lib/availability/projection-tables.ts`:
- Around line 37-50: 在 availBucket1m 的 Drizzle 表定义中补充与迁移对应的
idx_avail_bucket_1m_time 索引声明，使用 bucketStart 的降序排序，并保留现有主键定义不变。
In `@src/lib/availability/projection-worker.ts`:
- Around line 261-294: Refactor runCycle to remove the definite-assignment
assertion from current while preserving the finally check that clears
s.currentPromise only for the active cycle. Use a local token or named-cycle
structure to retain the reference identity without relying on let current!.
- Around line 11-13: 使 projection worker 的轮询间隔和批量大小支持环境变量覆盖，更新 BATCH 与 TICK_MS
的配置读取并保留现有默认值。调整轮询逻辑，使连续返回空批次时逐步增加间隔，最高退避至约 2 秒；检测到事件后立即恢复默认轮询间隔。
In `@tests/unit/lib/availability-service.test.ts`:
- Around line 42-49: Update the expectProjectionBucketReadPath test helper to
assert that p50LatencyMs, p95LatencyMs, and p99LatencyMs currently use the same
value as avgLatencyMs. Add coverage in the relevant projection result assertions
so a future switch to true percentile calculations fails until the expected
contract is updated.
- Around line 621-668: 在 getCurrentProviderStatus 的测试中补充 avail_current 返回
requestCount 为 0 的用例，断言状态为 unknown 且不会触发 avail_bucket_1m 回退；另补充 state 为 yellow
的用例，验证其按现有归一化规则转换为 green 或 red，并覆盖 projection-worker 写入该状态的兼容行为。
🪄 Autofix
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 Plus
Run ID: 5fb90f31-d915-4111-b378-c037f4635cae
📒 Files selected for processing (11)
drizzle/0120_availability_projection.sqldrizzle/meta/_journal.jsonsrc/instrumentation.tssrc/lib/availability/availability-service.tssrc/lib/availability/index.tssrc/lib/availability/projection-tables.tssrc/lib/availability/projection-worker.tssrc/lib/lifecycle/shutdown.tstests/unit/lib/availability-service.test.tstests/unit/lib/shutdown.test.tstests/unit/server-shutdown.test.ts
| CASE | ||
| WHEN SUM(latency_cnt) > 0 THEN (SUM(latency_sum_ms)::double precision / SUM(latency_cnt)) | ||
| ELSE 0 | ||
| END AS "p50LatencyMs", | ||
| CASE | ||
| WHEN SUM(latency_cnt) > 0 THEN (SUM(latency_sum_ms)::double precision / SUM(latency_cnt)) | ||
| ELSE 0 | ||
| END AS "p95LatencyMs", | ||
| CASE | ||
| WHEN SUM(latency_cnt) > 0 THEN (SUM(latency_sum_ms)::double precision / SUM(latency_cnt)) | ||
| ELSE 0 | ||
| END AS "p99LatencyMs", |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
p50LatencyMs、p95LatencyMs、p99LatencyMs 现在返回平均值，但外部类型未变化。
三个字段都被赋为 SUM(latency_sum_ms) / SUM(latency_cnt)，即算术平均。TimeBucketMetrics 的字段名仍然声称是百分位。API 消费者与前端图表会把平均值当作 p95 展示，在长尾延迟场景下会严重低估。
投影桶只保存 latency_cnt 与 latency_sum_ms，确实无法还原真实百分位。请在合并前明确处理方式：
- 在 API 响应或类型定义中标注这些字段当前为近似值；或
- 从响应中移除百分位字段，直到桶中加入 t-digest / 直方图；或
- 在
avail_bucket_1m中增加分桶直方图列以支持近似百分位。
保持字段名不变而悄悄改变语义，会让下游得到错误结论。
🤖 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/availability/availability-service.ts` around lines 296 - 307, Update
the latency metrics projection around the p50LatencyMs, p95LatencyMs, and
p99LatencyMs aliases so their average-based calculation is not exposed as true
percentiles. Choose one explicit contract: mark these fields as approximate in
the API response or TimeBucketMetrics type, remove them until histogram or
t-digest data is available, or add histogram-backed percentile calculation to
avail_bucket_1m; do not retain unchanged percentile semantics for
SUM(latency_sum_ms) / SUM(latency_cnt).
| export const outboxEvents = pgTable("outbox_events", { | ||
| id: bigint("id", { mode: "number" }).primaryKey().generatedByDefaultAsIdentity(), | ||
| eventId: uuid("event_id").notNull(), | ||
| eventType: text("event_type").notNull(), | ||
| aggregateType: text("aggregate_type").notNull(), | ||
| aggregateId: bigint("aggregate_id", { mode: "number" }).notNull(), | ||
| occurredAt: timestamp("occurred_at", { withTimezone: true }).notNull(), | ||
| payload: jsonb("payload").notNull(), | ||
| createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(), | ||
| publishedAt: timestamp("published_at", { withTimezone: true }), | ||
| attempts: integer("attempts").notNull().default(0), | ||
| lastError: text("last_error"), | ||
| }); |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Drizzle 表定义与迁移 SQL 不同步。 同一批投影表在两处独立描述，主键生成方式、唯一约束、默认值与索引均存在差异。若这些表被 drizzle.config 的 schema 输入包含，下一次 drizzle-kit generate 会产出一个试图修正差异的迁移，其中删除序列改为 identity 的操作具有破坏性。请让两侧成为同一份定义。
src/lib/availability/projection-tables.ts#L17-L29：为eventId补上.unique()与.defaultRandom()，并声明部分索引idx_outbox_events_unpublished。drizzle/0120_availability_projection.sql#L5-L5：把id的bigserial改为bigint GENERATED BY DEFAULT AS IDENTITY，与 TS 定义的generatedByDefaultAsIdentity()对齐；或反向把 TS 改为序列写法。src/lib/availability/projection-tables.ts#L37-L50：为availBucket1m声明迁移中已创建的bucket_start DESC索引idx_avail_bucket_1m_time。
📍 Affects 2 files
src/lib/availability/projection-tables.ts#L17-L29(this comment)drizzle/0120_availability_projection.sql#L5-L5src/lib/availability/projection-tables.ts#L37-L50
🤖 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/availability/projection-tables.ts` around lines 17 - 29, 使 Drizzle
表定义与迁移 SQL 保持一致：在 src/lib/availability/projection-tables.ts:17-29 的 outboxEvents
中为 eventId 增加唯一约束和随机默认值，并声明部分索引 idx_outbox_events_unpublished；在
drizzle/0120_availability_projection.sql:5 将 id 的 bigserial 改为与
generatedByDefaultAsIdentity() 对应的 bigint identity 定义；在
src/lib/availability/projection-tables.ts:37-50 的 availBucket1m 中声明迁移已有的
bucket_start DESC 索引 idx_avail_bucket_1m_time。
| for (const row of claimed) { | ||
| const payload = asPayload(row.payload); | ||
| const requestId = Number(payload.request_id); | ||
| const providerId = Number(payload.provider_id); | ||
| const outcome = String(payload.outcome || "excluded"); | ||
| const occurredAt = payload.occurred_at; | ||
| if (!Number.isFinite(requestId) || !Number.isFinite(providerId) || !occurredAt) { | ||
| await tx.execute(sql` | ||
| UPDATE outbox_events | ||
| SET published_at = now(), | ||
| attempts = attempts + 1, | ||
| last_error = 'invalid payload' | ||
| WHERE id = ${row.id} | ||
| `); | ||
| continue; | ||
| } | ||
|
|
||
| const inserted = await tx.execute(sql` | ||
| INSERT INTO proj_applied_requests (request_id, event_id) | ||
| VALUES (${requestId}, ${row.event_id}::uuid) | ||
| ON CONFLICT (request_id) DO NOTHING | ||
| RETURNING request_id | ||
| `); | ||
| const isFresh = Array.from(inserted as Iterable<unknown>).length > 0; | ||
|
|
||
| if (isFresh) { | ||
| const durationMs = | ||
| payload.duration_ms === null || payload.duration_ms === undefined | ||
| ? null | ||
| : Number(payload.duration_ms); | ||
| const successCnt = outcome === "success" ? 1 : 0; | ||
| const failureCnt = outcome === "failure" ? 1 : 0; | ||
| const excludedCnt = outcome === "excluded" ? 1 : 0; | ||
| const latencyCnt = | ||
| (outcome === "success" || outcome === "failure") && | ||
| durationMs !== null && | ||
| Number.isFinite(durationMs) | ||
| ? 1 | ||
| : 0; | ||
| const latencySum = | ||
| latencyCnt === 1 && durationMs !== null && Number.isFinite(durationMs) | ||
| ? Math.trunc(durationMs) | ||
| : 0; | ||
|
|
||
| await tx.execute(sql` | ||
| INSERT INTO avail_bucket_1m AS b ( | ||
| provider_id, | ||
| bucket_start, | ||
| success_cnt, | ||
| failure_cnt, | ||
| excluded_cnt, | ||
| latency_cnt, | ||
| latency_sum_ms, | ||
| last_request_at | ||
| ) VALUES ( | ||
| ${providerId}, | ||
| date_trunc('minute', ${occurredAt}::timestamptz), | ||
| ${successCnt}, | ||
| ${failureCnt}, | ||
| ${excludedCnt}, | ||
| ${latencyCnt}, | ||
| ${latencySum}, | ||
| ${occurredAt}::timestamptz | ||
| ) | ||
| ON CONFLICT (provider_id, bucket_start) DO UPDATE SET | ||
| success_cnt = b.success_cnt + EXCLUDED.success_cnt, | ||
| failure_cnt = b.failure_cnt + EXCLUDED.failure_cnt, | ||
| excluded_cnt = b.excluded_cnt + EXCLUDED.excluded_cnt, | ||
| latency_cnt = b.latency_cnt + EXCLUDED.latency_cnt, | ||
| latency_sum_ms = b.latency_sum_ms + EXCLUDED.latency_sum_ms, | ||
| last_request_at = GREATEST( | ||
| COALESCE(b.last_request_at, EXCLUDED.last_request_at), | ||
| EXCLUDED.last_request_at | ||
| ) | ||
| `); | ||
| touchedProviders.add(providerId); | ||
| applied += 1; | ||
| } | ||
|
|
||
| await tx.execute(sql` | ||
| UPDATE outbox_events | ||
| SET published_at = now(), | ||
| attempts = attempts + 1, | ||
| last_error = NULL | ||
| WHERE id = ${row.id} | ||
| `); | ||
| } |
There was a problem hiding this comment.
🚀 Performance & Scalability | 🟠 Major | 🏗️ Heavy lift
批处理循环对每条事件发起 2 到 3 次串行数据库往返。
BATCH 为 300。每条事件依次执行 INSERT proj_applied_requests、INSERT avail_bucket_1m 与 UPDATE outbox_events，全部串行 await。单批最多约 900 次往返，全部在一个事务内持有行锁。
建议改为集合式处理：用一条 INSERT ... SELECT ... FROM unnest(...) 批量写入 proj_applied_requests 并 RETURNING 出新鲜的 request_id，再用一条按 (provider_id, bucket_start) 预聚合的 INSERT ... ON CONFLICT 更新桶，最后用一条 UPDATE ... WHERE id = ANY(...) 标记 outbox。往返次数可降到常数级。
🤖 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/availability/projection-worker.ts` around lines 134 - 220, 重构处理
claimed 的逐条数据库操作，改为集合式批处理以避免每条事件串行执行多个 tx.execute。先批量校验并使用 unnest 一次写入
proj_applied_requests，保留 request_id 冲突去重并获取新插入记录；再按 provider_id 与 bucket_start
预聚合成功、失败、排除及延迟指标，批量执行 avail_bucket_1m 的插入或冲突更新；最后使用事件 ID 集合一次性更新
outbox_events，并继续正确维护 touchedProviders 与 applied。
| for (const providerId of touchedProviders) { | ||
| await tx.execute(sql` | ||
| INSERT INTO avail_current AS c ( | ||
| provider_id, state, availability, request_count, last_request_at, updated_at | ||
| ) | ||
| SELECT | ||
| ${providerId}, | ||
| CASE | ||
| WHEN COALESCE(SUM(b.success_cnt), 0) + COALESCE(SUM(b.failure_cnt), 0) <= 0 THEN 'unknown' | ||
| WHEN (COALESCE(SUM(b.success_cnt), 0)::float | ||
| / (COALESCE(SUM(b.success_cnt), 0) + COALESCE(SUM(b.failure_cnt), 0))) >= 0.8 THEN 'green' | ||
| WHEN (COALESCE(SUM(b.success_cnt), 0)::float | ||
| / (COALESCE(SUM(b.success_cnt), 0) + COALESCE(SUM(b.failure_cnt), 0))) >= 0.5 THEN 'yellow' | ||
| ELSE 'red' | ||
| END, | ||
| CASE | ||
| WHEN COALESCE(SUM(b.success_cnt), 0) + COALESCE(SUM(b.failure_cnt), 0) <= 0 THEN 0 | ||
| ELSE COALESCE(SUM(b.success_cnt), 0)::float | ||
| / (COALESCE(SUM(b.success_cnt), 0) + COALESCE(SUM(b.failure_cnt), 0)) | ||
| END, | ||
| (COALESCE(SUM(b.success_cnt), 0) + COALESCE(SUM(b.failure_cnt), 0))::int, | ||
| MAX(b.last_request_at), | ||
| now() | ||
| FROM avail_bucket_1m b | ||
| WHERE b.provider_id = ${providerId} | ||
| AND b.bucket_start >= now() - interval '5 minutes' | ||
| ON CONFLICT (provider_id) DO UPDATE SET | ||
| state = EXCLUDED.state, | ||
| availability = EXCLUDED.availability, | ||
| request_count = EXCLUDED.request_count, | ||
| last_request_at = EXCLUDED.last_request_at, | ||
| updated_at = now() | ||
| `); | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | 🏗️ Heavy lift
avail_current 缺少过期机制，provider 停止流量后状态长期显示为 green。 根因是 worker 只对本批次被触达的 provider 重算 avail_current，没有任何周期性衰减；读路径又只对完全缺行的 provider 回退到桶数据，因此陈旧行会被原样返回。改造前读路径直接扫描最近 15 分钟的 message_request，停流后会转为 unknown，这是一处行为回归。两处的时间窗口也不一致（5 分钟与 15 分钟）。
src/lib/availability/projection-worker.ts#L222-L255：增加周期性重算，覆盖所有有avail_current行的 provider，而不仅是touchedProviders；并把 5 分钟窗口抽取为与读路径共享的常量。src/lib/availability/availability-service.ts#L506-L521：把avail_current.updated_at纳入查询并加入新鲜度判断，超过窗口的行视为缺失并走avail_bucket_1m回退；回退窗口改用与 worker 相同的常量。
📍 Affects 2 files
src/lib/availability/projection-worker.ts#L222-L255(this comment)src/lib/availability/availability-service.ts#L506-L521
🤖 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/availability/projection-worker.ts` around lines 222 - 255, Update the
projection loop around the worker’s avail_current upsert to periodically
recompute every provider with an existing avail_current row, not only
touchedProviders, and extract the 5-minute freshness window into a shared
constant. In src/lib/availability/projection-worker.ts lines 222-255, use that
constant for recalculation; in src/lib/availability/availability-service.ts
lines 506-521, select avail_current.updated_at, treat rows older than the shared
window as missing, and use the same constant for avail_bucket_1m fallback.
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
| return {}; | ||
| } | ||
|
|
||
| async function processBatch(): Promise<number> { |
There was a problem hiding this comment.
[High] [TEST-MISSING-CRITICAL] projection-worker.ts (341 lines) ships with no unit tests
Why this is a problem: This module is the critical write path for availability monitoring — outbox claiming (FOR UPDATE SKIP LOCKED), proj_applied_requests dedup, the avail_bucket_1m upsert math, the avail_current recompute, the 48h bootstrapBackfill, and asPayload error handling. None of it is exercised by a test. CLAUDE.md states "All new features must have unit test coverage of at least 80%", and every comparable scheduler in this repo is tested (replay-cleanup.test.ts, tests/unit/lib/log-cleanup/, etc.). A regression here — wrong bucket math, a broken ON CONFLICT (request_id) DO NOTHING dedup, an off-by-one in the claim loop — would ship undetected and silently corrupt availability data.
Suggested fix: Add tests/unit/lib/availability/projection-worker.test.ts (mock db.transaction / db.execute) covering at minimum:
processBatch: fresh event increments the right 1m bucket and recomputesavail_currentprocessBatch: duplicateevent_idis skipped viaproj_applied_requestsON CONFLICT and NOT double-countedprocessBatch: invalid payload (missingrequest_id/provider_id/occurred_at) is marked published withlast_error = 'invalid payload'asPayload: object payload passes through; string JSON is parsed; unparseable string yields{}bootstrapBackfill: is a no-op whenprojection_metaalready haskey = 'backfill_done'
| now() | ||
| FROM avail_bucket_1m b | ||
| WHERE b.provider_id = ${providerId} | ||
| AND b.bucket_start >= now() - interval '5 minutes' |
There was a problem hiding this comment.
[Medium] [LOGIC-BUG] avail_current uses a 5-minute window while the read-path fallback uses 15 minutes
Why this is a problem: avail_current.state / request_count are aggregated over now() - interval '5 minutes' here, but when getCurrentProviderStatus() falls back to avail_bucket_1m (availability-service.ts) it aggregates over CURRENT_PROVIDER_STATUS_WINDOW_MINUTES (= 15). The same provider therefore reports status from a different time window depending on whether its avail_current row exists (5m) or is missing (15m fallback). The hard-coded 5 is also detached from any constant, so the two will keep drifting on future edits. This is additionally a silent behavior change: the "current" status window moved from the historical 15 minutes to 5 minutes on the primary path.
Suggested fix: drive both paths from one source of truth. Export the constant and reuse it:
// availability-service.ts
export const CURRENT_PROVIDER_STATUS_WINDOW_MINUTES = 15;
// projection-worker.ts
import { CURRENT_PROVIDER_STATUS_WINDOW_MINUTES } from "./availability-service";
// ...
AND b.bucket_start >= now() - (${CURRENT_PROVIDER_STATUS_WINDOW_MINUTES} * interval '1 minute')If 5 minutes is the intentionally tighter new window, lower CURRENT_PROVIDER_STATUS_WINDOW_MINUTES to 5 so the fallback matches — but make the choice explicit and identical on both sides.
| const stats = providerStats.get(provider.id)!; | ||
| const total = stats.greenCount + stats.redCount; | ||
| const availability = calculateAvailabilityScore(stats.greenCount, stats.redCount); | ||
| const stats = byId.get(provider.id); |
There was a problem hiding this comment.
[Medium] [LOGIC-BUG] Idle providers surface a frozen status instead of "unknown"
Why this is a problem: avail_current is recomputed only for providers in touchedProviders (i.e. when they just produced a new outbox event). Once a provider stops receiving traffic its row is never refreshed, so this read path keeps returning the last-computed state / requestCount indefinitely. The previous implementation scanned a live 15-minute window and returned unknown once the provider went quiet — which the surrounding comments still declare as required behavior ("No data = 'unknown', NOT 'green'! Must be honest"). The requestCount <= 0 guard below does NOT catch this, because the frozen row retains request_count > 0 from its last update.
Suggested fix: treat avail_current rows whose updated_at / last_request_at falls outside the status window as stale (requires also selecting c.updated_at AS "updatedAt"):
const stats = byId.get(provider.id);
const WINDOW_MS = CURRENT_PROVIDER_STATUS_WINDOW_MINUTES * 60 * 1000;
const updatedAtMs = stats?.updatedAt ? new Date(stats.updatedAt).getTime() : 0;
const isStale =
!stats ||
toFiniteNumber(stats.requestCount) <= 0 ||
(updatedAtMs > 0 && Date.now() - updatedAtMs > WINDOW_MS);
if (isStale) {
return {
providerId: provider.id,
providerName: provider.name,
status: "unknown" as AvailabilityStatus,
availability: 0,
requestCount: 0,
lastRequestAt: null,
};
}Alternatively, have the worker periodically recompute (or delete) avail_current rows for providers not seen recently.
There was a problem hiding this comment.
Code Review Summary
This PR introduces an outbox + 1-minute projection-bucket architecture for availability monitoring, moving the read path off direct message_request scans onto pre-aggregated avail_bucket_1m / avail_current tables — a sound scalability direction. The main gaps are (1) the new 341-line projection worker ships with no unit tests despite CLAUDE.md's 80% coverage rule, and (2) two semantic inconsistencies in how avail_current is computed and refreshed versus the read path, which can make the same provider report different windows or a stale "last-known" status.
PR Size: XL
- Lines changed: 1215 (815 additions / 400 deletions)
- Files changed: 11
Suggested splits (the layers are independently valid and would review more cleanly stacked):
- Schema/migration + DB trigger —
drizzle/0120_*.sql,projection-tables.ts - Projection worker + lifecycle wiring —
projection-worker.ts,instrumentation.ts,shutdown.ts - Read-path rewrite + test updates —
availability-service.ts,index.ts,availability-service.test.ts
Issues Found
| Category | Critical | High | Medium | Low |
|---|---|---|---|---|
| Logic/Bugs | 0 | 0 | 2 | 0 |
| Security | 0 | 0 | 0 | 0 |
| Error Handling | 0 | 0 | 0 | 0 |
| Types | 0 | 0 | 0 | 0 |
| Comments/Docs | 0 | 0 | 0 | 0 |
| Tests | 0 | 1 | 0 | 0 |
| Simplification | 0 | 0 | 0 | 0 |
Critical Issues (Must Fix)
None.
High Priority Issues (Should Fix)
- [Tests]
projection-worker.tshas no unit tests. The entire write path (outbox claiming,proj_applied_requestsdedup,avail_bucket_1mupsert,avail_currentrecompute, 48h backfill,asPayload) is untested, violating the CLAUDE.md 80% coverage rule. See inline comment onprojection-worker.ts:115.
Medium Priority Issues (Consider Fixing)
- [Logic]
avail_currentwindow (5m) vs. read-path fallback window (15m) mismatch. Same provider reports status from a different time window depending on whether itsavail_currentrow is present. See inline comment onprojection-worker.ts:247. - [Logic] Idle providers surface a frozen status.
avail_currentis recomputed only on new events, so an idle provider keeps returning its last-knownstate/requestCountinstead of"unknown", contradicting the code's documented "no data = unknown" invariant. See inline comment onavailability-service.ts:546.
Other Considerations (non-blocking, lower confidence)
- Unbounded growth:
avail_bucket_1m(~1 row/provider/min) andproj_applied_requests(1 row/request) have no retention/TTL job. The repo has cleanup patterns for comparable growing tables (probe-log-cleanup,replay-cleanup); consider adding a projection retention pass. - The
p50/p95/p99latency fields now all collapse to the mean (documented in a code comment as temporary until a sketch-based p95 lands) — downstream consumers reading those field names should be aware they no longer reflect true percentiles.
Review Coverage
- Logic and correctness
- Security (OWASP Top 10)
- Error handling
- Type safety
- Documentation accuracy
- Test coverage
- Code clarity
Automated review by Claude AI
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (5)
tests/unit/lib/availability/projection-worker.test.ts (3)
27-38: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win五个用例重复相同的三段
vi.doMock。请提取为公共辅助函数。
@/drizzle/db、@/lib/migrate、@/lib/logger的 mock 在每个用例中逐字重复。提取一个接收execute与transaction的辅助函数，可减少重复并降低后续新增依赖时的改动面。♻️ 建议的辅助函数
+function mockWorkerDeps(db: { execute: unknown; transaction: unknown }) { + const withAdvisoryLock = vi.fn(async (_n: string, fn: () => Promise<unknown>) => ({ + ran: true, + result: await fn(), + })); + vi.doMock("`@/drizzle/db`", () => ({ db })); + vi.doMock("`@/lib/migrate`", () => ({ withAdvisoryLock })); + vi.doMock("`@/lib/logger`", () => ({ + logger: { info: vi.fn(), warn: vi.fn(), error: vi.fn() }, + })); + return { withAdvisoryLock }; +}Also applies to: 79-90, 132-143, 172-183, 196-202
🤖 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/availability/projection-worker.test.ts` around lines 27 - 38, 在 projection-worker 测试中提取一个公共 mock 辅助函数，接收 execute 和 transaction 实现，并集中配置 `@/drizzle/db`、@/lib/migrate 与 `@/lib/logger` 的 mock；替换五个用例中重复的三段 vi.doMock 调用，保持现有 withAdvisoryLock 行为及各用例传入的数据库函数不变。
99-99: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win断言硬编码
15，与CURRENT_PROVIDER_STATUS_WINDOW_MINUTES脱钩。
projection-worker.ts通过sql.raw(String(CURRENT_PROVIDER_STATUS_WINDOW_MINUTES))生成该字面量。常量一旦调整，本断言会失败，但失败信息不指向根因。直接引用常量可让测试跟随实现。♻️ 建议的断言写法
+ const { CURRENT_PROVIDER_STATUS_WINDOW_MINUTES } = await import( + "`@/lib/availability/availability-service`" + ); expect( - texts.some((t) => t.includes("15 * interval '1 minute'")) + texts.some((t) => + t.includes(`${CURRENT_PROVIDER_STATUS_WINDOW_MINUTES} * interval '1 minute'`) + ) ).toBe(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/lib/availability/projection-worker.test.ts` at line 99, 更新 projection-worker 测试中的 SQL 断言，移除硬编码的 15，改为引用 CURRENT_PROVIDER_STATUS_WINDOW_MINUTES 生成期望的 interval 字面量，使断言与 projection-worker.ts 的实现保持同步。
192-209: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win回填与状态重算的关键分支仍未覆盖。
当前只覆盖
backfill_done已存在的 no-op 路径。以下分支仍无测试：
bootstrapBackfill在锁被其他实例占用时（withAdvisoryLock返回{ ran: false }）跳过并记录日志。- 回填按
BACKFILL_CHUNK_HOURS分块，且stopRequested为真时中断循环并不写入backfill_done。recomputeAvailCurrent把窗口内无流量的 provider 重置为unknown的第二条UPDATE。这些分支决定回填的正确性与状态衰减行为。需要我补写这些用例吗？
🤖 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/availability/projection-worker.test.ts` around lines 192 - 209, 补充 projection-worker 的分支测试：覆盖 bootstrapBackfill 在 withAdvisoryLock 返回 ran:false 时记录日志并跳过；模拟多个 BACKFILL_CHUNK_HOURS 分块并验证 stopRequested 为真时中断且不写入 backfill_done；同时覆盖 recomputeAvailCurrent 对窗口内无流量 provider 执行第二条 UPDATE 并重置为 unknown。复用现有 mock 与 __test__ 导出，分别断言锁、查询、更新及日志调用。src/lib/availability/projection-worker.ts (2)
334-336: 🎯 Functional Correctness | 🔵 Trivial | 💤 Low value
occurred_at使用字符串比较，可能取到较早的时间。
occurredAt是 payload 中的原始字符串。不同来源的时间格式可能不同，例如2026-04-13T08:03:12.000Z与2026-04-13T08:03:12+00:00。字符串比较在这种情况下会得出错误结果，last_request_at会低于真实值。改为时间戳比较更稳妥。♻️ 建议的比较方式
- if (occurredAt > prev.lastRequestAtIso) { + if (Date.parse(occurredAt) > Date.parse(prev.lastRequestAtIso)) { prev.lastRequestAtIso = occurredAt; }🤖 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/availability/projection-worker.ts` around lines 334 - 336, 在处理 lastRequestAtIso 的更新逻辑中，修改 occurredAt 与 prev.lastRequestAtIso 的比较方式，先将两个时间字符串解析为时间戳后再比较，避免不同 ISO 格式导致字典序错误；保留较新时间对应的 occurredAt 作为 prev.lastRequestAtIso。
109-146: 🚀 Performance & Scalability | 🔵 Trivial回填会执行约 400 次分块查询，请确认
message_request上的索引。
BACKFILL_RANGE_DAYS为 100，BACKFILL_CHUNK_HOURS为 6，循环共约 400 次。每次执行INSERT ... SELECT，其中包含对message_request.created_at的范围过滤、对providers的LEFT JOIN，以及对proj_applied_requests的NOT EXISTS。若message_request.created_at没有索引，每个分块都会触发全表扫描，首次启动会显著加重数据库负载，这与本 PR 要解决的问题相冲突。请确认存在
message_request(created_at)索引，并考虑在回填分块之间加入短暂让步，避免持续占满数据库。🤖 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/availability/projection-worker.ts` around lines 109 - 146, Ensure the backfill query used by enqueueBackfillChunk has an index on message_request.created_at to support its time-range filtering, adding the appropriate schema migration if needed. Also add a brief yield between iterations of bootstrapBackfill’s chunk loop so repeated INSERT ... SELECT operations do not continuously monopolize the database.
🤖 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/lib/availability/projection-worker.ts`:
- Around line 202-207: 在处理 providerIds 的相关事务中先生成稳定排序的 orderedIds，并在两处
providerIds.map 使用它；第一条聚合查询显式按 provider_id 排序，第二条 UPDATE avail_current 通过按
provider_id 排序的派生表驱动更新，以固定所有实例的加锁顺序。同步修改该文件中对应的另外两处 providerIds 使用点，保持批处理行为不变。
---
Nitpick comments:
In `@src/lib/availability/projection-worker.ts`:
- Around line 334-336: 在处理 lastRequestAtIso 的更新逻辑中，修改 occurredAt 与
prev.lastRequestAtIso 的比较方式，先将两个时间字符串解析为时间戳后再比较，避免不同 ISO 格式导致字典序错误；保留较新时间对应的
occurredAt 作为 prev.lastRequestAtIso。
- Around line 109-146: Ensure the backfill query used by enqueueBackfillChunk
has an index on message_request.created_at to support its time-range filtering,
adding the appropriate schema migration if needed. Also add a brief yield
between iterations of bootstrapBackfill’s chunk loop so repeated INSERT ...
SELECT operations do not continuously monopolize the database.
In `@tests/unit/lib/availability/projection-worker.test.ts`:
- Around line 27-38: 在 projection-worker 测试中提取一个公共 mock 辅助函数，接收 execute 和
transaction 实现，并集中配置 `@/drizzle/db`、@/lib/migrate 与 `@/lib/logger` 的
mock；替换五个用例中重复的三段 vi.doMock 调用，保持现有 withAdvisoryLock 行为及各用例传入的数据库函数不变。
- Line 99: 更新 projection-worker 测试中的 SQL 断言，移除硬编码的 15，改为引用
CURRENT_PROVIDER_STATUS_WINDOW_MINUTES 生成期望的 interval 字面量，使断言与
projection-worker.ts 的实现保持同步。
- Around line 192-209: 补充 projection-worker 的分支测试：覆盖 bootstrapBackfill 在
withAdvisoryLock 返回 ran:false 时记录日志并跳过；模拟多个 BACKFILL_CHUNK_HOURS 分块并验证
stopRequested 为真时中断且不写入 backfill_done；同时覆盖 recomputeAvailCurrent 对窗口内无流量
provider 执行第二条 UPDATE 并重置为 unknown。复用现有 mock 与 __test__ 导出，分别断言锁、查询、更新及日志调用。
🪄 Autofix
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 Plus
Run ID: 2e7d2589-88f3-4e93-b4c7-d098a0518b86
📒 Files selected for processing (11)
drizzle/0120_availability_projection.sqldrizzle/meta/0120_snapshot.jsonsrc/drizzle/schema.tssrc/lib/availability/availability-service.tssrc/lib/availability/index.tssrc/lib/availability/projection-tables.tssrc/lib/availability/projection-worker.tssrc/lib/availability/types.tssrc/lib/lifecycle/shutdown.tstests/unit/lib/availability-service.test.tstests/unit/lib/availability/projection-worker.test.ts
🚧 Files skipped from review as they are similar to previous changes (5)
- src/lib/availability/index.ts
- drizzle/0120_availability_projection.sql
- src/lib/lifecycle/shutdown.ts
- src/lib/availability/availability-service.ts
- tests/unit/lib/availability-service.test.ts
Sort provider/bucket ids before upsert and recompute so concurrent worker instances take avail_current locks in the same order.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 985f1dbf56
ℹ️ 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".
| WHERE ${requestConditions} | ||
| provider_id AS "providerId", | ||
| date_bin( | ||
| (${bucketSizeMinutes} * INTERVAL '1 minute'), |
There was a problem hiding this comment.
When bucketSizeMinutes is below 1 (the 15-minute dashboard computes 0.25 and the API still accepts 0.25), this date_bin runs against rows that already summarize a full minute, so a whole minute of requests is reported as a 15-second bucket and bucketEnd is also only 15 seconds later. Short-range availability heatmaps/API responses are therefore mislabeled and can overcount each slot; clamp the accepted/display bucket size to at least 1 minute, or keep sub-minute raw data, before aggregating these projection rows.
Useful? React with 👍 / 👎.
2 participants
Summary
avail_bucket_1m/avail_current），不再对message_request做逐行 outcome + percentile 聚合。outbox_events，进程内 worker 异步消费并维护桶；API 路由与代理写链路无需改动。0120_availability_projection，并在instrumentation/shutdown接入 worker 启停；单测对齐投影读路径。Problem
可用性监控的 provider + 时间范围聚合此前始终走
message_request全表扫描（即使命中部分索引，逐行fn_compute_message_request_success_rate_outcome+percentile_cont在大请求量下仍偏重）。本 PR 把"每次查询重算"改为"终态写入时增量投影"，从根本上消除该读路径对message_request的扫描。Related Issues & PRs:
status_code IS NOT NULL谓词对齐部分索引缓解了 CPU 飙升；本 PR 进一步让读路径不再扫描message_request，从根上消除该热点。fix(availability): restore status_code IS NOT NULL terminal filter。本 PR 在其终态语义基础上引入投影表，保留status_code IS NOT NULL作为 trigger 触发条件。fix: 可用性监控仅统计终态请求。该 PR 建立的status_code IS NOT NULL终态边界与部分索引，在本 PR 的 trigger 中沿用。fix(metrics): align success-rate outcome semantics。该 PR 引入的fn_compute_message_request_success_rate_outcome分类法，在本 PR 的 trigger / worker 中复用。Solution
读路径直接聚合预投影的 1 分钟桶（
avail_bucket_1m），写路径通过 outbox 模式异步维护桶数据，API 路由与代理写链路保持不变。设计要点:
FOR UPDATE SKIP LOCKEDproj_applied_requests(request_id)projection_meta.backfill_done）queryProviderAvailability/getCurrentProviderStatus对外类型与路由契约不变；getCurrentProviderStatus优先读avail_current，缺失时回退到avail_bucket_1m5 分钟窗口Changes
Core Changes
drizzle/0120_availability_projection.sql(+133)：新建outbox_events/outbox_processed/proj_applied_requests/avail_bucket_1m/avail_current/projection_meta表，以及message_request上的AFTER INSERT OR UPDATEtriggertrg_message_request_outbox（终态时写入 outbox 事件）。src/lib/availability/projection-worker.ts(+341，新增)：进程内 outbox 消费者，FOR UPDATE SKIP LOCKED批量拉取事件、增量更新 1 分钟桶与avail_current；启动时执行近 48h 回填。src/lib/availability/projection-tables.ts(+65，新增)：投影表的 Drizzle 定义。src/lib/availability/availability-service.ts(+132/-234)：queryProviderAvailability改为从avail_bucket_1m用date_bin聚合；getCurrentProviderStatus改为读avail_current，缺失时回退avail_bucket_1m。移除全部message_request扫描、fn_compute_message_request_success_rate_outcome内联调用与percentile_cont逻辑。Supporting Changes
src/instrumentation.ts(+22)：在两个启动分支接入startAvailabilityProjectionWorker()。src/lib/lifecycle/shutdown.ts(+12)：优雅停机时调用stopAvailabilityProjectionWorker()。src/lib/availability/index.ts：模块文档更新（投影读路径说明）。drizzle/meta/_journal.json：登记迁移0120_availability_projection。Tests
tests/unit/lib/availability-service.test.ts(+75/-162)：断言读路径来自avail_bucket_1m、含date_bin、不再出现message_request/fn_compute_message_request_success_rate_outcome/percentile_cont；新增avail_current缺失回退用例。tests/unit/lib/shutdown.test.ts(+20) /tests/unit/server-shutdown.test.ts(+3)：mock 并断言stopAvailabilityProjectionWorker在停机链路被调用。Migration & Breaking Changes
0120_availability_projectionbun run db:migrate或依赖AUTO_MIGRATE=truemessage_request新增AFTER INSERT OR UPDATEtriggeroutbox_events，写路径有少量额外开销status_code由 NULL 变非 NULL 时触发；幂等由proj_applied_requests保证avail_bucket_1m为空期间读路径无数据getCurrentProviderStatus对缺失 provider 回退avail_bucket_1m；无数据返回unknown（不会误判为 green）已知精度回归（临时）： p50 / p95 / p99 暂以平均延迟（
SUM(latency_sum_ms)/SUM(latency_cnt)）回填，待后续引入 sketch-based p95 后恢复。1 分钟桶只保存latency_cnt与latency_sum_ms，不保留原始分布。Test plan
bunx vitest run tests/unit/lib/availability-service.test.ts tests/unit/lib/shutdown.test.tsbun run db:migrate应用0120_availability_projection[AvailProjection] worker startedGET /api/availability/current、GET /api/availability?startTime=...&endTime=...返回结构与改前一致且延迟明显下降outbox_events被消费、avail_bucket_1m/avail_current更新stopAvailabilityProjectionWorkerChecklist
bun run test)bun run build/bun run lint/bun run typecheckDescription enhanced by Claude AI
Greptile Summary
The PR replaces availability queries over request logs with an outbox-driven one-minute projection and integrates its worker into application startup and shutdown.
Confidence Score: 4/5
The PR is not yet safe to merge because non-minute-aligned availability queries still return metrics for requests outside the requested interval.
The projection stores only whole-minute aggregates, while the read path truncates both supplied boundaries and includes the complete boundary buckets, so exact time-range queries return incorrect counts and derived availability metrics.
Files Needing Attention: src/lib/availability/availability-service.ts and drizzle/0120_availability_projection.sql
Important Files Changed
Sequence Diagram
Reviews (3): Last reviewed commit: "fix(availability): stable provider lock ..." | Re-trigger Greptile
Context used (3)