saddle-framework 0.3.35

//! Formal reserved DB handoff. The listener switch supplies the mandatory
//! original context; no context or account is manufactured in this layer.
use super::*;
use saddle_runtime::profusegw::{ProfuseGwReservedScopeFailure, ProfuseGwSuspendedScope};
use saddle_runtime::request_task::reserved::ReservedRequestFailure;

type Cancel = tokio::sync::futures::OwnedNotified;

// This guard is outside the borrowed D future. D's kernel is dropped first on
// cancellation, then this guard places the original physical terminal back in
// the supervisor-owned RequestState. It never signs NotUsed for Entered.
pub(super) struct ReservedDbOwner<T> {
    pub(super) serial: Option<SerialScope>,
    suspended: Option<ProfuseGwSuspendedScope<Cancel, T>>,
    state: Arc<RequestState>,
    process: Arc<saddle_db::internal::StartupManagedDatabaseProcessCapability>,
}
impl<T> ReservedDbOwner<T> {
    pub(super) async fn resolve(
        &mut self,
        result: Result<
            saddle_db::internal::ReservedDatabaseCompletion<Cancel, T>,
            ProfuseGwReservedScopeFailure,
        >,
        context: &saddle_runtime::request_task::reserved::ReservedTaskContext,
    ) -> Result<T, ScopeDatabaseError> {
        match result {
            Ok(completion) => {
                self.suspended = Some(completion.suspended);
                self.retain(completion.unretained);
                Ok(self.publish().await)
            }
            Err(error) => {
                let class = match &error {
                    ProfuseGwReservedScopeFailure::Preparation(reason) => {
                        use saddle_core::{
                            BoundedDiagnostic, BoundedDiagnosticCause, CaptureSite,
                            DiagnosticCategory, DiagnosticCode, DiagnosticStage,
                        };
                        let code = DiagnosticCode::new("service.database.preparation").unwrap();
                        self.retain(Some(context.view().source_description(
                            &PreparationDescription(reason),
                            BoundedDiagnostic::capture(
                                DiagnosticCategory::UnexpectedError,
                                CaptureSite::FirstObserved,
                                BoundedDiagnosticCause::new(DiagnosticStage::RequestDb, code),
                            ),
                            code,
                            self.state.diagnostic_handle.as_ref(),
                            saddle_observability::root_diagnostic::RootRequestEvent::Database,
                            Default::default(),
                        )));
                        ScopeDatabaseError::Resource
                    }
                    ProfuseGwReservedScopeFailure::Execution(reason) => match reason {
                        saddle_runtime::profusegw::ProfuseGwScopeFailure::Admission(error) => {
                            let failure=context.view().source_error(error,self.state.diagnostic_handle.as_ref(),saddle_core::DiagnosticStage::RequestDb,
                                saddle_observability::root_diagnostic::RootRequestEvent::Database,Default::default());
                            self.retain(Some(failure));
                            ScopeDatabaseError::Resource
                        },
                        saddle_runtime::profusegw::ProfuseGwScopeFailure::Stopped(
                            saddle_runtime::profusegw::ProfuseGwScopeStop::TimedOut,
                        ) => ScopeDatabaseError::TimedOut,
                        saddle_runtime::profusegw::ProfuseGwScopeFailure::Stopped(
                            saddle_runtime::profusegw::ProfuseGwScopeStop::Cancelled,
                        ) => ScopeDatabaseError::Cancelled,
                        saddle_runtime::profusegw::ProfuseGwScopeFailure::Panicked => {
                            ScopeDatabaseError::Panicked
                        }
                        _ => ScopeDatabaseError::State,
                    },
                };
                let suspended = self.process.finish_reserved_without_connection(
                    self.serial
                        .take()
                        .expect("preparation returns original owner"),
                    error,
                );
                let (_original, terminal) = suspended
                    .into_response_parts()
                    .unwrap_or_else(|_| std::process::abort());
                *self.state.terminal.lock().unwrap() = RequestTerminal::PostDatabase(terminal);
                self.state.completed.notify_one();
                Err(class)
            }
        }
    }
    async fn publish(&mut self) -> T {
        let suspended = self
            .suspended
            .as_mut()
            .expect("D completion saved before publication");
        let (terminal, value) = match suspended.resume().await {
            Ok((scope, value)) => (RequestTerminal::Serial(scope), value),
            Err(_) => {
                let (value, terminal) = self
                    .suspended
                    .take()
                    .unwrap()
                    .into_response_parts()
                    .unwrap_or_else(|_| std::process::abort());
                (RequestTerminal::PostDatabase(terminal), value)
            }
        };
        self.suspended = None;
        *self.state.terminal.lock().unwrap() = terminal;
        self.state.completed.notify_one();
        value
    }
    fn retain(&self, failure: Option<ReservedRequestFailure>) {
        if let Some(failure) = failure {
            let mut slot = self.state.reserved_failure.lock().unwrap();
            assert!(
                slot.is_none(),
                "pending original failure prohibits later DB execution"
            );
            *slot = Some(failure);
        }
    }
}
impl<T> Drop for ReservedDbOwner<T> {
    fn drop(&mut self) {
        let terminal = if let Some(suspended) = self.suspended.take() {
            let (_, terminal) = suspended
                .into_response_parts()
                .unwrap_or_else(|_| std::process::abort());
            Some(terminal)
        } else if let Some(scope) = self.serial.take() {
            let suspended = self.process.finish_reserved_without_connection(scope, ());
            let (_, terminal) = suspended
                .into_response_parts()
                .unwrap_or_else(|_| std::process::abort());
            Some(terminal)
        } else {
            None
        };
        if let Some(terminal) = terminal {
            *self.state.terminal.lock().unwrap() = RequestTerminal::PostDatabase(terminal);
            self.state.completed.notify_one();
        }
    }
}

impl DatabaseRequest {
    pub(super) fn reserved_owner<T>(
        &mut self,
    ) -> Result<
        (
            Arc<saddle_runtime::request_task::reserved::ReservedTaskContext>,
            ReservedDbOwner<T>,
        ),
        ScopeDatabaseError,
    > {
        let context = self
            .state
            .reserved_context
            .clone()
            .ok_or(ScopeDatabaseError::State)?;
        if self.state.reserved_failure.lock().unwrap().is_some() {
            return Err(ScopeDatabaseError::State);
        }
        let process = self
            .process
            .clone()
            .ok_or(ScopeDatabaseError::Unavailable)?;
        let serial = self.begin_serial()?;
        Ok((
            context,
            ReservedDbOwner {
                serial: Some(serial),
                suspended: None,
                state: self.state.clone(),
                process,
            },
        ))
    }
    pub(crate) fn take_reserved_failure(&self) -> Option<ReservedRequestFailure> {
        self.state.reserved_failure.lock().unwrap().take()
    }

    pub fn constructed_transaction<'a, B>(
        &'a mut self,
        isolation: saddle_db::internal::TransactionIsolation,
        body: B,
    ) -> impl std::future::Future<
        Output = saddle_db::internal::ScopeTransactionOutcome<B::Value, B::Error>,
    > + Send + 'a
    where
        B: saddle_db::internal::ReservedTransactionBody<Cancel> + Send + 'a,
    {
        saddle_admission::lazy_framework_future(move || {
            use saddle_db::internal::{ScopeTransactionAbort as A, ScopeTransactionOutcome as O};
            if self.state.reserved_context.is_none()
                || self.state.reserved_failure.lock().unwrap().is_some()
            {
                return Err(O::Rejected(A::Technical(ScopeDatabaseError::State)));
            }
            if self.process.is_none() {
                return Err(O::Rejected(A::Technical(ScopeDatabaseError::Unavailable)));
            }
            let memory = self.framework_future_memory();
            let reporter = self.shared();
            // The original body is not constructed until its actual storage
            // has been charged. No scope is moved before that reservation.
            saddle_admission::managed_framework_future(
                &memory,
                move || self.constructed_transaction_inner(isolation, body),
                |error| reporter.record_transaction_future_error_at_source(error),
            )
            .map_err(|_| O::Rejected(A::Technical(ScopeDatabaseError::Resource)))
        })
    }

    async fn constructed_transaction_inner<B>(
        &mut self,
        isolation: saddle_db::internal::TransactionIsolation,
        body: B,
    ) -> saddle_db::internal::ScopeTransactionOutcome<B::Value, B::Error>
    where
        B: saddle_db::internal::ReservedTransactionBody<Cancel> + Send,
    {
        use saddle_db::internal::{ScopeTransactionAbort as A, ScopeTransactionOutcome as O};
        let Some(context) = self.state.reserved_context.clone() else {
            return O::Rejected(A::Technical(ScopeDatabaseError::State));
        };
        if self.state.reserved_failure.lock().unwrap().is_some() {
            return O::Rejected(A::Technical(ScopeDatabaseError::State));
        }
        let Some(process) = self.process.clone() else {
            return O::Rejected(A::Technical(ScopeDatabaseError::Unavailable));
        };
        let scope = match self.begin_serial() {
            Ok(scope) => scope,
            Err(error) => return O::Rejected(A::Technical(error)),
        };
        let mut owner = ReservedDbOwner {
            serial: Some(scope),
            suspended: None,
            state: self.state.clone(),
            process: process.clone(),
        };
        match process
            .transaction_scope_reserved(&mut owner.serial, &context, isolation, body)
            .await
        {
            Ok(completion) => {
                owner.suspended = Some(completion.suspended);
                owner.retain(completion.unretained);
                owner.publish().await.outcome
            }
            Err(error) => {
                let class = match &error {
                    ProfuseGwReservedScopeFailure::Preparation(reason) => {
                        use saddle_core::{
                            BoundedDiagnostic, BoundedDiagnosticCause, CaptureSite,
                            DiagnosticCategory, DiagnosticCode, DiagnosticStage,
                        };
                        let code = DiagnosticCode::new("service.database.preparation").unwrap();
                        let failure = context.view().source_description(
                            &PreparationDescription(reason),
                            BoundedDiagnostic::capture(
                                DiagnosticCategory::UnexpectedError,
                                CaptureSite::FirstObserved,
                                BoundedDiagnosticCause::new(DiagnosticStage::RequestDb, code),
                            ),
                            code,
                            self.diagnostic_handle.as_ref(),
                            saddle_observability::root_diagnostic::RootRequestEvent::Database,
                            Default::default(),
                        );
                        owner.retain(Some(failure));
                        ScopeDatabaseError::Resource
                    }
                    ProfuseGwReservedScopeFailure::Execution(reason) => match reason {
                        saddle_runtime::profusegw::ProfuseGwScopeFailure::Admission(error) => {
                            let failure=context.view().source_error(error,self.diagnostic_handle.as_ref(),saddle_core::DiagnosticStage::RequestDb,
                                saddle_observability::root_diagnostic::RootRequestEvent::Database,Default::default());
                            owner.retain(Some(failure));
                            ScopeDatabaseError::Resource
                        },
                        saddle_runtime::profusegw::ProfuseGwScopeFailure::Stopped(
                            saddle_runtime::profusegw::ProfuseGwScopeStop::TimedOut,
                        ) => ScopeDatabaseError::TimedOut,
                        saddle_runtime::profusegw::ProfuseGwScopeFailure::Stopped(
                            saddle_runtime::profusegw::ProfuseGwScopeStop::Cancelled,
                        ) => ScopeDatabaseError::Cancelled,
                        saddle_runtime::profusegw::ProfuseGwScopeFailure::Panicked => {
                            ScopeDatabaseError::Panicked
                        }
                        saddle_runtime::profusegw::ProfuseGwScopeFailure::AlreadySupervised
                        | saddle_runtime::profusegw::ProfuseGwScopeFailure::ScopeAlreadyEntered => {
                            ScopeDatabaseError::State
                        }
                    },
                };
                // Keep the actual error alive through physical completion; the
                // reserved source consumer will project it at the caller boundary.
                let suspended =
                    process.finish_reserved_without_connection(owner.serial.take().unwrap(), error);
                let (_original, terminal) = suspended
                    .into_response_parts()
                    .unwrap_or_else(|_| std::process::abort());
                *owner.state.terminal.lock().unwrap() = RequestTerminal::PostDatabase(terminal);
                owner.state.completed.notify_one();
                O::Rejected(A::Technical(class))
            }
        }
    }
}

#[derive(Debug)]
struct PreparationDescription<'a>(&'a saddle_runtime::request_task::reserved::ReservedContextError);
impl std::fmt::Display for PreparationDescription<'_> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "database scope preparation: {:?}", self.0)
    }
}

#[cfg(test)]
impl DatabaseRequest {
    pub(crate) async fn control_transaction_storage_outcomes(&mut self) {
        use saddle_db::internal::{ReservedTransactionBody, ReservedDatabaseTransactionSession,
            ScopeTransactionAbort as A, ScopeTransactionOutcome as O, TransactionIsolation};
        struct Body<'a> { mode: &'a u8, _pin: std::marker::PhantomPinned }
        impl ReservedTransactionBody<Cancel> for Body<'_> {
            type Value = u8;
            type Error = u8;
            fn execute(self, _: &mut ReservedDatabaseTransactionSession<'_, '_, '_, Cancel>)
                -> impl Future<Output = Result<u8, A<u8>>> + Send {
                async move {
                    match *self.mode {
                        0 => Ok(7),
                        1 => Err(A::Business(9)),
                        _ => Err(A::Technical(ScopeDatabaseError::InvalidRow)),
                    }
                }
            }
        }
        for mode in [0, 1, 2] {
            let result = self.constructed_transaction(TransactionIsolation::ReadCommitted,
                Body { mode: &mode, _pin: std::marker::PhantomPinned }).await;
            match (mode, result) {
                (0, O::Committed(7)) | (1, O::Rejected(A::Business(9)))
                | (2, O::Rejected(A::Technical(ScopeDatabaseError::InvalidRow))) => {},
                (_, other) => panic!("original transaction outcome changed: {other:?}"),
            }
            assert!(matches!(*self.state.terminal.lock().unwrap(), RequestTerminal::Serial(_)),
                "original owner returns the real serial scope before notification");
        }
    }

    pub(crate) async fn control_transaction_storage_rejection(&mut self) {
        use std::sync::atomic::{AtomicUsize, Ordering};
        use saddle_db::internal::{ReservedTransactionBody, ReservedDatabaseTransactionSession,
            ScopeTransactionAbort as A, ScopeTransactionOutcome as O, TransactionIsolation};
        struct Body<'a> { calls: &'a AtomicUsize, _pin: std::marker::PhantomPinned }
        impl ReservedTransactionBody<Cancel> for Body<'_> {
            type Value = ();
            type Error = ();
            fn execute(self, _: &mut ReservedDatabaseTransactionSession<'_, '_, '_, Cancel>)
                -> impl Future<Output = Result<(), A<()>>> + Send {
                self.calls.fetch_add(1, Ordering::SeqCst);
                async move {
                    let large = [0u8; 64 * 1024];
                    std::future::pending::<()>().await;
                    std::hint::black_box(large);
                    std::hint::black_box(self);
                    Ok(())
                }
            }
        }
        struct Held([u8; 32 * 1024]);
        impl Future for Held {
            type Output = ();
            fn poll(self: std::pin::Pin<&mut Self>, _: &mut std::task::Context<'_>)
                -> std::task::Poll<()> {
                std::hint::black_box(&self.0);
                std::task::Poll::Pending
            }
        }
        assert!(self.process.is_some());
        assert!(self.state.reserved_context.is_some());
        assert!(matches!(*self.state.terminal.lock().unwrap(), RequestTerminal::Untouched(_)));
        let calls = AtomicUsize::new(0);
        // Borrowed !Unpin input is accepted; dropping the unpolled shell neither
        // builds the execute future nor consumes SQL scope or records a failure.
        drop(self.constructed_transaction(TransactionIsolation::ReadCommitted,
            Body { calls: &calls, _pin: std::marker::PhantomPinned }));
        assert_eq!(calls.load(Ordering::SeqCst), 0);
        assert!(self.state.reserved_failure.lock().unwrap().is_none());
        assert!(matches!(*self.state.terminal.lock().unwrap(), RequestTerminal::Untouched(_)));
        let memory = self.framework_future_memory();
        let mut held = Vec::new();
        loop {
            assert!(held.len() < 1024);
            match saddle_admission::managed_framework_future(&memory,
                || Held([0; 32 * 1024]), |_| {}) {
                Ok(future) => held.push(future),
                Err(saddle_admission::AdmissionError::FrameworkReserveExceeded { .. }) => break,
                Err(error) => panic!("unexpected saturation error: {error:?}"),
            }
        }
        let outcome = self.constructed_transaction(TransactionIsolation::ReadCommitted,
            Body { calls: &calls, _pin: std::marker::PhantomPinned }).await;
        assert!(matches!(outcome, O::Rejected(A::Technical(ScopeDatabaseError::Resource))));
        assert_eq!(calls.load(Ordering::SeqCst), 0, "no SQL business execute before storage charge");
        assert!(matches!(*self.state.terminal.lock().unwrap(), RequestTerminal::Untouched(_)),
            "reserve rejection must not call begin_serial");
        let original = {
            let failure = self.state.reserved_failure.lock().unwrap();
            let failure = failure.as_ref().expect("original RequestDb rejection retained");
            assert!(failure.original_was_written());
            serde_json::to_value(failure.occurrence()).unwrap()
        };
        drop(held);
        let next = self.constructed_transaction(TransactionIsolation::ReadCommitted,
            Body { calls: &calls, _pin: std::marker::PhantomPinned }).await;
        assert!(matches!(next, O::Rejected(A::Technical(ScopeDatabaseError::State))));
        assert_eq!(serde_json::to_value(self.state.reserved_failure.lock().unwrap().as_ref().unwrap().occurrence()).unwrap(), original,
            "later poll preserves the original failure rather than capturing again");
        assert_eq!(calls.load(Ordering::SeqCst), 0);
        assert!(matches!(*self.state.terminal.lock().unwrap(), RequestTerminal::Untouched(_)));
    }
}