minerva 0.2.0

//! Fixed-roster bootstrap admission and its seal-edge repair exhibit.

extern crate alloc;

use alloc::collections::BTreeMap;
use alloc::vec::Vec;
use core::num::NonZeroUsize;

use crate::metis::tests::support::dot as d;
use crate::metis::{
    Cut, Declaration, Dot, EpochAddress, EpochRefusal, LineageProofRecord, Stability,
};

use super::super::fabric::Fabric;
use super::super::replica::{JoinerBootstrapError, Replica};
use super::super::{Note, act, assert_converged, crash, fleet_of, lineage_of, resync};
use super::{ROSTER, horizon};

struct BootstrapFixture {
    fleet: BTreeMap<u32, Replica>,
    fabric: Fabric,
    address: EpochAddress,
    declaration: Declaration,
    replay: Note,
}

fn bootstrappable_fleet(depth: NonZeroUsize) -> BootstrapFixture {
    let mut fleet = fleet_of(&ROSTER, depth);
    let mut fabric = Fabric::new(0xF1EE_70B0, &ROSTER, 0);
    let mut replay = None;
    for station in ROSTER {
        let replica = fleet.get_mut(&station).expect("roster member");
        let mut outbox = Vec::new();
        let _ = replica.insert_visible(0, &mut outbox);
        let _ = replay.get_or_insert_with(|| {
            outbox
                .iter()
                .find(|note| matches!(note, Note::Old { .. }))
                .cloned()
                .expect("an insert emits old-addressed traffic")
        });
        fabric.post(station, outbox);
    }
    fabric.drain(&mut fleet);

    let mut declaration_out = Vec::new();
    let address = fleet
        .get_mut(&1)
        .expect("roster member")
        .try_declare(&mut declaration_out)
        .expect("the settled first plane declares");
    let declaration = declaration_out
        .iter()
        .find_map(|note| match note {
            Note::Declare { declaration } => Some(declaration.clone()),
            _ => None,
        })
        .expect("the declaration leaves through the outbox");
    fabric.post(1, declaration_out);
    fabric.drain(&mut fleet);
    assert_converged(&fleet);

    BootstrapFixture {
        fleet,
        fabric,
        address,
        declaration,
        replay: replay.expect("captured first-generation traffic"),
    }
}

fn install_bootstrap_joiner(fleet: &mut BTreeMap<u32, Replica>, depth: NonZeroUsize) {
    let checkpoint = fleet[&3]
        .joiner_checkpoint()
        .expect("station 3 stands exactly at the seal");
    let lineage = lineage_of(&fleet[&1]);
    let joiner = Replica::bootstrap(checkpoint, &ROSTER, depth, &lineage)
        .expect("the sealed checkpoint and proof describe one generation");
    assert_eq!(joiner.epochs(), fleet[&1].epochs());
    assert_eq!(joiner.effective_order(), fleet[&1].effective_order());
    let _ = fleet.insert(3, joiner);
}

fn assert_bootstrap_verdict_parity(
    fleet: &BTreeMap<u32, Replica>,
    address: EpochAddress,
    declaration: &Declaration,
    replay: &Note,
) {
    let replay_dot = match replay {
        Note::Old { dots, .. } => *dots.first().expect("an event note names its dot"),
        _ => unreachable!("captured an old-addressed note"),
    };
    let sealed = fleet[&1]
        .epochs()
        .sealed()
        .find(|sealed| sealed.declaration() == address)
        .expect("the first declaration is retained");
    let uncovered_dot = d(
        replay_dot.station(),
        sealed
            .sealed_join()
            .get(replay_dot.station())
            .checked_add(1)
            .expect("the test seal stays below the dot ceiling"),
    );
    let covered_verdict = fleet[&1].epochs().recognize(address, replay_dot);
    let uncovered_verdict = fleet[&1].epochs().recognize(address, uncovered_dot);
    let deliver_verdict = {
        let mut epochs = fleet[&1].epochs().clone();
        epochs.deliver(declaration.clone(), &Stability::new(ROSTER), &Cut::bottom())
    };
    for replica in fleet.values() {
        assert_eq!(
            replica.epochs().recognize(address, replay_dot),
            covered_verdict,
            "replica {} recognizes the covered old replay identically",
            replica.id()
        );
        assert_eq!(
            replica.epochs().recognize(address, uncovered_dot),
            uncovered_verdict,
            "replica {} refuses the uncovered old replay identically",
            replica.id()
        );
        let mut epochs = replica.epochs().clone();
        assert_eq!(
            epochs.deliver(declaration.clone(), &Stability::new(ROSTER), &Cut::bottom()),
            deliver_verdict,
            "replica {} absorbs the sealed declaration identically",
            replica.id()
        );
    }
}

fn cross_joiner_seal(fabric: &mut Fabric, fleet: &mut BTreeMap<u32, Replica>) {
    for station in ROSTER {
        let _ = act(fabric, fleet, station, |replica, out| {
            replica.insert_visible(0, out)
        });
    }
    fabric.drain(fleet);
    let address = act(fabric, fleet, 3, Replica::try_declare)
        .expect("the joiner declares the next generation");
    fabric.drain(fleet);
    assert_converged(fleet);
    for replica in fleet.values() {
        assert_eq!(replica.generation(), 3);
        let sealed = replica
            .epochs()
            .sealed()
            .find(|sealed| sealed.declaration() == address)
            .expect("the joiner's round sealed");
        assert!(
            sealed.contains(address),
            "the joiner's declaration participates in the sealed record"
        );
    }
}

fn repair_joiner_at_next_seal_edge(fleet: &mut BTreeMap<u32, Replica>, depth: NonZeroUsize) {
    let mut round = Vec::new();
    let _ = fleet
        .get_mut(&1)
        .expect("roster member")
        .try_declare(&mut round)
        .expect("the settled post-join fleet declares again");
    while let Some(note) = round.pop() {
        for station in ROSTER {
            if station == 3 && matches!(note, Note::Adoption { station: 1 | 2, .. }) {
                continue;
            }
            let mut out = Vec::new();
            fleet
                .get_mut(&station)
                .expect("roster member")
                .handle(&note, &mut out);
            round.extend(out);
        }
    }
    assert_eq!(
        ROSTER.map(|station| fleet[&station].generation()),
        [4, 4, 3],
        "the peers seal while the joiner waits on their adoption reports"
    );
    assert!(fleet[&3].adopted());

    crash(fleet, &ROSTER, depth, 3);
    assert!(
        fleet[&3].adopted(),
        "the joiner re-earns adoption from its fenced journal"
    );
    let mut ahead = Vec::new();
    fleet[&1].restate(&mut ahead);
    let next_floor = ahead
        .into_iter()
        .find(|note| matches!(note, Note::Report { generation: 4, .. }))
        .expect("the sealed peer reports its new floor first");
    fleet
        .get_mut(&3)
        .expect("the joiner remains in the roster")
        .handle(&next_floor, &mut Vec::new());
    assert_eq!(
        fleet[&3].parked_len(),
        1,
        "the next-generation floor parks until the sealed round is restated"
    );

    let mut edge_fabric = Fabric::new(0xF1EE_70B1, &ROSTER, 0);
    resync(&mut edge_fabric, fleet, 3);
    edge_fabric.drain(fleet);
    assert_converged(fleet);
    assert_eq!(fleet[&3].generation(), 4);
    assert_eq!(fleet[&3].parked_len(), 0);
}

/// A fixed-roster member can join from a sealed data checkpoint plus a
/// verified lineage proof. Its recognizer agrees with incumbents, it is a
/// required participant in the next seal, and the R-71 restatement lane
/// still repairs it when it later crashes at a seal edge.
#[test]
fn a_bootstrap_joiner_enters_the_fleet_and_crosses_the_next_seal() {
    let depth = horizon(3);
    let BootstrapFixture {
        mut fleet,
        mut fabric,
        address,
        declaration,
        replay,
    } = bootstrappable_fleet(depth);
    install_bootstrap_joiner(&mut fleet, depth);
    assert_bootstrap_verdict_parity(&fleet, address, &declaration, &replay);

    // The new member starts with no live floor authority. Ordinary repair
    // supplies current logs and reports; no checkpoint state is trusted as
    // a live witness.
    resync(&mut fabric, &fleet, 3);
    fabric.drain(&mut fleet);
    assert_converged(&fleet);

    cross_joiner_seal(&mut fabric, &mut fleet);
    repair_joiner_at_next_seal_edge(&mut fleet, depth);
}

/// Seals one generation from station 1, returning its address beside the
/// declaration that carried it.
fn seal_one_generation(
    fabric: &mut Fabric,
    fleet: &mut BTreeMap<u32, Replica>,
) -> (EpochAddress, Declaration) {
    for station in ROSTER {
        let _ = act(fabric, fleet, station, |replica, out| {
            replica.insert_visible(0, out)
        });
    }
    fabric.drain(fleet);
    let mut outbox = Vec::new();
    let address = fleet
        .get_mut(&1)
        .expect("roster member")
        .try_declare(&mut outbox)
        .expect("the settled plane declares");
    let declaration = outbox
        .iter()
        .find_map(|note| match note {
            Note::Declare { declaration } => Some(declaration.clone()),
            _ => None,
        })
        .expect("the declaration leaves through the outbox");
    fabric.post(1, outbox);
    fabric.drain(fleet);
    (address, declaration)
}

/// A dot exactly at a retained seal's join for `station`: the covered
/// replay whose duplicate verdict the recognizer owes.
fn covered_dot(replica: &Replica, address: EpochAddress, station: u32) -> Dot {
    let sealed = replica
        .epochs()
        .sealed()
        .find(|sealed| sealed.declaration() == address)
        .expect("the generation is retained");
    d(station, sealed.sealed_join().get(station))
}

/// Both old-addressed doors, read against one replica without disturbing
/// it: the recognizer verdict and the redelivery verdict.
fn old_addressed_verdicts(
    replica: &Replica,
    address: EpochAddress,
    dot: Dot,
    declaration: &Declaration,
) -> (Result<(), EpochRefusal>, Result<(), EpochRefusal>) {
    let recognized = replica.epochs().recognize(address, dot);
    let mut epochs = replica.epochs().clone();
    let delivered = epochs.deliver(declaration.clone(), &Stability::new(ROSTER), &Cut::bottom());
    (recognized, delivered)
}

/// A joiner admitted through a lineage proof *shorter* than its peers'
/// retained lineage carries a hole in its recognizer coverage: exactly
/// the generations the proof left out.
///
/// The distinction the door does not draw for the caller is between the
/// horizon a joiner *declares* (a transport fact: how far back a replay
/// can reach) and the coverage its proof *supplies*. `bootstrap` refuses
/// a proof longer than the declared horizon and accepts any shorter one,
/// so a caller that trims the proof — a checkpoint-body retention policy
/// is the live example, since a released body may take its seal record
/// with it — hands the joiner a machine that answers old-addressed
/// traffic differently from every incumbent.
///
/// What this pins, in the order it matters. Within the proof's coverage
/// both doors agree with the incumbents exactly. Below it they part, and
/// part *fail-closed*: the incumbents grant the duplicate verdict while
/// the joiner refuses [`EpochRefusal::BeyondHorizon`], which is the
/// machine declining to guess, never a wrong absorption. The hole does
/// not heal by catching up — the joiner never re-acquires a generation
/// its proof omitted — and the fleet stays converged on document state
/// throughout, so nothing here is a divergence of the replicated value.
/// It closes only by eviction, after exactly the number of further seals
/// that brings the joiner's own lineage to the declared horizon, at
/// which point both sides hold the same newest `horizon` generations and
/// answer identically again.
#[test]
fn a_truncated_bootstrap_proof_holds_a_bounded_recognizer_hole() {
    let depth = horizon(3);
    let mut fleet = fleet_of(&ROSTER, depth);
    let mut fabric = Fabric::new(0xF1EE_70B5, &ROSTER, 0);

    let (first, first_declaration) = seal_one_generation(&mut fabric, &mut fleet);
    let (second, second_declaration) = seal_one_generation(&mut fabric, &mut fleet);
    assert_converged(&fleet);
    assert_eq!(
        fleet[&1].epochs().sealed().count(),
        2,
        "the incumbents retain both sealed generations"
    );

    // The caller trims its proof to the newest generation alone: the
    // shape a body-retention policy produces once older bodies are
    // released.
    let full = lineage_of(&fleet[&1]);
    let newest = full.entries().last().expect("a retained seal").clone();
    let truncated =
        LineageProofRecord::try_new(Vec::from([newest])).expect("one entry is consecutive");

    let checkpoint = fleet[&3]
        .joiner_checkpoint()
        .expect("station 3 stands exactly at the seal");
    let joiner = Replica::bootstrap(checkpoint, &ROSTER, depth, &truncated)
        .expect("a trimmed proof still names the checkpoint's own generation");
    assert_eq!(
        joiner.epochs().sealed().count(),
        1,
        "the joiner's coverage is the proof's length, not its declared horizon"
    );
    assert_eq!(
        joiner.epochs().horizon(),
        fleet[&1].epochs().horizon(),
        "the declared horizon agrees; only the supplied coverage differs"
    );

    // Within the proof's coverage the two doors are indistinguishable.
    let inside = covered_dot(&fleet[&1], second, 1);
    assert_eq!(
        old_addressed_verdicts(&joiner, second, inside, &second_declaration),
        old_addressed_verdicts(&fleet[&1], second, inside, &second_declaration),
        "inside the proof's coverage the joiner answers exactly as the incumbents do"
    );

    // Below it they part, fail-closed.
    let outside = covered_dot(&fleet[&1], first, 1);
    assert_eq!(
        old_addressed_verdicts(&fleet[&1], first, outside, &first_declaration),
        (Ok(()), Ok(())),
        "the incumbents absorb the covered replay and the redelivered declaration"
    );
    assert_eq!(
        old_addressed_verdicts(&joiner, first, outside, &first_declaration),
        (
            Err(EpochRefusal::BeyondHorizon { epoch: first }),
            Err(EpochRefusal::BeyondHorizon { epoch: first })
        ),
        "the joiner refuses to guess on both doors rather than absorbing blind"
    );

    let _ = fleet.insert(3, joiner);
    resync(&mut fabric, &fleet, 3);
    fabric.drain(&mut fleet);
    assert_eq!(
        fleet[&1].text(),
        fleet[&3].text(),
        "the hole is in the recognizer, never in the replicated document"
    );

    // One further seal does not close it: the omitted generation is gone
    // for good, and the joiner is simply one generation short.
    let _ = seal_one_generation(&mut fabric, &mut fleet);
    assert_eq!(
        fleet[&3].epochs().recognize(first, outside),
        Err(EpochRefusal::BeyondHorizon { epoch: first }),
        "sealing forward never re-acquires an omitted generation"
    );
    assert_eq!(
        fleet[&1].epochs().recognize(first, outside),
        Ok(()),
        "the incumbents still hold it, so the fleet still disagrees"
    );

    // The second one does, by eviction: both sides now hold the same
    // newest `horizon` generations.
    let _ = seal_one_generation(&mut fabric, &mut fleet);
    assert_eq!(
        fleet[&1].epochs().sealed().count(),
        depth.get(),
        "the incumbents have filled their declared horizon"
    );
    assert_eq!(
        fleet[&3].epochs().recognize(first, outside),
        fleet[&1].epochs().recognize(first, outside),
        "eviction closes the hole: the omitted generation is now beyond everyone's horizon"
    );
    assert_converged(&fleet);
}

/// Bootstrap checkpoints name a completed seal, and the join boundary
/// rejects either authority when it does not describe the configured
/// station and generation.
#[test]
fn a_bootstrap_joiner_refuses_unsealed_and_mismatched_authorities() {
    let depth = horizon(3);
    assert!(
        Replica::new(3, &ROSTER, depth)
            .joiner_checkpoint()
            .is_none(),
        "birth is not a sealed data checkpoint"
    );

    let mut fleet = fleet_of(&ROSTER, depth);
    let mut fabric = Fabric::new(0xF1EE_70B2, &ROSTER, 0);
    for station in ROSTER {
        let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.insert_visible(0, out)
        });
    }
    fabric.drain(&mut fleet);
    let _ = act(&mut fabric, &mut fleet, 1, Replica::try_declare)
        .expect("the settled first plane declares");
    fabric.drain(&mut fleet);
    assert_converged(&fleet);

    let outside_roster = fleet[&3]
        .joiner_checkpoint()
        .expect("station 3 stands at the first seal");
    let stale_generation = fleet[&3]
        .joiner_checkpoint()
        .expect("station 3 stands at the first seal");
    let first_lineage = lineage_of(&fleet[&1]);
    // Under the roster inversion (PRD 0028 R4) the configured slice
    // grounds the *oldest* retained generation, so a wrong one surfaces
    // at the machine's own per-generation containment proof: the lineage
    // carries station 3's dots, which the misconfigured founding roster
    // cannot contain. The joiner door's own refusal
    // (`StationOutsideRoster`) now guards the derived roster instead ---
    // a checkpoint stamped for a station the walked lineage never
    // admitted.
    assert!(matches!(
        Replica::bootstrap(outside_roster, &[1, 2], depth, &first_lineage),
        Err(JoinerBootstrapError::Epoch(
            crate::metis::EpochBootstrapError::ForeignStation { station: 3, .. }
        ))
    ));
    assert!(
        fleet[&1].joiner_checkpoint_for(9).is_none(),
        "a checkpoint is never re-stamped for a station off the exporter's derived roster"
    );

    for station in ROSTER {
        let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.insert_visible(0, out)
        });
    }
    fabric.drain(&mut fleet);
    let _ = act(&mut fabric, &mut fleet, 1, Replica::try_declare)
        .expect("the settled second plane declares");
    fabric.drain(&mut fleet);
    assert_converged(&fleet);

    let second_lineage = lineage_of(&fleet[&1]);
    assert!(matches!(
        Replica::bootstrap(stale_generation, &ROSTER, depth, &second_lineage),
        Err(JoinerBootstrapError::GenerationMismatch {
            checkpoint: 2,
            lineage: 3,
        })
    ));
}