Orphan recovery — reclaiming messages left in-flight by a dead worker

The failure this exists to prevent

A worker that dies mid-task (deploy, OOM, SIGKILL) never acks or nacks the message it claimed. The tw_messages row stays state='running' with claimed_by pointing at a process that no longer exists.

That is worse than one lost task, because per-topic concurrency is computed from exactly those rows:

SELECT count(*) FROM tw_messages WHERE topic = ? AND state = 'running'

claim() gates on that count, so every orphan permanently burns one concurrency slot. Enough of them and the topic stops dispatching entirely. Restarting does not help: the fresh worker recomputes the limit from the same stale rows and boots already saturated.

Measured on brain (brain#72, #12): two orphans against a brain: 2 limit wedged the topic for 11 days, 287 messages deep, 52 user shares never processed. Nothing was red — the workers heartbeated, the queue looked "busy", and the submitting side reported healthy submissions the whole time.

The old watchdog.cleanup_stuck_tasks cannot fix this. It walks fp_task_reports — the history table — and marks dead-thread reports DEAD. The concurrency limit does not read history. Clearing those rows changes nothing, which is why the brain wedge survived both a restart and a report cleanup.

How the reaper works

Broker.reclaim_orphaned_messages() (implemented on PostgresBroker; a no-op on in-process brokers, whose queue dies with the process) finds running messages whose claimer has no fresh heartbeat and either requeues them (state='queued', claimed_by/claimed_at cleared) or, past orphan_max_attempts, dead-letters them.

It runs on the beat, in-process — schedule_loop calls it every orphan_reclaim_interval_seconds. Two reasons:

The beat is already leader-elected (#14), so one process reaps instead of every worker issuing the same UPDATE.
It must not be a queued task. The failure being repaired is a topic that cannot dispatch; a reaper that has to be dispatched to run could sit queued behind the very wedge it exists to clear.

The watchdog.reclaim_orphaned_messages task is still registered so an operator can trigger a pass from the UI, but it deliberately carries no schedule — the beat is the only automatic driver.

At-least-once: read this before enabling side-effectful tasks

Requeuing a message whose worker died mid-execution runs that task again, and the reaper cannot know how far the first attempt got. A task that had already charged a card or sent a mail will do it twice.

This is within the broker's existing contract — claim() increments attempts and nack(requeue=True) already re-delivers — but tasks with side effects that must not repeat have to be idempotent. The bound on the damage is orphan_max_attempts: past that many claims a message is dead-lettered instead of retried, so a poison task that kills whatever worker picks it up cannot loop forever taking each new worker with it.

Fail-closed guards

The reaper reclaims nothing, and logs at ERROR, when any of these holds:

Guard	Why
the live-worker set could not be read	a failed query must not read as "nobody is alive"
the live-worker set is empty	something alive is making the call, so an empty set means the liveness signal is broken — not that the fleet died and its work is free to requeue
the caller is not in the live set	the positive control: if the broker cannot see the process asking, heartbeats are not being written (or clocks disagree) and every worker looks dead
the message was claimed less than `orphan_claim_grace_seconds` ago	a worker that just claimed may not have written its first heartbeat yet; reaping here hands the same task to a second worker

Every skip increments fastpluggy_broker_orphan_reclaim_skipped_total. A reaper that fail-closes every pass is protecting nothing, and would otherwise look exactly like a reaper with nothing to do — hence the counter.

Locks are reclaimed in the same pass

The reaper also releases locks whose holder is gone (#18), under the same live-worker set, the same grace window and the same fail-closed guards.

Doing only the messages would be worse than doing neither. A worker that dies holding a lock_name never releases it — there is no TTL, and acquired_at was written but never read — so the requeued message would come back, fail to acquire a lock nobody could ever release, and be dropped again. One crash poisoned that lock name permanently, recoverable only by a manual force_release_lock.

The fail-closed rule matters even more here than for messages: releasing a lock that is genuinely held lets a second worker run the task the lock exists to serialise, which is the precise failure the lock was there to prevent. So an untrustworthy live-worker set releases nothing, exactly as it requeues nothing.

Applies to postgres and local. memory cannot orphan a lock (its holders die with it) and rabbitmq already expires a crashed holder's lock on a TTL and re-seeds the token. The count comes back as locks_released.

Settings

Setting	Default	Meaning
`orphan_reclaim_enabled`	`True`	run the reaper from the beat
`orphan_reclaim_interval_seconds`	`300`	how often the beat runs a pass
`orphan_reclaim_after_seconds`	`120`	heartbeat staleness before a claimer counts as dead
`orphan_claim_grace_seconds`	`60`	minimum age of a claim before it is eligible
`orphan_max_attempts`	`5`	claims before dead-lettering instead of requeueing

orphan_reclaim_after_seconds is deliberately not postgres_worker_ttl_seconds (24 h). That one governs when a dead worker's row is purged — bookkeeping, where a day is fine. An orphaned message burns a concurrency slot for as long as it sits there, so its threshold is measured in missed heartbeats, in the spirit of check_existing_beat's "2× the heartbeat interval".

Alerting

The 11-day outage was invisible, which is half the bug. What to watch:

# A topic wedging: the oldest in-flight claim only ever grows. A healthy topic's
# resets every time a task finishes, so this stays near the longest task runtime.
fastpluggy_broker_oldest_running_claim_seconds > 3600

# The reaper is refusing to run — orphans are NOT being recovered.
increase(fastpluggy_broker_orphan_reclaim_skipped_total[1h]) > 0

# Tasks are being re-run. Expected after a crash; a steady stream is not.
increase(fastpluggy_broker_orphans_reclaimed_total[1h]) > 0

# Poison tasks being retired.
increase(fastpluggy_broker_orphans_dead_lettered_total[1h]) > 0

fastpluggy_broker_oldest_running_claim_seconds is the one that would have caught brain on day one. It is deliberately independent of the reaper: it reads the same running rows the limit is computed from, so it rises whether the reaper is working, misconfigured, or fail-closed and silent. Note that a wedged topic and a merely saturated one are identical in every count — they differ only in age.

Counters are process-local (like the task telemetry counters) and are reported by whichever process serves the metrics endpoint, which in a FastPluggy app is also the one hosting the beat.

The catch-all: a schedule that stops producing (#23)

Everything above watches the broker. One rule watches the outcome instead, and it is the one to alert on first:

# A schedule that is triggered but produces no successful run.
fastpluggy_scheduled_task_since_success_seconds
  / fastpluggy_scheduled_task_interval_seconds > 3

Why it earns its place next to the claim-age gauge: it is blind to the cause, so it fires for all of them — an orphaned tw_locks row (#18), a task failing on every run, a wedged topic, a starved worker, a dead beat. The period is exported alongside the age precisely so the rule needs no per-task threshold.

It is not redundant with fastpluggy_schedule_overdue_seconds. That gauge is derived from last_attempt, which tasks/scheduler.py stamps before the submit — so it answers "is the beat beating?", and a task whose message is discarded on every tick keeps it pinned at 0. brain lost four schedules that way for 13 days (152 consecutive skipped runs of one of them) with that gauge reading healthy the whole time. Keep both: overdue catches a beat that stopped firing, this one catches firing that stopped producing.

Two deliberate limits, so they are not mistaken for bugs:

a schedule that has never succeeded emits no series at all — otherwise every fresh deploy would page while the task waits for its first turn. That leaves "registered but never ran once" uncovered here, by choice;
the success test is the report status, never the returned payload. A task may legitimately return {"skipped": "healthy"} and still have run.