fv-streams-engine 0.6.0

//! 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.

/// Every Rust code block in the README is compiled and run as a doctest.
#[cfg(doctest)]
#[doc = include_str!("../README.md")]
pub struct ReadmeDoctests;

use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};

use serde_json::{json, Value as J};

use fv_streams_types::settings::env;

mod batches;
pub mod compute;
pub use fv_streams_runtime::dataflow;
use fv_streams_runtime::{cluster, placement};
pub use fv_streams_types::decode;
mod keys;
use fv_streams_state as epochs;
mod ops;
mod pipeline;
mod rows;
mod spec;
pub mod steps;
use batches::*;
pub use fv_streams_types::{
    Binding, BuildSignal, ControlPlane, InlineSource, OutputSink, SinkCtx, SourceCtx, StageDef, Topology,
};
use keys::*;
use spec::*;

/// Processing-time clock (wall-clock epoch ms) for bounded idleness. Injected into the operators so
/// they stay pure and testable — event time can't measure the *absence* of events (see
/// [`fv_streams_ops::Watermark`]).
fn now_ms() -> i64 {
    chrono::Utc::now().timestamp_millis()
}

/// The process's CPU so far, folded by thread family: the engine's dataflow tasks (`fv-task`),
/// librdkafka's per-broker threads (`rdk:…`), the generator, the async runtime, the rest. A live
/// core-ms ledger — cheap (one walk of `/proc/self/task` per heartbeat), empty off Linux. Surfaced
/// on the build record's `metrics.cpuMs` so any host (the sandbox observe surface, a console) can
/// show where the cores went while the stream runs, not only at the end.
pub(crate) fn cpu_families() -> std::collections::BTreeMap<String, u64> {
    let mut families: std::collections::BTreeMap<String, u64> = std::collections::BTreeMap::new();
    for (name, ms) in &dataflow::process_cpu_by_thread() {
        // fold librdkafka's per-broker threads and the engine's per-task threads into their families.
        let family = if name.starts_with("rdk:") {
            name.trim_end_matches(|c: char| c.is_ascii_digit()).to_string()
        } else if name.starts_with("fv-task-") {
            "fv-task".to_string()
        } else {
            name.clone()
        };
        *families.entry(family).or_default() += ms;
    }
    families
}

/// `STREAM_TRACE_CPU=1`: print where the process's cores have gone so far, by thread family, on
/// every heartbeat — the engine's tasks, librdkafka's threads, the generator, the rest.
fn trace_cpu(build_id: &str) {
    if env("STREAM_TRACE_CPU", "0") != "1" {
        return;
    }
    let families = cpu_families();
    let total: u64 = families.values().sum();
    let line: Vec<String> = families.iter().map(|(k, v)| format!("{k}={v}")).collect();
    eprintln!("stream {build_id}: cpu ms total={total} {}", line.join(" "));
}

/// 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.
pub async fn run_stream(
    cp: std::sync::Arc<dyn ControlPlane>,
    build_id: String,
    pipeline_name: String,
    mut record: J,
) -> Result<(), String> {
    match run_stream_inner(cp.as_ref(), &build_id, &pipeline_name, &mut record).await {
        Ok(()) => Ok(()),
        Err(e) => {
            eprintln!("stream {build_id}: FAILED — {e}");
            record["status"] = json!("FAILED");
            record["error"] = json!(e);
            record["finishedAt"] = json!(chrono::Utc::now().to_rfc3339());
            cp.put_record(&build_id, &record).await;
            Err(e)
        }
    }
}

async fn run_stream_inner(
    cp: &dyn ControlPlane,
    build_id: &str,
    pipeline_name: &str,
    record: &mut J,
) -> Result<(), String> {
    // ── Resolve the pipeline into a stage CHAIN (multi-stage topologies). ──
    let topo = cp.topology(pipeline_name).await?;
    if topo.stages.is_empty() {
        return Err("pipeline has no transforms".into());
    }
    let plans = pipeline::plan(&topo)?;
    pipeline::run_pipeline(cp, build_id, pipeline_name, record, &topo, plans).await
}