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::{run_stream, BuildSignal, ControlPlane, SinkKind, SourceKind, StageDef, TopicBinding, Topology};
use serde_json::{json, Value};

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_topic(&self, dataset: &str) -> Result<TopicBinding, String> {
        Ok(TopicBinding { api_name: dataset.into(), topic: dataset.into(), brokers: "localhost:9092".into(), policy: Default::default(), source: SourceKind::Kafka, sink: SinkKind::Kafka })
    }
    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_topic — a dataset name to its topic and brokers.
• 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.