minerva 0.2.0

extern crate alloc;

use alloc::collections::{BTreeMap, BTreeSet};
use alloc::vec::Vec;

use super::{Cut, VersionVector};

mod abandonment;
mod departure;
mod unknown;

pub use abandonment::{AbandonRefusal, Abandoned};
pub use departure::{Departed, Departure, DepartureRefusal, Fenced};
pub use unknown::UnknownStation;

/// A cross-node causal-stability tracker: the meet of every roster member's
/// delivered cut, the *stability watermark* `min_n D_n` (PRD 0011).
///
/// Each member reports its delivered cut; the tracker folds each member's
/// reports by join and exposes their greatest lower bound. The
/// [`watermark`](Self::watermark) is the greatest cut *every* member has
/// passed, so an event at or below it is delivered everywhere and safe to
/// forget. This is the matrix-clock garbage-collection quantity over Mattern's
/// lattice of consistent cuts.
///
/// # The roster is the meaning
///
/// A meet is only as trustworthy as the family it ranges over, and the
/// failure direction is unrecoverable: a join error self-heals, but
/// forgetting below a wrongly-high watermark loses state a peer still needs.
/// So the roster is fixed at construction and off-roster reporters are
/// refused ([`UnknownStation`]); meeting over "peers heard from so far"
/// would move the family mid-flight. A member that stops reporting pins the
/// watermark at its last cut: the availability cost of a meet, borne
/// deliberately.
///
/// [`abandon`](Self::abandon) is the one door that narrows that family, and
/// it narrows the *family* rather than the roster, irreversibly. Arrival
/// still means a new tracker over a new roster
/// (`docs/metis-membership-departure.adoc`).
///
/// # Two things a report claims, and neither is checkable
///
/// *A cut claim.* A reported vector asserts a gap-free per-station prefix:
/// "I hold everything below these counts." A max-style high-water that runs
/// past holes is an upper bound of a cut, not a cut, and meeting upper
/// bounds over-claims stability. A caller whose receipt may hold holes
/// reports [`DotSet::floor`](crate::metis::DotSet::floor) instead.
///
/// *A durability claim.* The join that makes reporting order-robust also masks
/// amnesia. A member can crash, restart with less, and honestly re-report its
/// regressed cut. The tracker absorbs that report silently. The watermark may
/// already have licensed irreversible action on the higher claim. So report
/// only a cut that would survive a crash-restart. Report the durable floor,
/// never the delivered one. The recipe discharging that duty is the fleet's
/// rehydration recipe (PRD 0024, ruling R-51).
///
/// The tracker can verify neither (PRD 0011 R8), which is why both are
/// stated here rather than enforced below.
///
/// # Mechanism, not policy
///
/// The tracker computes the greatest safe cut and itself forgets nothing.
/// What is reclaimed below the watermark, and how reports travel, are the
/// caller's. The retention face of the bounded-resources boundary, beside
/// the backlog bound (`try_insert`) and the skew bound (`try_observe`).
///
/// # Totality
///
/// [`report`](Self::report) folds by join, so it is idempotent,
/// commutative, and monotone: duplicated, reordered, and stale reports
/// absorb, and the watermark never regresses. A genuinely regressed member
/// is a membership event the caller handles by rebuilding; the tracker will
/// not un-vouch on its behalf.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Stability {
    /// Per roster member, the join of every cut it has reported; bottom until the
    /// first report.
    reports: BTreeMap<u32, VersionVector>,
    /// Members the caller has abandoned: still on the roster, no longer in
    /// the family the meet ranges over. Monotone, and a subset of
    /// `reports`' keys by [`abandon`](Self::abandon)'s own refusal.
    abandoned: BTreeSet<u32>,
    /// Abandoned members whose slot absorbed a report afterwards. Nothing
    /// derived reads it; see [`resurgent`](Self::resurgent) for how narrow
    /// the claim is.
    resurgent: BTreeSet<u32>,
    /// Abandoned members whose departure carried a sealed [`Departed`], and
    /// the bound it agreed. A subset of `abandoned` by
    /// [`abandon_attested`](Self::abandon_attested)'s own refusal, and the
    /// only membership state anything outside this tracker consumes: the
    /// epoch rounds substitute this coordinate for the testimony the
    /// departed member will never give.
    attested: BTreeMap<u32, Departed>,
    /// Members whose slot has ever absorbed a bare charter
    /// [`report`](Self::report) rather than a witnessed
    /// [`report_cut`](Self::report_cut). Per station rather than one global
    /// flag, so the witness narrows with the family exactly as the meet
    /// does: the closure theorem asks only about the slots the meet ranges
    /// over, and abandoning the one member that reported bare restores
    /// [`watermark_cut`](Self::watermark_cut) rather than freezing it
    /// forever. Within a member it is still permanent --- a join cannot
    /// un-see an unwitnessed vector.
    unwitnessed: BTreeSet<u32>,
}

impl Stability {
    /// Creates a tracker over a fixed roster of `station_id`s, every member at
    /// bottom (nothing vouched, nothing stable). Duplicate ids collapse.
    ///
    /// An empty roster is degenerate: see [`watermark`](Self::watermark) for why it
    /// answers bottom rather than the vacuous "everything is stable".
    #[must_use]
    pub fn new(roster: impl IntoIterator<Item = u32>) -> Self {
        Self {
            reports: roster
                .into_iter()
                .map(|station| (station, VersionVector::new()))
                .collect(),
            abandoned: BTreeSet::new(),
            resurgent: BTreeSet::new(),
            attested: BTreeMap::new(),
            unwitnessed: BTreeSet::new(),
        }
    }

    /// Folds a roster member's reported delivered cut into its slot, by join.
    ///
    /// The fold absorbs rather than replaces, which makes reporting
    /// order-robust. A stale or duplicated report --- a gossip replay, a
    /// reordered channel --- can never lower a member's vouched cut. The
    /// watermark is therefore monotone non-decreasing, and that is what
    /// licenses acting on it.
    ///
    /// # Errors
    ///
    /// Refuses a `station` that is not on the roster, returning [`UnknownStation`]
    /// and leaving the tracker unchanged: an unheralded reporter is a membership
    /// change, the caller's policy.
    pub fn report(
        &mut self,
        station: u32,
        delivered: &VersionVector,
    ) -> Result<(), UnknownStation> {
        // The charter boundary (PRD 0011 R8): a bare vector is a *claim*, not a
        // witnessed cut, so folding one clears the witnessed gate for good. The
        // watermark stays correct; only watermark_cut's honesty is withdrawn.
        self.fold(station, delivered)?;
        let _ = self.unwitnessed.insert(station);
        Ok(())
    }

    /// Folds a roster member's reported cut into its slot, by join, *carrying
    /// the witness* to the tracker boundary instead of laundering through
    /// [`Cut::as_vector`].
    ///
    /// The witnessed twin of [`report`](Self::report). Taking a [`Cut`]
    /// rather than a bare vector lets the tracker know the slot holds a join
    /// of cuts, which is what [`watermark_cut`](Self::watermark_cut) reads.
    /// Both doors agree on the resulting watermark; only this one keeps the
    /// gap-freedom proof.
    ///
    /// # Errors
    ///
    /// [`UnknownStation`] for an off-roster station, tracker unchanged.
    pub fn report_cut(&mut self, station: u32, cut: &Cut) -> Result<(), UnknownStation> {
        self.fold(station, cut.as_vector())
    }

    /// The shared join fold, private so the witnessed gate is set only by the
    /// two public doors and never bypassed.
    fn fold(&mut self, station: u32, delivered: &VersionVector) -> Result<(), UnknownStation> {
        let vouched = self
            .reports
            .get_mut(&station)
            .ok_or(UnknownStation { station })?;
        *vouched = vouched.merge(delivered);
        // Inert, never refused: the fold lands in a slot nothing reads, so
        // the door stays total for a transport that cannot know a peer was
        // evicted. Recorded anyway, for the little it is worth: see
        // `resurgent` for why observing a report is not a verdict.
        if self.abandoned.contains(&station) {
            let _ = self.resurgent.insert(station);
        }
        Ok(())
    }

    /// The stability watermark: the greatest cut every roster member has passed,
    /// the n-ary [`meet`](VersionVector::meet) of the members' vouched cuts.
    ///
    /// The watermark is itself a consistent cut, because cuts are closed under
    /// pointwise min. It is monotone non-decreasing across
    /// [`report`](Self::report) and [`abandon`](Self::abandon). `abandon` only
    /// ever removes a floor. A surviving member still at bottom pins the
    /// watermark there.
    ///
    /// An *empty* family's mathematical meet is the lattice top --- vacuously
    /// "everything is stable" --- which a safe-to-forget quantity must never
    /// invent, so the degenerate answer is bottom. `O(roster x entries)`,
    /// computed on read.
    #[must_use]
    pub fn watermark(&self) -> VersionVector {
        let mut vouched = self.surviving().map(|(_, cut)| cut);
        let Some(first) = vouched.next() else {
            return VersionVector::new();
        };
        vouched.fold(first.clone(), |acc, cut| acc.meet(cut))
    }

    /// The surviving stations alone, ascending: the arrival round's
    /// family read (PRD 0028 R2). Crate-private: the epoch module fixes a
    /// round's family from the same set the meet ranges over, so an
    /// attested or abandoned member is not asked to endorse.
    pub(crate) fn surviving_family(&self) -> impl Iterator<Item = u32> + '_ {
        self.surviving().map(|(station, _)| station)
    }

    /// The family the meet actually ranges over: roster members not
    /// abandoned, ascending. The one place the narrowing is applied, so
    /// every derived quantity narrows together or not at all.
    fn surviving(&self) -> impl Iterator<Item = (u32, &VersionVector)> {
        self.reports
            .iter()
            .filter(|(station, _)| !self.abandoned.contains(*station))
            .map(|(&station, cut)| (station, cut))
    }

    /// The stability watermark *as a witnessed [`Cut`]*, or `None` if any
    /// *surviving* member's report arrived unwitnessed.
    ///
    /// A meet of witnessed cuts is itself a cut by closure, so over a
    /// witnessed family the watermark carries its gap-freedom proof and this
    /// hands it back branded --- ready to floor a delta or seed another
    /// tracker without re-witnessing.
    ///
    /// The `Option` is the closure theorem's precondition. The bare
    /// [`report`](Self::report) door accepts a vector the tracker cannot
    /// verify. Wrapping a mixed family's meet as a `Cut` would forge a
    /// gap-freedom witness from that unverified vector. That is exactly the
    /// laundering [`Cut`] exists to refuse. A caller wanting a guaranteed cut
    /// reports only through [`report_cut`](Self::report_cut); one mixing the
    /// two reads [`watermark`](Self::watermark) and carries the burden itself.
    /// The value inside `Some` always equals [`watermark`](Self::watermark);
    /// only the witness is conditional.
    ///
    /// An empty family yields `Some(Cut::bottom())`: bottom is a genuine
    /// cut, and never the vacuous lattice top.
    #[must_use]
    pub fn watermark_cut(&self) -> Option<Cut> {
        self.surviving()
            .all(|(station, _)| !self.unwitnessed.contains(&station))
            .then(|| Cut::from_witnessed(self.watermark()))
    }

    /// Iterates the roster's `(station_id, vouched cut)` slots in ascending station
    /// order, unreported members at bottom.
    ///
    /// The observability read: the member pinning the watermark (the straggler a
    /// caller may want to page, or evict via a new roster) is visible here.
    pub fn reports(&self) -> impl Iterator<Item = (u32, &VersionVector)> {
        self.reports.iter().map(|(&station, cut)| (station, cut))
    }

    /// Removes `station` from the family the meet ranges over, permanently,
    /// and hands back what the decision writes off.
    ///
    /// That member no longer pins [`watermark`](Self::watermark). This
    /// unfreezes a fleet stalled on a silent member, and it is also the whole
    /// of the danger: the meet now licenses forgetting state the member never
    /// delivered. Sound exactly when it is never coming back, which nobody can
    /// check --- so the door returns the loss instead of pretending.
    /// Idempotent in state; the returned bound is recomputed each call and
    /// rises with the survivors' join.
    ///
    /// Complete for the family a meet ranges over, and *incomplete* for the
    /// record an epoch consignment binds over: a departed member's own
    /// window traffic is no longer covered by a join built from the
    /// family's self-reports, and the consignment door fails closed on it.
    /// [`abandon_attested`](Self::abandon_attested) is the door that closes
    /// that half; this one narrows the meet and nothing else.
    ///
    /// # Errors
    ///
    /// [`AbandonRefusal::UnknownStation`] off the roster;
    /// [`AbandonRefusal::LastSurvivor`] for the family's last member.
    pub fn abandon(&mut self, station: u32) -> Result<Abandoned, AbandonRefusal> {
        let bound = self.narrow(station, None)?;
        Ok(Abandoned::new(station, bound))
    }

    /// Narrows the family *and* records the survivors' agreed bound, so the
    /// epoch rounds may substitute it for the departed member's testimony.
    ///
    /// The attested twin of [`abandon`](Self::abandon), and the only
    /// difference between them is what the tracker may hand onwards. A local
    /// bound is a number honest survivors disagree about, so nothing in the
    /// protocol may consume it; a [`Departed`] is the outcome of a round
    /// every survivor ran, so every survivor's seal record agrees at the
    /// departed coordinate. Both doors narrow identically, and the returned
    /// [`Abandoned`] still prices the departure *locally* --- it is the
    /// operator's read, never the protocol's.
    ///
    /// The attestation is enforced outside this tracker: the caller admits no
    /// dot above it ([`Departure::admits`]). A recorded bound with no fence
    /// behind it is the one way this door can lie, which is why the
    /// attestation is minted only by a completed round.
    ///
    /// Idempotent at the same bound, and refused at another: a round seals
    /// once, and re-attesting elsewhere would move a coordinate this
    /// replica's peers may already have sealed over.
    ///
    /// # Errors
    ///
    /// [`AbandonRefusal::UnknownStation`] off the roster;
    /// [`AbandonRefusal::LastSurvivor`] for the family's last member;
    /// [`AbandonRefusal::FamilyMismatch`] when the attestation ranged over a
    /// family that is not this tracker's, since "every survivor proposed" is
    /// a claim about *which* survivors; and [`AbandonRefusal::Reattested`]
    /// for a second attestation at a different bound. The tracker is
    /// unchanged in every case, and only this door raises the last two.
    pub fn abandon_attested(&mut self, departed: &Departed) -> Result<Abandoned, AbandonRefusal> {
        // Ordered so each refusal answers the question a caller asked. Off
        // the roster first, because nothing else is meaningful about a
        // non-member; then the latch, because a second attestation at another
        // bound is a contradiction whatever family it names; then the family,
        // which is a fact about this attestation rather than about the
        // station.
        let station = departed.station();
        if !self.reports.contains_key(&station) {
            return Err(AbandonRefusal::UnknownStation { station });
        }
        if let Some(held) = self.attested(station)
            && held != departed.bound()
        {
            return Err(AbandonRefusal::Reattested {
                station,
                held,
                offered: departed.bound(),
            });
        }
        let bound = self.narrow(station, Some(departed))?;
        let _ = self.attested.insert(station, departed.clone());
        Ok(Abandoned::new(station, bound))
    }

    /// The narrowing both departure doors share, returning the local
    /// write-off bound read *before* the family narrows.
    fn narrow(
        &mut self,
        station: u32,
        installing: Option<&Departed>,
    ) -> Result<VersionVector, AbandonRefusal> {
        if !self.reports.contains_key(&station) {
            return Err(AbandonRefusal::UnknownStation { station });
        }
        // Idempotence needs no special case, and it is worth seeing why
        // rather than guarding it: every condition below is about the family
        // that *remains*, which is `surviving` minus `station` --- and for a
        // station already abandoned that is `surviving` itself. A repeat
        // therefore evaluates each of them over exactly the set the first
        // call left behind, and answers the same.
        if self.surviving().all(|(surviving, _)| surviving == station) {
            return Err(AbandonRefusal::LastSurvivor { station });
        }
        // An attested bound is a promise that some survivor can still serve
        // everything below it. The condition is about who *remains*, not
        // about who leaves: a report is a durable floor and lags what a
        // member holds, so "the leaver had not reported that high" proves
        // nothing about whether it was the only holder. What must be true
        // after the narrowing is that somebody left behind reports through
        // the bound.
        // Whether an attestation is the one this family would have agreed,
        // checked after the family question above because an empty remainder
        // binds nothing and `LastSurvivor` is the truer answer there.
        if let Some(departed) = installing
            && !departed.binds(
                self.surviving()
                    .map(|(surviving, _)| surviving)
                    .filter(|&surviving| surviving != station),
            )
        {
            return Err(AbandonRefusal::FamilyMismatch { station });
        }
        let remaining: Vec<u32> = self
            .surviving()
            .map(|(surviving, _)| surviving)
            .filter(|&surviving| surviving != station)
            .collect();
        if let Some((&stranded, _)) = self
            .attested
            .iter()
            .find(|(_, departed)| !self.servable(departed, remaining.iter().copied()))
        {
            return Err(AbandonRefusal::StrandsAttestation { station, stranded });
        }
        let bound = self.abandonment_bound(station);
        let _ = self.abandoned.insert(station);
        Ok(bound)
    }

    /// The agreed bound for an attested departure, or `None` for a station
    /// that is present, unattested, or off the roster.
    ///
    /// The substitution read: a round that reads it supplies this coordinate
    /// itself instead of waiting for a report that is not coming.
    #[must_use]
    pub fn attested(&self, station: u32) -> Option<u64> {
        self.attested.get(&station).map(Departed::bound)
    }

    /// Whether some member of `family` can still serve `departed`'s bound:
    /// one that proposed through it in the round that agreed it, or one whose
    /// report has since reached it.
    ///
    /// Two arms because they answer at different times. A proposal is
    /// evidence of holding at round time, which no report can supply
    /// afterwards; a report is evidence of holding *now*, which is how the
    /// condition clears once the repair lane has carried the traffic to
    /// somebody else. A bottom bound is served by everyone.
    fn servable(&self, departed: &Departed, family: impl IntoIterator<Item = u32>) -> bool {
        let bound = departed.bound();
        bound == 0
            || family.into_iter().any(|member| {
                departed
                    .proposal_of(member)
                    .is_some_and(|held| held >= bound)
                    || self.reports[&member].get(departed.station()) >= bound
            })
    }

    /// The abandonment bound at `station`: the join of every *other*
    /// surviving member's vouched cut, restricted to `station`'s coordinate.
    ///
    /// Readable before the decision, which is the point: it prices an
    /// eviction. See [`Abandoned`] for what the number means.
    ///
    /// Bottom for a station off the roster, and that needs the explicit
    /// guard below rather than falling out: a member's reported cut is over
    /// whatever dots it delivered, so a survivor's vector can carry
    /// coordinates for stations this tracker never had on its roster, and
    /// restricting to one of those would price a departure that cannot
    /// happen.
    #[must_use]
    pub fn abandonment_bound(&self, station: u32) -> VersionVector {
        if !self.reports.contains_key(&station) {
            return VersionVector::new();
        }
        self.surviving()
            .filter(|&(surviving, _)| surviving != station)
            .fold(VersionVector::new(), |join, (_, cut)| join.merge(cut))
            .restrict([station])
    }

    /// Whether `station` has been abandoned. False for a station off the
    /// roster, which was never in the family to leave it.
    #[must_use]
    pub fn is_abandoned(&self, station: u32) -> bool {
        self.abandoned.contains(&station)
    }

    /// The abandoned members, ascending: the roster minus the family the
    /// meet ranges over.
    pub fn abandoned(&self) -> impl Iterator<Item = u32> + '_ {
        self.abandoned.iter().copied()
    }

    /// Abandoned members whose slot has absorbed a report *after* this
    /// tracker abandoned them, ascending.
    ///
    /// Read the claim exactly: a report was **observed** here after local
    /// abandonment. It is not a verdict that the member is alive. A report
    /// already in flight when the operator decided produces an entry with
    /// nothing wrong, and traffic on any other channel --- an epoch
    /// declaration, say --- produces none, because this tracker never sees
    /// it. So a non-empty answer is a prompt to look, and an empty one
    /// proves nothing. A verdict needs a fenced post-departure signal,
    /// which is charter work (`docs/metis-membership-departure.adoc`).
    ///
    /// It un-abandons nothing (nothing can) and is not an error.
    pub fn resurgent(&self) -> impl Iterator<Item = u32> + '_ {
        self.resurgent.iter().copied()
    }
}