saddle-runtime 0.3.18

//! Application-owned fixed post-driver finalizer slot.
//!
//! This is framework assembly surface only. It is public because Rust has no
//! cross-crate friend visibility and is not re-exported by the Saddle facade.

use std::{
    sync::{Arc, Mutex, MutexGuard, mpsc},
    time::Instant,
};

use saddle_admission::VerifiedPostDriverInstallBinding;
use saddle_core::{ErrorKind, Result, SaddleError};

use crate::compiled_route::OfficialCompiledDriverFinalizer;

enum InstallState {
    Unarmed,
    MustSubmit,
    Consumed,
}

struct State {
    install: InstallState,
    submitted: Option<OfficialCompiledDriverFinalizer>,
    consumed: bool,
}

/// Cloneable access to one Application-owned fixed slot. Clones do not own the
/// finalizer; they can only arm the contract or linearly submit its opaque
/// value during component shutdown.
#[derive(Clone)]
pub struct PendingDriverFinalizerSlot {
    state: Arc<Mutex<State>>,
}

impl PendingDriverFinalizerSlot {
    pub(crate) fn new() -> Self {
        Self {
            state: Arc::new(Mutex::new(State {
                install: InstallState::Unarmed,
                submitted: None,
                consumed: false,
            })),
        }
    }

    #[cfg(test)]
    pub(crate) fn is_unarmed_for_test(&self) -> bool {
        let state = lock(&self.state);
        matches!(state.install, InstallState::Unarmed)
            && state.submitted.is_none()
            && !state.consumed
    }

    /// Performs the final synchronous, infallible authority commit. All
    /// fallible/async work completed while the slot was still unarmed.
    #[doc(hidden)]
    pub fn commit_verified_install(&self, binding: VerifiedPostDriverInstallBinding) {
        let mut state = lock(&self.state);
        if !matches!(state.install, InstallState::Unarmed)
            || state.submitted.is_some()
            || state.consumed
        {
            std::process::abort();
        }
        let _binding = binding;
        state.install = InstallState::MustSubmit;
    }

    pub(crate) fn reserved_submit_handle(&self) -> MustSubmitDriverFinalizer {
        MustSubmitDriverFinalizer { slot: self.clone() }
    }

    /// Submits the sole opaque finalizer. Duplicate, unarmed or post-consume
    /// submission is a finite fail-closed contract violation.
    pub fn submit(&self, finalizer: OfficialCompiledDriverFinalizer) {
        let mut state = lock(&self.state);
        if !matches!(state.install, InstallState::MustSubmit)
            || state.submitted.is_some()
            || state.consumed
        {
            std::process::abort();
        }
        state.submitted = Some(finalizer);
    }

    pub(crate) fn finish(
        &self,
        runtime: tokio::runtime::Runtime,
        application_result: Result<()>,
        shutdown_deadline: Option<Instant>,
        lifecycle_observer: Option<(saddle_observability::Observer, String)>,
    ) -> Result<()> {
        let pending = {
            let mut state = lock(&self.state);
            if state.consumed {
                std::process::abort();
            }
            state.consumed = true;
            let must_submit = matches!(state.install, InstallState::MustSubmit);
            state.install = InstallState::Consumed;
            if must_submit {
                Some(
                    state
                        .submitted
                        .take()
                        .unwrap_or_else(|| std::process::abort()),
                )
            } else {
                None
            }
        };

        let finalizer_result = match pending {
            Some(pending) => {
                let Some(deadline) = shutdown_deadline else {
                    drop(runtime);
                    return application_result.and(Err(finalization_error()));
                };
                let (sender, receiver) = mpsc::sync_channel(1);
                std::thread::spawn(move || {
                    let result = crate::diagnostics::catching(
                        saddle_core::DiagnosticStage::FinalizerResource,
                        "runtime.post_driver",
                        None,
                        || {
                            pending
                                .bind_runtime(runtime)
                                .finish()
                                .map_err(|_| finalization_failure("runtime.finalizer_bind_failed"))
                                .and_then(|report| {
                                    #[cfg(test)]
                                    if std::env::var_os("RUNTIME_DIAGNOSTIC_FINALIZER_FAULT")
                                        .is_some()
                                    {
                                        // Actual post-driver worker, after physical
                                        // driver/ledger finalization, before result delivery.
                                        panic!("DIAGNOSTIC_PRIVATE_SENTINEL");
                                    }
                                    if report.ledger.healthy
                                        && !report.watermark.breached
                                        && !report.task_failed
                                    {
                                        Ok(())
                                    } else {
                                        Err(finalization_failure(if !report.ledger.healthy {
                                            "runtime.finalizer_ledger_unhealthy"
                                        } else if report.watermark.breached {
                                            "runtime.finalizer_watermark_breached"
                                        } else {
                                            "runtime.finalizer_task_failed"
                                        }))
                                    }
                                })
                        },
                    )
                    .unwrap_or_else(|d| Err(finalization_error().with_diagnostic(d)));
                    let _ = sender.send(result);
                });
                let wait_started = Instant::now();
                match receiver.recv_timeout(deadline.saturating_duration_since(wait_started)) {
                    Ok(result) => result,
                    Err(error) => {
                        let failed = receive_failure(error);
                        let failed = crate::diagnostics::cleanup(
                            failed,
                            application_result.as_ref().err(),
                            saddle_core::DiagnosticStage::FinalizerResource,
                        );
                        if matches!(error, mpsc::RecvTimeoutError::Timeout) {
                            if let Some((observer, application)) = lifecycle_observer {
                                observer.record_lifecycle_timeout(
                                application.as_str(),
                                saddle_observability::LifecycleTimeoutStage::PostDriverFinalization,
                                u64::try_from(wait_started.elapsed().as_millis())
                                    .unwrap_or(u64::MAX),
                            );
                            }
                        }
                        drop(failed);
                        std::process::abort()
                    }
                }
            }
            None => {
                drop(runtime);
                Ok(())
            }
        };
        let finalizer_result = finalizer_result.map_err(|error| {
            crate::diagnostics::cleanup(
                error,
                application_result.as_ref().err(),
                saddle_core::DiagnosticStage::FinalizerResource,
            )
        });
        application_result.and(finalizer_result)
    }
}

fn finalization_failure(code: &'static str) -> SaddleError {
    crate::diagnostics::attach(
        finalization_error(),
        saddle_core::DiagnosticStage::FinalizerResource,
        code,
    )
}

fn receive_failure(error: mpsc::RecvTimeoutError) -> SaddleError {
    finalization_failure(match error {
        mpsc::RecvTimeoutError::Timeout => "runtime.finalizer_timeout",
        mpsc::RecvTimeoutError::Disconnected => "runtime.finalizer_disconnected",
    })
}

/// Component-facing submission half. It conveys no Admission or identity
/// facts and can only submit into the already committed slot.
#[doc(hidden)]
pub struct MustSubmitDriverFinalizer {
    slot: PendingDriverFinalizerSlot,
}

impl MustSubmitDriverFinalizer {
    #[doc(hidden)]
    pub fn submit(self, finalizer: OfficialCompiledDriverFinalizer) {
        self.slot.submit(finalizer);
    }
}

fn lock(state: &Mutex<State>) -> MutexGuard<'_, State> {
    state.lock().unwrap_or_else(|_| std::process::abort())
}

fn finalization_error() -> SaddleError {
    SaddleError::new(
        ErrorKind::Infrastructure,
        "runtime.driver_finalization_failed",
        "the managed Runtime driver did not finalize cleanly",
    )
}

#[cfg(test)]
pub(crate) fn diagnostic_test_finish(
    runtime: tokio::runtime::Runtime,
    pending: OfficialCompiledDriverFinalizer,
    primary: SaddleError,
) -> SaddleError {
    let slot = PendingDriverFinalizerSlot::new();
    lock(&slot.state).install = InstallState::MustSubmit;
    slot.submit(pending);
    let result = slot.finish(
        runtime,
        Err(primary),
        Some(Instant::now() + std::time::Duration::from_secs(2)),
        None,
    );
    result.expect_err("primary error must remain the returned error")
}

#[cfg(test)]
mod diagnostic_tests {
    use super::*;
    #[test]
    fn disconnected_is_not_timeout() {
        for (error, expected) in [
            (
                mpsc::RecvTimeoutError::Disconnected,
                "runtime.finalizer_disconnected",
            ),
            (mpsc::RecvTimeoutError::Timeout, "runtime.finalizer_timeout"),
        ] {
            let error = receive_failure(error);
            let json = serde_json::to_value(error.diagnostic().unwrap()).unwrap();
            assert_eq!(json["causes"][0]["code"], expected);
        }
    }
}