minerva 0.2.0

//! The boundary disciplines, pinned as counterexamples: why the fleet's
//! replicas delete with a minted dot, why a declarer dry-runs the fold
//! before opening a window, and why the retirement round's gathering is
//! cut-gated and generation-scoped. Each pin is a shaped demonstration in
//! the store-algebra style: it exhibits the divergence or refusal that
//! makes the discipline load-bearing rather than stylistic.

extern crate alloc;

mod classification;
mod durability;
mod retirement;
mod writer;

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

use crate::metis::tests::support::dot as d;
use crate::metis::{Cut, DotSet};

use super::Note;
use super::replica::Replica;

/// The parked lane's bounded-record pin (the review-caught round-three
/// P2): a note that outruns its record parks *once*, however many
/// repair rounds re-state it while its prerequisite is still missing.
/// Without the park-once rule, every re-stated future-generation report
/// and every re-stated floor-blocked oath would count as effectful,
/// compounding the parked set, the volatile buffer, and the journal on
/// each repair; with it, duplicates neither park nor fence, and the
/// prerequisite's arrival retires the parked copy through the ordinary
/// retry loop.
#[test]
fn out_of_reach_protocol_notes_park_once_and_cure_on_the_prerequisite() {
    const ROSTER: [u32; 3] = [1, 2, 3];
    let depth = NonZeroUsize::new(2).expect("positive");
    let mut two = Replica::new(2, &ROSTER, depth);

    // A future-generation report (its window has not opened here) and a
    // floor-blocked retirement oath (its carried cut is not yet
    // delivered here), each re-stated five times, as a repeated repair
    // round would.
    let future = Note::Report {
        generation: 2,
        station: 3,
        cut: Cut::floor_of(&DotSet::new()),
    };
    let mut applied = DotSet::new();
    let _ = applied.insert(d(1, 1));
    let blocked = Note::Retire {
        generation: 1,
        station: 3,
        applied: applied.clone(),
        at: Cut::floor_of(&applied),
    };
    for _ in 0..5 {
        two.handle(&future, &mut Vec::new());
        two.handle(&blocked, &mut Vec::new());
    }
    assert_eq!(two.parked_len(), 2, "an exact duplicate re-parks nothing");
    assert_eq!(
        two.volatile_len() + two.journal_notes_len(),
        2,
        "each distinct out-of-reach note is recorded exactly once, fenced \
         or pending, however many repair rounds re-stated it"
    );

    // The prerequisite lands: station 1's weave raises the floor, the
    // parked oath retries through the ordinary loop and counts, and only
    // the future-generation report stays parked, waiting for its seal.
    let mut one = Replica::new(1, &ROSTER, depth);
    let mut outbox = Vec::new();
    let _ = one.insert_visible(0, &mut outbox);
    for note in &outbox {
        two.handle(note, &mut Vec::new());
    }
    assert_eq!(
        two.parked_len(),
        1,
        "the cut-gated oath retried and counted; the future report waits"
    );
}

/// The zero-counter restatement pin (the review-caught round-four gap):
/// the sealed join is a sparse vector, so a member that adopted at
/// counter zero (it minted nothing that generation) has no entry, and a
/// lineage restatement that walks the join's entries silently drops
/// exactly that member's adoption report. A laggard whose missing
/// prerequisite is that report would then never seal, with repair
/// sourced only from peers whose windows are long cleared. The repair
/// lane therefore walks the roster and reads each member through the
/// join (absent means zero), and this exhibit drives the exact wedge:
/// two peers seal, the laggard misses the silent member's adoption
/// report, crashes, and completes its seal from the sealed peers'
/// restatements alone.
#[test]
fn a_sealed_peer_restates_a_zero_counter_adoption_report() {
    const ROSTER: [u32; 3] = [1, 2, 3];
    let depth = NonZeroUsize::new(2).expect("positive");
    let mut one = Replica::new(1, &ROSTER, depth);
    let mut two = Replica::new(2, &ROSTER, depth);
    let mut three = Replica::new(3, &ROSTER, depth);

    // Stations 1 and 2 mint; station 3 never does (its adoption will
    // report counter zero and leave no sealed-join entry).
    let mut seed = Vec::new();
    let _ = one.insert_visible(0, &mut seed);
    let mut out_two = Vec::new();
    let _ = two.insert_visible(1, &mut out_two);
    seed.extend(out_two);
    let mut flood = seed;
    while let Some(note) = flood.pop() {
        for replica in [&mut one, &mut two, &mut three] {
            let mut out = Vec::new();
            replica.handle(&note, &mut out);
            flood.extend(out);
        }
    }

    // Station 1 declares; the rounds flood, except that station 3's
    // adoption report never reaches station 1 (lost in flight).
    let mut flood = Vec::new();
    let _ = one
        .try_declare(&mut flood)
        .expect("a settled watermark self-supports the declaration");
    while let Some(note) = flood.pop() {
        for replica in [&mut one, &mut two, &mut three] {
            if replica.id() == 1 && matches!(note, Note::Adoption { station: 3, .. }) {
                continue;
            }
            let mut out = Vec::new();
            replica.handle(&note, &mut out);
            flood.extend(out);
        }
    }
    assert_eq!(two.generation(), 2, "station 2 seals");
    assert_eq!(three.generation(), 2, "station 3 seals");
    assert_eq!(one.generation(), 1, "the laggard waits on the lost report");
    assert!(one.adopted(), "the laggard adopted before the loss");

    // The laggard crashes; repair comes only from peers whose windows
    // are sealed and cleared, so the lineage restatement is the one
    // source of the zero-counter adoption report.
    let journal = one.journal().clone();
    let mut one = Replica::rehydrate(1, &ROSTER, depth, &journal);
    let mut flood = Vec::new();
    for peer in [&two, &three] {
        peer.reserve_logs(&mut flood);
        peer.restate(&mut flood);
    }
    while let Some(note) = flood.pop() {
        let mut out = Vec::new();
        one.handle(&note, &mut out);
        for response in out {
            let mut echo = Vec::new();
            two.handle(&response, &mut echo);
            three.handle(&response, &mut echo);
            flood.extend(echo);
        }
    }
    assert_eq!(
        one.generation(),
        2,
        "the restated zero-counter adoption report completes the seal"
    );
    assert_eq!(one.seals, two.seals, "the seal record agrees");
    assert_eq!(one.effective_order(), two.effective_order());
}

/// The retention question the repair lane does not have (the
/// review-raised round-six P1, refuted by mechanism): peers can never
/// seal away the only copy of a window delta a member still needs.
/// In the group-communication taxonomy this pin is the executable form
/// of the seal's guarantee class, a whole-roster safe-prefix barrier
/// (safe indication plus reliable prefix in the survey vocabulary the
/// S265 intake reads; its citations are transcribed, not verified, but
/// this pin stands on the mechanism alone): nothing is licensed until
/// it is stable at every member, which is why the "missing sole copy"
/// schedule cannot be constructed against an honest fleet.
/// Old-plane mints happen only pre-adoption, so every window delta sits
/// inside its minter's own adoption cut, hence inside any seal's sealed
/// join; and the seal fires only when the watermark, the meet of every
/// member's reported floor, covers that join. A member missing the
/// delta pins the watermark below it and nothing seals anywhere, and
/// under the durable-floor duty a reported floor is a journaled floor,
/// so by the time any peer seals, every member durably holds every
/// window delta. Previous-generation log retention is therefore
/// unnecessary; this pin drives the exact corner.
#[test]
fn no_seal_outruns_a_member_missing_window_traffic() {
    const ROSTER: [u32; 3] = [1, 2, 3];
    let depth = NonZeroUsize::new(2).expect("positive");
    let mut one = Replica::new(1, &ROSTER, depth);
    let mut two = Replica::new(2, &ROSTER, depth);
    let mut three = Replica::new(3, &ROSTER, depth);

    // Seeds settle everywhere.
    let mut flood = Vec::new();
    let mut out = Vec::new();
    let _ = one.insert_visible(0, &mut out);
    flood.extend(out);
    let mut out = Vec::new();
    let _ = two.insert_visible(1, &mut out);
    flood.extend(out);
    let mut out = Vec::new();
    let _ = three.insert_visible(2, &mut out);
    flood.extend(out);
    while let Some(note) = flood.pop() {
        for replica in [&mut one, &mut two, &mut three] {
            let mut out = Vec::new();
            replica.handle(&note, &mut out);
            flood.extend(out);
        }
    }

    // The declaration reaches station 2 first, and 2 mints window
    // traffic before any round moves further: old-addressed, above the
    // cut, inside 2's future adoption cut.
    let mut declare_out = Vec::new();
    let _ = one
        .try_declare(&mut declare_out)
        .expect("a settled watermark self-supports the declaration");
    let mut two_out = Vec::new();
    for note in &declare_out {
        two.handle(note, &mut two_out);
    }
    let mut mint_out = Vec::new();
    let _ = two.insert_visible(0, &mut mint_out);
    let window_delta = mint_out
        .iter()
        .find(|note| matches!(note, Note::Old { .. }))
        .cloned()
        .expect("the window mint");

    // Everything floods, except station 3 never receives the window
    // delta: its sole copy is lost in flight.
    let mut flood = Vec::new();
    flood.extend(declare_out);
    flood.extend(two_out);
    flood.extend(mint_out);
    while let Some(note) = flood.pop() {
        for replica in [&mut one, &mut two, &mut three] {
            if replica.id() == 3 && note == window_delta {
                continue;
            }
            let mut out = Vec::new();
            replica.handle(&note, &mut out);
            flood.extend(out);
        }
    }
    for replica in [&one, &two, &three] {
        assert_eq!(
            replica.generation(),
            1,
            "replica {}: no seal while a member misses window traffic",
            replica.id()
        );
        assert_eq!(replica.epochs().sealed().count(), 0);
    }

    // The delta lands; the meets rise and the lifecycle completes.
    let mut flood = vec![window_delta];
    while let Some(note) = flood.pop() {
        for replica in [&mut one, &mut two, &mut three] {
            let mut out = Vec::new();
            replica.handle(&note, &mut out);
            flood.extend(out);
        }
    }
    for replica in [&one, &two, &three] {
        assert_eq!(replica.generation(), 2, "the delivered delta unblocks");
    }
    assert_eq!(one.effective_order(), two.effective_order());
    assert_eq!(two.effective_order(), three.effective_order());
}