fix(worker): bound BullMQ job retention to prevent Redis OOM - #1693
Merged
Merged
Conversation
Completed and failed jobs were retained by age only (14 days) with no count cap, so Redis memory grew with repo count times reindex frequency. - Window retention now caps completed jobs at 1 day / 5,000 and failed jobs at 14 days / 10,000 per queue. - Queues whose latest job is resolved by the web app (repo-index, repo-permission-sync, connection-sync, account-permission-sync) use a new latestPerResource retention: when a job starts, the job it supersedes for the same resource is removed, keyed by the queue's deduplication id, with a 7-day age backstop for orphans. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Contributor
|
Navigate logical layers of code changes, visualize relationships, and explore their blast radius. No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Essentials Run ID: 📒 Files selected for processing (1)
🚧 Files skipped from review as they are similar to previous changes (1)
Included review availability: This review used your included allowance. Your plan provides up to 8 included reviews per hour; 6 remain after this review. WalkthroughQueue specs now support window-based and latest-per-resource retention. The BullMQ client applies retention settings to jobs and schedulers. Workers track selected jobs and attempt to remove superseded jobs. ChangesQueue retention and latest-job tracking
Priority: ➖ Normal Estimated code review effort: 3 (Moderate) | ~25 minutes Change: Bug fix Sequence Diagram(s)sequenceDiagram
participant JobManager as BullMQJobManager
participant Workload
participant BullMQClient
participant Redis
JobManager->>Workload: Run onStarted
JobManager->>BullMQClient: Track job after onStarted
BullMQClient->>Redis: Read and store resource job ID
BullMQClient-->>JobManager: Return superseded job ID or null
JobManager->>Workload: Process job
Merge Risk: ⚪ Minimal · up to No actionable retention or latest-job tracking risk remains from the inspected changes. The PR is mergeable after normal checks. 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
![cubic-dev-ai[bot]](https://avatars.githubusercontent.com/in/1082092?s=60&v=4){kind=small}
There was a problem hiding this comment.
8 issues found and verified against the latest diff
Prompt for AI agents (unresolved issues)
Check if these issues are valid — if so, understand the root cause of each and fix them. If appropriate, use sub-agents to investigate and fix each issue separately.
<file name="packages/shared/src/bullmqClient.test.ts">
<violation number="1" location="packages/shared/src/bullmqClient.test.ts:275">
P3: This test can pass without exercising the locked-predecessor path: `trackLatestJob` may return `null` without calling `getJob` or `removeJob`. Assert both calls to prove the rejection is caught and the predecessor is left to the age backstop.</violation>
</file>
<file name="packages/shared/src/bullmqClient.ts">
<violation number="1" location="packages/shared/src/bullmqClient.ts:301">
P2: Check the `MULTI/EXEC` command results before removing the predecessor. An OOM-rejected `SET` can leave the pointer stale without rejecting `exec()`, bypassing the job manager's failure log and leaving later superseded jobs until age cleanup.</violation>
<violation number="2" location="packages/shared/src/bullmqClient.ts:304">
P3: The claim that the tracking record and the job expire together overstates the backstop. BullMQ only applies age-based removal when another job in the same queue reaches the same terminal state (as the removed queue.ts comment documented: a lone completed/failed job remains available past its age). When a resource stops running and no other job in that queue ever completes, the last job's `removeOnComplete.age` sweep never fires, so the job lingers after the 7-day tracking-key TTL expires and the comment's "resource that stops running leaves nothing behind" doesn't hold. The leak is bounded to one job per stopped resource, but consider also removing the tracked job explicitly when the key expires, or documenting that the backstop only fires while the queue stays active.</violation>
</file>
<file name="packages/backend/src/connectionSyncWorkload.test.ts">
<violation number="1" location="packages/backend/src/connectionSyncWorkload.test.ts:46">
P3: This mock gives `connection-sync` window retention, while the real queue uses `latestPerResource`, so tests expose the wrong retention contract. Match the real policy in this fixture.</violation>
</file>
<file name="packages/shared/src/queue.ts">
<violation number="1" location="packages/shared/src/queue.ts:103">
P2: A configured scheduler interval longer than seven days can outlive the resource’s latest retained job, leaving its `latest...JobId` pointer referencing a removed job. Keep this backstop above the supported maximum interval or enforce a seven-day maximum on those scheduler settings.</violation>
</file>
<file name="packages/backend/src/jobManager.test.ts">
<violation number="1" location="packages/backend/src/jobManager.test.ts:119">
P2: These tracking tests use a window-retained `connection-sync` workload, so the real client would no-op and the mock hides that mismatch. Configure the tracking-test workload with `latestPerResource` retention so it represents the production queue.</violation>
<violation number="2" location="packages/backend/src/jobManager.test.ts:335">
P2: These mock implementations leak into later tests because `vi.clearAllMocks()` does not reset implementations. Use one-shot implementations for both tracking tests or reset `trackLatestJob` in `beforeEach` to keep later tests isolated.</violation>
</file>
<file name="packages/backend/src/jobManager.ts">
<violation number="1" location="packages/backend/src/jobManager.ts:156">
P2: The invariant stated in the comment above this line doesn't hold for the `repo-index` workload: `createRepoIndexWorkload` defines no `onStarted`, so its `latestIndexingJobId` pointer is written inside `process` → `prepareRepoIndexJob`, which runs after this call. `trackLatestJob` therefore removes the predecessor from the queue before `latestIndexingJobId` points at the new job. If the new job's `process` fails before the pointer write (e.g., DB error in `prepareRepoIndexJob`) or the process crashes in the window, `repo.latestIndexingJobId` references a deleted job and `repo-index-status` returns `latestJob: null`. Add an `onStarted` hook to the repo-index workload (mirroring `connection-sync`/permission-sync) that persists `latestIndexingJobId` before tracking, so removal only happens after the pointer has moved.</violation>
</file>
Reply with feedback, questions, or to request a fix.
Re-trigger cubic
There was a problem hiding this comment.
3 existing issues remain and 1 new issue found across 7 files
Prompt for AI agents (unresolved issues)
Check if these issues are valid — if so, understand the root cause of each and fix them. If appropriate, use sub-agents to investigate and fix each issue separately.
<file name="CHANGELOG.md">
<violation number="1" location="CHANGELOG.md:20">
P3: The entry's "retaining only the latest job per repo, connection, and account" describes only the `latestPerResource` mode. `packages/shared/src/queue.ts` applies window retention to attachment-prune, audit-log-prune, and repo-cleanup (keepJobs: completed 1 day/5,000, failed 14 days/10,000), so those queues retain multiple jobs per repo, not only the latest. Consider wording that covers both modes.</violation>
</file>
Requires human review: Auto-approval blocked because this review re-detected 3 unresolved issues already reported by Cubic.
Re-trigger cubic
…seding trackLatestJob removes a resource's previous job right after onStarted, which assumes the parent's latest...JobId pointer has already moved. repo-index wrote latestIndexingJobId inside process instead, so an abort or DB error before that write could leave the pointer on a removed job. - Move the pointer write to an onStarted hook (updateMany keyed by repo id), matching the other latestPerResource workloads. - Reject a latestPerResource workload without onStarted at registration so the contract cannot be silently skipped by a future workload. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes using high effort and found 1 potential issue.
❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.
Reviewed by Cursor Bugbot for commit e46744e. Configure here.
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.
Fixes SOU-2311
Problem
A customer's Redis (Valkey) hit OOMKills and crash-looped.
DEFAULT_JOB_OPTIONS.keepJobsretained completed and failed BullMQ jobs by age only (14 days) with no count cap, so memory scaled withrepos × reindex runs × 14 days. With ~680 repos on the default hourly reindex that is ~228K completedrepo-indexjobs, each a hash plus a:logslist, matching the observed 935K keys / 1.52 GB.Change
Job retention is now a
retentionpolicy on each queue spec with two modes:window(default): BullMQ's count and age caps for the queue as a whole. Completed jobs keep 1 day / 5,000, failed jobs keep 14 days / 10,000. Used byrepo-cleanup,attachment-prune, andaudit-log-prune.latestPerResource: keeps one job per resource. When a job starts, the job manager records it undersourcebot:latest-job:<queue>:<resource>(keyed by the queue's existing deduplication id) and removes the job it supersedes. A 7-day age-only backstop reclaims jobs for resources that stop running, and the tracking key expires with the same TTL. Used byrepo-index,repo-permission-sync,connection-sync, andaccount-permission-sync, the queues whose latest job the web app resolves through alatest...JobIdpointer.Window retention alone cannot guarantee a resource's latest job survives, since the window is shared by every resource in the queue. The failed set in particular drives the failed-repo filter, the sync-issue popover, and retry-all, so a code host outage that fails every repo at once would overflow any fixed cap.
latestPerResourceturns that into a guarantee bounded only by the age backstop.Tracking happens inside the execution lock, right after
onStarted, so the database pointer has already moved before the predecessor is removed. Retries of the same job are a no-op, a still-locked predecessor is left to the backstop, and tracking failures are logged without failing the job.Rollout
Existing jobs drain by age only: anything older than 7 days is pruned at the first completions after upgrade (1,000 per completion), and the rest age out over the following week. Steady state after that is one job per resource. Schedulers pick up the new template on the next reconcile.
Testing
trackLatestJob(supersede, no predecessor, retry, locked predecessor, window no-op) and for the job manager ordering and failure handling.repo-indexrun for the same repo removed the first job and its logs key, and the tracking key pointed at the second job with a 7-day TTL.🤖 Generated with Claude Code
Note
Medium Risk
Changes core worker queue retention and job lifecycle ordering for sync/index queues; mistakes could drop job history the UI resolves via
latest…JobId, though processing is designed to continue if Redis tracking fails.Overview
BullMQ job retention is reworked so Redis memory no longer grows with repo/connection count. Queue specs now use a
retentionpolicy instead of flatkeepJobs:window(tighter global count/age caps on prune queues) orlatestPerResource(one finished job per deduplication resource, with a 7-day age backstop).For
latestPerResourcequeues (repo-index,connection-sync,account-permission-sync,repo-permission-sync), the job manager callstrackLatestJobafteronStarted(inside the execution lock) to record the current job in Redis and remove the superseded BullMQ job when possible; registration requires anonStartedhook that publishes the job id to the parent’slatest…JobIdfield. Repo indexing moveslatestIndexingJobIdupdates intoonStartedand drops the transactional write from job preparation. Tracking failures are logged and do not fail the running job.Reviewed by Cursor Bugbot for commit 0e4c41c. Bugbot is set up for automated code reviews on this repo. Configure here.
Summary by cubic
Fixes SOU-2311, which caused Redis OOMKills and crash-loops, by replacing age-only 14-day job retention with per-queue retention policies that cap memory growth.
repo-cleanup,attachment-prune, andaudit-log-prune.latestPerResourcemode keeps one job per resource, removing the job it supersedes when a new job starts, with a 7-day age backstop for resources that stop running.repo-index,repo-permission-sync,connection-sync, andaccount-permission-sync, whose latest job the web app resolves through alatest...JobIdpointer.Written for commit 5040f09. Summary will update on new commits.
Summary by CodeRabbit