Fair lanes & the ring

The shape — lanes, a ring, and the books

A lane per identity, a ring of the serviceable ones, two small books for concurrency. Fairness is these structures — not a hash over the id.

The grouped claim, in two beats

# echo_mq — EchoMQ.Lanes
# claim/3 reads the queue-wide pause flag first (a queue-wide pause stops the
# grouped claim too — distinct from pause/3, which parks ONE lane), then runs
# @gclaim. KEYS = [ring, active]; ARGV = [the queue base "emq:{q}:", the lease].
def claim(conn, queue, lease_ms) when is_integer(lease_ms) and lease_ms > 0 do
  if EchoMQ.Jobs.paused?(conn, queue) do
    :empty
  else
    keys = [Keyspace.queue_key(queue, "ring"), Keyspace.queue_key(queue, "active")]
    argv = [Keyspace.queue_key(queue, ""), Integer.to_string(lease_ms)]

    case Connector.eval(conn, @gclaim, keys, argv) do
      {:ok, []} -> :empty
      {:ok, [id, payload, att, group]} -> {:ok, {id, payload, att, group}}  # the group rides back
      other -> other
    end
  end
end

-- the @gclaim script — rotate one step, serve the head, lease, keep the books
-- ROTATE: LMOVE pops the head group of the ring and pushes it to the tail in
-- one atomic move, so the NEXT claim starts at the next group — round-robin,
-- constructed. The returned g is the group to serve now.
local g = redis.call('LMOVE', KEYS[1], KEYS[1], 'LEFT', 'RIGHT')
if not g then return {} end
local lane = ARGV[1] .. 'g:' .. g .. ':pending'
-- SERVE: pop this lane's oldest id. An emptied lane leaves the ring and the
-- claim answers empty (a producer will return it to the ring with a wake).
local popped = redis.call('ZPOPMIN', lane)
if #popped == 0 then
  redis.call('LREM', KEYS[1], 0, g)
  return {}
end
local id = popped[1]
local jk = ARGV[1] .. 'job:' .. id
-- LEASE: the same as the flat claim — attempts is the fencing token; the
-- active-set score IS the lease deadline, read off the server clock.
local att = redis.call('HINCRBY', jk, 'attempts', 1)
redis.call('HSET', jk, 'state', 'active')
local t = redis.call('TIME')
local now = t[1] * 1000 + math.floor(t[2] / 1000)
redis.call('ZADD', KEYS[2], now + tonumber(ARGV[2]), id)
-- BOOKS: count this claim in-flight; if the group is now at its glimit, or its
-- lane is now empty, drop it from the ring until a completion frees a slot.
local act = redis.call('HINCRBY', ARGV[1] .. 'gactive', g, 1)
local lim = redis.call('HGET', ARGV[1] .. 'glimit', g)
if lim and act >= tonumber(lim) then
  redis.call('LREM', KEYS[1], 0, g)
elseif redis.call('ZCARD', lane) == 0 then
  redis.call('LREM', KEYS[1], 0, g)
end
return {id, redis.call('HGET', jk, 'payload'), att, g}

One LMOVE turns the rota; one ZPOPMIN serves the head. Fairness is the turn of a list, not the output of a hash.

The concurrency limit parks a lane

A parked lane keeps its backlog and its in-flight work — it only leaves the rota. A completion that drops the group's in-flight below the limit returns it to the ring with a wake, so a parked consumer picks it up at once.

The control verbs

Every verb that makes a lane serviceable again pushes a wake onto emq:{q}:wake, so a parked consumer returns the moment work appears — the loop the next dive reads.

Pattern & implementation

The pattern says: round-robin over per-producer queues. The implementation is one LMOVE per claim, turning a ring of lanes.

Recap — the rota is a data structure

Next in this module: The consumer loop — reap, promote, drain, park, and a raising handler that becomes a typed retry.

References