Cancellation & checkpoints

Cancellation is cooperative and worker-side

# 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 one else's.
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 cleanly there — else continues.
def check!(token) do
  case check(token) do
    :ok -> :ok
    {:cancelled, reason} -> raise Cancelled, reason: reason
  end
end

A cooperative cancel is a message, not a kill. The handler decides where it is safe to stop; the token pin keeps the cancel its own.

extend_lock checkpoints the lease

EchoMQ.Jobs @extend_lock — 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_lock, keys, argv) do
    {:ok, 1} -> :ok                       # the lease re-scored to now + lease
    {:ok, -1} -> {:error, :gone}          # no such row
    {:error, {:server, "EMQSTALE" <> _}} -> {:error, :stale}  # not the current holder
    other -> other
  end
end

@extend_lock — the body

-- the @extend_lock script — re-score one lease, token-fenced, on the server clock
-- The token fence: the row's attempts IS the fencing token. Read it first.
local att = redis.call('HGET', KEYS[2], 'attempts')
if not att then return -1 end          -- no row: gone
if att ~= ARGV[2] then
  return redis.error_reply('EMQSTALE extend token mismatch')  -- a stale holder may not extend
end
-- The re-score: read the SERVER clock and set the active member to now + lease.
-- 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 checkpoint is a re-score, not a new lock. The lease is the active score; moving it ahead of now is all it takes to keep an alive handler from being reaped, and the token keeps it the holder's alone.

Stalled recovery is count-thresholded, on top of reap

EchoMQ.Stalled @sweep_stalled — the handle

# echo_mq — EchoMQ.Stalled — one sweep over the active set
# KEYS = [active, pending, dead]; ARGV = [the queue base, max_stalled, limit].
def check(conn, queue, opts \\ []) do
  max_stalled = Keyword.get(opts, :max_stalled, 1)
  limit = Keyword.get(opts, :limit, 100)

  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

@sweep_stalled — the body (flat path)

-- the @sweep_stalled script — count stalls, recover or dead-letter
local p = ARGV[1]                        -- the queue base root
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)
local exp = redis.call('ZRANGEBYSCORE', KEYS[1], '-inf', now, 'LIMIT', 0, lim)
local recovered = {}
local dead = {}
for _, id in ipairs(exp) do
  redis.call('ZREM', KEYS[1], id)        -- leave the active set
  local jk = p .. 'job:' .. id
  local st = redis.call('HINCRBY', jk, 'stalled', 1)  -- count this stall
  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 to pending
    redis.call('ZADD', KEYS[2], 0, id)
    redis.call('HSET', jk, 'state', 'pending')
    table.insert(recovered, id)
  end
end
return {recovered, dead}

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 is a poison job that earns the morgue.

Pattern & implementation

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.

Recap — control over the worker in hand

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

References