fv_streams_engine/
lib.rs

1//! The STREAM worker — runs a `kind=stream` build CONTINUOUSLY on the dataflow runtime
2//! (`pipeline`, `dataflow`): a topology's stages become tasks on threads connected by bounded
3//! in-memory edges — source tasks (a connector's: a Kafka topic's partitions, files, a table, a
4//! generator) produce Arrow batches, stateless steps run on the batch through
5//! `fv-value-datafusion`, the stateful operators (windows, sessions, ranks, joins) run as tasks
6//! sharded per origin or over vnode ranges behind an in-memory shuffle, and sink tasks (a
7//! connector's) write the output under deterministic keys. Connectors only at the edges — the
8//! engine carries no broker, store or catalog code.
9//!
10//! Correctness: epochs from the runtime — barriers ride every edge behind the data, a stateful
11//! task snapshots at each, the coordinator writes the epoch (the objects, then a manifest with
12//! every source's positions) to the epoch store, and only then do offsets and transactional sinks
13//! commit; a restart restores from the newest manifest. Time rides in band as watermarks. Never
14//! SUCCEEDED: it heartbeats while RUNNING and unwinds to STOPPED when the control plane requests
15//! it (the heartbeat echo carries the stop signal — one round-trip for liveness + control). Config
16//! errors fail the build loudly up front; per-row isolation drops only poison DATA.
17
18/// Every Rust code block in the README is compiled and run as a doctest.
19#[cfg(doctest)]
20#[doc = include_str!("../README.md")]
21pub struct ReadmeDoctests;
22
23use std::sync::atomic::{AtomicU64, Ordering};
24use std::sync::Arc;
25use std::time::{Duration, Instant};
26
27use serde_json::{json, Value as J};
28
29use fv_streams_types::settings::env;
30
31mod batches;
32pub mod compute;
33pub use fv_streams_runtime::dataflow;
34use fv_streams_runtime::{cluster, placement};
35pub use fv_streams_types::decode;
36mod keys;
37use fv_streams_state as epochs;
38mod ops;
39mod pipeline;
40mod rows;
41mod spec;
42pub mod steps;
43use batches::*;
44pub use fv_streams_types::{
45    Binding, BuildSignal, ControlPlane, InlineSource, OutputSink, SinkCtx, SourceCtx, StageDef, Topology,
46};
47use keys::*;
48use spec::*;
49
50/// Processing-time clock (wall-clock epoch ms) for bounded idleness. Injected into the operators so
51/// they stay pure and testable — event time can't measure the *absence* of events (see
52/// [`fv_streams_ops::Watermark`]).
53fn now_ms() -> i64 {
54    chrono::Utc::now().timestamp_millis()
55}
56
57/// The process's CPU so far, folded by thread family: the engine's dataflow tasks (`fv-task`),
58/// librdkafka's per-broker threads (`rdk:…`), the generator, the async runtime, the rest. A live
59/// core-ms ledger — cheap (one walk of `/proc/self/task` per heartbeat), empty off Linux. Surfaced
60/// on the build record's `metrics.cpuMs` so any host (the sandbox observe surface, a console) can
61/// show where the cores went while the stream runs, not only at the end.
62pub(crate) fn cpu_families() -> std::collections::BTreeMap<String, u64> {
63    let mut families: std::collections::BTreeMap<String, u64> = std::collections::BTreeMap::new();
64    for (name, ms) in &dataflow::process_cpu_by_thread() {
65        // fold librdkafka's per-broker threads and the engine's per-task threads into their families.
66        let family = if name.starts_with("rdk:") {
67            name.trim_end_matches(|c: char| c.is_ascii_digit()).to_string()
68        } else if name.starts_with("fv-task-") {
69            "fv-task".to_string()
70        } else {
71            name.clone()
72        };
73        *families.entry(family).or_default() += ms;
74    }
75    families
76}
77
78/// `STREAM_TRACE_CPU=1`: print where the process's cores have gone so far, by thread family, on
79/// every heartbeat — the engine's tasks, librdkafka's threads, the generator, the rest.
80fn trace_cpu(build_id: &str) {
81    if env("STREAM_TRACE_CPU", "0") != "1" {
82        return;
83    }
84    let families = cpu_families();
85    let total: u64 = families.values().sum();
86    let line: Vec<String> = families.iter().map(|(k, v)| format!("{k}={v}")).collect();
87    eprintln!("stream {build_id}: cpu ms total={total} {}", line.join(" "));
88}
89
90/// Claim-side entry: validate + run one stream build to its terminal state against ANY
91/// [`ControlPlane`] host. Any init error is reported as FAILED; a stop request lands as STOPPED.
92/// Never returns Err — it OWNS its record.
93pub async fn run_stream(
94    cp: std::sync::Arc<dyn ControlPlane>,
95    build_id: String,
96    pipeline_name: String,
97    mut record: J,
98) -> Result<(), String> {
99    match run_stream_inner(cp.as_ref(), &build_id, &pipeline_name, &mut record).await {
100        Ok(()) => Ok(()),
101        Err(e) => {
102            eprintln!("stream {build_id}: FAILED — {e}");
103            record["status"] = json!("FAILED");
104            record["error"] = json!(e);
105            record["finishedAt"] = json!(chrono::Utc::now().to_rfc3339());
106            cp.put_record(&build_id, &record).await;
107            Err(e)
108        }
109    }
110}
111
112async fn run_stream_inner(
113    cp: &dyn ControlPlane,
114    build_id: &str,
115    pipeline_name: &str,
116    record: &mut J,
117) -> Result<(), String> {
118    // ── Resolve the pipeline into a stage CHAIN (multi-stage topologies). ──
119    let topo = cp.topology(pipeline_name).await?;
120    if topo.stages.is_empty() {
121        return Err("pipeline has no transforms".into());
122    }
123    let plans = pipeline::plan(&topo)?;
124    pipeline::run_pipeline(cp, build_id, pipeline_name, record, &topo, plans).await
125}
fv_streams_engine/lib.rs

fv_streams_engine/
lib.rs
