saddle_runtime/

application.rs

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

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;
const DEFAULT_START_TIMEOUT: Duration = Duration::from_secs(30);
const DEFAULT_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(30);

/// Fixed wall-clock limits for the managed component lifecycle.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct LifecycleTimeouts {
    start: Duration,
    shutdown: Duration,
}

impl LifecycleTimeouts {
    pub fn from_millis(start_ms: u64, shutdown_ms: u64) -> Option<Self> {
        if start_ms == 0 || shutdown_ms == 0 {
            return None;
        }
        Some(Self {
            start: Duration::from_millis(start_ms),
            shutdown: Duration::from_millis(shutdown_ms),
        })
    }
}

impl Default for LifecycleTimeouts {
    fn default() -> Self {
        Self {
            start: DEFAULT_START_TIMEOUT,
            shutdown: DEFAULT_SHUTDOWN_TIMEOUT,
        }
    }
}

/// 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,
    deployment_resource_budget: Option<saddle_admission::DeploymentResourceBudget>,
    ingress_bridge_issued: AtomicBool,
    lifecycle_timeouts: LifecycleTimeouts,
    shutdown_deadline: Arc<std::sync::Mutex<Option<Instant>>>,
    lifecycle_observer: Option<(saddle_observability::Observer, String)>,
    #[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(),
            deployment_resource_budget: None,
            ingress_bridge_issued: AtomicBool::new(false),
            lifecycle_timeouts: LifecycleTimeouts::default(),
            shutdown_deadline: Arc::new(std::sync::Mutex::new(None)),
            lifecycle_observer: None,
            #[cfg(all(target_arch = "x86_64", target_os = "linux"))]
            pending_driver_finalizer: crate::post_driver::PendingDriverFinalizerSlot::new(),
        }
    }

    /// Installs the frozen process lifecycle policy before component startup.
    #[doc(hidden)]
    pub fn set_lifecycle_timeouts(&mut self, timeouts: LifecycleTimeouts) {
        self.lifecycle_timeouts = timeouts;
    }

    #[doc(hidden)]
    pub fn install_lifecycle_observer(
        &mut self,
        observer: saddle_observability::Observer,
        application: &str,
    ) {
        self.lifecycle_observer = Some((observer, application.to_owned()));
    }

    #[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
        })
    }

    /// Runs the formal process while retaining its one frozen deployment
    /// budget inside Runtime assembly. No read or replacement surface escapes.
    #[doc(hidden)]
    pub fn run_with_deployment_resource_budget<F, Fut>(
        budget: saddle_admission::DeploymentResourceBudget,
        bootstrap: F,
    ) -> Result<()>
    where
        F: FnOnce() -> Fut + Send + 'static,
        Fut: Future<Output = Result<Self>> + Send + 'static,
    {
        Self::run_with(move || async move {
            let mut application = bootstrap().await?;
            if application.deployment_resource_budget.is_some() {
                return Err(SaddleError::new(
                    ErrorKind::Conflict,
                    "runtime.deployment_resource_budget_already_installed",
                    "the deployment resource budget was already installed",
                ));
            }
            application.deployment_resource_budget = Some(budget);
            Ok(application)
        })
    }

    #[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 shutdown_deadline = Arc::clone(&application.shutdown_deadline);
            let lifecycle_observer = application.lifecycle_observer_handle();
            let result = application.run_until_shutdown(signal.wait()).await;
            Ok::<_, SaddleError>((finalizer, shutdown_deadline, lifecycle_observer, result))
        });
        match outcome {
            Ok((finalizer, shutdown_deadline, lifecycle_observer, result)) => {
                let deadline = *shutdown_deadline
                    .lock()
                    .unwrap_or_else(|poisoned| poisoned.into_inner());
                finalizer.finish(runtime, result, deadline, lifecycle_observer)
            }
            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_started = Instant::now();
            let start = crate::diagnostics::task(
                async { component.start().await },
                saddle_core::DiagnosticStage::StartupListener,
                "runtime.component_start",
            );
            tokio::pin!(start);
            let mut shutdown_during_start = None;
            let start_result = tokio::select! {
                biased;
                signal_result = &mut shutdown => {
                    let deadline = Instant::now() + self.lifecycle_timeouts.shutdown;
                    self.set_shutdown_deadline(deadline);
                    shutdown_during_start = Some((signal_result, deadline));
                    // A component may have partially initialized before its
                    // start future yielded. Bound that in-progress start by
                    // both lifecycle budgets, then include it in rollback.
                    let start_deadline = std::cmp::min(
                        start_started + self.lifecycle_timeouts.start,
                        deadline,
                    );
                    match tokio::time::timeout_at(start_deadline.into(), start).await {
                        Ok(result) => result,
                        Err(_) => Err(lifecycle_timeout_error("component_start")),
                    }
                }
                start_result = tokio::time::timeout(self.lifecycle_timeouts.start, &mut start) => {
                    start_result.unwrap_or_else(|_| Err(lifecycle_timeout_error("component_start")))
                },
            };

            if let Err(error) = start_result {
                // Fix cleanup's budget BEFORE any diagnostic capture/submit.
                let deadline = shutdown_during_start
                    .as_ref()
                    .map(|(_, deadline)| *deadline)
                    .unwrap_or_else(|| Instant::now() + self.lifecycle_timeouts.shutdown);
                self.set_shutdown_deadline(deadline);
                let error = crate::diagnostics::attach(
                    error,
                    saddle_core::DiagnosticStage::StartupListener,
                    "runtime.component_start_failed",
                );
                crate::diagnostics::report(&error);
                self.record_timeout(&error, start_started.elapsed());
                self.requests.begin_draining();
                let cleanup_count = if error.code() == "runtime.lifecycle_timeout.component_start" {
                    started + 1
                } else {
                    started
                };
                let drain_result = timeout_at(
                    deadline,
                    self.requests.wait_until_drained(),
                    "request_drain",
                )
                .await;
                if drain_result.is_ok() {
                    let _ = self
                        .shutdown_components(cleanup_count, deadline, Some(&error))
                        .await;
                } else if let Err(cleanup) = drain_result {
                    crate::diagnostics::cleanup(
                        cleanup,
                        Some(&error),
                        saddle_core::DiagnosticStage::ShutdownComponent,
                    );
                }
                self.requests.mark_stopped();
                return Err(error);
            }
            started += 1;

            if let Some((signal_result, deadline)) = shutdown_during_start {
                self.requests.begin_draining();
                let drain_result = timeout_at(
                    deadline,
                    self.requests.wait_until_drained(),
                    "request_drain",
                )
                .await;
                let signal_result = signal_result.map_err(|e| {
                    let e = crate::diagnostics::attach(
                        e,
                        saddle_core::DiagnosticStage::ShutdownComponent,
                        "runtime.shutdown_signal_failed",
                    );
                    crate::diagnostics::report(&e);
                    e
                });
                let shutdown_result = if drain_result.is_ok() {
                    self.shutdown_components(started, deadline, signal_result.as_ref().err())
                        .await
                } else {
                    drain_result.map_err(|e| {
                        crate::diagnostics::cleanup(
                            e,
                            signal_result.as_ref().err(),
                            saddle_core::DiagnosticStage::ShutdownComponent,
                        )
                    })
                };
                self.requests.mark_stopped();
                return signal_result.and(shutdown_result);
            }
        }

        self.requests.mark_ready();
        let signal_result = shutdown.await;
        let deadline = Instant::now() + self.lifecycle_timeouts.shutdown;
        self.set_shutdown_deadline(deadline);
        self.requests.begin_draining();
        let signal_result = signal_result.map_err(|e| {
            let e = crate::diagnostics::attach(
                e,
                saddle_core::DiagnosticStage::ShutdownComponent,
                "runtime.shutdown_signal_failed",
            );
            crate::diagnostics::report(&e);
            e
        });
        let drain_result = timeout_at(
            deadline,
            self.requests.wait_until_drained(),
            "request_drain",
        )
        .await;
        if let Err(error) = &drain_result {
            self.record_timeout(error, self.lifecycle_timeouts.shutdown);
        }
        let shutdown_result = if drain_result.is_ok() {
            self.shutdown_components(started, deadline, signal_result.as_ref().err())
                .await
        } else {
            drain_result.map_err(|e| {
                crate::diagnostics::cleanup(
                    e,
                    signal_result.as_ref().err(),
                    saddle_core::DiagnosticStage::ShutdownComponent,
                )
            })
        };
        self.requests.mark_stopped();

        signal_result.and(shutdown_result)
    }

    fn set_shutdown_deadline(&self, deadline: Instant) {
        *self
            .shutdown_deadline
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(deadline);
    }

    pub(crate) fn shutdown_deadline_handle(&self) -> Arc<std::sync::Mutex<Option<Instant>>> {
        Arc::clone(&self.shutdown_deadline)
    }

    pub(crate) fn lifecycle_observer_handle(
        &self,
    ) -> Option<(saddle_observability::Observer, String)> {
        self.lifecycle_observer.clone()
    }

    fn record_timeout(&self, error: &SaddleError, elapsed: Duration) {
        let stage = match error.code() {
            "runtime.lifecycle_timeout.component_start" => {
                saddle_observability::LifecycleTimeoutStage::ComponentStart
            }
            "runtime.lifecycle_timeout.request_drain" => {
                saddle_observability::LifecycleTimeoutStage::RequestDrain
            }
            "runtime.lifecycle_timeout.component_shutdown" => {
                saddle_observability::LifecycleTimeoutStage::ComponentShutdown
            }
            _ => return,
        };
        if let Some((observer, application)) = &self.lifecycle_observer {
            observer.record_lifecycle_timeout(
                application.as_str(),
                stage,
                u64::try_from(elapsed.as_millis()).unwrap_or(u64::MAX),
            );
        }
    }

    async fn shutdown_components(
        &self,
        started: usize,
        deadline: Instant,
        primary: Option<&SaddleError>,
    ) -> Result<()> {
        let mut first_error = None;
        for component in self.components[..started].iter().rev() {
            let result = match timeout_at(
                deadline,
                crate::diagnostics::task(
                    async { component.shutdown().await },
                    saddle_core::DiagnosticStage::ShutdownComponent,
                    "runtime.component_shutdown",
                ),
                "component_shutdown",
            )
            .await
            {
                Ok(result) => result,
                Err(error) => Err(error),
            };
            if let Err(error) = result {
                let error = crate::diagnostics::cleanup(
                    error,
                    primary.or(first_error.as_ref()),
                    saddle_core::DiagnosticStage::ShutdownComponent,
                );
                self.record_timeout(&error, self.lifecycle_timeouts.shutdown);
                if first_error.is_none() {
                    first_error = Some(error);
                }
                if Instant::now() >= deadline {
                    break;
                }
            }
        }
        first_error.map_or(Ok(()), Err)
    }
}

async fn timeout_at<T>(
    deadline: Instant,
    future: impl Future<Output = T>,
    stage: &'static str,
) -> Result<T> {
    let remaining = deadline.saturating_duration_since(Instant::now());
    tokio::time::timeout(remaining, future)
        .await
        .map_err(|_| lifecycle_timeout_error(stage))
}

fn lifecycle_timeout_error(stage: &'static str) -> SaddleError {
    SaddleError::new(
        ErrorKind::Infrastructure,
        match stage {
            "component_start" => "runtime.lifecycle_timeout.component_start",
            "request_drain" => "runtime.lifecycle_timeout.request_drain",
            "component_shutdown" => "runtime.lifecycle_timeout.component_shutdown",
            _ => "runtime.lifecycle_timeout.post_driver",
        },
        format!("managed lifecycle stage '{stage}' exceeded its wall-clock deadline"),
    )
}

#[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(all(target_arch = "x86_64", target_os = "linux"))]
pub(crate) fn claim_owned_runtime() -> Result<tokio::runtime::Runtime> {
    Application::claim_process_runtime()?;
    build_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 BlockingShutdownComponent {
        events: Arc<Mutex<Vec<String>>>,
    }

    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 BlockingShutdownComponent {
        fn name(&self) -> &'static str {
            "blocking-shutdown"
        }

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

        fn shutdown(&self) -> LifecycleFuture<'_> {
            Box::pin(async move {
                self.events
                    .lock()
                    .unwrap()
                    .push("shutdown:blocking-shutdown".into());
                std::future::pending().await
            })
        }
    }

    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()
            .enable_time()
            .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");
    }

    #[test]
    fn blocked_component_start_times_out_and_rolls_back_started_components() {
        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 mut application = Application::new();
            application.set_lifecycle_timeouts(LifecycleTimeouts::from_millis(10, 100).unwrap());
            application.register(component("first", &events)).unwrap();
            application
                .register(BlockingStartComponent {
                    events: Arc::clone(&events),
                    started: Mutex::new(Some(started_tx)),
                    release: Mutex::new(Some(release_rx)),
                })
                .unwrap();

            let running = tokio::spawn(application.run_until_shutdown(std::future::pending()));
            started_rx.await.unwrap();
            let error = running.await.unwrap().unwrap_err();
            assert_eq!(error.code(), "runtime.lifecycle_timeout.component_start");
            assert_eq!(
                *events.lock().unwrap(),
                [
                    "start:first",
                    "start:blocking",
                    "shutdown:blocking",
                    "shutdown:first"
                ]
            );
        });
    }

    #[test]
    fn blocked_component_shutdown_uses_one_total_deadline_and_is_not_clean() {
        test_runtime().block_on(async {
            let events = Arc::new(Mutex::new(Vec::new()));
            let mut application = Application::new();
            application.set_lifecycle_timeouts(LifecycleTimeouts::from_millis(100, 10).unwrap());
            application
                .register(BlockingShutdownComponent {
                    events: Arc::clone(&events),
                })
                .unwrap();
            let shutdown = shutdown_when_ready(application.request_lifecycle());
            let error = application.run_until_shutdown(shutdown).await.unwrap_err();
            assert_eq!(error.code(), "runtime.lifecycle_timeout.component_shutdown");
            assert_eq!(
                *events.lock().unwrap(),
                ["start:blocking-shutdown", "shutdown:blocking-shutdown"]
            );
        });
    }
}