saddle-runtime 0.3.0-alpha.2

//! 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,
    ObservabilityQueuePairAuthority, OfficialTokioDomain, PairedStartupPlan, PendingStartupPlan,
    PrepairObservabilityQueueDomainOwner, PreparedObservabilityQueueOwner,
    ProcessAllocationProfile, ProcessLedger, StartupActualFacts, StartupClaim, StartupClaimKind,
    StartupContinuationAdapter, StartupContinuationClaim, StartupPairingError,
    VerifiedStartupContinuationOwner, VerifiedTransportStartupClaim, bind_startup_continuation,
};
use saddle_core::{ComponentLifecycle, ErrorKind, Result as SaddleResult, SaddleError};
use saddle_observability::file::NormalizedSignedProviderFilesystemRuntimeEvidence;

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

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

    fn construct(
        self,
        required: DbCreditProfile,
    ) -> impl std::future::Future<Output = Result<Self::Owner, Self::Error>> + Send;
}

#[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<F> {
    database_factory: F,
    db_required: DbCreditProfile,
    tokio_domain: OfficialTokioDomain,
    runtime: tokio::runtime::Runtime,
    allocation: ProcessAllocationProfile,
    ledger: ProcessLedger,
    termination: DerivedTerminationPlan,
    transaction: Transaction,
    observability: StartupObservabilityPreparation,
}

#[doc(hidden)]
pub struct StartupObservabilityPreparation {
    physical: NormalizedSignedProviderFilesystemRuntimeEvidence,
    queue: PrepairObservabilityQueueDomainOwner,
    pair_authority: ObservabilityQueuePairAuthority,
}

#[doc(hidden)]
pub enum StartupObservabilityOwner {
    Prepared(StartupObservabilityPreparation),
    Started(Arc<crate::observability::StartupObservabilityOwner>),
}

#[doc(hidden)]
// The production physical owner remains inline and allocation-free. Boxing it
// would create an unaccounted startup allocation solely to optimize the
// evidence-only Disabled variant.
#[allow(clippy::large_enum_variant)]
pub enum StartupObservabilityInput {
    #[cfg(test)]
    Disabled,
    Production(NormalizedSignedProviderFilesystemRuntimeEvidence),
}

impl<F> ActualStartupOwners<F> {
    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 allocation(&self) -> &ProcessAllocationProfile {
        &self.allocation
    }
    pub fn termination(&self) -> DerivedTerminationPlan {
        self.termination
    }
}

/// 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,
}

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>,
    transport_binding: ([u8; 32], u64, [u8; 32], [u8; 32]),
    observability: Option<StartupObservabilityOwner>,
}

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>,
        ([u8; 32], u64, [u8; 32], [u8; 32]),
        Option<StartupObservabilityOwner>,
    ) {
        (
            self.database,
            self.db_domain,
            self.tokio_domain,
            self.allocation,
            self.ledger,
            self.termination,
            self.bound,
            self.transport_binding,
            self.observability,
        )
    }
}

#[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 {
    type Ready: ReadyBootstrapInstall<D, C>;
    type Error;

    fn component_names(&self) -> &'static [&'static str];
    fn prepare_install(
        &mut self,
    ) -> impl std::future::Future<Output = Result<(), Self::Error>> + Send;
    fn into_ready(self) -> Self::Ready;
}

#[doc(hidden)]
pub trait ReadyBootstrapInstall<D, C>: Sized {
    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 VerifiedPreparedBootstrapTransaction<D, C, P> {
    database: D,
    db_domain: Option<DbPermitDomain>,
    tokio_domain: OfficialTokioDomain,
    allocation: ProcessAllocationProfile,
    ledger: ProcessLedger,
    termination: DerivedTerminationPlan,
    pending: saddle_admission::PostDriverInstallPending,
    remainder: saddle_admission::PostDriverContinuationRemainder<C>,
    transport_binding: ([u8; 32], u64, [u8; 32], [u8; 32]),
    observability: Option<StartupObservabilityOwner>,
    prepared: P,
    receipt: BootstrapReservationReceipt,
    post_driver_binding: saddle_admission::VerifiedPostDriverInstallBinding,
}

#[doc(hidden)]
pub struct ReadyBootstrapTransaction<D, C, P> {
    batch: PreparedBootstrapBatch,
    post_driver_binding: saddle_admission::VerifiedPostDriverInstallBinding,
    _marker: std::marker::PhantomData<(D, C, P)>,
}

#[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,
    finalizer: Option<crate::post_driver::MustSubmitDriverFinalizer>,
}

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

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

impl BootstrapBatchSeal {
    pub fn take_driver_finalizer(&mut self) -> crate::post_driver::MustSubmitDriverFinalizer {
        self.finalizer
            .take()
            .unwrap_or_else(|| std::process::abort())
    }

    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 async fn prepare_observability(self) -> Result<Self, StartupBootstrapFailure<D, C>> {
        if matches!(
            self.owners.observability,
            Some(StartupObservabilityOwner::Started(_))
        ) {
            Ok(self)
        } else {
            std::process::abort()
        }
    }

    pub async fn prepare_install(&mut self) -> Result<(), P::Error> {
        self.prepared.prepare_install().await
    }

    pub fn verify_post_driver_binding(
        self,
    ) -> Result<
        VerifiedPreparedBootstrapTransaction<D, C, P>,
        BootstrapPrepareFailure<D, C, P::Error>,
    > {
        let PreparedBootstrapTransaction {
            owners,
            prepared,
            receipt,
        } = self;
        let StartupBootstrapOwners {
            database,
            db_domain,
            tokio_domain,
            allocation,
            ledger,
            termination,
            bound,
            transport_binding,
            observability,
        } = owners;
        let (pending, remainder) = match bound.into_post_driver_prepare_parts() {
            Ok(parts) => parts,
            Err(bound) => {
                return Err(BootstrapPrepareFailure {
                    owners: StartupBootstrapOwners {
                        database,
                        db_domain,
                        tokio_domain,
                        allocation,
                        ledger,
                        termination,
                        bound,
                        transport_binding,
                        observability,
                    },
                    error: BootstrapPrepareError::InvalidComponentSet,
                });
            }
        };
        let (post_driver_binding, pending) = match pending.verify_for_runtime_prepare() {
            Ok(parts) => parts,
            Err(pending) => {
                let bound =
                    BoundStartupContinuation::restore_after_post_driver_cancel(pending, remainder)
                        .unwrap_or_else(|_| std::process::abort());
                return Err(BootstrapPrepareFailure {
                    owners: StartupBootstrapOwners {
                        database,
                        db_domain,
                        tokio_domain,
                        allocation,
                        ledger,
                        termination,
                        bound,
                        transport_binding,
                        observability,
                    },
                    error: BootstrapPrepareError::InvalidComponentSet,
                });
            }
        };
        Ok(VerifiedPreparedBootstrapTransaction {
            database,
            db_domain,
            tokio_domain,
            allocation,
            ledger,
            termination,
            pending,
            remainder,
            transport_binding,
            observability,
            prepared,
            receipt,
            post_driver_binding,
        })
    }

    /// Rolls a fully prepared but unpublished transaction back through the
    /// official runner when the final external authorization fails.
    pub fn fail(self, error: SaddleError) -> StartupBootstrapFailure<D, C> {
        self.owners.fail(error)
    }
}

impl<D, C, P> VerifiedPreparedBootstrapTransaction<D, C, P>
where
    P: PreparedBootstrapInstall<D, C>,
{
    pub fn finish_prepare_install(self) -> ReadyBootstrapTransaction<D, C, P::Ready> {
        let Self {
            database,
            db_domain,
            tokio_domain,
            allocation,
            ledger,
            termination,
            pending,
            remainder,
            transport_binding,
            observability,
            prepared,
            receipt,
            post_driver_binding,
        } = self;
        let ready = prepared.into_ready();
        let bound = BoundStartupContinuation::restore_after_post_driver_cancel(pending, remainder)
            .unwrap_or_else(|_| std::process::abort());
        let application = Application::new();
        let finalizer = application.reserved_post_driver_submit();
        let batch = ready.install(
            StartupBootstrapOwners {
                database,
                db_domain,
                tokio_domain,
                allocation,
                ledger,
                termination,
                bound,
                transport_binding,
                observability,
            },
            BootstrapBatchSeal {
                _receipt: receipt,
                application,
                finalizer: Some(finalizer),
            },
        );
        ReadyBootstrapTransaction {
            batch,
            post_driver_binding,
            _marker: std::marker::PhantomData,
        }
    }

    pub fn fail(self, error: SaddleError) -> StartupBootstrapFailure<D, C> {
        let bound = BoundStartupContinuation::restore_after_post_driver_cancel(
            self.pending,
            self.remainder,
        )
        .unwrap_or_else(|_| std::process::abort());
        StartupBootstrapOwners {
            database: self.database,
            db_domain: self.db_domain,
            tokio_domain: self.tokio_domain,
            allocation: self.allocation,
            ledger: self.ledger,
            termination: self.termination,
            bound,
            transport_binding: self.transport_binding,
            observability: self.observability,
        }
        .fail(error)
    }
}

impl<D, C, P> ReadyBootstrapTransaction<D, C, P>
where
    P: ReadyBootstrapInstall<D, C>,
{
    pub fn commit(self) -> Application {
        self.batch
            .application
            .commit_post_driver_install(self.post_driver_binding);
        self.batch.application
    }
}

/// 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, P, F, Fut>(
    owners: ActualStartupOwners<P>,
    adapter: A,
    continuation: A::ContinuationOwner,
    bootstrap: F,
) -> SaddleResult<()>
where
    A: StartupContinuationAdapter,
    A::ContinuationOwner: VerifiedStartupContinuationOwner,
    P: StartupDbPoolFactory + Send + 'static,
    P::Owner: Send + 'static,
    P::Error: Send + 'static,
    F: FnOnce(
        StartupBootstrapOwners<P::Owner, A::ContinuationOwner>,
        VerifiedTransportRuntimeProfile,
    ) -> Fut,
    Fut: std::future::Future<
            Output = Result<Application, StartupBootstrapFailure<P::Owner, A::ContinuationOwner>>,
        >,
{
    Application::claim_process_runtime()?;
    let ActualStartupOwners {
        database_factory,
        db_required,
        tokio_domain,
        runtime,
        allocation,
        ledger,
        termination,
        transaction,
        observability,
    } = owners;

    let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
        runtime.block_on(async {
            let writer = match crate::observability::StartupObservabilityOwner::prepare(
                observability.physical,
                observability.queue,
                observability.pair_authority,
                termination.writer_shutdown_budget,
            )
            .await
            {
                Ok(writer) => Arc::new(writer),
                Err(_) => {
                    return Err(PreauthorizationFailure {
                        database: None,
                        db_domain: None,
                        tokio_domain,
                        allocation,
                        ledger,
                        transaction: Some(transaction),
                        writer: None,
                        error: runner_error("runtime.observability_prepared_start_failed"),
                    });
                }
            };
            let database = match database_factory.construct(db_required).await {
                Ok(database) => database,
                Err(_) => {
                    return Err(PreauthorizationFailure {
                        database: None,
                        db_domain: None,
                        tokio_domain,
                        allocation,
                        ledger,
                        transaction: Some(transaction),
                        writer: Some(writer),
                        error: runner_error("runtime.database_prepared_start_failed"),
                    });
                }
            };
            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 {
                match ledger.prepare_db_domain(actual_db) {
                    Ok(domain) => Some(domain),
                    Err(_) => {
                        return Err(PreauthorizationFailure {
                            database: Some(database),
                            db_domain: None,
                            tokio_domain,
                            allocation,
                            ledger,
                            transaction: Some(transaction),
                            writer: Some(writer),
                            error: runner_error("runtime.database_domain_prepare_failed"),
                        });
                    }
                }
            };
            let mut transaction = transaction;
            if transaction
                .complete(
                    StartupClaimKind::DbPool,
                    StartupActualFacts::DbPool {
                        connections: actual_db.connections,
                        operations: actual_db.operations,
                    },
                )
                .is_err()
            {
                return Err(PreauthorizationFailure {
                    database: Some(database),
                    db_domain,
                    tokio_domain,
                    allocation,
                    ledger,
                    transaction: Some(transaction),
                    writer: Some(writer),
                    error: runner_error("runtime.database_receipt_failed"),
                });
            }
            let authority = match writer.take_pair_authority() {
                Ok(authority) => authority,
                Err(_) => {
                    return Err(PreauthorizationFailure {
                        database: Some(database),
                        db_domain,
                        tokio_domain,
                        allocation,
                        ledger,
                        transaction: Some(transaction),
                        writer: Some(writer),
                        error: runner_error("runtime.observability_health_failed"),
                    });
                }
            };
            let (mut paired, continuation_claim) =
                match transaction.pair_with_observability_queue(authority) {
                    Ok(parts) => parts,
                    Err((_error, transaction)) => {
                        return Err(PreauthorizationFailure {
                            database: Some(database),
                            db_domain,
                            tokio_domain,
                            allocation,
                            ledger,
                            transaction: Some(transaction),
                            writer: Some(writer),
                            error: runner_error("runtime.startup_pair_failed"),
                        });
                    }
                };
            let transport_binding = paired.transport_binding();
            let transport_claim = match paired.claim_transport_startup_profile() {
                Ok(claim) => claim,
                Err(_) => {
                    return Err(PreauthorizationFailure {
                        database: Some(database),
                        db_domain,
                        tokio_domain,
                        allocation,
                        ledger,
                        transaction: None,
                        writer: Some(writer),
                        error: runner_error("runtime.transport_profile_claim_failed"),
                    });
                }
            };
            let runtime_profile = match verify_transport_profile_parts(
                &tokio_domain,
                termination,
                transport_binding,
                transport_claim,
            ) {
                Ok(profile) => profile,
                Err(_) => {
                    return Err(PreauthorizationFailure {
                        database: Some(database),
                        db_domain,
                        tokio_domain,
                        allocation,
                        ledger,
                        transaction: None,
                        writer: Some(writer),
                        error: runner_error("runtime.transport_profile_verify_failed"),
                    });
                }
            };
            let bound = match bind_startup_continuation(
                paired,
                continuation_claim,
                adapter,
                continuation,
            ) {
                Ok(bound) => bound,
                Err(_) => {
                    return Err(PreauthorizationFailure {
                        database: Some(database),
                        db_domain,
                        tokio_domain,
                        allocation,
                        ledger,
                        transaction: None,
                        writer: Some(writer),
                        error: runner_error("runtime.startup_continuation_bind_failed"),
                    });
                }
            };
            let bootstrap_owners = StartupBootstrapOwners {
                database,
                db_domain,
                tokio_domain,
                allocation,
                ledger,
                termination,
                bound,
                transport_binding,
                observability: Some(StartupObservabilityOwner::Started(writer)),
            };
            let signal = match ShutdownSignal::register() {
                Ok(signal) => signal,
                Err(error) => {
                    return Err(PreauthorizationFailure::from_bootstrap(
                        bootstrap_owners,
                        error,
                    ));
                }
            };
            let application = bootstrap(bootstrap_owners, runtime_profile)
                .await
                .map_err(PreauthorizationFailure::from_bootstrap_failure)?;
            let finalizer = application.pending_driver_finalizer();
            let result = application.run_until_shutdown(signal.wait()).await;
            Ok((finalizer, result))
        })
    }));
    match outcome {
        Ok(Ok((finalizer, result))) => finalizer.finish(runtime, result),
        Ok(Err(failure)) => failure.finish(runtime),
        Err(_) => std::process::abort(),
    }
}

struct PreauthorizationFailure<D> {
    database: Option<D>,
    db_domain: Option<DbPermitDomain>,
    tokio_domain: OfficialTokioDomain,
    allocation: ProcessAllocationProfile,
    ledger: ProcessLedger,
    transaction: Option<Transaction>,
    writer: Option<Arc<crate::observability::StartupObservabilityOwner>>,
    error: SaddleError,
}

impl<D> PreauthorizationFailure<D> {
    fn from_bootstrap<C>(owners: StartupBootstrapOwners<D, C>, error: SaddleError) -> Self {
        let StartupBootstrapOwners {
            database,
            db_domain,
            tokio_domain,
            allocation,
            ledger,
            termination: _,
            bound: _,
            transport_binding: _,
            observability,
        } = owners;
        let writer = match observability {
            Some(StartupObservabilityOwner::Started(writer)) => Some(writer),
            _ => None,
        };
        Self {
            database: Some(database),
            db_domain,
            tokio_domain,
            allocation,
            ledger,
            transaction: None,
            writer,
            error,
        }
    }

    fn from_bootstrap_failure<C>(failure: StartupBootstrapFailure<D, C>) -> Self {
        Self::from_bootstrap(failure.owners, failure.error)
    }

    fn finish(self, runtime: tokio::runtime::Runtime) -> SaddleResult<()> {
        if let Some(writer) = &self.writer {
            if runtime.block_on(writer.shutdown_prepared()).is_err() {
                return Err(runner_error("runtime.observability_rollback_failed"));
            }
        }
        drop(self.writer);
        drop(self.transaction);
        drop((self.database, self.db_domain, self.tokio_domain));
        drop(runtime);
        let allocation_result = self.allocation.finish();
        let ledger_result = self.ledger.try_shutdown();
        if allocation_result.is_err() || ledger_result.is_err() {
            Err(runner_error("runtime.preauthorization_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 prepare_observability_queue(
        &mut self,
        ledger: &ProcessLedger,
    ) -> Result<PreparedObservabilityQueueOwner, StartupPairingError> {
        let index = self
            .active
            .iter()
            .position(|claim| claim.kind() == StartupClaimKind::ObservabilityQueue)
            .expect("observability queue claim is active");
        let claim = self.active.swap_remove(index);
        claim.prepare_observability_queue(ledger)
    }

    // The inline Transaction is returned intact for allocation-free rollback.
    #[allow(clippy::result_large_err)]
    fn pair_with_observability_queue(
        mut self,
        authority: ObservabilityQueuePairAuthority,
    ) -> Result<(PairedStartupPlan, StartupContinuationClaim), (StartupPairingError, Self)> {
        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 = match pending.continuation_claim() {
            Ok(continuation) => continuation,
            Err(error) => {
                self.pending = Some(pending);
                self.completed = Some(completed);
                return Err((error, self));
            }
        };
        let paired = match pending.pair_with_observability_queue(completed, authority) {
            Ok(paired) => paired,
            Err((error, mut pending, mut completed, authority)) => {
                if pending
                    .restore_observability_pair_authority(&mut completed, authority)
                    .is_err()
                {
                    std::process::abort();
                }
                self.pending = Some(pending);
                self.completed = Some(completed);
                return Err((error, self));
            }
        };
        Ok((paired, 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,
    observability: StartupObservabilityInput,
) -> Result<ActualStartupOwners<F>, 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 (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 observability = match observability {
        #[cfg(test)]
        StartupObservabilityInput::Disabled => {
            return Err(StartupAssemblyError::InvalidRequirement);
        }
        StartupObservabilityInput::Production(physical) => {
            tx.requirement(StartupClaimKind::ObservabilityQueue)?;
            let queue = tx.prepare_observability_queue(&ledger)?;
            let (queue, pair_authority) = queue.into_prepair_parts();
            StartupObservabilityPreparation {
                physical,
                queue,
                pair_authority,
            }
        }
    };
    Ok(ActualStartupOwners {
        database_factory,
        db_required,
        tokio_domain,
        runtime,
        allocation,
        ledger,
        termination,
        transaction: tx,
        observability,
    })
}

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,
        future::{Future, poll_fn},
        pin::Pin,
        process::{Command, Stdio},
        sync::{
            Arc,
            atomic::{AtomicBool, Ordering},
        },
        task::{Context, Poll, Wake, Waker},
        time::Instant,
    };

    use super::*;

    struct BorrowedPendingInstall {
        ready: Arc<AtomicBool>,
        panic_on_poll: Arc<AtomicBool>,
        listener_bound: Arc<AtomicBool>,
        authorized: Arc<AtomicBool>,
        published: Arc<AtomicBool>,
        spawned: Arc<AtomicBool>,
    }

    struct BorrowedReadyInstall;

    impl ReadyBootstrapInstall<(), ()> for BorrowedReadyInstall {
        fn install(
            self,
            _owners: StartupBootstrapOwners<(), ()>,
            _seal: BootstrapBatchSeal,
        ) -> PreparedBootstrapBatch {
            unreachable!("the borrow/cancel contract test never installs components")
        }
    }

    impl PreparedBootstrapInstall<(), ()> for BorrowedPendingInstall {
        type Ready = BorrowedReadyInstall;
        type Error = ();

        fn component_names(&self) -> &'static [&'static str] {
            &["observability", "database", "http1-production"]
        }

        async fn prepare_install(&mut self) -> Result<(), Self::Error> {
            poll_fn(|_| {
                self.listener_bound.store(true, Ordering::Release);
                assert!(!self.authorized.load(Ordering::Acquire));
                assert!(!self.published.load(Ordering::Acquire));
                assert!(!self.spawned.load(Ordering::Acquire));
                assert!(
                    !self.panic_on_poll.load(Ordering::Acquire),
                    "deterministic borrowed prepare panic"
                );
                if self.ready.load(Ordering::Acquire) {
                    Poll::Ready(Ok(()))
                } else {
                    Poll::Pending
                }
            })
            .await
        }

        fn into_ready(self) -> Self::Ready {
            BorrowedReadyInstall
        }
    }

    struct NoopWake;

    impl Wake for NoopWake {
        fn wake(self: Arc<Self>) {}
    }

    fn poll_once<F: Future>(future: Pin<&mut F>) -> Poll<F::Output> {
        let waker = Waker::from(Arc::new(NoopWake));
        future.poll(&mut Context::from_waker(&waker))
    }

    #[test]
    fn borrowed_prepare_pending_drop_recovers_outer_owners_and_retries() {
        let ready = Arc::new(AtomicBool::new(false));
        let panic_on_poll = Arc::new(AtomicBool::new(false));
        let listener_bound = Arc::new(AtomicBool::new(false));
        let authorized = Arc::new(AtomicBool::new(false));
        let published = Arc::new(AtomicBool::new(false));
        let spawned = Arc::new(AtomicBool::new(false));
        let application = Application::new();
        let pending_owner = Box::new(17_u8);
        let startup_owner = Box::new(23_u8);
        let mut prepared = BorrowedPendingInstall {
            ready: Arc::clone(&ready),
            panic_on_poll: Arc::clone(&panic_on_poll),
            listener_bound: Arc::clone(&listener_bound),
            authorized: Arc::clone(&authorized),
            published: Arc::clone(&published),
            spawned: Arc::clone(&spawned),
        };

        let mut future = Box::pin(prepared.prepare_install());
        assert!(poll_once(future.as_mut()).is_pending());
        assert!(application.post_driver_is_unarmed_for_test());
        assert!(listener_bound.load(Ordering::Acquire));
        assert!(!authorized.load(Ordering::Acquire));
        assert!(!published.load(Ordering::Acquire));
        assert!(!spawned.load(Ordering::Acquire));
        drop(future);

        // The outer transaction still owns every linear value after cancelling
        // the borrowed Future, so it can recover and retry the original set.
        let recovered = (pending_owner, application, startup_owner, prepared);
        let (pending_owner, application, startup_owner, mut prepared) = recovered;
        assert_eq!(*pending_owner, 17);
        assert_eq!(*startup_owner, 23);
        assert!(application.post_driver_is_unarmed_for_test());

        ready.store(true, Ordering::Release);
        let mut retry = Box::pin(prepared.prepare_install());
        assert!(poll_once(retry.as_mut()).is_ready());
        drop(retry);
        let _ready_install = prepared.into_ready();

        // Unwind while the borrowed Future is Pending has the same ownership
        // property: no linear outer value was moved into that Future.
        panic_on_poll.store(true, Ordering::Release);
        ready.store(false, Ordering::Release);
        let mut panicking = BorrowedPendingInstall {
            ready,
            panic_on_poll,
            listener_bound,
            authorized,
            published,
            spawned,
        };
        let unwind = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            let mut future = Box::pin(panicking.prepare_install());
            let _ = poll_once(future.as_mut());
        }));
        assert!(unwind.is_err());
        assert!(application.post_driver_is_unarmed_for_test());
        let _still_owned_after_unwind = (&mut panicking, pending_owner, startup_owner);
    }

    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 = ();

        async fn construct(
            self,
            mut required: DbCreditProfile,
        ) -> Result<Self::Owner, Self::Error> {
            if self.drift {
                required.connections = required.connections.saturating_sub(1);
            }
            Ok(DbOwner {
                profile: required,
                dropped: self.dropped,
            })
        }
    }

    #[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 error = assemble_actual_startup_owners(
            crate::resource_envelope::tests::pending_plan(),
            DbFactory {
                drift: false,
                dropped: Arc::clone(&dropped),
            },
            StartupObservabilityInput::Disabled,
        )
        .err()
        .unwrap();
        assert!(matches!(error, StartupAssemblyError::InvalidRequirement));
        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),
            },
            StartupObservabilityInput::Disabled,
        )
        .err()
        .unwrap();
        assert!(matches!(error, StartupAssemblyError::InvalidRequirement));
        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 first = assemble_actual_startup_owners(
            crate::resource_envelope::tests::pending_plan(),
            DbFactory {
                drift: false,
                dropped: Arc::new(AtomicBool::new(false)),
            },
            StartupObservabilityInput::Disabled,
        )
        .err()
        .unwrap();
        assert!(matches!(first, StartupAssemblyError::InvalidRequirement));
    }

    #[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 ["disabled-input", "disabled-input-repeat"] {
            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));
            };
            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 = mode.to_str().unwrap();
        assert!(matches!(mode, "disabled-input" | "disabled-input-repeat"));
        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: false,
                dropped: Arc::clone(&dropped),
            },
            StartupObservabilityInput::Disabled,
        )
        .err()
        .unwrap();
        assert!(matches!(error, StartupAssemblyError::InvalidRequirement));
        assert!(!dropped.load(Ordering::Acquire));
    }
}