saddle-db 0.2.0-rc.19

//! Sealed work proof for the stock-sqlx normal pool-return path.
//!
//! This module deliberately contains no duration. It describes the finite
//! protocol work that a deployment DB service attestation must cover. A work
//! bound alone cannot turn an unresponsive server into a finite latency bound.

/// Versioned identity for the exact stock-sqlx normal-return work domain.
///
/// The identity is private data returned only by Database. Generated topology
/// can carry it to the startup solver, but cannot forge a different work
/// shape or attach a duration to it.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct DbNormalReturnWorkProof {
    identity: [u8; 32],
    sqlx_major: u16,
    sqlx_minor: u16,
    sqlx_patch: u16,
    max_ping_commands: u8,
    max_protocol_writes: u8,
    max_protocol_reads: u8,
    requires_open_pool: bool,
    requires_no_after_release_hook: bool,
    requires_no_max_lifetime: bool,
    requires_zero_min_connections: bool,
    requires_deployment_service_attestation: bool,
    supports_budget_to_poison_discard: bool,
}

impl DbNormalReturnWorkProof {
    const SQLX_0_8_6_MYSQL: Self = Self {
        identity: [
            0x53, 0x44, 0x4c, 0x2d, 0x44, 0x42, 0x2d, 0x52, 0x45, 0x54, 0x55, 0x52, 0x4e, 0x2d,
            0x30, 0x31, 0x2d, 0x53, 0x51, 0x4c, 0x58, 0x2d, 0x30, 0x38, 0x30, 0x36, 0x2d, 0x4d,
            0x59, 0x53, 0x51, 0x4c,
        ],
        sqlx_major: 0,
        sqlx_minor: 8,
        sqlx_patch: 6,
        max_ping_commands: 1,
        max_protocol_writes: 1,
        max_protocol_reads: 1,
        requires_open_pool: true,
        requires_no_after_release_hook: true,
        requires_no_max_lifetime: true,
        requires_zero_min_connections: true,
        requires_deployment_service_attestation: true,
        // The sqlx return Future owns Floating<Live>; dropping that Future
        // synchronously drops the raw socket and pool size guard. Production
        // still needs Runtime to provide the sealed solver-derived budget
        // Future to select this branch.
        supports_budget_to_poison_discard: true,
    };

    pub const fn identity(self) -> [u8; 32] {
        self.identity
    }

    pub const fn sqlx_version(self) -> (u16, u16, u16) {
        (self.sqlx_major, self.sqlx_minor, self.sqlx_patch)
    }

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

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

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

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

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

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

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

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

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

/// Returns the Database-owned normal-return work-domain proof.
#[doc(hidden)]
pub const fn db_normal_return_work_proof() -> DbNormalReturnWorkProof {
    DbNormalReturnWorkProof::SQLX_0_8_6_MYSQL
}

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

    use saddle_core::ComponentLifecycle;
    use saddle_observability::Observer;
    use sqlx::Executor;

    use super::*;
    use crate::{Database, DatabaseConfig};

    struct FixedBudget {
        pending_polls: Option<u8>,
    }

    impl FixedBudget {
        const fn pending() -> Self {
            Self {
                pending_polls: None,
            }
        }

        const fn after_one_poll() -> Self {
            Self {
                pending_polls: Some(1),
            }
        }
    }

    impl Future for FixedBudget {
        type Output = ();

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

    #[derive(Clone, Copy, Debug, Eq, PartialEq)]
    enum ReturnOutcome {
        Returned,
        PoisonDiscarded,
    }

    async fn return_or_discard<R, B>(return_to_pool: R, budget: B) -> ReturnOutcome
    where
        R: Future<Output = ()>,
        B: Future<Output = ()>,
    {
        let mut return_to_pool = pin!(return_to_pool);
        let mut budget = pin!(budget);
        std::future::poll_fn(|context| {
            if return_to_pool.as_mut().poll(context).is_ready() {
                return Poll::Ready(ReturnOutcome::Returned);
            }
            if budget.as_mut().poll(context).is_ready() {
                // Returning drops the still-Pending sqlx Future in this same
                // task. That Future owns Floating<Live>, so raw socket and
                // pool size guard are synchronously poison-discarded.
                return Poll::Ready(ReturnOutcome::PoisonDiscarded);
            }
            Poll::Pending
        })
        .await
    }

    async fn database(url: &str) -> Database {
        Database::connect(
            DatabaseConfig::new(url).max_connections(1),
            Observer::with_writer(Default::default(), std::io::sink()).unwrap(),
        )
        .await
        .unwrap()
    }

    #[test]
    fn proof_has_identity_and_work_units_but_no_duration() {
        let proof = db_normal_return_work_proof();
        assert_ne!(proof.identity(), [0; 32]);
        assert_eq!(proof.sqlx_version(), (0, 8, 6));
        assert_eq!(proof.max_ping_commands(), 1);
        assert_eq!(proof.max_protocol_writes(), 1);
        assert_eq!(proof.max_protocol_reads(), 1);
        assert!(proof.requires_open_pool());
        assert!(proof.requires_no_after_release_hook());
        assert!(proof.requires_no_max_lifetime());
        assert!(proof.requires_zero_min_connections());
        assert!(proof.requires_deployment_service_attestation());
        assert!(proof.supports_budget_to_poison_discard());
        assert_eq!(std::mem::size_of::<DbNormalReturnWorkProof>(), 48);
    }

    #[tokio::test]
    async fn real_mariadb_normal_return_or_unresponsive_budget_discards_once() {
        let Ok(url) = env::var("SADDLE_TEST_DATABASE_URL") else {
            eprintln!("skipping normal-return proof: SADDLE_TEST_DATABASE_URL is not set");
            return;
        };
        let server_pid: u32 = env::var("SADDLE_TEST_DATABASE_SERVER_PID")
            .expect("real unresponsive proof requires the isolated MariaDB pid")
            .parse()
            .unwrap();

        let normal = database(&url).await;
        let mut connection = normal.pool.try_acquire().unwrap();
        connection.execute("SELECT 1").await.unwrap();
        let outcome = return_or_discard(connection.return_to_pool(), FixedBudget::pending()).await;
        assert_eq!(outcome, ReturnOutcome::Returned);
        assert_eq!((normal.pool.size(), normal.pool.num_idle()), (1, 1));
        normal.shutdown().await.unwrap();

        let unresponsive = database(&url).await;
        let mut connection = unresponsive.pool.try_acquire().unwrap();
        connection.execute("SELECT 1").await.unwrap();
        let return_to_pool = connection.return_to_pool();
        assert!(
            Command::new("kill")
                .args(["-STOP", &server_pid.to_string()])
                .status()
                .unwrap()
                .success()
        );
        let outcome = return_or_discard(return_to_pool, FixedBudget::after_one_poll()).await;
        assert_eq!(outcome, ReturnOutcome::PoisonDiscarded);
        assert_eq!(
            (unresponsive.pool.size(), unresponsive.pool.num_idle()),
            (0, 0)
        );
        assert!(
            Command::new("kill")
                .args(["-CONT", &server_pid.to_string()])
                .status()
                .unwrap()
                .success()
        );
        unresponsive.shutdown().await.unwrap();
    }
}