Batches · dive 03

Batch lease extension

Re-score the many under one clock read. The lease is the active score, so checkpointing many is one script — and it returns the ones it did not extend.

A worker holding several jobs at once checkpoints all their leases together, on one clock. extend_locks/4 re-scores every live lease under one server-clock read and answers the failed list: the ids whose row is gone, whose token is stale, or which are no longer in active. The lease in this Queue is the active-set score, so extending it is re-scoring the active member; there is no separate lock. On a node that loads the native module, the same keys and arguments run as one command, EMQ.EXTENDS. Step the batch on the right, then read the two beats, the contrast with the single verb, and the native twin.

The two beats Single vs batch The native twin Pattern & implementation References

Extend four leases · step the members

JOB0KHTOWnGLuA · token 1, attempts 1, in active → re-scored → ZADD active now+lease
failed after member 1: []
a live lease: ZADD active now+lease, the same now as every other member

Batch lease extension, in two beats

Read it the way the protocol is built: the handle, the Elixir verb that gates the ids and runs the script, and the body, the Lua that re-scores the members.

Beat one — the named handle. extend_locks/4 gates every id at Keyspace.job_key/2 first, puts the lease at ARGV[1] and flattens the {id, token} pairs after it, and runs EchoMQ.Jobs @extend_locks2 with two declared keys: the active set and the queue base emq:{q}:. It answers {:ok, failed}.

# echo_mq — EchoMQ.Jobs
# extend_locks/4: many leases, one script, one server-clock read.
def extend_locks(conn, queue, held, lease_ms)
    when is_list(held) and is_integer(lease_ms) and lease_ms > 0 do
  # every id is gated at the key builder before the wire: an ill-formed id
  # raises here and never reaches a key
  Enum.each(held, fn {id, _token} -> Keyspace.job_key(queue, id) end)

  # each {id, token} becomes two ARGV slots; the token travels as a decimal string
  pairs =
    Enum.flat_map(held, fn {id, token} -> [id, Integer.to_string(token)] end)

  # ARGV[1] is the lease; the pairs start at ARGV[2]
  argv = [Integer.to_string(lease_ms) | pairs]

  # KEYS[1] is the active set; KEYS[2] is the queue base, the declared root
  # every row key is derived from
  keys = [Keyspace.queue_key(queue, "active"), Keyspace.queue_key(queue, "")]

  case Connector.eval(conn, @extend_locks2, keys, argv) do
    # the ids whose lease was not extended
    {:ok, failed} when is_list(failed) -> {:ok, failed}
    other -> other
  end
end

Beat two — the script body. @extend_locks2 reads the clock once, then walks the pairs from ARGV[2]. A member is re-scored only when three checks pass in order: its row exists, the row's attempts equals the caller's token, and ZSCORE finds the id in active.

-- the @extend_locks2 script — KEYS[1] = active, KEYS[2] = the queue base emq:{q}:
-- ARGV[1] = lease_ms, then (id, token) pairs from ARGV[2]
local base = KEYS[2]
local lease = tonumber(ARGV[1])
-- one server-clock read for the whole batch
local t = redis.call('TIME')
local now = t[1] * 1000 + math.floor(t[2] / 1000)
local failed = {}
local i = 2
-- i < #ARGV: a trailing id with no token after it is never read
while i < #ARGV do
  local id = ARGV[i]
  local token = ARGV[i + 1]
  -- the row key is derived from the declared root, so it shares the queue's hashtag
  local jk = base .. 'job:' .. id
  local att = redis.call('HGET', jk, 'attempts')
  -- a row, the holder's token (a string compare), and membership of active
  if att and att == token and redis.call('ZSCORE', KEYS[1], id) then
    -- the lease is the active score: re-score the member to now + lease
    redis.call('ZADD', KEYS[1], now + lease, id)
  else
    -- no row, a stale token, or no longer in active: reported, never re-scored
    table.insert(failed, id)
  end
  i = i + 2
end
return failed

The membership guard is the lesson of the live script. A lease that lapsed and was reaped back to pending keeps its attempts, so its holder's token still matches; without the ZSCORE check, the extension would put the id back in active while it also waits in pending. With it, the id joins failed and is never re-scored, so a reaped lease cannot be resurrected in bulk.

Batch lease extension is one script and one clock read: re-lease every live member at the shared now + lease, and return the ids that had no row, a stale token, or no place in active.

Single vs batch — collect, don't error

The single extend_lock/5 re-scores one member under @extend_lock2, with the same three checks in the same order, and answers each with its own error: no row → {:error, :gone}, a stale token → {:error, :stale} (the EMQSTALE wire class), not in active → {:error, :not_active}. The batch does not error on one bad id: it collects the id into failed, so one bad lease among many does not abort the others. Pick a job's fate to read both answers, and the native replies beside them.

extend_lock/5 vs extend_locks/4
	single `extend_lock/5`	batch `extend_locks/4`
ids	one	many
keys declared	`[active, job_row]`	`[active, base]`; each row key derived in-script: `base .. 'job:' .. id`
clock reads	one	one, shared across all ids
no row	errors — `{:error, :gone}`	collected — the id joins `failed`
a stale token	errors — `{:error, :stale}`	collected — the id joins `failed`
not in `active`	errors — `{:error, :not_active}`	collected — the id joins `failed`
return	`:ok` \| `{:error, …}`	`{:ok, failed}`
native twin	`EMQ.EXTEND`	`EMQ.EXTENDS`

A job's fate, two verbs · read both answers

single: :ok · batch: re-scored
native: EMQ.EXTEND → 1 · EMQ.EXTENDS → the id is absent from the reply; both stamp lease_epoch
the row exists, the token matches, the id is in active: both verbs re-score it at now + lease

The single verb errors because there is nothing to collect: one id, one answer. The batch collects because it has many ids and one result: a bad lease becomes an entry in failed, and the rest are re-leased on the same clock.

The native twin — `EMQ.EXTENDS`

On a node that loads the native module, extend_locks/4 through a wire: 6 connector, or any raw client, sends the same keys and arguments as one command. Over the default wire 4, extend_locks/4 evaluates the script.

# scripted (wire 4): what extend_locks/4 sends, through Connector.eval/5
EVALSHA <sha> 2 emq:{q}:active emq:{q}: <lease_ms> <id> <token> …
# native (wire 6): the same keys and arguments, as one command
EMQ.EXTENDS 2 emq:{q}:active emq:{q}: <lease_ms> <id> <token> …

Both reply with an array of the ids not extended, in input order. The differences a caller can see:

Every check before any write. The handler reads every verdict first, then writes. A refusal writes nothing, including a WRONGTYPE row anywhere in the batch; the script keeps the ZADDs it made before the row it failed on.

An unpaired trailing id is refused (ERR EMQ.EXTENDS wrong number of arguments), where the script's while i < #ARGV ignores it. extend_locks/4 never sends one, because it flattens whole pairs.

The shape is checked: numkeys must be 2, KEYS[1] must be the active set of the base in KEYS[2], and lease_ms must be a positive integer.

Each row is opened as the caller. The handler stamps every extended row, so a restricted user needs write access to the job rows: a caller that may only read them gets NOPERM, where wire 4 served it.

lease_epoch is stamped on every extended row with the server's run_id; the script writes no such field.

The clock is the server's cached command time, the value TIME reports, read once, so every extended member shares one score, as in the script.

It does not read group, so a lane job's lease extends as a flat job's does. The handler, verbatim from the branch head:

// trafikey — EMQ.EXTENDS, the native twin of EchoMQ.Jobs @extend_locks2
pub(crate) fn extends(ctx: &Context, args: Vec<ValkeyString>) -> ValkeyResult {
    let s = host::shape("EMQ.EXTENDS", &args, 2)?;
    // the lease plus whole (id, token) pairs is an odd count: an unpaired id is refused
    if s.argc() % 2 == 0 {
        return Err(errors::arg_count(s.verb));
    }
    // … the layout check and the lease
    let caller = host::fence(ctx)?;
    // one deadline for the whole batch, from the server's cached clock
    let at = host::deadline(s.verb, host::now_ms()?, lease)?;
    let mut live: Vec<(&[u8], Vec<u8>)> = Vec::new();
    let mut failed: Vec<ValkeyValue> = Vec::new();
    let mut i = 2;
    // the read pass: every verdict before any write
    while i < s.argc() {
        let (id, token) = (s.arg(i), s.arg(i + 1));
        // the row is derived from the base and opened as the caller: NOPERM refuses here
        let row = q.job(id);
        caller.open(&row, Access::Read)?;
        // the script's three checks, in the script's order: a row, the token, membership
        if lease_verdict(ctx, s.key(1), &row, id, token)? == Lease::Live {
            live.push((id, row));
        } else {
            failed.push(host::bulk(id));
        }
        i += 2;
    }
    // the write pass, only when a lease is live
    if !live.is_empty() {
        for (_, row) in &live {
            caller.open(row, Access::Write)?;
        }
        let run_id = epoch::current()?;
        let score = at.to_string();
        let w = host::begin_writes(ctx)?;
        // each live lease: the re-score, then the server epoch on the row
        for (id, row) in &live {
            w.call(cmd::ZADD, &[s.key(1), score.as_bytes(), id])?;
            w.call(cmd::HSET, &[row.as_slice(), b"lease_epoch", run_id])?;
        }
    }
    Ok(ValkeyValue::Array(failed))
}

The mechanism — the fence, the cached clock, the epoch stamp — is taught in One transition, two executors.

Pattern & implementation

the pattern (Redis Patterns Applied)

A batch over a sorted set should read its reference point — the clock — once, so every member is scored against the same instant, not a drifting one. Reliable queues teaches the reliable-claim move; the one-clock-many-members angle is the near side of the door.

the implementation (echo_mq)

EchoMQ.Jobs.extend_locks/4 runs @extend_locks2, which reads the server TIME once and re-scores every member whose row, token and membership check out, returning the failed list — where the single extend_lock/5 errors on each.

The pattern says: read the clock once, score the many against it. The implementation is extend_locks/4 — one TIME, many re-scores, and the failed list for the rest.

Recap — the many, under one clock

Batch lease extension re-scores the many under one clock read: gate every id, walk the held pairs, re-lease each live member at the shared now + lease, and return the ids that could not be extended: no row, a stale token, or no longer in active. The single verb errors on a bad id; the batch collects it. Natively, EMQ.EXTENDS takes the same keys and arguments, checks everything before it writes, and stamps lease_epoch. That closes the Batches module.

Back to the module: Batches — the three bulk verbs and the one rule they share. The next module in The Queue covers the control plane beyond enqueue and claim.

References

Where the batch lease extension is grounded, and the commands the script uses.

Sources

Valkey — ZADD — the active-set re-score that extends a lease (the score is the deadline).
Valkey — ZSCORE — the membership guard: a missing member answers nil.
Redis — EVALSHA — the run-by-SHA dispatch the batch script runs under.
Valkey — HSET — the row write whose attempts field is the fencing token the script checks.
Valkey — Documentation — the substrate of record EchoMQ is backed by.

Related in this course

Batches — the module this dive belongs to.
Bulk flows — compose many flows in one call, flow by flow.
Claim & the lease — the lease the active-set score is, claimed for one job.
Cancellation & checkpoints — the single extend_lock/5 as a checkpoint.
One transition, two executors — the native mechanism: checks before writes, the cached clock, lease_epoch.
redis-patterns · Reliable queues — the near side of the door.