obzenflow_runtime 0.2.4

// SPDX-License-Identifier: MIT OR Apache-2.0
// SPDX-FileCopyrightText: 2025-2026 ObzenFlow Contributors
// https://obzenflow.dev

//! Runtime-owned bootstrap settings shared across infra, DSL, adapters, and runtime.
//!
//! These values are resolved once by the hosting shell (typically `FlowApplication`)
//! and then read by lower layers during flow build and execution.

use std::path::PathBuf;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, OnceLock, RwLock};
use std::time::Duration;
use std::{fmt, mem};

use obzenflow_core::journal::archive::ReplayArchive;

#[cfg(any(test, feature = "test-support"))]
use tokio::sync::{Mutex as TokioMutex, MutexGuard as TokioMutexGuard};

thread_local! {
    static INSTALL_OWNER: u8 = const { 0 };
}

#[cfg(any(test, feature = "test-support"))]
fn bootstrap_test_mutex() -> &'static TokioMutex<()> {
    static LOCK: OnceLock<TokioMutex<()>> = OnceLock::new();
    LOCK.get_or_init(|| TokioMutex::new(()))
}

/// Controls whether a pipeline starts automatically after materialisation.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum StartupMode {
    Auto,
    Manual,
}

impl StartupMode {
    /// Returns `true` when the host must manually start the pipeline.
    pub fn is_manual(self) -> bool {
        matches!(self, Self::Manual)
    }
}

/// The operator's archive verb (FLOWIP-120n): bounded reconstruction that
/// drains, or catch-up that continues live.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ReplayVerb {
    /// `--replay-from`.
    Replay,
    /// `--resume-from`.
    Resume,
}

/// Replay settings needed to bootstrap a flow from an existing run archive.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ReplayBootstrap {
    /// Path to a run archive directory containing `run_manifest.json`.
    pub archive_path: PathBuf,
    /// When true, allow replay from an incomplete archive.
    pub allow_incomplete_archive: bool,
    /// When true, resume proceeds past non-idempotent or undeclared sinks
    /// instead of refusing (FLOWIP-120n F16).
    pub allow_duplicate_sink_delivery: bool,
    /// Which verb selected the archive.
    pub verb: ReplayVerb,
}

/// A resolved bootstrap snapshot shared across infra, DSL, adapters, and runtime.
///
/// These settings are resolved by the hosting shell (typically `FlowApplication`)
/// and installed for the lifetime of a single run via [`install_bootstrap_config`].
#[derive(Clone)]
pub struct BootstrapConfig {
    /// Graceful shutdown timeout for pipeline drain and stage teardown.
    pub shutdown_timeout: Duration,
    /// Host startup behaviour for pipeline execution.
    pub startup_mode: StartupMode,
    /// Optional replay bootstrap configuration.
    pub replay: Option<ReplayBootstrap>,
    /// The opened replay/resume input archive, when `replay` is set (FLOWIP-120u).
    ///
    /// The host opens it at config resolution (the concrete opener lives in
    /// infra, below the DSL in the dependency graph) and the build consumes it.
    pub replay_archive: Option<Arc<dyn ReplayArchive>>,
}

impl Default for BootstrapConfig {
    fn default() -> Self {
        Self {
            shutdown_timeout: Duration::from_secs(30),
            startup_mode: StartupMode::Auto,
            replay: None,
            replay_archive: None,
        }
    }
}

// Manual impls: the archive handle is a live resource, excluded from equality
// and rendered by presence in Debug.
impl PartialEq for BootstrapConfig {
    fn eq(&self, other: &Self) -> bool {
        self.shutdown_timeout == other.shutdown_timeout
            && self.startup_mode == other.startup_mode
            && self.replay == other.replay
    }
}

impl Eq for BootstrapConfig {}

impl fmt::Debug for BootstrapConfig {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("BootstrapConfig")
            .field("shutdown_timeout", &self.shutdown_timeout)
            .field("startup_mode", &self.startup_mode)
            .field("replay", &self.replay)
            .field(
                "replay_archive",
                &self.replay_archive.as_ref().map(|_| "<present>"),
            )
            .finish()
    }
}

fn storage() -> &'static RwLock<BootstrapConfig> {
    static BOOTSTRAP: OnceLock<RwLock<BootstrapConfig>> = OnceLock::new();
    BOOTSTRAP.get_or_init(|| RwLock::new(BootstrapConfig::default()))
}

/// Read the currently-installed bootstrap snapshot.
pub fn bootstrap_config() -> BootstrapConfig {
    storage()
        .read()
        .expect("bootstrap config lock poisoned")
        .clone()
}

/// Overwrite the process bootstrap snapshot.
///
/// Prefer [`install_bootstrap_config`] for host runs so the previous value is restored
/// on drop. This setter is primarily intended for benchmarks and tests.
pub fn set_bootstrap_config(config: BootstrapConfig) {
    *storage().write().expect("bootstrap config lock poisoned") = config;
}

/// Install a bootstrap snapshot for the lifetime of the returned guard.
///
/// This keeps bootstrap settings scoped to a single host run instead of
/// permanently caching the first configuration seen in the process.
///
/// Bootstrap settings are process-global, so only one active install is supported
/// at a time. Overlapping installs will wait until the prior guard is dropped.
pub fn install_bootstrap_config(config: BootstrapConfig) -> BootstrapConfigGuard {
    let active_install = ActiveInstallLease::acquire();

    let previous = {
        let mut bootstrap = storage().write().expect("bootstrap config lock poisoned");
        mem::replace(&mut *bootstrap, config)
    };

    BootstrapConfigGuard {
        previous: Some(previous),
        active_install,
    }
}

/// Attempt to install a bootstrap snapshot without waiting.
///
/// When another install is already active, this returns `Err(config)` and leaves
/// the currently-installed bootstrap unchanged.
pub fn try_install_bootstrap_config(
    config: BootstrapConfig,
) -> Result<BootstrapConfigGuard, BootstrapConfig> {
    let active_install = match ActiveInstallLease::try_acquire() {
        Some(lease) => lease,
        None => return Err(config),
    };

    let previous = {
        let mut bootstrap = storage().write().expect("bootstrap config lock poisoned");
        mem::replace(&mut *bootstrap, config)
    };

    Ok(BootstrapConfigGuard {
        previous: Some(previous),
        active_install,
    })
}

/// Restores the prior bootstrap snapshot on drop.
pub struct BootstrapConfigGuard {
    previous: Option<BootstrapConfig>,
    active_install: ActiveInstallLease,
}

impl Drop for BootstrapConfigGuard {
    fn drop(&mut self) {
        if let Some(previous) = self.previous.take() {
            *storage().write().expect("bootstrap config lock poisoned") = previous;
        }
        let _ = &self.active_install;
    }
}

/// Graceful shutdown timeout for pipeline drain and teardown.
pub fn shutdown_timeout() -> Duration {
    bootstrap_config().shutdown_timeout
}

/// Returns `true` when the pipeline should not auto-run on materialisation.
pub fn startup_mode_manual() -> bool {
    bootstrap_config().startup_mode.is_manual()
}

/// Optional replay bootstrap parameters.
pub fn replay_bootstrap() -> Option<ReplayBootstrap> {
    bootstrap_config().replay
}

/// The opened replay/resume input archive, when the host installed one (FLOWIP-120u).
pub fn replay_archive() -> Option<Arc<dyn ReplayArchive>> {
    bootstrap_config().replay_archive
}

fn active_install_owner() -> &'static AtomicUsize {
    static ACTIVE_INSTALL: OnceLock<AtomicUsize> = OnceLock::new();
    ACTIVE_INSTALL.get_or_init(|| AtomicUsize::new(0))
}

fn current_owner_id() -> usize {
    INSTALL_OWNER.with(|marker| marker as *const u8 as usize)
}

struct ActiveInstallLease;

impl ActiveInstallLease {
    fn try_acquire() -> Option<Self> {
        let owner = active_install_owner();
        let me = current_owner_id();

        // Catch the most likely deadlock case: re-entrant install on the same thread.
        // This cannot be made perfectly robust without passing an explicit run context.
        if owner.load(Ordering::Acquire) == me {
            panic!("install_bootstrap_config() does not support nested installs");
        }

        owner
            .compare_exchange(0, me, Ordering::AcqRel, Ordering::Acquire)
            .ok()
            .map(|_| Self)
    }

    fn acquire() -> Self {
        let owner = active_install_owner();
        let me = current_owner_id();

        // Catch the most likely deadlock case: re-entrant install on the same thread.
        // This cannot be made perfectly robust without passing an explicit run context.
        if owner.load(Ordering::Acquire) == me {
            panic!("install_bootstrap_config() does not support nested installs");
        }

        let mut attempts: u32 = 0;
        loop {
            if let Some(lease) = Self::try_acquire() {
                return lease;
            }

            attempts = attempts.saturating_add(1);
            if attempts <= 10 {
                std::hint::spin_loop();
            } else if attempts <= 100 {
                std::thread::yield_now();
            } else {
                std::thread::sleep(Duration::from_millis(1));
            }
        }
    }
}

impl Drop for ActiveInstallLease {
    fn drop(&mut self) {
        active_install_owner().store(0, Ordering::Release);
    }
}

#[cfg(test)]
pub(crate) fn bootstrap_test_lock() -> TokioMutexGuard<'static, ()> {
    bootstrap_test_mutex().blocking_lock()
}

#[cfg(any(test, feature = "test-support"))]
pub(crate) async fn bootstrap_test_lock_async() -> TokioMutexGuard<'static, ()> {
    bootstrap_test_mutex().lock().await
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn bootstrap_defaults_are_sensible() {
        let _lock = bootstrap_test_lock();
        let _guard = install_bootstrap_config(BootstrapConfig::default());

        let bootstrap = bootstrap_config();
        assert_eq!(bootstrap.shutdown_timeout, Duration::from_secs(30));
        assert_eq!(bootstrap.startup_mode, StartupMode::Auto);
        assert_eq!(bootstrap.replay, None);
    }

    #[test]
    fn bootstrap_guard_restores_previous_config() {
        let _lock = bootstrap_test_lock();
        let baseline = BootstrapConfig {
            shutdown_timeout: Duration::from_secs(45),
            startup_mode: StartupMode::Auto,
            replay: None,
            replay_archive: None,
        };
        set_bootstrap_config(baseline.clone());

        {
            let _guard = install_bootstrap_config(BootstrapConfig {
                shutdown_timeout: Duration::from_secs(5),
                startup_mode: StartupMode::Manual,
                replay: Some(ReplayBootstrap {
                    archive_path: PathBuf::from("/tmp/archive"),
                    allow_incomplete_archive: true,
                    allow_duplicate_sink_delivery: false,
                    verb: ReplayVerb::Replay,
                }),
                replay_archive: None,
            });

            assert_eq!(shutdown_timeout(), Duration::from_secs(5));
            assert!(startup_mode_manual());
        }

        assert_eq!(bootstrap_config(), baseline);
    }

    #[test]
    fn nested_install_panics() {
        let _lock = bootstrap_test_lock();
        let _guard = install_bootstrap_config(BootstrapConfig::default());

        let result = std::panic::catch_unwind(|| {
            let _nested = install_bootstrap_config(BootstrapConfig::default());
        });

        assert!(result.is_err());
    }

    #[test]
    fn overlapping_installs_are_serialised() {
        use std::sync::mpsc;

        let _lock = bootstrap_test_lock();
        let guard = install_bootstrap_config(BootstrapConfig::default());

        let (started_tx, started_rx) = mpsc::channel::<()>();
        let (acquired_tx, acquired_rx) = mpsc::channel::<()>();

        let handle = std::thread::spawn(move || {
            started_tx.send(()).unwrap();
            let _guard = install_bootstrap_config(BootstrapConfig::default());
            acquired_tx.send(()).unwrap();
        });

        started_rx.recv_timeout(Duration::from_secs(1)).unwrap();
        std::thread::sleep(Duration::from_millis(25));
        assert!(acquired_rx.try_recv().is_err());

        drop(guard);
        acquired_rx.recv_timeout(Duration::from_secs(5)).unwrap();
        handle.join().unwrap();
    }

    #[test]
    fn try_install_fails_fast_when_an_install_is_active() {
        use std::sync::mpsc;

        let _lock = bootstrap_test_lock();
        let guard = install_bootstrap_config(BootstrapConfig::default());

        let (tx, rx) = mpsc::channel::<bool>();
        let handle = std::thread::spawn(move || {
            let result = try_install_bootstrap_config(BootstrapConfig::default());
            tx.send(result.is_err()).unwrap();
        });

        assert!(rx.recv_timeout(Duration::from_secs(1)).unwrap());
        drop(guard);
        handle.join().unwrap();
    }
}