fv-streams-engine 0.6.0

The FusionVault Streams engine: runs a stream pipeline continuously over Kafka with N independent consumer threads, stateful operators from fv-streams-ops, checkpointed state, exactly-once output, and Kinetics compute steps. Hosted through one small ControlPlane trait.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95

# fv-streams-engine

The [FusionVault Streams](https://github.com/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`](https://crates.io/crates/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.

```rust,no_run
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`](https://crates.io/crates/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](https://github.com/FusionVault/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.