minerva 0.2.0

//! Monotone epoch drives, transition opening, and seal checkpointing.

extern crate alloc;

use alloc::vec::Vec;

use crate::metis::{
    Cut, Departed, Dot, DotSet, DotStore, EpochAddress, EpochIdeal, EpochRefusal, EpochStratum,
    Retirement, Stability, Vouched,
};

use super::super::{Note, OldDelta};
use super::{Journal, Moves, Replica, Text, Transition, fold_native};

impl Replica {
    // ---- the monotone drives -----------------------------------------------

    /// Runs every monotone step that new knowledge can have enabled:
    /// report a risen floor, confirm reachable candidates, adopt a fixed
    /// winner, drain the native gate, seal a complete window. Loops because
    /// a seal opens a new generation whose floor wants reporting.
    pub(super) fn poll(&mut self, out: &mut Vec<Note>) {
        loop {
            // A sealed departure the tracker refused is retried here rather
            // than in a lane of its own: what unblocks it is a survivor's
            // report reaching the stranded coordinate, and reports arrive
            // through the same drive.
            self.settle_departures();
            self.report(out);
            self.acknowledge_removals(out);
            self.confirm_candidates(out);
            self.try_adopt(out);
            self.drain_gate();
            if !self.try_seal() {
                break;
            }
        }
    }

    /// The removals this replica has applied, the honest acknowledgement a
    /// retirement round gathers: every text dot it has seen and no longer
    /// shows (context minus visible support), MINUS the anchor pins. A
    /// surviving movement testimony's placement anchor is a retention pin:
    /// the recension re-places the moved element through that locus, so
    /// excising it would silently re-decide the move (the moved element
    /// falls back to its birth place), a GC that moves text. Withholding
    /// the acknowledgement pins the dot at the roster meet until the
    /// testimony itself is retired, exactly the named consumer's
    /// cursor-anchor discipline composed with the move machine.
    fn applied_removals(&self) -> DotSet {
        let visible = self.text.store().support();
        let mut pinned = DotSet::new();
        for (_, testimony) in self.moves.store() {
            // The anchor is a payload seat (R-91): a raw coordinate naming
            // the non-dot zero pins nothing, as the old refused insert did.
            if let Some(anchor) = testimony.to.anchor.dot()
                && let Ok(anchor) = Dot::try_from(anchor)
            {
                let _ = pinned.insert(anchor);
            }
        }
        let mut applied = DotSet::new();
        for dot in self.text.context().difference(&visible) {
            if !pinned.contains(dot) {
                let _ = applied.insert(dot);
            }
        }
        applied
    }

    /// Publishes this replica's applied-removal acknowledgement when it
    /// changed: the retirement round's own-station half, folded locally
    /// and broadcast with the delivered floor the gathering gate checks.
    fn acknowledge_removals(&mut self, out: &mut Vec<Note>) {
        let applied = self.applied_removals();
        if self.acked == applied {
            return;
        }
        self.retirement
            .acknowledge(self.id, &applied)
            .expect("self is a roster member");
        let note = Note::Retire {
            generation: self.generation,
            station: self.id,
            applied: applied.clone(),
            at: Cut::floor_of(&self.have),
        };
        // The oath duty: the acknowledgement's basis (the applied
        // removals and everything below the carried floor) is fenced
        // before the oath leaves. A restarted replica that had forgotten
        // its own published oath would re-anchor through a dot its peers
        // lawfully excised (the boundary exhibit).
        self.emit(out, note);
        self.acked = applied;
    }

    fn report(&mut self, out: &mut Vec<Note>) {
        let floor = self.have.floor();
        if self.reported.as_ref() == Some(&floor) {
            return;
        }
        let cut = Cut::floor_of(&self.have);
        self.stability
            .report_cut(self.id, &cut)
            .expect("self is a roster member");
        let note = Note::Report {
            generation: self.generation,
            station: self.id,
            cut,
        };
        // The durable-floor duty: the fence runs before the report
        // leaves, so the claimed floor never exceeds the journal (a
        // report is a durability claim beside its cut claim; PRD 0011
        // R8's second face). The join-fold at every peer would absorb a
        // post-crash regression silently, so the only sound place to
        // stop one is before the higher claim is ever emitted.
        self.emit(out, note);
        self.reported = Some(floor);
    }

    fn confirm_candidates(&mut self, out: &mut Vec<Note>) {
        let floor = self.have.floor();
        let candidates: Vec<(EpochAddress, Dot)> = self
            .epochs
            .candidates()
            .map(|declaration| (declaration.address(), declaration.dot()))
            .collect();
        for (address, dot) in candidates {
            if !self.confirmed.contains(&address) && floor.get(dot.station()) >= dot.counter() {
                let cut = Cut::floor_of(&self.have);
                self.epochs
                    .confirm(address, &Vouched::trust(self.id, cut.clone()))
                    .expect("a floor covering the declaration confirms it");
                let _ = self.confirmed.insert(address);
                let note = Note::Confirm {
                    generation: self.generation,
                    epoch: address,
                    station: self.id,
                    cut,
                };
                self.emit(out, note);
            }
        }
    }

    fn try_adopt(&mut self, out: &mut Vec<Note>) {
        if self.transition.is_some() {
            return;
        }
        match self.epochs.adopt(self.id, self.counter, &self.stability) {
            Ok(adopted) => {
                let note = Note::Adoption {
                    generation: self.generation,
                    epoch: adopted.address(),
                    station: self.id,
                    counter: self.counter,
                };
                // The round duty: adoption's basis (the declaration, the
                // confirmation round, the reports the watermark read) is
                // fenced before the report leaves. The seal is licensed
                // by this emission, so a restart must re-derive the same
                // adoption from the journal alone; the `Adopted` witness
                // itself is deliberately never persisted, it is re-earned
                // by re-running `adopt` over the rebuilt record.
                self.emit(out, note);
                self.open_transition(adopted);
            }
            // `Unconfirmed` is the ordinary wait. `UnassentedAdmission`
            // is the R-89 counterweight, surfaced rather than crashed: a
            // lawfully tagged winner waits at this member's own adopt
            // door until its operator endorses the boundary, because
            // assent is a caller act and the machine refuses to invent
            // it. A stall, not an error --- the cure is this member's
            // own `open_arrival`
            // (`scenarios::arrival::an_unassenting_member_stalls_the_window_until_its_operator_admits`).
            Err(EpochRefusal::Unconfirmed | EpochRefusal::UnassentedAdmission { .. }) => {}
            Err(refusal) => panic!(
                "replica {}: adoption refused unexpectedly: {refusal:?}",
                self.id
            ),
        }
    }

    /// Adoption: rebuild the sealed stratum through the fused construction
    /// (one decision replay serving the fold, the maintained surface, and
    /// the projection), move both products out through the transition door
    /// (the native plane and the shadow, no clone, S272), and replay the
    /// window traffic already delivered through the batch doors (one
    /// maintenance refresh per plane, the S270 latency rule).
    fn open_transition(&mut self, adopted: crate::metis::Adopted) {
        let cut = adopted.cut().clone();
        let (text_at_cut, moves_at_cut) = self.rebuild_at(&cut);
        let stratum = EpochStratum::new(&adopted, text_at_cut, moves_at_cut)
            .expect("the rebuild carries exactly the sealed stratum");
        let transition = stratum.into_transition();
        let pristine = Text::from_store(transition.native);
        let mut shadow = transition.shadow;
        let vector = cut.as_vector();
        let above_cut = |dots: &[Dot]| {
            dots.iter()
                .any(|dot| dot.counter() > vector.get(dot.station()))
        };
        let window_text = self.log.iter().filter_map(|(dots, delta)| match delta {
            OldDelta::Text(delta) if above_cut(dots) => Some(delta.as_ref()),
            _ => None,
        });
        shadow
            .deliver_text_batch(window_text)
            .expect("already-delivered window text replays through the shadow");
        let window_moves = self.log.iter().filter_map(|(dots, delta)| match delta {
            OldDelta::Moves(delta) if above_cut(dots) => Some(delta),
            _ => None,
        });
        shadow
            .deliver_moves_batch(window_moves)
            .expect("already-delivered window movement replays through the shadow");
        // Refoundation coalesces this station's survivors to `1..=live`;
        // above-cut local mints translate affinely after them.
        let live = pristine.store().woven().high_water_of(self.id);
        let window_mints = self
            .counter
            .checked_sub(vector.get(self.id))
            .expect("the local allocator covers the adopted cut");
        let native_counter = live
            .checked_add(window_mints)
            .expect("the refounded allocator stays below the dot ceiling");
        self.gate
            .adopt_epoch(&adopted)
            .expect("one adoption per transition buffer");
        self.transition = Some(Transition {
            adopted,
            shadow,
            native_text: pristine,
            native_moves: Moves::new(),
            native_counter,
            native_recorded: DotSet::new(),
            native_log: Vec::new(),
        });
    }

    fn drain_gate(&mut self) {
        if let Some(transition) = self.transition.as_mut() {
            while let Some(released) = self.gate.pop_ready_event() {
                let stamp = released.stamp();
                let (dots, delta) = released.into_payload();
                fold_native(transition, &dots, &delta, Some(stamp));
            }
        }
    }

    /// The seal: take the shadow's consignment as the next generation's base
    /// (the stage-four document-plane carry: window births woven at their
    /// judged places, moved images where the old replay put them, deleted
    /// images as order tombstones, floors born gap-free), annex the
    /// protocol-plane declaration dots, and open the next generation.
    /// Rebuilds the tracker for the new generation and re-applies the
    /// caller's departures to it.
    ///
    /// Departure is configuration, not machine state: re-declare it or the
    /// new generation freezes on the member the last one evicted
    /// (`membership::a_rebuilt_tracker_starts_with_a_whole_family_again`).
    ///
    /// An attested departure re-declares at the departed station's
    /// compacted prefix in `opening`, the plane the seal founds: the bound
    /// its round agreed is a counter in the plane this seal just
    /// re-minted, the member's surviving elements were coalesced to
    /// `1..=held` in the new base every member agrees on, and it minted
    /// nothing since --- so the re-declared coordinate is testimony the
    /// base itself gives (`Departed::refounded`). Bottom only when nothing
    /// of the member's survived; a bottom carry under surviving elements
    /// understates the coordinate and wedges the *next* seal at its
    /// consignment door, which is how the original bottom-only carry was
    /// caught (`scenarios::arrival::an_evictees_survivors_cross_the_second_boundary`).
    /// The fence crosses with it, or it would admit in the new plane
    /// exactly the dots the new attestation writes off. A round still
    /// *open* at the boundary is dropped: its proposals were counters in
    /// the old plane, and re-opening it costs nothing.
    fn carry_departures(&mut self, opening: &DotSet) {
        self.stability = Stability::new(self.roster.iter().copied());
        for &station in &self.departed {
            let _ = self.stability.abandon(station).expect("a family remains");
        }
        // Attestations re-key over the family the new plane opens with:
        // an admission-bearing seal widened `self.roster` before this call,
        // and an attestation still keyed to the narrow family would refuse
        // the rebuilt tracker's family check (`Departed::binds`) at every
        // seal after the first admission. Coverage and possession part in
        // the re-key: an old-family member carries the base prefix it
        // sealed over, an admitted joiner enters at zero (its admission
        // proves nothing about what it holds) and earns holder status
        // through its own report (PRD 0028 falsifier 8: the S340 cure
        // relocated, not retired; the split is the third review round's
        // P1).
        self.departures = self
            .departures
            .values()
            .filter_map(|round| {
                Some((
                    round.station(),
                    round.refounded(self.roster.iter().copied(), opening)?,
                ))
            })
            .collect();
        self.attested = self
            .attested
            .iter()
            .map(|departed| departed.refounded(self.roster.iter().copied(), opening))
            .collect();
        for departed in &self.attested {
            let _ = self
                .stability
                .abandon_attested(departed)
                .expect("a family remains");
        }
    }

    fn try_seal(&mut self) -> bool {
        // A wedged replica performs no further seals. Without this the next
        // call would take an already-taken transition and trip the
        // local-adoption expectation, turning an observed wedge into a
        // harness crash one step later.
        if self.halted.is_some() {
            return false;
        }
        let Some(sealed) = self.epochs.try_seal(&self.stability) else {
            return false;
        };
        let sealed = sealed.clone();
        let address = sealed.declaration();
        let mut transition = self
            .transition
            .take()
            .expect("a local seal implies local adoption");
        assert_eq!(
            self.gate.pending_len(),
            0,
            "replica {}: no native event may be left behind at the seal",
            self.id
        );
        let projection = transition.shadow.projection().clone();
        // The consignment door derives the generation's protocol-plane dots
        // (the declarations) from the seal witness itself and checks the
        // joint coverage of the sealed join before folding (the
        // review-caught completeness hole: a genuine seal over a shadow
        // missing one delivered delta must refuse, never silently drop,
        // and nothing caller-claimed may stand in for the delivery).
        let (consignment, consigned) = match transition.shadow.consign(&sealed) {
            Ok(accepted) => accepted,
            // Under an honest fabric this is unreachable and must stay a
            // panic: every other arm's laws rest on it. The `byzantine`
            // exposure module is the one caller that arms `halt_on_refusal`,
            // because a forked voice makes the refusal the *expected*
            // outcome and the harness must be able to observe it rather than
            // die on it (`Replica::halted`).
            // The wedge is *asymmetric*, and that is part of the finding
            // rather than a harness artifact: `Epochs::try_seal` already
            // advanced the lineage before this door refused, and a seal does
            // not roll back. So a halted replica's epoch ledger stands one
            // generation ahead of its document plane, permanently. The
            // transition is put back so the state is defined rather than
            // half-taken, and `halted` wedges the seal path above.
            Err((shadow, refusal)) if self.halt_on_refusal => {
                self.halted = Some(refusal);
                // The door hands the shadow back rather than consuming it on
                // refusal, so the window survives even though the ledger has
                // moved on. That is the asymmetry: recoverable at the
                // shadow, not at the seal.
                transition.shadow = *shadow;
                self.transition = Some(transition);
                return false;
            }
            Err((_, refusal)) => panic!(
                "replica {}: an honestly sealed window consigns, got {refusal:?}",
                self.id
            ),
        };
        // The certification duty, discharged: the record becomes
        // certifiable exactly here, and never at `try_seal`.
        self.certified.push(consigned);
        let (base_text, base_moves, _map) = consignment.into_parts();

        // The stage-four agreement pin: the consigned base reads exactly
        // the frozen projection, at every replica, before any native delta
        // folds over it.
        assert_eq!(
            &base_text.store().order()[..],
            projection.order(),
            "replica {}: the consignment must open on the sealed projection",
            self.id
        );

        // The next generation: base plus this window's native log.
        let mut text = base_text.clone();
        let mut moves = base_moves.clone();
        for (_, delta, _) in &transition.native_log {
            match delta {
                OldDelta::Text(delta) => text.merge_from(delta),
                OldDelta::Moves(delta) => moves.merge_from(delta),
            }
        }
        self.generation += 1;
        // The rebuild reads the roster from the record just sealed
        // (PRD 0028 R1): an admission-bearing seal widened the machine's
        // roster, and the harness copy follows it here, before the
        // stability and retirement trackers rebuild, so both families
        // widen in the same act --- the R-87 question answered: widening
        // only ever lowers a meet (a silent joiner pins it at bottom),
        // where a tracker left narrow would license claims the joiner
        // never made.
        self.roster = self.epochs.roster().collect();
        let opening = base_text.context().merge(base_moves.context());
        self.carry_departures(&opening);
        self.counter = transition.native_counter;
        self.have = text.context().merge(moves.context());
        self.text = text;
        self.moves = moves;
        self.base_text = base_text;
        self.base_moves = base_moves;
        self.log = transition
            .native_log
            .drain(..)
            .map(|(dots, delta, _)| (dots, delta))
            .collect();
        self.recorded = self.base_text.context().merge(self.base_moves.context());
        for (dots, _) in &self.log {
            for &dot in dots {
                let _ = self.recorded.insert(dot);
            }
        }
        self.reported = None;
        // Retirement evidence is generation-scoped: the refound swept the
        // old plane's retired identity wholesale, the next plane is a new
        // dot space (whose small counters recur), and a carried meet would
        // license excising a recurring dot before the new generation's own
        // round has vouched for it.
        self.retirement = Retirement::new(self.roster.iter().copied());
        self.acked = DotSet::new();
        self.confirmed.clear();
        self.declaration_dots.clear();
        self.gate = EpochIdeal::default();
        self.gated = DotSet::new();
        self.seals.push((address, projection));
        // The seal checkpoint: the one durable-store rewrite the recipe
        // asks for, at the one moment everything it holds is computed
        // anyway. The consigned base pair and the sealed lineage are
        // compaction output that the swept history can no longer
        // re-derive, so they must be durable before this replica
        // participates in the new generation (the H1 caveat priced); the
        // journal restarts at the new base plus this window's native log
        // as the opening notes. Notes still parked (a faster peer's
        // next-generation traffic) return to the volatile buffer: they
        // re-fence with the first new-generation emission, and a crash
        // before that loses them to the repair lane, like any receipt.
        self.journal = Journal {
            generation: self.generation,
            base_text: self.base_text.clone(),
            base_moves: self.base_moves.clone(),
            epochs: self.epochs.clone(),
            counter: self.counter,
            // Minted at the seal, so it exceeds every stamp this replica
            // spoke in the sealed window (the clock minted them all and
            // is monotone); the re-foundation re-bases ranks, so those
            // stamps survive nowhere else a restart could walk.
            clock: self.clock.now(0u16),
            seals: self.seals.clone(),
            departed: self.departed.clone(),
            attested: self.attested.iter().map(Departed::station).collect(),
            notes: self
                .log
                .iter()
                .map(|(dots, delta)| Note::Old {
                    generation: self.generation,
                    dots: dots.clone(),
                    delta: delta.clone(),
                })
                .collect(),
        };
        self.volatile = self.parked.clone();
        true
    }
}