jonnify·redis /redis-patterns/queues/stalled-recovery/two-phase-mark-recover

R3.03.2 · Stalled recovery — dive

Two-phase mark/recover — count the stalls, bound the recovery.

Detection finds a job whose lease lapsed. Recovery has a harder decision: should this job go back in line, or has it stalled so many times that returning it again only re-poisons the queue? The mark layer counts the stalls and bounds them.

Grounding: echo/apps/echo_mq/lib/echo_mq/stalled.ex · EchoMQ.Stalled.check/3 · the per-job stalled field (HINCRBY), max_stalled the dead-letter threshold, the emq:{q}:dead morgue.

One increment, one branch. Select a job by its stall count to read where the sweep sends it against max_stalled = 1. Below the threshold recover; at it dead-letter.
HINCRBY stalledin-script count recoverpending / lane dead-letteremq:{q}:dead 0 → 1 1 → 2 grouped
readout · Select a job. The sweep increments stalled in the script, then recovers below max_stalled or dead-letters at it.
§1

The reaper recovers; the sweep counts

jobs.ex · stalled.ex

EchoMQ.Jobs.reap/2 is the unconditional layer — it returns any expired-lease job to pending once, no count, the at-least-once floor. EchoMQ.Stalled.check/3 is the count-thresholded layer that rides on top. Each pass it examines the same expired prefix, but for each job it increments a per-job stalled field on the row (HINCRBY ... 'stalled' 1) and branches on the result against max_stalled (default 1).

The stalled field is the mark. It is not a separate set or a separate key — it is a field on the job's row, so marking a job costs nothing the row did not already hold. A job's stall history travels with the job, and a recovered job carries its count into the next claim.

echo/apps/echo_mq/lib/echo_mq/stalled.ex · @sweep_stalled, the mark-and-branchverbatim
local st = redis.call('HINCRBY', jk, 'stalled', 1)   ← this job's stall count, this pass
if st >= maxst then
  redis.call('HSET', jk, 'state', 'dead')
  redis.call('HSET', jk, 'last_error', 'stalled')
  redis.call('ZADD', KEYS[3], 0, id)                 ← emq:{q}:dead · dead-lettered
  redis.call('HINCRBY', p .. 'metrics:failed', 'count', 1)
else
  redis.call('ZADD', KEYS[2], 0, id)                 ← emq:{q}:pending · recovered
  redis.call('HSET', jk, 'state', 'pending')
end

The reaper is the floor: every dead worker's job comes back. The sweep is the cap: a job that keeps stalling is counted, and at the threshold it is given up rather than recovered forever.

§2

Recover below, dead-letter at

stalled.ex · :max_stalled

The branch is the two phases in one decision.

  • Below max_stalled — recover. ZADD emq:{q}:pending, HSET state pending. The job goes back to be claimed again. A grouped job (a fair lane) recovers into its lane, emq:{q}:g:<group>:pending, mirroring the reaper's group branch, so lane fairness is preserved across a recovery.
  • At or above max_stalled — dead-letter. HSET state dead, ZADD emq:{q}:dead, and the metrics:failed counter ticks. The job stops cycling; it is now a record in the morgue rather than a member of the working set.

max_stalled is the budget. Set to 1, a job is recovered once and dead-lettered on its second stall — the strict default. Raise it for a workload where a transient stall (a slow downstream, a brief pause) is expected and a job deserves several recoveries before it is given up. The threshold separates the worker died (recover, it was not the job's fault) from this job kills workers (dead-letter, the job is poison).

The budget at three settings. Select a max_stalled value to read how many recoveries a job gets before the morgue.
budget 11 recovery, then dead budget 32 recoveries, then dead budget 100many recoveries; a poison job cycles
readout · Select a budget. A job is recovered up to max_stalled - 1 times, then dead-lettered on the stall that reaches the threshold.

Notes on ValkeyHINCRBY increments the per-job stalled field on the row in one step — the field travels with the job, no separate counter key — valkey.io/commands/hincrby.

§3

Idempotent across a restart

stalled.ex · :transient child

The sweep is idempotent over the active set. If the sweep process crashes and its :transient child restarts, the next pass re-reads ZRANGEBYSCORE active -inf now and re-applies the branch — a job already recovered has left active, so it is not examined again; a job not yet recovered is recovered on the re-sweep. No double-recovery, no loss across a restart, because the move reads the live set, not a journal of what the last pass intended to do.

detect

A job's lease lapsed; the reaper finds it on active. The pattern guarantees at-least-once delivery — the job must come back.

then bound the recovery

EchoMQ.Stalled increments the job's stalled field, recovers it below max_stalled into pending (or its lane), and dead-letters it at or above the threshold into emq:{q}:dead — a poison job is bounded, not recovered forever.

A plain reaper recovers a poison job forever; the stall count turns “recover” into “recover a bounded number of times, then retire.”

A job in emq:{q}:dead is a durable record of a failure, not a working-set member. When the bus trims the stream that carries the queue's history, EchoStore.StreamArchive folds the trimmed segments into the durable Graft floor on CubDB at a reserved high page range and on to Tigris — deep history without resident memory, queryable beside the live tail. The morgue is the dial's first detent, held in Valkey; the archive is the next.

/echo-persistence

The durability dial, in full

Where a dead-lettered job's history lands: EchoStore.StreamArchive into the Graft floor on CubDB, on to Tigris — the four-tier ladder taught end to end.

 /echomq/queue

The state machine the sweep moves

The Queue pillar teaches the pending / active / dead sets and the inline Lua transitions, the sweep among them, in depth.
§4

References

sources · related

Sources

Related in this course