saddle-runtime 0.2.0-rc.5

//! Framework-internal C4 Transport ingress capability.
//!
//! This module is public only because Rust has no cross-crate friend
//! visibility. The Saddle facade owns construction and may pass the narrow
//! capability to its Transport integration. It is not a stable or business
//! API and is intentionally not re-exported by the facade.

use std::{
    future::Future,
    hint::spin_loop,
    marker::PhantomData,
    pin::Pin,
    sync::{
        Mutex, MutexGuard, TryLockError,
        atomic::{AtomicBool, Ordering},
    },
    task::{Context, Poll},
};

use saddle_admission::{
    AdmissionError, AttemptOutcome as AdmissionOutcome, AuditReport, ManagedBytes, ProcessLedger,
    RequestEnvelope, RequestMemory, ResourceConfig, WaitRegistration, WaitRemoval,
};
use tokio::task::{JoinError, JoinHandle};

use crate::{
    RequestLifecycle,
    request::{RequestClaim, RequestGuard},
};

const TASK_STORAGE_BOUND: usize = 384;
type TaskOutput = Result<AuditReport, AdmissionError>;

/// Startup-only, non-stable sizing input for the internal capability.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct CapabilityConfig {
    pub managed_capacity: usize,
    pub entry_reserve: usize,
    pub max_active_requests: usize,
    pub task_registry_capacity: usize,
    pub wait_capacity: usize,
    pub ingress_capacity: usize,
    pub system_estimate: usize,
    pub safety_margin: usize,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum CapabilityBuildError {
    InvalidConfiguration,
    SizeOverflow,
    TaskLayoutTooLarge,
}

/// Fixed non-owning identity returned after successful execution publication.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct RequestIdentity {
    slot: usize,
    generation: u64,
}

/// Fixed one-shot retry token. Its Admission identity remains private.
#[derive(Eq, PartialEq)]
pub struct RetryToken {
    registration: WaitRegistration,
}

/// Linear owner retained by Transport only while Admission is waiting.
///
/// Private fields prevent construction, registration substitution or access
/// to the underlying Admission identity.
pub struct WaitingBody<B> {
    registration: WaitRegistration,
    body: B,
}

impl<B> WaitingBody<B> {
    pub fn is_retry_for(&self, token: &RetryToken) -> bool {
        self.registration == token.registration
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum RejectReason {
    IngressFull,
    RegistryFull,
    Admission,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum StopReason {
    Shutdown,
    RuntimeUnavailable,
    Admission,
    InvalidRetry,
}

/// The only outcomes visible to the facade-owned Transport integration.
pub enum AttemptOutcome<B> {
    Accepted(RequestIdentity),
    Registered(WaitingBody<B>),
    Reject(RejectReason, B),
    Stop(StopReason, B),
}

pub struct CancelOutcome<B> {
    body: B,
    removed: bool,
}

impl<B> CancelOutcome<B> {
    pub fn was_registered(&self) -> bool {
        self.removed
    }

    pub fn into_body(self) -> B {
        self.body
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum CompletionOutcome {
    Completed,
    Failed,
    Stale,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct ShutdownOutcome {
    pub healthy: bool,
    pub task_failed: bool,
}

/// Narrow construction-time access to the admitted request account.
///
/// It cannot be stored in the returned `'static` execution future and does
/// not expose the account, envelope, ledger, domain or lifecycle guard.
pub struct RequestResources<'a> {
    memory: &'a RequestMemory,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ResourceError {
    Rejected,
}

impl RequestResources<'_> {
    pub fn try_bytes(&self, value: &[u8]) -> Result<ManagedBytes, ResourceError> {
        self.memory
            .try_bytes(value)
            .map_err(|_| ResourceError::Rejected)
    }
}

struct PublishedTask {
    generation: u64,
    handle: JoinHandle<TaskOutput>,
    lifecycle: RequestGuard,
}

struct CapabilityState {
    requests: RequestLifecycle,
    ledger: Option<ProcessLedger>,
    published: Box<[Option<PublishedTask>]>,
    generations: Box<[u64]>,
}

/// Facade-owned internal capability around the unique C4 coordinator.
///
/// `B` is Transport's owned body source, `E` is the facade-owned execution
/// factory and `F` is the single managed future shape. No executor, envelope
/// or general spawn operation is exposed.
pub struct TransportCapability<B, E, F> {
    ingress: Box<[AtomicBool]>,
    state: Mutex<CapabilityState>,
    executor: E,
    closed: AtomicBool,
    shutdown_complete: AtomicBool,
    marker: PhantomData<fn(B) -> F>,
}

impl<B, E, F> TransportCapability<B, E, F>
where
    B: Send + 'static,
    E: for<'a> Fn(B, RequestResources<'a>) -> F + Send + Sync + 'static,
    F: Future + Send + 'static,
{
    pub fn try_new(config: CapabilityConfig, executor: E) -> Result<Self, CapabilityBuildError> {
        if config.max_active_requests == 0
            || config.wait_capacity == 0
            || config.ingress_capacity == 0
            || config.task_registry_capacity
                < config
                    .max_active_requests
                    .checked_add(1)
                    .ok_or(CapabilityBuildError::SizeOverflow)?
            || config.managed_capacity == 0
            || config.entry_reserve == 0
            || config.system_estimate == 0
            || config.safety_margin == 0
        {
            return Err(CapabilityBuildError::InvalidConfiguration);
        }
        if std::mem::size_of::<RequestEnvelope<F>>() > TASK_STORAGE_BOUND {
            return Err(CapabilityBuildError::TaskLayoutTooLarge);
        }

        let accounts = config.max_active_requests;
        let framework_reserve = accounts
            .checked_mul(std::mem::size_of::<F>())
            .ok_or(CapabilityBuildError::SizeOverflow)?;
        let task_reserve = config
            .task_registry_capacity
            .checked_mul(TASK_STORAGE_BOUND)
            .ok_or(CapabilityBuildError::SizeOverflow)?;
        let process_state_reserve = ProcessLedger::minimum_process_state_reserve_with_waiters(
            accounts,
            config.wait_capacity,
        )
        .map_err(map_build_error)?;
        let process_limit = [
            config.managed_capacity,
            config.entry_reserve,
            framework_reserve,
            task_reserve,
            process_state_reserve,
            config.system_estimate,
            config.safety_margin,
        ]
        .into_iter()
        .try_fold(0_usize, usize::checked_add)
        .ok_or(CapabilityBuildError::SizeOverflow)?;
        let ledger = ProcessLedger::new_with_waiters(
            ResourceConfig {
                managed_capacity: config.managed_capacity,
                entry_reserve: config.entry_reserve,
                framework_reserve,
                task_reserve,
                process_state_reserve,
                system_estimate: config.system_estimate,
                safety_margin: config.safety_margin,
                process_limit,
                max_active_requests: accounts,
            },
            config.wait_capacity,
        )
        .map_err(map_build_error)?;
        let requests = RequestLifecycle::new();
        requests.mark_ready();

        Ok(Self {
            ingress: (0..config.ingress_capacity)
                .map(|_| AtomicBool::new(false))
                .collect(),
            state: Mutex::new(CapabilityState {
                requests,
                ledger: Some(ledger),
                published: (0..config.task_registry_capacity).map(|_| None).collect(),
                generations: vec![0; config.task_registry_capacity].into_boxed_slice(),
            }),
            executor,
            closed: AtomicBool::new(false),
            shutdown_complete: AtomicBool::new(false),
            marker: PhantomData,
        })
    }

    pub fn attempt(&self, managed_limit: usize, body: B) -> AttemptOutcome<B> {
        self.attempt_inner(managed_limit, body)
    }

    pub fn claim_retry(&self) -> Option<RetryToken> {
        if self.closed.load(Ordering::Acquire) {
            return None;
        }
        let _ingress = self.enter_control();
        let state = self.spin_state();
        if self.closed.load(Ordering::Acquire) {
            state
                .ledger
                .as_ref()
                .expect("ledger exists until consuming shutdown")
                .stop_waiting();
            return None;
        }
        state
            .ledger
            .as_ref()
            .expect("ledger exists until consuming shutdown")
            .claim_retry_hint()
            .map(|hint| RetryToken {
                registration: hint.registration(),
            })
    }

    pub fn retry(
        &self,
        managed_limit: usize,
        waiting: WaitingBody<B>,
        token: RetryToken,
    ) -> AttemptOutcome<B> {
        if waiting.registration != token.registration {
            let state = self.spin_state();
            state
                .ledger
                .as_ref()
                .expect("ledger exists until consuming shutdown")
                .stop_waiting();
            return AttemptOutcome::Stop(StopReason::InvalidRetry, waiting.body);
        }
        self.attempt_inner(managed_limit, waiting.body)
    }

    pub fn cancel(&self, waiting: WaitingBody<B>) -> CancelOutcome<B> {
        let _ingress = self.enter_control();
        let state = self.spin_state();
        let removed = state
            .ledger
            .as_ref()
            .expect("ledger exists until consuming shutdown")
            .cancel_wait(waiting.registration)
            == WaitRemoval::Removed;
        CancelOutcome {
            body: waiting.body,
            removed,
        }
    }

    pub fn complete(&self, identity: RequestIdentity) -> Completion<'_, B, E, F> {
        Completion {
            capability: self,
            identity,
            finished: false,
        }
    }

    pub fn begin_shutdown(&self) -> usize {
        self.closed.store(true, Ordering::Release);
        let _ingress = self.enter_control();
        let state = self.spin_state();
        state.requests.begin_draining();
        state
            .ledger
            .as_ref()
            .expect("ledger exists until consuming shutdown")
            .stop_waiting()
    }

    pub async fn shutdown(mut self, cancel: bool) -> ShutdownOutcome {
        self.begin_shutdown();
        let state = self
            .state
            .get_mut()
            .unwrap_or_else(|_| std::process::abort());
        if cancel {
            for task in state.published.iter().flatten() {
                task.handle.abort();
            }
        }

        let mut task_failed = false;
        for slot in &mut state.published {
            if let Some(task) = slot.take() {
                task_failed |= task_result(task.handle.await) != CompletionOutcome::Completed;
                drop(task.lifecycle);
            }
        }
        state.requests.wait_until_drained().await;
        state.requests.mark_stopped();
        let ledger = state
            .ledger
            .take()
            .expect("shutdown consumes the final ledger owner");
        let healthy = ledger
            .try_shutdown()
            .map(|snapshot| snapshot.healthy)
            .unwrap_or(false);
        self.shutdown_complete.store(true, Ordering::Release);
        ShutdownOutcome {
            healthy,
            task_failed,
        }
    }

    fn attempt_inner(&self, managed_limit: usize, body: B) -> AttemptOutcome<B> {
        if self.closed.load(Ordering::Acquire) {
            return AttemptOutcome::Stop(StopReason::Shutdown, body);
        }
        if tokio::runtime::Handle::try_current().is_err() {
            return AttemptOutcome::Stop(StopReason::RuntimeUnavailable, body);
        }
        let Some(_ingress) = self.try_enter() else {
            return AttemptOutcome::Reject(RejectReason::IngressFull, body);
        };
        let mut state = self.spin_state();
        if self.closed.load(Ordering::Acquire) {
            return AttemptOutcome::Stop(StopReason::Shutdown, body);
        }
        let Some(publish_slot) = state.published.iter().position(Option::is_none) else {
            return AttemptOutcome::Reject(RejectReason::RegistryFull, body);
        };
        let claim = match state.requests.try_claim() {
            Ok(claim) => claim,
            Err(_) => return AttemptOutcome::Stop(StopReason::Shutdown, body),
        };
        let mut body = Some(body);
        let ledger = state
            .ledger
            .as_ref()
            .expect("ledger exists until consuming shutdown");
        let admission = ledger.attempt_or_register(managed_limit, TASK_STORAGE_BOUND, |memory| {
            (self.executor)(
                body.take().expect("factory runs at most once"),
                RequestResources { memory },
            )
        });
        match admission {
            AdmissionOutcome::Ready(envelope) => {
                let generation = state.generations[publish_slot]
                    .checked_add(1)
                    .unwrap_or_else(|| std::process::abort());
                state.generations[publish_slot] = generation;
                let lifecycle = RequestClaim::publish(claim);
                let handle = tokio::spawn(envelope);
                state.published[publish_slot] = Some(PublishedTask {
                    generation,
                    handle,
                    lifecycle,
                });
                AttemptOutcome::Accepted(RequestIdentity {
                    slot: publish_slot,
                    generation,
                })
            }
            AdmissionOutcome::Registered(registration) => {
                drop(claim);
                AttemptOutcome::Registered(WaitingBody {
                    registration,
                    body: body.expect("registration does not invoke factory"),
                })
            }
            AdmissionOutcome::Reject(_) => {
                drop(claim);
                AttemptOutcome::Reject(
                    RejectReason::Admission,
                    body.expect("reject does not invoke factory"),
                )
            }
            AdmissionOutcome::Stop(_) => {
                drop(claim);
                ledger.stop_waiting();
                AttemptOutcome::Stop(
                    StopReason::Admission,
                    body.unwrap_or_else(|| std::process::abort()),
                )
            }
        }
    }

    fn try_enter(&self) -> Option<IngressGuard<'_, B, E, F>> {
        self.ingress
            .iter()
            .enumerate()
            .find_map(|(slot, occupied)| {
                occupied
                    .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
                    .ok()
                    .map(|_| IngressGuard {
                        capability: self,
                        slot,
                    })
            })
    }

    fn enter_control(&self) -> IngressGuard<'_, B, E, F> {
        loop {
            if let Some(guard) = self.try_enter() {
                return guard;
            }
            spin_loop();
        }
    }

    fn spin_state(&self) -> MutexGuard<'_, CapabilityState> {
        loop {
            match self.state.try_lock() {
                Ok(state) => return state,
                Err(TryLockError::WouldBlock) => spin_loop(),
                Err(TryLockError::Poisoned(_)) => std::process::abort(),
            }
        }
    }
}

impl<B, E, F> Drop for TransportCapability<B, E, F> {
    fn drop(&mut self) {
        if !self.shutdown_complete.load(Ordering::Acquire) {
            std::process::abort();
        }
    }
}

pub struct Completion<'a, B, E, F> {
    capability: &'a TransportCapability<B, E, F>,
    identity: RequestIdentity,
    finished: bool,
}

impl<B, E, F> Future for Completion<'_, B, E, F>
where
    B: Send + 'static,
    E: for<'a> Fn(B, RequestResources<'a>) -> F + Send + Sync + 'static,
    F: Future + Send + 'static,
{
    type Output = CompletionOutcome;

    fn poll(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Self::Output> {
        if self.finished {
            return Poll::Ready(CompletionOutcome::Stale);
        }
        let outcome = {
            let mut state = self.capability.spin_state();
            let Some(slot) = state.published.get_mut(self.identity.slot) else {
                self.finished = true;
                return Poll::Ready(CompletionOutcome::Stale);
            };
            let Some(task) = slot.as_mut() else {
                self.finished = true;
                return Poll::Ready(CompletionOutcome::Stale);
            };
            if task.generation != self.identity.generation {
                self.finished = true;
                return Poll::Ready(CompletionOutcome::Stale);
            }
            match Pin::new(&mut task.handle).poll(context) {
                Poll::Pending => return Poll::Pending,
                Poll::Ready(result) => {
                    let task = slot.take().expect("polled task remains in its slot");
                    drop(task.lifecycle);
                    task_result(result)
                }
            }
        };
        self.finished = true;
        Poll::Ready(outcome)
    }
}

struct IngressGuard<'a, B, E, F> {
    capability: &'a TransportCapability<B, E, F>,
    slot: usize,
}

impl<B, E, F> Drop for IngressGuard<'_, B, E, F> {
    fn drop(&mut self) {
        self.capability.ingress[self.slot].store(false, Ordering::Release);
    }
}

fn task_result(result: Result<TaskOutput, JoinError>) -> CompletionOutcome {
    match result {
        Ok(Ok(_)) => CompletionOutcome::Completed,
        Ok(Err(_)) | Err(_) => CompletionOutcome::Failed,
    }
}

fn map_build_error(error: AdmissionError) -> CapabilityBuildError {
    match error {
        AdmissionError::SizeOverflow => CapabilityBuildError::SizeOverflow,
        _ => CapabilityBuildError::InvalidConfiguration,
    }
}

#[cfg(test)]
mod tests {
    use std::{
        env,
        os::unix::process::ExitStatusExt,
        pin::Pin,
        process::{Command, Stdio},
        sync::{
            Arc, Barrier,
            atomic::{AtomicUsize, Ordering},
        },
        task::{Context, Poll},
        thread,
        time::{Duration, Instant},
    };

    use super::*;

    #[derive(Debug)]
    struct Body {
        drops: Arc<AtomicUsize>,
        executions: Arc<AtomicUsize>,
    }

    impl Drop for Body {
        fn drop(&mut self) {
            self.drops.fetch_add(1, Ordering::SeqCst);
        }
    }

    struct BodyFuture {
        body: Option<Body>,
        pending: bool,
    }

    struct NeverFuture {
        body: Option<Body>,
    }

    impl Future for NeverFuture {
        type Output = ();

        fn poll(self: Pin<&mut Self>, _context: &mut Context<'_>) -> Poll<()> {
            Poll::Pending
        }
    }

    impl Drop for NeverFuture {
        fn drop(&mut self) {
            drop(self.body.take());
        }
    }

    impl Future for BodyFuture {
        type Output = ();

        fn poll(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<()> {
            if self.pending {
                self.pending = false;
                context.waker().wake_by_ref();
                return Poll::Pending;
            }
            let body = self.body.take().expect("body executes exactly once");
            body.executions.fetch_add(1, Ordering::SeqCst);
            drop(body);
            Poll::Ready(())
        }
    }

    fn config(accounts: usize, waiters: usize, ingress: usize) -> CapabilityConfig {
        CapabilityConfig {
            managed_capacity: accounts * 8,
            entry_reserve: 64,
            max_active_requests: accounts,
            task_registry_capacity: accounts + 1,
            wait_capacity: waiters,
            ingress_capacity: ingress,
            system_estimate: 4_096,
            safety_margin: 2_048,
        }
    }

    fn body(drops: &Arc<AtomicUsize>, executions: &Arc<AtomicUsize>) -> Body {
        Body {
            drops: Arc::clone(drops),
            executions: Arc::clone(executions),
        }
    }

    fn capability(
        accounts: usize,
        waiters: usize,
        ingress: usize,
    ) -> TransportCapability<Body, impl Fn(Body, RequestResources<'_>) -> BodyFuture, BodyFuture>
    {
        TransportCapability::try_new(
            config(accounts, waiters, ingress),
            |body: Body, _resources: RequestResources<'_>| BodyFuture {
                body: Some(body),
                pending: true,
            },
        )
        .unwrap()
    }

    #[tokio::test]
    async fn accepted_publishes_body_to_the_only_managed_execution_path() {
        let drops = Arc::new(AtomicUsize::new(0));
        let executions = Arc::new(AtomicUsize::new(0));
        let capability = capability(1, 1, 1);
        let identity = match capability.attempt(8, body(&drops, &executions)) {
            AttemptOutcome::Accepted(identity) => identity,
            _ => panic!("capacity must accept"),
        };
        assert_eq!(executions.load(Ordering::SeqCst), 0);
        assert_eq!(drops.load(Ordering::SeqCst), 0);
        assert_eq!(
            capability.complete(identity).await,
            CompletionOutcome::Completed
        );
        assert_eq!(
            capability.complete(identity).await,
            CompletionOutcome::Stale
        );
        assert_eq!(drops.load(Ordering::SeqCst), 1);
        let shutdown = capability.shutdown(false).await;
        assert!(shutdown.healthy);
        assert!(!shutdown.task_failed);
    }

    #[tokio::test]
    async fn finished_but_unreaped_task_keeps_the_fixed_registry_slot() {
        let drops = Arc::new(AtomicUsize::new(0));
        let executions = Arc::new(AtomicUsize::new(0));
        let capability = capability(1, 1, 1);
        let identity = match capability.attempt(8, body(&drops, &executions)) {
            AttemptOutcome::Accepted(identity) => identity,
            _ => panic!("first request must accept"),
        };
        while executions.load(Ordering::SeqCst) == 0 {
            tokio::task::yield_now().await;
        }
        let second = match capability.attempt(8, body(&drops, &executions)) {
            AttemptOutcome::Accepted(identity) => identity,
            _ => panic!("spare fixed registry slot must accept"),
        };
        while executions.load(Ordering::SeqCst) < 2 {
            tokio::task::yield_now().await;
        }
        let rejected = match capability.attempt(8, body(&drops, &executions)) {
            AttemptOutcome::Reject(RejectReason::RegistryFull, body) => body,
            _ => panic!("full unreaped registry must reject without a second queue"),
        };
        drop(rejected);
        assert_eq!(
            capability.complete(identity).await,
            CompletionOutcome::Completed
        );
        assert_eq!(
            capability.complete(second).await,
            CompletionOutcome::Completed
        );
        assert!(capability.shutdown(false).await.healthy);
        assert_eq!(drops.load(Ordering::SeqCst), 3);
    }

    #[tokio::test]
    async fn registered_owner_retries_only_with_the_matching_private_token() {
        let drops = Arc::new(AtomicUsize::new(0));
        let executions = Arc::new(AtomicUsize::new(0));
        let capability = capability(1, 1, 1);
        let active = match capability.attempt(8, body(&drops, &executions)) {
            AttemptOutcome::Accepted(identity) => identity,
            _ => panic!("first request must accept"),
        };
        let waiting = match capability.attempt(8, body(&drops, &executions)) {
            AttemptOutcome::Registered(waiting) => waiting,
            _ => panic!("capacity miss must register atomically"),
        };
        assert_eq!(
            capability.complete(active).await,
            CompletionOutcome::Completed
        );
        let token = capability.claim_retry().unwrap();
        assert!(waiting.is_retry_for(&token));
        let retried = match capability.retry(8, waiting, token) {
            AttemptOutcome::Accepted(identity) => identity,
            _ => panic!("matching claimed retry must publish"),
        };
        assert_eq!(
            capability.complete(retried).await,
            CompletionOutcome::Completed
        );
        assert_eq!(executions.load(Ordering::SeqCst), 2);
        assert_eq!(drops.load(Ordering::SeqCst), 2);
        assert!(capability.shutdown(false).await.healthy);
    }

    #[tokio::test]
    async fn mismatched_linear_retry_stops_waiting_without_losing_body_owners() {
        let drops = Arc::new(AtomicUsize::new(0));
        let executions = Arc::new(AtomicUsize::new(0));
        let capability = capability(1, 2, 1);
        let active = match capability.attempt(8, body(&drops, &executions)) {
            AttemptOutcome::Accepted(identity) => identity,
            _ => panic!("first request must accept"),
        };
        let first = match capability.attempt(8, body(&drops, &executions)) {
            AttemptOutcome::Registered(waiting) => waiting,
            _ => panic!("first waiter must register"),
        };
        let second = match capability.attempt(8, body(&drops, &executions)) {
            AttemptOutcome::Registered(waiting) => waiting,
            _ => panic!("second waiter must register"),
        };
        assert_eq!(
            capability.complete(active).await,
            CompletionOutcome::Completed
        );
        let token = capability.claim_retry().unwrap();
        assert!(first.is_retry_for(&token));
        assert!(!second.is_retry_for(&token));
        let second_body = match capability.retry(8, second, token) {
            AttemptOutcome::Stop(StopReason::InvalidRetry, body) => body,
            _ => panic!("mismatched retry must stop, not self-retry"),
        };
        drop(second_body);
        drop(capability.cancel(first).into_body());
        assert!(capability.shutdown(false).await.healthy);
        assert_eq!(drops.load(Ordering::SeqCst), 3);
    }

    #[tokio::test]
    async fn consuming_shutdown_aborts_and_joins_pending_execution() {
        let drops = Arc::new(AtomicUsize::new(0));
        let executions = Arc::new(AtomicUsize::new(0));
        let capability = TransportCapability::try_new(
            config(1, 1, 1),
            |body: Body, _resources: RequestResources<'_>| NeverFuture { body: Some(body) },
        )
        .unwrap();
        assert!(matches!(
            capability.attempt(8, body(&drops, &executions)),
            AttemptOutcome::Accepted(_)
        ));
        let shutdown = capability.shutdown(true).await;
        assert!(shutdown.healthy);
        assert!(shutdown.task_failed);
        assert_eq!(drops.load(Ordering::SeqCst), 1);
        assert_eq!(executions.load(Ordering::SeqCst), 0);
    }

    #[tokio::test]
    async fn cancel_returns_only_the_transport_body_and_removes_registration() {
        let drops = Arc::new(AtomicUsize::new(0));
        let executions = Arc::new(AtomicUsize::new(0));
        let capability = capability(1, 1, 1);
        let active = match capability.attempt(8, body(&drops, &executions)) {
            AttemptOutcome::Accepted(identity) => identity,
            _ => panic!("first request must accept"),
        };
        let waiting = match capability.attempt(8, body(&drops, &executions)) {
            AttemptOutcome::Registered(waiting) => waiting,
            _ => panic!("second request must wait"),
        };
        let cancelled = capability.cancel(waiting);
        assert!(cancelled.was_registered());
        drop(cancelled.into_body());
        assert_eq!(
            capability.complete(active).await,
            CompletionOutcome::Completed
        );
        assert_eq!(executions.load(Ordering::SeqCst), 1);
        assert_eq!(drops.load(Ordering::SeqCst), 2);
        assert!(capability.shutdown(false).await.healthy);
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn shutdown_and_attempt_publish_or_stop_without_detaching() {
        for _ in 0..64 {
            let drops = Arc::new(AtomicUsize::new(0));
            let executions = Arc::new(AtomicUsize::new(0));
            let capability = Arc::new(capability(1, 1, 2));
            let start = Arc::new(Barrier::new(3));
            let attempt_capability = Arc::clone(&capability);
            let attempt_start = Arc::clone(&start);
            let attempt_drops = Arc::clone(&drops);
            let attempt_executions = Arc::clone(&executions);
            let runtime = tokio::runtime::Handle::current();
            let attempt = thread::spawn(move || {
                let _runtime = runtime.enter();
                attempt_start.wait();
                attempt_capability.attempt(8, body(&attempt_drops, &attempt_executions))
            });
            let shutdown_capability = Arc::clone(&capability);
            let shutdown_start = Arc::clone(&start);
            let shutdown = thread::spawn(move || {
                shutdown_start.wait();
                shutdown_capability.begin_shutdown()
            });
            start.wait();
            let outcome = attempt.join().unwrap();
            shutdown.join().unwrap();
            match outcome {
                AttemptOutcome::Accepted(identity) => {
                    assert_eq!(
                        capability.complete(identity).await,
                        CompletionOutcome::Completed
                    );
                }
                AttemptOutcome::Stop(_, body) | AttemptOutcome::Reject(_, body) => drop(body),
                AttemptOutcome::Registered(waiting) => {
                    drop(capability.cancel(waiting).into_body());
                }
            }
            let capability = Arc::try_unwrap(capability).ok().unwrap();
            assert!(capability.shutdown(false).await.healthy);
            assert_eq!(drops.load(Ordering::SeqCst), 1);
        }
    }

    #[test]
    fn capability_surface_contains_no_admission_owners() {
        assert_eq!(std::mem::size_of::<RequestIdentity>(), 16);
        assert_eq!(std::mem::size_of::<RetryToken>(), 16);
        fn assert_send_static<T: Send + 'static>() {}
        assert_send_static::<WaitingBody<Body>>();
    }

    #[test]
    fn abnormal_capability_drop_is_finite_fail_closed_in_subprocess() {
        const CHILD: &str = "c4_capability::tests::abnormal_capability_drop_child";
        let mut child = Command::new(env::current_exe().unwrap())
            .args(["--exact", CHILD, "--nocapture"])
            .env("SADDLE_C4_CAPABILITY_ABORT_CHILD", "1")
            .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() >= Duration::from_secs(5) {
                child.kill().unwrap();
                child.wait().unwrap();
                panic!("capability Drop did not terminate within five seconds");
            }
            thread::sleep(Duration::from_millis(10));
        };
        assert_eq!(status.signal(), Some(6));
    }

    #[test]
    fn abnormal_capability_drop_child() {
        if env::var_os("SADDLE_C4_CAPABILITY_ABORT_CHILD").is_none() {
            return;
        }
        let drops = Arc::new(AtomicUsize::new(0));
        let executions = Arc::new(AtomicUsize::new(0));
        let runtime = tokio::runtime::Builder::new_current_thread()
            .build()
            .unwrap();
        runtime.block_on(async {
            let capability = capability(1, 1, 1);
            assert!(matches!(
                capability.attempt(8, body(&drops, &executions)),
                AttemptOutcome::Accepted(_)
            ));
            drop(capability);
        });
        unreachable!("abnormal capability Drop must abort");
    }
}