fv-streams-engine

The FusionVault Streams engine: it runs one stream pipeline continuously over Kafka. N independent consumer threads in one group each own the partitions Kafka assigns them and run a self-contained consume → transform → emit → commit loop, so throughput scales with partitions and cores and there is no shared poll thread to serialise on. Stateful steps use fv-streams-ops operators, one instance per assigned partition, checkpointed to a compacted state topic; with exactly-once on, emissions and checkpoints ride one Kafka transaction.

Host it

The engine asks its host four things through one trait; everything else is the engine's.

use std::sync::Arc;
use async_trait::async_trait;
use fv_streams_engine::dataflow::{Out, Poll, Sink, Source};
use fv_streams_engine::{run_stream, Binding, BuildSignal, ControlPlane, InlineSource, OutputSink, SinkCtx, SourceCtx, StageDef, Topology};
use serde_json::{json, Value};

/// A source that yields nothing and ends. A real host hands the engine a connector's source —
/// `fv-streams-connectors` resolves `connector = "…"` to one — or its own.
struct Nothing;
impl Source for Nothing {
    fn poll(&mut self, _out: &mut Out) -> Poll { Poll::Done }
    fn on_barrier(&mut self, _epoch: u64) -> Vec<u8> { Vec::new() }
}
impl InlineSource for Nothing {
    fn name(&self) -> String { "nothing".into() }
    fn tasks(&self) -> usize { 1 }
    fn open(&self, _cx: SourceCtx) -> Result<Box<dyn Source + Send>, String> { Ok(Box::new(Nothing)) }
}

/// A sink that discards what it gets (a connector's sink writes it somewhere).
struct Discard;
impl Sink for Discard {
    fn on_data(&mut self, _batch: arrow::array::RecordBatch) {}
    fn on_barrier(&mut self, _epoch: u64) -> bool { true }
    fn on_eos(&mut self) {}
}
impl OutputSink for Discard {
    fn open(&self, _cx: SinkCtx) -> Result<Box<dyn Sink>, String> { Ok(Box::new(Discard)) }
}

struct MyHost;

#[async_trait]
impl ControlPlane for MyHost {
    async fn topology(&self, _pipeline: &str) -> Result<Topology, String> {
        Ok(Topology { stages: vec![StageDef {
            inputs: vec!["events".into()],
            output: "out".into(),
            steps: vec![json!({ "op": "filter", "expression": "price > 100" })],
        }]})
    }
    async fn resolve_dataset(&self, dataset: &str) -> Result<Binding, String> {
        Ok(Binding { api_name: dataset.into(), source: Arc::new(Nothing), sink: Arc::new(Discard) })
    }
    async fn heartbeat(&self, _build_id: &str) -> BuildSignal { BuildSignal::Continue }
    async fn put_record(&self, _build_id: &str, record: &Value) { println!("{record}"); }
}

# tokio::runtime::Runtime::new().unwrap().block_on(async {
let record = json!({ "id": "demo", "pipeline": "demo", "kind": "stream", "status": "RUNNING" });
run_stream(Arc::new(MyHost), "demo".into(), "demo".into(), record).await;
# });

• topology — the pipeline as stages of steps in the engine's vocabulary (the engine validates them and fails loudly before consuming anything).
• resolve_dataset — a dataset name to its source and sink factories: a connector's (fv-streams-connectors resolves connector = "…" / sink = "…" through one registry) or the host's own.
• heartbeat — called on a cadence; returning Stop triggers the ordered shutdown (final checkpoint, flush, commit). ContinueUnreachable keeps the stream alive when the host is down.
• put_record — the live build record (status, consumed / emitted / dropped counters).

fv-streams is this trait answered from a TOML file; a platform answers it over HTTP. Same engine, same semantics.

Steps

A stage is inline steps around at most one operator step:

• inline: select, rename, drop, filter, applyExpression (the value dialect, per row, poison rows isolated and counted, never a stalled stream);
• compute: wasm / container — a Kinetics transform by ref, resolved from FV_TRANSFORMS_DIR;
• stateful: windowedAggregate, sessionAggregate, streamJoin, topN, lastN.

Settings are STREAM_* environment variables (consumers, batch size, checkpoint interval, exactly-once, offset reset, idle timeouts); the repository README has the table.

Apache-2.0.