liminal-server 0.25.4

use super::{ConnectionRuntime, RawFd, Receiver, channel};

impl ConnectionRuntime {
    /// R7: records one serviced slice for `pid`. Bumped at the head of every
    /// slice; the park-flip's quiescence assertion reads [`Self::slice_count`].
    #[cfg(test)]
    pub(in crate::server::connection) fn record_slice(&self, pid: u64) {
        let count = if let Ok(mut counts) = self.slice_counts.lock() {
            let count = counts.entry(pid).or_insert(0);
            *count += 1;
            *count
        } else {
            return;
        };
        if let Ok(mut observers) = self.slice_observers.lock()
            && let Some(observer) = observers.remove(&pid)
        {
            let _ = observer.send(count);
        }
    }

    #[cfg(test)]
    pub(super) fn observe_next_slice(&self, pid: u64) -> Receiver<u64> {
        let (sender, receiver) = channel();
        if let Ok(mut observers) = self.slice_observers.lock() {
            observers.insert(pid, sender);
        }
        receiver
    }

    #[cfg(test)]
    pub(super) fn observe_next_park(&self, pid: u64) -> Receiver<u64> {
        let (sender, receiver) = channel();
        if let Ok(mut observers) = self.park_observers.lock() {
            observers.insert(pid, sender);
        }
        receiver
    }

    #[cfg(test)]
    pub(super) fn observe_settled_park(&self, pid: u64) -> Receiver<u64> {
        let (sender, receiver) = channel();
        let Ok(counts) = self.slice_counts.lock() else {
            return receiver;
        };
        let current = counts.get(&pid).copied().unwrap_or(0);
        let Ok(parks) = self.park_counts.lock() else {
            return receiver;
        };
        if current > 0 && parks.get(&pid).copied() == Some(current) {
            let _ = sender.send(current);
        } else if let Ok(mut observers) = self.park_observers.lock() {
            observers.insert(pid, sender);
        }
        drop(parks);
        drop(counts);
        receiver
    }

    /// Reports the kinds of the frames `pid` has written to its transport,
    /// once its outbound buffer has drained (R3b F8, ruling 2a).
    #[cfg(test)]
    pub(in crate::server::connection) fn record_frames_written(
        &self,
        pid: u64,
        kinds: Vec<liminal::protocol::FrameType>,
    ) {
        if kinds.is_empty() {
            return;
        }
        let observer = self
            .frame_observers
            .lock()
            .ok()
            .and_then(|observers| observers.get(&pid).cloned());
        if let Some(observer) = observer {
            let _ = observer.send(kinds);
        }
    }

    /// Every later report of the frames `pid` writes to its transport.
    #[cfg(test)]
    pub(super) fn observe_frames_written(
        &self,
        pid: u64,
    ) -> Receiver<Vec<liminal::protocol::FrameType>> {
        let (sender, receiver) = channel();
        if let Ok(mut observers) = self.frame_observers.lock() {
            observers.insert(pid, sender);
        }
        receiver
    }

    #[cfg(test)]
    pub(in crate::server::connection) fn record_park(&self, pid: u64) {
        let count = self.slice_count(pid);
        if let Ok(mut parks) = self.park_counts.lock() {
            parks.insert(pid, count);
        }
        if let Ok(mut observers) = self.park_observers.lock()
            && let Some(observer) = observers.remove(&pid)
        {
            let _ = observer.send(count);
        }
    }

    #[cfg(test)]
    pub(super) fn queue_next_outbound_capacity(&self, capacity: usize) {
        if let Ok(mut capacities) = self.next_outbound_capacities.lock() {
            capacities.push_back(capacity);
        }
    }

    #[cfg(test)]
    pub(in crate::server::connection) fn take_next_outbound_capacity(&self) -> Option<usize> {
        self.next_outbound_capacities
            .lock()
            .ok()
            .and_then(|mut capacities| capacities.pop_front())
    }

    #[cfg(test)]
    pub(super) fn install_participant_holdback_pause(&self, pid: u64) -> Receiver<()> {
        let (sender, receiver) = channel();
        if let Ok(mut pauses) = self.participant_holdback_pauses.lock() {
            pauses.insert(pid, sender);
        }
        receiver
    }

    #[cfg(test)]
    pub(in crate::server::connection) fn pause_participant_holdback(&self, pid: u64) -> bool {
        self.participant_holdback_pauses
            .lock()
            .ok()
            .and_then(|mut pauses| pauses.remove(&pid))
            .is_some_and(|sender| {
                let _ = sender.send(());
                true
            })
    }

    /// R7: slices serviced by connection `pid` since spawn (test instrument).
    #[cfg(test)]
    pub(in crate::server::connection) fn slice_count(&self, pid: u64) -> u64 {
        self.slice_counts
            .lock()
            .map_or(0, |counts| counts.get(&pid).copied().unwrap_or(0))
    }

    #[cfg(test)]
    pub(super) fn observe_process_stream_drop(&self, fd: RawFd) -> Receiver<()> {
        let (sender, receiver) = channel();
        if let Ok(mut observers) = self.process_stream_drop_observers.lock() {
            observers.insert(fd, sender);
        }
        receiver
    }

    /// Publishes the process-owned stream's real drop boundary to a waiting test.
    #[cfg(test)]
    pub(in crate::server::connection) fn record_process_stream_drop(&self, fd: RawFd) {
        let observer = self
            .process_stream_drop_observers
            .lock()
            .ok()
            .and_then(|mut observers| observers.remove(&fd));
        if let Some(observer) = observer {
            let _ = observer.send(());
        }
    }
}