minerva 0.2.0

extern crate alloc;

use alloc::vec::Vec;
use core::num::NonZeroUsize;

use super::super::{Causal, Event, Gate, VersionVector};
use super::Ideal;

impl<T, G: Gate> Ideal<T, G> {
    /// Buffers an event, refusing to grow the live backlog past `capacity`.
    ///
    /// [`insert`](Self::insert) buffers unconditionally, so an event whose
    /// dependencies never arrive grows `pending` without bound (PRD 0005 names
    /// no built-in bound). `try_insert` is the additive bounded path beside
    /// that infallible default, the `try_*` escape hatch the error model
    /// describes (target architecture): it caps the *live* backlog and hands a
    /// rejected event back, so the caller applies its own give-up policy
    /// (retry later, shed, escalate). The buffer owns the *bound*; the give-up
    /// *policy* stays the caller's. `capacity` is a property of the *call*,
    /// not a stored field. Each call reads it fresh against the live backlog.
    /// Pass a stable value: a varying value is a caller bug the type cannot
    /// catch. [`pending_len`](Self::pending_len) can exceed a `capacity`
    /// passed on an earlier call.
    ///
    /// A [`stale`](Gate::stale) event (a duplicate or replay) is dropped and
    /// counts against nothing, exactly as in [`insert`](Self::insert). When
    /// the backlog is already at `capacity`, the buffer first reclaims entries
    /// a release has since made [`stale`](Gate::stale). Such an entry is a
    /// pre-delivery duplicate whose twin was already released; reclaiming it
    /// is sound because [`Gate::stale`] is terminal.
    /// The bound therefore counts only events that can still be
    /// delivered. That `O(pending)` reclamation scan is paid only on the full
    /// path, never the common one.
    ///
    /// # Errors
    ///
    /// Returns [`AtCapacity`] (carrying the rejected `event` and the `capacity`) when the live
    /// backlog is still at `capacity` after reclamation. The rejected `event` is not buffered and
    /// the buffer's delivery behavior is unchanged, though the full-path reclamation above may
    /// have dropped provably-dead entries, so [`pending_len`](Self::pending_len) can fall.
    #[must_use = "a rejected event is returned and must be handled"]
    pub fn try_insert(
        &mut self,
        event: Event<T, G::Dep>,
        capacity: NonZeroUsize,
    ) -> Result<(), AtCapacity<T, G::Dep>> {
        let sender = event.stamp.station_id();
        if G::stale(&self.progress, sender, &event.deps) {
            return Ok(());
        }
        if self.pending.len() >= capacity.get() {
            self.prune_stale();
            if self.pending.len() >= capacity.get() {
                return Err(AtCapacity { event, capacity });
            }
        }
        self.pending.push(event);
        Ok(())
    }

    /// Drops pending events a release has made provably-dead.
    ///
    /// Behavior-preserving for delivery because [`Gate::stale`] is terminal: a once-stale event
    /// stays stale and is never deliverable, so reclaiming it on the strength of current
    /// progress changes no delivery outcome. `O(pending)`; called only when a bounded
    /// [`try_insert`](Self::try_insert) finds the backlog full.
    fn prune_stale(&mut self) {
        let progress = &self.progress;
        self.pending.retain(|event| {
            let sender = event.stamp.station_id();
            let stale = G::stale(progress, sender, &event.deps);
            debug_assert!(
                !stale || !G::deliverable(progress, sender, &event.deps),
                "Gate broke stale/deliverable disjointness: a pending event was both",
            );
            !stale
        });
    }
}

impl<T> Ideal<T, Causal> {
    /// Buffers an event, replacing one caller-approved causally maximal resident when full.
    ///
    /// `evictable` observes only candidates whose dot is absent from the causal past of
    /// the input and every other pending event. Candidates are presented in
    /// [`Kairos`](crate::kairos::Kairos) order; the first accepted candidate is replaced
    /// atomically and returned. The operation does not advance delivered progress.
    ///
    /// The returned victim remains the caller's responsibility to retry, quarantine,
    /// reroute, or audit. Exact candidate selection costs `O(pending^2)` time plus
    /// `O(pending)` temporary indices only on the full path.
    ///
    /// Replacement is deliberately absent from non-causal buffers:
    ///
    /// ```compile_fail
    /// use core::num::NonZeroUsize;
    /// use minerva::metis::{Event, FifoIdeal};
    ///
    /// fn replace<T>(buffer: &mut FifoIdeal<T>, event: Event<T, u64>) {
    ///     let capacity = NonZeroUsize::new(1).unwrap();
    ///     let _ = buffer.try_insert_evicting_where(event, capacity, |_| true);
    /// }
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`AtCapacity`] with the unchanged input when no safe candidate
    /// is approved, or when the live backlog still exceeds `capacity` after
    /// reclamation (a `capacity` lowered since an earlier call): eviction
    /// frees one slot, never several, so no candidate is offered. Stale
    /// reclamation may still reduce [`pending_len`](Self::pending_len).
    #[must_use = "the outcome may carry an evicted event or unchanged refused input"]
    pub fn try_insert_evicting_where<F>(
        &mut self,
        event: Event<T>,
        capacity: NonZeroUsize,
        mut evictable: F,
    ) -> Result<BoundedInsert<T>, AtCapacity<T>>
    where
        F: FnMut(&Event<T>) -> bool,
    {
        let sender = event.stamp.station_id();
        if Causal::stale(&self.progress, sender, &event.deps) {
            return Ok(BoundedInsert::DroppedStale);
        }
        if self.pending.len() < capacity.get() {
            self.pending.push(event);
            return Ok(BoundedInsert::Buffered);
        }

        self.prune_stale();
        if self.pending.len() < capacity.get() {
            self.pending.push(event);
            return Ok(BoundedInsert::Buffered);
        }
        if self.pending.len() > capacity.get() {
            return Err(AtCapacity { event, capacity });
        }

        let mut candidates: Vec<usize> = (0..self.pending.len()).collect();
        candidates.sort_unstable_by_key(|&index| self.pending[index].stamp);
        let Some(index) = candidates.into_iter().find(|&index| {
            self.is_causally_maximal(index, &event) && evictable(&self.pending[index])
        }) else {
            return Err(AtCapacity { event, capacity });
        };

        let evicted = self.pending.swap_remove(index);
        self.pending.push(event);
        Ok(BoundedInsert::Evicted { evicted })
    }

    fn is_causally_maximal(&self, candidate: usize, arrival: &Event<T>) -> bool {
        let victim = &self.pending[candidate];
        !Self::depends_on(arrival, victim)
            && self
                .pending
                .iter()
                .enumerate()
                .all(|(index, event)| index == candidate || !Self::depends_on(event, victim))
    }

    fn depends_on(event: &Event<T>, predecessor: &Event<T>) -> bool {
        let predecessor_dot = Self::dot(predecessor);

        Self::dot(event) != predecessor_dot
            && event.deps.get(predecessor_dot.0) >= predecessor_dot.1
    }

    fn dot(event: &Event<T>) -> (u32, u64) {
        let station = event.stamp.station_id();
        (station, event.deps.get(station))
    }
}

/// Outcome of [`CausalIdeal::try_insert_evicting_where`](super::CausalIdeal::try_insert_evicting_where).
///
/// A replaced event is returned because local buffer eviction is not a replicated
/// retraction and disposition remains caller policy.
#[derive(Debug, Clone)]
#[must_use = "an evicted event remains the caller's responsibility"]
pub enum BoundedInsert<T> {
    /// The input now occupies one pending slot.
    Buffered,
    /// The input was already terminally stale and occupies no slot.
    DroppedStale,
    /// The input replaced one caller-approved causally maximal resident.
    Evicted {
        /// The locally removed event, unchanged.
        evicted: Event<T>,
    },
}

/// The rejected outcome of a bounded insert: the backlog stayed at or
/// above `capacity`, so the `event` is handed back unbuffered.
///
/// Above is reachable: a `capacity` lowered since an earlier call finds
/// the backlog already past it.
///
/// Produced by [`Ideal::try_insert`] and
/// [`try_insert_evicting_where`](Ideal::try_insert_evicting_where), after
/// each has reclaimed what a release made provably dead.
///
/// The buffer enforces the *bound*; the give-up *policy* (retry, shed, escalate) is the
/// caller's (PRD 0005). Carrying the rejected event lets the caller re-route it without
/// rebuilding it. Delivery behavior is unchanged, though reclamation may have dropped
/// provably-dead entries (see [`try_insert`](Ideal::try_insert)).
#[derive(Debug, Clone, thiserror::Error)]
#[error("buffer at or above capacity {capacity}: the event was not buffered")]
pub struct AtCapacity<T, D = VersionVector> {
    /// The event that was not buffered.
    pub event: Event<T, D>,
    /// The bound the live backlog had reached.
    pub capacity: NonZeroUsize,
}