Reaper startup scan misses stalled jobs once the worker's registration expires

Start with src/consumer.ts around the registerWorker call at line 110, then read refreshWorker implementations and the Reaper's #checkStalledJobs startup scan. Use the supplied reproduction as the initial failing case. Done means live consumers remain visible while running and a restarted Reaper can recover processing jobs even after their worker registration expires.

A Reaper's startup scan only looks at workers returned by storage.getWorkers(), but consumers register their worker once, with a TTL of visibilityTimeout * 2, and never refresh it. Once that TTL passes, a still-running worker drops out of the list. If a Reaper then starts (e.g. in a process restarted after a crash), it never finds that worker's in-flight job, which stays processing forever.

A Reaper that was already running when the job started isn't affected: it has a timer for the job from the processing event and recovers it. The problem is the startup scan, which is exactly what recovers jobs after a restart.

Reproduction

@platformatic/[email protected] (also on main), Node 26. A handler that never settles keeps the job in the worker's processing list, standing in for a worker process that died mid-job:

import { MemoryStorage, Queue, Reaper } from '@platformatic/job-queue'
import { setTimeout as sleep } from 'node:timers/promises'

const visibilityTimeout = 1000
const storage = new MemoryStorage()

// The handler never settles: the job stays in the worker's processing list,
// as it would if the worker process had died mid-job.
let jobStarted
const started = new Promise(resolve => { jobStarted = resolve })
const worker = new Queue({ storage, visibilityTimeout, workerId: 'worker-1' })
worker.execute(() => { jobStarted(); return new Promise(() => {}) })
await worker.start()

console.log('workers right after start:', await storage.getWorkers())
await sleep(visibilityTimeout * 2 + 500)
console.log('workers 2.5s later (worker still running):', await storage.getWorkers())

await worker.enqueue('job-1', {})
await started

// A Reaper that starts now (as it would in a restarted process) relies on its
// startup scan to find the in-flight job.
const reaper = new Reaper({ storage, visibilityTimeout })
reaper.on('stalled', id => console.log('recovered', id))
await reaper.start()

await sleep(visibilityTimeout * 3)
console.log('job-1 state:', (await worker.getStatus('job-1'))?.state)

await reaper.stop()
await worker.stop()
await storage.disconnect()

Output:

workers right after start: [ 'worker-1' ]
workers 2.5s later (worker still running): []
job-1 state: processing

Without the sleep(visibilityTimeout * 2 + 500), the same script prints recovered job-1.

I saw the same expiry with a running consumer on RedisStorage and PgStorage (listed right after start(), gone 2.5s later with visibilityTimeout: 1000). On Redis the TTL is on the whole workers hash, so another worker registering later extends it.

Cause

Consumer.start() calls registerWorker(workerId, visibilityTimeout * 2) once (consumer.ts#L110). Nothing calls refreshWorker(), which every storage implements. The Reaper's startup scan (#checkStalledJobs) only walks getWorkers().

Fix

Two parts are needed:

1. Consumers refresh their registration while running, e.g. every visibilityTimeout / 2, and stop in stop(). This keeps live workers visible.
2. The startup scan shouldn't depend on live registrations at all. A worker that crashed stops refreshing, so its registration still expires, and if the restart happens after that, its jobs are still missed. The scan needs a way to enumerate processing lists regardless of registration (and treat unregistered owners as dead).

I'm happy to send a PR for this.