Scheduling & recurrence

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

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 measured on the SERVER clock (see @schedule).
def enqueue_at(conn, queue, job_id, payload, run_at_ms)
    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)
end

def enqueue_in(conn, queue, job_id, payload, delay_ms)
    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)
end

defp schedule(conn, queue, job_id, payload, mode, value) do
  # KEYS = [the job row, the schedule set]; the values ride ARGV.
  keys = [Keyspace.job_key(queue, job_id), Keyspace.queue_key(queue, "schedule")]

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

@schedule — the body

-- the @schedule script — admit a job, fenced out of pending until a run-at score
-- 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: the score. 'in' reads the SERVER clock and adds the delay,
-- so the delay is measured on the same clock promote and reap read — never the
-- caller's. 'at' takes the literal run-at value.
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); the id is parked on the
-- schedule set at the run-at score. Same row, same slot — fenced, not duplicated.
redis.call('HSET', KEYS[1], 'state', 'scheduled', 'attempts', '0', 'payload', ARGV[2])
redis.call('ZADD', KEYS[2], score, ARGV[1])
return 1

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

EchoMQ.Jobs @promote — the body

-- the @promote script — release every due scheduled job to pending
-- KEYS[1] = the schedule set, KEYS[2] = pending; ARGV[1] = the queue base root,
-- ARGV[2] = the batch limit. The clock is the server's, never the caller's.
local p = ARGV[1]
local t = redis.call('TIME')
local now = t[1] * 1000 + math.floor(t[2] / 1000)
-- The due read: every member scored at or before now, oldest-due first.
local due = redis.call('ZRANGEBYSCORE', KEYS[1], '-inf', now, 'LIMIT', 0, tonumber(ARGV[2]))
for _, id in ipairs(due) do
  redis.call('ZREM', KEYS[1], id)        -- leave the schedule set
  local jk = p .. 'job:' .. id
  local g = redis.call('HGET', jk, 'group')
  if g then
    redis.call('ZADD', p .. 'g:' .. g .. ':pending', 0, id)  -- a grouped job rejoins its lane
    -- (the lane re-wakes the ring when its concurrency allows — the lanes branch)
  else
    redis.call('ZADD', KEYS[2], 0, id)   -- a flat job rejoins pending at score 0
  end
  redis.call('HSET', jk, 'state', 'pending')
end
return #due                            -- the count released this pass

Promote and reap share the server-clock read: promote releases due scheduled jobs, the reaper returns expired leases. Both run on the consumer's beat; both price time on the wire's clock, never the caller's.

Backoff is a curve computed above the wire

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

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 climbing past the ceiling.
  raw = base * Bitwise.bsl(1, attempts - 1)
  min(raw, cap)
end

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

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

Backoff is a deterministic function, not a sleeping process. Compute the delay above the wire, hand the wire a number, and the retry cadence becomes testable.

Recurrence mints a fresh job per occurrence

EchoMQ.Repeat @register — the body

-- the @register script — register a cadence under a name, idempotently
-- KEYS[1] = the repeat ZSET, KEYS[2] = the repeat:<name> record hash.
-- ARGV[1] = name, ARGV[2] = every_ms, ARGV[3] = template, ARGV[4] = first-run ms.
if redis.call('EXISTS', KEYS[2]) == 1 then
  return 0                              -- a live name already registered — no-op
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

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 says: park a delayed job on a scored set, sweep it due, price the retry above the wire. The implementation is @schedule + @promote + Backoff + Repeat.

Recap — time is a control

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

References