saddle-runtime 0.2.0-rc.6

//! Isolated C4 attempt-ingress and publication model.
//!
//! This stays crate-private and uses a controlled Admission stand-in. It
//! proves the Runtime ownership and concurrency contract without connecting
//! the production transport or freezing an API.

#![allow(dead_code)]

use std::sync::{
    Mutex, MutexGuard, TryLockError,
    atomic::{AtomicBool, Ordering},
};

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

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
struct CompletionIdentity {
    slot: usize,
    generation: u64,
}

#[derive(Debug)]
enum AttemptResult<O> {
    Accepted(CompletionIdentity),
    Reject(O),
    Stop(O),
}

#[derive(Debug)]
struct Published<O> {
    generation: u64,
    ownership: O,
    lifecycle: RequestGuard,
}

#[derive(Debug)]
struct CoordinatorState<O> {
    requests: RequestLifecycle,
    admission_available: usize,
    published: Vec<Option<Published<O>>>,
    generations: Vec<u64>,
}

/// A fixed-capacity, non-queuing ingress around the unique publish owner.
///
/// Callers first claim one preallocated ingress slot and then use `try_lock`.
/// Contention is rejected synchronously; it never creates an async waiter,
/// channel node, registration, or per-attempt task. Only a later Admission
/// slice that atomically stores a generation-tagged registration may return
/// the C4 `Registered` outcome.
#[derive(Debug)]
struct AttemptCoordinator<O> {
    ingress: Box<[AtomicBool]>,
    state: Mutex<CoordinatorState<O>>,
    closed: AtomicBool,
}

impl<O> AttemptCoordinator<O> {
    fn new(
        requests: RequestLifecycle,
        ingress_capacity: usize,
        publish_capacity: usize,
        admission_capacity: usize,
    ) -> Self {
        assert!(ingress_capacity > 0);
        assert!(publish_capacity > 0);
        assert!(admission_capacity <= publish_capacity);
        Self {
            ingress: (0..ingress_capacity)
                .map(|_| AtomicBool::new(false))
                .collect(),
            state: Mutex::new(CoordinatorState {
                requests,
                admission_available: admission_capacity,
                published: (0..publish_capacity).map(|_| None).collect(),
                generations: vec![0; publish_capacity],
            }),
            closed: AtomicBool::new(false),
        }
    }

    fn attempt(&self, ownership: O) -> AttemptResult<O> {
        if self.closed.load(Ordering::Acquire) {
            return AttemptResult::Stop(ownership);
        }
        let Some(_ingress) = self.try_enter() else {
            return self.at_capacity(ownership);
        };
        let mut state = match self.state.try_lock() {
            Ok(state) => state,
            Err(TryLockError::WouldBlock) => return self.at_capacity(ownership),
            Err(TryLockError::Poisoned(_)) => std::process::abort(),
        };

        let Some(slot) = state.published.iter().position(Option::is_none) else {
            return AttemptResult::Reject(ownership);
        };
        let claim = match state.requests.try_claim() {
            Ok(claim) => claim,
            Err(_) => return AttemptResult::Stop(ownership),
        };
        if state.admission_available == 0 {
            return AttemptResult::Reject(ownership);
        }

        // This is the complete synchronous publication section. No await,
        // callback, transport I/O, queue, or fallible allocation occurs after
        // the unpublished phase claim. The preallocated slot becomes the only
        // owner of both the transferred input and lifecycle publication.
        state.admission_available -= 1;
        let generation = state.generations[slot]
            .checked_add(1)
            .unwrap_or_else(|| std::process::abort());
        state.generations[slot] = generation;
        state.published[slot] = Some(Published {
            generation,
            ownership,
            lifecycle: RequestClaim::publish(claim),
        });
        AttemptResult::Accepted(CompletionIdentity { slot, generation })
    }

    fn complete(&self, identity: CompletionIdentity) -> bool {
        let mut state = self.lock_state();
        let Some(slot) = state.published.get_mut(identity.slot) else {
            return false;
        };
        let Some(published) = slot.take() else {
            return false;
        };
        if published.generation != identity.generation {
            state.published[identity.slot] = Some(published);
            return false;
        }
        state.admission_available += 1;
        drop(published.lifecycle);
        drop(published.ownership);
        true
    }

    fn begin_shutdown(&self) -> bool {
        // Closing ingress is the shutdown side of the arbitration. An attempt
        // that already entered may finish publication; every later attempt
        // observes Stop.
        self.closed.store(true, Ordering::Release);
        let Some(_ingress) = self.try_enter() else {
            return false;
        };
        let state = match self.state.try_lock() {
            Ok(state) => state,
            Err(TryLockError::WouldBlock) => return false,
            Err(TryLockError::Poisoned(_)) => std::process::abort(),
        };
        state.requests.begin_draining();
        true
    }

    fn finish_shutdown(&self) -> bool {
        let state = self.lock_state();
        if state.published.iter().any(Option::is_some) {
            return false;
        }
        state.requests.mark_stopped();
        true
    }

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

    fn at_capacity(&self, ownership: O) -> AttemptResult<O> {
        if self.closed.load(Ordering::Acquire) {
            return AttemptResult::Stop(ownership);
        }
        AttemptResult::Reject(ownership)
    }

    fn lock_state(&self) -> MutexGuard<'_, CoordinatorState<O>> {
        self.state.lock().unwrap_or_else(|_| std::process::abort())
    }
}

struct IngressGuard<'a, O> {
    ingress: &'a AttemptCoordinator<O>,
    slot: usize,
}

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

#[cfg(test)]
mod tests {
    use std::{
        sync::{
            Arc, Barrier,
            atomic::{AtomicUsize, Ordering},
        },
        thread,
    };

    use super::*;
    use crate::ApplicationPhase;

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

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

    fn ready_coordinator(
        ingress: usize,
        published: usize,
        admission: usize,
    ) -> AttemptCoordinator<OwnedInput> {
        let requests = RequestLifecycle::new();
        requests.mark_ready();
        AttemptCoordinator::new(requests, ingress, published, admission)
    }

    fn input(drops: &Arc<AtomicUsize>) -> OwnedInput {
        OwnedInput {
            drops: Arc::clone(drops),
        }
    }

    #[test]
    fn success_returns_only_identity_and_transfers_the_unique_owner() {
        let drops = Arc::new(AtomicUsize::new(0));
        let coordinator = ready_coordinator(1, 1, 1);
        let identity = match coordinator.attempt(input(&drops)) {
            AttemptResult::Accepted(identity) => identity,
            _ => panic!("attempt must publish"),
        };
        assert_eq!(drops.load(Ordering::SeqCst), 0);
        assert!(coordinator.complete(identity));
        assert_eq!(drops.load(Ordering::SeqCst), 1);
        assert!(coordinator.begin_shutdown());
        assert!(coordinator.finish_shutdown());
    }

    #[test]
    fn capacity_and_serialization_contention_reject_without_queueing() {
        let drops = Arc::new(AtomicUsize::new(0));
        let coordinator = ready_coordinator(1, 1, 1);
        let occupied = coordinator.try_enter().unwrap();
        assert!(matches!(
            coordinator.attempt(input(&drops)),
            AttemptResult::Reject(_)
        ));
        assert_eq!(drops.load(Ordering::SeqCst), 1);
        drop(occupied);
        assert!(coordinator.begin_shutdown());
        assert!(coordinator.finish_shutdown());
    }

    #[test]
    fn admission_miss_rolls_back_claim_and_returns_ownership() {
        let drops = Arc::new(AtomicUsize::new(0));
        let coordinator = ready_coordinator(1, 1, 0);
        assert!(matches!(
            coordinator.attempt(input(&drops)),
            AttemptResult::Reject(_)
        ));
        assert_eq!(
            coordinator.lock_state().requests.phase(),
            ApplicationPhase::Ready
        );
        assert!(coordinator.begin_shutdown());
        assert!(coordinator.finish_shutdown());
        assert_eq!(drops.load(Ordering::SeqCst), 1);
    }

    #[test]
    fn shutdown_and_attempt_have_only_publish_or_stop_outcomes() {
        for _ in 0..1_000 {
            let drops = Arc::new(AtomicUsize::new(0));
            let coordinator = Arc::new(ready_coordinator(2, 1, 1));
            let start = Arc::new(Barrier::new(3));

            let attempt_coordinator = Arc::clone(&coordinator);
            let attempt_start = Arc::clone(&start);
            let attempt_drops = Arc::clone(&drops);
            let attempt = thread::spawn(move || {
                attempt_start.wait();
                attempt_coordinator.attempt(input(&attempt_drops))
            });

            let shutdown_coordinator = Arc::clone(&coordinator);
            let shutdown_start = Arc::clone(&start);
            let shutdown = thread::spawn(move || {
                shutdown_start.wait();
                while !shutdown_coordinator.begin_shutdown() {
                    thread::yield_now();
                }
            });

            start.wait();
            let result = attempt.join().unwrap();
            shutdown.join().unwrap();
            match result {
                AttemptResult::Accepted(identity) => {
                    assert_eq!(
                        coordinator.lock_state().requests.phase(),
                        ApplicationPhase::Draining
                    );
                    assert!(coordinator.complete(identity));
                }
                AttemptResult::Stop(ownership) => drop(ownership),
                AttemptResult::Reject(_) => {
                    panic!("shutdown race has no capacity result")
                }
            }
            assert!(coordinator.finish_shutdown());
            assert_eq!(drops.load(Ordering::SeqCst), 1);
        }
    }

    #[test]
    fn stale_non_owning_identity_cannot_complete_reused_slot() {
        let drops = Arc::new(AtomicUsize::new(0));
        let coordinator = ready_coordinator(1, 1, 1);
        let first = match coordinator.attempt(input(&drops)) {
            AttemptResult::Accepted(identity) => identity,
            _ => panic!("first attempt must publish"),
        };
        assert!(coordinator.complete(first));
        let second = match coordinator.attempt(input(&drops)) {
            AttemptResult::Accepted(identity) => identity,
            _ => panic!("second attempt must publish"),
        };
        assert_ne!(first, second);
        assert!(!coordinator.complete(first));
        assert!(coordinator.complete(second));
        assert!(coordinator.begin_shutdown());
        assert!(coordinator.finish_shutdown());
        assert_eq!(drops.load(Ordering::SeqCst), 2);
    }

    #[test]
    fn all_storage_is_fixed_before_the_first_attempt() {
        let drops = Arc::new(AtomicUsize::new(0));
        let coordinator = ready_coordinator(3, 2, 2);
        {
            let state = coordinator.lock_state();
            assert_eq!(coordinator.ingress.len(), 3);
            assert_eq!(state.published.len(), 2);
            assert_eq!(state.published.capacity(), 2);
            assert_eq!(state.generations.len(), 2);
            assert_eq!(state.generations.capacity(), 2);
        }

        for _ in 0..1_000 {
            let identity = match coordinator.attempt(input(&drops)) {
                AttemptResult::Accepted(identity) => identity,
                _ => panic!("fixed storage must remain reusable"),
            };
            assert!(coordinator.complete(identity));
        }
        let state = coordinator.lock_state();
        assert_eq!(state.published.capacity(), 2);
        assert_eq!(state.generations.capacity(), 2);
        drop(state);
        assert!(coordinator.begin_shutdown());
        assert!(coordinator.finish_shutdown());
        assert_eq!(drops.load(Ordering::SeqCst), 1_000);
    }
}