Stalled recovery

A job is stalled when a worker took it — it sits in active — but failed to complete it or renew its lock before the lock's PX TTL expired. That happens when the worker process crashes, the machine loses power, a network failure blocks lock renewal, or the job processor blocks without yielding. EchoMQ.StalledChecker finds such jobs and recovers them.

The checker is a GenServer driving one timer:

# default stalled_interval = 30000 ms
timer_ref = Process.send_after(self(), :check_stalled, stalled_interval)

Its public surface is start_link/1, check(connection, queue, opts \\ []) (arity 3, a manual trigger returning {:ok, %{recovered: n, failed: n}}), and job_stalled?(connection, queue, job_id, opts \\ []) (arity 4, a single-job probe).

On each :check_stalled, do_check lists the active jobs and probes their locks. The stall signal is the absence of the lock key for a job still in active. Pick an active job below to read the probe and its verdict.

defp do_check(connection, ctx, max_stalled_count) do
  case RedisConnection.command(connection, ["LRANGE", Keys.active(ctx), 0, -1]) do
    {:ok, []}      -> {:ok, %{recovered: 0, failed: 0}}
    {:ok, job_ids} -> check_jobs_stalled(connection, ctx, job_ids, max_stalled_count)
  end
end

# one EXISTS per active id, run as a single pipeline; exists == 0 means no lock
stalled_jobs =
  Enum.zip(job_ids, results)
  |> Enum.filter(fn {_id, exists} -> exists == 0 end)
  |> Enum.map(fn {id, _} -> id end)

EchoMQ uses two-phase detection so it never fails a job on a single bad reading:

Mark phase — a job with no valid lock is moved into the stalled SET. Recover phase — on the next check, a job still in the stalled set is either requeued (→ wait) or moved to failed, decided by max_stalled_count. The two phases prevent a false positive from a transient timing window: a job that briefly looked lockless but is in fact healthy clears itself on the next pass instead of being failed.

The recovery itself is moveStalledJobsToWait-8.lua — the -8 is the numkeys. The canonical wrapper EchoMQ.Scripts.move_stalled_jobs_to_wait/4 passes eight keys:

def move_stalled_jobs_to_wait(conn, ctx, max_stalled_count, opts \\ []) do
  keys = [
    Keys.stalled(ctx), Keys.wait(ctx), Keys.active(ctx), Keys.failed(ctx),
    Keys.stalled_check(ctx), Keys.meta(ctx), Keys.paused(ctx), Keys.marker(ctx)
  ]                                          -- eight keys → moveStalledJobsToWait-8.lua
  ...
end

The recovery result the checker returns is {:ok, %{recovered: count, failed: count}} — recovered is how many jobs were requeued to wait, failed how many had exhausted their stall count and were moved to failed. The checker emits telemetry for each: :stalled_recovered and :stalled_failed.

The default max_stalled_count is 1 because a job that stalls more than once usually signals a real bug — repeated crashes on specific job data, resource exhaustion, an external-service failure. So the default fails the job rather than looping forever; raising it is generally not recommended. Investigate why the job stalls instead.

The stall signal — the absence of the lock key for a job still in active (L1) — and the recovery script moveStalledJobsToWait-8.lua (L2) are frozen and shared, the same in every runtime. The Elixir checker — LRANGE active, a pipelined EXISTS-lock probe, the recovery script, all on one GenServer timer — is the Elixir runtime's own L3/L4.

The two-phase recovery sweep and the Redis commands the checker issues.