minerva 0.2.0

//! Construction, durable checkpoints, rehydration, and repair reads.

extern crate alloc;

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

use crate::kairos::{Clock, TickCounter};
use crate::metis::{Cut, DotSet, EpochIdeal, Epochs, LineageProofRecord, Retirement, Stability};

use super::super::Note;
use super::{
    JoinerBootstrapError, JoinerCheckpoint, Journal, Moves, Replica, Text, observe_pair_ranks,
};

impl Replica {
    pub fn new(id: u32, roster: &[u32], horizon: NonZeroUsize) -> Self {
        let clock = Clock::with_default_config(TickCounter::new(), id).unwrap();
        let clock_mark = clock.now(0u16);
        Self {
            id,
            roster: roster.to_vec(),
            clock,
            epochs: Epochs::new(roster.iter().copied(), horizon),
            generation: 1,
            stability: Stability::new(roster.iter().copied()),
            counter: 0,
            text: Text::new(),
            moves: Moves::new(),
            base_text: Text::new(),
            base_moves: Moves::new(),
            log: Vec::new(),
            have: DotSet::new(),
            recorded: DotSet::new(),
            reported: None,
            retirement: Retirement::new(roster.iter().copied()),
            acked: DotSet::new(),
            confirmed: BTreeSet::new(),
            declaration_dots: Vec::new(),
            gate: EpochIdeal::default(),
            gated: DotSet::new(),
            transition: None,
            parked: Vec::new(),
            seals: Vec::new(),
            journal: Journal {
                generation: 1,
                base_text: Text::new(),
                base_moves: Moves::new(),
                epochs: Epochs::new(roster.iter().copied(), horizon),
                counter: 0,
                clock: clock_mark,
                seals: Vec::new(),
                departed: BTreeSet::new(),
                attested: Vec::new(),
                notes: Vec::new(),
            },
            volatile: Vec::new(),
            halt_on_refusal: false,
            halted: None,
            certified: Vec::new(),
            departed: BTreeSet::new(),
            departures: BTreeMap::new(),
            attested: Vec::new(),
            resurgences: Vec::new(),
        }
    }

    /// Exports the local data checkpoint when the replica stands exactly
    /// at a seal, before any current-generation data or window state can
    /// make the checkpoint incomplete.
    pub(in crate::metis::tests::fleet) fn joiner_checkpoint(&self) -> Option<JoinerCheckpoint> {
        self.joiner_checkpoint_for(self.id)
    }

    /// [`Replica::joiner_checkpoint`] re-stamped for `station`: the same
    /// sealed data checkpoint, exported for a member other than this one
    /// --- an admitted joiner above all, a station that never held a
    /// checkpoint of its own (PRD 0028's fleet carriage).
    ///
    /// The re-stamp changes exactly two fields. The station names the
    /// intended holder, which [`Replica::bootstrap`] checks against the
    /// lineage-derived roster. The allocator basis becomes *that
    /// station's* high water in the checkpointed plane --- zero for a
    /// fresh joiner, whose pre-admission coordinate is bottom by the
    /// lineage rule --- because handing the joiner the exporter's counter
    /// would open its dot sequence at a gap the floor could never close.
    /// Everything else (the base pair, the clock watermark, the seal
    /// agreement record) is generation state, identical at every honest
    /// member by construction, so the exporter's copy is the joiner's.
    ///
    /// Two membership refusals guard the door beside the at-seal guard.
    /// The target must not be attested or abandoned here: an epoch roster
    /// never removes a seat, so the membership read alone would happily
    /// re-stamp for an evicted identity. And the live attestation state
    /// must still be the journal's seal-time state: an attestation lands
    /// mid-generation without disturbing any at-seal field, so a
    /// checkpoint exported past one would omit the departure and a
    /// bootstrap from it would restore the evicted member as active while
    /// every peer fences it. The refusal is conservative and cheap: the
    /// next seal re-journals the attestations and export resumes.
    pub(in crate::metis::tests::fleet) fn joiner_checkpoint_for(
        &self,
        station: u32,
    ) -> Option<JoinerCheckpoint> {
        let member = self.epochs.roster().any(|seat| seat == station)
            && self.attested(station).is_none()
            && !self.departed.contains(&station);
        let attestations_at_seal = self
            .attested
            .iter()
            .map(crate::metis::Departed::station)
            .eq(self.journal.attested.iter().copied());
        let at_seal = !self.journal.seals.is_empty()
            && self.transition.is_none()
            && self.generation == self.journal.generation
            && self.epochs == self.journal.epochs
            && self.text == self.journal.base_text
            && self.moves == self.journal.base_moves
            && self.counter == self.journal.counter
            && self.seals == self.journal.seals;
        (member && attestations_at_seal && at_seal).then(|| JoinerCheckpoint {
            station,
            generation: self.journal.generation,
            base_text: self.journal.base_text.clone(),
            base_moves: self.journal.base_moves.clone(),
            counter: if station == self.id {
                self.journal.counter
            } else {
                self.journal
                    .base_text
                    .context()
                    .merge(self.journal.base_moves.context())
                    .high_water_of(station)
            },
            clock: self.journal.clock,
            seals: self.journal.seals.clone(),
            attested: self.journal.attested.clone(),
        })
    }

    /// Builds a fresh replica from the two bootstrap authorities: sealed
    /// local data and a verified peer lineage.
    ///
    /// `roster` grounds the *oldest* retained generation, per the machine
    /// door's own contract: the proof's admissions widen it forward, and
    /// the replica's live roster is *derived* from the walked lineage
    /// rather than configured (PRD 0028 R4's inversion at the harness
    /// door). The checkpoint's station is therefore checked against the
    /// derived roster --- which is what admits a joiner whose station no
    /// founding roster ever named, through the very proof that admits it.
    pub(in crate::metis::tests::fleet) fn bootstrap(
        checkpoint: JoinerCheckpoint,
        roster: &[u32],
        horizon: NonZeroUsize,
        lineage: &LineageProofRecord,
    ) -> Result<Self, JoinerBootstrapError> {
        let epochs = Epochs::bootstrap(roster.iter().copied(), horizon, lineage)?;
        let derived: Vec<u32> = epochs.roster().collect();
        if !derived.contains(&checkpoint.station) {
            return Err(JoinerBootstrapError::StationOutsideRoster {
                station: checkpoint.station,
            });
        }
        let lineage_generation = epochs.snapshot().generation();
        if checkpoint.generation != lineage_generation {
            return Err(JoinerBootstrapError::GenerationMismatch {
                checkpoint: checkpoint.generation,
                lineage: lineage_generation,
            });
        }

        let mut replica = Self::new(checkpoint.station, &derived, horizon);
        replica.clock.observe(checkpoint.clock);
        replica.epochs = epochs;
        replica.generation = checkpoint.generation;
        replica.counter = checkpoint.counter;
        replica.text = checkpoint.base_text.clone();
        replica.moves = checkpoint.base_moves.clone();
        replica.base_text = checkpoint.base_text.clone();
        replica.base_moves = checkpoint.base_moves.clone();
        replica.have = checkpoint
            .base_text
            .context()
            .merge(checkpoint.base_moves.context());
        replica.recorded = replica.have.clone();
        replica.seals = checkpoint.seals.clone();
        // The certifiable grade a joiner arrives holding. It consigned
        // none of these records itself: they came as *verified peer
        // claims*, checked against an authenticated checkpoint certificate
        // one layer up, which is a stronger warrant than local acceptance
        // and one Minerva cannot re-derive. That is the first of exactly
        // two honest callers of `Consigned::vouch`.
        replica.vouch_restored_lineage();
        replica.journal = Journal {
            generation: checkpoint.generation,
            base_text: checkpoint.base_text,
            base_moves: checkpoint.base_moves,
            epochs: replica.epochs.clone(),
            counter: checkpoint.counter,
            clock: checkpoint.clock,
            seals: checkpoint.seals,
            departed: BTreeSet::new(),
            attested: checkpoint.attested.clone(),
            notes: Vec::new(),
        };
        // The attested departures ride the checkpoint: agreed protocol
        // objects the seat's world carries, without which the joiner
        // holds a meet the evicted member pins forever and window rounds
        // no substitution covers. The bounds re-derive from the restored
        // base's own compacted prefixes, like every restore. Bare
        // abandonments deliberately do not ride: each operator reads its
        // own (R-84).
        replica.restore_departures(&BTreeSet::new(), &checkpoint.attested);
        observe_pair_ranks(&replica.clock, &replica.text, &replica.moves);
        Ok(replica)
    }

    /// The modeled durable store as it stands right now: what a crash at
    /// this instant would leave on disk.
    pub const fn journal(&self) -> &Journal {
        &self.journal
    }

    /// The current generation's log length: one entry per unique data
    /// note under the bounded-record rule, however much redelivery and
    /// repair traffic flowed.
    pub fn log_len(&self) -> usize {
        self.log.len()
    }

    /// How many data notes (old or native deltas) the journal holds: the
    /// bounded-record read, exactly one per unique event.
    pub fn journal_data_notes(&self) -> usize {
        self.journal
            .notes
            .iter()
            .filter(|note| matches!(note, Note::Old { .. } | Note::Native { .. }))
            .count()
    }

    /// The journal's total fenced-note count: under effect-based fencing
    /// this grows only with genuine state change, so a no-effect repair
    /// round leaves it exactly where it was.
    pub fn journal_notes_len(&self) -> usize {
        self.journal.notes.len()
    }

    /// The parked set's size: bounded by *distinct* out-of-reach notes
    /// under the park-once rule, however often a repair round re-states
    /// them.
    pub fn parked_len(&self) -> usize {
        self.parked.len()
    }

    /// The volatile buffer's size: what a crash right now would lose.
    /// Effect-based fencing keeps it bounded by genuine new information.
    pub fn volatile_len(&self) -> usize {
        self.volatile.len()
    }

    /// Received-but-unfenced data notes: the sharp settled-state read.
    /// A drained fleet fences every data note (the last novel delta
    /// raises the floor, the floor change emits a report, and the
    /// emission fences), so only protocol tails may sit volatile, the
    /// bounded loss the repair lane re-states.
    pub fn volatile_data_notes(&self) -> usize {
        self.volatile
            .iter()
            .filter(|note| matches!(note, Note::Old { .. } | Note::Native { .. }))
            .count()
    }

    /// Rebuilds a replica from its journal alone: the rehydration
    /// recipe's mechanical half, and deliberately nothing but a replay.
    ///
    /// The checkpoint seeds the generation, the base pair, the epoch
    /// lineage, and the allocator basis; every fenced note then replays
    /// through the ordinary inbound door with the outbox discarded, so
    /// recovery is re-derivation by the very code paths delivery uses,
    /// never a second semantics. The pairs, the have-set, the log, the
    /// stability and retirement trackers, the parked set, the window
    /// state, and the [`Adopted`](crate::metis::Adopted) witness itself
    /// (re-earned by `adopt` re-running over the rebuilt record, exactly
    /// because it is unforgeable and was never persisted) all fall out of
    /// the replay. Afterwards the allocators and the clock re-derive
    /// their high waters from the rebuilt planes; under the mint duty the
    /// journal covers every dot this replica ever emitted, so the
    /// re-derived allocator can never re-mint one.
    ///
    /// What a crash loses is the volatile buffer: receipts since the last
    /// fence. Losing them silences this replica (floors sit lower, rounds
    /// stall) but never un-says it, and the repair lane (`resync`)
    /// restores liveness by re-serving logs and re-stating the idempotent
    /// round positions.
    pub fn rehydrate(id: u32, roster: &[u32], horizon: NonZeroUsize, journal: &Journal) -> Self {
        Self::rehydrate_inner(id, roster, horizon, journal, false)
    }

    /// [`Replica::rehydrate`] with the observable halt armed *before* the
    /// journal replays.
    ///
    /// The `byzantine` exposure module is the only caller. Arming after
    /// `rehydrate` returns is too late to observe a durable wedge: the
    /// replay itself re-derives the seal that refuses, so the strict panic
    /// fires inside the constructor and the harness never gets to look.
    pub fn rehydrate_tolerating_refusals(
        id: u32,
        roster: &[u32],
        horizon: NonZeroUsize,
        journal: &Journal,
    ) -> Self {
        Self::rehydrate_inner(id, roster, horizon, journal, true)
    }

    fn rehydrate_inner(
        id: u32,
        roster: &[u32],
        horizon: NonZeroUsize,
        journal: &Journal,
        halt_on_refusal: bool,
    ) -> Self {
        let mut replica = Self::new(id, roster, horizon);
        replica.halt_on_refusal = halt_on_refusal;
        replica.generation = journal.generation;
        replica.epochs = journal.epochs.clone();
        // The roster comes from the restored record (PRD 0028 R4): an
        // admission-bearing seal widened it past the caller's birth
        // roster, and the checkpoint's own lineage names the family the
        // trackers must range over. Same-station restore is the caller
        // trusting its own durable state (the `Epochs::rehydrate`
        // posture), so the derivation is a read, not a check.
        replica.roster = replica.epochs.roster().collect();
        replica.stability = Stability::new(replica.roster.iter().copied());
        replica.retirement = Retirement::new(replica.roster.iter().copied());
        replica.base_text = journal.base_text.clone();
        replica.base_moves = journal.base_moves.clone();
        replica.text = journal.base_text.clone();
        replica.moves = journal.base_moves.clone();
        replica.have = journal
            .base_text
            .context()
            .merge(journal.base_moves.context());
        replica.recorded = replica.have.clone();
        replica.counter = journal.counter;
        replica.seals = journal.seals.clone();
        // The second honest caller: the checkpoint this lineage came out
        // of was written *at a seal*, which the seal path reaches only
        // past the consignment door, so a restored lineage is a lineage
        // whose consignments were accepted before the crash. The caller
        // vouches for its own storage, exactly the posture
        // `Epochs::rehydrate` already takes. Records sealed *after* the
        // checkpoint are re-derived by the replay below and must earn the
        // grade again at the door: that is why a wedge survives a restart.
        replica.vouch_restored_lineage();
        replica.journal = Journal {
            generation: journal.generation,
            base_text: journal.base_text.clone(),
            base_moves: journal.base_moves.clone(),
            epochs: journal.epochs.clone(),
            counter: journal.counter,
            clock: journal.clock,
            seals: journal.seals.clone(),
            departed: journal.departed.clone(),
            attested: journal.attested.clone(),
            notes: Vec::new(),
        };
        // The departures the checkpoint fenced. The bare half is the caller
        // configuration R-84 records as a re-declaration duty; the attested
        // half is a promise peers have sealed over, so it is restored here
        // rather than waited for --- at the restored base's compacted
        // prefix, which is where the seal that wrote this checkpoint left
        // every attestation.
        replica.restore_departures(&journal.departed, &journal.attested);
        // The clock re-derives its high water before anything can mint:
        // the checkpoint watermark dominates every stamp spoken before
        // the checkpoint (including the sealed window's, whose ranks the
        // re-foundation re-based away), and the journaled notes carry
        // every post-checkpoint stamp that lives in no plane (a native
        // delete's stamp, a declaration's rank).
        replica.clock.observe(journal.clock);
        for note in &journal.notes {
            match note {
                Note::Native { stamp, .. } => replica.clock.observe(*stamp),
                Note::Declare { declaration } => replica.clock.observe(declaration.rank()),
                _ => {}
            }
        }
        let mut discard = Vec::new();
        for note in &journal.notes {
            replica.handle(note, &mut discard);
            discard.clear();
        }
        // The replay's own tail fence: the journal ends where it began.
        replica.fence();
        // The allocators re-derive past everything the journal shows
        // emitted (own dots are delivered locally at mint, so the rebuilt
        // planes carry them all).
        replica.counter = replica.counter.max(replica.have.high_water_of(id));
        if let Some(transition) = replica.transition.as_mut() {
            let own_native = transition
                .native_log
                .iter()
                .flat_map(|(dots, _, _)| dots.iter().copied())
                .filter(|dot| dot.station() == id)
                .map(crate::metis::Dot::counter)
                .max();
            if let Some(highest) = own_native {
                transition.native_counter = transition.native_counter.max(highest);
            }
        }
        // The rebuilt planes' ranks complete the clock re-derivation
        // (beside the checkpoint watermark and note stamps above), so no
        // post-restart stamp undercuts one this replica already spoke.
        observe_pair_ranks(&replica.clock, &replica.text, &replica.moves);
        if let Some(transition) = &replica.transition {
            observe_pair_ranks(
                &replica.clock,
                &transition.native_text,
                &transition.native_moves,
            );
        }
        for declaration in replica.epochs.candidates() {
            replica.clock.observe(declaration.rank());
        }
        replica
    }

    /// Re-states this replica's current round positions for a peer that
    /// restarted: the repair lane's protocol half. Everything here is
    /// idempotent by the protocol's own folds (reports and confirmations
    /// join, acknowledgements gather under the cut gate, declarations and
    /// adoption reports absorb as duplicates), so re-stating is always
    /// lawful, whether or not the receiver lost anything.
    ///
    /// A sealed window's adoption round is reconstructed from the
    /// lineage: the sealed join's per-member entry *is* that member's
    /// adoption report (the join of own-station cuts is one vector with
    /// each member's own counter), so a peer that sealed and cleared its
    /// window can still re-state the round a restarted laggard needs to
    /// finish. The restarted member necessarily adopted before the fleet
    /// could seal, so under the round duty its journal already holds the
    /// declaration and the confirmation round; what no live traffic will
    /// repeat are the adoption reports *and the generation's floor
    /// standing*, and both are reconstructed here from the sealed record.
    /// The floor half is the S289 schedule tape's wedge: a member whose
    /// crash regresses its tracker after the fleet seals can re-earn
    /// adoption from its journal yet never pass its own seal gate (the
    /// roster-meet watermark), because its peers' live reports speak a
    /// later generation and lawfully park. The sealed join doubles as the
    /// cure: the seal fired only after the roster meet covered it, so
    /// "member `m`'s floor covered the join" is a *proven under-claim*
    /// for every member (the S286 relay under-claim license, the
    /// transport of claims), and re-stating it as each member's own claim
    /// raises the laggard's stale rows exactly to the seal gate.
    pub fn restate(&self, out: &mut Vec<Note>) {
        out.push(Note::Report {
            generation: self.generation,
            station: self.id,
            cut: Cut::floor_of(&self.have),
        });
        out.push(Note::Retire {
            generation: self.generation,
            station: self.id,
            applied: self.acked.clone(),
            at: Cut::floor_of(&self.have),
        });
        for declaration in self.epochs.candidates() {
            out.push(Note::Declare {
                declaration: declaration.clone(),
            });
        }
        let floor = self.have.floor();
        for &address in &self.confirmed {
            let dot = address.declaration();
            if floor.get(dot.station()) >= dot.counter() {
                out.push(Note::Confirm {
                    generation: self.generation,
                    epoch: address,
                    station: self.id,
                    cut: Cut::floor_of(&self.have),
                });
            }
        }
        if let Some(transition) = &self.transition {
            out.push(Note::Adoption {
                generation: self.generation,
                epoch: transition.adopted.address(),
                station: self.id,
                counter: self.counter,
            });
        }
        // An open arrival round's own endorsement, re-stated: a peer that
        // crashed between receiving this word and fencing it holds an
        // incomplete round no live traffic will ever complete (an
        // endorsement is spoken once), so the repair lane re-serves it.
        // Idempotent like every restatement here: the fold latches, a
        // repeat absorbs, and a receiver past the base drops it as void.
        if let Some(round) = self.epochs.arrival() {
            out.push(Note::Endorse {
                by: self.id,
                admission: round.admission().clone(),
                commitment: *round.commitment(),
            });
        }
        for sealed in self.epochs.sealed() {
            let address = sealed.declaration();
            // Walk the roster, not the join's entries: the sealed join is
            // a sparse vector, so a member that adopted at counter zero
            // (no mints of its own that generation) has no entry, yet its
            // adoption report is exactly as load-bearing for a laggard's
            // round completeness as anyone's (the review-caught gap,
            // pinned by the zero-counter restatement exhibit).
            for &station in &self.roster {
                out.push(Note::Adoption {
                    generation: address.generation(),
                    epoch: address,
                    station,
                    counter: sealed.sealed_join().get(station),
                });
            }
            // The generation's floor standing, re-stated as every
            // member's proven under-claim (the doc paragraph above; the
            // S286 license): same-generation rows fold at the laggard,
            // and every advanced receiver drops the stale generation.
            for &station in &self.roster {
                out.push(Note::Report {
                    generation: address.generation(),
                    station,
                    cut: Cut::from_witnessed(sealed.sealed_join().clone()),
                });
            }
        }
    }

    /// Re-serves every logged delta of the planes this replica holds: the
    /// repair lane's data half, the anti-entropy duty the recipe assigns
    /// to the consumer's transport. The lane is a pure read: old-plane
    /// deltas re-carry their generation, native deltas re-carry the
    /// adopted address under their *original* stamps, and nothing is
    /// minted, because a freshly minted repair stamp would be
    /// spoken-but-unfenced, the exact class the recipe forbids (a crash
    /// right after would re-derive the clock below it and the next mint
    /// could re-speak it). An entry learned as a sealed peer's
    /// next-generation old traffic never carried a stamp and re-serves
    /// in that same stampless shape.
    pub fn reserve_logs(&self, out: &mut Vec<Note>) {
        for (dots, delta) in &self.log {
            out.push(Note::Old {
                generation: self.generation,
                dots: dots.clone(),
                delta: delta.clone(),
            });
        }
        if let Some(transition) = &self.transition {
            for (dots, delta, stamp) in &transition.native_log {
                match stamp {
                    Some(stamp) => out.push(Note::Native {
                        epoch: transition.adopted.address(),
                        stamp: *stamp,
                        dots: dots.clone(),
                        delta: delta.clone(),
                    }),
                    None => out.push(Note::Old {
                        generation: self.generation + 1,
                        dots: dots.clone(),
                        delta: delta.clone(),
                    }),
                }
            }
        }
    }
}