Conversation
- Replace slow LEFT JOIN backfill discovery (scanned all logs) with sequential hour-based iteration using generate_series and the created_at index - Add concurrency guard to prevent overlapping runs from stacking up DB connections - Change log-process log level to trace Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Replace frontier-based backfill (max hour + 1) with generate_series + NOT EXISTS to find actual unprocessed hours, including gaps left by previous partial runs. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
WalkthroughRefactored the project stats aggregator service to transition from LEFT JOIN-based bucket detection to per-hour processing. Added hour-truncation utilities, reworked the backfill loop to iterate over hourly ranges, and implemented a module-level concurrency guard to prevent overlapping invocations. Changes
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~20 minutes Possibly related PRs
🚥 Pre-merge checks | ✅ 3✅ Passed checks (3 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing touches
🧪 Generate unit tests (beta)
No actionable comments were generated in the recent review. 🎉 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 |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Fix all issues with AI agents
In `@apps/worker/src/services/project-stats-aggregator.ts`:
- Around line 434-449: The current hours generation marks an hour as processed
if any row exists in projectHourlyStats for that hour, which skips
partially-processed hours; to fix, change the logic so the missing-check is done
per-project: either (A) modify the SQL that populates hoursToProcess to
join/compare distinct projects-with-logs against distinct projects-with-stats
per hour (include project identifier in the NOT EXISTS check against
projectHourlyStats and select only hours where some project is missing), or (B)
keep the generate_series query but inside the per-hour processing loop filter
the projects list by performing a NOT EXISTS check per (hour, project) against
projectHourlyStats (use projectHourlyStats.hourTimestamp and project id column)
so only projects that lack a stats row are processed; update references to
hoursToProcess, projectHourlyStats, projectHourlyStats.hourTimestamp and the
per-hour loop that iterates over STATS_BATCH_SIZE accordingly.
🧹 Nitpick comments (2)
apps/worker/src/services/project-stats-aggregator.ts (2)
424-424: Unboundedgenerate_serieswhenSTATS_BACKFILL_DAYS=0.When backfill days is set to 0 (unlimited),
rangeStartbecomes1970-01-01 00:00:00, producing ~480K+ hour slots in the series. PostgreSQL'sgenerate_serieswill materialize this before applying theNOT EXISTSfilter andLIMIT. On a system with few missing hours this is still fast, but it's worth noting the potential for a large intermediate set.Consider falling back to the earliest log
created_atinstead of epoch when unlimited backfill is configured, e.g. a quickSELECT to_char(date_trunc('hour', min(created_at)), ...) FROM logto bound the range.
456-484: Sequential per-project, per-hour processing issues 4 separate queries — consider batching.For each project in each hour, 4 sequential
INSERT ... ON CONFLICTcalls are made. During initial backfill of many hours with many projects, this could be very slow. Since this is a background job it won't block users, but it does extend the window during which the concurrency guard blocks subsequent runs.Consider wrapping the 4 recalculate calls per project in a single transaction to reduce round-trips, or processing all projects for an hour in parallel (e.g.,
Promise.allwith a concurrency limit).
| const hoursToProcess = await database.execute<{ | ||
| hour_timestamp: string; | ||
| }>(sql` | ||
| SELECT to_char(h, 'YYYY-MM-DD HH24:MI:SS') AS hour_timestamp | ||
| FROM generate_series( | ||
| ${rangeStart}::timestamp, | ||
| ${currentHourStart}::timestamp - interval '1 hour', | ||
| interval '1 hour' | ||
| ) AS h | ||
| WHERE NOT EXISTS ( | ||
| SELECT 1 FROM ${projectHourlyStats} | ||
| WHERE ${projectHourlyStats.hourTimestamp} = h | ||
| ) | ||
| .groupBy(log.projectId, sql`date_trunc('hour', ${log.createdAt})`) | ||
| .orderBy(sql`date_trunc('hour', ${log.createdAt}) ASC`) | ||
| .limit(STATS_BATCH_SIZE); | ||
| ORDER BY h ASC | ||
| LIMIT ${STATS_BATCH_SIZE} | ||
| `); |
There was a problem hiding this comment.
Backfill skips hours that are only partially aggregated (some projects missing).
The NOT EXISTS check on Line 443–446 considers an hour "processed" if any project_hourly_stats row exists for that hour_timestamp, regardless of project. If a previous run aggregated project A for hour X but crashed before processing project B, that hour will never be retried by Phase 1. Phase 2 (stale detection) also won't help because project B has no existing stats row to compare against.
Add the project dimension to the check, or change the approach to compare distinct projects-with-logs against distinct projects-with-stats per hour. One lightweight fix: instead of filtering hours globally, keep the current hour list but inside the per-hour loop, use NOT EXISTS per-project to skip only projects already aggregated:
Proposed approach sketch
Keep the generate_series query as-is to find candidate hours (hours where at least one project may be missing), but change the NOT EXISTS to look for hours where at least one project is missing stats:
SELECT to_char(h, 'YYYY-MM-DD HH24:MI:SS') AS hour_timestamp
FROM generate_series(
${rangeStart}::timestamp,
${currentHourStart}::timestamp - interval '1 hour',
interval '1 hour'
) AS h
-WHERE NOT EXISTS (
- SELECT 1 FROM ${projectHourlyStats}
- WHERE ${projectHourlyStats.hourTimestamp} = h
-)
+WHERE EXISTS (
+ SELECT 1 FROM ${log}
+ WHERE ${log.createdAt} >= h
+ AND ${log.createdAt} < h + interval '1 hour'
+ LIMIT 1
+)
+AND NOT EXISTS (
+ SELECT 1 FROM ${projectHourlyStats}
+ WHERE ${projectHourlyStats.hourTimestamp} = h
+)
ORDER BY h ASC
LIMIT ${STATS_BATCH_SIZE}Alternatively, within the per-hour loop (lines 460–478), filter out projects that already have a stats row for that hour so partially-processed hours are completed on future runs — though that requires a separate change to include partially-processed hours in the candidate list.
🤖 Prompt for AI Agents
In `@apps/worker/src/services/project-stats-aggregator.ts` around lines 434 - 449,
The current hours generation marks an hour as processed if any row exists in
projectHourlyStats for that hour, which skips partially-processed hours; to fix,
change the logic so the missing-check is done per-project: either (A) modify the
SQL that populates hoursToProcess to join/compare distinct projects-with-logs
against distinct projects-with-stats per hour (include project identifier in the
NOT EXISTS check against projectHourlyStats and select only hours where some
project is missing), or (B) keep the generate_series query but inside the
per-hour processing loop filter the projects list by performing a NOT EXISTS
check per (hour, project) against projectHourlyStats (use
projectHourlyStats.hourTimestamp and project id column) so only projects that
lack a stats row are processed; update references to hoursToProcess,
projectHourlyStats, projectHourlyStats.hourTimestamp and the per-hour loop that
iterates over STATS_BATCH_SIZE accordingly.
There was a problem hiding this comment.
Pull request overview
Improves the worker’s hourly stats aggregation by changing how backfill discovers missing aggregation windows, reducing log-processing verbosity, and preventing overlapping refresh runs from piling up database work.
Changes:
- Downgrade per-log processing output from
infototrace. - Replace the backfill “find missing buckets” query with an hour-based
generate_series+NOT EXISTSapproach and per-hour project discovery. - Add an in-process
isRunningguard to skip concurrent refresh executions.
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 5 comments.
| File | Description |
|---|---|
| apps/worker/src/worker.ts | Lowers log processing verbosity to reduce log volume. |
| apps/worker/src/services/project-stats-aggregator.ts | Reworks backfill discovery/iteration logic and adds an overlap guard for refresh runs. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| ), | ||
| ); | ||
|
|
||
| if (projects.length === 0) { |
There was a problem hiding this comment.
The generated hour list can include hours that have no logs. In that case projects.length === 0 and the code continues without writing any stats row, so that hour will remain "missing" forever and will be selected again on the next run (potentially blocking progress at the start of the range). Consider adding an EXISTS filter on the logs table in the SQL that selects hoursToProcess, or otherwise skipping hours with no logs in a way that won’t reselect them indefinitely.
| if (projects.length === 0) { | |
| if (projects.length === 0) { | |
| logger.info( | |
| `[backfill] Skipping hour ${i + 1}/${hoursToProcess.rows.length}: ${hourTimestamp} (no projects with logs)`, | |
| ); | |
| totalBucketsProcessed++; |
| for (let i = 0; i < hoursToProcess.rows.length; i++) { | ||
| const hourTimestamp = hoursToProcess.rows[i].hour_timestamp; | ||
|
|
||
| // Find all projects with logs in this hour | ||
| const projects = await database | ||
| .selectDistinct({ projectId: log.projectId }) | ||
| .from(log) | ||
| .where( | ||
| and( | ||
| sql`${log.createdAt} >= ${hourTimestamp}::timestamp`, | ||
| sql`${log.createdAt} < ${hourTimestamp}::timestamp + interval '1 hour'`, | ||
| ), | ||
| ); | ||
|
|
||
| if (projects.length === 0) { | ||
| continue; | ||
| } | ||
|
|
||
| for (const { projectId } of projects) { | ||
| await recalculateProjectHourlyStats(projectId, hourTimestamp); | ||
| await recalculateProjectHourlyModelStats(projectId, hourTimestamp); | ||
| await recalculateApiKeyHourlyStats(projectId, hourTimestamp); | ||
| await recalculateApiKeyHourlyModelStats(projectId, hourTimestamp); | ||
| } |
There was a problem hiding this comment.
STATS_BATCH_SIZE is now applied to number of hours, but the work per hour is unbounded (it scales with number of projects that had logs in that hour, and each project triggers 4 aggregation queries). This can cause a single run to take much longer than before and make progress unpredictable. Consider adding a secondary cap (e.g., max project-hour buckets per run) or restructuring to limit the number of (projectId, hourTimestamp) buckets processed per invocation.
| `[backfill] Processed bucket ${i + 1}/${backfillBuckets.length}: project=${bucket.projectId} hour=${bucket.hourTimestamp}`, | ||
| `[backfill] Processed hour ${i + 1}/${hoursToProcess.rows.length}: ${hourTimestamp} (${projects.length} projects)`, | ||
| ); | ||
| totalBucketsProcessed++; |
There was a problem hiding this comment.
totalBucketsProcessed is incremented once per processed hour in the backfill phase, but in the stale phase it is incremented by the number of project-hour buckets. This mixes units while the log line reports "total buckets processed", which can make metrics misleading. Consider incrementing by the number of project-hour buckets processed in backfill (e.g., projects.length) or renaming/logging separate counters for hours vs buckets.
| totalBucketsProcessed++; | |
| totalBucketsProcessed += projects.length; |
| // Phase 1: Backfill — iterate hours sequentially from the frontier. | ||
| // Instead of scanning all logs with a LEFT JOIN (slow on large tables), | ||
| // we find the last processed hour and walk forward, processing each hour | ||
| // individually using the created_at index for fast lookups. |
There was a problem hiding this comment.
The updated backfill comment says it "find[s] the last processed hour and walk[s] forward", but the implementation generates all hours in the window and filters via NOT EXISTS (no frontier-based start). Please align the comment with the actual approach, or implement the described frontier logic to avoid confusion during future maintenance.
| // Phase 1: Backfill — iterate hours sequentially from the frontier. | |
| // Instead of scanning all logs with a LEFT JOIN (slow on large tables), | |
| // we find the last processed hour and walk forward, processing each hour | |
| // individually using the created_at index for fast lookups. | |
| // Phase 1: Backfill — identify and process unaggregated hourly buckets. | |
| // Instead of scanning all logs with a LEFT JOIN (slow on large tables), | |
| // we generate all hours in the backfill window and use NOT EXISTS against | |
| // the small, indexed stats tables to find hours that still need aggregation. |
| // Find hours with gaps: generate all hours in the backfill window, | ||
| // then exclude hours that already have stats rows. | ||
| // This is fast because generate_series is instant and the NOT EXISTS | ||
| // checks against the small project_hourly_stats table (indexed). | ||
| const hoursToProcess = await database.execute<{ | ||
| hour_timestamp: string; | ||
| }>(sql` | ||
| SELECT to_char(h, 'YYYY-MM-DD HH24:MI:SS') AS hour_timestamp | ||
| FROM generate_series( | ||
| ${rangeStart}::timestamp, | ||
| ${currentHourStart}::timestamp - interval '1 hour', | ||
| interval '1 hour' | ||
| ) AS h | ||
| WHERE NOT EXISTS ( | ||
| SELECT 1 FROM ${projectHourlyStats} | ||
| WHERE ${projectHourlyStats.hourTimestamp} = h | ||
| ) |
There was a problem hiding this comment.
Backfill gap detection is currently only checking for existence of any row in project_hourly_stats for a given hour (WHERE projectHourlyStats.hourTimestamp = h). Since the table is unique per (projectId, hourTimestamp), this will incorrectly treat an hour as “processed” if just one project has stats for that hour, leaving gaps for other projects undiscovered. Consider driving the backfill off missing (projectId, hourTimestamp) pairs (e.g., join projects-with-logs per hour and NOT EXISTS on both columns) or otherwise ensure per-project gaps are detected.
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
generate_series starting from a non-hour-aligned timestamp (e.g. 20:27:48) produced non-aligned hour buckets. Each worker restart had a different offset, creating new rows that didn't match previous ones via NOT EXISTS, causing counts to grow indefinitely. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Summary
generate_series+NOT EXISTSagainst the small stats tableNOT EXISTSto find actual unprocessed hours instead of frontier-based approach (max(hour_timestamp) + 1) which skipped gaps left by previous partial runsisRunningflag) to prevent overlapping runs from stacking up DB connections when a batch takes longer than the refresh intervaltraceTest plan
pnpm build:corepasses🤖 Generated with Claude Code
Summary by CodeRabbit
Bug Fixes
Chores