Crate fv_streams_engine 

The STREAM worker — runs a kind=stream build CONTINUOUSLY on the dataflow runtime (pipeline, dataflow): a topology’s stages become tasks on threads connected by bounded in-memory edges — source tasks (a connector’s: a Kafka topic’s partitions, files, a table, a generator) produce Arrow batches, stateless steps run on the batch through fv-value-datafusion, the stateful operators (windows, sessions, ranks, joins) run as tasks sharded per origin or over vnode ranges behind an in-memory shuffle, and sink tasks (a connector’s) write the output under deterministic keys. Connectors only at the edges — the engine carries no broker, store or catalog code.

Correctness: epochs from the runtime — barriers ride every edge behind the data, a stateful task snapshots at each, the coordinator writes the epoch (the objects, then a manifest with every source’s positions) to the epoch store, and only then do offsets and transactional sinks commit; a restart restores from the newest manifest. Time rides in band as watermarks. Never SUCCEEDED: it heartbeats while RUNNING and unwinds to STOPPED when the control plane requests it (the heartbeat echo carries the stop signal — one round-trip for liveness + control). Config errors fail the build loudly up front; per-row isolation drops only poison DATA.

Modules

compute

The worker’s compute runtime: a wasm / container pipeline step selects a Kinetics transform by ref; fv_plan::KineticsRunner types the rows, runs it, and returns rows. This module only decides which transforms and backends the worker has: the roots from FV_TRANSFORMS_DIR, and both shipped backends.

dataflow

THE DATAFLOW RUNTIME (Phase 2): a pipeline as tasks on OS threads connected by bounded in-memory edges that carry Arrow batches and the two control signals — watermarks and checkpoint barriers — in band (design/2026-09-15-phase2-dataflow-scoping.md).

decode

COLUMNAR DECODE at the source edge: Kafka messages → one Arrow RecordBatch per partition per poll, under the source’s schema and bad-data policy.

steps

INLINE STEPS ON THE BATCH: select / rename / drop / filter / applyExpression applied to an Arrow RecordBatch, with the value dialect compiled through fv-value-datafusion into DataFusion physical expressions (vectorised over the columns; a sub-expression the translator cannot reproduce exactly runs as a per-row UDF inside the same expression, so semantics never change — only speed).

Structs

Binding

A dataset resolved to what reads it and what writes it. An input dataset’s source opens the source tasks; an output dataset’s sink builds the sink tasks; the other half is simply unused. Not Eq: the factories ride Arcs.

SinkCtx

What a sink is built with: its task id and its index sink within the output (the files it names); the emitted counter; whether output is exactly-once (visibility gated on the epoch commit); the checkpoint epoch a restored run continues from (None: a fresh run starts the output over); the barrier Acker; and the run’s identity (a transactional id is pipeline + the task). Built by the engine only (#[non_exhaustive]: new fields arrive as setters).

SourceCtx

What a source task is built with: its index task of tasks; the positions the checkpoint recorded for it under the source’s name (start: partition → next position, None when the run is fresh or the checkpoint has none for this task — restoring tells the two apart); the event-time column it stamps watermarks from; the run’s identity for a source that names itself to a broker (a consumer group is pipeline + stage); and the counter of rows it drops as bad data. Built by the engine only (#[non_exhaustive]: new fields arrive as new setters, never as a break).

StageDef

One stage of a stream topology: transform steps (the engine’s vocabulary — inline / compute / windowedAggregate / sessionAggregate / streamJoin, parsed and validated by the engine, not the host) reading inputs datasets and producing the output dataset.

Topology

A stream topology: its stages, in order.

Enums

BuildSignal

What one heartbeat tells the engine to do next.

Traits

ControlPlane

The host contract. Implementations must be cheap to clone behind an Arc and safe to call from the engine’s single control task (the task threads never touch it).

InlineSource

A source provided from outside the engine — a connector’s (a Kafka topic, files, a database table, the Nexmark generator) or a host’s own. The runtime opens tasks of them; task t owns its share of the input (a topic’s partitions p % tasks == t, a generator’s events, a listing’s files), so any task count gives the same stream. A source still stamps watermarks, takes barriers and records its position like any other: the checkpoint protocol does not know the difference.

OutputSink

A sink provided from outside the engine — a connector’s (a Kafka topic, files, Iceberg, Flight, the blackhole) or a host’s own (a fan-out, a test collector) — the symmetric counterpart of InlineSource. The runtime builds one per sink task, at start, on the worker that owns the task (a durable sink’s directory must be opened by its one owner); a builder that fails fails the start before any task runs, naming the sink.

Functions

run_stream

Claim-side entry: validate + run one stream build to its terminal state against ANY ControlPlane host. Any init error is reported as FAILED; a stop request lands as STOPPED. Never returns Err — it OWNS its record.