Who moves time today

Two sweeps, sent by a client

# the two sweeps on a module-loaded node, from a wire 6 connection or a raw client: the EVALSHA tails, the sha replaced by the name
EMQ.PROMOTE 2 emq:{orders}:schedule emq:{orders}:pending emq:{orders}: 100
# the reply: the number of due members selected, ghosts included
EMQ.STALLED.CHECK 3 emq:{orders}:active emq:{orders}:pending emq:{orders}:dead emq:{orders}: 1 100
# the reply: [[recovered ids], [dead ids]]

// trafikey — EMQ.PROMOTE: one clock reading, the due members by score, then one write pass
pub(crate) fn promote(ctx: &Context, args: Vec<ValkeyString>) -> ValkeyResult {
    let s = host::shape("EMQ.PROMOTE", &args, 2)?;
    s.expect_argc(2)?;
    let q = promote_queue(&s).ok_or_else(|| s.layout_error())?;
    let batch = parse_positive(s.arg(2)).ok_or_else(|| errors::positive(s.verb, "batch"))?;
    let caller = host::fence(ctx)?;
    let (schedule, pending) = (s.key(1), s.key(2));

    // Read pass (rule 15): one `now`, then Lua's `ZRANGEBYSCORE KEYS[1] -inf now LIMIT 0 batch`.
    let (now, limit) = (host::now_ms()?.to_string(), batch.to_string());
    let selected = match host::read(
        ctx,
        cmd::ZRANGE,
        &[
            schedule,
            b"-inf",
            now.as_bytes(),
            b"BYSCORE",
            b"LIMIT",
            b"0",
            limit.as_bytes(),
        ],
    )? {
        Reply::Array(items) => items,
        _ => return Err(errors::emq(errors::HOST_REPLY)),
    };
    // … each selected member: TYPE, the lane check, a ghost or a live row
    if members.is_empty() {
        return Ok(ValkeyValue::Integer(0));
    }
    // … pending is typed and the members counted, then the write guard
    let w = host::begin_writes(ctx)?;
    apply(&w, &promote_plan(&q, schedule, pending, &members))?;
    Ok(ValkeyValue::Integer(count))
}

A sweep is a command like any other: it runs when it arrives, moves what its one reading makes due, and is gone.

One clock reading per call

// trafikey — host::now_ms: the command's cached server time, the value TIME reports, read once
/// The server's millisecond time (G0.8): `server.ustime / 1000`, the value TIME reports and
/// frozen for the command, so it equals Lua's `t[1]*1000 + floor(t[2]/1000)`. Read it once.
pub fn now_ms() -> Result<i64, ValkeyError> {
    let micros = api(unsafe { raw::RedisModule_CachedMicroseconds })?;
    Ok(unsafe { micros() } / 1000)
}

-- EchoMQ.Jobs @promote — one TIME call, then the due members by score, up to the batch
local p = ARGV[1]
local t = redis.call('TIME')
local now = t[1] * 1000 + math.floor(t[2] / 1000)
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)
  local jk = p .. 'job:' .. id
  -- … a lane row joins its lane (and the ring, with a wake, when the lane can be served); a flat row joins pending
  redis.call('HSET', jk, 'state', 'pending')
end
return #due

The server supplies a reading, never a wake-up: how late a due job leaves schedule is set by how often a client calls and how much each call may move.

The processes that send them

# echo_mq — EchoMQ.Consumer's beat: recover, promote, drain, then park until a wake or the beat
defp loop(s) do
  check_control()
  exit_if_ttl_elapsed(s)

  with {:ok, s} <- recover(s),
       {:ok, s} <- promote(s),
       {:ok, s} <- drain(s) do
    # … a healthy beat resets the degraded streak
    s = reset_degraded(s)
    park(s)
    loop(s)
  else
    # … a wire error on any leg skips the rest of the beat and backs off
    {:degraded, s} ->
      Process.sleep(degraded_sleep(s))
      loop(s)
  end
end
# … the recovery leg: @reap by default, @sweep_stalled with recovery :stalled, neither with :off
defp promote(s) do
  case Jobs.promote(s.conn, s.queue, s.pump_batch) do
    {:ok, _} -> {:ok, s}
    {:error, reason} -> {:degraded, degrade(s, :promote, reason)}
  end
end

# echo_mq — EchoMQ.Stalled, the opt-in sweep process: a client-side timer, re-armed after every sweep
@impl true
def handle_info(:sweep, s) do
  _ = check(s.conn, s.queue, s.opts)
  {:noreply, arm(s)}
end

defp arm(s) do
  Process.send_after(self(), :sweep, s.interval_ms)
  s
end

Today every clock that moves a job runs in a client: a beat, a tick or an interval in an Elixir process on either wire, a timer in the Go client’s worker, or a person at a raw client; the server only answers what each call reads.

Pattern & implementation

The next dive turns from time to waiting: how a parked worker learns that work arrived.

References