bug(cohorts): non-static cohorts stop refreshing permanently — fixed jobId blocks every cohortRefresh tick

[niajkitir]

Summary

After the first successful compute, non-static cohorts stop being recomputed permanently. The cohortRefresh cron keeps firing on schedule but every enqueue is a silent no-op. The only way to refresh a cohort is the Refresh button in the UI, and one click buys exactly one working cron cycle before it re-freezes.

Affects main as of today. bullmq ^5.63.0.

Root cause

cohortRefreshCronJob and enqueueCohortCompute both enqueue with a fixed jobId:

// apps/worker/src/jobs/cron.cohort-refresh.ts
cohortComputeQueue.add(
  'cohortCompute',
  { cohortId: cohort.id },
  {
    jobId: `cohort-${cohort.id}`,
    removeOnComplete: { age: 3600 },
    removeOnFail: { age: 86400 },
  },
)

BullMQ's add short-circuits while a job with that id still has a Redis record:

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

The key detail: removeOnComplete: { age } is not a TTL. There is no expiry on the job hash and nothing on a timer. removeJobsByMaxAge runs only from inside moveToFinished / removeJobsOnFail — i.e. only as a side effect of some job in that queue finishing — and it collects predecessors only, since the finishing job's own zset score equals the trim timestamp:

-- moveToFinished-14.lua
if maxCount ~= 0 then
    rcall("ZADD", targetSet, timestamp, jobId)
    rcall("HSET", jobIdKey, ..., "finishedOn", timestamp)   -- hash KEPT
    if maxAge ~= nil then
        removeJobsByMaxAge(timestamp, maxAge, targetSet, prefix)
    end
else
    removeJobKeys(jobIdKey)          -- removeOnComplete: true self-deletes
end

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 use the same blocked ids. Once every non-static cohort holds a finished record, nothing can ever be added, so nothing finishes, so nothing is ever collected. Permanent, not a 1h delay.

removeCohortComputeJob in the UI Refresh path (remove() then add()) is the only escape. Because removeJobsByMaxAge trims the whole target zset by score, one manual refresh also evicts every other cohort's hour-old record — so the next tick works for everyone, they all complete, all write fresh records, and it deadlocks again. That is why the symptom reads as intermittent rather than broken.

Evidence from a production instance

6 non-static cohorts, worker healthy:

• bull:cron:repeat:cohortRefresh armed, ~4400 recorded firings, next run scheduled.
• bull:cohortCompute:wait and :active both empty.
• 4 completed job records ~45h old despite age: 3600, all with ttl = -1 (no expiry set).
• Newest cohorts.lastComputedAt was 2 days stale.

The boundary case is the clearest fingerprint: on one day, 3 cohorts computed at 08:08 and 2 at 08:30. The 08:08 group blocked itself at the 08:30 tick; the 08:30 pair only ran because the 08:08 completions had trimmed their older records.

Suggested fix

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

await cohortComputeQueue.add(
  'cohortCompute',
  { cohortId },
  {
    deduplication: { id: `cohort-${cohortId}` },
    removeOnComplete: { age: 3600, count: 100 },
    removeOnFail: { age: 86400 },
  },
)

Measured against a real Redis:

ids equal (deduped while in flight) = true
dedup key pttl                      = -1
dedup key after completion          = 0     -> next add lands in `wait`
dedup key after terminal failure    = 0     -> next add lands in `wait`
failedReason retained               = yes

This keeps the failure record, which a plain remove()-before-add() would destroy at the following tick.

Two smaller notes while in there:

• cohortRefreshCronJob duplicates the options of enqueueCohortCompute instead of calling it, so the two can drift — fixing only the helper misses the cron.
• removeOnComplete: true would also work (it takes the self-deleting branch), but loses completed-job visibility.