Raxol. Harness. StreamCadence
(Raxol v2.6.1)
View Source
The streaming render cadence layer: decouples token ingest (network rate, unbounded) from render egress (cadence-throttled, bounded per flush) so a live agent's token stream never floods the terminal or starves input.
1. What this is
When a live agent streams LLM tokens, Raxol.Agent.Backend.HTTP.stream/2
produces text-chunk deltas at network rate -- bursts of hundreds per
second are normal. Painting on every delta floods the terminal
(redundant repaints of the same live tail) and starves input handling
in whatever process applies the paint. This module sits between that
producer and a Raxol.Harness.Surface-shaped consumer (apply deltas
to the live tail, repaint the footer): unbounded ingest via
ingest/2 (the GenServer mailbox is the buffer -- it never blocks
the producer; load-shedding, not backpressure: excess above
:max_pending is dropped, not pushed back), cadence-throttled egress
via Raxol.Harness.CadencePolicy (the pure decision function),
bounded per-flush drain (Raxol.Harness.CadencePolicy.drain_count/2).
To be plain about the flow-control model: this layer never applies backpressure toward the producer. It is a cadence + load-shedding layer -- ingest is always accepted, and excess above the watermark is shed per section 3. A supervision instrument must stay alive and current: the newest deltas ARE the live tail's value, and blocking (call-based) ingest would stall the SSE reader process and cascade into transport timeouts -- the wrong failure mode. Visibly lossy above the watermark beats dead.
2. The owner-consumption contract (the seam)
Out of the box this module enforces NO input priority. The
default :input_check always returns false, so the source-side
hold described below is OFF until a caller wires a real check. The
:input_check seam is MANDATORY, not optional, for the input-first
behavior -- and the other mandatory half is the owner handling its
input messages before {:render_batch, ...} messages in its own
receive.
Flush delivery is a message the owner consumes -- by default
{:render_batch, batch} sent to the :owner pid -- never a blocking
call into the owner. This server never forces the owner to handle a
batch synchronously.
Input priority is enforced in plain OTP terms, not by this module
reaching into the owner's mailbox: the owner (the future live-session
loop) is responsible for keeping input ahead of paints by handling
its own input messages before {:render_batch, ...} messages --
selective receive matching input patterns first, or draining pending
input before applying a batch. Belt-and-suspenders: the :input_check
option lets the cadence policy hold token flushes at the source
(:yield_to_input) while input is pending, so batches don't even
enter the owner's mailbox ahead of input in the common case.
A custom :sink must preserve the non-blocking property (a
send, never a GenServer.call into the owner) -- this server has
no timeout protection against a slow sink.
3. Ordering / loss contract
The guarantee is: lossless below the :max_pending watermark;
explicit, in-band loss above it.
Below the watermark, batches arrive in ingest order and the
concatenation of every delivered batch equals the exact ingest
sequence -- no delta dropped, duplicated, or reordered. Each batch
carries at most Raxol.Harness.CadencePolicy.max_drain_per_flush/0
deltas (or the configured override).
At or above the watermark, the OLDEST pending deltas are shed
(drop-oldest: the queue head is history, the queue tail is the live
view). Each shed emits an [:raxol, :harness, :stream_cadence, :overflow] telemetry event at drop time (measurements
%{dropped, pending_count}, metadata %{max_pending}), and the next
flushed batch begins with a {:cadence_dropped, n} marker sitting
exactly at the position of the loss (the dropped items were the queue
head). The marker is NOT counted against the drain bound, so a batch
carries at most max_drain_per_flush + 1 elements when loss
occurred. {:cadence_dropped, non_neg_integer()} is a documented
RESERVED element type in the batch stream; consumers must handle it.
A naive sink that assumes every batch element is a delta (say, one
that joins binaries) will fail loudly at the first loss -- that is
deliberate: in a supervision instrument, silently rendering a
gapless stream over shed data is worse than a crash that names the
unhandled loss.
The pending queue is therefore always bounded.
4. Throughput ceiling (deliberate design point)
Steady-state egress is bounded at max_drain_per_flush items per
flush_interval_ms -- the shipped defaults put that at 32 items per
16ms, i.e. 2,000 deltas/sec, comfortably above any real LLM token
rate. This diverges from the Rust reference design, where message
application and painting are decoupled (unbounded application,
throttled paint): here the flush message itself IS the paint
trigger, so the drain bound doubles as an application bound. An
instantaneous 1,000-delta backlog drains in roughly 0.5s of bounded
paints at the default cadence. Call flush_now/1 at end-of-stream
(the SSE :sse_done moment) to skip that wait and flush the tail
immediately instead of leaving it up to 16ms stale.
In the other direction, rendering waits on input for at most ~one
frame interval: the consecutive-yield budget
(Raxol.Harness.CadencePolicy.max_consecutive_yields/0) forces a
flush through even under continuous input.
5. Not wired yet
The live session loop that would own one of these servers, apply
{:render_batch, batch} to a live tail, and repaint the footer does
not exist yet. This module and the contract above ship standalone,
ahead of that wiring.
Summary
Functions
Returns a specification to start this module under a supervisor.
Forces an immediate full drain of everything currently pending,
ignoring cadence and the input gate. Delivers all pending items now,
as consecutive sink calls of at most max_drain_per_flush items
each, in ingest order. Intended for the end-of-stream moment (SSE
:sse_done) so the tail never sits cadence-stale.
Callback implementation for Raxol.Core.Behaviours.BaseManager.handle_manager_call/3.
Enqueues delta for eventual flush. Always a cast -- the unbounded
mailbox is the buffer, so ingest never blocks the producer
(load-shedding, not backpressure: excess above :max_pending is
dropped, not pushed back -- see moduledoc section 3).
Functions
Returns a specification to start this module under a supervisor.
See Supervisor.
@spec flush_now(GenServer.server()) :: :ok
Forces an immediate full drain of everything currently pending,
ignoring cadence and the input gate. Delivers all pending items now,
as consecutive sink calls of at most max_drain_per_flush items
each, in ingest order. Intended for the end-of-stream moment (SSE
:sse_done) so the tail never sits cadence-stale.
Callback implementation for Raxol.Core.Behaviours.BaseManager.handle_manager_call/3.
@spec ingest(GenServer.server(), term()) :: :ok
Enqueues delta for eventual flush. Always a cast -- the unbounded
mailbox is the buffer, so ingest never blocks the producer
(load-shedding, not backpressure: excess above :max_pending is
dropped, not pushed back -- see moduledoc section 3).