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. Toggle the gate on the right, then run a claim.

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

The separate gate · toggle pause, then 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.

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: the claim script stays byte-unchanged, and the gate lives above it in Jobs.claim/3 (and Lanes.claim/3, so a queue-wide pause gates both the flat and the grouped claim). A paused queue stops serving work without losing a single member.

EchoMQ.Admin pause/2 & resume/2 — 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 subsequent 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 — set the paused field on the queue's meta hash
-- KEYS[1] is emq:{q}:meta. No set is moved: 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

Because the gate is a field and not a set move, the @claim and @gclaim scripts are unchanged by pause: they keep their exact transition, and the read of the paused field happens above them in the claim function. 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 reads, not a move on a set. The backlog survives a pause intact, and the claim script never has to know the queue was paused.

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/3 — 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 + 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
    local jk = base .. 'job:' .. id      -- the row key derives from the base root
    redis.call('DEL', jk, jk .. ':logs')        -- row + its logs subkey
  end
  redis.call('DEL', setkey)                  -- then the set
  return #ids
end
local n = wipe(KEYS[2])                    -- pending
if KEYS[3] then n = n + wipe(KEYS[3]) end -- schedule, when asked
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. It is bounded per invocation by a budget: it answers :more while work remains (call again) and :ok when the queue is gone.

EchoMQ.Admin obliterate/3 — the handle

# echo_mq — EchoMQ.Admin — destroy a PAUSED queue, bounded by a per-call budget
# KEYS = [meta, the queue base root]; ARGV = [force?, budget]. :more / :ok / refusals.
def obliterate(conn, queue, opts \\ []) do
  force = if Keyword.get(opts, :force, false), do: "force", else: ""
  budget = Keyword.get(opts, :budget, 1000)
  keys = [Keyspace.queue_key(queue, "meta"), Keyspace.queue_key(queue, "")]

  case Connector.eval(conn, @obliterate, keys, [force, Integer.to_string(budget)]) do
    {:ok, 0} -> :ok                            # the queue is gone
    {:ok, 1} -> :more                          # budget spent, work remains — call again
    {: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 — refuse unless paused; then delete, bounded by budget
local meta = KEYS[1]
local base = KEYS[2]
if redis.call('HGET', meta, 'paused') == false then
  return redis.error_reply('EMQSTATE not paused')   -- order matters: pause before obliterate
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
-- ...then pending / schedule / dead the same way, each ZREM'd under the budget;
-- return 1 (:more) the moment the budget is spent. When every set is empty,
-- the lane structures, repeat records, metrics hashes and meta are DEL'd,
-- and the script returns 0 (:ok) — the queue is gone.

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 wire on a queue with a large backlog; the operator loops on :more until :ok.

Obliterate refuses a live queue. Paused, no jobs in flight, 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/4 — the handle

# echo_mq — EchoMQ.Jobs — write the progress field AND publish a progress event
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
    {:ok, -1} -> {:error, :gone}      # no such row
    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
if redis.call('EXISTS', KEYS[1]) == 0 then return -1 end  -- no row: gone
redis.call('HSET', KEYS[1], 'progress', ARGV[1])
-- ARGV[3] is the queue base; 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 five 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 — ZREM the job from whichever of the four sets holds it, delete the row and its :logs subkey, and release a held dedup key when its id is supplied. It refuses a locked job (emq:{q}:job:<id>:lock present) with {:error, :locked}, leaving it untouched.
reprocess_job/3 — move a dead job back to pending: clear last_error, set state = pending, add it to the pending set. It refuses a job not in dead with {:error, :not_dead}.

The operator console · pick a verb and a queue state

verb queue / job state

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.

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 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). 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.
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.
The Lua layer — how each operator verb is one atomic, declared-key script.
The Protocol — the keyspace and the branded id every verb's keys derive from.
redis-patterns · Time, delay & priority — the operational delay/schedule family.