minerva 0.2.0

//! Inbound dispatch, replay classification, and protocol-note handling.

extern crate alloc;

use alloc::vec::Vec;

use crate::kairos::Kairos;
use crate::metis::{
    Admission, ArrivalRefusal, Cut, Declaration, Dot, Endorsement, EpochAddress, EpochRefusal,
    Event, SealedEpoch, Vouched,
};

use super::super::{Note, OldDelta};
use super::{Probe, Replica, fold_native};

impl Replica {
    // ---- the inbound door --------------------------------------------------

    pub fn handle(&mut self, note: &Note, out: &mut Vec<Note>) {
        // Every received note that taught this replica anything enters
        // the volatile buffer: it becomes durable at the next fence (no
        // later than this replica's next emission), and a crash before
        // that fence loses it, which is exactly the loss the repair lane
        // is scoped to cure. Fencing is *effect-based* (the
        // bounded-record rule): a data note buffers only when it carries
        // a novel event dot, and a protocol note only when dispatching
        // it changed the round state it addresses or parked for later,
        // so redelivery, duplication, and repeated repair rounds never
        // compound the journal. The probes are exact, not heuristic: the
        // trackers are compared whole, so an absorbed join, a duplicate
        // candidate, or a stale acknowledgement fences nothing.
        let parked_before = self.parked.len();
        let probe = match note {
            // A note the fence refuses teaches this replica nothing, so it
            // must not enter the journal either: a restart replaying it
            // against a fence rebuilt at bottom would fold exactly the
            // traffic the live fence rejected.
            Note::Old { .. } | Note::Native { .. } => {
                Probe::Data(self.data_note_novel(note) && !self.note_refused(note))
            }
            Note::Report { .. } => Probe::Stability(self.stability.clone()),
            Note::Retire { .. } => Probe::Retirement(self.retirement.clone()),
            Note::Declare { .. }
            | Note::Confirm { .. }
            | Note::Adoption { .. }
            | Note::Endorse { .. } => Probe::Epochs(alloc::boxed::Box::new(self.epochs.clone())),
            Note::Depart { .. } => Probe::Departures(self.departures.clone()),
        };
        self.dispatch(note);
        let effectful = match probe {
            Probe::Data(novel) => novel,
            Probe::Stability(before) => before != self.stability,
            Probe::Retirement(before) => before != self.retirement,
            Probe::Epochs(before) => *before != self.epochs,
            Probe::Departures(before) => before != self.departures,
        } || self.parked.len() > parked_before;
        if effectful {
            self.volatile.push(note.clone());
        }
        self.drive(out);
    }

    /// Alternates the monotone drives with parked-note retries until
    /// neither makes progress: a poll may seal and open the very
    /// generation a parked note was waiting for, and a retried note may
    /// enable the next poll step, so a single pass of either strands
    /// the other's enablements. Shared with the operator doors whose act
    /// is itself an unparking event --- `open_arrival` above all, since a
    /// parked peer endorsement waits on exactly this replica's own round
    /// opening.
    pub(super) fn drive(&mut self, out: &mut Vec<Note>) {
        loop {
            self.poll(out);
            if self.parked.is_empty() {
                break;
            }
            let parked: Vec<Note> = core::mem::take(&mut self.parked);
            let before = parked.len();
            for note in &parked {
                self.dispatch(note);
            }
            if self.parked.len() == before {
                break;
            }
        }
    }

    fn dispatch(&mut self, note: &Note) {
        match note {
            Note::Old {
                generation,
                dots,
                delta,
            } => self.on_old(*generation, dots, delta, note),
            Note::Native {
                epoch,
                stamp,
                dots,
                delta,
            } => self.on_native(*epoch, *stamp, dots, delta),
            Note::Report {
                generation,
                station,
                cut,
            } => match generation.cmp(&self.generation) {
                core::cmp::Ordering::Equal => self
                    .stability
                    .report_cut(*station, cut)
                    .expect("every reporter is a roster member"),
                core::cmp::Ordering::Greater => self.park(note),
                core::cmp::Ordering::Less => {}
            },
            Note::Retire {
                generation,
                station,
                applied,
                at,
            } => match generation.cmp(&self.generation) {
                core::cmp::Ordering::Equal => {
                    // The gathering gate: the acknowledgement counts only
                    // once this replica has delivered the acker's floor.
                    // Every write the acker minted while a removed element
                    // was still visible to it precedes its acknowledgement
                    // (own mints are in the own floor), so covering `at`
                    // proves no in-flight write can still anchor through a
                    // dot the meet would license; an early-counted
                    // acknowledgement is exactly the stranded-anchor
                    // divergence the boundary pin exhibits.
                    if at
                        .as_vector()
                        .partial_cmp(&self.have.floor())
                        .is_some_and(core::cmp::Ordering::is_le)
                    {
                        self.retirement
                            .acknowledge(*station, applied)
                            .expect("every acker is a roster member");
                    } else {
                        self.park(note);
                    }
                }
                core::cmp::Ordering::Greater => self.park(note),
                // A sealed generation's retirement evidence is void: the
                // re-foundation swept its retired identity wholesale.
                core::cmp::Ordering::Less => {}
            },
            Note::Depart {
                generation,
                departing,
                by,
                prefix,
            } => match generation.cmp(&self.generation) {
                core::cmp::Ordering::Equal => self.on_depart(*departing, *by, *prefix),
                core::cmp::Ordering::Greater => self.park(note),
                // A sealed generation's proposal is void: the bound it
                // agreed is a counter in a dot space the re-foundation
                // re-minted (`Departed::refounded`).
                core::cmp::Ordering::Less => {}
            },
            Note::Endorse {
                by,
                admission,
                commitment,
            } => self.on_endorse(*by, admission, *commitment, note),
            Note::Declare { declaration } => self.on_declare(declaration, note),
            Note::Confirm {
                generation,
                epoch,
                station,
                cut,
            } => self.on_confirm(*generation, *epoch, *station, cut, note),
            Note::Adoption {
                generation,
                epoch,
                station,
                counter,
            } => self.on_adoption(*generation, *epoch, *station, *counter, note),
        }
    }

    /// Folds a survivor's departure proposal, opening the round if this
    /// replica has not. Its own proposal is the operator's, never a peer's:
    /// hearing that somebody else is evicting a member does not evict it
    /// here, so the round completes only where every survivor's operator
    /// agreed.
    fn on_depart(&mut self, departing: u32, by: u32, prefix: u64) {
        let own = self.id;
        let Ok(round) = self.enter_departure(departing) else {
            return;
        };
        // An own proposal reaches this door only in replay: the fabric never
        // delivers a note back to its minter. Re-installing it at its
        // *recorded* value rather than re-deriving it from rebuilt state is
        // the whole point of journaling it --- peers priced their bound on
        // the number this replica emitted, and a restart may not lower it
        // (`Departure::reinstate`).
        let folded = if by == own {
            round.reinstate(prefix)
        } else {
            round.fold(&Vouched::trust(by, prefix))
        };
        if folded.is_err() {
            return;
        }
        self.settle_departures();
    }

    /// Folds a peer's arrival endorsement, or re-opens this replica's own
    /// round in replay. The boundary names its plane itself: an admission
    /// over the base sealed at generation `g` addresses `g + 1`'s window.
    ///
    /// The refusal ladder, classified for the harness. A word for a plane
    /// this replica sealed past is void (the seal superseded the round, so
    /// a stale endorsement is structurally unspendable). A word ahead of
    /// this replica parks, like any note that outran its record. At the
    /// current plane, `NoRound` parks --- opening is the operator's policy
    /// act, so the peer's word waits for it --- and `AttestedDeparture`
    /// drops: the freeze, a departed identity's word is not spendable.
    /// `UnknownStation` drops too, because abandonment is per-replica
    /// configuration: a member another operator abandoned locally still
    /// speaks here with a wider family, and its word for a slot this
    /// round does not hold teaches nothing (the R-89 counterweight's
    /// heterogeneous-family schedule, reachable honestly). Everything
    /// else is dishonest under an honest fabric and stays a panic.
    fn on_endorse(&mut self, by: u32, admission: &Admission, commitment: [u8; 32], note: &Note) {
        let window = admission.base().generation() + 1;
        match window.cmp(&self.generation) {
            core::cmp::Ordering::Greater => self.park(note),
            core::cmp::Ordering::Less => {}
            core::cmp::Ordering::Equal if by == self.id => {
                // Own word, so this is the rehydration replay (the fabric
                // never delivers a note back to its minter): re-open the
                // round exactly as the journaled mint did.
                let _ = self
                    .epochs
                    .open_arrival(admission.clone(), commitment, self.id, &self.stability)
                    .expect("a journaled endorsement re-opens its own round");
            }
            core::cmp::Ordering::Equal => {
                let claim = Endorsement::from_parts(admission.clone(), commitment);
                match self
                    .epochs
                    .fold_endorsement(&Vouched::trust(by, claim), &self.stability)
                {
                    // A fold, a dropped departed word (the freeze), and a
                    // dropped foreign-family word are all terminal here;
                    // only `NoRound` waits, on this operator's own act.
                    Ok(())
                    | Err(
                        ArrivalRefusal::AttestedDeparture { .. }
                        | ArrivalRefusal::UnknownStation { .. },
                    ) => {}
                    Err(ArrivalRefusal::NoRound) => self.park(note),
                    Err(refusal) => panic!(
                        "replica {}: unexpected endorsement refusal {refusal:?}",
                        self.id
                    ),
                }
            }
        }
    }

    fn on_old(&mut self, generation: u64, dots: &[Dot], delta: &OldDelta, note: &Note) {
        let refused = if generation > self.generation {
            // A sealed peer's next-generation traffic is judged in the plane
            // it addresses, not this one.
            self.fenced_native(dots)
        } else {
            self.fenced(dots)
        };
        if refused.is_some() {
            // The fence: nothing of a departing member's above the promise
            // this replica made enters any plane. Dropping is the whole
            // mechanism --- a delta that never folds cannot be in a shadow
            // the consignment door then finds above the seal.
            return;
        }
        if generation == self.generation + 1 {
            // The sender sealed first. The seal it performed proves every
            // member adopted, so this replica holds the next plane already.
            let transition = self
                .transition
                .as_mut()
                .expect("a peer's post-seal traffic implies local adoption");
            fold_native(transition, dots, delta, None);
            return;
        }
        assert!(
            generation <= self.generation + 1,
            "replica {}: no honest sender runs two generations ahead",
            self.id
        );
        if generation < self.generation {
            self.recognize_replay(generation, dots, note);
            return;
        }
        // The bounded-record rule: a redelivered data note still folds
        // (the pair's merge is idempotent, and a condensed tombstone may
        // lawfully re-learn from a replay: the hygiene lifecycle's
        // cycle), but only a novel note appends to the log; without this
        // gate, redelivery and repeated repair rounds would compound the
        // record geometrically across crash cycles.
        let novel = dots.iter().any(|&dot| !self.recorded.contains(dot));
        for &dot in dots {
            let _ = self.have.insert(dot);
            let _ = self.recorded.insert(dot);
            // Replaying an own note advances the allocator exactly as
            // minting it did: the rehydration replay routes own mints
            // through this door, and the basis must be current before
            // any journaled protocol note re-runs `adopt` (the adoption
            // report carries the counter). Live traffic never delivers a
            // note back to its minter, so this arm fires only in replay.
            if dot.station() == self.id {
                self.counter = self.counter.max(dot.counter());
            }
        }
        if let Some(transition) = self.transition.as_mut() {
            match delta {
                OldDelta::Text(delta) => {
                    transition
                        .shadow
                        .deliver_text(delta)
                        .expect("honest window text folds through the shadow");
                }
                OldDelta::Moves(delta) => {
                    transition
                        .shadow
                        .deliver_moves(delta)
                        .expect("honest window movement folds through the shadow");
                }
            }
        } else {
            self.fold_current(delta);
            if novel {
                self.log.push((dots.to_vec(), delta.clone()));
            }
        }
    }

    fn recognize_replay(&self, generation: u64, dots: &[Dot], note: &Note) {
        let address = self
            .epochs
            .sealed()
            .find(|sealed| sealed.declaration().generation() == generation)
            .map(SealedEpoch::declaration);
        // Below the retained horizon the machine refuses to guess; the
        // beyond-horizon scenario asserts that refusal explicitly.
        let Some(address) = address else { return };
        for &dot in dots {
            self.epochs
                .recognize(address, dot)
                .unwrap_or_else(|refusal| {
                    panic!(
                        "replica {}: an honest old replay within the horizon \
                     must read as a duplicate, got {refusal:?} for {note:?}",
                        self.id
                    )
                });
        }
    }

    fn on_native(&mut self, epoch: EpochAddress, stamp: Kairos, dots: &[Dot], delta: &OldDelta) {
        if self.fenced_native(dots).is_some() {
            // The successor plane's fence. A departing member's coordinate
            // there is bottom, so none of its native traffic may enter the
            // transition and ride the seal into the next generation's base
            // --- which would sit above the substituted coordinate the next
            // round's record names.
            return;
        }
        if let Some(transition) = self.transition.as_mut()
            && transition.adopted.address() == epoch
        {
            // Own native mints replayed at rehydration advance the native
            // allocator, like own old mints through `on_old`.
            for &dot in dots {
                if dot.station() == self.id {
                    transition.native_counter = transition.native_counter.max(dot.counter());
                }
            }
            fold_native(transition, dots, delta, Some(stamp));
            return;
        }
        // A native note that does not name the held transition falls
        // through rather than asserting: under scaffold-free schedules a
        // duplicated note from an earlier, already-sealed window lawfully
        // redelivers after this replica opened a later transition (the
        // S289 tape's second catch; the old assert declared that lawful
        // redelivery dishonest), and the sealed branch below already owns
        // exactly that verdict. A note naming a window this replica has
        // not reached gates like any pre-adoption arrival.
        if self
            .epochs
            .sealed()
            .any(|sealed| sealed.declaration() == epoch)
        {
            if epoch.generation() + 1 == self.generation {
                // That transition sealed here: its native plane is this
                // replica's current old plane. The bounded-record rule
                // applies as in `on_old`: duplicates fold, only novelty
                // appends.
                let novel = dots.iter().any(|&dot| !self.recorded.contains(dot));
                for &dot in dots {
                    let _ = self.have.insert(dot);
                    let _ = self.recorded.insert(dot);
                }
                self.fold_current(delta);
                if novel {
                    self.log.push((dots.to_vec(), delta.clone()));
                }
            }
            // An older sealed plane's native replay is that generation's
            // old traffic; the Old lane's recognizer owns the verdict.
            return;
        }
        // Not adopted yet: hold it behind the winner's address, one copy
        // per event (the bounded-record rule for the pre-adoption gate,
        // which would otherwise hold and re-release duplicates).
        let novel = dots.iter().any(|&dot| !self.gated.contains(dot));
        if !novel {
            return;
        }
        for &dot in dots {
            let _ = self.gated.insert(dot);
        }
        self.gate.insert(Event {
            stamp,
            deps: epoch,
            payload: (dots.to_vec(), delta.clone()),
        });
    }

    fn on_declare(&mut self, declaration: &Declaration, note: &Note) {
        if self.fenced(&[declaration.dot()]).is_some() {
            return;
        }
        let generation = declaration.address().generation();
        if generation > self.generation {
            self.park(note);
            return;
        }
        if generation < self.generation {
            self.epochs
                .deliver(declaration.clone(), &self.stability, &self.base_frontier())
                .expect("a retained sealed declaration replays as a duplicate");
            return;
        }
        self.epochs
            .deliver(declaration.clone(), &self.stability, &self.base_frontier())
            .expect("an honest concurrent declaration enters as a candidate");
        let dot = declaration.dot();
        let _ = self.have.insert(dot);
        if dot.station() == self.id {
            // An own declaration replayed at rehydration advances the
            // allocator, like any own mint (see `on_old`).
            self.counter = self.counter.max(dot.counter());
        }
        self.declaration_dots.push(dot);
    }

    fn on_confirm(
        &mut self,
        generation: u64,
        epoch: EpochAddress,
        station: u32,
        cut: &Cut,
        note: &Note,
    ) {
        if generation > self.generation {
            self.park(note);
            return;
        }
        match self
            .epochs
            .confirm(epoch, &Vouched::trust(station, cut.clone()))
        {
            Ok(()) => {}
            // The report outran its declaration; redelivery cures it.
            Err(EpochRefusal::AddressMiss { .. }) if generation == self.generation => {
                self.park(note);
            }
            Err(refusal) => {
                assert!(
                    generation < self.generation,
                    "replica {}: unexpected confirmation refusal {refusal:?}",
                    self.id
                );
            }
        }
    }

    fn on_adoption(
        &mut self,
        generation: u64,
        epoch: EpochAddress,
        station: u32,
        counter: u64,
        note: &Note,
    ) {
        if generation > self.generation {
            self.park(note);
            return;
        }
        match self
            .epochs
            .adopt_report(epoch, &Vouched::trust(station, counter))
        {
            Ok(()) => {}
            Err(EpochRefusal::AddressMiss { .. }) if generation == self.generation => {
                self.park(note);
            }
            Err(refusal) => {
                assert!(
                    generation < self.generation,
                    "replica {}: unexpected adoption-report refusal {refusal:?}",
                    self.id
                );
            }
        }
    }
}