beamr 0.19.2

//! Bounded, insertion-ordered store of process exit tombstones.
//!
//! A *tombstone* records the [`ExitReason`] of a process that has died. It is
//! the load-bearing exit-detection signal: [`Scheduler::run_until_exit`] parks
//! on a condvar and only returns once it observes the dead pid's tombstone, and
//! [`Scheduler::peek_exit_reason`] / the link/monitor already-dead guards read
//! it to discover a process has gone.
//!
//! Historically this was an unbounded `DashMap<u64, ExitReason>`: a tombstone
//! was written on every process death and *never* removed for the lifetime of
//! the scheduler. Under a workload that spawns a fresh process per connection
//! (or per request), that map grows without bound — a slow but real leak.
//!
//! [`BoundedTombstones`] caps the live tombstone count at [`TOMBSTONE_CAPACITY`]
//! entries using a pure insertion-order (FIFO) eviction policy: when a new
//! tombstone would push the count past the cap, the *oldest* tombstone is
//! evicted. The cap is deliberately huge (64Ki entries, low single-digit MB)
//! so that eviction is invisible to every legitimate reader:
//!
//! * `run_until_exit` always targets a pid whose tombstone was *just* inserted
//!   to wake that very caller; FIFO eviction only reclaims the oldest entries
//!   once [`TOMBSTONE_CAPACITY`] *newer* exits have accumulated, which cannot
//!   happen inside the sub-10ms condvar wake window — so a blocked
//!   `run_until_exit` can never miss its real exit.
//! * `peek_exit_reason` and the link/monitor guards observe recently-dead pids
//!   in practice (a just-closed connection, never one buried 64Ki exits deep),
//!   so for them too the cap is effectively unreachable.
//!
//! The additive finalization ledger has a different retention contract: its
//! complete owned `(reason, term)` value is retained until consumed, even if the
//! legacy tombstone is evicted. Taking releases the owned term but deliberately
//! leaves a compact per-pid token for the scheduler lifetime. That token makes
//! outcome installation and event publication exactly-once across both outcome
//! consumption and tombstone eviction. Callers must drain outcomes to bound the
//! retained owned-term payload; the token ledger itself grows by one entry per
//! finalized pid and is the same map, not a second unbounded store. The legacy
//! result and diagnostic satellites remain bounded with the legacy tombstone,
//! preserving their existing semantics.

use super::exit_events::{
    ExitEvent, ExitEventPublisher, ExitEventSubscription, ExitWatchRegistry, StoreWatch,
};
use crate::ets::copy::OwnedTerm;
use crate::process::ExitReason;
use dashmap::DashMap;
use std::collections::VecDeque;
use std::sync::{Arc, Mutex};

/// Maximum number of live exit tombstones retained at once.
///
/// At ~16 bytes per entry (a `u64` pid plus a `Copy` [`ExitReason`]) plus
/// DashMap overhead, 65536 entries caps the tombstone map at low single-digit
/// MB while leaving an enormous safety margin: a process that exited would have
/// to be followed by 65,536 *further* exits before its tombstone is reclaimed.
/// That dwarfs any plausible window of concurrently-interesting recently-dead
/// pids (a server with thousands of in-flight connections still has its
/// just-closed connection's tombstone well within the most-recent 64Ki), so the
/// FIFO eviction policy is effectively invisible to every legitimate reader
/// while still hard-bounding memory.
pub(super) const TOMBSTONE_CAPACITY: usize = 65_536;

/// Durable additive state for one process's first terminal transition.
///
/// `outcome` becomes `None` when taken, but the entry itself is the permanent
/// publication token. `reason` is therefore the authoritative additive reason
/// used by both the outcome and event even if a later cleanup overwrites the
/// bounded legacy tombstone's compatibility value.
struct FinalizedOutcome {
    reason: ExitReason,
    outcome: Option<OwnedTerm>,
}

/// A bounded, insertion-ordered concurrent map from pid to [`ExitReason`].
///
/// Reads are lock-free via the inner [`DashMap`] and preserve the exact
/// `Option`-returning semantics callers rely on (a miss returns `None`, same as
/// an unknown pid). Inserts additionally record the pid in a FIFO order queue
/// and, on overflow, evict the oldest pid — returning it so the caller can
/// evict the paired satellite entries.
pub(super) struct BoundedTombstones {
    reasons: DashMap<u64, ExitReason>,
    /// Complete take-once outcomes plus their durable publication tokens. The
    /// owned term is retained until consumed; the compact token remains for the
    /// scheduler lifetime and closes the eviction/consumption TOCTOU. Its 40-byte
    /// per-pid value residue is the deliberate, lawful price of process-lifetime
    /// exactly-once (excluding the `DashMap` key and bucket overhead).
    outcomes: DashMap<u64, FinalizedOutcome>,
    /// Insertion order of currently-live pids, oldest at the front. Guarded
    /// independently of the DashMap shards; it also serializes writers so an
    /// complete outcome is always visible before its legacy tombstone.
    order: Mutex<VecDeque<u64>>,
    capacity: usize,
    events: ExitEventPublisher,
    /// Notification-only one-shot exit watches (EXIT-001). Lives beside the
    /// durable `outcomes` record it reads; registration never takes `order`.
    watches: Arc<ExitWatchRegistry>,
}

impl BoundedTombstones {
    /// Create a store with the default [`TOMBSTONE_CAPACITY`].
    pub(super) fn new() -> Self {
        Self::with_capacity(TOMBSTONE_CAPACITY)
    }

    /// Create a store with an explicit capacity. `capacity` must be non-zero;
    /// a zero capacity is clamped to 1 so the structure always stores at least
    /// the most recent tombstone.
    pub(super) fn with_capacity(capacity: usize) -> Self {
        Self {
            reasons: DashMap::new(),
            outcomes: DashMap::new(),
            order: Mutex::new(VecDeque::new()),
            capacity: capacity.max(1),
            events: ExitEventPublisher::new(),
            watches: Arc::new(ExitWatchRegistry::new()),
        }
    }

    /// Read the exit reason for `pid`, or `None` if no tombstone is present.
    ///
    /// Lock-free and non-consuming: the tombstone is left in place. Takes the
    /// pid by reference to mirror the [`DashMap`] this replaced, keeping call
    /// sites unchanged.
    pub(super) fn get(&self, pid: &u64) -> Option<ExitReason> {
        self.reasons.get(pid).map(|entry| *entry)
    }

    /// Whether a tombstone exists for `pid`.
    pub(super) fn contains_key(&self, pid: &u64) -> bool {
        self.reasons.contains_key(pid)
    }

    /// Consume the complete retained outcome for `pid` exactly once.
    pub(super) fn take_outcome(&self, pid: &u64) -> Option<(ExitReason, OwnedTerm)> {
        let mut finalized = self.outcomes.get_mut(pid)?;
        let outcome = finalized.outcome.take()?;
        Some((finalized.reason, outcome))
    }

    /// Create the scheduler's sole exit-event subscription.
    pub(super) fn subscribe(&self) -> Option<ExitEventSubscription> {
        self.events.subscribe()
    }

    /// Non-consuming read of the authoritative additive exit reason.
    ///
    /// Served from the durable `FinalizedOutcome` token, which survives both
    /// legacy-tombstone eviction and outcome consumption — the unlosable
    /// already-dead source EXIT-001 D4 requires. `None` means no terminal
    /// transition has been recorded for `pid`.
    pub(super) fn finalized_reason(&self, pid: &u64) -> Option<ExitReason> {
        self.outcomes.get(pid).map(|finalized| finalized.reason)
    }

    /// Register-then-check: arm a one-shot watch for `pid`, then consult the
    /// durable record (EXIT-001 D3).
    ///
    /// Registration precedes the check because publication happens outside
    /// the writer mutex: check-then-register can miss an exit that installed
    /// its record but has not yet published, while register-then-check at
    /// worst observes both the record and a concurrent fire — which the
    /// one-shot slot absorbs (the reported answer wins; the armed slot is
    /// deregistered and a racing fire into it is a no-op).
    pub(super) fn watch(&self, pid: u64) -> StoreWatch {
        let watch = self.watches.register(pid);
        if let Some(reason) = self.finalized_reason(&pid) {
            // The just-armed slot deregisters when `watch` drops here; a fire
            // that already removed the pid's entry makes that a no-op, and a
            // fire's send into the slot is absorbed — the record answer wins.
            return StoreWatch::AlreadyExited(reason);
        }
        StoreWatch::Armed(watch)
    }

    /// Number of pids with at least one live watch. Test/diagnostic helper.
    #[cfg(test)]
    pub(super) fn watched_pid_count(&self) -> usize {
        self.watches.watched_pid_count()
    }

    #[cfg(test)]
    pub(super) fn install_event_publication_gate(
        &self,
    ) -> super::exit_events::ExitEventPublicationObserver {
        self.events.install_publication_gate()
    }

    #[cfg(test)]
    pub(super) fn clear_event_publication_gate(&self) {
        self.events.clear_publication_gate();
    }

    /// Insert a legacy tombstone without publishing an additive outcome.
    ///
    /// Used by internal lifecycle tests that need to simulate an already-dead
    /// process. Production exits use [`Self::insert_outcome`].
    #[cfg(test)]
    pub(super) fn insert(&self, pid: u64, reason: ExitReason) -> Option<u64> {
        self.insert_inner(pid, reason, None)
    }

    /// Insert a tombstone together with a complete retained outcome.
    ///
    /// The first terminal caller atomically owns the durable additive token and
    /// publishes one retained outcome followed by one event. Later callers may
    /// preserve the historical legacy behavior by overwriting or restoring the
    /// bounded tombstone, but cannot change the authoritative additive reason,
    /// re-arm a consumed outcome, or publish another event. Consequently a
    /// legacy `get` after overlapping cleanup may report that later cleanup's
    /// compatibility reason; the additive outcome and event remain coherent.
    pub(super) fn insert_outcome(
        &self,
        pid: u64,
        reason: ExitReason,
        outcome: OwnedTerm,
    ) -> Option<u64> {
        self.insert_inner(pid, reason, Some(outcome))
    }

    fn insert_inner(
        &self,
        pid: u64,
        reason: ExitReason,
        outcome: Option<OwnedTerm>,
    ) -> Option<u64> {
        let mut order = match self.order.lock() {
            Ok(guard) => guard,
            Err(poisoned) => poisoned.into_inner(),
        };
        if self.reasons.contains_key(&pid) {
            self.reasons.insert(pid, reason);
            return None;
        }

        // The writer mutex makes this durable-token check-and-install atomic
        // across competing terminal callers. Unlike the bounded legacy reason,
        // the token survives both eviction and outcome consumption.
        let publish_event = if let Some(outcome) = outcome {
            if self.outcomes.contains_key(&pid) {
                false
            } else {
                // Keep this order: a reader can never observe the legacy
                // tombstone before the exactly-once outcome, and the event
                // follows both.
                self.outcomes.insert(
                    pid,
                    FinalizedOutcome {
                        reason,
                        outcome: Some(outcome),
                    },
                );
                true
            }
        } else {
            false
        };
        self.reasons.insert(pid, reason);
        order.push_back(pid);
        let mut evicted = None;
        if order.len() > self.capacity {
            // Loop to skip any pid already removed from the legacy map. The
            // finalized entry deliberately remains; taking only releases its
            // owned term and never removes its publication token.
            while let Some(oldest) = order.pop_front() {
                if let Some((evicted_pid, _)) = self.reasons.remove(&oldest) {
                    evicted = Some(evicted_pid);
                    break;
                }
            }
        }
        drop(order);

        if publish_event {
            self.events.publish(ExitEvent::Exited { pid, reason });
            // EXIT-001 D6 / OQ-A(outside): watch fires are APPENDED after the
            // existing publication, outside the writer mutex, carrying the
            // authoritative additive reason by value. Nothing is inserted
            // between outcome installation and the existing publish, and
            // nothing is reordered — the ordering two other repos build on.
            self.watches.fire(pid, reason);
        }
        evicted
    }

    /// Number of live tombstones. Test/diagnostic helper.
    #[cfg(test)]
    pub(super) fn len(&self) -> usize {
        self.reasons.len()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::term::Term;
    use std::sync::Barrier;
    use std::time::Duration;

    const EVENT_TIMEOUT: Duration = Duration::from_secs(10);

    fn insert(store: &BoundedTombstones, pid: u64, reason: ExitReason) -> Option<u64> {
        store.insert_outcome(pid, reason, OwnedTerm::immediate(Term::NIL))
    }

    #[test]
    fn finalized_outcome_residue_is_size_bounded_and_payload_free_after_take() {
        const PINNED_RETAINED_VALUE_BYTES: usize = 40;

        // Measured size on the supported 64-bit layout is 40 bytes. Pin that
        // ceiling so the permanent exactly-once token cannot silently grow.
        assert!(
            core::mem::size_of::<FinalizedOutcome>() <= PINNED_RETAINED_VALUE_BYTES,
            "retained value grew beyond {PINNED_RETAINED_VALUE_BYTES} bytes"
        );

        let store = BoundedTombstones::with_capacity(1);
        let pid = 1;
        let payload = OwnedTerm::from_allocations(Term::NIL, vec![vec![0_u64].into_boxed_slice()]);
        assert_eq!(payload.allocation_count(), 1, "test payload must allocate");
        store.insert_outcome(pid, ExitReason::Normal, payload);

        let (reason, payload) = store.take_outcome(&pid).expect("outcome is takeable");
        assert_eq!(reason, ExitReason::Normal);
        assert_eq!(payload.allocation_count(), 1, "take returns the allocation");
        drop(payload);

        let retained = store.outcomes.get(&pid).expect("token remains durable");
        assert!(
            retained.outcome.is_none(),
            "take_exit_outcome must leave no OwnedTerm or payload allocation in the ledger"
        );
    }

    /// (a) Inserting far more than the cap keeps the live count bounded at the
    /// cap and never above it.
    #[test]
    fn insert_over_cap_stays_bounded() {
        let cap = 8;
        let store = BoundedTombstones::with_capacity(cap);
        for pid in 0..1_000u64 {
            insert(&store, pid, ExitReason::Normal);
            assert!(
                store.len() <= cap,
                "len {} exceeded cap {} after inserting pid {}",
                store.len(),
                cap,
                pid
            );
        }
        assert_eq!(store.len(), cap, "store settles exactly at the cap");
    }

    /// (b) The most-recent tombstones survive and read back their reason.
    #[test]
    fn most_recent_survive_and_are_readable() {
        let cap = 8;
        let store = BoundedTombstones::with_capacity(cap);
        for pid in 0..100u64 {
            // Vary the reason so we also confirm the right value comes back.
            let reason = if pid % 2 == 0 {
                ExitReason::Normal
            } else {
                ExitReason::Kill
            };
            insert(&store, pid, reason);
        }
        // The last `cap` pids (92..=99) must all be present with their reason.
        for pid in 92..100u64 {
            let expected = if pid % 2 == 0 {
                ExitReason::Normal
            } else {
                ExitReason::Kill
            };
            assert_eq!(
                store.get(&pid),
                Some(expected),
                "recent pid {pid} must survive with its reason"
            );
            assert!(store.contains_key(&pid));
        }
    }

    /// (c) The oldest tombstones are evicted — `get` returns `None` for them —
    /// while recent ones still return `Some`, preserving exact Option
    /// semantics (a miss is indistinguishable from an unknown pid).
    #[test]
    fn oldest_are_evicted_recent_retained() {
        let cap = 4;
        let store = BoundedTombstones::with_capacity(cap);
        for pid in 0..10u64 {
            insert(&store, pid, ExitReason::Normal);
        }
        // Oldest 6 (0..=5) evicted.
        for pid in 0..6u64 {
            assert_eq!(store.get(&pid), None, "old pid {pid} must be evicted");
            assert!(!store.contains_key(&pid));
        }
        // Newest 4 (6..=9) retained.
        for pid in 6..10u64 {
            assert_eq!(
                store.get(&pid),
                Some(ExitReason::Normal),
                "recent pid {pid} must be retained"
            );
        }
    }

    /// A re-insert (overwrite) of a live pid must not duplicate it in the FIFO
    /// order, must update the reason, and must not evict a different live pid.
    #[test]
    fn overwrite_does_not_duplicate_or_misevict() {
        let cap = 3;
        let store = BoundedTombstones::with_capacity(cap);
        insert(&store, 1, ExitReason::Normal);
        insert(&store, 2, ExitReason::Normal);
        insert(&store, 3, ExitReason::Normal);
        // Overwrite the oldest; reason updates, order is unchanged.
        insert(&store, 1, ExitReason::Kill);
        assert_eq!(store.get(&1), Some(ExitReason::Kill));
        assert_eq!(store.len(), cap);
        // Next fresh insert evicts pid 1 (still the oldest by first-insert
        // order), not pid 2 or 3.
        let insertion = insert(&store, 4, ExitReason::Normal);
        assert_eq!(insertion, Some(1));
        assert_eq!(store.get(&1), None, "first-inserted pid is the one evicted");
        assert_eq!(store.get(&2), Some(ExitReason::Normal));
        assert_eq!(store.get(&3), Some(ExitReason::Normal));
        assert_eq!(store.get(&4), Some(ExitReason::Normal));
        assert_eq!(store.len(), cap);
        let (reason, _term) = store
            .take_outcome(&1)
            .expect("legacy overwrite and eviction leave outcome retained");
        assert_eq!(reason, ExitReason::Normal);
        assert!(store.take_outcome(&1).is_none(), "outcome is take-once");
    }

    #[test]
    fn duplicate_after_eviction_preserves_original_untaken_outcome_and_emits_no_event() {
        let store = BoundedTombstones::with_capacity(2);
        let subscription = store.subscribe().expect("first subscriber");

        store.insert_outcome(
            1,
            ExitReason::Normal,
            OwnedTerm::immediate(Term::small_int(11)),
        );
        assert_eq!(
            subscription.recv_timeout(EVENT_TIMEOUT),
            Ok(ExitEvent::Exited {
                pid: 1,
                reason: ExitReason::Normal,
            })
        );
        for pid in 2..=3 {
            store.insert_outcome(
                pid,
                ExitReason::Normal,
                OwnedTerm::immediate(Term::small_int(pid as i64)),
            );
            match subscription.recv_timeout(EVENT_TIMEOUT) {
                Ok(ExitEvent::Exited { pid: event_pid, .. }) if event_pid == pid => {}
                other => {
                    panic!("expected exit event for pid {pid}, got {other:?}")
                }
            }
            assert!(store.take_outcome(&pid).is_some());
        }
        assert_eq!(store.get(&1), None, "pid 1 tombstone was evicted");

        store.insert_outcome(
            1,
            ExitReason::Kill,
            OwnedTerm::immediate(Term::small_int(99)),
        );

        let (reason, outcome) = store
            .take_outcome(&1)
            .expect("first terminal transition remains takeable");
        assert_eq!(reason, ExitReason::Normal);
        assert_eq!(outcome.root().as_small_int(), Some(11));
        assert_eq!(
            subscription.recv_timeout(Duration::ZERO),
            Err(super::super::ExitEventRecvError::Timeout),
            "duplicate finalization cannot emit another event"
        );
    }

    // ===== EXIT-001 walls (store level) =====

    fn armed(watch: StoreWatch) -> super::super::ExitWatch {
        match watch {
            StoreWatch::Armed(watch) => watch,
            StoreWatch::AlreadyExited(reason) => {
                panic!("expected an armed watch, got AlreadyExited({reason:?})")
            }
        }
    }

    /// W2 — REGISTRATION/PUBLICATION RACE (EXIT-001 D3). A watch registered
    /// while a terminal transition is parked at the existing post-send
    /// rendezvous observes EXACTLY one notification: not zero (the lost-wake
    /// face of a check that missed the installed record) and not two (the
    /// duplicate face of a snapshot scheme that reports the record AND leaves
    /// a slot armed for the in-flight fire).
    #[test]
    fn w2_watch_registered_during_inflight_publication_observes_exactly_one() {
        let store = BoundedTombstones::with_capacity(8);
        let subscription = store.subscribe().expect("first subscriber");
        let observer = store.install_event_publication_gate();

        std::thread::scope(|scope| {
            let publisher = scope.spawn(|| {
                insert(&store, 1, ExitReason::Kill);
            });
            observer.wait_for_publication(EVENT_TIMEOUT);
            // Parked post-send: the outcome is installed and the event is
            // sent, but publish() has not returned and no watch has fired.
            // Capture only — no assert may panic while the parked publisher's
            // release depends on a later line (wedge law, ruled 17:51Z): the
            // registration happens at the park; its answer is judged after
            // release + join, where a red terminates instead of wedging.
            let at_park_registration = store.watch(1);
            observer.release_publication(EVENT_TIMEOUT);
            publisher.join().expect("terminal caller completes");
            store.clear_event_publication_gate();

            let mut notifications = 0_usize;
            let armed_watch = match at_park_registration {
                StoreWatch::AlreadyExited(reason) => {
                    assert_eq!(reason, ExitReason::Kill, "record answer carries the reason");
                    notifications += 1;
                    None
                }
                StoreWatch::Armed(watch) => Some(watch),
            };
            if let Some(watch) = armed_watch {
                if let Ok((pid, reason)) = watch.recv_timeout(EVENT_TIMEOUT) {
                    assert_eq!((pid, reason), (1, ExitReason::Kill));
                    notifications += 1;
                }
                // Any duplicate delivery would have to already be queued: the
                // fire path completed before the join above. A zero-timeout
                // second receive is therefore deterministic, not a sleep.
                if watch.recv_timeout(Duration::ZERO).is_ok() {
                    notifications += 1;
                }
            }
            assert_eq!(
                notifications, 1,
                "a registration racing an in-flight publication must observe exactly one \
                 notification (0 = lost wake, 2+ = duplicate); observed {notifications}"
            );
        });
        let _ = subscription.recv_timeout(Duration::ZERO);
    }

    /// W3 — MANY WATCHERS, ONE PID: every watch fires, none starves, and the
    /// drainer's exactly-once `take_outcome` is untouched AFTER all fires.
    #[test]
    fn w3_many_watchers_one_pid_all_notified_and_drainer_undisturbed() {
        let store = BoundedTombstones::with_capacity(8);
        let subscription = store.subscribe().expect("first subscriber");
        let watches: Vec<_> = (0..8).map(|_| armed(store.watch(5))).collect();

        store.insert_outcome(
            5,
            ExitReason::Kill,
            OwnedTerm::immediate(Term::small_int(9)),
        );

        for (index, watch) in watches.iter().enumerate() {
            assert_eq!(
                watch.recv_timeout(EVENT_TIMEOUT),
                Ok((5, ExitReason::Kill)),
                "watch {index} of 8 must be notified"
            );
        }
        assert_eq!(
            subscription.recv_timeout(EVENT_TIMEOUT),
            Ok(ExitEvent::Exited {
                pid: 5,
                reason: ExitReason::Kill,
            }),
            "the exclusive subscriber still receives the event"
        );
        let (reason, value) = store.take_outcome(&5).expect(
            "watches are notification-only: the outcome must still be takeable after every fire",
        );
        assert_eq!(reason, ExitReason::Kill);
        assert_eq!(value.root().as_small_int(), Some(9));
        assert!(store.take_outcome(&5).is_none(), "take stays exactly-once");
    }

    /// W4 — NO LEAK ON ABANDONMENT: the registry clears on BOTH paths (drop
    /// and fire), including under a mixed workload.
    #[test]
    fn w4_registry_holds_no_entries_after_abandonment_or_fire() {
        let store = BoundedTombstones::with_capacity(8);

        // Abandonment face: watches on a pid that never exits.
        let abandoned: Vec<_> = (0..5).map(|_| armed(store.watch(77))).collect();
        assert_eq!(
            store.watched_pid_count(),
            1,
            "precondition: pid 77 is watched"
        );
        drop(abandoned);
        assert_eq!(
            store.watched_pid_count(),
            0,
            "dropped watches must deregister themselves"
        );

        // Fire face: the pid's entry clears even while a handle is live.
        let fired = armed(store.watch(6));
        insert(&store, 6, ExitReason::Normal);
        assert_eq!(
            fired.recv_timeout(EVENT_TIMEOUT),
            Ok((6, ExitReason::Normal))
        );
        assert_eq!(
            store.watched_pid_count(),
            0,
            "a fired pid's registry entry must be cleared even while its handle is alive"
        );
        drop(fired);
        assert_eq!(store.watched_pid_count(), 0);

        // Mixed workload (R2 acceptance): fired + abandoned + never-exiting.
        let fired_a = armed(store.watch(10));
        let never_exits = armed(store.watch(11));
        let fired_b = armed(store.watch(10));
        insert(&store, 10, ExitReason::Kill);
        assert_eq!(
            fired_a.recv_timeout(EVENT_TIMEOUT),
            Ok((10, ExitReason::Kill))
        );
        assert_eq!(
            fired_b.recv_timeout(EVENT_TIMEOUT),
            Ok((10, ExitReason::Kill))
        );
        drop((fired_a, never_exits, fired_b));
        assert_eq!(
            store.watched_pid_count(),
            0,
            "mixed fired/abandoned/never-exiting workload must leave the registry empty"
        );
    }

    /// W5 — EXCLUSIVE SUBSCRIBER UNDISTURBED: event sequence and `Lagged`
    /// behaviour are identical with watches live on the same pids.
    #[test]
    fn w5_exclusive_subscription_sequence_and_lag_unchanged_by_watches() {
        let store = BoundedTombstones::with_capacity(8);
        let subscription = store.subscribe().expect("first subscriber");
        let watch_one = armed(store.watch(1));
        let watch_two = armed(store.watch(2));

        for pid in 1..=3 {
            insert(&store, pid, ExitReason::Normal);
        }
        for pid in 1..=3 {
            assert_eq!(
                subscription.recv_timeout(EVENT_TIMEOUT),
                Ok(ExitEvent::Exited {
                    pid,
                    reason: ExitReason::Normal,
                }),
                "subscriber sequence must be unchanged and in order"
            );
        }
        assert_eq!(
            watch_one.recv_timeout(EVENT_TIMEOUT),
            Ok((1, ExitReason::Normal))
        );
        assert_eq!(
            watch_two.recv_timeout(EVENT_TIMEOUT),
            Ok((2, ExitReason::Normal))
        );

        // Lagged unchanged: overflow the bounded event queue with a watch
        // armed on a pid inside the overflow batch. The watch must still fire
        // even though the subscriber lags.
        let lag_pid = 5_000;
        let lag_watch = armed(store.watch(lag_pid));
        let overflow = super::super::EXIT_EVENT_CAPACITY as u64 + 8;
        for pid in 100..(100 + overflow) {
            insert(&store, pid, ExitReason::Normal);
        }
        insert(&store, lag_pid, ExitReason::Kill);
        assert_eq!(
            subscription.recv_timeout(EVENT_TIMEOUT),
            Ok(ExitEvent::Lagged),
            "overflow must still surface as the typed Lagged marker"
        );
        assert_eq!(
            lag_watch.recv_timeout(EVENT_TIMEOUT),
            Ok((lag_pid, ExitReason::Kill)),
            "a watch must fire even when the exclusive subscriber lags"
        );
    }

    /// R3 acceptance — the already-dead source survives legacy eviction: a
    /// finalized pid whose bounded tombstone was FIFO-evicted still answers
    /// `AlreadyExited` with the authoritative reason, because the answer is
    /// served from the durable outcome record.
    #[test]
    fn watch_after_legacy_eviction_answers_from_durable_record() {
        let store = BoundedTombstones::with_capacity(2);
        insert(&store, 1, ExitReason::Kill);
        insert(&store, 2, ExitReason::Normal);
        insert(&store, 3, ExitReason::Normal);
        assert_eq!(
            store.get(&1),
            None,
            "precondition: legacy tombstone evicted"
        );
        assert_eq!(
            store.finalized_reason(&1),
            Some(ExitReason::Kill),
            "precondition: durable record survives eviction"
        );

        match store.watch(1) {
            StoreWatch::AlreadyExited(reason) => assert_eq!(
                reason,
                ExitReason::Kill,
                "the durable record's reason is authoritative"
            ),
            StoreWatch::Armed(watch) => {
                let outcome = watch.recv_timeout(Duration::from_secs(1));
                panic!(
                    "blocked past the deadline: an evicted-but-finalized pid armed a watch \
                     (recv gave {outcome:?}) instead of answering AlreadyExited(Kill) — the \
                     already-dead source must be the durable record, not the bounded tombstone"
                );
            }
        }
    }

    /// REVIEW POINT 2(a) — a watch registered AFTER the drainer consumed the
    /// outcome still reports the correct reason: the `FinalizedOutcome` token
    /// retains `reason` for the scheduler lifetime once its term is taken.
    #[test]
    fn watch_after_outcome_consumed_reports_reason_from_token() {
        let store = BoundedTombstones::with_capacity(4);
        insert(&store, 9, ExitReason::Kill);
        assert!(
            store.take_outcome(&9).is_some(),
            "precondition: outcome consumed"
        );
        assert!(store.take_outcome(&9).is_none());

        match store.watch(9) {
            StoreWatch::AlreadyExited(reason) => assert_eq!(reason, ExitReason::Kill),
            StoreWatch::Armed(watch) => {
                let outcome = watch.recv_timeout(Duration::from_secs(1));
                panic!(
                    "blocked past the deadline: a consumed-outcome pid armed a watch \
                     (recv gave {outcome:?}) instead of AlreadyExited — the token's \
                     retained reason must serve the answer after take"
                );
            }
        }
    }

    /// R2 acceptance — the documented publication order (retained outcome →
    /// legacy tombstone → event → watch fires) asserted by a test, using the
    /// existing post-send rendezvous: while publication is parked, the
    /// outcome and tombstone are installed and the watch has NOT yet fired;
    /// after release, it fires.
    #[test]
    fn publication_order_outcome_tombstone_event_then_watch_fire() {
        let store = BoundedTombstones::with_capacity(8);
        let subscription = store.subscribe().expect("first subscriber");
        let watch = armed(store.watch(1));
        let observer = store.install_event_publication_gate();

        std::thread::scope(|scope| {
            let publisher = scope.spawn(|| {
                insert(&store, 1, ExitReason::Kill);
            });
            observer.wait_for_publication(EVENT_TIMEOUT);
            // At the park: CAPTURE ONLY (wedge law, ruled 17:51Z). The prior
            // form asserted here, and a failed assert panicked inside this
            // scope while the parked publisher's release channel lived outside
            // it — the failure path was an infinite hang, discovered under the
            // publish-before-install mutation (wedge run in the EXIT-001
            // evidence). Every capture below is non-blocking at the bytes:
            // finalized_reason/get read DashMaps, the watch receive uses a
            // zero timeout, and the parked publisher holds only `order`.
            let at_park_finalized = store.finalized_reason(&1);
            let at_park_legacy = store.get(&1);
            let at_park_watch_fire = watch.recv_timeout(Duration::ZERO);
            observer.release_publication(EVENT_TIMEOUT);
            publisher.join().expect("terminal caller completes");
            store.clear_event_publication_gate();

            assert_eq!(
                at_park_finalized,
                Some(ExitReason::Kill),
                "outcome must be installed before the event publication returns"
            );
            assert_eq!(
                at_park_legacy,
                Some(ExitReason::Kill),
                "legacy tombstone must be installed before the event publication returns"
            );
            assert_eq!(
                at_park_watch_fire,
                Err(super::super::ExitEventRecvError::Timeout),
                "watch fires must come strictly AFTER the existing event publication"
            );
            assert_eq!(
                watch.recv_timeout(EVENT_TIMEOUT),
                Ok((1, ExitReason::Kill)),
                "the watch fires once publication completes"
            );
        });
        assert_eq!(
            subscription.recv_timeout(EVENT_TIMEOUT),
            Ok(ExitEvent::Exited {
                pid: 1,
                reason: ExitReason::Kill,
            })
        );
    }

    #[test]
    fn concurrent_terminal_callers_publish_one_authoritative_outcome_and_event() {
        let store = BoundedTombstones::with_capacity(4);
        let subscription = store.subscribe().expect("first subscriber");
        let start = Barrier::new(3);

        std::thread::scope(|scope| {
            let normal = scope.spawn(|| {
                start.wait();
                store.insert_outcome(
                    1,
                    ExitReason::Normal,
                    OwnedTerm::immediate(Term::small_int(10)),
                );
            });
            let killed = scope.spawn(|| {
                start.wait();
                store.insert_outcome(
                    1,
                    ExitReason::Kill,
                    OwnedTerm::immediate(Term::small_int(20)),
                );
            });
            start.wait();
            normal.join().expect("normal finalizer completes");
            killed.join().expect("kill finalizer completes");
        });

        let (event_reason, expected_value) = match subscription.recv_timeout(EVENT_TIMEOUT) {
            Ok(ExitEvent::Exited { pid: 1, reason }) => match reason {
                ExitReason::Normal => (reason, 10),
                ExitReason::Kill => (reason, 20),
                other => panic!("unexpected authoritative reason {other:?}"),
            },
            other => panic!("expected one exit event, got {other:?}"),
        };
        let (outcome_reason, outcome) = store
            .take_outcome(&1)
            .expect("authoritative outcome is installed once");
        assert_eq!(outcome_reason, event_reason);
        assert_eq!(outcome.root().as_small_int(), Some(expected_value));
        assert!(
            store.take_outcome(&1).is_none(),
            "the losing finalizer cannot install another outcome"
        );
        assert_eq!(
            subscription.recv_timeout(Duration::ZERO),
            Err(super::super::ExitEventRecvError::Timeout),
            "the losing finalizer cannot emit an event"
        );
        let later_legacy_reason = match event_reason {
            ExitReason::Normal => ExitReason::Kill,
            ExitReason::Kill => ExitReason::Normal,
            _ => unreachable!("event reason was restricted above"),
        };
        assert_eq!(
            store.get(&1),
            Some(later_legacy_reason),
            "legacy overwrite is compatibility-only; the additive reason is authoritative"
        );
    }
}