saddle-db 0.2.0-rc.5

//! Production adapter from Runtime's verified DB startup claim to the single
//! Saddle-managed physical pool.

use saddle_admission::DbCreditProfile;
use saddle_core::{ComponentLifecycle, LifecycleFuture};
use saddle_observability::Observer;
use saddle_runtime::startup_assembly::{StartupDbPoolFactory, StartupDbPoolOwner};

use crate::{Database, DatabaseConfig, SaddleError};

/// Linear Database-owned input for Runtime's actual-owner assembly.
///
/// Fields are private and this type is deliberately not `Clone`: the facade
/// can hand one configuration owner to the one official assembly path.
#[doc(hidden)]
pub struct StartupManagedDatabaseFactory {
    config: Option<DatabaseConfig>,
    observer: Observer,
}

impl StartupManagedDatabaseFactory {
    pub fn new(config: Option<DatabaseConfig>, observer: Observer) -> Self {
        Self { config, observer }
    }
}

/// Opaque physical DB owner retained by Runtime until finalization/drop.
///
/// It exposes only the two actual facts required by Runtime's pairing
/// protocol. The managed Database and its raw sqlx pool remain private.
///
/// ```compile_fail
/// # use saddle_db::internal::StartupManagedDatabaseOwner;
/// fn duplicate(owner: StartupManagedDatabaseOwner) {
///     let _second = owner.clone();
/// }
/// ```
///
/// ```compile_fail
/// # use saddle_admission::DbCreditProfile;
/// # use saddle_db::internal::StartupManagedDatabaseOwner;
/// fn forge() -> StartupManagedDatabaseOwner {
///     StartupManagedDatabaseOwner { database: None, profile: DbCreditProfile { connections: 0, operations: 0 } }
/// }
/// ```
#[doc(hidden)]
pub struct StartupManagedDatabaseOwner {
    database: Option<Database>,
    profile: DbCreditProfile,
}

/// One-shot scoped view issued only while the unique startup owner is being
/// consumed by the Facade bootstrap.
///
/// The view cannot be constructed, cloned, replayed, or returned from the
/// higher-ranked bootstrap callback. A managed handle obtained from it remains
/// bounded by the returned owner's shutdown, which permanently closes the
/// shared pool before Runtime finalization.
///
/// ```compile_fail
/// # use saddle_db::internal::StartupManagedDatabaseBootstrap;
/// fn forge<'a>() -> StartupManagedDatabaseBootstrap<'a> {
///     StartupManagedDatabaseBootstrap { database: None }
/// }
/// ```
#[doc(hidden)]
pub struct StartupManagedDatabaseBootstrap<'a> {
    database: Option<&'a Database>,
}

impl StartupManagedDatabaseBootstrap<'_> {
    /// Consumes the one-shot view and returns the existing managed capability.
    /// It never constructs, reconnects, or exposes a raw sqlx pool.
    pub fn existing(self) -> Option<Database> {
        self.database.cloned()
    }
}

impl StartupManagedDatabaseOwner {
    /// Runs one Facade-owned bootstrap consumer and returns the same unique
    /// physical/finalization owner. The HRTB prevents the scoped view from
    /// escaping in `R`; panic/loss drops the owner instead of publishing it.
    pub fn bootstrap<R>(
        self,
        consume: impl for<'a> FnOnce(StartupManagedDatabaseBootstrap<'a>) -> R,
    ) -> (Self, R) {
        let result = consume(StartupManagedDatabaseBootstrap {
            database: self.database.as_ref(),
        });
        (self, result)
    }
}

/// ```compile_fail
/// # use saddle_db::internal::StartupManagedDatabaseOwner;
/// fn replay(owner: StartupManagedDatabaseOwner) {
///     let (owner, ()) = owner.bootstrap(|_| ());
///     let _first = owner.bootstrap(|_| ());
///     let _replay = owner.bootstrap(|_| ());
/// }
/// ```
const _: () = ();

impl Drop for StartupManagedDatabaseOwner {
    fn drop(&mut self) {
        // Make the unique physical owner release explicit. There is no
        // replacement factory, reconnect loop, or second pool.
        drop(self.database.take());
    }
}

impl StartupDbPoolOwner for StartupManagedDatabaseOwner {
    fn connection_capacity(&self) -> usize {
        self.profile.connections
    }

    fn operation_capacity(&self) -> usize {
        self.profile.operations
    }
}

impl ComponentLifecycle for StartupManagedDatabaseOwner {
    fn name(&self) -> &'static str {
        "database"
    }

    fn start(&self) -> LifecycleFuture<'_> {
        Box::pin(std::future::ready(Ok(())))
    }

    fn shutdown(&self) -> LifecycleFuture<'_> {
        Box::pin(async move {
            match self.database.as_ref() {
                Some(database) => database.close().await,
                None => Ok(()),
            }
        })
    }
}

/// Stable failure boundary for construction before Runtime pairs the DB
/// actual-facts receipt.
#[derive(Debug)]
#[doc(hidden)]
pub enum StartupManagedDatabaseError {
    ConfigurationMismatch,
    CapacityOverflow,
    Connect(SaddleError),
}

impl StartupDbPoolFactory for StartupManagedDatabaseFactory {
    type Owner = StartupManagedDatabaseOwner;
    type Error = StartupManagedDatabaseError;

    fn construct(
        self,
        runtime: &tokio::runtime::Runtime,
        required: DbCreditProfile,
    ) -> Result<Self::Owner, Self::Error> {
        if required.connections != required.operations {
            return Err(StartupManagedDatabaseError::ConfigurationMismatch);
        }

        if required.connections == 0 {
            return if self.config.is_none() {
                Ok(StartupManagedDatabaseOwner {
                    database: None,
                    profile: required,
                })
            } else {
                Err(StartupManagedDatabaseError::ConfigurationMismatch)
            };
        }

        let config = self
            .config
            .ok_or(StartupManagedDatabaseError::ConfigurationMismatch)?;
        let connections = u32::try_from(required.connections)
            .map_err(|_| StartupManagedDatabaseError::CapacityOverflow)?;
        let config = config.verified_connections(connections);
        let database = runtime
            .block_on(Database::connect(config, self.observer))
            .map_err(StartupManagedDatabaseError::Connect)?;
        Ok(StartupManagedDatabaseOwner {
            database: Some(database),
            profile: required,
        })
    }
}

#[cfg(test)]
mod tests {
    use std::{
        env, io,
        process::{Command, Stdio},
        time::{Duration, Instant},
    };

    use saddle_admission::{
        RouteCapacityFact, StartupContinuationAdapter, VerifiedDbTerminationServiceProofOwner,
        VerifiedFilesystemTerminationServiceProofOwner, VerifiedGeneratedStartupFactsOwner,
        VerifiedSchedulerTerminationServiceProofOwner, VerifiedStartupContinuationOwner,
        VerifiedSupervisorTerminationServiceProofOwner,
    };
    use saddle_core::{ComponentLifecycle, ErrorKind, LifecycleFuture, SaddleError};
    use saddle_observability::ObserverConfig;
    use saddle_runtime::{
        Application,
        resource_envelope::{DeploymentResourceAttestation, FiniteStartupPolicy},
        startup_assembly::{assemble_actual_startup_owners, run_with_actual_startup_owners},
    };
    use sqlx::{Connection, mysql::MySqlConnection};

    use super::*;

    fn runtime() -> tokio::runtime::Runtime {
        tokio::runtime::Builder::new_multi_thread()
            .worker_threads(1)
            .enable_all()
            .build()
            .unwrap()
    }

    fn observer() -> Observer {
        Observer::with_writer(ObserverConfig::default(), io::sink()).unwrap()
    }

    #[test]
    fn zero_profile_requires_no_database_configuration() {
        let owner = StartupManagedDatabaseFactory::new(None, observer())
            .construct(
                &runtime(),
                DbCreditProfile {
                    connections: 0,
                    operations: 0,
                },
            )
            .unwrap();
        assert!(owner.database.is_none());
        assert_eq!(owner.connection_capacity(), 0);
        assert_eq!(owner.operation_capacity(), 0);
        let (owner, database) = owner.bootstrap(|database| database.existing());
        assert!(database.is_none());
        drop(owner);

        let result = StartupManagedDatabaseFactory::new(
            Some(DatabaseConfig::new("mysql://localhost/unused")),
            observer(),
        )
        .construct(
            &runtime(),
            DbCreditProfile {
                connections: 0,
                operations: 0,
            },
        );
        assert!(matches!(
            result,
            Err(StartupManagedDatabaseError::ConfigurationMismatch)
        ));
    }

    #[test]
    fn asymmetric_and_overflowing_profiles_fail_before_connect() {
        let asymmetric = StartupManagedDatabaseFactory::new(
            Some(DatabaseConfig::new("mysql://localhost/unused")),
            observer(),
        )
        .construct(
            &runtime(),
            DbCreditProfile {
                connections: 1,
                operations: 2,
            },
        );
        assert!(matches!(
            asymmetric,
            Err(StartupManagedDatabaseError::ConfigurationMismatch)
        ));

        if usize::BITS > u32::BITS {
            let overflow = StartupManagedDatabaseFactory::new(
                Some(DatabaseConfig::new("mysql://localhost/unused")),
                observer(),
            )
            .construct(
                &runtime(),
                DbCreditProfile {
                    connections: u32::MAX as usize + 1,
                    operations: u32::MAX as usize + 1,
                },
            );
            assert!(matches!(
                overflow,
                Err(StartupManagedDatabaseError::CapacityOverflow)
            ));
        }
    }

    #[test]
    fn connect_failure_returns_without_a_physical_owner() {
        let result = StartupManagedDatabaseFactory::new(
            Some(
                DatabaseConfig::new("mysql://root@127.0.0.1:1/unreachable")
                    .acquire_timeout(Duration::from_millis(50)),
            ),
            observer(),
        )
        .construct(
            &runtime(),
            DbCreditProfile {
                connections: 1,
                operations: 1,
            },
        );
        assert!(matches!(
            result,
            Err(StartupManagedDatabaseError::Connect(_))
        ));
    }

    #[test]
    fn real_mariadb_constructs_exact_single_pool_and_drop_releases_owner() {
        let Ok(url) = env::var("SADDLE_TEST_DATABASE_URL") else {
            eprintln!("skipping startup pool adapter: SADDLE_TEST_DATABASE_URL is not set");
            return;
        };
        let runtime = runtime();
        let mut admin = runtime.block_on(MySqlConnection::connect(&url)).unwrap();
        let baseline = runtime
            .block_on(
                sqlx::query_scalar::<_, i64>(
                    "SELECT COUNT(*) FROM information_schema.PROCESSLIST WHERE DB = DATABASE()",
                )
                .fetch_one(&mut admin),
            )
            .unwrap();
        let owner = StartupManagedDatabaseFactory::new(
            Some(DatabaseConfig::new(&url).max_connections(99)),
            observer(),
        )
        .construct(
            &runtime,
            DbCreditProfile {
                connections: 2,
                operations: 2,
            },
        )
        .unwrap();
        assert_eq!(owner.connection_capacity(), 2);
        assert_eq!(owner.operation_capacity(), 2);
        let database = owner.database.as_ref().unwrap();
        assert_eq!(database.pool.size(), 2);
        assert_eq!(database.pool.num_idle(), 2);
        let (owner, bootstrap_database) = owner.bootstrap(|database| database.existing());
        let bootstrap_database = bootstrap_database.unwrap();
        assert_eq!(bootstrap_database.pool.size(), 2);
        drop(bootstrap_database);
        let with_pool = runtime
            .block_on(
                sqlx::query_scalar::<_, i64>(
                    "SELECT COUNT(*) FROM information_schema.PROCESSLIST WHERE DB = DATABASE()",
                )
                .fetch_one(&mut admin),
            )
            .unwrap();
        assert_eq!(with_pool, baseline + 2);
        drop(owner);
        runtime.block_on(async {
            for _ in 0..50 {
                let current = sqlx::query_scalar::<_, i64>(
                    "SELECT COUNT(*) FROM information_schema.PROCESSLIST WHERE DB = DATABASE()",
                )
                .fetch_one(&mut admin)
                .await
                .unwrap();
                if current == baseline {
                    return;
                }
                tokio::time::sleep(Duration::from_millis(20)).await;
            }
            panic!("dropping the unique startup owner did not close its physical pool");
        });

        let lost = StartupManagedDatabaseFactory::new(Some(DatabaseConfig::new(&url)), observer())
            .construct(
                &runtime,
                DbCreditProfile {
                    connections: 2,
                    operations: 2,
                },
            )
            .unwrap();
        let panic = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            let _ = lost.bootstrap::<()>(|_| panic!("lost bootstrap consumer"));
        }));
        assert!(panic.is_err());
        runtime.block_on(async {
            for _ in 0..50 {
                let current = sqlx::query_scalar::<_, i64>(
                    "SELECT COUNT(*) FROM information_schema.PROCESSLIST WHERE DB = DATABASE()",
                )
                .fetch_one(&mut admin)
                .await
                .unwrap();
                if current == baseline {
                    return;
                }
                tokio::time::sleep(Duration::from_millis(20)).await;
            }
            panic!("lost bootstrap capability did not drop its physical owner");
        });
        runtime.block_on(admin.close()).unwrap();
        runtime.shutdown_background();
    }

    struct Deployment {
        db: usize,
        cpuset: Vec<usize>,
    }

    impl DeploymentResourceAttestation for Deployment {
        fn cpu_quota_us(&self) -> u64 {
            800_000
        }
        fn cpu_period_us(&self) -> u64 {
            100_000
        }
        fn effective_cpuset(&self) -> &[usize] {
            &self.cpuset
        }
        fn memory_max_bytes(&self) -> usize {
            16_000_000
        }
        fn memory_high_bytes(&self) -> usize {
            8_000_000
        }
        fn saddle_logical_memory_bytes(&self) -> usize {
            4_000_000
        }
        fn saddle_requested_memory_bytes(&self) -> usize {
            4_000_000
        }
        fn registration_credits(&self) -> usize {
            260
        }
        fn event_credits(&self) -> usize {
            260
        }
        fn db_connection_credits(&self) -> usize {
            self.db
        }
        fn db_operation_credits(&self) -> usize {
            self.db
        }
        fn public_time_policy_ms(&self) -> [u64; 2] {
            [2_000, 8_000]
        }
        fn resource_attestation(&self) -> [u8; 32] {
            [1; 32]
        }
        fn public_time_policy_attestation(&self) -> [u8; 32] {
            [2; 32]
        }
        fn environment_identity(&self) -> [u8; 32] {
            [1; 32]
        }
        fn supervisor_attestation(&self) -> [u8; 32] {
            [3; 32]
        }
        fn build_identity(&self) -> [u8; 32] {
            [4; 32]
        }
        fn gate_identity(&self) -> [u8; 32] {
            [5; 32]
        }
    }

    struct Generated<const DB: usize>;
    const ROUTES: [RouteCapacityFact; 1] = [RouteCapacityFact {
        route_id: 1,
        managed_commitment_bytes: 1_024,
        managed_objects_peak: 2,
        framework_bytes: 512,
        task_storage_bytes: 4_096,
        response_carrier_bytes: 512,
        db_connections: 1,
        db_operations: 1,
    }];
    const NO_DB_ROUTES: [RouteCapacityFact; 1] = [RouteCapacityFact {
        route_id: 1,
        managed_commitment_bytes: 1_024,
        managed_objects_peak: 2,
        framework_bytes: 512,
        task_storage_bytes: 4_096,
        response_carrier_bytes: 512,
        db_connections: 0,
        db_operations: 0,
    }];

    impl<const DB: usize> VerifiedGeneratedStartupFactsOwner for Generated<DB> {
        fn costs_attestation(&self) -> [u8; 32] {
            [6; 32]
        }
        fn cost_vectors(&self) -> [[usize; 7]; 5] {
            [
                [0, 16_000, 16_000, 3, 3, 0, 0],
                [100, 100, 100, 0, 0, 0, 0],
                [0, 20_000, 20_000, 1, 1, 0, 0],
                [0, 1_000, 1_000, 1, 1, 0, 0],
                [0, 1_000, 1_000, 0, 0, 1, 1],
            ]
        }
        fn allocator_costs(&self) -> [usize; 2] {
            [4_000, 4_000]
        }
        fn support_attestation(&self) -> [u8; 32] {
            [7; 32]
        }
        fn support_limits(&self) -> [usize; 4] {
            [8, 128, 128, 64]
        }
        fn route_set_attestation(&self) -> [u8; 32] {
            [8; 32]
        }
        fn routes(&self) -> &'static [RouteCapacityFact] {
            if DB == 0 { &NO_DB_ROUTES } else { &ROUTES }
        }
        fn closure_constants(&self) -> [usize; 6] {
            [1, 2, 3, 64, 1_024, 512]
        }
        fn termination_topology(&self) -> [u64; 4] {
            [1, 1, 1, 2]
        }
        fn termination_topology_identity(&self) -> [u8; 32] {
            [9; 32]
        }
        fn db_return_work_identity(&self) -> [u8; 32] {
            [10; 32]
        }
        fn writer_work_identity(&self) -> [u8; 32] {
            [11; 32]
        }
        fn runtime_work_identity(&self) -> [u8; 32] {
            [12; 32]
        }
    }

    struct DbService;
    struct FilesystemService;
    struct SchedulerService;
    struct SupervisorService;

    impl VerifiedDbTerminationServiceProofOwner for DbService {
        fn work_identity(&self) -> [u8; 32] {
            [10; 32]
        }
        fn service_attestation(&self) -> [u8; 32] {
            [13; 32]
        }
        fn max_service_nanos(&self) -> u64 {
            2_000_000
        }
    }
    impl VerifiedFilesystemTerminationServiceProofOwner for FilesystemService {
        fn work_identity(&self) -> [u8; 32] {
            [11; 32]
        }
        fn service_attestation(&self) -> [u8; 32] {
            [14; 32]
        }
        fn max_service_nanos(&self) -> u64 {
            1_000_000
        }
    }
    impl VerifiedSchedulerTerminationServiceProofOwner for SchedulerService {
        fn runtime_work_identity(&self) -> [u8; 32] {
            [12; 32]
        }
        fn service_attestation(&self) -> [u8; 32] {
            [15; 32]
        }
        fn max_delivery_nanos(&self) -> u64 {
            1_000_000
        }
    }
    impl VerifiedSupervisorTerminationServiceProofOwner for SupervisorService {
        fn service_attestation(&self) -> [u8; 32] {
            [16; 32]
        }
        fn shutdown_delivery_nanos(&self) -> u64 {
            1_000_000
        }
        fn delivery_headroom_nanos(&self) -> u64 {
            1_000_000
        }
    }

    struct Continuation;
    impl VerifiedStartupContinuationOwner for Continuation {
        fn generated_facts_identity(&self) -> [u8; 32] {
            [6; 32]
        }
        fn build_identity(&self) -> [u8; 32] {
            [4; 32]
        }
        fn artifact_identity(&self) -> [u8; 32] {
            [17; 32]
        }
    }

    struct Adapter;
    impl StartupContinuationAdapter for Adapter {
        type ContinuationOwner = Continuation;
        fn adapter_provenance(&self) -> [u8; 32] {
            [9; 32]
        }
    }

    fn pending(db: usize) -> saddle_admission::PendingStartupPlan {
        fn create<const DB: usize>(db: usize) -> saddle_admission::PendingStartupPlan {
            saddle_runtime::resource_envelope::create_runtime_startup_plan(
                FiniteStartupPolicy::Balanced,
                [9; 32],
                Deployment {
                    db,
                    cpuset: (0..8).collect(),
                },
                Generated::<DB>,
                (
                    DbService,
                    FilesystemService,
                    SchedulerService,
                    SupervisorService,
                ),
            )
            .unwrap()
        }
        if db == 0 {
            create::<0>(db)
        } else {
            create::<1>(db)
        }
    }

    struct DatabaseComponent {
        owner: StartupManagedDatabaseOwner,
        bootstrap_database: Option<Database>,
        shutdown_error: bool,
    }

    impl ComponentLifecycle for DatabaseComponent {
        fn name(&self) -> &'static str {
            "startup-database-owner"
        }
        fn start(&self) -> LifecycleFuture<'_> {
            assert_eq!(
                self.owner.connection_capacity(),
                self.owner.operation_capacity()
            );
            assert_eq!(
                self.bootstrap_database.is_some(),
                self.owner.connection_capacity() != 0
            );
            Box::pin(async {
                Command::new("sh")
                    .args(["-c", "kill -TERM $PPID"])
                    .status()
                    .unwrap();
                Ok(())
            })
        }
        fn shutdown(&self) -> LifecycleFuture<'_> {
            Box::pin(async move {
                ComponentLifecycle::shutdown(&self.owner).await?;
                if self.shutdown_error {
                    Err(SaddleError::new(
                        ErrorKind::Infrastructure,
                        "test.database_shutdown_failed",
                        "test database shutdown failed",
                    ))
                } else {
                    Ok(())
                }
            })
        }
    }

    #[test]
    fn official_runner_holds_real_database_owner_across_all_exits() {
        let Ok(url) = env::var("SADDLE_TEST_DATABASE_URL") else {
            eprintln!("skipping startup runner joint: SADDLE_TEST_DATABASE_URL is not set");
            return;
        };
        const CHILD: &str = "startup_pool::tests::official_runner_real_database_owner_child";
        for mode in [
            "normal",
            "bootstrap-error",
            "shutdown-error",
            "drop",
            "db0",
            "connect-failure",
        ] {
            let mut child = Command::new(env::current_exe().unwrap())
                .args(["--exact", CHILD, "--nocapture"])
                .env("SADDLE_STARTUP_POOL_RUNNER_CHILD", mode)
                .env("SADDLE_TEST_DATABASE_URL", &url)
                .stdin(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(20) {
                    child.kill().unwrap();
                    child.wait().unwrap();
                    panic!("real DB runner child timed out: {mode}");
                }
                std::thread::sleep(Duration::from_millis(10));
            };
            assert!(status.success(), "real DB runner child failed: {mode}");
        }
    }

    #[test]
    fn official_runner_real_database_owner_child() {
        let Some(mode) = env::var_os("SADDLE_STARTUP_POOL_RUNNER_CHILD") else {
            return;
        };
        let mode = mode.to_str().unwrap();
        let url = env::var("SADDLE_TEST_DATABASE_URL").unwrap();
        let db = if mode == "db0" { 0 } else { 64 };
        let baseline = (db != 0).then(|| database_session_count(&url));
        let config = if db == 0 {
            None
        } else if mode == "connect-failure" {
            Some(DatabaseConfig::new("mysql://root@127.0.0.1:1/unreachable"))
        } else {
            Some(DatabaseConfig::new(&url))
        };
        let actual = assemble_actual_startup_owners(
            pending(db),
            StartupManagedDatabaseFactory::new(config, observer()),
        );
        if mode == "connect-failure" {
            assert!(matches!(
                actual,
                Err(
                    saddle_runtime::startup_assembly::StartupAssemblyError::Database(
                        StartupManagedDatabaseError::Connect(_)
                    )
                )
            ));
            wait_for_database_sessions(&url, baseline.unwrap());
            return;
        }
        let actual = actual.unwrap();
        if mode == "drop" {
            drop(actual);
            wait_for_database_sessions(&url, baseline.unwrap());
            return;
        }
        let (paired, claim, owners) = actual.into_pairing_parts();
        let result = run_with_actual_startup_owners(
            owners,
            paired,
            claim,
            Adapter,
            Continuation,
            |owners| async move {
                if mode == "bootstrap-error" {
                    return Err(owners.fail(SaddleError::new(
                        ErrorKind::Infrastructure,
                        "test.database_bootstrap_failed",
                        "test database bootstrap failed",
                    )));
                }
                let (database, db_domain, tokio_domain, allocation, ledger, _, bound) =
                    owners.into_parts();
                drop((db_domain, tokio_domain, allocation, ledger, bound));
                let (database, bootstrap_database) =
                    database.bootstrap(|database| database.existing());
                let mut application = Application::new();
                application
                    .register(DatabaseComponent {
                        owner: database,
                        bootstrap_database,
                        shutdown_error: mode == "shutdown-error",
                    })
                    .unwrap();
                Ok(application)
            },
        );
        match mode {
            "bootstrap-error" => {
                assert_eq!(result.unwrap_err().code(), "test.database_bootstrap_failed")
            }
            "shutdown-error" => {
                assert_eq!(result.unwrap_err().code(), "test.database_shutdown_failed")
            }
            _ => result.unwrap(),
        }
        if let Some(baseline) = baseline {
            wait_for_database_sessions(&url, baseline);
        }
    }

    fn database_session_count(url: &str) -> i64 {
        let runtime = runtime();
        runtime.block_on(async {
            let mut connection = MySqlConnection::connect(url).await.unwrap();
            let count = sqlx::query_scalar::<_, i64>(
                "SELECT COUNT(*) FROM information_schema.PROCESSLIST WHERE DB = DATABASE()",
            )
            .fetch_one(&mut connection)
            .await
            .unwrap();
            connection.close().await.unwrap();
            count
        })
    }

    fn wait_for_database_sessions(url: &str, expected: i64) {
        for _ in 0..50 {
            if database_session_count(url) == expected {
                return;
            }
            std::thread::sleep(Duration::from_millis(20));
        }
        panic!("official runner did not release the unique physical DB pool");
    }
}