R6.06 · Flow Control & Scale — the chapter workshop

One bus, fair under load.

The chapter capstone: codemojex under load. R6 taught five ways to keep a queue stable when load arrives — rate, fairness, groups, batches, worker concurrency — and codemojex, the Telegram code-breaking game, applies all five at once on the real EchoMQ bus.

Grounding: echo/apps/codemojex (Codemojex.RateLimiter, Codemojex.Guesses.submit/3, Codemojex.NotificationWorker, Codemojex.ScoreWorker, Codemojex.CommandWorker, Codemojex.Notifier) on echo/apps/echo_mq (EchoMQ.Lanes, EchoMQ.Jobs.enqueue_in/5 & enqueue_many/4, EchoMQ.Consumer, EchoMQ.Metrics.get_rate_limit_ttl/3) — every surface verified on disk. The engine is Valkey.

The pipeline under load. A guess or a notification flows in through four sequential gates, each an R6 technique on a real codemojex surface; the batch path (R6.04) is the cross-cutting fifth. Hover or select a gate to read what it does.

readout · Select a gate to read its R6 technique and the real codemojex surface that applies it. Traffic enters left, settles right; over-budget sends branch down to a durable deferral.

The five flow-control techniques are not five systems; they are gates on one pipeline — four in line, the batch path across all of them. The guess that floods in is the same identity that comes out scored, and every gate it passes is a real surface in the consumer.

§1

The scenario — a room floods, the engine stays fair

codemojex/notification_worker.ex

A popular room fills with players hammering guesses; a finished game fans a burst of result notifications back out to Telegram. Two pipelines, one bus. Without flow control, the loudest room starves the quiet ones, the engine sends faster than Telegram allows and gets throttled, and a worker stall backs everything up. R6's five techniques each address one of those failures.

The honest spine is already written into the consumer. Codemojex.NotificationWorker's own moduledoc names three layers of control — Fairness, Rate, Delivery — the delivery side is the same shape as the guess side, so one model fits both:

# Codemojex.NotificationWorker moduledoc — the three layers, verbatim
#   1. Fairness — notifications are enqueued on a fair lane per chat
#      (Codemojex.Notifier), so one chat's burst cannot starve others.
#   2. Rate — before each send the handler takes a token from Codemojex.RateLimiter
#      (global ~30/s, per-chat ~1/s). Over budget ... it re-enqueues the same job
#      with EchoMQ.Jobs.enqueue_in/5 ... and acks — deferred, not dropped.
#   3. Delivery — classifies the Telegram result: success acks; a transient
#      failure re-enqueues with capped backoff; a permanent failure acks-and-drops.

Each section that follows recalls one R6 module, then lands it on the real codemojex surface — rate, then fairness, then groups, then the batch path, then the pool of workers.

§2

R6.01 rate — a token bucket sized for Telegram

codemojex/rate_limiter.ex · R6.01

Rate limiting caps work to a budget per unit of time. R6.01 taught two algorithms with the same job: the fixed-window counter (the real emq:{cm}:limiter, spent down to max over its own PTTL window) and the token bucket (a refilling allowance — bursts up to the bucket, sustained at the refill rate). codemojex applies the token bucket on the delivery side. Codemojex.RateLimiter is sized for Telegram's send limits: a global bucket (about 30 messages per second, burst 30) and a per-chat bucket (about one per second, burst 3), and a send is allowed only when both grant a token. Those rates are the limits Telegram enforces — configuration, not measured throughput.

# Codemojex.RateLimiter.take/2 — the delivery gate (echo/apps/codemojex/lib/codemojex/rate_limiter.ex)
RateLimiter.take(chat)
#=> :ok            both buckets grant — send now
#=> {:wait, ms}    a bucket is empty — the smallest delay before a retry could succeed

The buckets refill lazily from elapsed time, so the limiter holds no timers and one process serves every chat. The NotificationWorker does not block the consumer when over budget: it re-enqueues the same notification with EchoMQ.Jobs.enqueue_in/5 after the reported wait and acks — the notification stays durable on the bus, deferred, not dropped. On the work side, the server-side emq:{cm}:limiter fixed-window counter is the same idea in Valkey, read by EchoMQ.Metrics.get_rate_limit_ttl/3 (the remaining window in ms, the over-budget back-off) — one budget shared across every bot worker because it lives in the engine, not in any process.

codemojex/notification_worker.exfrozen

# the over-budget path — defer, don't block (verbatim shape)
case RateLimiter.take(chat) do
  :ok        -> deliver(job, msg, chat, text, opts, attempt)
  {:wait, ms} -> requeue(msg, ms); :ok       # EchoMQ.Jobs.enqueue_in/5 after ms, then ack
end

The rate gate is the ceiling: the token bucket admits a burst up to the bucket, then holds to the refill rate, and an over-budget send is re-enqueued for later, never dropped — the budget is the same whether it is the in-process bucket or the engine's emq:{cm}:limiter counter.

§3

R6.02 fairness — each player rides their own lane

codemojex/game.ex · R6.02

Fairness under load is keeping every producer's work moving when one floods the queue. A single FIFO line lets one heavy producer starve the rest; R6.02 taught the fix — a queue is a set of lanes, each a per-group pending set, and every claim rotates the ring one step before serving, so service spreads across lanes by construction. codemojex enqueues each guess on the player's lane. Codemojex.Guesses.submit/3 validates the guess, charges the room's currency, then enqueues a branded JOB keyed by the player's PLR:

# Codemojex.Guesses.submit/3 — a guess on the player's fair lane (echo/apps/codemojex/lib/codemojex/game.ex)
job = EchoData.BrandedId.generate!("JOB")
payload = :erlang.term_to_binary({:guess, game, player, guess})
Lanes.enqueue(Bus.conn(), "cm", player, job, payload)   # the PLR is the lane group

The module's own words: "the lane is named by the player's PLR, so the bus rotates service across players and one keyboard masher cannot starve the field." Notifications are fair the same way: Codemojex.Notifier enqueues each notification on a fair lane keyed by chat id, so the bus spreads a chat's notifications behind the rate limit rather than firing them at once — the NotificationWorker's layer 1, fairness per chat. R6.02's retarget holds here: there is no numeric per-job priority; "served more" is a property of the identity (a higher lane weight), not a number stapled to the work, and mint order plus the rota already give fairness.

Fairness keeps the room honest: the loud player and the quiet one each have a lane, the ring serves both in turn, and the flood is bounded to its own lane — never the whole queue.

§4

R6.03 groups — the lane is the tenant

codemojex/game.ex · R6.03

A group is a tenant given its own lane. R6.03 taught two controls over a tenant lane: round-robin across tenants (the ring rotation of R6.02) and a per-group concurrency ceiling, plus parking one tenant without touching the rest of the queue. In codemojex the tenant is a player (a PLR) — its lane is the unit of control. Codemojex.Guesses exposes that control directly, each call delegating to EchoMQ.Lanes:

# Codemojex.Guesses — per-tenant control without touching the rest of the queue (game.ex)
def pause(player),  do: Lanes.pause(Bus.conn(),  "cm", player)
def resume(player), do: Lanes.resume(Bus.conn(), "cm", player)
def depth(player),  do: Lanes.depth(Bus.conn(),  "cm", player)

pause/1 parks one player's lane — held while the rest of the field runs; resume/1 unparks it; depth/1 reads how much work that one lane has pending. One tenant is held while the others run, which is exactly the manuscript's "group-aware pause and resume act on a lane without touching the queue." The concurrency ceiling and the de-ring mechanics (EchoMQ.Lanes.limit/4 → glimit/gactive) are the bus's, taught in the EchoMQ Queue pillar; codemojex reaches the per-tenant control through the lane.

The five gates, one bus. Select a technique to read which codemojex surface applies it and which R6 module it comes from, all over the cm queue.

readout · Select a technique. Rate caps the budget, fairness spreads the rota, groups give per-tenant control, the batch path bundles a burst, and the consumer pool drains the queues — one bus, five gates.

Static verdicts — 1 rate → RateLimiter.take/2 (defer enqueue_in/5) · 2 fairness → submit/3 → Lanes.enqueue/5 · 3 groups → pause/resume/depth/1 · 4 batches → enqueue_many/4 + bclaim/3 (pattern-level) · 5 concurrency → Consumer pool, one ZPOPMIN per claim.

The lane is the tenant: a player's whole pipeline is one lane the system can rotate, weight, pause, or read — control aimed at the identity, not stapled to each job.

§5

R6.04 + R6.05 — the burst path and the pool of workers

echo_mq/jobs.ex · codemojex/game.ex · R6.04 + R6.05

R6.04 taught the pipelined batch: many writes in one wire round-trip, each item atomic and idempotent, the batch not a transaction — partial verdicts are the honest model, by design over MULTI/EXEC rollback. codemojex enqueues one guess per submit/3 today, not a batch — so this technique is pattern-level here, the path the bot could take when a burst of inbound commands arrives. A burst could be admitted in one flush via EchoMQ.Jobs.enqueue_many/4 (a per-item idempotent script EVALSHA'd once per item through EchoWire.Pipe, flushed once, a verdict per item — :enqueued / :duplicate / {:error, :kind} — in input order) and drained in one batch claim via EchoMQ.Lanes.bclaim/3 (rotate to a serviceable lane, serve up to a batch of heads on one shared TIME lease). One round-trip in, one round-trip out.

R6.05 grounds the workers. The BEAM gives cheap concurrency — thousands of processes — but the wire is the shared resource: real parallelism on Valkey is pool width (N sockets), not N processes sharing one connector. The trio is EchoMQ.Consumer (the claim loop that parks on a wake key rather than polling), EchoMQ.Pool (the concurrency primitive), and the per-claim ZPOPMIN fetch (one job per claim, one round-trip), amortized by EchoMQ.Lanes.bclaim/3. codemojex runs a Consumer loop per queue:

Codemojex.ScoreWorker.handle/1 drains the cm guess queue — "EchoMQ.Consumer drains the guess queue through Lanes.claim, the player id arriving as the lane group." It scores, writes a GES, and for a classic game publishes a scored event.
Codemojex.NotificationWorker.handle/1 drains the cm.notify queue and delivers each notification under the three layers of control.
Codemojex.CommandWorker.handle/1 drains the cm.bot.commands lane — keeping inbound command handling on the bus makes it durable, per-chat ordered, and replayable, with the reply going back out through Codemojex.Notifier so it inherits the same rate limiting and retries.

Each claim is one ZPOPMIN round-trip — the per-claim fetch is the throughput ceiling, and pool width sets the floor. A busy event (many rooms guessing at once) is the case where batch claim and pool width matter.

echo_mq/lanes.ex · codemojex/game.exfrozen

submit/3   Lanes.enqueue(conn, "cm", player, JOB, payload)   ← one guess, the PLR lane
claim/3    LMOVE ring ring + ZPOPMIN lane                      ← one job per round-trip
bclaim/3   k × ZPOPMIN on one shared TIME lease               ← amortize the fetch (R6.04)
Consumer   ScoreWorker / NotificationWorker / CommandWorker    ← a loop per queue, parked on a wake

Notes on ValkeyEvery key of the cm queue — the lanes, the ring, the limiter counter — carries the queue's {cm} hash tag, so they all hash (CRC16 of the brace bytes) to one of the 16384 cluster slots and co-locate. That co-location is what keeps a multi-key claim or limiter script legal (no CROSSSLOT) and makes the limiter counter one shared key every bot worker increments — valkey.io/topics/cluster-spec.

The batch path is the burst option and the pool is the throughput floor: one round-trip admits a burst, one shared lease drains a batch, and a pool of parked consumers turns cheap BEAM concurrency into real parallelism on the wire.

§6

Putting it together — one pipeline, every gate

the application · the five gates

Trace one guess and one notification end to end:

A guess enters at submit/3, is charged and validated, and is enqueued on the player's fair lane (Lanes.enqueue, R6.02). The ring rotates and a ScoreWorker Consumer claims one head (Lanes.claim, R6.05), scores it, and writes a GES. If that player's lane is paused (R6.03), the ring skips it.
A notification enters at Notifier.notify on a fair lane per chat (R6.02), is claimed by the NotificationWorker Consumer (R6.05), and hits the rate gate — RateLimiter.take/2 (R6.01). On :ok it is delivered; on {:wait, ms} it is re-enqueued with EchoMQ.Jobs.enqueue_in/5 after the wait and acked — deferred, not dropped.

Five techniques, one bus: the rate gate is the ceiling, the lanes keep it fair, the groups give per-tenant control, the batch path is the burst option, and the pool of consumers is the floor. The identity — the branded JOB, GES, NOT, PLR ids — is the thread that ties the guess to its lane, its score, and its receipt.

the chapter

Five flow-control techniques: rate-limit to a budget (R6.01), fair lanes drained by a rota (R6.02), per-tenant groups (R6.03), a pipelined burst path (R6.04), and a pool of claim loops (R6.05). Each keeps the queue stable against a different failure under load.

codemojex’s pipeline under load

Codemojex.RateLimiter.take/2 defers over-budget sends with EchoMQ.Jobs.enqueue_in/5; Codemojex.Guesses.submit/3 enqueues each guess on the player’s PLR lane (EchoMQ.Lanes.enqueue/5), pause/1 / resume/1 / depth/1 give per-tenant control, and a pool of EchoMQ.Consumer loops (ScoreWorker, NotificationWorker, CommandWorker) drains every queue — all over the cm queues.

The five flow-control techniques are not five systems; they are five gates on one pipeline. The guess that floods in is the same identity that comes out scored, and the chapter’s whole catalog is working at once on one bus.

/echomq/queue

References

sources · related

Sources

Valkey — Cluster specification — the {cm} hash tag forces the queue’s keys onto one of the 16384 slots so a multi-key claim or limiter script stays legal.
Valkey — INCR — the atomic counter behind a fixed-window rate limit; the spend-down the emq:{cm}:limiter cap is built on.
Valkey — PTTL — the time until the limiter window resets; the over-budget back-off EchoMQ.Metrics.get_rate_limit_ttl/3 returns.
Valkey — ZPOPMIN — the per-claim pop behind EchoMQ.Lanes.claim/3; one job per round-trip, the ceiling pool width amortizes.
Valkey — Pipelining — many commands in one round-trip, the wire mechanic behind the burst path EchoMQ.Jobs.enqueue_many/4 takes.
Stripe — Scaling your API with rate limiters — the token-bucket and fixed-window algorithms in production, the family Codemojex.RateLimiter applies.

Related in this course

R6.01 · Rate limitingthe budget-per-window pattern the delivery gate applies
R6.02 · Fairness under loadthe rota that keeps one player from starving the field
R6.03 · Groups & multi-tenant fairnessthe lane as the tenant; pause and resume one without the rest
R6.04 · Batches & pipeliningthe one-round-trip burst path the bot could take
R6.05 · Worker concurrencythe pool of claim loops that sets the throughput floor
EchoMQ · the Queuethe scaling layer in depth
BCS · the Busthe fair-lanes architecture the figures draw from