saddle-db 0.2.0-rc.19

//! Production C6 contract for one statically generated optional query.
//!
//! Service hands one opaque admitted execution to this Database-owned,
//! same-task sqlx/Runtime finalizer state machine. Dynamic SQL, pool waiting,
//! Skill generation, write, transaction and multi-row APIs stay outside this
//! closed path.

use std::{
    any::TypeId, future::Future, marker::PhantomData, mem, panic::AssertUnwindSafe, pin::pin,
    task::Poll,
};

use saddle_admission::DbRequestPermit;
use saddle_runtime::db_finalizer::{
    DbQueryPoll, DbQueryTransition, DbTransitionRequest, drive_db_finalizer_with_transition,
};
use sqlx::{
    MySql, Row,
    mysql::{MySqlArguments, MySqlRow},
    query::Query,
};

use crate::Database;
use sealed::{ParameterShape as _, RowShape as _};

const MAX_STATIC_OPERATION_BYTES: usize = 128;
const MAX_STATIC_SQL_BYTES: usize = 65_536;

pub(crate) mod sealed {
    use sqlx::{MySql, mysql::MySqlArguments, query::Query};

    use super::MySqlRow;

    pub trait Field: Sized {
        fn bind<'q>(
            &'q self,
            query: Query<'q, MySql, MySqlArguments>,
        ) -> Query<'q, MySql, MySqlArguments>;

        fn decode(row: &MySqlRow, column: &mut usize) -> std::result::Result<Self, ()>;
    }

    pub trait ParameterShape {
        fn bind<'q>(
            &'q self,
            query: Query<'q, MySql, MySqlArguments>,
        ) -> Query<'q, MySql, MySqlArguments>;
    }

    pub trait RowShape: Sized {
        fn decode(row: &MySqlRow) -> std::result::Result<Self, ()>;
    }
}

/// Saddle-owned scalar fields accepted by the generated DB ABI.
pub trait ManagedDbField: sealed::Field + Send + Sync + 'static {}

#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
#[repr(transparent)]
pub struct DbU64(pub u64);

impl sealed::Field for DbU64 {
    fn bind<'q>(
        &'q self,
        query: Query<'q, MySql, MySqlArguments>,
    ) -> Query<'q, MySql, MySqlArguments> {
        query.bind(self.0)
    }

    fn decode(row: &MySqlRow, column: &mut usize) -> std::result::Result<Self, ()> {
        let value = row.try_get(*column).map_err(|_| ())?;
        *column += 1;
        Ok(Self(value))
    }
}
impl ManagedDbField for DbU64 {}

#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
#[repr(transparent)]
pub struct DbI64(pub i64);

impl sealed::Field for DbI64 {
    fn bind<'q>(
        &'q self,
        query: Query<'q, MySql, MySqlArguments>,
    ) -> Query<'q, MySql, MySqlArguments> {
        query.bind(self.0)
    }

    fn decode(row: &MySqlRow, column: &mut usize) -> std::result::Result<Self, ()> {
        let value = row.try_get(*column).map_err(|_| ())?;
        *column += 1;
        Ok(Self(value))
    }
}
impl ManagedDbField for DbI64 {}

#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
#[repr(transparent)]
pub struct DbBool(pub bool);

impl sealed::Field for DbBool {
    fn bind<'q>(
        &'q self,
        query: Query<'q, MySql, MySqlArguments>,
    ) -> Query<'q, MySql, MySqlArguments> {
        query.bind(self.0)
    }

    fn decode(row: &MySqlRow, column: &mut usize) -> std::result::Result<Self, ()> {
        let value = row.try_get(*column).map_err(|_| ())?;
        *column += 1;
        Ok(Self(value))
    }
}
impl ManagedDbField for DbBool {}

/// Inline bytes with a compile-time capacity. Construction never allocates.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct FixedDbBytes<const N: usize> {
    bytes: [u8; N],
    length: usize,
}

impl<const N: usize> FixedDbBytes<N> {
    pub const fn empty() -> Self {
        Self {
            bytes: [0; N],
            length: 0,
        }
    }

    pub fn try_from_slice(value: &[u8]) -> Result<Self, QueryOptionalContractError> {
        if value.len() > N {
            return Err(QueryOptionalContractError::ValueTooLarge);
        }
        let mut output = Self::empty();
        output.bytes[..value.len()].copy_from_slice(value);
        output.length = value.len();
        Ok(output)
    }

    pub const fn capacity(&self) -> usize {
        N
    }

    pub const fn len(&self) -> usize {
        self.length
    }

    pub const fn is_empty(&self) -> bool {
        self.length == 0
    }

    pub fn as_slice(&self) -> &[u8] {
        &self.bytes[..self.length]
    }
}

impl<const N: usize> sealed::Field for FixedDbBytes<N> {
    fn bind<'q>(
        &'q self,
        query: Query<'q, MySql, MySqlArguments>,
    ) -> Query<'q, MySql, MySqlArguments> {
        query.bind(self.as_slice())
    }

    fn decode(row: &MySqlRow, column: &mut usize) -> std::result::Result<Self, ()> {
        let value: &[u8] = row.try_get(*column).map_err(|_| ())?;
        *column += 1;
        Self::try_from_slice(value).map_err(|_| ())
    }
}
impl<const N: usize> ManagedDbField for FixedDbBytes<N> {}

/// Fixed-layout product used by generated parameter and row shapes.
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
pub struct DbPair<A: ManagedDbField, B: ManagedDbField>(pub A, pub B);

impl<A: ManagedDbField, B: ManagedDbField> sealed::Field for DbPair<A, B> {
    fn bind<'q>(
        &'q self,
        query: Query<'q, MySql, MySqlArguments>,
    ) -> Query<'q, MySql, MySqlArguments> {
        self.1.bind(self.0.bind(query))
    }

    fn decode(row: &MySqlRow, column: &mut usize) -> std::result::Result<Self, ()> {
        Ok(Self(A::decode(row, column)?, B::decode(row, column)?))
    }
}
impl<A: ManagedDbField, B: ManagedDbField> ManagedDbField for DbPair<A, B> {}

/// A concrete generated parameter tuple. It contains no `String`, `Vec`,
/// dynamic SQL, driver value, or allocator handle.
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
#[repr(transparent)]
pub struct ManagedQueryParameters<T: ManagedDbField>(pub T);

impl<T: ManagedDbField> sealed::ParameterShape for ManagedQueryParameters<T> {
    fn bind<'q>(
        &'q self,
        query: Query<'q, MySql, MySqlArguments>,
    ) -> Query<'q, MySql, MySqlArguments> {
        self.0.bind(query)
    }
}

pub trait QueryOptionalParameterShape:
    sealed::ParameterShape + Send + Sync + Sized + 'static
{
}

impl<T: ManagedDbField> QueryOptionalParameterShape for ManagedQueryParameters<T> {}

/// One concrete generated row. Column order and field types are in its Rust
/// type rather than discovered through a dynamic row accessor.
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
#[repr(transparent)]
pub struct ManagedQueryRow<T: ManagedDbField>(pub T);

impl<T: ManagedDbField> sealed::RowShape for ManagedQueryRow<T> {
    fn decode(row: &MySqlRow) -> std::result::Result<Self, ()> {
        let mut column = 0;
        let value = T::decode(row, &mut column)?;
        if column != row.len() {
            return Err(());
        }
        Ok(Self(value))
    }
}

pub trait QueryOptionalRowShape: sealed::RowShape + Send + Sync + Sized + 'static {}

impl<T: ManagedDbField> QueryOptionalRowShape for ManagedQueryRow<T> {}

/// The only C6.1 result cardinality: zero or one concrete managed row.
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
#[repr(transparent)]
pub struct ManagedOptionalRow<R: QueryOptionalRowShape>(pub Option<R>);

impl<R: QueryOptionalRowShape> ManagedOptionalRow<R> {
    pub const fn none() -> Self {
        Self(None)
    }

    pub const fn some(row: R) -> Self {
        Self(Some(row))
    }

    pub const fn as_ref(&self) -> Option<&R> {
        self.0.as_ref()
    }

    pub fn into_option(self) -> Option<R> {
        self.0
    }
}

/// Implemented only in generated artifact code for a statically known query.
///
/// `SQL` and `OPERATION` are startup/codegen facts. They are never supplied to
/// an admitted invocation.
pub trait StaticQueryOptionalOperation: Send + 'static {
    type Parameters: QueryOptionalParameterShape;
    type Row: QueryOptionalRowShape;

    const OPERATION: &'static str;
    const SQL: &'static str;
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum QueryOptionalContractError {
    InvalidOperation,
    InvalidSql,
    EmptyShape,
    SizeOverflow,
    ValueTooLarge,
}

/// Fixed-width production failures for the closed `query_optional` executor.
///
/// The 0.1 [`crate::SaddleError`] facade owns a `String` and therefore cannot
/// cross an admitted 0.2 query path. Generated Service code maps this closed
/// classification only after Database Finalizing has completed.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum QueryOptionalExecutionError {
    ConnectionUnavailable,
    QueryFailed,
    Cancelled,
    Shutdown,
    InvalidRow,
    FinalizerFailed,
}

impl QueryOptionalExecutionError {
    pub const fn code(self) -> &'static str {
        match self {
            Self::ConnectionUnavailable => "db.connection_unavailable",
            Self::QueryFailed => "db.query_failed",
            Self::Cancelled => "db.query_cancelled",
            Self::Shutdown => "db.query_shutdown",
            Self::InvalidRow => "db.invalid_column",
            Self::FinalizerFailed => "db.finalizer_failed",
        }
    }
}

/// Machine-readable physical finalization facts for generated profiles.
///
/// Cancel, shutdown and panic never await server I/O: the checked-out pool
/// connection is synchronously converted back into a pool size permit and its
/// poisoned raw socket is dropped in the same Finalizing poll. Normal query
/// completion may still use sqlx's explicit return-and-ping path and therefore
/// requires Runtime's independent termination bound.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct QueryOptionalFinalizationProof {
    cancel_external_io_awaits: u8,
    shutdown_external_io_awaits: u8,
    panic_external_io_awaits: u8,
    releases_pool_size_before_permit: bool,
    normal_return_requires_termination_bound: bool,
}

impl QueryOptionalFinalizationProof {
    const PRODUCTION: Self = Self {
        cancel_external_io_awaits: 0,
        shutdown_external_io_awaits: 0,
        panic_external_io_awaits: 0,
        releases_pool_size_before_permit: true,
        normal_return_requires_termination_bound: true,
    };

    pub const fn cancel_external_io_awaits(self) -> u8 {
        self.cancel_external_io_awaits
    }

    pub const fn shutdown_external_io_awaits(self) -> u8 {
        self.shutdown_external_io_awaits
    }

    pub const fn panic_external_io_awaits(self) -> u8 {
        self.panic_external_io_awaits
    }

    pub const fn releases_pool_size_before_permit(self) -> bool {
        self.releases_pool_size_before_permit
    }

    pub const fn normal_return_requires_termination_bound(self) -> bool {
        self.normal_return_requires_termination_bound
    }
}

/// Returns the sealed proof consumed by generated Runtime profiles.
#[doc(hidden)]
pub const fn query_optional_finalization_proof() -> QueryOptionalFinalizationProof {
    QueryOptionalFinalizationProof::PRODUCTION
}

/// Domain demand contributed by one operation to a generated route plan.
///
/// The route generator combines dependencies with `max`, not by counting call
/// sites. Admission will later reserve this demand atomically; this slice does
/// not create a permit or queue.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct QueryOptionalCreditDemand {
    connections: u32,
    operations: u32,
}

impl QueryOptionalCreditDemand {
    const ONE: Self = Self {
        connections: 1,
        operations: 1,
    };

    pub const fn connections(self) -> u32 {
        self.connections
    }

    pub const fn operations(self) -> u32 {
        self.operations
    }

    pub const fn merge_route_max(self, other: Self) -> Self {
        Self {
            connections: if self.connections > other.connections {
                self.connections
            } else {
                other.connections
            },
            operations: if self.operations > other.operations {
                self.operations
            } else {
                other.operations
            },
        }
    }
}

/// Type-derived contribution to the immutable route resource plan.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct QueryOptionalLayout {
    operation: TypeId,
    parameter_bytes: usize,
    optional_row_bytes: usize,
    credits: QueryOptionalCreditDemand,
}

impl QueryOptionalLayout {
    pub const fn parameter_bytes(self) -> usize {
        self.parameter_bytes
    }

    pub const fn optional_row_bytes(self) -> usize {
        self.optional_row_bytes
    }

    pub const fn credits(self) -> QueryOptionalCreditDemand {
        self.credits
    }

    pub fn belongs_to<O: StaticQueryOptionalOperation>(self) -> bool {
        self.operation == TypeId::of::<O>()
    }
}

/// Statically typed proof for one generated operation.
///
/// There is no numeric, string, SQL, or identity constructor. Generated code
/// obtains it by binding the operation marker and its concrete ABI types.
#[derive(Clone, Copy, Debug)]
pub struct QueryOptionalOperationProof<O: StaticQueryOptionalOperation> {
    layout: QueryOptionalLayout,
    _operation: PhantomData<fn() -> O>,
}

impl<O: StaticQueryOptionalOperation> QueryOptionalOperationProof<O> {
    pub fn bind() -> Result<Self, QueryOptionalContractError> {
        validate_operation(O::OPERATION)?;
        validate_sql(O::SQL)?;
        let parameter_bytes = mem::size_of::<O::Parameters>();
        let optional_row_bytes = mem::size_of::<ManagedOptionalRow<O::Row>>();
        if parameter_bytes == 0 || optional_row_bytes == 0 {
            return Err(QueryOptionalContractError::EmptyShape);
        }
        parameter_bytes
            .checked_add(optional_row_bytes)
            .ok_or(QueryOptionalContractError::SizeOverflow)?;
        Ok(Self {
            layout: QueryOptionalLayout {
                operation: TypeId::of::<O>(),
                parameter_bytes,
                optional_row_bytes,
                credits: QueryOptionalCreditDemand::ONE,
            },
            _operation: PhantomData,
        })
    }

    pub const fn layout(&self) -> QueryOptionalLayout {
        self.layout
    }

    /// Binds concrete parameters to the same operation type as the proof.
    pub fn invocation(self, parameters: O::Parameters) -> QueryOptionalInvocation<O> {
        QueryOptionalInvocation {
            layout: self.layout,
            parameters,
            _operation: PhantomData,
        }
    }
}

/// Pre-execution value that cannot carry dynamic SQL or a mismatched ABI.
pub struct QueryOptionalInvocation<O: StaticQueryOptionalOperation> {
    layout: QueryOptionalLayout,
    parameters: O::Parameters,
    _operation: PhantomData<fn() -> O>,
}

impl<O: StaticQueryOptionalOperation> QueryOptionalInvocation<O> {
    pub const fn layout(&self) -> QueryOptionalLayout {
        self.layout
    }

    pub fn parameters(&self) -> &O::Parameters {
        &self.parameters
    }

    pub fn into_parameters(self) -> O::Parameters {
        self.parameters
    }
}

/// Linear, pre-execution ownership accepted by the future Database executor.
///
/// It exposes neither the admitted permit nor query inputs. Service's
/// generated compiled dispatcher is the only production constructor, and
/// Database consumes this whole value into its fixed query/finalizer state
/// machine.
pub struct QueryOptionalExecution<O: StaticQueryOptionalOperation> {
    permit: DbRequestPermit,
    invocation: QueryOptionalInvocation<O>,
}

impl<O: StaticQueryOptionalOperation> QueryOptionalExecution<O> {
    /// Cross-crate internal constructor used only by Service's sealed,
    /// consuming handoff. It never exposes a raw permit.
    #[doc(hidden)]
    pub fn from_compiled_handoff(
        permit: DbRequestPermit,
        invocation: QueryOptionalInvocation<O>,
    ) -> Self {
        Self { permit, invocation }
    }

    fn into_parts(self) -> (DbRequestPermit, QueryOptionalInvocation<O>) {
        (self.permit, self.invocation)
    }
}

impl Database {
    /// Executes one generated, statically sealed zero-or-one-row query.
    ///
    /// The admitted permit and physical connection remain in this request
    /// Future until Runtime observes the owning return-to-pool Future finish.
    #[doc(hidden)]
    pub async fn execute_query_optional<O, C, S>(
        &self,
        execution: QueryOptionalExecution<O>,
        cancel: C,
        shutdown: S,
    ) -> Result<ManagedOptionalRow<O::Row>, QueryOptionalExecutionError>
    where
        O: StaticQueryOptionalOperation,
        C: Future + Unpin + Send + 'static,
        S: Future + Unpin + Send + 'static,
    {
        let pool = self.pool.clone();
        drive_db_finalizer_with_transition(cancel, shutdown, move |transition| {
            execute_with_transition(pool, execution, transition)
        })
        .await
        .map_err(|_| QueryOptionalExecutionError::FinalizerFailed)
        .and_then(|result| result)
    }
}

async fn execute_with_transition<O, C, S>(
    pool: sqlx::MySqlPool,
    execution: QueryOptionalExecution<O>,
    mut transition: DbQueryTransition<C, S>,
) -> std::result::Result<
    saddle_runtime::db_finalizer::DbFinalizingOutput<
        Result<ManagedOptionalRow<O::Row>, QueryOptionalExecutionError>,
        impl Future<Output = ()> + Send + 'static,
    >,
    saddle_admission::AdmissionError,
>
where
    O: StaticQueryOptionalOperation,
    C: Future + Unpin + Send + 'static,
    S: Future + Unpin + Send + 'static,
{
    let (permit, invocation) = execution.into_parts();
    let mut connection = pool.try_acquire();
    let (value, physical) = if let Some(connection) = connection.as_mut() {
        let query = async {
            let result = invocation
                .parameters()
                .bind(sqlx::query::<MySql>(O::SQL))
                .fetch_optional(&mut **connection)
                .await;
            result
                .map_err(map_execution_error)
                .and_then(|row| decode_optional::<O>(row.as_ref()))
        };
        let mut query = pin!(query);
        let outcome = std::future::poll_fn(|context| {
            match std::panic::catch_unwind(AssertUnwindSafe(|| {
                transition.poll_query(&permit, query.as_mut(), context)
            })) {
                Ok(Poll::Ready(output)) => Poll::Ready(Ok(output)),
                Ok(Poll::Pending) => Poll::Pending,
                Err(panic) => Poll::Ready(Err(panic)),
            }
        })
        .await;
        match outcome {
            Ok(DbQueryPoll::Ready(result)) => {
                let physical = if result
                    .as_ref()
                    .is_err_and(|error| error.requires_poison_discard())
                {
                    PhysicalFinalization::PoisonDiscard
                } else {
                    PhysicalFinalization::ReturnToPool
                };
                (result, physical)
            }
            Ok(DbQueryPoll::Transition(DbTransitionRequest::Cancel)) => (
                Err(QueryOptionalExecutionError::Cancelled),
                PhysicalFinalization::PoisonDiscard,
            ),
            Ok(DbQueryPoll::Transition(DbTransitionRequest::Shutdown)) => (
                Err(QueryOptionalExecutionError::Shutdown),
                PhysicalFinalization::PoisonDiscard,
            ),
            Err(_) => (
                Err(QueryOptionalExecutionError::QueryFailed),
                PhysicalFinalization::PoisonDiscard,
            ),
        }
    } else {
        (
            Err(QueryOptionalExecutionError::ConnectionUnavailable),
            PhysicalFinalization::ReturnToPool,
        )
    };
    let return_to_pool = async move {
        if let Some(mut connection) = connection.take() {
            match physical {
                PhysicalFinalization::ReturnToPool => connection.return_to_pool().await,
                PhysicalFinalization::PoisonDiscard => {
                    // sqlx calls this an attached pool connection "detach".
                    // It is a synchronous ownership conversion, not a detached
                    // task: dropping the raw MySQL connection closes its socket
                    // and dropping sqlx's guard decrements pool size here.
                    drop(connection.detach());
                }
            }
        }
    };
    transition.begin_finalizing(value, permit, return_to_pool)
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum PhysicalFinalization {
    ReturnToPool,
    PoisonDiscard,
}

fn decode_optional<O: StaticQueryOptionalOperation>(
    row: Option<&MySqlRow>,
) -> Result<ManagedOptionalRow<O::Row>, QueryOptionalExecutionError> {
    row.map(|row| O::Row::decode(row).map(ManagedOptionalRow::some))
        .transpose()
        .map(|row| row.unwrap_or_else(ManagedOptionalRow::none))
        .map_err(|_| QueryOptionalExecutionError::InvalidRow)
}

fn map_execution_error(error: sqlx::Error) -> QueryOptionalExecutionError {
    match error {
        sqlx::Error::PoolTimedOut | sqlx::Error::PoolClosed | sqlx::Error::Io(_) => {
            QueryOptionalExecutionError::ConnectionUnavailable
        }
        _ => QueryOptionalExecutionError::QueryFailed,
    }
}

impl QueryOptionalExecutionError {
    fn requires_poison_discard(self) -> bool {
        matches!(
            self,
            Self::ConnectionUnavailable | Self::QueryFailed | Self::Cancelled | Self::Shutdown
        )
    }
}

fn validate_operation(operation: &str) -> Result<(), QueryOptionalContractError> {
    if operation.is_empty()
        || operation.len() > MAX_STATIC_OPERATION_BYTES
        || !operation
            .bytes()
            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b'-'))
    {
        return Err(QueryOptionalContractError::InvalidOperation);
    }
    Ok(())
}

fn validate_sql(sql: &str) -> Result<(), QueryOptionalContractError> {
    if sql.trim().is_empty() || sql.len() > MAX_STATIC_SQL_BYTES {
        return Err(QueryOptionalContractError::InvalidSql);
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use std::{
        env, io,
        pin::Pin,
        process::Command,
        task::{Context, Poll},
        time::{Duration, Instant},
    };

    use saddle_admission::{
        AdmissionError, DbCreditProfile, DbRouteCreditDemand, DbRouteResources, EntryIoAuditPlan,
        EntryReadPoll, ManagedBytes, ManagedResponse, OfficialTokioEntryIoAttemptOutcome,
        OfficialTokioRegistrationProfile, ProcessLedger, RequestMemory, ResourceConfig,
        ResponseWritePoll,
    };
    use saddle_core::ComponentLifecycle;
    use saddle_observability::{Observer, ObserverConfig};
    use sqlx::{Connection, mysql::MySqlConnection};

    use super::*;

    struct FindUser;

    impl StaticQueryOptionalOperation for FindUser {
        type Parameters = ManagedQueryParameters<DbU64>;
        type Row = ManagedQueryRow<DbPair<DbU64, DbBool>>;

        const OPERATION: &'static str = "users.find";
        const SQL: &'static str = "SELECT id, active FROM users WHERE id = ?";
    }

    struct FindOrder;

    impl StaticQueryOptionalOperation for FindOrder {
        type Parameters = ManagedQueryParameters<DbU64>;
        type Row = ManagedQueryRow<DbU64>;

        const OPERATION: &'static str = "orders.find";
        const SQL: &'static str = "SELECT id FROM orders WHERE id = ?";
    }

    #[test]
    fn proof_is_bound_to_static_operation_and_concrete_shapes() {
        let proof = QueryOptionalOperationProof::<FindUser>::bind().unwrap();
        assert!(proof.layout().belongs_to::<FindUser>());
        assert!(!proof.layout().belongs_to::<FindOrder>());
        assert_eq!(proof.layout().parameter_bytes(), mem::size_of::<DbU64>());
        assert_eq!(
            proof.layout().optional_row_bytes(),
            mem::size_of::<ManagedOptionalRow<<FindUser as StaticQueryOptionalOperation>::Row>>()
        );
        assert_eq!(proof.layout().credits().connections(), 1);
        assert_eq!(proof.layout().credits().operations(), 1);

        let invocation = proof.invocation(ManagedQueryParameters(DbU64(7)));
        assert_eq!(invocation.parameters().0, DbU64(7));
    }

    #[test]
    fn route_credit_composition_uses_max_not_call_count() {
        let user = QueryOptionalOperationProof::<FindUser>::bind()
            .unwrap()
            .layout()
            .credits();
        let order = QueryOptionalOperationProof::<FindOrder>::bind()
            .unwrap()
            .layout()
            .credits();
        let route = user.merge_route_max(order).merge_route_max(user);
        assert_eq!(route.connections(), 1);
        assert_eq!(route.operations(), 1);
    }

    #[test]
    fn fixed_bytes_reject_growth_without_allocation() {
        let value = FixedDbBytes::<4>::try_from_slice(b"1234").unwrap();
        assert_eq!(value.as_slice(), b"1234");
        assert_eq!(
            FixedDbBytes::<4>::try_from_slice(b"12345"),
            Err(QueryOptionalContractError::ValueTooLarge)
        );
    }

    #[derive(Debug)]
    struct EmptySql;

    impl StaticQueryOptionalOperation for EmptySql {
        type Parameters = ManagedQueryParameters<DbU64>;
        type Row = ManagedQueryRow<DbU64>;

        const OPERATION: &'static str = "users.find";
        const SQL: &'static str = "";
    }

    #[test]
    fn invalid_generated_facts_fail_at_startup_binding() {
        assert_eq!(
            QueryOptionalOperationProof::<EmptySql>::bind().unwrap_err(),
            QueryOptionalContractError::InvalidSql
        );
    }

    struct ProductionSuccess;
    struct ProductionNone;
    struct ProductionError;
    struct ProductionSlow;
    struct ProductionPanic;
    struct ProductionLockWait;

    macro_rules! u64_operation {
        ($operation:ty, $name:literal, $sql:literal) => {
            impl StaticQueryOptionalOperation for $operation {
                type Parameters = ManagedQueryParameters<DbU64>;
                type Row = ManagedQueryRow<DbU64>;

                const OPERATION: &'static str = $name;
                const SQL: &'static str = $sql;
            }
        };
    }

    u64_operation!(
        ProductionSuccess,
        "production.success",
        "SELECT CAST(? AS UNSIGNED)"
    );
    u64_operation!(
        ProductionNone,
        "production.none",
        "SELECT CAST(? AS UNSIGNED) WHERE FALSE"
    );
    u64_operation!(
        ProductionError,
        "production.error",
        "SELECT id FROM saddle_c6_missing_table WHERE id = ?"
    );
    u64_operation!(
        ProductionSlow,
        "production.slow",
        "SELECT CAST(? AS UNSIGNED) FROM (SELECT SLEEP(0.05)) AS delayed"
    );
    u64_operation!(
        ProductionLockWait,
        "production.lock_wait",
        "SELECT id FROM saddle_c6_lock_wait WHERE id = ? FOR UPDATE"
    );

    struct PanicRow;

    impl sealed::RowShape for PanicRow {
        fn decode(_: &MySqlRow) -> std::result::Result<Self, ()> {
            panic!("generated row decode panic")
        }
    }

    impl QueryOptionalRowShape for PanicRow {}

    impl StaticQueryOptionalOperation for ProductionPanic {
        type Parameters = ManagedQueryParameters<DbU64>;
        type Row = PanicRow;

        const OPERATION: &'static str = "production.panic";
        const SQL: &'static str = "SELECT CAST(? AS UNSIGNED)";
    }

    struct EntryConnection;
    struct ReadyRead;
    struct ReadyWrite;

    impl EntryReadPoll<EntryConnection> for ReadyRead {
        fn poll_read(
            &mut self,
            _: &mut EntryConnection,
            memory: &RequestMemory,
            _: &mut Context<'_>,
        ) -> Poll<std::result::Result<ManagedBytes, AdmissionError>> {
            Poll::Ready(memory.try_bytes(&[]))
        }
    }

    impl ResponseWritePoll<EntryConnection> for ReadyWrite {
        fn poll_write(
            &mut self,
            _: &mut EntryConnection,
            response: &ManagedResponse,
            _: &mut Context<'_>,
        ) -> Poll<std::result::Result<(), AdmissionError>> {
            assert!(response.as_slice().is_empty());
            Poll::Ready(Ok(()))
        }
    }

    struct FixedSignal {
        polls_before_ready: Option<u8>,
    }

    impl FixedSignal {
        const fn pending() -> Self {
            Self {
                polls_before_ready: None,
            }
        }

        const fn after_pending_poll() -> Self {
            Self {
                polls_before_ready: Some(1),
            }
        }
    }

    impl Future for FixedSignal {
        type Output = ();

        fn poll(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<()> {
            match self.polls_before_ready {
                None => Poll::Pending,
                Some(0) => Poll::Ready(()),
                Some(remaining) => {
                    self.polls_before_ready = Some(remaining - 1);
                    context.waker().wake_by_ref();
                    Poll::Pending
                }
            }
        }
    }

    fn production_config(
        registration: OfficialTokioRegistrationProfile,
        task_reserve: usize,
    ) -> ResourceConfig {
        let process_state_reserve = ProcessLedger::minimum_process_state_reserve_with_waiters(1, 1)
            .unwrap()
            + ProcessLedger::official_tokio_state_reserve(registration).unwrap();
        ResourceConfig {
            managed_capacity: 4_096,
            entry_reserve: 64,
            framework_reserve: 1024 * 1024,
            task_reserve,
            process_state_reserve,
            system_estimate: 1024 * 1024,
            safety_margin: 1024 * 1024,
            process_limit: 4_096
                + 64
                + 1024 * 1024
                + task_reserve
                + process_state_reserve
                + 2 * 1024 * 1024,
            max_active_requests: 1,
        }
    }

    fn entry_plan() -> EntryIoAuditPlan {
        EntryIoAuditPlan::locked_linux_x86_64_tokio_1_53_1(64, 64, &[]).unwrap()
    }

    #[derive(Clone, Copy)]
    enum TestServerAction {
        Responsive,
        Stop(u32),
        Terminate(u32),
    }

    async fn run_production_mode<O, C, S>(
        url: &str,
        invocation: QueryOptionalInvocation<O>,
        cancel: C,
        shutdown: S,
        expected_code: Option<&'static str>,
        expected_row: bool,
    ) where
        O: StaticQueryOptionalOperation,
        O::Parameters: Unpin,
        C: Future<Output = ()> + Unpin + Send + 'static,
        S: Future<Output = ()> + Unpin + Send + 'static,
    {
        run_production_mode_with_server(
            url,
            invocation,
            cancel,
            shutdown,
            expected_code,
            expected_row,
            TestServerAction::Responsive,
        )
        .await;
    }

    #[allow(
        clippy::too_many_arguments,
        reason = "the fixture keeps every physical finalization outcome explicit"
    )]
    async fn run_production_mode_with_server<O, C, S>(
        url: &str,
        invocation: QueryOptionalInvocation<O>,
        cancel: C,
        shutdown: S,
        expected_code: Option<&'static str>,
        expected_row: bool,
        server_action: TestServerAction,
    ) where
        O: StaticQueryOptionalOperation,
        O::Parameters: Unpin,
        C: Future<Output = ()> + Unpin + Send + 'static,
        S: Future<Output = ()> + Unpin + Send + 'static,
    {
        let observer = Observer::with_writer(ObserverConfig::default(), io::sink()).unwrap();
        let database = Database::connect(
            crate::DatabaseConfig::new(url).max_connections(1),
            observer.clone(),
        )
        .await
        .unwrap();
        assert_eq!((database.pool.size(), database.pool.num_idle()), (1, 1));

        let registration = OfficialTokioRegistrationProfile {
            listener: 1,
            transport_connections: 1,
            runtime_fixed: 1,
        };
        let task_reserve = 1024 * 1024;
        let ledger =
            ProcessLedger::new_with_waiters(production_config(registration, task_reserve), 1)
                .unwrap();
        let runtime_domain = ledger.prepare_official_tokio_domain(registration).unwrap();
        let allocation = ledger
            .prepare_process_allocation_profile(usize::MAX)
            .unwrap();
        let db_domain = ledger
            .prepare_db_domain(DbCreditProfile {
                connections: 1,
                operations: 1,
            })
            .unwrap();
        let demand = DbRouteCreditDemand::new(1, 1).unwrap();
        let database_for_task = database.clone();

        let outcome = ledger.attempt_official_tokio_entry_io(
            &runtime_domain,
            DbRouteResources::required(&db_domain, demand),
            4_096,
            task_reserve,
            entry_plan(),
            entry_plan(),
            |_| (EntryConnection, ReadyRead, ReadyWrite),
            move |_, permit, memory| {
                let execution = QueryOptionalExecution::from_compiled_handoff(
                    permit.expect("DB route owns one permit"),
                    invocation,
                );
                let response = memory.try_response(&[]).unwrap();
                async move {
                    let result = database_for_task
                        .execute_query_optional(execution, cancel, shutdown)
                        .await;
                    match expected_code {
                        Some(code) => match result {
                            Ok(_) => panic!("query exit unexpectedly succeeded"),
                            Err(error) if code == "db.cancel_or_disconnect" => assert!(
                                matches!(
                                    error,
                                    QueryOptionalExecutionError::Cancelled
                                        | QueryOptionalExecutionError::ConnectionUnavailable
                                ),
                                "disconnect race returned unexpected error: {error:?}"
                            ),
                            Err(error) => assert_eq!(error.code(), code),
                        },
                        None => assert_eq!(result.unwrap().as_ref().is_some(), expected_row),
                    }
                    response
                }
            },
        );
        let envelope = match outcome {
            OfficialTokioEntryIoAttemptOutcome::Ready(envelope) => envelope,
            _ => panic!("fixed production DB resources must admit"),
        };
        let layout = (
            std::mem::size_of_val(&envelope),
            std::mem::align_of_val(&envelope),
        );
        assert!(layout.0 <= task_reserve);
        assert!(layout.1.is_power_of_two());
        eprintln!(
            "operation={} envelope_size={} envelope_align={} tested_reserve={}",
            O::OPERATION,
            layout.0,
            layout.1,
            task_reserve
        );
        match server_action {
            TestServerAction::Responsive => {}
            TestServerAction::Stop(pid) => {
                assert!(
                    Command::new("kill")
                        .args(["-STOP", &pid.to_string()])
                        .status()
                        .unwrap()
                        .success()
                );
            }
            TestServerAction::Terminate(pid) => {
                assert!(
                    Command::new("kill")
                        .args(["-TERM", &pid.to_string()])
                        .status()
                        .unwrap()
                        .success()
                );
            }
        }
        let finalization_started = Instant::now();
        let (envelope, task_slot) = envelope.into_runtime_parts();
        tokio::spawn(envelope).await.unwrap().unwrap();
        drop(task_slot);
        let finalization_elapsed = finalization_started.elapsed();
        if let TestServerAction::Stop(pid) = server_action {
            assert!(
                Command::new("kill")
                    .args(["-CONT", &pid.to_string()])
                    .status()
                    .unwrap()
                    .success()
            );
        }
        if !matches!(server_action, TestServerAction::Responsive) {
            assert!(
                finalization_elapsed < Duration::from_secs(1),
                "poison-discard exceeded the existing generated attempt bound: {finalization_elapsed:?}"
            );
        }

        assert_eq!(
            database.pool.num_idle(),
            database.pool.size() as usize,
            "physical return or close must precede request completion"
        );
        assert_eq!(database.pool.size(), u32::from(expected_code.is_none()));
        assert_eq!(db_domain.snapshot().unwrap().connections_in_use, 0);
        assert_eq!(db_domain.snapshot().unwrap().operations_in_use, 0);
        drop(db_domain);
        drop(runtime_domain);
        assert!(!allocation.finish().unwrap().breached);
        assert_eq!(ledger.try_shutdown().unwrap().active_accounts, 0);
        database.shutdown().await.unwrap();
    }

    #[tokio::test]
    async fn real_mariadb_production_query_optional_closes_six_exits() {
        let Ok(url) = env::var("SADDLE_TEST_DATABASE_URL") else {
            eprintln!("skipping production query_optional: SADDLE_TEST_DATABASE_URL is not set");
            return;
        };
        run_production_mode(
            &url,
            QueryOptionalOperationProof::<ProductionSuccess>::bind()
                .unwrap()
                .invocation(ManagedQueryParameters(DbU64(7))),
            FixedSignal::pending(),
            FixedSignal::pending(),
            None,
            true,
        )
        .await;
        run_production_mode(
            &url,
            QueryOptionalOperationProof::<ProductionNone>::bind()
                .unwrap()
                .invocation(ManagedQueryParameters(DbU64(7))),
            FixedSignal::pending(),
            FixedSignal::pending(),
            None,
            false,
        )
        .await;
        run_production_mode(
            &url,
            QueryOptionalOperationProof::<ProductionError>::bind()
                .unwrap()
                .invocation(ManagedQueryParameters(DbU64(7))),
            FixedSignal::pending(),
            FixedSignal::pending(),
            Some("db.query_failed"),
            false,
        )
        .await;
        run_production_mode(
            &url,
            QueryOptionalOperationProof::<ProductionPanic>::bind()
                .unwrap()
                .invocation(ManagedQueryParameters(DbU64(7))),
            FixedSignal::pending(),
            FixedSignal::pending(),
            Some("db.query_failed"),
            false,
        )
        .await;
        run_production_mode(
            &url,
            QueryOptionalOperationProof::<ProductionSlow>::bind()
                .unwrap()
                .invocation(ManagedQueryParameters(DbU64(7))),
            FixedSignal::after_pending_poll(),
            FixedSignal::pending(),
            Some("db.query_cancelled"),
            false,
        )
        .await;
        run_production_mode(
            &url,
            QueryOptionalOperationProof::<ProductionSlow>::bind()
                .unwrap()
                .invocation(ManagedQueryParameters(DbU64(7))),
            FixedSignal::pending(),
            FixedSignal::after_pending_poll(),
            Some("db.query_shutdown"),
            false,
        )
        .await;
    }

    #[tokio::test]
    async fn real_mariadb_cancel_discards_lock_wait_unresponsive_and_disconnected_sockets() {
        let Ok(url) = env::var("SADDLE_TEST_DATABASE_URL") else {
            eprintln!("skipping bounded DB finalization: SADDLE_TEST_DATABASE_URL is not set");
            return;
        };
        let server_pid = env::var("SADDLE_TEST_DATABASE_SERVER_PID")
            .expect("bounded finalization fixture requires the MariaDB PID")
            .parse::<u32>()
            .unwrap();

        let mut blocker = MySqlConnection::connect(&url).await.unwrap();
        sqlx::query(
            "CREATE TABLE IF NOT EXISTS saddle_c6_lock_wait \
             (id BIGINT UNSIGNED PRIMARY KEY, value BIGINT UNSIGNED NOT NULL)",
        )
        .execute(&mut blocker)
        .await
        .unwrap();
        sqlx::query(
            "INSERT INTO saddle_c6_lock_wait VALUES (7, 1) \
             ON DUPLICATE KEY UPDATE value = VALUES(value)",
        )
        .execute(&mut blocker)
        .await
        .unwrap();
        sqlx::query("BEGIN").execute(&mut blocker).await.unwrap();
        sqlx::query("UPDATE saddle_c6_lock_wait SET value = value + 1 WHERE id = 7")
            .execute(&mut blocker)
            .await
            .unwrap();
        run_production_mode(
            &url,
            QueryOptionalOperationProof::<ProductionLockWait>::bind()
                .unwrap()
                .invocation(ManagedQueryParameters(DbU64(7))),
            FixedSignal::after_pending_poll(),
            FixedSignal::pending(),
            Some("db.query_cancelled"),
            false,
        )
        .await;
        sqlx::query("ROLLBACK").execute(&mut blocker).await.unwrap();

        sqlx::query("BEGIN").execute(&mut blocker).await.unwrap();
        sqlx::query("UPDATE saddle_c6_lock_wait SET value = value + 1 WHERE id = 7")
            .execute(&mut blocker)
            .await
            .unwrap();
        run_production_mode(
            &url,
            QueryOptionalOperationProof::<ProductionLockWait>::bind()
                .unwrap()
                .invocation(ManagedQueryParameters(DbU64(7))),
            FixedSignal::pending(),
            FixedSignal::after_pending_poll(),
            Some("db.query_shutdown"),
            false,
        )
        .await;
        sqlx::query("ROLLBACK").execute(&mut blocker).await.unwrap();
        blocker.close().await.unwrap();

        run_production_mode_with_server(
            &url,
            QueryOptionalOperationProof::<ProductionSlow>::bind()
                .unwrap()
                .invocation(ManagedQueryParameters(DbU64(7))),
            FixedSignal::after_pending_poll(),
            FixedSignal::pending(),
            Some("db.query_cancelled"),
            false,
            TestServerAction::Stop(server_pid),
        )
        .await;

        run_production_mode_with_server(
            &url,
            QueryOptionalOperationProof::<ProductionSlow>::bind()
                .unwrap()
                .invocation(ManagedQueryParameters(DbU64(7))),
            FixedSignal::after_pending_poll(),
            FixedSignal::pending(),
            Some("db.cancel_or_disconnect"),
            false,
            TestServerAction::Terminate(server_pid),
        )
        .await;
    }
}