minerva 0.2.0

extern crate alloc;

mod capacity;
mod released;

use alloc::vec::Vec;

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

pub use capacity::{AtCapacity, BoundedInsert};
pub use released::Released;

/// The scalar per-source FIFO buffer: an [`Ideal`] over the [`Fifo`] gate. Named, not
/// silently defaulted, the mirror of [`CausalIdeal`].
pub type FifoIdeal<T> = Ideal<T, Fifo>;

/// An *order ideal* of delivered events: a delivery buffer that releases each event only
/// once it is *stable*.
///
/// Accepts [`Event`]s in any order and never delivers one before the events it
/// depends on. The machine is fixed and the *gate* `G` varies: the buffer
/// always holds back, releases the stable frontier in stamp-minimal order, and
/// forgets it, while the gate decides *when* an event is stable.
///
/// Under [`Causal`] the delivered set is downward-closed under
/// happens-before. This is an order ideal in the lattice sense, and each release
/// grows the down-set by one. Under the scalar [`Fifo`] gate it is per-source
/// FIFO. The stamp linearization of the concurrent frontier belongs to the
/// machine rather than the gate, so it is identical under every gate: this is
/// where the total stamp order and a partial stability order interlock
/// (PRD 0005, PRD 0009).
///
/// # The gate
///
/// Against the gate's monotone [`progress`](Self::progress), an event is
/// releasable iff [`Gate::deliverable`], release folds it in via
/// [`Gate::advance`], and one that can never become deliverable is dropped via
/// [`Gate::stale`].
///
/// # The frontier
///
/// Several events can be deliverable at once. They are the ideal's *cover*:
/// the minimal not-yet-delivered elements at its leading edge. They are not the
/// generating antichain `max(I)` of maximal delivered elements, which the
/// buffer never materializes. Release takes the stamp-minimal first, a
/// deterministic linearization that cannot contradict happens-before since
/// `a -> b` implies `a.stamp < b.stamp`. Release order is therefore a function
/// of the event multiset, never of arrival order.
///
/// # Totality
///
/// With the shipped gates, [`insert`](Self::insert) and
/// [`pop_ready`](Self::pop_ready) never fail or panic: out-of-order events
/// wait, duplicates and stale replays are dropped. A caller-defined gate runs
/// inline as trusted mechanism code. If one of its methods panics, the unwind
/// can expose a partially advanced transition. Do not reuse a buffer after you
/// catch such a panic. An event whose dependencies never arrive waits forever,
/// so [`pending_len`](Self::pending_len) is exposed for monitoring; bounding
/// the wait is a deployment concern.
pub struct Ideal<T, G: Gate> {
    /// The gate's monotone progress, advanced only by releasing events. Causal: the
    /// delivered vector `D`, the per-sender count of released events, starting at bottom.
    progress: G::Progress,
    /// Events buffered out of order, awaiting stability. A flat scan finds the next
    /// deliverable; indexing pending by blocking dependency is a later optimization
    /// (PRD 0005).
    pending: Vec<Event<T, G::Dep>>,
}

/// The causal delivery buffer: an [`Ideal`] over the [`Causal`] gate, the PRD 0005 contract.
/// The common case, spelled as an alias so the gate is *named*, not silently defaulted.
pub type CausalIdeal<T> = Ideal<T, Causal>;

impl<T> Ideal<T, Causal> {
    /// Creates an empty causal buffer: nothing delivered, nothing pending.
    #[must_use]
    pub const fn new() -> Self {
        Self {
            progress: VersionVector::new(),
            pending: Vec::new(),
        }
    }

    /// The buffer's delivered vector `D`: the per-sender count of released events. The
    /// causal-specific reading of [`progress`](Self::progress).
    #[must_use]
    pub const fn delivered(&self) -> &VersionVector {
        &self.progress
    }
}

impl<T, G: Gate> Ideal<T, G> {
    /// Advances the gate from an external progress witness.
    ///
    /// Some gates learn progress independently of releasing a buffered
    /// event. An epoch gate, for example, opens when the confirmation
    /// watermark licenses adoption. The caller supplies the same descriptor
    /// the gate would have folded after release; the gate remains the sole
    /// authority for interpreting it and must preserve its monotonicity law.
    pub(crate) fn advance_progress(&mut self, sender: u32, dep: &G::Dep) {
        G::advance(&mut self.progress, sender, dep);
        self.pending
            .retain(|event| !G::stale(&self.progress, event.stamp.station_id(), &event.deps));
    }

    /// Buffers an event for eventual ordered release.
    ///
    /// Out-of-order events wait until they become stable. An event the gate reports as
    /// [`stale`](Gate::stale) (a duplicate or replay that can never become deliverable) is
    /// dropped. Total: never fails.
    pub fn insert(&mut self, event: Event<T, G::Dep>) {
        let sender = event.stamp.station_id();
        if G::stale(&self.progress, sender, &event.deps) {
            return;
        }
        self.pending.push(event);
    }

    /// Releases the next deliverable *event* (the [`Kairos`](crate::kairos::Kairos)-minimal
    /// stable one), advancing the gate's progress. Returns `None` when nothing is currently
    /// deliverable.
    ///
    /// Like [`pop_ready`](Self::pop_ready) but hands back a [`Released`]
    /// witness over the whole [`Event`], `stamp` and dependency `deps`
    /// included, not the payload alone. The witness is minted only after the
    /// selected occurrence is removed and [`Gate::advance`] returns. The order
    /// the buffer just imposed is a *linearization* of a concurrent frontier,
    /// not the causal order itself. The payload stream alone cannot separate
    /// the two cases: an adjacent pair may be causally forced (`a -> b`), or
    /// it may be a `Kairos` tiebreak between concurrent events. A caller that
    /// needs the distinction (a resolver, an audit trail, a downstream
    /// re-orderer) reads `deps` to recover the partial order the total
    /// sequence flattened (`docs/metis-production-recording-decision.adoc`).
    ///
    /// Releasing one event can make others deliverable; a caller drains by looping until this
    /// returns `None`.
    pub fn pop_ready_event(&mut self) -> Option<Released<T, G>> {
        let index = self.next_deliverable()?;
        Some(self.release_index(index))
    }

    /// Releases the [`Kairos`](crate::kairos::Kairos)-minimal event in the current
    /// frontier that `eligible` admits.
    ///
    /// The gate still owns readiness and the buffer still owns progress.
    /// `eligible` sees only events [`Gate::deliverable`] reports ready now: it
    /// may filter that live set, never release a blocked event or advance
    /// progress. With nothing eligible this returns `None` and changes
    /// nothing; an ineligible ready event stays pending for a later call.
    ///
    /// Selection and release happen under one mutable borrow, so no copied
    /// frontier or caller-held token can go stale between them. Among eligible
    /// events the canonical stamp order is preserved. Readiness is
    /// re-evaluated after every predicate call, and if safe interior mutation
    /// changes the frontier, selection restarts in canonical order without
    /// consulting the predicate twice for one event.
    ///
    /// `eligible` is decision policy, not authority: [`Released`] witnesses
    /// the gate transition and does not brand the predicate's reason.
    /// Authorization and irreversible effects stay the caller's.
    /// `O(pending²)` worst case, which is what keeps selection live and
    /// canonical even for a stateful `FnMut` that changes readiness.
    pub fn pop_ready_event_where<F>(&mut self, mut eligible: F) -> Option<Released<T, G>>
    where
        F: FnMut(&Event<T, G::Dep>) -> bool,
    {
        let index = self.next_deliverable_where(&mut eligible)?;
        if !self.is_deliverable(&self.pending[index]) {
            return None;
        }
        Some(self.release_index(index))
    }

    /// Removes one verified-ready event, advances progress, then mints its witness.
    fn release_index(&mut self, index: usize) -> Released<T, G> {
        let event = self.pending.swap_remove(index);
        G::advance(&mut self.progress, event.stamp.station_id(), &event.deps);
        Released::new(event)
    }

    /// Releases the next deliverable payload (the [`Kairos`](crate::kairos::Kairos)-minimal
    /// stable event), advancing the gate's progress. Returns `None` when nothing is currently
    /// deliverable.
    ///
    /// The payload-only projection of [`pop_ready_event`](Self::pop_ready_event); the common
    /// case, when the caller wants the decided stream and not the stamp or dependencies.
    /// Releasing one event can make others deliverable; a caller drains by looping
    /// until this returns `None`.
    pub fn pop_ready(&mut self) -> Option<T> {
        self.pop_ready_event().map(Released::into_payload)
    }

    /// The number of events currently buffered awaiting their dependencies.
    ///
    /// Exposed for bounded-resource monitoring: a steadily growing value signals
    /// dependencies that are not arriving (PRD 0005 names no built-in bound). An *upper* bound
    /// under pre-delivery duplicate floods: the unbounded [`insert`](Self::insert) never prunes a
    /// dead twin, so a count may include provably-dead entries until a full
    /// [`try_insert`](Self::try_insert) reclaims them (sound because
    /// [`Gate::stale`] is terminal).
    #[must_use]
    pub const fn pending_len(&self) -> usize {
        self.pending.len()
    }

    /// The gate's progress state. Causal: the delivered vector `D`, also read via the
    /// domain-named [`delivered`](Ideal::delivered).
    #[must_use]
    pub const fn progress(&self) -> &G::Progress {
        &self.progress
    }

    /// The *frontier*: the events the gate reports deliverable right now, the set it
    /// currently leaves ready.
    ///
    /// Generically the gate's *enabled set*: the buffered events
    /// [`Gate::deliverable`] admits against the current
    /// [`progress`](Self::progress). Epistemic and local --- the minimal
    /// deliverable events *that have arrived* --- never the global minimum of
    /// the undelivered order, since an event minimal in the causal order but
    /// not yet buffered cannot appear.
    ///
    /// Under [`Causal`] it also has the order-theoretic reading: pairwise
    /// concurrent in a duplicate-free history, and the *cover* of the
    /// delivered ideal whose release would extend the down-set. Progress is
    /// then a point in the lattice of consistent cuts and the frontier its
    /// outgoing cover edges --- neither the ideal's interior nor its trailing
    /// `max(I)`. [`Fifo`] has the same reading per source.
    ///
    /// But the [`Gate`] trait exposes no event order, so "antichain" is a
    /// property of *those* gates and not of every `G`: a threshold gate can
    /// enable events that are not pairwise incomparable in any order the
    /// buffer knows (`docs/metis-production-recording-decision.adoc`). At this
    /// point the gate reports nothing further, and the next choice is open.
    /// [`pop_ready_event_where`](Self::pop_ready_event_where) filters this
    /// live set atomically. Richer resolution stays caller policy above that
    /// narrow mechanism (PRD 0008).
    ///
    /// Read-only. [`pop_ready`](Self::pop_ready) releases the stamp-minimal
    /// member of exactly this set. The buffer does not deduplicate, so a
    /// pre-delivery duplicate can appear beside its twin, sharing a dot
    /// rather than being concurrent. `O(pending)`, filtered lazily.
    pub fn frontier(&self) -> impl Iterator<Item = &Event<T, G::Dep>> {
        self.pending
            .iter()
            .filter(|&event| self.is_deliverable(event))
    }

    /// Index of the [`Kairos`]-minimal deliverable pending event, if any. `O(pending)`.
    fn next_deliverable(&self) -> Option<usize> {
        self.pending
            .iter()
            .enumerate()
            .filter(|&(_, event)| self.is_deliverable(event))
            .min_by_key(|&(_, event)| event.stamp)
            .map(|(index, _)| index)
    }

    /// Index of the [`Kairos`]-minimal eligible deliverable event, if any.
    ///
    /// Readiness is live rather than snapshotted. After each predicate call the
    /// scan restarts, retaining that event's decision. This admits newly ready
    /// lower candidates, skips newly blocked candidates, and bounds predicate
    /// calls to once per pending event.
    fn next_deliverable_where<F>(&self, eligible: &mut F) -> Option<usize>
    where
        F: FnMut(&Event<T, G::Dep>) -> bool,
    {
        let mut candidates: Vec<usize> = (0..self.pending.len()).collect();
        candidates.sort_unstable_by_key(|&index| self.pending[index].stamp);
        let mut decisions = Vec::with_capacity(candidates.len());
        decisions.resize(candidates.len(), None);

        loop {
            let mut invoked = false;
            for (&index, decision) in candidates.iter().zip(&mut decisions) {
                if !self.is_deliverable(&self.pending[index]) {
                    continue;
                }

                match *decision {
                    Some(true) => return Some(index),
                    Some(false) => {}
                    None => {
                        *decision = Some(eligible(&self.pending[index]));
                        invoked = true;
                        break;
                    }
                }
            }

            if !invoked {
                return None;
            }
        }
    }

    /// Whether `event` is deliverable against the current progress, as the gate decides.
    fn is_deliverable(&self, event: &Event<T, G::Dep>) -> bool {
        G::deliverable(&self.progress, event.stamp.station_id(), &event.deps)
    }
}

impl<T, G: Gate> Default for Ideal<T, G> {
    /// An empty buffer over the gate, with progress at bottom.
    fn default() -> Self {
        Self {
            progress: G::Progress::default(),
            pending: Vec::new(),
        }
    }
}