Trafikist · module 04

Blocking, timers and the bus

Today a client drives every sweep: the module answers each command and returns, and nothing in it waits or ticks; it appends to the bus only when a client sends a bus verb.

A job system has three kinds of work that no single job request carries. Waiting: a worker with nothing to do has to learn that work arrived. Time: a scheduled job has to move to pending when it falls due, and an expired lease has to be recovered. Broadcast: the bus has to append, publish and count failures. In EchoMQ all three run in client processes today. Elixir processes send the sweeps on a beat, park on Valkey's own blocking pop, and compose the bus from Valkey built-ins; trafikey serves EMQ.PROMOTE and EMQ.STALLED.CHECK to whichever client sends them, an Elixir connection opened with wire: 6 among them, and registers three bus verbs that no Elixir code sends. Pick a duty to see who carries it today. Where the specifications take each of the three is marked in a forward block, after the present-tense lesson it extends.

The module answers and returns Who moves time today The bus, from built-ins The dives References

The module answers and returns

A Valkey module can park a client until a key is signalled ready, signal a key itself and arm a timer that fires inside the server. trafikey does none of the three. Its load gate refuses the load when a host function on its list is missing, and the list names no blocked-client, ready-signal or timer function; neither does any call in the crate's source, and no row of its registration table carries the blocking command flag. Every EMQ.* command reads, checks, writes and replies inside one call. The two commands that move time-driven work are ordinary write commands with a bound on how much one call does:

// trafikey — registry::TABLE, two rows: the time-driven sweeps are plain write commands
// a client sends, each bounded by an argument (batch, limit)
    emq_promote => crate::emq::scheduling::promote, Row { name: c"EMQ.PROMOTE", family: Family::Emq, arity: 6,
        flags: "write", keys: EVAL_KEYS, acl: c"write slow", caps: &["promote"],
        summary: Some("Move due scheduled flat jobs to pending, bounded by batch (emq6.8 G8, @promote)"),
        complexity: Some("O(M log N) for M due jobs"), since: Some("0.1.0") };
// …
    emq_stalled_check => crate::emq::recovery::stalled_check, Row { name: c"EMQ.STALLED.CHECK", family: Family::Emq, arity: 8,
        flags: "write", keys: EVAL_KEYS, acl: c"write slow", caps: &["stalled.check"],
        summary: Some("Counted stall sweep: recover or dead-letter expired leases; restart expiries exempt (emq6.8 G15, @sweep_stalled)"),
        complexity: Some("O(M log N) for M expired leases"), since: Some("0.1.0") };

EMQ.PROMOTE moves the due flat members of schedule to pending, at most batch of them, and EMQ.STALLED.CHECK recovers or dead-letters the expired leases in active, at most limit of them. A due job stays in schedule until a client sends the first, and an expired lease stays in active until a client sends the second or its holder settles it. Each reads the server's clock once, when the call arrives. The first dive follows those two calls and the clock they read.

A module command runs when a client sends it and ends with its reply; between calls, nothing in the module runs.

Who moves time today

The Elixir library sends the sweeps from its own processes, as scripts on its default wire: 4 connection. EchoMQ.Consumer runs a beat, one second by default: recover expired leases (with the reaper, unless configured otherwise), promote due jobs, drain the lanes, then park. EchoMQ.Metronome takes over the beat and the single park for a pool of consumers. EchoMQ.Pump promotes on a tick of its own, and EchoMQ.Stalled can run the counted sweep on an interval. The waiting is Valkey's own: a parked consumer sends BLPOP on the queue's wake list, with the beat as its timeout. The durable bus's reclaim is a built-in too, sent by EchoMQ.StreamConsumer on each pass of its loop.

On a connection opened with wire: 6, EchoMQ.Jobs.promote/3 sends EMQ.PROMOTE and EchoMQ.Stalled.check/3 sends EMQ.STALLED.CHECK in place of their scripts. EchoMQ.Jobs.reap/2 and the lane claim a consumer drains with have no verb and are refused, so a beat that runs the reaper stops there and backs off, and a consumer claims nothing. No production caller opens such a connection today.

Who carries each duty today · pick a duty

promote due jobs · schedule → pending
the library: EchoMQ.Jobs.promote/3 evaluates @promote, sent each beat by a standalone EchoMQ.Consumer or by EchoMQ.Metronome for a pool, and each tick by EchoMQ.Pump; up to 100 members a call by default
trafikey: EMQ.PROMOTE makes the same move for flat rows, bounded by batch, and a due lane row refuses the whole call; on a wire: 6 connection Jobs.promote/3 sends it in place of @promote: EchoMQ.Pump sends it each tick, and a beat sends it only when it does not run the reaper, which is refused
between calls, a due job stays in schedule

Every sweep today has a client process behind it and a beat or an interval in that process; the server contributes one clock reading per call.

The bus, composed from built-ins

The library's bus sends no module command. EchoMQ.Bus and EchoMQ.Stream evaluate no script either: every bus operation is one or more Valkey built-ins the library sends, on either wire. PUBLISH carries the ephemeral tier, XADD the durable tier; XREADGROUP, XACK and XAUTOCLAIM serve the consumer groups; XTRIM applies retention. Most of these are one command, and so one server turn, already. Three operations are compositions: the dead-letter failure count is an INCR followed by a separate EXPIRE, a batch append is a pipeline whose replies the library drops, and a park appends to the dead-letter stream before the consumer acknowledges the source entry. The envelope's field layout comes from the Elixir codec alone, since the server stores whatever field pairs an XADD carries.

The module registers three bus verbs that do the compositions' work in one server turn: EMQ.APPEND writes a batch whole or not at all, EMQ.PUBLISH appends one envelope after checking its fields, and EMQ.FAILURES.INCR sets a count and its TTL together. No Elixir code sends them, on either wire. The third dive breaks the three compositions and reads the three verbs.

The bus runs on Valkey's own commands; where an operation needs two of them, the gap between the two is the client's.

The dives

Three dives, each starting from what runs today in trafikey and echo/apps/echo_mq, then marking where the specifications take it.

built page ready to read

01built

Who moves time today

EMQ.PROMOTE and EMQ.STALLED.CHECK run only when called; the native clock is the command's cached time, read once.

02built

The blocking claim

A wake only asks a waiter to check again; every path that makes work claimable has to ring, or the waiter learns at its timeout.

03built

The native bus

Most bus operations are one Valkey built-in already; three compositions leave a window between commands, and three native verbs write in one turn.

Pattern & implementation

the pattern (Redis Patterns Applied)

Park, don't poll: a worker blocks on a key and a producer wakes it; a sorted set scored by fire time is the clock, and a sweep promotes what is due. Blocking vs polling and The delayed queue teach both.

the implementation (EchoMQ and trafikey)

Elixir processes park on BLPOP of the wake list and send the promotion and recovery scripts on a beat; trafikey serves the same sweeps as commands, sent by a client, each reading the server's clock once.

Take the dives in order: time first, then waiting, then the bus.

Up next

→built

Who moves time today

Start the module: the two sweeps, their senders and the clock each reads.

←built

Failures and recovery

The native stall sweep and the reaper with no native twin, the lesson this module assumes.

After this module, TRA and native values reads the module's second command family, and the workshop runs one transition on both executors.

References

Sources

Valkey — Modules API reference — the blocked-client, key-readiness and timer calls a module can make, and the command flags.
Valkey — BLPOP — the blocking pop a parked consumer sends on the wake list.
Valkey — ZRANGEBYSCORE — the selection by score that every promotion and recovery sweep runs.
Valkey — XAUTOCLAIM — the stream reclaim the durable bus sends on each loop pass.
Valkey — EXPIRE — the second command of the dead-letter failure count.

Related in this course

The job core, natively — the previous module: the thirteen verbs, two of which this module follows.
The native module — the Protocol's introduction to the module and its command form.
The consumer loop — the Queue's reading of the beat that recovers, promotes, drains and parks.
Scheduling and recurrence — the schedule set and the promotion script in the Queue.
The consumer group — the Bus pillar's group reader, its blocking read and its reclaim.
redis-patterns · Blocking vs polling — park, don't poll: the near side of the door.