Lifecycle controls · dive 02

Cancellation & checkpoints

The lease is a score; a checkpoint moves it; a cancel is a message a handler chooses to honour.

A claimed job is held by a worker for the length of its lease. Three controls act on that worker while it runs. Cancel it cooperatively — ask a long handler to stop its own work at a safe point, without a forced kill mid-transaction. Checkpoint its lease — re-score the active member to a fresh deadline so a long-but-alive handler is not reaped. Recover the ones that stalled — sweep for leases that lapsed without a checkpoint, recover them, and dead-letter the ones that keep stalling. The checkpoint and the sweep have native twins (EMQ.EXTEND, EMQ.STALLED.CHECK); the cancel never touches a script. Drag "now" on the right, checkpoint and reap, then read the three below.

Cooperative cancel extend_lock The stalled sweep Pattern & implementation References

The lease timeline · drag now, checkpoint, reap

now (seconds from claim): 0

now = 0s, deadline = 30s → alive (now ≤ deadline: a reap pass and the stalled sweep skip it)

Cancellation is cooperative and worker-side

EchoMQ.Cancel is a host-side primitive with no wire identity. The token is a plain make_ref/0; cancellation is a process message. new/0 mints a token. cancel/3 sends {:emq_cancel, token, reason} to a pid's mailbox. check/1 is a non-blocking receive after 0 that answers {:cancelled, reason} if a cancellation for this token is waiting, else :ok. check!/1 raises EchoMQ.Cancel.Cancelled for a checkpoint-style handler.

# echo_mq — EchoMQ.Cancel — a cooperative token, worker-side, no wire identity
# new/0 mints a token: a unique reference that names one cancellation.
def new, do: make_ref()

# …
# cancel/3 sends the cancel message to a pid's mailbox; reason is optional.
# The handler picks it up at its next check/1; nothing interrupts it here.
def cancel(pid, token, reason \\ nil) when is_pid(pid) do
  send(pid, {:emq_cancel, token, reason})
  :ok
end

# …
# check/1 is O(1): a receive after 0. The ^token pin matches only THIS token,
# so a handler catches its own cancellation and no other.
def check(token) do
  receive do
    {:emq_cancel, ^token, reason} -> {:cancelled, reason}
  after
    0 -> :ok
  end
end

# …
# check!/1 raises EchoMQ.Cancel.Cancelled when cancelled, so a handler that
# calls it at a safe point aborts there, and otherwise continues.
def check!(token) do
  case check(token) do
    :ok -> :ok
    {:cancelled, reason} -> raise Cancelled, reason: reason
  end
end

Cooperative means a handler that never checks completes normally — cancellation does not interrupt it. The ^token pin keeps a handler to its own cancellation. A long handler calls check!/1 at safe points and aborts where it is asked to, leaving no half-finished transaction. The token itself touches no key, no script and no set.

Cancel by id. request/4 is the verb that reaches a job by queue and id from anywhere on the node. It reads the row's state with one host-side HGET and dispatches on it, composing what already ships rather than adding a script.

# echo_mq — EchoMQ.Cancel
# request/4 cancels a job by queue + id from anywhere on the node: one host-side
# read of the row state (an HGET, not a transition), then a dispatch on it.
def request(conn, queue, job_id, reason \\ nil) do
  # the id is gated at the key builder, before the wire
  row = Keyspace.job_key(queue, job_id)

  case Connector.command(conn, ["HGET", row, "state"]) do
    {:ok, nil} ->
      {:error, :gone}

    # unclaimed: removed through Jobs.remove_job/4, which runs @remove_job2
    {:ok, state} when state in ["pending", "scheduled"] ->
      case Jobs.remove_job(conn, queue, job_id) do
        :ok -> {:ok, :removed}
        {:error, _} = err -> err
      end

    # held by a worker: a cancel event on emq:{q}:events, delivered at most once
    {:ok, "active"} ->
      case Events.publish(conn, queue, "cancel", job_id, reason: reason_field(reason)) do
        :ok -> {:ok, :signalled}
        other -> other
      end

    {:ok, _other_state} ->
      {:error, :not_cancellable}

    other ->
      other
  end
end

A pending or scheduled job is removed outright ({:ok, :removed}). An active job is only signalled ({:ok, :signalled}): a Consumer started with the :cancel option mints a token per claim, hands it to the handler, and forwards a matching cancel event to that token with cancel/3. The event rides the fire-and-forget events channel, so a publish with no live subscriber is lost: the signal is at most once. A handler that honours it answers the verdict {:cancel, reason}, and the Consumer settles the job through Jobs.complete/5 with the result "cancelled:" <> reason — terminal, never retried. A row in any other state answers {:error, :not_cancellable}; a missing row answers {:error, :gone}. There is no cancelled state on the wire: a cancelled job ends as a completion whose result carries the tag.

A cooperative cancel is a message, not a kill. The handler chooses where it is safe to stop; the token pin keeps the cancel its own, and request/4 reaches it by id.

extend_lock checkpoints the lease

The lease is the active-set score. extend_lock/5 runs @extend_lock2: read the row's attempts, fence the token (a stale token answers EMQSTALE), check that the id is still a member of the active set, read the server TIME, and ZADD the active set at now + lease for this id. It re-scores the member to a fresh deadline.

Beat one — the named handle. extend_lock/5 builds the active set and the row, passes the id, token and lease, and maps four replies.

EchoMQ.Jobs @extend_lock2 — the handle

# echo_mq — EchoMQ.Jobs
# extend_lock/5 re-scores the active member to a fresh deadline. KEYS = [the
# active set, the job row]; ARGV = [id, the attempts token, the lease ms].
def extend_lock(conn, queue, job_id, token, lease_ms)
    when is_integer(lease_ms) and lease_ms > 0 do
  keys = [Keyspace.queue_key(queue, "active"), Keyspace.job_key(queue, job_id)]
  argv = [job_id, Integer.to_string(token), Integer.to_string(lease_ms)]

  case Connector.eval(conn, @extend_lock2, keys, argv) do
    # the lease re-scored to now + lease
    {:ok, 1} -> :ok
    # no such row
    {:ok, -1} -> {:error, :gone}
    # the token holds, but the id has left the active set
    {:ok, -3} -> {:error, :not_active}
    # not the current holder
    {:error, {:server, "EMQSTALE" <> _}} -> {:error, :stale}
    other -> other
  end
end

Beat two — the script body. The checks run in a fixed order: the row, the token, then membership. So an id that is both stale and out of active answers EMQSTALE, never -3.

@extend_lock2 — the body

-- the @extend_lock2 script — KEYS[1] = active, KEYS[2] = the job row
-- ARGV[1] = id, ARGV[2] = the token, ARGV[3] = the lease ms
-- 1. the row: its attempts field is the fencing token; no row answers -1
local att = redis.call('HGET', KEYS[2], 'attempts')
if not att then return -1 end
-- 2. the token fence: a stale holder may not extend
if att ~= ARGV[2] then
  return redis.error_reply('EMQSTALE extend token mismatch')
end
-- 3. membership: an id no longer in active answers -3, so a reaped lease
-- cannot be brought back
if not redis.call('ZSCORE', KEYS[1], ARGV[1]) then return -3 end
-- 4. the re-score, on the SERVER clock. The lease is the active score, so
-- there is no separate lock string to refresh.
local t = redis.call('TIME')
local now = t[1] * 1000 + math.floor(t[2] / 1000)
redis.call('ZADD', KEYS[1], now + tonumber(ARGV[3]), ARGV[1])
return 1

A long-but-alive handler calls extend_lock/5 periodically so its deadline sits ahead of now, and a reap pass or a stall sweep skips it. Because the lease is the active score, there is no separate lock string to refresh: the checkpoint re-scores the one member that already is the lease. The fences decide who may move it. A stale token (the job was reclaimed and claimed again, so attempts moved on) answers {:error, :stale}. A lapsed lease that no pass has collected yet is still a member, so a late checkpoint still answers :ok. Once a reap pass has taken the id out of active, the token still matches but the membership check fails, and the checkpoint answers {:error, :not_active} instead of resurrecting a lease a reap pass has already handed back. Try it in the timeline above: drag past 30s, reap, then checkpoint.

Native twin · EMQ.EXTEND

# the scripted line — extend_lock/5 runs EchoMQ.Jobs @extend_lock2, EVALSHA-first on wire 4
EVALSHA <sha> 2 emq:{q}:active emq:{q}:job:<id> <id> <token> <lease_ms>
# the native line — the same tail, the sha replaced by the command name
EMQ.EXTEND    2 emq:{q}:active emq:{q}:job:<id> <id> <token> <lease_ms>

extend_lock/5 sends the native line through a wire: 6 connector, as a raw client on a module-loaded node can; over the default wire 4 it evaluates the script. Same keys, same argument order and, on the ordinary path, the same replies, checked in the same order: -1 for a missing row, EMQSTALE extend token mismatch for a stale holder, -3 for an id no longer in active, 1 for a re-score. A caller sees no difference in those four replies. The handler refuses two inputs the script accepts: a lease that is not a positive integer (ERR EMQ.EXTEND lease_ms must be a positive integer; the script runs tonumber, and only the Elixir guard checks it) and keys that do not match the queue's layout. On a re-score it also stamps the row's lease_epoch with the server's run id, the field the native stall sweep reads. Every check runs before the first write. The mechanism: Commands, not scripts.

A checkpoint is a re-score, not a new lock. The token keeps it the holder's alone, and the membership check refuses to revive a reaped lease.

Stalled recovery is count-thresholded, on top of reap

The reaper (Jobs.reap/2) returns any expired-lease job from active to pending once, with no count — crash recovery. EchoMQ.Stalled is the count-thresholded layer beside it. @sweep_stalled reads the expired members on the server clock, and for each one increments a per-job stalled field. A job below the max_stalled threshold is recovered — back to pending, or to its lane if it is grouped. A job at or above the threshold is dead-lettered.

Beat one — the named handle. The defaults are module attributes: a threshold of 1, at most 100 expired members per pass, and a 30-second interval for the opt-in periodic process. check/3 runs one sweep and answers {:ok, %{recovered: [id], dead: [id]}} — the ids returned to pending and the ids dead-lettered this pass.

EchoMQ.Stalled @sweep_stalled — the handle

# echo_mq — EchoMQ.Stalled — the defaults are module attributes
@default_max_stalled 1
@default_interval_ms 30_000
@default_limit 100

# …
# check/3 runs ONE sweep. KEYS = [active, pending, dead];
# ARGV = [the queue base, max_stalled, limit].
def check(conn, queue, opts \\ []) do
  max_stalled = Keyword.get(opts, :max_stalled, @default_max_stalled)
  limit = Keyword.get(opts, :limit, @default_limit)

  keys = [
    Keyspace.queue_key(queue, "active"),
    Keyspace.queue_key(queue, "pending"),
    Keyspace.queue_key(queue, "dead")
  ]

  argv = [
    Keyspace.queue_key(queue, ""),
    Integer.to_string(max_stalled),
    Integer.to_string(limit)
  ]

  case Connector.eval(conn, @sweep_stalled, keys, argv) do
    {:ok, [recovered, dead]} when is_list(recovered) and is_list(dead) ->
      {:ok, %{recovered: recovered, dead: dead}}

    other ->
      other
  end
end

Beat two — the script body. Per expired id: leave active, count the stall, then fork on the threshold. A grouped job also gives back its lane's active slot; its return to the lane and the ring is elided here, and the flat path is shown in full.

@sweep_stalled — the body

-- the @sweep_stalled script — KEYS[1] = active, KEYS[2] = pending, KEYS[3] = dead
-- ARGV[1] = the queue base, ARGV[2] = max_stalled, ARGV[3] = limit
local p = ARGV[1]
local maxst = tonumber(ARGV[2])
local lim = tonumber(ARGV[3])
local t = redis.call('TIME')
local now = t[1] * 1000 + math.floor(t[2] / 1000)
-- every active member whose lease has lapsed (score at or before now), up to the limit
local exp = redis.call('ZRANGEBYSCORE', KEYS[1], '-inf', now, 'LIMIT', 0, lim)
local recovered = {}
local dead = {}
for _, id in ipairs(exp) do
  -- leave the active set
  redis.call('ZREM', KEYS[1], id)
  local jk = p .. 'job:' .. id
  local g = redis.call('HGET', jk, 'group')
  -- count this stall; on an id with no row, HINCRBY creates one
  local st = redis.call('HINCRBY', jk, 'stalled', 1)
  if g then
    -- a grouped job gives back its lane's active slot
    local act = redis.call('HINCRBY', p .. 'gactive', g, -1)
    if act <= 0 then redis.call('HDEL', p .. 'gactive', g) end
  end
  if st >= maxst then
    -- at or above the threshold: dead-letter
    redis.call('HSET', jk, 'state', 'dead')
    redis.call('HSET', jk, 'last_error', 'stalled')
    redis.call('ZADD', KEYS[3], 0, id)
    redis.call('HINCRBY', p .. 'metrics:failed', 'count', 1)
    table.insert(dead, id)
  else
    -- below the threshold: recover
    if g then
      -- … a grouped job rejoins its lane, and the ring when the lane may run
    else
      redis.call('ZADD', KEYS[2], 0, id)
    end
    redis.call('HSET', jk, 'state', 'pending')
    table.insert(recovered, id)
  end
end
return {recovered, dead}

Who runs it. A sweep runs only when a client process calls check/3. A Consumer started with recovery: :stalled runs it as the recovery leg of its beat, in place of the reap: one recovery mechanism per beat, chosen at start. The opt-in EchoMQ.Stalled process runs it on a host timer, every interval_ms. And an operator calls it directly for a one-shot recovery.

# echo_mq — EchoMQ.Stalled — the opt-in process: a host timer that calls check/3
def handle_info(:sweep, s) do
  _ = check(s.conn, s.queue, s.opts)
  {:noreply, arm(s)}
end

defp arm(s) do
  # the next sweep, interval_ms from now, on the BEAM's timer
  Process.send_after(self(), :sweep, s.interval_ms)
  s
end

The stalled-sweep ladder · step the passes

max_stalled: 1

stalled = 0, max_stalled = 1 — run a sweep pass to lapse the lease and count the stall

With max_stalled above 1, a job that stalls once is recovered, and a job that keeps stalling — a poison job that hangs every worker that picks it up — is dead-lettered once its count reaches the threshold. At the default of 1, the first counted stall dead-letters: the reaper is the recover-once path, the stalled sweep the give-up path. The count is the difference between a transient slow handler and a job that should die.

Native twin · EMQ.STALLED.CHECK

# the scripted line — check/3 runs EchoMQ.Stalled @sweep_stalled, EVALSHA-first on wire 4
EVALSHA <sha> 3 emq:{q}:active emq:{q}:pending emq:{q}:dead emq:{q}: <max_stalled> <limit>
# the native line — the same tail, the sha replaced by the command name
EMQ.STALLED.CHECK 3 emq:{q}:active emq:{q}:pending emq:{q}:dead emq:{q}: <max_stalled> <limit>

Same keys and arguments, and the reply is the script's pair, [[recovered], [dead]]. A caller can see four differences. A lease whose row has no lease_epoch, or one stamped by an earlier server run, is exempt: it is recovered without counting and listed with the recovered, so a lease from before a restart never counts as a stall; the native claim and extend stamp that field with the server's run id. A ghost — an active id with no row — is only removed from active and appears in neither list, and no row is written for it; the script's HINCRBY creates a skeleton row, which at the default threshold is dead-lettered and listed under dead. A grouped expired row refuses the whole call with EMQSTATE stalled lane not native, where the script returns it to its lane. And max_stalled and limit are validated (ERR EMQ.STALLED.CHECK max_stalled must be a positive integer). Every read and check comes before the first write. The restart exemption is taught in One transition, two executors.

Worked example. A worker claims a job with a 30-second lease, runs for 90 seconds, and checkpoints with extend_lock/5 every 20 seconds. Its active deadline always sits ahead of now, so neither the reaper nor the stalled sweep touches it. A second worker claims a different job and crashes silently. Its lease lapses; the scripted sweep increments its stalled field to 1, at the default threshold, so it is dead-lettered with last_error = "stalled".

Recovery counts how many times a job has stalled before it gives up on it. One stall is a slow worker; repeated stalls on the same job mark a poison job for the dead set.

Pattern & implementation

the pattern (Redis Patterns Applied)

A reliable queue checkpoints a worker's lease and reclaims abandoned work on a recovery pass. Time, delay & priority and Reliable queues teach the lease-and-recover machinery over Redis.

the implementation (echo_mq)

The lease is the active score, extend_lock/5 re-scores it under a token fence and a membership check, and EchoMQ.Stalled is the count-thresholded sweep beside the reaper — recover below a threshold, dead-letter at it. The cancel is host-side, off the wire entirely.

The pattern says: checkpoint the lease, reclaim the abandoned, give up on the poison. The implementation is extend_lock, the reaper, and the count-thresholded Stalled sweep.

The durable floor

When the sweep gives up on a poison job it dead-letters it, kept for inspection in memory until an operator reprocesses, removes or trims it (trimming deletes; it does not archive). What outlives memory is a declared stream, such as the queue's completions stream a recording Consumer (:record) writes: when a deployment declares that stream to EchoStore.StreamArchive.Driver, the driver folds each slice into the durable EchoStore.Graft floor before it trims it: CubDB, and Tigris when a remote is configured. That floor is taught in full in Echo Persistence.

Recap — control over the worker in hand

The lease is a score; a checkpoint moves it (extend_lock/5 and @extend_lock2, token-fenced and membership-checked on the server clock, with EMQ.EXTEND as its native twin); a cancel is a message a handler chooses to honour (EchoMQ.Cancel, off the wire, reached by id through request/4); recovery counts how many times a job has stalled before it dead-letters it (EchoMQ.Stalled and EMQ.STALLED.CHECK, which exempts leases from before a restart). Control over the worker in hand is three small, fenced moves. The next dive widens the lens to the whole queue and one job's row — the operator's runbook.

Next in lifecycle controls: The operator plane — EchoMQ.Admin pause / resume / drain / obliterate, and the per-job verbs on EchoMQ.Jobs.

References

Where the cancel, the lease checkpoint and the stalled sweep are grounded, and the commands their bodies use.

Sources

Valkey — ZADD — the active-set re-score that is the lease checkpoint.
Valkey — ZSCORE — the membership check a checkpoint runs before it re-scores.
Valkey — HINCRBY — the per-job stalled count the sweep increments.
Valkey — ZRANGEBYSCORE — the expired-lease read the sweep runs on the server clock.
Redis — EVALSHA — the load-once dispatch the checkpoint and sweep run by SHA.
Valkey — Documentation — the substrate of record EchoMQ is backed by.

Related in this course

Lifecycle controls — the module this dive belongs to.
The Queue — the state machine, the active set whose score is the lease.
Commands, not scripts — how a native twin checks before it writes.
One transition, two executors — the lease epoch and the restart exemption.
redis-patterns · Time, delay & priority — the lease/recover family.
Echo Persistence — the durable floor an archived stream folds into.