PSv2: Async jobs hang forever when NATS tasks exhaust max_deliver without posting results

[mihow]

Summary

When an external ML worker pulls tasks from a NATS-backed async job but fails to post results back, the tasks silently die in NATS after exhausting max_deliver retries. Django is never notified, no error is logged on the job record, and the job remains in STARTED status indefinitely.

Reproduction

1. Create an async_api job with N images
2. Have a worker pull tasks via GET /jobs/{id}/tasks/
3. Worker fails to process or crashes — never calls POST /jobs/{id}/result/
4. Wait for NATS ack_wait (30s default) × max_deliver (5) = ~2.5 minutes per message
5. All messages become dead in NATS
6. Job stays STARTED forever with 0 errors logged

Observed Behavior

• NATS consumer state: num_pending=0, num_ack_pending=0, num_redelivered=756
• Job progress: 168/925 processed, 757 remaining, 0 failed
• Job logs: empty (no errors, no warnings)
• Job status: STARTED (never transitions to FAILURE)

Root Cause

The result-processing path (process_nats_pipeline_result) is the only place that updates job progress and logs errors. When a worker pulls a task but never posts results, this code path is never invoked. NATS handles retries internally and eventually drops the message — but there is no callback, webhook, or polling mechanism for Django to detect that messages have been permanently dropped.

The _fail_job() function added in #1162 only triggers when Redis state is missing during result processing. It does not cover the case where result processing never happens at all.

Proposed Solution

Add a stale consumer detection mechanism. Two possible approaches:

Option A: Check inside the /tasks/ endpoint

When reserve_tasks() returns an empty list, check the NATS consumer state. If num_pending == 0 and num_ack_pending == 0 but the job still has remaining images (from Redis or the Job progress), mark the job as FAILURE with a descriptive error message.

Pros: Runs naturally as workers poll, no extra infrastructure.
Cons: Requires a worker to keep polling; if all workers stop, the check never runs.

Option B: Periodic Celery beat task

Add a beat task that runs every few minutes, queries all STARTED async_api jobs, checks their NATS consumer state, and fails any job where the consumer is exhausted but progress is incomplete.

Pros: Catches stalled jobs even if no workers are polling. Can also detect jobs where the NATS stream was deleted (e.g., container restart with ephemeral storage).
Cons: Adds a periodic task and requires NATS connectivity from the beat worker.

Option C: Both

Use Option A for fast detection during active polling, and Option B as a safety net.

Additional Context

• Related PR: Support ML async job cancellation, fail jobs on redis errors #1162 (async job cancellation + Redis error handling)
• NATS consumer config: max_deliver=5, ack_wait=30s (configurable via NATS_TASK_TTR)
• The job's dispatch_mode is async_api and pipeline is set
• JetStream storage is ephemeral (/tmp/nats/jetstream), so a NATS container restart also causes silent data loss — the beat task (Option B) would catch this too

Acceptance Criteria

• A job whose NATS tasks have all been exhausted (dead) is detected and marked as FAILURE
• An error message is logged on the job record explaining that tasks were dropped
• The detection works even if no external workers are actively polling
• Existing tests pass; new test covers the dead-message scenario

[mihow]

Revisiting this one in a troubleshooting session today and wanted to add concrete evidence from a production incident plus a note on the existing DLQ hook that appears to be wired only halfway.

Production evidence from today

We cancelled 11 async_api jobs whose NATS consumers had exhausted max_deliver=5 with no result posts. NATS state across the group looked like:

Metric	Typical value
num_pending	0
num_ack_pending	0
num_redelivered	100–2266
delivered.consumer_seq	488 → 23,580
Stream msgs	0 (retention-expired) or 434 (still within 24h max_age)
Django Job.status	STARTED, PENDING, CREATED, CANCELING, or REVOKED — never FAILURE

One consumer had a delivered.consumer_seq of 23,580 — we had roughly 100 other inert job streams sitting around with exhausted consumers. GPU workers were cycling over the 11 still-Django-active ones in a tight loop (Processing job X → empty NATS fetch → next job → repeat every ~5s), burning CPU and producing logs that looked identical to healthy polling.

Cause of the exhaustion in this particular incident: a cascade of worker OOMs during a memory-pressure window a few days ago. Workers pulled tasks, crashed mid-batch before acking, ack_wait=30s × max_deliver=5 cycled through in ~2.5 minutes per message, then the consumer was dead. Django never learned.

The memory issue that triggered the cascade is tracked separately (tuning work at RolnickLab/ami-data-companion#138 / RolnickLab/ami-data-companion#139). This ticket's bug is orthogonal: even if nothing ever OOMs again, any worker crash, network partition, or upstream bug that causes N acks to be lost will leave a Django job stuck STARTED forever with no alert.

The DLQ advisory hook exists but isn't called by anything in a loop

nats_queue.py:fetch_dead_letter_ids (introduced in #1175) already subscribes to $JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.> on the shared advisory stream and correctly resolves each event to an image_id. It's just never called from a background task.

The only caller today is python manage.py check_dead_letter_queue <job_id> — a human-invoked CLI. No celerybeat task, no long-running subscriber, no periodic sweep. So the infrastructure is built but dormant; max_deliver advisories accumulate in the advisory stream (up to its 24h retention) and nobody reads them.

This feels like ~half a day of wiring to close the loop, not a new system.

Why state-polling (num_pending == 0 && num_ack_pending == 0) is a poor substitute

Option A in the original issue description floats the idea of checking consumer state inside the /tasks/ endpoint when reserve_tasks() returns empty. That's tempting but pending=0 && ack_pending=0 is true in several non-failure states:

Scenario	pending	ack_pending	Actual state
Job completed successfully, everything acked	0	0	SUCCESS
Dead consumer (all msgs past max_deliver)	0	0	FAILURE
Freshly created consumer, nothing published yet	0	0	setup window
Worker just pulled everything, briefly hasn't processed	0	0	healthy, in flight
Quiescent between batches	0	0	healthy idle

To disambiguate you'd need to additionally check ack_floor.stream_seq < stream.last_seq (there are un-acked messages above ack_floor) AND verify the job has been STARTED longer than some initial-setup window. Even then, it's a poll-based heuristic with a detection lag.

By contrast, a max_deliver advisory is unambiguous the moment it fires — it names a specific message seq that has been delivered N times and will never be delivered again.

Proposed fix

Two components, both small, both using mechanisms that already exist:

1. Wire the advisory listener into a background loop

A celerybeat task (every 30s) or a dedicated long-running subscriber that calls TaskQueueManager.fetch_dead_letter_ids() on the shared advisory stream, and for each event:

• Records the image as permanently failed in the job's progress
• Increments a "dead letter" count for the job
• If dead_letter_count >= total_queued_images → job.update_status(FAILURE) + cleanup_async_job_if_needed(job)

This is push-based (scales with failure rate, not job count) and catches the per-message exhaustion the moment it happens.

2. Add a job.last_worker_activity_at heartbeat + stale sweep

A timestamp updated on every successful task ack path, plus an hourly celerybeat sweep that marks jobs FAILURE if they've been STARTED for > N minutes with no activity. This catches the other failure mode: workers silently dying in a way that produces so many advisories at once that they overflow the 24h advisory stream, or NATS itself being unreachable so the advisories never arrive.

Belt + suspenders: (1) catches per-message max_deliver exhaustion; (2) catches silent worker death / NATS outages. The two failure modes need different detection and each alone isn't enough.

#1025 ("Add periodic status check for incomplete jobs") is the right home for (2). This ticket is the right home for (1), and specifically for wiring the existing DLQ hook from #1175 into a loop.

Related work

• Add periodic status check for incomplete jobs  #1025 — periodic status check for incomplete jobs (the heartbeat side)
• Fail fast on NATS errors #1174 — fail fast on NATS unreachable (the transport-error side)
• Stop and fail jobs where the ongoing error rate exceeds a threshold #1176 — stop and fail jobs where ongoing error rate exceeds a threshold
• fix: PSv2 - Tasks are queued, worker sees job but no tasks #1123 — symptomatic report from February of the same "worker sees job but no tasks" behavior
• Support for NATS dead letter queue #1175 — closed PR that added the fetch_dead_letter_ids infrastructure