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.
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 = [the ring, the active set, the base emq:{q}:] — every key declared
    keys = [
      Keyspace.queue_key(queue, "ring"),
      Keyspace.queue_key(queue, "active"),
      Keyspace.queue_key(queue, "")
    ]

    # ARGV = [the lease window in ms]
    argv = [Integer.to_string(lease_ms)]

    case Connector.eval(conn, @gclaim2, keys, argv) do
      {:ok, []} -> :empty
      # the group rides back beside the job, so the worker settles the right lane
      {:ok, [id, payload, att, group]} -> {:ok, {id, payload, att, group}}
      other -> other
    end
  end
end

-- the @gclaim2 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
-- the lane key is derived from the declared root KEYS[3], on the queue's slot
local lane = KEYS[3] .. 'g:' .. g .. ':pending'
-- SERVE: pop this lane's oldest id. An emptied lane leaves the ring and the
-- claim answers empty; the next admission to it returns it 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 = KEYS[3] .. 'job:' .. id
-- the ghost guard: an id with no row is consumed by the pop; nothing is leased
-- or counted, the lane leaves the ring only if it emptied, and the claim
-- answers empty
if redis.call('EXISTS', jk) == 0 then
  if redis.call('ZCARD', lane) == 0 then redis.call('LREM', KEYS[1], 0, g) end
  return {}
end
-- 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[1]), 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 it is serviceable again.
local act = redis.call('HINCRBY', KEYS[3] .. 'gactive', g, 1)
local lim = redis.call('HGET', KEYS[3] .. '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, when the lane is unpaused and still holds work, 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.

Lanes run only as scripts

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