saddle_runtime/

application.rs

use std::{
    future::Future,
    sync::Arc,
    sync::atomic::{AtomicBool, Ordering},
};

use saddle_core::{ComponentLifecycle, ErrorKind, Result, SaddleError};

use crate::RequestLifecycle;

static RUNTIME_STARTED: AtomicBool = AtomicBool::new(false);
const WORKER_THREADS: usize = 2;
const MAX_IO_EVENTS_PER_TICK: usize = 5;

/// A complete Saddle application hosted by the process-wide async runtime.
///
/// This is an assembly API, not a general-purpose async executor: it exposes no
/// Tokio handle, task spawning, runtime configuration, or arbitrary `block_on`.
pub struct Application {
    components: Vec<Arc<dyn ComponentLifecycle>>,
    requests: RequestLifecycle,
    ingress_bridge_issued: AtomicBool,
    #[cfg(all(target_arch = "x86_64", target_os = "linux"))]
    pending_driver_finalizer: crate::post_driver::PendingDriverFinalizerSlot,
}

impl Application {
    /// Creates an empty application assembly.
    pub fn new() -> Self {
        Self {
            components: Vec::new(),
            requests: RequestLifecycle::new(),
            ingress_bridge_issued: AtomicBool::new(false),
            #[cfg(all(target_arch = "x86_64", target_os = "linux"))]
            pending_driver_finalizer: crate::post_driver::PendingDriverFinalizerSlot::new(),
        }
    }

    #[cfg(all(target_arch = "x86_64", target_os = "linux"))]
    pub(crate) fn install_prevalidated_components(
        &mut self,
        components: Vec<Arc<dyn ComponentLifecycle>>,
    ) {
        debug_assert!(self.components.is_empty());
        self.components = components;
    }

    #[cfg(all(target_arch = "x86_64", target_os = "linux"))]
    #[doc(hidden)]
    pub fn pending_driver_finalizer(&self) -> crate::post_driver::PendingDriverFinalizerSlot {
        self.pending_driver_finalizer.clone()
    }

    #[cfg(all(test, target_arch = "x86_64", target_os = "linux"))]
    pub(crate) fn post_driver_is_unarmed_for_test(&self) -> bool {
        self.pending_driver_finalizer.is_unarmed_for_test()
    }

    #[cfg(all(target_arch = "x86_64", target_os = "linux"))]
    #[doc(hidden)]
    #[allow(clippy::result_large_err)]
    pub fn commit_post_driver_install(
        &self,
        binding: saddle_admission::VerifiedPostDriverInstallBinding,
    ) {
        self.pending_driver_finalizer
            .commit_verified_install(binding)
    }

    pub(crate) fn reserved_post_driver_submit(
        &self,
    ) -> crate::post_driver::MustSubmitDriverFinalizer {
        self.pending_driver_finalizer.reserved_submit_handle()
    }

    /// Returns the request lifecycle shared with Saddle's Service adapter.
    pub fn request_lifecycle(&self) -> RequestLifecycle {
        self.requests.clone()
    }

    /// Returns a read-only observer of the framework's unique lifecycle
    /// state. The observer cannot admit requests or mutate readiness.
    pub fn health(&self) -> crate::ApplicationHealth {
        self.requests.health()
    }

    /// Reserves the fixed alpha.1 Ingress execution bridge attached to this
    /// application's existing 0.2 request lifecycle.
    #[doc(hidden)]
    pub fn managed_ingress_bridge(
        &self,
        capacity: usize,
    ) -> Option<crate::alpha1_ingress::ManagedIngressBridge> {
        if capacity == 0 {
            return None;
        }
        if self
            .ingress_bridge_issued
            .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
            .is_err()
        {
            return None;
        }
        crate::alpha1_ingress::ManagedIngressBridge::new(self.requests.clone(), capacity)
    }

    /// Registers a framework component for managed startup and shutdown.
    ///
    /// Components start in registration order and stop in reverse order.
    pub fn register<C>(&mut self, component: C) -> Result<()>
    where
        C: ComponentLifecycle + 'static,
    {
        self.register_shared(Arc::new(component))
    }

    /// Registers an already shared framework component.
    pub fn register_shared(&mut self, component: Arc<dyn ComponentLifecycle>) -> Result<()> {
        if self
            .components
            .iter()
            .any(|registered| registered.name() == component.name())
        {
            return Err(SaddleError::new(
                ErrorKind::Conflict,
                "runtime.duplicate_component",
                format!("component '{}' is already registered", component.name()),
            ));
        }
        self.components.push(component);
        Ok(())
    }

    /// Runs the application on Saddle's single process-wide async runtime.
    ///
    /// The call blocks the process entry thread until SIGINT or, on Unix,
    /// SIGTERM. Shutdown first closes request admission, then waits for every
    /// admitted request, and finally stops components in reverse order.
    pub fn run(self) -> Result<()> {
        Self::run_with(|| async move { Ok(self) })
    }

    /// Creates the application inside Saddle's process-wide async runtime and
    /// then runs it until shutdown.
    ///
    /// This is the framework assembly path for components whose initialization
    /// performs async I/O. Business code is not given a runtime handle or an
    /// executor through this API.
    pub fn run_with<F, Fut>(bootstrap: F) -> Result<()>
    where
        F: FnOnce() -> Fut + Send + 'static,
        Fut: Future<Output = Result<Self>> + Send + 'static,
    {
        if RUNTIME_STARTED
            .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
            .is_err()
        {
            return Err(SaddleError::new(
                ErrorKind::Conflict,
                "runtime.already_started",
                "the Saddle runtime has already started in this process",
            ));
        }

        let runtime = build_runtime()?;

        #[cfg(all(target_arch = "x86_64", target_os = "linux"))]
        {
            Self::run_with_owned_runtime(runtime, bootstrap)
        }

        #[cfg(not(all(target_arch = "x86_64", target_os = "linux")))]
        runtime.block_on(async {
            let signal = ShutdownSignal::register()?;
            bootstrap_and_run(bootstrap, signal.wait()).await
        })
    }

    #[cfg(all(target_arch = "x86_64", target_os = "linux"))]
    pub(crate) fn claim_process_runtime() -> Result<()> {
        if RUNTIME_STARTED
            .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
            .is_err()
        {
            return Err(SaddleError::new(
                ErrorKind::Conflict,
                "runtime.already_started",
                "the Saddle runtime has already started in this process",
            ));
        }
        Ok(())
    }

    #[cfg(all(target_arch = "x86_64", target_os = "linux"))]
    pub(crate) fn run_with_owned_runtime<F, Fut>(
        runtime: tokio::runtime::Runtime,
        bootstrap: F,
    ) -> Result<()>
    where
        F: FnOnce() -> Fut,
        Fut: Future<Output = Result<Self>>,
    {
        let outcome = runtime.block_on(async {
            let signal = ShutdownSignal::register()?;
            let application = bootstrap().await?;
            let finalizer = application.pending_driver_finalizer();
            let result = application.run_until_shutdown(signal.wait()).await;
            Ok::<_, SaddleError>((finalizer, result))
        });
        match outcome {
            Ok((finalizer, result)) => finalizer.finish(runtime, result),
            Err(error) => {
                drop(runtime);
                Err(error)
            }
        }
    }

    pub(crate) async fn run_until_shutdown<F>(self, shutdown: F) -> Result<()>
    where
        F: Future<Output = Result<()>>,
    {
        tokio::pin!(shutdown);
        let signal_before_start = tokio::select! {
            biased;
            signal_result = &mut shutdown => Some(signal_result),
            _ = std::future::ready(()) => None,
        };
        if let Some(signal_result) = signal_before_start {
            self.requests.begin_draining();
            self.requests.wait_until_drained().await;
            self.requests.mark_stopped();
            return signal_result;
        }

        let mut started = 0;

        for component in &self.components {
            let start = component.start();
            tokio::pin!(start);
            let mut shutdown_during_start = None;
            let start_result = tokio::select! {
                biased;
                signal_result = &mut shutdown => {
                    shutdown_during_start = Some(signal_result);
                    // ComponentLifecycle does not define cancellation-safe
                    // startup. Finish the in-progress start before rollback so
                    // partially initialized resources can be shut down safely.
                    start.await
                }
                start_result = &mut start => start_result,
            };

            if let Err(error) = start_result {
                self.requests.begin_draining();
                self.requests.wait_until_drained().await;
                let _ = self.shutdown_components(started).await;
                self.requests.mark_stopped();
                return Err(error);
            }
            started += 1;

            if let Some(signal_result) = shutdown_during_start {
                self.requests.begin_draining();
                self.requests.wait_until_drained().await;
                let shutdown_result = self.shutdown_components(started).await;
                self.requests.mark_stopped();
                return signal_result.and(shutdown_result);
            }
        }

        self.requests.mark_ready();
        let signal_result = shutdown.await;
        self.requests.begin_draining();
        self.requests.wait_until_drained().await;
        let shutdown_result = self.shutdown_components(started).await;
        self.requests.mark_stopped();

        signal_result.and(shutdown_result)
    }

    async fn shutdown_components(&self, started: usize) -> Result<()> {
        let mut first_error = None;
        for component in self.components[..started].iter().rev() {
            if let Err(error) = component.shutdown().await {
                if first_error.is_none() {
                    first_error = Some(error);
                }
            }
        }
        first_error.map_or(Ok(()), Err)
    }
}

#[cfg(any(test, not(all(target_arch = "x86_64", target_os = "linux"))))]
async fn bootstrap_and_run<F, Fut, S>(bootstrap: F, shutdown: S) -> Result<()>
where
    F: FnOnce() -> Fut,
    Fut: Future<Output = Result<Application>>,
    S: Future<Output = Result<()>>,
{
    let application = bootstrap().await?;
    application.run_until_shutdown(shutdown).await
}

impl Default for Application {
    fn default() -> Self {
        Self::new()
    }
}

fn build_runtime() -> Result<tokio::runtime::Runtime> {
    tokio::runtime::Builder::new_multi_thread()
        .worker_threads(WORKER_THREADS)
        .max_io_events_per_tick(MAX_IO_EVENTS_PER_TICK)
        .enable_all()
        .build()
        .map_err(|_| {
            SaddleError::new(
                ErrorKind::Infrastructure,
                "runtime.initialization_failed",
                "failed to initialize the Saddle async runtime",
            )
        })
}

#[cfg(unix)]
pub(crate) struct ShutdownSignal {
    interrupt: tokio::signal::unix::Signal,
    terminate: tokio::signal::unix::Signal,
}

#[cfg(unix)]
impl ShutdownSignal {
    /// Registers both listeners synchronously before any component starts.
    pub(crate) fn register() -> Result<Self> {
        Ok(Self {
            interrupt: tokio::signal::unix::signal(tokio::signal::unix::SignalKind::interrupt())
                .map_err(|_| signal_error())?,
            terminate: tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
                .map_err(|_| signal_error())?,
        })
    }

    pub(crate) async fn wait(mut self) -> Result<()> {
        tokio::select! {
            _ = self.interrupt.recv() => Ok(()),
            _ = self.terminate.recv() => Ok(()),
        }
    }
}

#[cfg(windows)]
struct ShutdownSignal {
    ctrl_c: tokio::signal::windows::CtrlC,
    ctrl_break: tokio::signal::windows::CtrlBreak,
}

#[cfg(windows)]
impl ShutdownSignal {
    /// Registers both listeners synchronously before any component starts.
    fn register() -> Result<Self> {
        Ok(Self {
            ctrl_c: tokio::signal::windows::ctrl_c().map_err(|_| signal_error())?,
            ctrl_break: tokio::signal::windows::ctrl_break().map_err(|_| signal_error())?,
        })
    }

    async fn wait(mut self) -> Result<()> {
        tokio::select! {
            _ = self.ctrl_c.recv() => Ok(()),
            _ = self.ctrl_break.recv() => Ok(()),
        }
    }
}

fn signal_error() -> SaddleError {
    SaddleError::new(
        ErrorKind::Infrastructure,
        "runtime.signal_registration_failed",
        "failed to register the application shutdown signal",
    )
}

#[cfg(test)]
mod tests {
    use std::sync::Mutex;

    use saddle_core::LifecycleFuture;

    use super::*;
    use crate::ApplicationPhase;

    struct RecordingComponent {
        name: &'static str,
        events: Arc<Mutex<Vec<String>>>,
        start_error: bool,
        shutdown_error: bool,
    }

    struct BlockingStartComponent {
        events: Arc<Mutex<Vec<String>>>,
        started: Mutex<Option<tokio::sync::oneshot::Sender<()>>>,
        release: Mutex<Option<tokio::sync::oneshot::Receiver<()>>>,
    }

    struct HealthAwareListener {
        health: crate::ApplicationHealth,
        events: Arc<Mutex<Vec<String>>>,
    }

    impl ComponentLifecycle for RecordingComponent {
        fn name(&self) -> &'static str {
            self.name
        }

        fn start(&self) -> LifecycleFuture<'_> {
            Box::pin(async move {
                self.events
                    .lock()
                    .unwrap()
                    .push(format!("start:{}", self.name));
                if self.start_error {
                    Err(test_error("start failed"))
                } else {
                    Ok(())
                }
            })
        }

        fn shutdown(&self) -> LifecycleFuture<'_> {
            Box::pin(async move {
                self.events
                    .lock()
                    .unwrap()
                    .push(format!("shutdown:{}", self.name));
                if self.shutdown_error {
                    Err(test_error("shutdown failed"))
                } else {
                    Ok(())
                }
            })
        }
    }

    impl ComponentLifecycle for BlockingStartComponent {
        fn name(&self) -> &'static str {
            "blocking"
        }

        fn start(&self) -> LifecycleFuture<'_> {
            Box::pin(async move {
                self.events
                    .lock()
                    .unwrap()
                    .push("start:blocking".to_owned());
                let started = self.started.lock().unwrap().take().unwrap();
                let release = self.release.lock().unwrap().take().unwrap();
                started.send(()).unwrap();
                release.await.unwrap();
                Ok(())
            })
        }

        fn shutdown(&self) -> LifecycleFuture<'_> {
            Box::pin(async move {
                self.events
                    .lock()
                    .unwrap()
                    .push("shutdown:blocking".to_owned());
                Ok(())
            })
        }
    }

    impl ComponentLifecycle for HealthAwareListener {
        fn name(&self) -> &'static str {
            "health-aware-listener"
        }

        fn start(&self) -> LifecycleFuture<'_> {
            Box::pin(async move {
                let snapshot = self.health.snapshot();
                assert!(snapshot.is_live());
                assert!(!snapshot.is_ready());
                assert_eq!(snapshot.phase(), ApplicationPhase::Starting);
                self.events
                    .lock()
                    .unwrap()
                    .push("listener:accepting".into());
                Ok(())
            })
        }

        fn shutdown(&self) -> LifecycleFuture<'_> {
            Box::pin(async move {
                let snapshot = self.health.snapshot();
                assert!(snapshot.is_live());
                assert!(!snapshot.is_ready());
                assert_eq!(snapshot.phase(), ApplicationPhase::Draining);
                self.events.lock().unwrap().push("listener:stopped".into());
                Ok(())
            })
        }
    }

    fn component(name: &'static str, events: &Arc<Mutex<Vec<String>>>) -> RecordingComponent {
        RecordingComponent {
            name,
            events: Arc::clone(events),
            start_error: false,
            shutdown_error: false,
        }
    }

    fn test_error(message: &'static str) -> SaddleError {
        SaddleError::new(ErrorKind::Infrastructure, "test.failure", message)
    }

    fn test_runtime() -> tokio::runtime::Runtime {
        tokio::runtime::Builder::new_current_thread()
            .build()
            .expect("test runtime must build")
    }

    async fn shutdown_when_ready(requests: RequestLifecycle) -> Result<()> {
        while requests.phase() != crate::ApplicationPhase::Ready {
            tokio::task::yield_now().await;
        }
        Ok(())
    }

    #[test]
    fn components_start_in_order_and_shutdown_in_reverse() {
        let events = Arc::new(Mutex::new(Vec::new()));
        let mut application = Application::new();
        application.register(component("db", &events)).unwrap();
        application.register(component("service", &events)).unwrap();
        let shutdown = shutdown_when_ready(application.request_lifecycle());

        test_runtime()
            .block_on(application.run_until_shutdown(shutdown))
            .unwrap();

        assert_eq!(
            *events.lock().unwrap(),
            [
                "start:db",
                "start:service",
                "shutdown:service",
                "shutdown:db"
            ]
        );
    }

    #[test]
    fn health_uses_the_unique_lifecycle_and_clears_ready_before_listener_shutdown() {
        test_runtime().block_on(async {
            let events = Arc::new(Mutex::new(Vec::new()));
            let mut application = Application::new();
            let health = application.health();
            let initial = health.snapshot();
            assert!(initial.is_live());
            assert!(!initial.is_ready());
            assert_eq!(initial.phase(), ApplicationPhase::Starting);

            application
                .register(HealthAwareListener {
                    health: health.clone(),
                    events: Arc::clone(&events),
                })
                .unwrap();
            let shutdown_health = health.clone();
            application
                .run_until_shutdown(async move {
                    loop {
                        let snapshot = shutdown_health.snapshot();
                        if snapshot.is_ready() {
                            assert!(snapshot.is_live());
                            assert_eq!(snapshot.phase(), ApplicationPhase::Ready);
                            return Ok(());
                        }
                        tokio::task::yield_now().await;
                    }
                })
                .await
                .unwrap();

            let stopped = health.snapshot();
            assert!(!stopped.is_live());
            assert!(!stopped.is_ready());
            assert_eq!(stopped.phase(), ApplicationPhase::Stopped);
            assert_eq!(
                *events.lock().unwrap(),
                ["listener:accepting", "listener:stopped"]
            );
        });
    }

    #[test]
    fn async_bootstrap_runs_before_early_shutdown_prevents_component_start() {
        let events = Arc::new(Mutex::new(Vec::new()));
        let bootstrap_events = Arc::clone(&events);

        test_runtime()
            .block_on(bootstrap_and_run(
                move || async move {
                    bootstrap_events
                        .lock()
                        .unwrap()
                        .push("bootstrap".to_owned());
                    let mut application = Application::new();
                    application.register(component("component", &bootstrap_events))?;
                    Ok(application)
                },
                std::future::ready(Ok(())),
            ))
            .unwrap();

        assert_eq!(*events.lock().unwrap(), ["bootstrap"]);
    }

    #[test]
    fn failed_async_bootstrap_does_not_start_components() {
        let error = test_runtime()
            .block_on(bootstrap_and_run(
                || async { Err(test_error("bootstrap failed")) },
                std::future::pending(),
            ))
            .unwrap_err();
        assert_eq!(error.message(), "bootstrap failed");
    }

    #[test]
    fn startup_failure_rolls_back_only_started_components() {
        let events = Arc::new(Mutex::new(Vec::new()));
        let mut application = Application::new();
        application.register(component("first", &events)).unwrap();
        let mut failing = component("failing", &events);
        failing.start_error = true;
        application.register(failing).unwrap();
        application.register(component("never", &events)).unwrap();
        let shutdown = shutdown_when_ready(application.request_lifecycle());

        let error = test_runtime()
            .block_on(application.run_until_shutdown(shutdown))
            .unwrap_err();

        assert_eq!(error.message(), "start failed");
        assert_eq!(
            *events.lock().unwrap(),
            ["start:first", "start:failing", "shutdown:first"]
        );
    }

    #[test]
    fn shutdown_continues_after_a_component_error() {
        let events = Arc::new(Mutex::new(Vec::new()));
        let mut application = Application::new();
        application.register(component("first", &events)).unwrap();
        let mut failing = component("second", &events);
        failing.shutdown_error = true;
        application.register(failing).unwrap();
        let shutdown = shutdown_when_ready(application.request_lifecycle());

        let error = test_runtime()
            .block_on(application.run_until_shutdown(shutdown))
            .unwrap_err();

        assert_eq!(error.message(), "shutdown failed");
        assert_eq!(
            *events.lock().unwrap(),
            [
                "start:first",
                "start:second",
                "shutdown:second",
                "shutdown:first"
            ]
        );
    }

    #[test]
    fn duplicate_component_names_are_rejected() {
        let events = Arc::new(Mutex::new(Vec::new()));
        let mut application = Application::new();
        application.register(component("db", &events)).unwrap();

        let error = application.register(component("db", &events)).unwrap_err();
        assert_eq!(error.code(), "runtime.duplicate_component");
    }

    #[test]
    fn application_shutdown_waits_for_an_admitted_request() {
        test_runtime().block_on(async {
            let application = Application::new();
            let requests = application.request_lifecycle();
            let (release, released) = tokio::sync::oneshot::channel();

            let shutdown = async move {
                shutdown_when_ready(requests.clone()).await?;
                let request = requests
                    .try_accept()
                    .expect("application is ready before waiting for shutdown");
                tokio::spawn(async move {
                    released.await.unwrap();
                    drop(request);
                });
                Ok(())
            };
            let running = tokio::spawn(application.run_until_shutdown(shutdown));

            tokio::task::yield_now().await;
            assert!(!running.is_finished());
            release.send(()).unwrap();
            running.await.unwrap().unwrap();
        });
    }

    #[test]
    fn signal_failure_before_start_prevents_component_startup() {
        let events = Arc::new(Mutex::new(Vec::new()));
        let mut application = Application::new();
        application.register(component("service", &events)).unwrap();

        let error = test_runtime()
            .block_on(application.run_until_shutdown(async { Err(signal_error()) }))
            .unwrap_err();

        assert_eq!(error.code(), "runtime.signal_registration_failed");
        assert!(events.lock().unwrap().is_empty());
    }

    #[test]
    fn shutdown_during_startup_stops_starting_and_rolls_back() {
        test_runtime().block_on(async {
            let events = Arc::new(Mutex::new(Vec::new()));
            let (started_tx, started_rx) = tokio::sync::oneshot::channel();
            let (release_tx, release_rx) = tokio::sync::oneshot::channel();
            let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel();
            let mut application = Application::new();
            application
                .register(BlockingStartComponent {
                    events: Arc::clone(&events),
                    started: Mutex::new(Some(started_tx)),
                    release: Mutex::new(Some(release_rx)),
                })
                .unwrap();
            application.register(component("never", &events)).unwrap();

            let running = tokio::spawn(application.run_until_shutdown(async move {
                shutdown_rx.await.unwrap();
                Ok(())
            }));
            started_rx.await.unwrap();
            shutdown_tx.send(()).unwrap();
            tokio::task::yield_now().await;
            release_tx.send(()).unwrap();

            running.await.unwrap().unwrap();
            assert_eq!(
                *events.lock().unwrap(),
                ["start:blocking", "shutdown:blocking"]
            );
        });
    }

    #[test]
    fn managed_runtime_provides_an_async_io_driver() {
        build_runtime()
            .unwrap()
            .block_on(async {
                tokio::net::TcpListener::bind((std::net::Ipv4Addr::LOCALHOST, 0)).await
            })
            .expect("service listeners require the managed async I/O driver");
    }
}