fix(cohorts): stop non-static cohorts from freezing permanently after first compute

[lindesvard]

Fixes #424.

The bug

cohortRefreshCronJob and enqueueCohortCompute both enqueued with a fixed jobId: cohort-<id>. BullMQ's add short-circuits while any Redis record for that id still exists:

-- addStandardJob-9.lua
if rcall("EXISTS", jobIdKey) == 1 then
    return handleDuplicatedJob(...)   -- never reaches storeJob / LPUSH to wait
end

The key detail is that removeOnComplete: { age } is not a TTL. Nothing expires on a timer — removeJobsByMaxAge runs only from inside moveToFinished/removeJobsOnFail, i.e. as a side effect of some job in that queue finishing, and it collects predecessors only.

So the states form a cycle:

• deleting cohort-X's record requires some job in the queue to finish ≥1h after it,
• a job can only finish if it was added,
• it can only be added if no record exists for its jobId.

cohortCompute has only three producers — the cron, cohort create/update, and the UI Refresh — and the first two used the same blocked ids. Once every non-static cohort holds a completed record, nothing can ever be added, so nothing finishes, so nothing is ever collected. Permanent, not a 1h delay. The cron keeps firing on schedule and every enqueue is a silent no-op.

This also explains why it looked intermittent: removeCohortComputeJob (remove() then add()) was the only escape, and because removeJobsByMaxAge trims the whole target zset by score, one manual Refresh evicted every other cohort's hour-old record too — so the next tick worked for everyone, they all completed, all wrote fresh records, and it deadlocked again.

The fix

Deduplicate on the cohort rather than pinning a jobId:

await cohortComputeQueue.add(
  'cohortCompute',
  { cohortId },
  { deduplication: { id: `cohort-${cohortId}` } },
);

With no ttl, the deduplication key is released by moveToFinished on completion or failure — removeDeduplicationKeyIfNeededOnFinalization is called above the completed/failed branching — so it collapses only a compute that is genuinely still in flight, and finished records stop gating anything.

Verification

Ran against a real Redis on bullmq@5.63.0, comparing both enqueue shapes over three ticks with no age-trim opportunity in between:

BEFORE (jobId: `cohort-A`)
  processed 1 time(s) over 3 ticks  ->  STUCK
  counts: {"wait":0,"active":0,"completed":1,"failed":0}
AFTER  (deduplication: { id: `cohort-A` })
  processed 3 time(s) over 3 ticks  ->  OK
  counts: {"wait":0,"active":0,"completed":3,"failed":0}

In-flight dedup:
  second add while active started a 2nd run? no (correct)
  dedup key pttl while in flight = -1 (expect -1, no TTL)
  dedup key exists after completion = 0 (expect 0 -> next add lands in wait)

After terminal failure:
  dedup key exists = 0 (expect 0)
  re-enqueue ran again? YES (correct)
  failedReason retained = boom

Also in this PR

Both raised in the issue as worth doing while in here:

• cohortRefreshCronJob now calls enqueueCohortCompute instead of duplicating its options. Fixing only the helper would have missed the cron, which is the producer that actually matters.
• removeCohortComputeJob is deleted, along with its two call sites in cohort.ts (update and refresh). It only existed to work around the fixed jobId; with dedup a stale record no longer blocks anything, and remove() before add() would destroy the failure record that dedup deliberately preserves. Behaviour is equivalent — remove() never removed active jobs anyway, and a waiting deduped job re-reads the cohort definition when it runs.
• removeOnComplete/removeOnFail gain a count bound on the queue defaults, since age alone only trims when another job finishes.
• Dropped two stray console.logs in the cohort router that were sitting on the changed lines.

removeOnComplete: true would also have fixed the deadlock (it takes the self-deleting branch) but loses completed-job visibility, so it isn't used here.

Checks

tsc --noEmit reports zero errors in every touched file (cohort.service.ts, queues.ts, routers/cohort.ts, cron.cohort-refresh.ts). Remaining repo-wide errors are pre-existing in notification.service.ts and insights/store.ts.

https://claude.ai/code/session_01ETRr6KYLYATzBVaLRZcwwk

Summary by CodeRabbit

• Improvements
  • Improved reliability of automatic and manual cohort refresh processing.
  • Prevented unnecessary duplicate cohort computation jobs.
  • Added limits to retained completed and failed background jobs to help manage queue storage.
  • Simplified cohort updates and refreshes by removing redundant job-removal steps.

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro Plus

Run ID: 2e05e911-faa0-4112-b2d3-15ef7163e1cb

📥 Commits

Reviewing files that changed from the base of the PR and between f72310c and 5403feb.

📒 Files selected for processing (4)

• apps/worker/src/jobs/cron.cohort-refresh.ts
• packages/db/src/services/cohort.service.ts
• packages/queue/src/queues.ts
• packages/trpc/src/routers/cohort.ts

💤 Files with no reviewable changes (1)

• packages/trpc/src/routers/cohort.ts

Included review availability: Your plan includes up to 8 reviews per rolling hour; 6 remain after this review.

---

📝 Walkthrough

Walkthrough

Cohort computation producers now use enqueueCohortCompute, which deduplicates only in-flight work by cohort. The queue also limits retained completed and failed jobs by age and count. Explicit job removal was deleted from cohort update and refresh paths.

Changes

Cohort compute refresh

Layer / File(s)	Summary
Cohort-specific enqueue flow packages/db/src/services/cohort.service.ts, apps/worker/src/jobs/cron.cohort-refresh.ts, packages/trpc/src/routers/cohort.ts	enqueueCohortCompute uses a cohort-specific BullMQ deduplication key. Cron, cohort updates, and manual refreshes enqueue computation without explicit job removal.
Bounded completed-job retention packages/queue/src/queues.ts	cohortComputeQueue limits completed and failed jobs by both retention age and a maximum count of 100.

Estimated code review effort: 3 (Moderate) | ~20 minutes

Merge Risk: ⚪ Minimal · up to 5403f

The PR changes cohort compute scheduling so completed jobs no longer permanently block future refreshes; no actionable merge-blocking risk remains beyond normal checks and review.

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 25.00% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (4 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title clearly identifies the cohort refresh bug and the fix's purpose.
Linked Issues check	✅ Passed	The changes implement cohort-based deduplication, shared enqueueing, workaround removal, and queue count bounds required by issue #424.
Out of Scope Changes check	✅ Passed	All changes support issue #424 by fixing refresh deduplication, queue cleanup, or related obsolete code.

✨ Finishing Touches 💡 1

📝 Generate docstrings 💡

• Create stacked PR
• Commit on current branch

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch fix/cohort-refresh-deadlock-424

---

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.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}