liminal-server 0.23.1

use super::{Arc, ConnectionRuntime, Ordering};
#[cfg(test)]
use super::{Barrier, PreWaitBarrier, ReclaimBarrier};

impl ConnectionRuntime {
    #[cfg(test)]
    pub(super) fn install_pre_wait_barrier(&self) -> (Arc<Barrier>, Arc<Barrier>) {
        let armed = Arc::new(Barrier::new(2));
        let release = Arc::new(Barrier::new(2));
        if let Ok(mut slot) = self.pre_wait_barrier.lock() {
            *slot = Some(PreWaitBarrier {
                armed: Arc::clone(&armed),
                release: Arc::clone(&release),
            });
        }
        (armed, release)
    }

    /// Installs the pid-specific reclamation gate (oracle 26) and returns its
    /// `(reached, release, done)` endpoints. The harness rendezvouses `reached`
    /// to pin the dead-but-tracked window, `release` to let the reclaim proceed,
    /// and `done` to observe the `remove()` funnel completing.
    #[cfg(test)]
    pub(in crate::server::connection) fn install_reclaim_barrier(
        &self,
        pid: u64,
    ) -> (Arc<Barrier>, Arc<Barrier>, Arc<Barrier>) {
        let reached = Arc::new(Barrier::new(2));
        let release = Arc::new(Barrier::new(2));
        let done = Arc::new(Barrier::new(2));
        if let Ok(mut slot) = self.reclaim_barrier.lock() {
            *slot = Some(ReclaimBarrier {
                pid,
                reached: Arc::clone(&reached),
                release: Arc::clone(&release),
                done: Arc::clone(&done),
            });
        }
        (reached, release, done)
    }

    /// Takes the installed reclamation gate iff it targets `pid` (one-use). Any
    /// other pid's delivery is ungated, so unrelated exits reclaim immediately.
    #[cfg(test)]
    pub(super) fn stage_reclaim_barrier(&self, pid: u64) -> Option<ReclaimBarrier> {
        let mut slot = self.reclaim_barrier.lock().ok()?;
        if slot.as_ref().is_some_and(|barrier| barrier.pid == pid) {
            slot.take()
        } else {
            None
        }
    }

    /// Installs the one-use drain-park gate (oracles 18, 19) and returns its
    /// `(armed, release)` endpoints. Staged in the drain waiter's observe->park
    /// window so a harness can deliver an exit strictly between the completion
    /// observation and the park. Entirely `#[cfg(test)]`; changes no production
    /// wait semantics.
    #[cfg(test)]
    pub(in crate::server::connection) fn install_drain_park_barrier(
        &self,
    ) -> (Arc<Barrier>, Arc<Barrier>) {
        let armed = Arc::new(Barrier::new(2));
        let release = Arc::new(Barrier::new(2));
        if let Ok(mut slot) = self.drain_park_barrier.lock() {
            *slot = Some(PreWaitBarrier {
                armed: Arc::clone(&armed),
                release: Arc::clone(&release),
            });
        }
        (armed, release)
    }

    /// Runs the one-use drain-park gate, if installed. One-use so only the staged
    /// park rendezvouses; every later park in the same waiter runs ungated.
    #[cfg(test)]
    pub(super) fn run_drain_park_barrier(&self) {
        let barrier = self
            .drain_park_barrier
            .lock()
            .ok()
            .and_then(|mut slot| slot.take());
        let Some(barrier) = barrier else {
            return;
        };
        barrier.armed.wait();
        barrier.release.wait();
    }

    /// Test-only count of drain waiter wakes that observed a real removal.
    #[cfg(test)]
    pub(in crate::server::connection) fn drain_exit_wakes(&self) -> u64 {
        self.drain_exit_wakes.load(Ordering::SeqCst)
    }

    /// Test-only count of drain waiter deadline expirations.
    #[cfg(test)]
    pub(in crate::server::connection) fn drain_deadline_hits(&self) -> u64 {
        self.drain_deadline_hits.load(Ordering::SeqCst)
    }

    /// Runs a one-use deterministic test gate after arm. Returns whether the gate
    /// was installed so only that staged probe contributes to observability.
    #[cfg(test)]
    pub(in crate::server::connection) fn run_pre_wait_barrier(&self) -> bool {
        let barrier = self
            .pre_wait_barrier
            .lock()
            .ok()
            .and_then(|mut slot| slot.take());
        let Some(barrier) = barrier else {
            return false;
        };
        barrier.armed.wait();
        barrier.release.wait();
        true
    }

    /// Installs a one-use barrier in `enqueue_control`'s insert->wake window
    /// (S8 staging: lets a test act as the control-drain consumer between the
    /// queue insertion and the wake attempt) and returns its test endpoints.
    #[cfg(test)]
    pub(in crate::server::connection) fn install_pre_wake_barrier(
        &self,
    ) -> (Arc<Barrier>, Arc<Barrier>) {
        let armed = Arc::new(Barrier::new(2));
        let release = Arc::new(Barrier::new(2));
        if let Ok(mut slot) = self.pre_wake_barrier.lock() {
            *slot = Some(PreWaitBarrier {
                armed: Arc::clone(&armed),
                release: Arc::clone(&release),
            });
        }
        (armed, release)
    }

    /// Runs the one-use insert->wake test gate, if installed.
    #[cfg(test)]
    pub(in crate::server::connection) fn run_pre_wake_barrier(&self) {
        let barrier = self
            .pre_wake_barrier
            .lock()
            .ok()
            .and_then(|mut slot| slot.take());
        if let Some(barrier) = barrier {
            barrier.armed.wait();
            barrier.release.wait();
        }
    }

    #[cfg(test)]
    pub(in crate::server::connection) fn record_pre_wait_probe_hit(&self) {
        self.pre_wait_probe_hits.fetch_add(1, Ordering::AcqRel);
    }

    #[cfg(test)]
    pub(super) fn pre_wait_probe_hits(&self) -> u64 {
        self.pre_wait_probe_hits.load(Ordering::Acquire)
    }
}