Lifecycle controls · dive 03

The operator plane

An operator's runbook drives a live queue with real transitions: over the whole queue, and over one job's row.

Two scopes. Queue scope (EchoMQ.Admin) pauses and resumes claiming on the entire queue, drains the pending backlog, and obliterates a paused queue down to its keyspace footprint. Job scope (EchoMQ.Jobs) reaches into one row to replace its payload, write progress, append logs, remove it, or reprocess a dead one. Every verb is a fenced transition that either succeeds or refuses cleanly with a typed reason. The admin verbs run only as scripts; of the per-job verbs, reprocess has a native twin (EMQ.REPROCESS). Toggle the gate on the right, then run a claim.

Pause is a gate Drain the backlog Obliterate, bounded The per-job verbs Reprocess Pattern & implementation References

The separate gate · toggle pause, then claim

queue running · a claim serves the head of pending — run a claim

Pause is a separate gate, so the claim script never changes

EchoMQ.Admin.pause/2 runs @pause: HSET the paused field on emq:{q}:meta. resume/2 runs @resume: HDEL that field. The pause is a field on the queue's meta hash — not a move on any set.

EchoMQ.Admin @pause & @resume — the handle

# echo_mq — EchoMQ.Admin — a claim gate over the WHOLE queue, the meta field form
# pause/2 sets the paused field; the claim path reads it first and answers :empty.
def pause(conn, queue) do
  case Connector.eval(conn, @pause, [Keyspace.queue_key(queue, "meta")], []) do
    {:ok, 1} -> :ok
    other -> other
  end
end

# …
# resume/2 clears the field, so a later claim serves the head of pending again.
def resume(conn, queue) do
  case Connector.eval(conn, @resume, [Keyspace.queue_key(queue, "meta")], []) do
    {:ok, 1} -> :ok
    other -> other
  end
end

@pause & @resume — the bodies

-- the @pause script — KEYS[1] is emq:{q}:meta. No set moves: the backlog is untouched.
redis.call('HSET', KEYS[1], 'paused', '1')
return 1
-- …
-- the @resume script — clear the field; claiming resumes at the head of pending
redis.call('HDEL', KEYS[1], 'paused')
return 1

The claim path reads that field first and answers :empty when it is set, even with a non-empty pending set; the pending backlog is untouched. This is the separate-gate form: Jobs.claim/3 calls paused?/2 — one HGET in its own round trip — before it evaluates @claim2, and Lanes.claim/3 reads the same field before @gclaim2, so a queue-wide pause gates both the flat and the grouped claim.

# echo_mq — EchoMQ.Jobs — the gate lives in the claim function, above the script
def claim(conn, queue, lease_ms) when is_integer(lease_ms) and lease_ms > 0 do
  # the pause read comes first, in its own round trip
  if paused?(conn, queue) do
    :empty
  else
    keys = [
      Keyspace.queue_key(queue, "pending"),
      Keyspace.queue_key(queue, "active"),
      Keyspace.queue_key(queue, "")
    ]

    argv = [Integer.to_string(lease_ms)]

    # the live claim script, byte-unchanged by pause
    case Connector.eval(conn, @claim2, keys, argv) do
      {:ok, []} -> :empty
      {:ok, [id, payload, att]} -> {:ok, {id, payload, att}}
      other -> other
    end
  end
end

# …
# any non-nil paused field means paused; a wire fault reads as not paused
def paused?(conn, queue) do
  case Connector.command(conn, ["HGET", Keyspace.queue_key(queue, "meta"), "paused"]) do
    {:ok, nil} -> false
    {:ok, _} -> true
    _ -> false
  end
end

Because the gate is a field and not a set move, the live claim scripts (@claim2, @gclaim2) carry no pause logic: they keep their exact transition, and the read of the paused field happens above them in the claim function. The read and the claim are two round trips, not one atomic step. The native claim, EMQ.CLAIM, folds the same field read into the command. Pause and resume are both idempotent — pausing a paused queue and resuming a running one each answer :ok and change nothing observable.

Pause is a field the claim function reads, not a move on a set. The backlog survives a pause intact, and the claim script carries no pause logic at all.

Drain empties the backlog; active and the cadence survive

drain/3 runs @drain: for the pending set (and the schedule set when asked), read the members, delete each job's row and its :logs subkey, then delete the set. It answers {:ok, n}, the count drained. Two things it does not touch: the active set (those jobs are in flight) and the repeat registry (a registered repeatable keeps producing after a drain).

EchoMQ.Admin @drain — the handle

# echo_mq — EchoMQ.Admin — empty the backlog; active and the repeat registry survive
# KEYS = [the queue base, pending, (schedule when include_schedule: true)].
def drain(conn, queue, opts \\ []) do
  base = Keyspace.queue_key(queue, "")

  keys =
    [base, Keyspace.queue_key(queue, "pending")] ++
      if Keyword.get(opts, :include_schedule, false),
        do: [Keyspace.queue_key(queue, "schedule")],
        else: []

  case Connector.eval(conn, @drain, keys, []) do
    {:ok, n} when is_integer(n) -> {:ok, n}
    other -> other
  end
end

@drain — the body

-- the @drain script — wipe a set's rows and logs, then the set itself
-- KEYS[1] = the queue base root; KEYS[2] = pending; KEYS[3] = schedule (optional)
local base = KEYS[1]
local function wipe(setkey)
  local ids = redis.call('ZRANGE', setkey, 0, -1)
  for _, id in ipairs(ids) do
    -- the row key derives from the base root
    local jk = base .. 'job:' .. id
    -- the row and its logs subkey
    redis.call('DEL', jk, jk .. ':logs')
  end
  -- then the set
  redis.call('DEL', setkey)
  return #ids
end
-- pending, then the schedule set when asked
local n = wipe(KEYS[2])
if KEYS[3] then n = n + wipe(KEYS[3]) end
return n

A drain clears the backlog of already-enqueued work; it does not cancel work in hand or stop the cadence that produces more. An operator who has paused a queue and wants to clear what is queued without disturbing the jobs being worked or the registered repeatables reaches for drain/3.

Drain spares what is in flight and the cadence that produces more. It empties the queued backlog and nothing else — active and the repeat registry are left alone.

Obliterate destroys a paused queue, bounded

obliterate/3 runs @obliterate. It refuses a non-paused queue (EMQSTATE not paused → {:error, :not_paused}) and, unless forced, a queue with live active jobs (EMQSTATE active jobs present → {:error, :active}). Otherwise it deletes every state set (pending/active/schedule/dead), every reachable job row, and the auxiliary keys — the metrics hashes, the lane structures, the repeat records, and the meta hash with the paused flag. Each call is bounded by a budget, the module attribute @default_budget (1000) unless the caller passes :budget: it answers :more while work remains (call again) and :ok when the queue is gone.

EchoMQ.Admin @obliterate — the handle

# echo_mq — EchoMQ.Admin — destroy a PAUSED queue, bounded by a per-call budget
@default_budget 1000

# …
# KEYS = [meta, the queue base root]; ARGV = [force?, budget].
def obliterate(conn, queue, opts \\ []) do
  force = if Keyword.get(opts, :force, false), do: "force", else: ""
  budget = Keyword.get(opts, :budget, @default_budget)
  keys = [Keyspace.queue_key(queue, "meta"), Keyspace.queue_key(queue, "")]

  case Connector.eval(conn, @obliterate, keys, [force, Integer.to_string(budget)]) do
    # the queue is gone
    {:ok, 0} -> :ok
    # the budget is spent and work remains: call again
    {:ok, 1} -> :more
    {:error, {:server, "EMQSTATE not paused" <> _}} -> {:error, :not_paused}
    {:error, {:server, "EMQSTATE active" <> _}} -> {:error, :active}
    other -> other
  end
end

@obliterate — the body (the guards and the bounded state-set sweep)

-- the @obliterate script — KEYS[1] = meta, KEYS[2] = the queue base root
-- ARGV[1] = 'force' or '', ARGV[2] = the budget
local meta = KEYS[1]
local base = KEYS[2]
-- refuse unless paused: the order is pause, then obliterate
if redis.call('HGET', meta, 'paused') == false then
  return redis.error_reply('EMQSTATE not paused')
end
local force = ARGV[1] == 'force'
local budget = tonumber(ARGV[2])

local function del_job(id)
  local jk = base .. 'job:' .. id
  redis.call('DEL', jk, jk .. ':logs', jk .. ':lock')
end

-- active first: refuse on live active jobs unless forced
local active = base .. 'active'
local act = redis.call('ZRANGE', active, 0, budget - 1)
if #act > 0 and not force then
  return redis.error_reply('EMQSTATE active jobs present')
end
for _, id in ipairs(act) do del_job(id) end
if #act > 0 then
  redis.call('ZREM', active, unpack(act))
  budget = budget - #act
  -- 1 is :more: the budget is spent, call again
  if budget <= 0 then return 1 end
end

-- the remaining state sets, bounded
for _, t in ipairs({'pending', 'schedule', 'dead'}) do
  local sk = base .. t
  local ids = redis.call('ZRANGE', sk, 0, budget - 1)
  for _, id in ipairs(ids) do del_job(id) end
  if #ids > 0 then
    redis.call('ZREM', sk, unpack(ids))
    budget = budget - #ids
    if budget <= 0 then return 1 end
  end
end
-- … every state set is empty: the lanes (their rows first, under the same budget),
-- the repeat records, the metrics hashes and meta are deleted
return 0

The order matters: a queue must be paused before it can be obliterated, and (unforced) it must have no jobs in flight. The destructive verb refuses to run on a live queue. There is no completed or failed set to destroy — the metrics hashes are the throughput record, deleted as keys like everything else. Bounding the work by a budget keeps one obliterate call from blocking the server on a queue with a large backlog; the operator loops on :more until :ok.

Obliterate refuses a live queue. Paused, no jobs in flight unless forced, bounded by a budget — the destructive verb is gated by the same preconditions an operator would check by hand.

The per-job verbs reach into one row

The per-job plane lives on EchoMQ.Jobs, beside the state machine it extends. Each is one inline script that declares its keys; the branded id is gated at the key builder; a missing job answers a typed {:error, :gone} and changes nothing. The most expressive of them, update_progress/4, both writes the row and emits an event on the per-queue events channel.

Beat one — the named handle. update_progress/4 declares the row key, passes the progress value, the job id and the queue base, and maps the return.

EchoMQ.Jobs @update_progress — the handle

# echo_mq — EchoMQ.Jobs — write the progress field AND publish a progress event
# KEYS = [the row]; ARGV = [the progress value, the id, the queue base].
def update_progress(conn, queue, job_id, progress) when is_binary(progress) do
  keys = [Keyspace.job_key(queue, job_id)]
  argv = [progress, job_id, Keyspace.queue_key(queue, "")]

  case Connector.eval(conn, @update_progress, keys, argv) do
    {:ok, 1} -> :ok
    # no such row
    {:ok, -1} -> {:error, :gone}
    other -> other
  end
end

Beat two — the script body. After the field write, the script PUBLISHes on the per-queue events channel emq:{q}:events a cjson-encoded object. The event name rides the payload's event field, so one channel per queue carries every lifecycle event. A subscriber-less PUBLISH is a no-op.

@update_progress — the body

-- the @update_progress script — write the field, then publish the event
-- no row: gone, and nothing is written
if redis.call('EXISTS', KEYS[1]) == 0 then return -1 end
redis.call('HSET', KEYS[1], 'progress', ARGV[1])
-- ARGV[3] is the queue base, so ARGV[3] .. 'events' is the per-queue events channel
redis.call('PUBLISH', ARGV[3] .. 'events',
  cjson.encode({event = 'progress', job = ARGV[2], progress = ARGV[1]}))
return 1

The other verbs follow the same shape — declared keys, the id gated at the key builder, a typed refusal that changes nothing:

update_data/4 — replace the row's payload field (:ok | {:error, :gone}).
add_log/5 — RPUSH a line onto the job's logs list (emq:{q}:job:<id>:logs) and answer the count; with a keep argument it LTRIMs to the last keep lines. get_job_logs/3 reads the list in append order ({:ok, []} when empty).
remove_job/4 runs @remove_job2 — ZREM the job from whichever of the four sets holds it and, for a grouped job, from its lane, giving back the lane's active slot when the job was in flight; then delete the row and its :logs subkey, and release a held dedup key when its id is supplied. It refuses a locked job (a emq:{q}:job:<id>:lock key or a lock field on the row) with {:error, :locked}, leaving it untouched.
reprocess_job/3 runs @reprocess2 — the next section.

The operator console · pick a verb and a queue state

verb queue / job state

EchoMQ.Admin.pause/2 · @pause
precondition: none (idempotent)
:ok — paused field set; the pending backlog is untouched
native: none, a script only; EMQ.CLAIM reads the same paused field

Every per-job verb is a fenced transition that refuses cleanly: :gone for a missing row, :locked for a held job, :not_dead for a reprocess of a live job. The operator plane is the runbook, in real transitions.

Reprocess returns a dead job to the queue

reprocess_job/3 is the "retry a failed job" verb: it moves a dead job back to pending, clears last_error, and writes state = pending. The counters stay on the row. It refuses a job not in dead with {:error, :not_dead}, and a missing row with {:error, :gone}. A reprocessed job on a paused queue lands in pending and waits there: the transition does not read the pause flag, the claim does.

Beat one — the named handle. The live script is @reprocess2, with the queue base declared as KEYS[4].

EchoMQ.Jobs @reprocess2 — the handle

# echo_mq — EchoMQ.Jobs — move one dead job back to pending
# KEYS = [the row, dead, pending, the queue base]; ARGV = [id].
def reprocess_job(conn, queue, job_id) do
  keys = [
    Keyspace.job_key(queue, job_id),
    Keyspace.queue_key(queue, "dead"),
    Keyspace.queue_key(queue, "pending"),
    Keyspace.queue_key(queue, "")
  ]

  case Connector.eval(conn, @reprocess2, keys, [job_id]) do
    {:ok, 1} -> :ok
    {:ok, -1} -> {:error, :gone}
    {:error, {:server, "EMQSTATE" <> _}} -> {:error, :not_dead}
    other -> other
  end
end

Beat two — the script body. The ZREM on the dead set is the dead check, and a grouped job goes back to its lane rather than the flat pending set, so a grouped claim serves it and the lane's active count stays balanced.

@reprocess2 — the body

-- the @reprocess2 script — KEYS[1] = the row, KEYS[2] = dead, KEYS[3] = pending,
-- KEYS[4] = the queue base; ARGV[1] = id
local jk = KEYS[1]
-- no row: gone
if redis.call('EXISTS', jk) == 0 then return -1 end
local id = ARGV[1]
-- the ZREM is the dead check: a job not in dead is refused before any write
if redis.call('ZREM', KEYS[2], id) ~= 1 then
  return redis.error_reply('EMQSTATE not dead')
end
local base = KEYS[4]
-- the counters stay; only the error and the state change
redis.call('HDEL', jk, 'last_error')
redis.call('HSET', jk, 'state', 'pending')
local g = redis.call('HGET', jk, 'group')
if g then
  -- a grouped job returns to its lane, not the flat set
  local lane = base .. 'g:' .. g .. ':pending'
  redis.call('ZADD', lane, 0, id)
  -- … an unpaused lane under its limit rejoins the ring and pushes a wake token
else
  -- a flat job returns to pending at score 0
  redis.call('ZADD', KEYS[3], 0, id)
end
return 1

Native twin · EMQ.REPROCESS

# the scripted line — reprocess_job/3 runs EchoMQ.Jobs @reprocess2, EVALSHA-first on wire 4
EVALSHA <sha> 4 emq:{q}:job:<id> emq:{q}:dead emq:{q}:pending emq:{q}: <id>
# the native line — the same tail, the sha replaced by the command name
EMQ.REPROCESS 4 emq:{q}:job:<id> emq:{q}:dead emq:{q}:pending emq:{q}: <id>

reprocess_job/3 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 @reprocess2. Same keys and argument, and for a flat row the same replies and writes: 1 with ZREM dead, HDEL last_error, HSET state pending and ZADD pending at score 0, with attempts, failures and stalled kept; -1 for a missing row; EMQSTATE not dead for a job not in dead. The handler tests dead membership with a read before it writes, where the script's test is the ZREM itself, with the same reply. A grouped dead row is refused with EMQSTATE reprocess lane not native, where the script returns it to its lane and the ring. Every check runs before the first write. The mechanism: Commands, not scripts.

Reprocess is the one operator verb with a native twin. Both executors refuse a job that is not dead before they write, and only the script carries a grouped job back to its lane.

Pattern & implementation

the pattern (Redis Patterns Applied)

Operating a queue at scale — pausing intake, clearing a backlog, retrying a failure — is the operational side of the delay/schedule/priority family. Time, delay & priority teaches it, and Reliable queues the machinery underneath.

the implementation (echo_mq)

EchoMQ.Admin is the queue-scope plane: pause as a separate gate, drain that spares the active set and the cadence, bounded obliterate that refuses a live queue. The per-job verbs on EchoMQ.Jobs reach into one row, each one fenced and typed.

The pattern says: pause intake, clear the backlog, mend one job, retire the queue. The implementation is Admin's four queue-scope verbs and the six per-job verbs on Jobs.

Recap — control over the whole queue, and one row

Pause is a field the claim function reads, so the claim script never changes (Admin.pause/2 / resume/2); drain spares what is in flight and the cadence that produces more (drain/3); obliterate refuses a live queue and is bounded by a budget (obliterate/3); and every per-job verb is a fenced transition that refuses cleanly (update_data, update_progress with its events publish, add_log/get_job_logs, remove_job, reprocess_job, whose native twin is EMQ.REPROCESS). This dive closes lifecycle controls: time, the worker in hand, and now the whole queue and one row.

Back to the module: Lifecycle controls — scheduling and recurrence, cancellation and checkpoints, and the operator plane — the control plane beyond enqueue, claim, complete.

References

Where the operator verbs are grounded, and the commands their bodies run.

Sources

Valkey — HSET — the meta field pause sets and the row fields the per-job verbs write.
Valkey — HDEL — the field resume clears, and the error reprocess clears.
Valkey — PUBLISH — the progress event update_progress emits on the events channel.
Redis — EVALSHA — the load-once dispatch every operator verb runs 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 operator verbs act on.
Commands, not scripts — how a native twin checks before it writes.
One transition, two executors — the claim with the pause read folded in.
redis-patterns · Time, delay & priority — the operational delay/schedule family.