saddle-runtime 0.2.0-rc.3

//! Linear construction of the six Runtime startup owners.
//!
//! Requirements may be inspected to size a real owner, but a receipt is made
//! only from facts observed from the constructed owner.  This module contains
//! the sole pairing path; it deliberately offers no numeric or closure adapter.

use std::{sync::Arc, time::Duration};

use saddle_admission::{
    BoundStartupContinuation, CompletedStartupClaims, DbCreditProfile, DbPermitDomain,
    OfficialTokioDomain, PairedStartupPlan, PendingStartupPlan, ProcessAllocationProfile,
    ProcessLedger, StartupActualFacts, StartupClaim, StartupClaimKind, StartupContinuationAdapter,
    StartupContinuationClaim, StartupPairingError, VerifiedStartupContinuationOwner,
    VerifiedTransportStartupClaim, bind_startup_continuation,
};
use saddle_core::{ComponentLifecycle, ErrorKind, Result as SaddleResult, SaddleError};

use crate::{Application, application::ShutdownSignal};

#[doc(hidden)]
pub trait StartupDbPoolFactory: Sized {
    type Owner: StartupDbPoolOwner;
    type Error;

    fn construct(
        self,
        runtime: &tokio::runtime::Runtime,
        required: DbCreditProfile,
    ) -> Result<Self::Owner, Self::Error>;
}

#[doc(hidden)]
pub trait StartupDbPoolOwner: Sized {
    fn connection_capacity(&self) -> usize;
    fn operation_capacity(&self) -> usize;
}

#[doc(hidden)]
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct DerivedTerminationPlan {
    request_timeout: Duration,
    db_return_budget: Duration,
    finalization_requirement: Duration,
    writer_shutdown_budget: Duration,
    component_shutdown_budget: Duration,
    runtime_termination_requirement: Duration,
    shutdown_grace: Duration,
    public_policy_attestation: [u8; 32],
    generated_requirements_attestation: [u8; 32],
    supervisor_attestation: [u8; 32],
}

impl DerivedTerminationPlan {
    pub fn request_timeout(&self) -> Duration {
        self.request_timeout
    }
    pub fn db_return_budget(&self) -> Duration {
        self.db_return_budget
    }
    pub fn finalization_requirement(&self) -> Duration {
        self.finalization_requirement
    }
    pub fn runtime_termination_requirement(&self) -> Duration {
        self.runtime_termination_requirement
    }
    pub fn shutdown_grace(&self) -> Duration {
        self.shutdown_grace
    }

    fn facts(self) -> StartupActualFacts {
        StartupActualFacts::Termination {
            request_timeout_ms: millis(self.request_timeout),
            db_return_budget_ms: millis(self.db_return_budget),
            finalization_requirement_ms: millis(self.finalization_requirement),
            writer_shutdown_budget_ms: millis(self.writer_shutdown_budget),
            component_shutdown_budget_ms: millis(self.component_shutdown_budget),
            runtime_termination_requirement_ms: millis(self.runtime_termination_requirement),
            shutdown_grace_ms: millis(self.shutdown_grace),
            public_policy_attestation: self.public_policy_attestation,
            generated_requirements_attestation: self.generated_requirements_attestation,
            supervisor_attestation: self.supervisor_attestation,
        }
    }
}

fn millis(value: Duration) -> u64 {
    u64::try_from(value.as_millis()).expect("verified startup duration fits u64")
}

#[doc(hidden)]
pub struct ActualStartupOwners<D> {
    database: D,
    db_domain: Option<DbPermitDomain>,
    tokio_domain: OfficialTokioDomain,
    runtime: tokio::runtime::Runtime,
    allocation: ProcessAllocationProfile,
    ledger: ProcessLedger,
    termination: DerivedTerminationPlan,
    paired: PairedStartupPlan,
    continuation: StartupContinuationClaim,
}

impl<D> ActualStartupOwners<D> {
    pub fn runtime(&self) -> &tokio::runtime::Runtime {
        &self.runtime
    }
    pub fn ledger(&self) -> &ProcessLedger {
        &self.ledger
    }
    pub fn tokio_domain(&self) -> &OfficialTokioDomain {
        &self.tokio_domain
    }
    pub fn db_domain(&self) -> Option<&DbPermitDomain> {
        self.db_domain.as_ref()
    }
    pub fn allocation(&self) -> &ProcessAllocationProfile {
        &self.allocation
    }
    pub fn database(&self) -> &D {
        &self.database
    }
    pub fn termination(&self) -> DerivedTerminationPlan {
        self.termination
    }

    pub fn issue_transport_profile(
        &mut self,
    ) -> Result<VerifiedTransportRuntimeProfile, TransportRuntimeProfileError> {
        let binding = self.paired.transport_binding();
        let claim = self
            .paired
            .claim_transport_startup_profile()
            .map_err(|_| TransportRuntimeProfileError::DuplicateOrForeignClaim)?;
        verify_transport_profile_parts(&self.tokio_domain, self.termination, binding, claim)
    }

    pub fn into_pairing_parts(
        self,
    ) -> (
        PairedStartupPlan,
        StartupContinuationClaim,
        StartupRuntimeOwners<D>,
    ) {
        let binding = self.paired.transport_binding();
        (
            self.paired,
            self.continuation,
            StartupRuntimeOwners {
                database: self.database,
                db_domain: self.db_domain,
                tokio_domain: self.tokio_domain,
                runtime: self.runtime,
                allocation: self.allocation,
                ledger: self.ledger,
                termination: self.termination,
                transport_binding: binding,
            },
        )
    }
}

#[doc(hidden)]
pub struct StartupRuntimeOwners<D> {
    database: D,
    db_domain: Option<DbPermitDomain>,
    tokio_domain: OfficialTokioDomain,
    runtime: tokio::runtime::Runtime,
    allocation: ProcessAllocationProfile,
    ledger: ProcessLedger,
    termination: DerivedTerminationPlan,
    transport_binding: ([u8; 32], u64, [u8; 32], [u8; 32]),
}

/// Linear transport profile derived from the paired StartupPlan and checked
/// against the physical Runtime owner. It is framework-internal and exposes no
/// constructor or Clone path to generated/business code.
///
/// ```compile_fail
/// # use saddle_runtime::startup_assembly::VerifiedTransportRuntimeProfile;
/// fn replay(value: VerifiedTransportRuntimeProfile) {
///     let first = value;
///     let _second = value;
///     drop(first);
/// }
/// ```
#[doc(hidden)]
#[derive(Debug)]
pub struct VerifiedTransportRuntimeProfile {
    plan_digest: [u8; 32],
    owner_generation: u64,
    build_identity: [u8; 32],
    route_set_attestation: [u8; 32],
    head_deadline: Duration,
    attempt_deadline: Duration,
    finalization_termination_bound: Duration,
    task_storage_bound: usize,
}

#[doc(hidden)]
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum TransportRuntimeProfileError {
    DuplicateOrForeignClaim,
    InvalidIdentity,
    ActualOwnerDrift,
    InvalidBound,
}

impl<D> StartupRuntimeOwners<D> {
    pub fn runtime(&self) -> &tokio::runtime::Runtime {
        &self.runtime
    }
    pub fn ledger(&self) -> &ProcessLedger {
        &self.ledger
    }
    pub fn tokio_domain(&self) -> &OfficialTokioDomain {
        &self.tokio_domain
    }
    pub fn db_domain(&self) -> Option<&DbPermitDomain> {
        self.db_domain.as_ref()
    }
    pub fn allocation(&self) -> &ProcessAllocationProfile {
        &self.allocation
    }
    pub fn database(&self) -> &D {
        &self.database
    }
    pub fn termination(&self) -> DerivedTerminationPlan {
        self.termination
    }

    pub fn verify_transport_profile(
        &self,
        claim: VerifiedTransportStartupClaim,
    ) -> Result<VerifiedTransportRuntimeProfile, TransportRuntimeProfileError> {
        verify_transport_profile_parts(
            &self.tokio_domain,
            self.termination,
            self.transport_binding,
            claim,
        )
    }
}

fn verify_transport_profile_parts(
    tokio_domain: &OfficialTokioDomain,
    termination: DerivedTerminationPlan,
    expected_binding: ([u8; 32], u64, [u8; 32], [u8; 32]),
    claim: VerifiedTransportStartupClaim,
) -> Result<VerifiedTransportRuntimeProfile, TransportRuntimeProfileError> {
    let (
        plan_digest,
        owner_generation,
        build_identity,
        route_set_attestation,
        task_capacity,
        task_storage_bound,
        request_timeout_ms,
        finalization_requirement_ms,
    ) = claim.facts();
    if [plan_digest, build_identity, route_set_attestation].contains(&[0; 32])
        || owner_generation == 0
    {
        return Err(TransportRuntimeProfileError::InvalidIdentity);
    }
    if (
        plan_digest,
        owner_generation,
        build_identity,
        route_set_attestation,
    ) != expected_binding
    {
        return Err(TransportRuntimeProfileError::DuplicateOrForeignClaim);
    }
    let snapshot = tokio_domain
        .snapshot()
        .map_err(|_| TransportRuntimeProfileError::ActualOwnerDrift)?;
    if snapshot.task_capacity != task_capacity {
        return Err(TransportRuntimeProfileError::ActualOwnerDrift);
    }
    let request_timeout = Duration::from_millis(request_timeout_ms);
    let finalization = Duration::from_millis(finalization_requirement_ms);
    if task_storage_bound == 0
        || request_timeout.is_zero()
        || finalization.is_zero()
        || request_timeout != termination.request_timeout
        || finalization != termination.finalization_requirement
    {
        return Err(TransportRuntimeProfileError::InvalidBound);
    }
    Ok(VerifiedTransportRuntimeProfile {
        plan_digest,
        owner_generation,
        build_identity,
        route_set_attestation,
        head_deadline: request_timeout,
        attempt_deadline: request_timeout,
        finalization_termination_bound: finalization,
        task_storage_bound,
    })
}

impl VerifiedTransportRuntimeProfile {
    #[doc(hidden)]
    pub fn identities(&self) -> ([u8; 32], u64, [u8; 32], [u8; 32]) {
        (
            self.plan_digest,
            self.owner_generation,
            self.build_identity,
            self.route_set_attestation,
        )
    }

    #[doc(hidden)]
    pub fn into_transport_parts(self) -> (Duration, Duration, Duration, usize) {
        (
            self.head_deadline,
            self.attempt_deadline,
            self.finalization_termination_bound,
            self.task_storage_bound,
        )
    }
}

/// Owners delivered to the sole async bootstrap. The type is consuming and
/// non-Clone; production assembly must move its ledger domains into the one
/// request coordinator before returning `Application`.
#[doc(hidden)]
pub struct StartupBootstrapOwners<D, C> {
    database: D,
    db_domain: Option<DbPermitDomain>,
    tokio_domain: OfficialTokioDomain,
    allocation: ProcessAllocationProfile,
    ledger: ProcessLedger,
    termination: DerivedTerminationPlan,
    bound: BoundStartupContinuation<C>,
}

impl<D, C> StartupBootstrapOwners<D, C> {
    pub fn fail(self, error: SaddleError) -> StartupBootstrapFailure<D, C> {
        StartupBootstrapFailure {
            owners: self,
            error,
        }
    }

    #[allow(clippy::type_complexity)]
    pub fn into_parts(
        self,
    ) -> (
        D,
        Option<DbPermitDomain>,
        OfficialTokioDomain,
        ProcessAllocationProfile,
        ProcessLedger,
        DerivedTerminationPlan,
        BoundStartupContinuation<C>,
    ) {
        (
            self.database,
            self.db_domain,
            self.tokio_domain,
            self.allocation,
            self.ledger,
            self.termination,
            self.bound,
        )
    }
}

#[doc(hidden)]
pub struct StartupBootstrapFailure<D, C> {
    owners: StartupBootstrapOwners<D, C>,
    error: SaddleError,
}

#[doc(hidden)]
pub struct BootstrapOwnerView<'a, D, C> {
    database: &'a D,
    termination: DerivedTerminationPlan,
    bound: &'a BoundStartupContinuation<C>,
}

impl<'a, D, C> BootstrapOwnerView<'a, D, C> {
    pub fn database(&self) -> &'a D {
        self.database
    }
    pub fn termination(&self) -> DerivedTerminationPlan {
        self.termination
    }
    pub fn bound(&self) -> &'a BoundStartupContinuation<C> {
        self.bound
    }
}

#[doc(hidden)]
pub trait BootstrapInstallAdapter<D, C>: Sized {
    type Prepared: PreparedBootstrapInstall<D, C>;
    type Error;

    fn prepare(self, owners: BootstrapOwnerView<'_, D, C>) -> Result<Self::Prepared, Self::Error>;
}

#[doc(hidden)]
pub trait PreparedBootstrapInstall<D, C>: Sized {
    fn component_names(&self) -> &'static [&'static str];
    fn install(
        self,
        owners: StartupBootstrapOwners<D, C>,
        seal: BootstrapBatchSeal,
    ) -> PreparedBootstrapBatch;
}

#[doc(hidden)]
pub struct BootstrapTransaction<D, C> {
    owners: StartupBootstrapOwners<D, C>,
}

#[doc(hidden)]
pub struct PreparedBootstrapTransaction<D, C, P> {
    owners: StartupBootstrapOwners<D, C>,
    prepared: P,
    receipt: BootstrapReservationReceipt,
}

#[doc(hidden)]
pub struct BootstrapPrepareFailure<D, C, E> {
    owners: StartupBootstrapOwners<D, C>,
    error: BootstrapPrepareError<E>,
}

#[doc(hidden)]
pub enum BootstrapPrepareError<E> {
    Adapter(E),
    InvalidComponentSet,
}

type BootstrapPrepareResult<D, C, P, E> =
    Result<PreparedBootstrapTransaction<D, C, P>, BootstrapPrepareFailure<D, C, E>>;

impl<D, C, E> BootstrapPrepareFailure<D, C, E> {
    pub fn error(&self) -> &BootstrapPrepareError<E> {
        &self.error
    }

    pub fn into_runner_failure(self, error: SaddleError) -> StartupBootstrapFailure<D, C> {
        self.owners.fail(error)
    }
}

#[doc(hidden)]
pub struct BootstrapBatchSeal {
    _receipt: BootstrapReservationReceipt,
    application: Application,
}

#[derive(Clone, Copy)]
struct BootstrapReservationReceipt {
    _private: (),
}

#[doc(hidden)]
pub struct PreparedBootstrapBatch {
    application: Application,
}

impl BootstrapBatchSeal {
    pub fn pending_driver_finalizer(&self) -> crate::post_driver::PendingDriverFinalizerSlot {
        self.application.pending_driver_finalizer()
    }

    pub fn install(
        mut self,
        components: Vec<Arc<dyn ComponentLifecycle>>,
    ) -> PreparedBootstrapBatch {
        self.application.install_prevalidated_components(components);
        PreparedBootstrapBatch {
            application: self.application,
        }
    }
}

impl<D, C> BootstrapTransaction<D, C> {
    pub fn new(owners: StartupBootstrapOwners<D, C>) -> Self {
        Self { owners }
    }

    pub fn prepare<A>(self, adapter: A) -> BootstrapPrepareResult<D, C, A::Prepared, A::Error>
    where
        A: BootstrapInstallAdapter<D, C>,
    {
        let view = BootstrapOwnerView {
            database: &self.owners.database,
            termination: self.owners.termination,
            bound: &self.owners.bound,
        };
        let prepared = match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            adapter.prepare(view)
        })) {
            Ok(Ok(prepared)) => prepared,
            Ok(Err(error)) => {
                return Err(BootstrapPrepareFailure {
                    owners: self.owners,
                    error: BootstrapPrepareError::Adapter(error),
                });
            }
            Err(_) => std::process::abort(),
        };
        let names = prepared.component_names();
        const REQUIRED: [&str; 3] = ["observability", "database", "http1-production"];
        if names.is_empty()
            || names.iter().any(|name| name.is_empty())
            || names
                .iter()
                .enumerate()
                .any(|(index, name)| names[..index].contains(name))
            || REQUIRED.iter().any(|required| !names.contains(required))
        {
            return Err(BootstrapPrepareFailure {
                owners: self.owners,
                error: BootstrapPrepareError::InvalidComponentSet,
            });
        }
        Ok(PreparedBootstrapTransaction {
            owners: self.owners,
            prepared,
            receipt: BootstrapReservationReceipt { _private: () },
        })
    }
}

impl<D, C, P> PreparedBootstrapTransaction<D, C, P>
where
    P: PreparedBootstrapInstall<D, C>,
{
    pub fn commit(self) -> Application {
        let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            self.prepared.install(
                self.owners,
                BootstrapBatchSeal {
                    _receipt: self.receipt,
                    application: Application::new(),
                },
            )
        }));
        match outcome {
            Ok(batch) => batch.application,
            Err(_) => std::process::abort(),
        }
    }
}

/// Runs the application on the Runtime which was already constructed and
/// receipted by the startup plan. No Runtime handle or general spawn surface
/// escapes this function.
#[doc(hidden)]
pub fn run_with_actual_startup_owners<A, D, F, Fut>(
    owners: StartupRuntimeOwners<D>,
    paired: PairedStartupPlan,
    continuation_claim: StartupContinuationClaim,
    adapter: A,
    continuation: A::ContinuationOwner,
    bootstrap: F,
) -> SaddleResult<()>
where
    A: StartupContinuationAdapter,
    A::ContinuationOwner: VerifiedStartupContinuationOwner,
    D: Send + 'static,
    F: FnOnce(StartupBootstrapOwners<D, A::ContinuationOwner>) -> Fut,
    Fut: std::future::Future<
            Output = Result<Application, StartupBootstrapFailure<D, A::ContinuationOwner>>,
        >,
{
    Application::claim_process_runtime()?;
    let bound = bind_startup_continuation(paired, continuation_claim, adapter, continuation)
        .map_err(|_| runner_error("runtime.startup_continuation_bind_failed"))?;

    let StartupRuntimeOwners {
        database,
        db_domain,
        tokio_domain,
        runtime,
        allocation,
        ledger,
        termination,
        transport_binding: _,
    } = owners;
    let bootstrap_owners = StartupBootstrapOwners {
        database,
        db_domain,
        tokio_domain,
        allocation,
        ledger,
        termination,
        bound,
    };

    let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
        runtime.block_on(async {
            let signal = match ShutdownSignal::register() {
                Ok(signal) => signal,
                Err(error) => return Err(bootstrap_owners.fail(error)),
            };
            let application = bootstrap(bootstrap_owners).await?;
            let finalizer = application.pending_driver_finalizer();
            let result = application.run_until_shutdown(signal.wait()).await;
            Ok::<_, StartupBootstrapFailure<D, A::ContinuationOwner>>((finalizer, result))
        })
    }));
    match outcome {
        Ok(Ok((finalizer, result))) => finalizer.finish(runtime, result),
        Ok(Err(failure)) => failure.finish(runtime),
        Err(_) => std::process::abort(),
    }
}

impl<D, C> StartupBootstrapFailure<D, C> {
    fn finish(self, runtime: tokio::runtime::Runtime) -> SaddleResult<()> {
        let StartupBootstrapOwners {
            database,
            db_domain,
            tokio_domain,
            allocation,
            ledger,
            termination: _,
            bound: _,
        } = self.owners;
        drop((database, db_domain, tokio_domain));
        drop(runtime);
        let allocation_result = allocation.finish();
        let ledger_result = ledger.try_shutdown();
        if allocation_result.is_err() || ledger_result.is_err() {
            Err(runner_error("runtime.bootstrap_rollback_failed"))
        } else {
            Err(self.error)
        }
    }
}

fn runner_error(code: &'static str) -> SaddleError {
    SaddleError::new(
        ErrorKind::Infrastructure,
        code,
        "the verified Runtime startup owner could not be run",
    )
}

#[doc(hidden)]
#[derive(Debug)]
pub enum StartupAssemblyError<E> {
    Pairing(StartupPairingError),
    InvalidRequirement,
    Runtime,
    Ledger,
    Database(E),
}

impl<E> From<StartupPairingError> for StartupAssemblyError<E> {
    fn from(value: StartupPairingError) -> Self {
        Self::Pairing(value)
    }
}

struct Transaction {
    pending: Option<PendingStartupPlan>,
    completed: Option<CompletedStartupClaims>,
    active: Vec<StartupClaim>,
}

impl Transaction {
    fn new(pending: PendingStartupPlan) -> Self {
        let completed = pending.receipt_set();
        Self {
            pending: Some(pending),
            completed: Some(completed),
            active: Vec::new(),
        }
    }

    fn requirement(
        &mut self,
        kind: StartupClaimKind,
    ) -> Result<StartupActualFacts, StartupPairingError> {
        let claim = self
            .pending
            .as_mut()
            .expect("transaction is live")
            .claim(kind)?;
        let expected = claim.expected();
        self.active.push(claim);
        Ok(expected)
    }

    fn complete(
        &mut self,
        kind: StartupClaimKind,
        actual: StartupActualFacts,
    ) -> Result<(), StartupPairingError> {
        let index = self
            .active
            .iter()
            .position(|claim| claim.kind() == kind)
            .expect("startup claim is active");
        let claim = self.active.swap_remove(index);
        let receipt = claim.complete(actual)?;
        self.completed
            .as_mut()
            .expect("transaction is live")
            .insert(receipt)
            .map_err(|(error, _)| error)
    }

    fn pair(
        mut self,
    ) -> Result<(PairedStartupPlan, StartupContinuationClaim), StartupPairingError> {
        debug_assert!(self.active.is_empty());
        let mut pending = self.pending.take().expect("transaction is live");
        let completed = self.completed.take().expect("transaction is live");
        let continuation = pending.continuation_claim()?;
        Ok((pending.pair(completed)?, continuation))
    }
}

impl Drop for Transaction {
    fn drop(&mut self) {
        let Some(pending) = self.pending.as_mut() else {
            return;
        };
        for claim in self.active.drain(..) {
            let _ = claim.rollback(pending);
        }
        if let Some(completed) = self.completed.take() {
            let _ = completed.rollback(pending);
        }
    }
}

/// Constructs and retains every physical owner before returning a pairable
/// continuation claim.  Dropping on any error destroys constructed resources;
/// no publish token can be obtained from a partial transaction.
#[doc(hidden)]
pub fn assemble_actual_startup_owners<F>(
    pending: PendingStartupPlan,
    database_factory: F,
) -> Result<ActualStartupOwners<F::Owner>, StartupAssemblyError<F::Error>>
where
    F: StartupDbPoolFactory,
{
    let mut tx = Transaction::new(pending);

    let (workers, events) = match tx.requirement(StartupClaimKind::Runtime)? {
        StartupActualFacts::Runtime {
            worker_threads,
            event_credits,
        } => (worker_threads, event_credits),
        _ => return Err(StartupAssemblyError::InvalidRequirement),
    };
    if workers == 0 || events == 0 {
        return Err(StartupAssemblyError::InvalidRequirement);
    }
    let runtime = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(workers)
        .max_io_events_per_tick(events)
        .enable_all()
        .build()
        .map_err(|_| StartupAssemblyError::Runtime)?;
    tx.complete(
        StartupClaimKind::Runtime,
        StartupActualFacts::Runtime {
            worker_threads: workers,
            event_credits: events,
        },
    )?;

    let wait_slots = match tx.requirement(StartupClaimKind::WaitOwners)? {
        StartupActualFacts::WaitOwners { wait_slots } => wait_slots,
        _ => return Err(StartupAssemblyError::InvalidRequirement),
    };

    let (config, registration, task_slots) =
        match tx.requirement(StartupClaimKind::LedgerRegistries)? {
            StartupActualFacts::LedgerRegistries {
                resource_config,
                registration_profile,
                task_slots,
            } => (resource_config, registration_profile, task_slots),
            _ => return Err(StartupAssemblyError::InvalidRequirement),
        };
    let ledger = ProcessLedger::new_with_waiters(config, wait_slots)
        .map_err(|_| StartupAssemblyError::Ledger)?;
    if ledger.wait_snapshot().capacity != wait_slots {
        return Err(StartupAssemblyError::Ledger);
    }
    tx.complete(
        StartupClaimKind::WaitOwners,
        StartupActualFacts::WaitOwners {
            wait_slots: ledger.wait_snapshot().capacity,
        },
    )?;
    let tokio_domain = ledger
        .prepare_official_tokio_domain(registration)
        .map_err(|_| StartupAssemblyError::Ledger)?;
    let tokio_snapshot = tokio_domain
        .snapshot()
        .map_err(|_| StartupAssemblyError::Ledger)?;
    if tokio_snapshot.task_capacity != task_slots {
        return Err(StartupAssemblyError::Ledger);
    }
    tx.complete(
        StartupClaimKind::LedgerRegistries,
        StartupActualFacts::LedgerRegistries {
            resource_config: config,
            registration_profile: registration,
            task_slots: tokio_snapshot.task_capacity,
        },
    )?;

    let db_required = match tx.requirement(StartupClaimKind::DbPool)? {
        StartupActualFacts::DbPool {
            connections,
            operations,
        } => DbCreditProfile {
            connections,
            operations,
        },
        _ => return Err(StartupAssemblyError::InvalidRequirement),
    };
    let database = database_factory
        .construct(&runtime, db_required)
        .map_err(StartupAssemblyError::Database)?;
    let actual_db = DbCreditProfile {
        connections: database.connection_capacity(),
        operations: database.operation_capacity(),
    };
    let db_domain = if actual_db.connections == 0 && actual_db.operations == 0 {
        None
    } else {
        Some(
            ledger
                .prepare_db_domain(actual_db)
                .map_err(|_| StartupAssemblyError::Ledger)?,
        )
    };
    tx.complete(
        StartupClaimKind::DbPool,
        StartupActualFacts::DbPool {
            connections: actual_db.connections,
            operations: actual_db.operations,
        },
    )?;

    let (requested_bytes, profile_attestation) =
        match tx.requirement(StartupClaimKind::Watermark)? {
            StartupActualFacts::Watermark {
                requested_bytes,
                profile_attestation,
            } => (requested_bytes, profile_attestation),
            _ => return Err(StartupAssemblyError::InvalidRequirement),
        };
    if requested_bytes == usize::MAX || requested_bytes == 0 || profile_attestation == [0; 32] {
        return Err(StartupAssemblyError::InvalidRequirement);
    }
    let allocation = ledger
        .prepare_process_allocation_profile(requested_bytes)
        .map_err(|_| StartupAssemblyError::Ledger)?;
    tx.complete(
        StartupClaimKind::Watermark,
        StartupActualFacts::Watermark {
            requested_bytes,
            profile_attestation,
        },
    )?;

    let termination = termination_from(tx.requirement(StartupClaimKind::Termination)?)
        .ok_or(StartupAssemblyError::InvalidRequirement)?;
    tx.complete(StartupClaimKind::Termination, termination.facts())?;

    let (paired, continuation) = tx.pair()?;
    Ok(ActualStartupOwners {
        database,
        db_domain,
        tokio_domain,
        runtime,
        allocation,
        ledger,
        termination,
        paired,
        continuation,
    })
}

fn termination_from(facts: StartupActualFacts) -> Option<DerivedTerminationPlan> {
    let StartupActualFacts::Termination {
        request_timeout_ms,
        db_return_budget_ms,
        finalization_requirement_ms,
        writer_shutdown_budget_ms,
        component_shutdown_budget_ms,
        runtime_termination_requirement_ms,
        shutdown_grace_ms,
        public_policy_attestation,
        generated_requirements_attestation,
        supervisor_attestation,
    } = facts
    else {
        return None;
    };
    let values = [
        request_timeout_ms,
        db_return_budget_ms,
        finalization_requirement_ms,
        writer_shutdown_budget_ms,
        component_shutdown_budget_ms,
        runtime_termination_requirement_ms,
        shutdown_grace_ms,
    ];
    if values.contains(&0)
        || [
            public_policy_attestation,
            generated_requirements_attestation,
            supervisor_attestation,
        ]
        .contains(&[0; 32])
    {
        return None;
    }
    Some(DerivedTerminationPlan {
        request_timeout: Duration::from_millis(request_timeout_ms),
        db_return_budget: Duration::from_millis(db_return_budget_ms),
        finalization_requirement: Duration::from_millis(finalization_requirement_ms),
        writer_shutdown_budget: Duration::from_millis(writer_shutdown_budget_ms),
        component_shutdown_budget: Duration::from_millis(component_shutdown_budget_ms),
        runtime_termination_requirement: Duration::from_millis(runtime_termination_requirement_ms),
        shutdown_grace: Duration::from_millis(shutdown_grace_ms),
        public_policy_attestation,
        generated_requirements_attestation,
        supervisor_attestation,
    })
}

#[cfg(test)]
mod tests {
    use std::{
        env,
        os::unix::process::ExitStatusExt,
        process::{Command, Stdio},
        sync::{
            Arc,
            atomic::{AtomicBool, Ordering},
        },
        time::Instant,
    };

    use saddle_admission::{StartupContinuationAdapter, VerifiedStartupContinuationOwner};
    use saddle_core::{ComponentLifecycle, LifecycleFuture};

    use super::*;

    struct DbFactory {
        drift: bool,
        dropped: Arc<AtomicBool>,
    }

    struct DbOwner {
        profile: DbCreditProfile,
        dropped: Arc<AtomicBool>,
    }

    impl Drop for DbOwner {
        fn drop(&mut self) {
            self.dropped.store(true, Ordering::Release);
        }
    }

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

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

    impl StartupDbPoolFactory for DbFactory {
        type Owner = DbOwner;
        type Error = ();

        fn construct(
            self,
            runtime: &tokio::runtime::Runtime,
            mut required: DbCreditProfile,
        ) -> Result<Self::Owner, Self::Error> {
            runtime.block_on(std::future::ready(()));
            if self.drift {
                required.connections = required.connections.saturating_sub(1);
            }
            Ok(DbOwner {
                profile: required,
                dropped: self.dropped,
            })
        }
    }

    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 ContinuationAdapter;

    impl StartupContinuationAdapter for ContinuationAdapter {
        type ContinuationOwner = Continuation;

        fn adapter_provenance(&self) -> [u8; 32] {
            [9; 32]
        }
    }

    struct SignalComponent {
        shutdown_error: bool,
    }

    struct TestBootstrapAdapter {
        mode: &'static str,
    }

    struct TestPreparedBootstrap {
        mode: &'static str,
    }

    impl BootstrapInstallAdapter<DbOwner, Continuation> for TestBootstrapAdapter {
        type Prepared = TestPreparedBootstrap;
        type Error = &'static str;

        fn prepare(
            self,
            _owners: BootstrapOwnerView<'_, DbOwner, Continuation>,
        ) -> Result<Self::Prepared, Self::Error> {
            match self.mode {
                "prepare-first-error" => Err("first"),
                "prepare-second-error" => Err("second"),
                "prepare-panic" => panic!("pre-commit panic must fail closed"),
                _ => Ok(TestPreparedBootstrap { mode: self.mode }),
            }
        }
    }

    impl PreparedBootstrapInstall<DbOwner, Continuation> for TestPreparedBootstrap {
        fn component_names(&self) -> &'static [&'static str] {
            &["observability", "database", "http1-production"]
        }

        fn install(
            self,
            owners: StartupBootstrapOwners<DbOwner, Continuation>,
            seal: BootstrapBatchSeal,
        ) -> PreparedBootstrapBatch {
            if self.mode == "commit-panic" {
                panic!("commit panic must fail closed");
            }
            let (database, db_domain, tokio_domain, allocation, ledger, _, bound) =
                owners.into_parts();
            drop((bound, database, db_domain, tokio_domain));
            let slot = seal.pending_driver_finalizer();
            slot.arm();
            slot.submit(crate::admission::test_startup_driver_finalizer(
                ledger, allocation,
            ));
            seal.install(vec![Arc::new(SignalComponent {
                shutdown_error: self.mode == "shutdown-error",
            })])
        }
    }

    impl ComponentLifecycle for SignalComponent {
        fn name(&self) -> &'static str {
            "runner-signal"
        }

        fn start(&self) -> LifecycleFuture<'_> {
            Box::pin(async {
                Command::new("sh")
                    .args(["-c", "kill -TERM $PPID"])
                    .status()
                    .unwrap();
                Ok(())
            })
        }

        fn shutdown(&self) -> LifecycleFuture<'_> {
            Box::pin(std::future::ready(if self.shutdown_error {
                Err(runner_error("test.shutdown_failed"))
            } else {
                Ok(())
            }))
        }
    }

    #[test]
    fn six_actual_owners_pair_only_after_real_construction() {
        let _profile_test = crate::admission::tests::OFFICIAL_TOKIO_PROFILE_TEST
            .lock()
            .unwrap();
        let dropped = Arc::new(AtomicBool::new(false));
        let mut owners = assemble_actual_startup_owners(
            crate::resource_envelope::tests::pending_plan(),
            DbFactory {
                drift: false,
                dropped: Arc::clone(&dropped),
            },
        )
        .unwrap();
        assert!(
            owners.runtime().handle().runtime_flavor()
                == tokio::runtime::RuntimeFlavor::MultiThread
        );
        assert!(owners.ledger().wait_snapshot().capacity > 0);
        assert!(owners.tokio_domain().snapshot().unwrap().task_capacity > 0);
        assert!(owners.db_domain().is_some());
        assert_ne!(
            owners
                .allocation()
                .snapshot()
                .unwrap()
                .approved_threshold_bytes,
            usize::MAX
        );
        let termination = owners.termination();
        let profile = owners.issue_transport_profile().unwrap();
        let (plan, generation, build, routes) = profile.identities();
        assert_ne!(plan, [0; 32]);
        assert_ne!(generation, 0);
        assert_ne!(build, [0; 32]);
        assert_ne!(routes, [0; 32]);
        let (head, attempt, finalization, task_storage) = profile.into_transport_parts();
        assert_eq!(head, termination.request_timeout());
        assert_eq!(attempt, termination.request_timeout());
        assert_eq!(finalization, termination.finalization_requirement());
        assert_ne!(task_storage, 0);
        assert_eq!(
            owners.issue_transport_profile().unwrap_err(),
            TransportRuntimeProfileError::DuplicateOrForeignClaim
        );
        assert!(!dropped.load(Ordering::Acquire));
        drop(owners);
        assert!(dropped.load(Ordering::Acquire));
    }

    #[test]
    fn db_actual_drift_drops_physical_owner_and_cannot_pair() {
        let _profile_test = crate::admission::tests::OFFICIAL_TOKIO_PROFILE_TEST
            .lock()
            .unwrap();
        let dropped = Arc::new(AtomicBool::new(false));
        let error = assemble_actual_startup_owners(
            crate::resource_envelope::tests::pending_plan(),
            DbFactory {
                drift: true,
                dropped: Arc::clone(&dropped),
            },
        )
        .err()
        .unwrap();
        assert!(matches!(
            error,
            StartupAssemblyError::Pairing(StartupPairingError::ActualFactsDrift)
        ));
        assert!(dropped.load(Ordering::Acquire));
    }

    #[test]
    fn transport_profile_rejects_foreign_generation_before_attach() {
        let _profile_test = crate::admission::tests::OFFICIAL_TOKIO_PROFILE_TEST
            .lock()
            .unwrap();
        let mut first = assemble_actual_startup_owners(
            crate::resource_envelope::tests::pending_plan(),
            DbFactory {
                drift: false,
                dropped: Arc::new(AtomicBool::new(false)),
            },
        )
        .unwrap();
        let foreign = first.paired.claim_transport_startup_profile().unwrap();
        drop(first);
        let second = assemble_actual_startup_owners(
            crate::resource_envelope::tests::pending_plan(),
            DbFactory {
                drift: false,
                dropped: Arc::new(AtomicBool::new(false)),
            },
        )
        .unwrap();
        assert_eq!(
            verify_transport_profile_parts(
                &second.tokio_domain,
                second.termination,
                second.paired.transport_binding(),
                foreign,
            )
            .unwrap_err(),
            TransportRuntimeProfileError::DuplicateOrForeignClaim
        );
    }

    #[test]
    fn official_runner_normal_failure_and_panic_are_finite_in_subprocesses() {
        const CHILD: &str = "startup_assembly::tests::official_runner_child";
        const DEADLINE: Duration = Duration::from_secs(10);

        for mode in [
            "normal",
            "prepare-first-error",
            "prepare-second-error",
            "shutdown-error",
            "prepare-panic",
            "commit-panic",
        ] {
            let mut child = Command::new(env::current_exe().unwrap())
                .args(["--exact", CHILD, "--nocapture"])
                .env("SADDLE_ACTUAL_OWNER_RUNNER_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() >= DEADLINE {
                    child.kill().unwrap();
                    child.wait().unwrap();
                    panic!("runner child {mode} exceeded {DEADLINE:?}");
                }
                std::thread::sleep(Duration::from_millis(10));
            };
            if matches!(mode, "prepare-panic" | "commit-panic") {
                assert_eq!(status.signal(), Some(6));
            } else {
                assert!(status.success(), "runner child {mode} failed: {status}");
            }
        }
    }

    #[test]
    fn official_runner_child() {
        let Some(mode) = env::var_os("SADDLE_ACTUAL_OWNER_RUNNER_CHILD") else {
            return;
        };
        let mode = match mode.to_str().unwrap() {
            "normal" => "normal",
            "prepare-first-error" => "prepare-first-error",
            "prepare-second-error" => "prepare-second-error",
            "shutdown-error" => "shutdown-error",
            "prepare-panic" => "prepare-panic",
            "commit-panic" => "commit-panic",
            other => panic!("unexpected runner mode {other}"),
        };
        let _profile_test = crate::admission::tests::OFFICIAL_TOKIO_PROFILE_TEST
            .lock()
            .unwrap();
        let dropped = Arc::new(AtomicBool::new(false));
        let actual = assemble_actual_startup_owners(
            crate::resource_envelope::tests::pending_plan(),
            DbFactory {
                drift: false,
                dropped: Arc::clone(&dropped),
            },
        )
        .unwrap();
        let (paired, claim, owners) = actual.into_pairing_parts();
        let result = run_with_actual_startup_owners(
            owners,
            paired,
            claim,
            ContinuationAdapter,
            Continuation,
            |owners| async move {
                let transaction = BootstrapTransaction::new(owners);
                match transaction.prepare(TestBootstrapAdapter { mode }) {
                    Ok(prepared) => Ok(prepared.commit()),
                    Err(failure) => Err(failure.into_runner_failure(runner_error(match mode {
                        "prepare-first-error" => "test.prepare_first_failed",
                        "prepare-second-error" => "test.prepare_second_failed",
                        _ => "test.prepare_failed",
                    }))),
                }
            },
        );
        match mode {
            "prepare-first-error" => {
                assert_eq!(result.unwrap_err().code(), "test.prepare_first_failed")
            }
            "prepare-second-error" => {
                assert_eq!(result.unwrap_err().code(), "test.prepare_second_failed")
            }
            "shutdown-error" => assert_eq!(result.unwrap_err().code(), "test.shutdown_failed"),
            _ => result.unwrap(),
        }
        assert!(dropped.load(Ordering::Acquire));
    }
}