spate-core 0.1.0

//! Pipeline runtime: pinned driver threads, the source controller, and
//! process assembly.
//!
//! Thread anatomy (see `docs/DESIGN.md` § Process anatomy):
//!
//! ```text
//! main (run())          controller (std thread)      driver 0..N (std threads)
//!   metrics/admin/io  ──  owns Source + Checkpointer ── own lanes + chain
//!   joins everything      poll_events / commit tick     poll → push_batch → route
//!                         pause/resume application      backpressure ticks
//! ```
//!
//! Communication: the controller sends [`ThreadControl`] messages to
//! drivers (lane assignment, drain barriers); drivers send [`DriverEvent`]
//! requests back (pause/resume — only the controller touches the
//! [`Source`](crate::source::Source) — and fatal reports). All channels are
//! unbounded crossbeam channels: control traffic is rare and must never
//! block a poll loop.
//!
//! Shutdown (also the full-revocation path, per DESIGN.md § Shutdown):
//! SIGTERM → controller stops event polling and sends `Shutdown` to every
//! driver → each driver flushes its chain, drops its lanes, and arrives at
//! the barrier → main joins driver threads (dropping chains closes the
//! shard queues) → the sink drains under the remaining deadline → the
//! controller runs a final drain + commit + `flush_commits` → the process
//! reports an [`ExitReport`]. A sink that cannot flush by the deadline is
//! abandoned loudly; unacknowledged offsets are never committed, so the
//! data replays after restart (at-least-once).

mod builder;
mod controller;
mod driver;
mod runtime;

pub use builder::{BuildError, ChainCtx, Pipeline, PipelineError, SinkOptions};
pub use runtime::{PipelineRuntime, RuntimeOptions, ShutdownHandle, StartError, metrics_settings};

use crate::error::FatalError;
use crate::record::PartitionId;
use crate::sink::ShardQueues;
use crate::source::{DrainBarrier, LaneId};
use std::time::Instant;

/// Control messages the controller sends to a driver thread.
pub(crate) enum ThreadControl<L> {
    /// Take ownership of a newly assigned lane.
    AddLane(L),
    /// Stop and drop the listed lanes (revocation): flush the chain, then
    /// arrive at `barrier` once per stopped lane before `deadline`.
    StopLanes {
        lanes: Vec<LaneId>,
        barrier: DrainBarrier,
        deadline: Instant,
    },
    /// Flush the chain's partial terminal state now, without waiting out
    /// the idle-flush lull
    /// ([`SourceEvent::CommitReady`](crate::source::SourceEvent::CommitReady)).
    ///
    /// A lane that reached genuine end-of-input has no more data coming, so
    /// its tail would otherwise sit in the chain until `idle_flush` elapses
    /// — unacknowledged, and therefore blocking the completion of whatever
    /// unit of work it belongs to. Best-effort and unsynchronised: the
    /// controller does not wait, and a blocked chain simply retries on the
    /// ordinary lull check.
    FlushNow,
    /// Drop the listed lanes without flushing or synchronising
    /// ([`SourceEvent::LanesRetired`](crate::source::SourceEvent::LanesRetired)):
    /// their input is fully delivered, acknowledged *and* committed, so
    /// nothing of theirs can sit unflushed in the chain. Pure bookkeeping —
    /// no barrier, no deadline, no `flush_until`. Forcing a flush here would
    /// fragment sink batches and park the thread once per completed unit of
    /// work; sources with anything in flight must use
    /// [`ThreadControl::StopLanes`] instead.
    DropLanes { lanes: Vec<LaneId> },
    /// Stop everything (shutdown): flush the chain, drop all lanes, arrive
    /// once at `barrier`, and exit the thread.
    Shutdown {
        barrier: DrainBarrier,
        deadline: Instant,
    },
}

/// Requests and reports a driver thread sends the controller.
#[derive(Debug)]
pub(crate) enum DriverEvent {
    /// Backpressure tripped: pause these lanes at the source.
    PauseLanes { lanes: Vec<LaneId> },
    /// Backpressure cleared: resume these lanes.
    ResumeLanes { lanes: Vec<LaneId> },
    /// The chain failed or panicked; the pipeline must stop.
    Fatal { thread: usize, error: FatalError },
}

/// What the sink reported when draining at shutdown — the sink layer's
/// [`DrainReport`](crate::sink::DrainReport), re-exported so assemblies
/// hand `SinkPool::drain`'s result straight through.
pub use crate::sink::DrainReport;

/// The sink half the runtime drives: the shared shard-queue handle plus a
/// drain hook invoked once at shutdown with the remaining drain budget.
///
/// Built by the sink layer (`SinkPool`) or by tests; the runtime is
/// deliberately ignorant of worker internals.
pub struct SinkRuntime {
    /// Sending side of the per-shard chunk queues, one entry per installed
    /// sink (the runtime only uses capacity introspection for the
    /// backpressure resume gate; the chain's terminal stage holds clones).
    /// A single-sink pipeline has one entry.
    pub queues: Vec<ShardQueues>,
    /// Drain the sinks: flush what's pending within the budget, fail the
    /// acknowledgements of anything abandoned, and report. For multi-sink
    /// pipelines this is the composed hook that drains every sink.
    pub drain: SinkDrainFn,
    /// Optional connectivity probe (e.g. `SinkPool::probe_all`). The
    /// runtime probes at startup and then periodically, driving the
    /// sinks-connected half of `/readyz`. Without a probe the flag is set
    /// unconditionally.
    pub probe: Option<SinkProbeFn>,
}

impl std::fmt::Debug for SinkRuntime {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SinkRuntime")
            .field("queues", &self.queues)
            .finish_non_exhaustive()
    }
}

/// Boxed sink drain hook — defined next to the sink layer that produces
/// it, re-exported here where the runtime consumes it.
pub use crate::sink::{SinkDrainFn, SinkProbeFn};

/// Terminal state of a pipeline run.
#[derive(Clone, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub enum ExitState {
    /// Drained and committed cleanly: a requested shutdown (SIGTERM or
    /// programmatic), or a bounded source reporting
    /// [`SourceEvent::Drained`](crate::source::SourceEvent::Drained) after
    /// exhausting its input. For the drained case this additionally
    /// guarantees every batch was acknowledged and the final watermark
    /// commit persisted (anything less exits `Failed`).
    Completed,
    /// A fatal error stopped the pipeline; the process should exit
    /// non-zero.
    Failed(FatalErrorReport),
}

/// Owned copy of the fatal error carried in the exit report.
#[derive(Clone, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub struct FatalErrorReport {
    /// Component that failed.
    pub component: String,
    /// Human-readable cause.
    pub reason: String,
}

impl std::fmt::Display for FatalErrorReport {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "pipeline failed in {}: {}", self.component, self.reason)
    }
}

impl std::error::Error for FatalErrorReport {}

/// Outcome of [`PipelineRuntime::run`].
#[derive(Debug)]
#[non_exhaustive]
pub struct ExitReport {
    /// How the run ended.
    pub state: ExitState,
    /// The sink's drain report (absent when the sink drain hook could not
    /// run, e.g. the I/O runtime was already gone).
    pub sink_drain: Option<DrainReport>,
    /// The last committed watermark per partition, as reported by the
    /// final commit.
    pub final_watermarks: Vec<(PartitionId, i64)>,
}

impl ExitReport {
    /// Log the outcome — state, sink drain, final watermarks — at the level
    /// matching it (`info` for a clean exit, `error` for a failure).
    pub fn log(&self) {
        match &self.state {
            ExitState::Completed => tracing::info!(
                state = ?self.state,
                drain = ?self.sink_drain,
                watermarks = ?self.final_watermarks,
                "pipeline finished"
            ),
            ExitState::Failed(failure) => tracing::error!(
                component = %failure.component,
                reason = %failure.reason,
                drain = ?self.sink_drain,
                watermarks = ?self.final_watermarks,
                "pipeline failed"
            ),
        }
    }

    /// The process exit code this outcome maps to: `0` for a clean exit,
    /// `1` for a failure.
    #[must_use]
    pub fn exit_code(&self) -> i32 {
        match self.state {
            ExitState::Completed => 0,
            ExitState::Failed(_) => 1,
        }
    }

    /// The report as a `Result`, so a `main` can `?` a failed run: a clean
    /// exit passes the report through, a failure returns the
    /// [`FatalErrorReport`] as the error.
    pub fn ok(self) -> Result<ExitReport, FatalErrorReport> {
        match &self.state {
            ExitState::Completed => Ok(self),
            ExitState::Failed(failure) => Err(failure.clone()),
        }
    }
}

#[cfg(all(test, not(loom)))]
pub(crate) mod fakes;
#[cfg(all(test, not(loom)))]
mod tests;