Hacktoberfest 2026: le issue che i maintainer hanno segnato per ottobre, aperte e adatte ai principianti. Sfoglia le issue Hacktoberfest

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

Aperta
  • Qualcosa non va?
#51 0 commenti 0 reazioni 0 assegnatari Vedi su GitHub

Nessuno ha ancora preso questa issue.

Valutazione

Difficoltà
4/5
Tempo stimato
3-5 giorni
Idoneità per principianti
48/100
Tipo di issue
Bug
Chiarezza
Abbastanza chiara
Stato di attività
Attiva
Stack tecnologico
node.js, postgresql, redis, typescript
Ambito
backend, databases, distributed-systems

Direzione di ricerca

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.

Scritto dal modello di indicizzazione a partire dal testo della issue.

Descrizione

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.

Lingua principale
TypeScript
Stelle
371
Fork
11
Nessuna PR unita negli ultimi 30g

Preparare l'ambiente

  • Include un Dockerfile o un file Docker Compose
  • Nessun modello di pull request
  • Nessuna guida per i contributori

Come iniziare

  1. Leggi tutta la issue e poi la guida ai contributi del progetto.
  2. Commenta sulla issue per dire che te ne occupi tu — evita che due persone facciano lo stesso lavoro.
  3. Fai un fork del repository e lavora su un branch.
  4. Apri una pull request che faccia riferimento al numero della issue.

Altre issue di platformatic/job-queue

Tutte le issue di platformatic/job-queue

Issue simili

Altre issue su TypeScript

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.