Module stream 

Real-time access-record stream (DW-121, feature analysis section 5 “Platform”): the record plumbing that mirrors the notice plumbing of DW-044. Where the event bus carries a handful of discrete operational events per minute (breaker transitions, ejections, config publishes), this pipeline carries ONE record per completed request — the raw firehose an operator points at their own warehouse or SIEM when the embedded analytics store (DW-043) is not the system of record. It complements, never replaces: the two are configured independently (a deployment can run either, both, or neither) and share nothing at runtime — different channels, different tasks, different failure domains.

Placement

The stream lives in the events domain beside the webhook deliverer because it IS an outbound delivery pipeline: the sink reuses the DW-044 delivery engine (webhook::deliver_with_retry — one total timeout per delivery, exponential backoff, the transient status set) and consumes the access record type from observability, both of which the events domain may import. The analytics domain cannot (its dependency set is config/observability/extensions), and the firehose needs nothing from the SQLite store — moving it there would force a dependency-graph change to share an engine that already lives one module over.

Wire format (stable shape)

One NDJSON body per flushed batch — one JSON object per line, one line per record, application/x-ndjson:

{"id":"rec-18f3c2a1b9d0-00000a","gateway":"dwara-8213-18f3c2910b07",
 "timestamp":"2026-08-30T09:00:00.123Z","request_id":"req-...",
 "listener":"edge","route":"billing","consumer":"acme",
 "upstream":"billing-v1","endpoint":"10.0.0.4:8443","method":"GET",
 "path":"/v1/invoices","status":200,"duration_ms":4.2,"attempts":1,
 "rate_limited":false,"broken":false,"shed":false,
 "dimensions":{"plan":"gold"}}

The field list is the access record’s redacted-by-construction set: path never carries a query string, there are no headers, no credentials, and dimensions carries only config-declared header-sourced tags (DW-043’s grammar, capped at 16 x 128 bytes). id is process-unique and monotonic (rec-<hex unix ms>-<hex seq>) and gateway is the process instance label — together they give a receiver per-instance ordering even across restarts. A record whose serialized line exceeds MAX_RECORD_BYTES (only possible via absurd path/dimension lengths — the same bound class as a webhook envelope) is dropped and counted, never truncated: a receiver should never see a malformed line.

Batching: one delivery per flushed batch

The flusher accumulates records and flushes on the first of: batch_max records queued, the batch byte cap MAX_BATCH_BYTES reached, or flush_ms elapsed since the batch’s first record (all three read live from the current generation). ONE delivery — with its whole retry cycle — is made per batch, not per record: a warehouse ingest endpoint wants throughput-shaped traffic, and per-record delivery would multiply the sink’s load by the batch factor for zero information gain. Deliveries are made STRICTLY IN ORDER, inline in the flusher task: a record firehose’s arrival order is a real receiver expectation (unlike alert events, which the deliverer dispatches as independent tasks). The cost of that ordering is deliberate and bounded: one slow batch holds the queue back by at most its timeout_ms budget, after which it fails and the queue moves on — and a queue that outgrows its buffer bound drops at OFFER time (counted), never blocks the request path.

Emission contract: never backpressure the dataplane

AccessRecordStream::offer is an enabled-flag check plus a bounded-channel try_send. When the queue is full (or the flusher is stalled behind a slow batch) the record is DROPPED and the drop counted (AccessRecordStream::dropped, surfaced as the dwara_access_records_dropped_total gauge at scrape time). There is no blocking wait and no unbounded buffer: a dead sink degrades the stream’s completeness counters, never the gateway’s latency. When the current generation has no compiled sink the flag is off and offer returns before allocating anything — an unconfigured stream costs one relaxed atomic load per request.

The sink seam

RecordSink is deliberately small: one method, one batch, one boolean answer. webhook is the one shipped implementation; a Kafka producer is the documented second slot, deferred by the lean-deps rule (the same decision that keeps Parquet out of DW-156 backlog territory: a sink that drags a client library must earn its dependency weight). Sinks are compiled per config generation (a kafka variant would compile its client the same way) and pushed to the flusher over a watch channel — a reload retargets the stream with no restart, mid-batch boundaries aside.

Lifecycle

The channel and its capacity are constructed once per process (the capacity is a boot-time property; dwara-bin always constructs the stream so a live reload can arm it) and the flusher is one background task stopped by the shutdown watch, which drains what is already queued into one final flush attempt before returning. The gateway is not a durable queue: a record offered but not yet delivered at process death is lost with only its counters as the witness.

Structs

AccessRecordStream

The bounded record queue plus its counters and enabled flag. Share via Arc; the dataplane holds it behind an ArcSwapOption set once at boot.

KafkaRecordSink

The Kafka batch sink (SCALE-08, #187, analytics_stream.sink.kafka): produces each flushed batch as NDJSON messages to a Kafka topic via a Kafka REST Proxy. The REST Proxy endpoint receives the batch as an HTTP POST with the Content-Type: application/vnd.kafka.binary.v2+json media type; each NDJSON line becomes a Kafka message value (base64 encoded per the REST Proxy spec). Reuses the DW-044 webhook delivery engine’s retry/budget shape — one delivery (with its retries) per batch.

StreamRecord

One completed request as it crosses the stream channel: the access record’s owned copy plus the identity the flusher stamps into the envelope. Constructed on the request-completion path (the same allocation class as the analytics store’s raw-record copy).

StreamTargets

The per-generation compiled stream state (DW-121): pushed to the flusher over a watch channel by the dataplane’s refresh. An empty sinks list is the disabled state — the offer path checks a flag the refresh derives from exactly this, so an unconfigured stream never queues a record.

WebhookRecordSink

The webhook batch sink (DW-121, analytics_stream.sink.webhook): one POST per flushed batch, body the NDJSON lines, media type application/x-ndjson, delivered through the shared DW-044 engine (one total timeout per batch, exponential backoff, the transient status set — see events::webhook for the contract).

Constants

MAX_BATCH_BYTES

Hard cap on one flushed batch (bytes), the byte half of the flush trigger (with batch_max records, whichever comes first). Batches are delivered inline and in order, so this is also the flusher’s per-delivery memory bound.

MAX_RECORD_BYTES

Hard cap on one serialized record line (bytes). The field list is bounded (path length and 16x128-byte dimensions are the only growth axes), so this only binds absurd shapes; an over-cap line is dropped and counted, never sent truncated.

Traits

RecordSink

One flushed batch’s delivery destination (DW-121). Deliberately small — one method, one batch, one answer — because the pipeline’s guarantees (ordering, bounded memory, drop-and-count) live in the FLUSHER, not in the sink: an implementation only decides where a batch goes and whether the receiver took it.

Functions

compile_stream_targets

Compile one generation’s stream state from the config block. A sink whose compilation fails is skipped with a LOUD error (fail closed — never delivered with placeholder bytes, never fatal to the generation); if that leaves no sinks, the stream is disabled and records drop at offer time with the counters as the witness.

refresh_stream_gauges

Refresh the stream’s observation gauges at scrape time (the event bus’s model: the offer path bumps plain atomics, and only the scrape couples them to the registry).

run_stream_flusher

The flusher loop (DW-121): drain the stream channel, assemble ordered batches, and deliver each batch — inline, in order, one retry cycle per batch — to the current generation’s sinks. Stops on shutdown after draining what is already queued into one final flush attempt (the gateway is not a durable queue; an offer undelivered at process death is lost with its counters as the witness).