saddle-runtime 0.2.0-rc.3

//! Framework-internal DB finalizer phase.
//!
//! This protocol is public only to let the approved Database assembly provide
//! one concrete, monomorphized query and return-to-pool future. It is not a
//! business API and is not re-exported by the Saddle facade.

use std::{
    future::Future,
    pin::{Pin, pin},
    task::{Context, Poll},
};

use saddle_admission::{AdmissionError, DbFinalizerRoleClaim, DbRequestPermit, ManagedResponse};

/// The only value with which a DB query phase may complete.
///
/// Construction consumes the current-account permit and atomically changes
/// its Admission role to `Finalizing`. The physical return future and logical
/// role proof consequently cannot be separated.
#[doc(hidden)]
pub struct DbFinalizingOutput<T, F> {
    value: T,
    return_to_pool: F,
    claim: DbFinalizerRoleClaim,
}

impl<F> DbFinalizingOutput<ManagedResponse, F>
where
    F: Future<Output = ()> + Send + 'static,
{
    /// Compatibility constructor for the existing HTTP response path.
    ///
    /// Typed Database results must use [`DbQueryTransition::begin_finalizing`]
    /// so cancel and shutdown cannot be bypassed.
    #[doc(hidden)]
    pub fn begin(
        response: ManagedResponse,
        permit: DbRequestPermit,
        return_to_pool: F,
    ) -> Result<Self, AdmissionError> {
        let claim = permit.begin_finalizing()?;
        Ok(Self {
            value: response,
            return_to_pool,
            claim,
        })
    }
}

/// The closed cooperative inputs that may win while a DB query is pending.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
#[doc(hidden)]
pub enum DbTransitionRequest {
    Cancel,
    Shutdown,
}

/// Result of one atomic query-versus-transition poll.
#[doc(hidden)]
pub enum DbQueryPoll<T> {
    Ready(T),
    Transition(DbTransitionRequest),
}

/// Runtime-owned, linear transition capability for one admitted DB query.
///
/// The concrete cancel and shutdown futures are stored inline. The capability
/// has no public constructor and is not cloneable. A Database state machine
/// must consume it to enter `Finalizing`; dropping it aborts the process, so an
/// outer Future drop or task abort cannot silently discard a live connection.
#[must_use = "the DB transition owner must be consumed into Finalizing"]
#[doc(hidden)]
pub struct DbQueryTransition<C, S> {
    cancel: C,
    shutdown: S,
    selected: Option<DbTransitionRequest>,
    active: bool,
}

impl<C, S> DbQueryTransition<C, S>
where
    C: Future + Unpin,
    S: Future + Unpin,
{
    /// Polls the concrete query and both cooperative inputs as one state step.
    ///
    /// Query completion wins when it is already Ready in this poll. Otherwise
    /// shutdown is checked before cancel, and the selected transition remains
    /// sticky until this owner is consumed into `Finalizing`.
    #[doc(hidden)]
    pub fn poll_query<Q>(
        &mut self,
        permit: &DbRequestPermit,
        query: Pin<&mut Q>,
        context: &mut Context<'_>,
    ) -> Poll<DbQueryPoll<Q::Output>>
    where
        Q: Future,
    {
        if let Poll::Ready(output) = permit.poll_query(query, context) {
            return Poll::Ready(DbQueryPoll::Ready(output));
        }
        self.poll_cancel_or_shutdown(context)
            .map(DbQueryPoll::Transition)
    }

    fn poll_cancel_or_shutdown(&mut self, context: &mut Context<'_>) -> Poll<DbTransitionRequest> {
        if let Some(selected) = self.selected {
            return Poll::Ready(selected);
        }
        if Pin::new(&mut self.shutdown).poll(context).is_ready() {
            self.selected = Some(DbTransitionRequest::Shutdown);
            return Poll::Ready(DbTransitionRequest::Shutdown);
        }
        if Pin::new(&mut self.cancel).poll(context).is_ready() {
            self.selected = Some(DbTransitionRequest::Cancel);
            return Poll::Ready(DbTransitionRequest::Cancel);
        }
        Poll::Pending
    }

    /// Atomically consumes the query transition owner and Admission permit.
    ///
    /// `T` and the physical return future remain concrete and inline in the
    /// one request Future. No business callback is polled in Finalizing.
    #[doc(hidden)]
    pub fn begin_finalizing<T, F>(
        mut self,
        value: T,
        permit: DbRequestPermit,
        return_to_pool: F,
    ) -> Result<DbFinalizingOutput<T, F>, AdmissionError>
    where
        T: Send + 'static,
        F: Future<Output = ()> + Send + 'static,
    {
        let claim = permit.begin_finalizing()?;
        self.active = false;
        Ok(DbFinalizingOutput {
            value,
            return_to_pool,
            claim,
        })
    }
}

impl<C, S> Drop for DbQueryTransition<C, S> {
    fn drop(&mut self) {
        if self.active {
            std::process::abort();
        }
    }
}

struct FinalizationGuard {
    complete: bool,
}

impl Drop for FinalizationGuard {
    fn drop(&mut self) {
        if !self.complete {
            // A managed DB task may only disappear after physical return and
            // logical reconciliation. Abort is non-catchable even during an
            // existing unwind, so no JoinHandle/drop path can continue after
            // losing the role.
            std::process::abort();
        }
    }
}

/// Drives the one task-local `Query -> Finalizing` state machine.
///
/// `Q` handles its cooperative cancel/error/panic inputs and must produce a
/// `DbFinalizingOutput`. Runtime then owns the only poll path for the physical
/// return future and releases Admission credits only after it is ready. The
/// returned opaque future is concrete and inline; this function performs no
/// boxing, spawn, channel or queue operation.
#[doc(hidden)]
pub async fn drive_db_finalizer<T, Q, F>(query: Q) -> Result<T, AdmissionError>
where
    T: Send + 'static,
    Q: Future<Output = Result<DbFinalizingOutput<T, F>, AdmissionError>> + Send + 'static,
    F: Future<Output = ()> + Send + 'static,
{
    let mut guard = FinalizationGuard { complete: false };
    let output = query.await?;
    let mut return_to_pool = pin!(output.return_to_pool);
    std::future::poll_fn(|context| {
        output
            .claim
            .poll_connection_return(return_to_pool.as_mut(), context)
    })
    .await;
    output.claim.complete_after_connection_return()?;
    guard.complete = true;
    Ok(output.value)
}

/// Creates the only transition owner and drives one typed DB result to
/// physical return in the same request task.
///
/// `factory` is monomorphized assembly, not a stored callback. Its concrete
/// Future, cooperative inputs, result and return Future all contribute to the
/// enclosing task layout reviewed by Admission.
#[doc(hidden)]
pub async fn drive_db_finalizer_with_transition<T, C, S, B, Q, F>(
    cancel: C,
    shutdown: S,
    factory: B,
) -> Result<T, AdmissionError>
where
    T: Send + 'static,
    C: Future + Unpin + Send + 'static,
    S: Future + Unpin + Send + 'static,
    B: FnOnce(DbQueryTransition<C, S>) -> Q,
    Q: Future<Output = Result<DbFinalizingOutput<T, F>, AdmissionError>> + Send + 'static,
    F: Future<Output = ()> + Send + 'static,
{
    let transition = DbQueryTransition {
        cancel,
        shutdown,
        selected: None,
        active: true,
    };
    drive_db_finalizer(factory(transition)).await
}

/// Polls one concrete query future only while the current request owns its DB
/// Query role. This is the enumerated DB framework phase; it does not grant
/// privilege to the surrounding generic business future.
#[doc(hidden)]
pub async fn drive_db_query<Q>(permit: &DbRequestPermit, query: Q) -> Q::Output
where
    Q: Future,
{
    let mut query = pin!(query);
    std::future::poll_fn(|context| permit.poll_query(query.as_mut(), context)).await
}

#[cfg(test)]
mod tests {
    use std::{
        env,
        future::{Future, Pending, Ready},
        os::unix::process::ExitStatusExt,
        pin::pin,
        process::{Command, Stdio},
        task::{Context, Poll, Waker},
        time::{Duration, Instant},
    };

    use super::*;

    async fn pending_query()
    -> Result<DbFinalizingOutput<ManagedResponse, Ready<()>>, AdmissionError> {
        std::future::pending().await
    }

    async fn pending_transition_query(
        _transition: DbQueryTransition<Pending<()>, Pending<()>>,
    ) -> Result<DbFinalizingOutput<ManagedResponse, Ready<()>>, AdmissionError> {
        std::future::pending().await
    }

    #[test]
    fn lost_db_role_is_finite_fail_closed_in_subprocess() {
        const CHILD: &str = "db_finalizer::tests::lost_db_role_child";
        for mode in [
            "direct",
            "during-unwind",
            "transition-direct",
            "transition-during-unwind",
        ] {
            let mut child = Command::new(env::current_exe().unwrap())
                .args(["--exact", CHILD, "--nocapture"])
                .env("SADDLE_DB_FINALIZER_LOST_ROLE_CHILD", mode)
                .stdin(Stdio::null())
                .stdout(Stdio::null())
                .stderr(Stdio::null())
                .spawn()
                .unwrap();
            let started = Instant::now();
            let status = loop {
                if let Some(status) = child.try_wait().unwrap() {
                    break status;
                }
                if started.elapsed() >= Duration::from_secs(5) {
                    child.kill().unwrap();
                    child.wait().unwrap();
                    panic!("lost DB role ({mode}) exceeded finite deadline");
                }
                std::thread::sleep(Duration::from_millis(10));
            };
            assert_eq!(status.signal(), Some(6));
        }
    }

    #[test]
    fn lost_db_role_child() {
        let Some(mode) = env::var_os("SADDLE_DB_FINALIZER_LOST_ROLE_CHILD") else {
            return;
        };
        let transition = mode.to_string_lossy().starts_with("transition-");
        let mut future = pin!(async {
            if transition {
                drive_db_finalizer_with_transition(
                    std::future::pending(),
                    std::future::pending(),
                    pending_transition_query,
                )
                .await
            } else {
                drive_db_finalizer(pending_query()).await
            }
        });
        let waker = Waker::noop();
        let mut context = Context::from_waker(waker);
        assert!(matches!(future.as_mut().poll(&mut context), Poll::Pending));
        if mode.to_string_lossy().ends_with("during-unwind") {
            let _future = future;
            panic!("existing unwind must not discard the DB finalizer role");
        }
        // Returning drops the hidden stack-pinned future without an unwind.
        // Its armed guard must abort before control reaches the caller.
    }
}