Lifecycle controls · dive 01

Scheduling & recurrence

A schedule is a score, not a second queue. A backoff is a value, not a sleep. A recurrence is a registration, not a loop.

A queue's default is now: an enqueued job is claimable the moment it lands in pending. Four controls move that in time. Schedule admits a job now but fences it out of pending until a run-at instant, and promote releases it once due. Delay sends a job in hand back to the schedule without touching its attempts, which the next claim increments as after any retry. Backoff prices a delay as a pure curve, and recurrence registers a cadence that mints a fresh job for every occurrence. Schedule, promote and delay are scripts in EchoMQ.Jobs with native twins on a module-loaded node; backoff and recurrence run on the host. Drag "now" on the right, then read the beats below.

The visibility fence promote delay The backoff curve Recurrence Pattern & implementation References

The schedule clock · drag now

now (seconds from t0): 0

now = 0s → promote releases 0 · all 3 jobs still fenced (every score > now)

The schedule set is a visibility fence, not a second queue

Both enqueue_at/6 and enqueue_in/6 run one script, @schedule, through the private schedule/7. It is the enqueue transition with one change: the row is written state = scheduled and the id goes on the schedule set (emq:{q}:schedule) at the run-at score instead of pending. The member is the same branded id, so byte order is still mint order: once a scheduled job is released, it sorts among the pending jobs by its mint instant, not its release instant.

Beat one — the named handle. The two public verbs differ only in the mode token ("at" or "in") and the guard on the value: any integer for an absolute run-at, a non-negative integer for a delay. The trailing opts carries :via, the module the eval dispatches through, EchoMQ.Connector unless the caller passes a pool.

EchoMQ.Jobs @schedule — the handle

# echo_mq — EchoMQ.Jobs
# Two public verbs, one private body. enqueue_at takes an absolute run-at ms;
# enqueue_in takes a relative delay the script prices on the SERVER clock.
def enqueue_at(conn, queue, job_id, payload, run_at_ms, opts \\ [])
    when is_binary(job_id) and is_binary(payload) and is_integer(run_at_ms) do
  schedule(conn, queue, job_id, payload, "at", run_at_ms, opts)
end

# …

def enqueue_in(conn, queue, job_id, payload, delay_ms, opts \\ [])
    when is_binary(job_id) and is_binary(payload) and is_integer(delay_ms) and delay_ms >= 0 do
  schedule(conn, queue, job_id, payload, "in", delay_ms, opts)
end

# …
# opts carries :via, the dispatch module: EchoMQ.Connector unless a pool is passed.
defp schedule(conn, queue, job_id, payload, mode, value, opts) do
  via = Keyword.get(opts, :via, Connector)
  # KEYS = [the job row, the schedule set]; ARGV = [id, payload, mode, value].
  keys = [Keyspace.job_key(queue, job_id), Keyspace.queue_key(queue, "schedule")]

  case via.eval(conn, @schedule, keys, [job_id, payload, mode, Integer.to_string(value)]) do
    # 1: the row is written scheduled and parked on the schedule set
    {:ok, 1} -> {:ok, :scheduled}
    # 0: the id already has a row, an idempotent no-op
    {:ok, 0} -> {:ok, :duplicate}
    {:error, {:server, "EMQKIND" <> _}} -> {:error, :kind}
    other -> other
  end
end

Beat two — the script body. @schedule shares the enqueue gate and idempotency, then diverges on the score. In mode 'in' the score is the server TIME plus the delay; in any other mode it is the literal value. The caller's clock prices only an absolute run-at.

@schedule — the body

-- the @schedule script — KEYS[1] = the job row, KEYS[2] = the schedule set
-- ARGV[1] = id, ARGV[2] = payload, ARGV[3] = 'in' or 'at', ARGV[4] = the value
-- The same branded-id gate as enqueue: the id must be JOB-namespaced.
if string.sub(ARGV[1], 1, 3) ~= 'JOB' then
  return redis.error_reply('EMQKIND job id must be JOB-namespaced')
end
-- The same idempotency as enqueue: a row that already exists is a no-op.
if redis.call('EXISTS', KEYS[1]) == 1 then
  return 0
end
-- The one divergence is the score. 'in' reads the SERVER clock and adds the
-- delay; any other mode takes the value as an absolute run-at.
local score
if ARGV[3] == 'in' then
  local t = redis.call('TIME')
  local now = t[1] * 1000 + math.floor(t[2] / 1000)
  score = now + tonumber(ARGV[4])
else
  score = tonumber(ARGV[4])
end
-- The row carries state = scheduled (not pending), attempts 0 and the payload;
-- the id is parked on the schedule set at the run-at score.
redis.call('HSET', KEYS[1], 'state', 'scheduled', 'attempts', '0', 'payload', ARGV[2])
redis.call('ZADD', KEYS[2], score, ARGV[1])
return 1

A scheduled job is not a different kind of job in a different queue. It is the same row, on the same slot, fenced out of pending by a score a promote pass reads. The fence is a number; there is nothing to move back, because there was never a second structure.

Native twin · EMQ.SCHEDULE

# the scripted line — enqueue_at/6 and enqueue_in/6 run EchoMQ.Jobs @schedule, EVALSHA-first on wire 4
EVALSHA <sha> 2 emq:{q}:job:<id> emq:{q}:schedule <id> <payload> <in|at> <value>
# the native line — the same tail, the sha replaced by the command name
EMQ.SCHEDULE  2 emq:{q}:job:<id> emq:{q}:schedule <id> <payload> <in|at> <due>
# the native verb with no script: rescore a job already on the schedule set
EMQ.RESCHEDULE 2 emq:{q}:job:<id> emq:{q}:schedule <id> <in|at> <due>

enqueue_at/6 and enqueue_in/6 send the native line through a wire: 6 connector, as a raw client on a module-loaded node can; over the default wire 4 they evaluate the script. Same keys, same argument order. The replies are the script's: 1 admitted, 0 for an id that already has a row (nothing written), and EMQKIND job id must be JOB-namespaced. For valid input the two agree. The handler checks the shape, the mode, the due and the fence before it reads a key, so a refusal writes nothing, and it refuses two inputs the script admits: a mode other than in or at (ERR EMQ.SCHEDULE mode must be in or at; the script prices any mode but 'in' as absolute) and a negative or non-integer due (ERR EMQ.SCHEDULE due must be a non-negative integer; enqueue_at/6 guards only is_integer, so a negative run-at reaches the script).

EMQ.RESCHEDULE has no scripted twin and no Elixir function sends it. It rescores a job already on the schedule set and keeps its row: it writes only ZADD … XX and answers 1, -1 for a missing row, or EMQSTATE reschedule not scheduled. How a command checks before it writes: Commands, not scripts.

A schedule is a score on the one schedule set. The run-at instant fences the job out of pending; the mint instant still orders it once it is released.

promote releases the due ones, on the server clock

promote/3 runs @promote: read the server TIME, take the schedule members scored at or before now (ZRANGEBYSCORE schedule -inf now LIMIT 0 batch), and move each one to pending at score 0 — or, for a grouped job, to its lane, waking the ring when the lane is unpaused and under its limit — with state = pending. It returns how many it released.

Beat one — the named handle. The queue base rides in ARGV, and the reply is passed through raw.

EchoMQ.Jobs @promote — the handle

# echo_mq — EchoMQ.Jobs
# KEYS = [the schedule set, pending]; ARGV = [the queue base, the batch].
# The reply is the script's count, passed through raw: {:ok, n}.
def promote(conn, queue, batch) when is_integer(batch) and batch > 0 do
  keys = [Keyspace.queue_key(queue, "schedule"), Keyspace.queue_key(queue, "pending")]

  Connector.eval(conn, @promote, keys, [Keyspace.queue_key(queue, ""), Integer.to_string(batch)])
end

@promote — the body

-- the @promote script — KEYS[1] = the schedule set, KEYS[2] = pending
-- ARGV[1] = the queue base (the row and lane keys derive from it), ARGV[2] = the batch
local p = ARGV[1]
local t = redis.call('TIME')
local now = t[1] * 1000 + math.floor(t[2] / 1000)
-- The due read, on the server clock: every member scored at or before now, up to the batch.
local due = redis.call('ZRANGEBYSCORE', KEYS[1], '-inf', now, 'LIMIT', 0, tonumber(ARGV[2]))
for _, id in ipairs(due) do
  -- leave the schedule set
  redis.call('ZREM', KEYS[1], id)
  local jk = p .. 'job:' .. id
  local g = redis.call('HGET', jk, 'group')
  if g then
    -- a grouped job rejoins its lane
    local lane = p .. '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 rejoins pending at score 0
    redis.call('ZADD', KEYS[2], 0, id)
  end
  -- the row is written pending, whether or not a row was there
  redis.call('HSET', jk, 'state', 'pending')
end
-- the count released this pass
return #due

Promote is bounded by a batch and idempotent: a member already moved is gone from the schedule set, so a second pass does not move it twice. Because the score is read against the server clock, a job admitted with enqueue_in/6 is released on the same clock that priced its delay — no caller-clock skew between admission and release.

Nothing promotes on its own. A client process calls promote/3: the Consumer's beat runs its recovery leg, then promote/3, then the claims; in a pool, the Metronome runs the reap and the promote once per beat for the queue; the opt-in Pump's tick promotes, fires due repeats and delivers flow completions, and does not reap.

Native twin · EMQ.PROMOTE

# the scripted line — promote/3 runs EchoMQ.Jobs @promote, EVALSHA-first on wire 4
EVALSHA <sha> 2 emq:{q}:schedule emq:{q}:pending emq:{q}: <batch>
# the native line — the same tail, the sha replaced by the command name
EMQ.PROMOTE   2 emq:{q}:schedule emq:{q}:pending emq:{q}: <batch>

The reply is the count of selected members, as the script's. Three things differ, all visible to a caller. A ghost — a scheduled id with no row — is only removed from the schedule set; the script adds it to pending and its HSET creates a skeleton row. A grouped due row refuses the whole call with EMQSTATE promote lane not native and nothing is written; the script routes it to its lane. And the batch is validated (ERR EMQ.PROMOTE batch must be a positive integer) where the script runs tonumber. The handler reads every selected row before its first write, so a refusal writes nothing. The mechanism: Commands, not scripts.

Promote and reap read the same server clock: promote releases due scheduled jobs, the reaper returns expired leases. Both run only when a client process calls them.

delay sends a job in hand back to the schedule

delay/6 is the one move from active to the schedule set: the worker holding the token re-scores its job to run later without touching attempts: the next claim increments it as after any retry, and the delay itself never dead-letters. @delay checks the JOB prefix, fences the token against the row's attempts, removes the id from active, writes state = scheduled, and parks the id at the same 'in'/'at' score @schedule computes. It is not @schedule: that script refuses an existing row and resets attempts to 0, while @delay keeps the row, its attempts and its payload.

EchoMQ.Jobs @delay — the handle

# echo_mq — EchoMQ.Jobs
# The token is the attempts value the claim returned. mode: :at switches the
# value to an absolute run-at; the default prices it on the server clock.
def delay(conn, queue, job_id, token, delay_ms, opts \\ [])
    when is_binary(job_id) and is_integer(token) and is_integer(delay_ms) do
  mode = if Keyword.get(opts, :mode, :in) == :at, do: "at", else: "in"

  # KEYS = [the job row, the active set, the schedule set], all three declared.
  keys = [
    Keyspace.job_key(queue, job_id),
    Keyspace.queue_key(queue, "active"),
    Keyspace.queue_key(queue, "schedule")
  ]

  argv = [job_id, Integer.to_string(token), mode, Integer.to_string(delay_ms)]

  case Connector.eval(conn, @delay, keys, argv) do
    {:ok, 1} -> :ok
    # the more specific EMQSTALE text is matched first
    {:error, {:server, "EMQSTALE job gone" <> _}} -> {:error, :gone}
    {:error, {:server, "EMQSTALE" <> _}} -> {:error, :stale}
    {:error, {:server, "EMQKIND" <> _}} -> {:error, :kind}
    other -> other
  end
end

@delay — the body

-- the @delay script — KEYS[1] = the job row, KEYS[2] = active, KEYS[3] = the schedule set
-- ARGV[1] = id, ARGV[2] = the token, ARGV[3] = 'in' or 'at', ARGV[4] = the value
if string.sub(ARGV[1], 1, 3) ~= 'JOB' then
  return redis.error_reply('EMQKIND job id must be JOB-namespaced')
end
-- The token fence: the row's attempts, compared as a string.
local att = redis.call('HGET', KEYS[1], 'attempts')
if not att then return redis.error_reply('EMQSTALE job gone') end
if att ~= ARGV[2] then
  return redis.error_reply('EMQSTALE delay token mismatch')
end
-- Release the lease. A member already gone from active makes this a no-op,
-- and the script schedules the job all the same.
redis.call('ZREM', KEYS[2], ARGV[1])
-- Only the state changes: attempts and the payload stay on the row.
redis.call('HSET', KEYS[1], 'state', 'scheduled')
local score
if ARGV[3] == 'in' then
  local t = redis.call('TIME')
  local now = t[1] * 1000 + math.floor(t[2] / 1000)
  score = now + tonumber(ARGV[4])
else
  score = tonumber(ARGV[4])
end
redis.call('ZADD', KEYS[3], score, ARGV[1])
return 1

Native twin · EMQ.DELAY

# the scripted line — delay/6 runs EchoMQ.Jobs @delay, EVALSHA-first on wire 4
EVALSHA <sha> 3 emq:{q}:job:<id> emq:{q}:active emq:{q}:schedule <id> <token> <in|at> <value>
# the native line — the same tail, the sha replaced by the command name
EMQ.DELAY     3 emq:{q}:job:<id> emq:{q}:active emq:{q}:schedule <id> <token> <in|at> <due>

For a live holder the reply is 1 and the writes are the script's: ZREM active, HSET state scheduled, ZADD on the schedule set, with attempts, failures, stalled and the payload untouched. A missing row, a stale token and a foreign id answer the script's texts (EMQSTALE job gone, EMQSTALE delay token mismatch, EMQKIND …). Three refusals are native only: a token holder no longer in active answers EMQSTATE delay not active, where the script's ZREM is a no-op and the job is scheduled anyway; a grouped row answers EMQSTATE delay lane not native; and a negative delay answers the due text, where delay/6 guards only is_integer. Every check runs before the first write.

Delay is the inverse of a claim: it releases a lease and mints nothing, so the job returns to the schedule with its attempt history intact.

Backoff is a pure curve, and who calls it

Backoff.delay_ms(policy, attempts) is a pure function: no process, no clock, no I/O. Three policies. {:fixed, ms} is the same delay every attempt. {:exponential, base, cap} is base * 2^(attempts-1), clamped at cap, so the curve climbs and then holds. {:jitter, inner} is a uniform random delay in 0..inner_delay — the full-jitter form that spreads a retry storm.

# echo_mq — EchoMQ.Backoff — a pure function of (policy, attempts) to a literal ms
# {:fixed, ms}: the same delay on every attempt.
def delay_ms({:fixed, ms}, attempts)
    when is_integer(ms) and ms >= 0 and is_integer(attempts) and attempts >= 1,
    do: ms

# {:exponential, base, cap}: the curve climbs, then holds at the cap.
def delay_ms({:exponential, base, cap}, attempts)
    when is_integer(base) and base >= 1 and is_integer(cap) and cap >= 1 and
           is_integer(attempts) and attempts >= 1 do
  # base * 2^(attempts-1), clamped at cap; the shift is exact-integer and
  # the cap holds the curve from overflowing past the ceiling.
  raw = base * Bitwise.bsl(1, attempts - 1)
  min(raw, cap)
end

# {:jitter, inner}: a uniform draw in 0..bound, where the inner policy is the bound.
def delay_ms({:jitter, inner}, attempts) when is_integer(attempts) and attempts >= 1 do
  bound = delay_ms(inner, attempts)
  if bound <= 0, do: 0, else: :rand.uniform(bound + 1) - 1
end

The module's doctests run as tests and are the curve's contract:

iex> EchoMQ.Backoff.delay_ms({:exponential, 100, 10_000}, 1)
100
iex> EchoMQ.Backoff.delay_ms({:exponential, 100, 10_000}, 3)
400
iex> EchoMQ.Backoff.delay_ms({:exponential, 100, 10_000}, 20)
10000

The backoff curve · plot delay over attempts

{:fixed, 1000}
attempt→ms (delay): 1→1000 2→1000 3→1000 4→1000 5→1000 6→1000 7→1000 8→1000
the same 1000 ms on every attempt — a flat curve

Who calls it. The wire takes a literal delay: Jobs.retry/7 carries a number, and @retry parks the job at now + delay on the server clock. The shipped Consumer hands it a fixed number — a failed handler is retried after retry_delay_ms (default 1_000), the same wait on every attempt. The Consumer and the Metronome call Backoff for something else: after a wire error on a leg, the loop sleeps on an exponential curve over its beat, indexed by the streak of degraded beats. The conformance suite's poison-job drill is the one caller that prices a job retry on the curve.

# echo_mq — EchoMQ.Consumer
# A failed handler: Jobs.retry/7 gets the fixed retry_delay_ms (default 1_000).
defp retry(s, id, att, reason) do
  err = to_string(reason)
  result = Jobs.retry(s.conn, s.queue, id, att, s.retry_delay_ms, s.max_attempts, err)
# …
# A degraded beat (a wire error on a leg): the loop sleeps on the curve,
# indexed by the streak of degraded beats.
defp degraded_sleep(%{beat_ms: beat_ms, degraded: streak}) do
  Backoff.delay_ms({:exponential, beat_ms, max(30_000, beat_ms)}, streak)
end

Backoff is a deterministic function, not a sleeping process. A caller that wants a retry curve computes it above the wire and hands Jobs.retry/7 the number; the shipped Consumer hands it a fixed one.

Recurrence mints a fresh job per occurrence

Repeat.register/6 writes a registration under a name: a record hash emq:{q}:repeat:<name> carrying every_ms and a payload template, and a member on the emq:{q}:repeat sorted set scored by its next-run millisecond. cancel/3 removes both. due/3 reads the names scored at or before now, each with its record, as host commands rather than a script. advance/4 sets a name's next run to now plus its period, and count/2 is the registry depth.

Every score in the registry is priced on the host clock: register/6, due/3 and advance/4 all read the BEAM's System.system_time(:millisecond), where the schedule set and promote read the server's TIME. A late tick does not accumulate as a backlog of runs: the next run is measured from the moment the Pump advances the name, not from the old score.

EchoMQ.Repeat @register — the handle

# echo_mq — EchoMQ.Repeat
# register/6: the first run is the HOST clock now plus first_in_ms.
def register(conn, queue, name, every_ms, template, first_in_ms \\ 0)
    when is_binary(name) and is_integer(every_ms) and every_ms > 0 and
           is_binary(template) and is_integer(first_in_ms) and first_in_ms >= 0 do
  keys = [Keyspace.queue_key(queue, "repeat"), repeat_key(queue, name)]
  first_at = now_ms() + first_in_ms

  argv = [name, Integer.to_string(every_ms), template, Integer.to_string(first_at)]

  case Connector.eval(conn, @register, keys, argv) do
    {:ok, 1} -> {:ok, :registered}
    {:ok, 0} -> {:ok, :exists}
    other -> other
  end
end

# …
# advance/4: the next run is the host clock now plus the period, not the old
# score plus the period. The Pump calls it after it enqueues an occurrence.
def advance(conn, queue, name, every_ms)
    when is_binary(name) and is_integer(every_ms) and every_ms > 0 do
  keys = [Keyspace.queue_key(queue, "repeat"), repeat_key(queue, name)]
  next_at = now_ms() + every_ms

  case Connector.eval(conn, @advance, keys, [name, Integer.to_string(next_at)]) do
    {:ok, 1} -> {:ok, :advanced}
    # the record was cancelled mid-sweep: the script drops the dangling member
    {:ok, 0} -> {:ok, :absent}
    other -> other
  end
end

# …
# The one clock every repeat score is priced on: the BEAM host's wall clock.
defp now_ms, do: System.system_time(:millisecond)

@register — the body

-- the @register script — KEYS[1] = the repeat set, KEYS[2] = the repeat:<name> record
-- ARGV[1] = name, ARGV[2] = every_ms, ARGV[3] = template, ARGV[4] = the first-run ms
-- A live name already registered is a no-op.
if redis.call('EXISTS', KEYS[2]) == 1 then
  return 0
end
-- The record carries the period and the payload template the Pump stamps.
redis.call('HSET', KEYS[2], 'every_ms', ARGV[2], 'template', ARGV[3])
-- The name is the registry member, scored by its first-run instant.
redis.call('ZADD', KEYS[1], tonumber(ARGV[4]), ARGV[1])
return 1

The cadence is the Pump's: on each tick it reads the due registrations, mints a fresh JOB id on the host, enqueues an occurrence carrying the template, then advances the name. Each occurrence is a first-class, browsable, mint-ordered job, because the id mints fresh per occurrence — never a reused row, which would break both the order theorem and the dedup semantics. The enqueue comes before the advance, so a Pump that crashes between the two, or a second Pump that reads the same due record, enqueues the occurrence twice: the semantics are at least once per due period, and a consumer that needs one effect dedups on its own key.

Worked example. Register a daily report with every_ms = 86_400_000. The registry holds one member. When it comes due, the Pump mints a new JOB-namespaced id, enqueues an occurrence carrying the template, and advances the score to the host's now plus one day. Tomorrow's run is a different id, a different row and a different mint instant; the registration is unchanged.

A recurrence is one registration the Pump sweeps, not a loop that holds state. Every run is its own branded job, so a recurring stream is browsable and ordered like any other.

Pattern & implementation

the pattern (Redis Patterns Applied)

A delayed or scheduled job is a sorted set scored by run-at time, swept by a promote pass; a retry curve is computed client-side. Time, delay & priority teaches the delay/schedule family over Redis.

the implementation (echo_mq)

@schedule parks on the schedule set at the run-at score, @promote releases the due ones on the server clock, @delay returns a job in hand to the schedule, Backoff.delay_ms/2 is the curve a caller computes, and EchoMQ.Repeat is the cadence on the host clock.

The pattern says: park a delayed job on a scored set, sweep it due, price the retry above the wire. The implementation is @schedule + @promote + @delay + Backoff + Repeat.

Recap — time is a control

A schedule is a score, not a second queue — @schedule parks the row on the schedule set (a relative delay priced on the server clock, an absolute run-at on the caller's) and @promote releases it due on the server clock, both with a native twin (EMQ.SCHEDULE, EMQ.PROMOTE) that checks before it writes. @delay and EMQ.DELAY send a job in hand back to the schedule without touching attempts (unlike @schedule, which resets it to 0); the next claim increments it as after any retry, and the delay itself never dead-letters. EMQ.RESCHEDULE rescores a scheduled job and has no Elixir twin. A backoff is a value, not a sleep, and the shipped Consumer retries on a fixed delay. A recurrence is a registration, not a loop — Repeat mints a fresh job per occurrence, on the host clock. The next dive moves from time to the worker in hand.

Next in lifecycle controls: Cancellation & checkpoints — the cooperative cancellation token, extend_lock as a lease checkpoint, and the count-thresholded stalled sweep.

References

Where the schedule, promote, delay and recurrence transitions are grounded, and the commands their bodies use.

Sources

Valkey — ZADD — the schedule-set and repeat-set insertion at the run-at score.
Valkey — ZRANGEBYSCORE — the due read promote runs against the schedule set.
Valkey — TIME — the server clock the schedule, promote and delay scripts read.
Valkey — HSET — the scheduled row write and the repeat record fields.
Redis — EVALSHA — the load-once dispatch the schedule verbs 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 consumer loop — the beat that calls promote and paces a degraded leg.
Commands, not scripts — how a native twin checks before it writes.
redis-patterns · Time, delay & priority — the delay/schedule/priority family.