minerva 0.2.0

//! Fleet lifecycles pinned as examples: each drives the whole shipped
//! stack across a specific hazard the charter names.

extern crate alloc;

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

use crate::metis::EpochRefusal;
use crate::metis::tests::support::dot as d;

use super::fabric::Fabric;
use super::replica::Replica;
use super::{Note, act, assert_converged, fleet_of};

mod arrival;
mod bootstrap;
mod recovery;

/// The second generation's own hygiene round: a fresh element, a fresh
/// delete, fresh acknowledgements gathered by the fresh tracker, one
/// excision everywhere, convergence whole.
fn second_generation_round(fabric: &mut Fabric, fleet: &mut BTreeMap<u32, Replica>) {
    let _ = act(fabric, fleet, 3, |replica, out| {
        replica.insert_visible(0, out)
    });
    fabric.drain(fleet);
    let deleted = act(fabric, fleet, 1, |replica, out| {
        replica.delete_visible(0, out)
    });
    assert!(deleted.is_some(), "the new plane has an element to delete");
    fabric.drain(fleet);
    for &station in &ROSTER {
        assert_eq!(
            condense_at(fleet, station),
            1,
            "the second generation's round excises its own tombstone"
        );
    }
    assert_converged(fleet);
}

/// One local condense at `station`, returning its excision count.
fn condense_at(fleet: &mut BTreeMap<u32, Replica>, station: u32) -> usize {
    fleet.get_mut(&station).expect("roster member").condense()
}

const ROSTER: [u32; 3] = [1, 2, 3];

fn horizon(depth: usize) -> NonZeroUsize {
    NonZeroUsize::new(depth).expect("a positive horizon")
}

/// The honest fleet crosses one boundary with every kind of traffic in
/// flight: old edits before the cut, a window birth, a window delete, a
/// window move judged in the old topology, native mints held back by the
/// epoch gate, then the seal. Everyone converges, the projection carries
/// the window judgment, and a replayed old delta absorbs.
#[test]
fn the_honest_fleet_crosses_one_boundary_and_converges() {
    let mut fleet = fleet_of(&ROSTER, horizon(2));
    let mut fabric = Fabric::new(0xF1EE_7001, &ROSTER, 8);

    // Seed content from all three stations under an interleaved schedule.
    let mut replayable = None;
    for (turn, &(station, at)) in [(1u32, 0usize), (2, 1), (3, 2), (1, 1), (2, 0), (3, 4)]
        .iter()
        .enumerate()
    {
        let replica = fleet.get_mut(&station).expect("roster member");
        let mut outbox = Vec::new();
        let _ = replica.insert_visible(at, &mut outbox);
        if turn == 0 {
            replayable = outbox
                .iter()
                .find(|note| matches!(note, Note::Old { .. }))
                .cloned();
        }
        fabric.post(station, outbox);
        for _ in 0..2 {
            let _ = fabric.step(&mut fleet);
        }
    }
    fabric.drain(&mut fleet);
    assert_converged(&fleet);
    let baseline = fleet[&1].effective_order().len();
    assert_eq!(baseline, 6, "all six seeded elements are visible");

    // Declare over the settled watermark.
    let _ = act(&mut fabric, &mut fleet, 1, |replica, out| {
        replica.try_declare(out)
    })
    .expect("a settled watermark licenses the declaration");

    // Window traffic minted before the declaration lands anywhere: a birth,
    // a delete of sealed content, and a move judged in the old topology.
    let _ = act(&mut fabric, &mut fleet, 2, |replica, out| {
        replica.insert_visible(3, out)
    });
    let deleted = act(&mut fabric, &mut fleet, 3, |replica, out| {
        replica.delete_visible(1, out)
    })
    .expect("the seeded document has an element to delete");
    let _ = act(&mut fabric, &mut fleet, 3, |replica, out| {
        replica.move_visible(4, Some(0), out)
    })
    .expect("the seeded document has an element to move");

    // Let the rounds run until someone adopts, then mint native traffic
    // that the laggards must hold behind the epoch gate.
    let mut adopter = None;
    while adopter.is_none() {
        assert!(
            fabric.step(&mut fleet),
            "the confirmation round must complete from in-flight traffic"
        );
        adopter = ROSTER.iter().copied().find(|id| fleet[id].adopted());
    }
    let adopter = adopter.expect("just found");
    let _ = act(&mut fabric, &mut fleet, adopter, |replica, out| {
        replica.insert_visible(0, out)
    });
    let _ = act(&mut fabric, &mut fleet, adopter, |replica, out| {
        replica.delete_visible(0, out)
    });

    fabric.drain(&mut fleet);
    assert_converged(&fleet);
    for replica in fleet.values() {
        assert_eq!(replica.generation(), 2, "the window sealed everywhere");
        assert_eq!(replica.seals.len(), 1);
    }
    // The projection carries the window judgment: birth in, delete out.
    let (_, projection) = &fleet[&1].seals[0];
    assert_eq!(
        projection.order().len(),
        baseline,
        "one window birth and one window delete leave the count unchanged"
    );
    // The consignment carries the projection whole into the next generation's
    // base (the stage-four document-plane carry, ruling R-50), and the
    // adopter's native insert and delete cancelled each other, so the new
    // plane opens reading exactly the sealed projection.
    assert_eq!(fleet[&1].effective_order(), projection.order());

    // A replayed pre-boundary delta reads as a duplicate everywhere and
    // changes nothing.
    let before: Vec<u8> = fleet[&1].text().store().to_bytes();
    let replay = replayable.expect("the first seed op emitted an old note");
    fabric.broadcast(1, &replay);
    fabric.drain(&mut fleet);
    assert_eq!(fleet[&1].text().store().to_bytes(), before);
    assert_converged(&fleet);
    // The deleted element's old identity stays answerable while retained.
    let sealed = fleet[&1]
        .epochs()
        .sealed()
        .next()
        .expect("one sealed epoch");
    assert_eq!(sealed.declaration().generation(), 1);
    assert_eq!(
        fleet[&1].epochs().recognize(sealed.declaration(), deleted),
        Ok(()),
        "a sealed-plane dot below the sealed join is a duplicate"
    );
}

/// Window births and moves cross the seal alive (the stage-four
/// document-plane carry, ruling R-50): laggards keep weaving and moving
/// old-addressed content while the fleet adopts, the shadow judges every
/// delivery in the old topology, the seal consigns the judged outcome
/// whole, the new plane opens reading exactly the frozen projection at
/// every replica, and the carried content re-founds cleanly at the next
/// boundary.
#[test]
fn window_births_and_moves_cross_the_seal_alive() {
    let mut fleet = fleet_of(&ROSTER, horizon(2));
    let mut fabric = Fabric::new(0xF1EE_7007, &ROSTER, 5);

    for &(station, at) in &[(1u32, 0usize), (2, 1), (3, 2)] {
        let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.insert_visible(at, out)
        });
    }
    fabric.drain(&mut fleet);
    assert_converged(&fleet);
    assert_eq!(fleet[&1].effective_order().len(), 3);

    let _ = act(&mut fabric, &mut fleet, 1, |replica, out| {
        replica.try_declare(out)
    })
    .expect("a settled watermark licenses the declaration");

    // Old-addressed window traffic minted before anyone adopts: a birth
    // at the front, a move of the visual tail to the head, and a second
    // birth mid-document. No native traffic in this scenario, so the next
    // plane must open reading exactly the sealed projection.
    let _ = act(&mut fabric, &mut fleet, 2, |replica, out| {
        replica.insert_visible(0, out)
    });
    let moved = act(&mut fabric, &mut fleet, 3, |replica, out| {
        replica.move_visible(2, None, out)
    });
    assert!(
        moved.is_some(),
        "the seeded document has an element to move"
    );
    let _ = act(&mut fabric, &mut fleet, 3, |replica, out| {
        replica.insert_visible(1, out)
    });

    fabric.drain(&mut fleet);
    assert_converged(&fleet);
    for replica in fleet.values() {
        assert_eq!(replica.generation(), 2, "the window sealed everywhere");
        let (_, projection) = &replica.seals[0];
        assert_eq!(
            projection.order().len(),
            5,
            "both window births survive the boundary"
        );
        assert_eq!(
            replica.effective_order(),
            projection.order(),
            "the new plane opens on the sealed projection"
        );
    }

    // The carried plane is an ordinary plane: the next generation edits
    // it (a weave, a move over carried content, a delete), settles, and
    // re-founds again, sweeping what the first boundary carried.
    let _ = act(&mut fabric, &mut fleet, 2, |replica, out| {
        replica.insert_visible(2, out)
    });
    let moved = act(&mut fabric, &mut fleet, 1, |replica, out| {
        replica.move_visible(4, Some(0), out)
    });
    assert!(moved.is_some(), "the carried plane moves like any other");
    let deleted = act(&mut fabric, &mut fleet, 3, |replica, out| {
        replica.delete_visible(1, out)
    });
    assert!(
        deleted.is_some(),
        "the carried plane deletes like any other"
    );
    fabric.drain(&mut fleet);
    assert_converged(&fleet);

    let _ = act(&mut fabric, &mut fleet, 3, |replica, out| {
        replica.try_declare(out)
    })
    .expect("the settled second-generation watermark licenses the declaration");
    fabric.drain(&mut fleet);
    assert_converged(&fleet);
    for replica in fleet.values() {
        assert_eq!(replica.generation(), 3, "the second boundary sealed");
        assert_eq!(replica.seals.len(), 2);
        let (_, projection) = &replica.seals[1];
        assert_eq!(
            replica.effective_order(),
            projection.order(),
            "a quiesced second window opens on its own projection"
        );
        assert_eq!(replica.effective_order().len(), 5);
    }
}

/// Two genuinely concurrent declarations fix the same winner at every
/// replica; the loser is retired as a candidate of the sealed epoch, and
/// its replays absorb.
#[test]
fn competing_declarations_fix_one_winner_everywhere() {
    let mut fleet = fleet_of(&ROSTER, horizon(2));
    let mut fabric = Fabric::new(0xF1EE_7002, &ROSTER, 7);

    for &(station, at) in &[(1u32, 0usize), (2, 0), (3, 1)] {
        let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.insert_visible(at, out)
        });
    }
    fabric.drain(&mut fleet);

    // Both declare before either delivery lands: concurrent candidates.
    let first = act(&mut fabric, &mut fleet, 1, |replica, out| {
        replica.try_declare(out)
    })
    .expect("station 1 declares over the settled watermark");
    let second = act(&mut fabric, &mut fleet, 3, |replica, out| {
        replica.try_declare(out)
    })
    .expect("station 3 declares concurrently");
    assert_ne!(first, second);

    fabric.drain(&mut fleet);
    assert_converged(&fleet);
    for replica in fleet.values() {
        assert_eq!(replica.generation(), 2);
        let sealed = replica.epochs().sealed().next().expect("one sealed epoch");
        assert!(
            sealed.contains(first) && sealed.contains(second),
            "both candidates retire into the sealed epoch"
        );
        let winner = sealed.declaration();
        assert!(
            winner == first || winner == second,
            "the winner is one of the two candidates"
        );
    }
}

/// A severed member freezes the epoch, not the fleet: the window stays
/// open, nothing corrupts, and healing lets the seal fire (the charter's
/// R8 liveness honesty).
#[test]
fn a_silent_member_freezes_the_seal_until_it_heals() {
    let mut fleet = fleet_of(&ROSTER, horizon(2));
    let mut fabric = Fabric::new(0xF1EE_7003, &ROSTER, 0);

    for &(station, at) in &[(1u32, 0usize), (2, 1), (3, 2)] {
        let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.insert_visible(at, out)
        });
    }
    fabric.drain(&mut fleet);

    fabric.sever(&[3]);
    let _ = act(&mut fabric, &mut fleet, 1, |replica, out| {
        replica.try_declare(out)
    })
    .expect("the watermark was witnessed before the severance");
    let _ = act(&mut fabric, &mut fleet, 2, |replica, out| {
        replica.delete_visible(0, out)
    });
    fabric.drain(&mut fleet);

    assert!(fabric.in_flight() > 0, "traffic waits at the severance");
    for &station in &[1u32, 2] {
        assert!(
            !fleet[&station].adopted(),
            "no adoption without the silent member's confirmation"
        );
        assert_eq!(fleet[&station].generation(), 1);
        assert!(fleet[&station].seals.is_empty());
    }

    fabric.heal();
    fabric.drain(&mut fleet);
    assert_converged(&fleet);
    for replica in fleet.values() {
        assert_eq!(replica.generation(), 2, "healing lets the seal fire");
    }
}

/// Two full re-foundations advance the lineage; a horizon of one evicts
/// the first epoch, whose address then refuses `BeyondHorizon` identically
/// at every replica, while the retained epoch keeps answering duplicates.
#[test]
fn two_epochs_advance_the_lineage_and_the_horizon_truncates() {
    let mut fleet = fleet_of(&ROSTER, horizon(1));
    let mut fabric = Fabric::new(0xF1EE_7004, &ROSTER, 6);

    let mut first_gen_note = None;
    for &(station, at) in &[(1u32, 0usize), (2, 1), (3, 0), (2, 2)] {
        let replica = fleet.get_mut(&station).expect("roster member");
        let mut outbox = Vec::new();
        let _ = replica.insert_visible(at, &mut outbox);
        if first_gen_note.is_none() {
            first_gen_note = outbox
                .iter()
                .find(|note| matches!(note, Note::Old { .. }))
                .cloned();
        }
        fabric.post(station, outbox);
    }
    fabric.drain(&mut fleet);

    let first_address = act(&mut fabric, &mut fleet, 2, |replica, out| {
        replica.try_declare(out)
    })
    .expect("first declaration");
    let _ = act(&mut fabric, &mut fleet, 3, |replica, out| {
        replica.delete_visible(0, out)
    });
    fabric.drain(&mut fleet);
    for replica in fleet.values() {
        assert_eq!(replica.generation(), 2);
    }

    // Second epoch over the re-founded plane.
    let mut second_gen_dot = None;
    for &(station, at) in &[(3u32, 0usize), (1, 1)] {
        let dot = act(&mut fabric, &mut fleet, station, |replica, out| {
            replica.insert_visible(at, out)
        });
        second_gen_dot = Some(dot);
    }
    fabric.drain(&mut fleet);
    let second_address = act(&mut fabric, &mut fleet, 3, |replica, out| {
        replica.try_declare(out)
    })
    .expect("second declaration over the re-founded plane");
    assert_eq!(second_address.generation(), 2);
    fabric.drain(&mut fleet);
    assert_converged(&fleet);

    for replica in fleet.values() {
        assert_eq!(replica.generation(), 3, "two seals landed");
        assert_eq!(replica.seals.len(), 2);
        assert_eq!(
            replica.epochs().sealed().count(),
            1,
            "the horizon of one retains only the newest seal"
        );
        assert_eq!(
            replica.epochs().recognize(first_address, d(1, 1)),
            Err(EpochRefusal::BeyondHorizon {
                epoch: first_address
            }),
            "below the horizon the machine refuses to guess"
        );
        let dot = second_gen_dot.expect("second-generation content minted");
        assert_eq!(
            replica.epochs().recognize(second_address, dot),
            Ok(()),
            "the retained epoch answers its duplicates"
        );
    }

    // A first-generation replay from outside the horizon changes nothing.
    let before: Vec<u8> = fleet[&1].text().store().to_bytes();
    let replay = first_gen_note.expect("a first-generation note was captured");
    fabric.broadcast(1, &replay);
    fabric.drain(&mut fleet);
    assert_eq!(fleet[&1].text().store().to_bytes(), before);
    assert_converged(&fleet);
}

/// The whole hygiene lifecycle rides the fleet and crosses a boundary:
/// pin, excise the target, retract the testimony dotted, unpin, excise
/// the anchor, then re-found. A surviving movement testimony's placement
/// anchor is withheld from the retirement acknowledgements (excising it
/// would silently re-decide the move: GC must never move text), the
/// dotted retirement releases the pin, and the excision counts are exact
/// at every stage, so a condense that reaches past its license is loud
/// here. A duplicated weave re-learns an excised locus and the next
/// condense excises it again; byte identity holds at every settled point
/// and across the seal.
#[test]
fn the_hygiene_lifecycle_pins_excises_and_crosses_the_boundary() {
    let mut fleet = fleet_of(&ROSTER, horizon(2));
    let mut fabric = Fabric::new(0xF1EE_7006, &ROSTER, 0);

    // Seed a, b, c left to right, then d at the front. Settle after each
    // weave so every anchor choice is deterministic.
    let mut c_weave = None;
    for &(station, at) in &[(1u32, 0usize), (2, 1), (3, 2), (1, 0)] {
        let replica = fleet.get_mut(&station).expect("roster member");
        let mut outbox = Vec::new();
        let dot = replica.insert_visible(at, &mut outbox);
        if station == 3 {
            assert_eq!(dot, d(3, 1), "c is station 3's first weave");
            c_weave = outbox
                .iter()
                .find(|note| matches!(note, Note::Old { .. }))
                .cloned();
        }
        fabric.post(station, outbox);
        fabric.drain(&mut fleet);
    }
    let c_weave = c_weave.expect("c's weave note was captured");
    assert_eq!(fleet[&1].effective_order().len(), 4, "d, a, b, c visible");

    // Station 2 moves c to sit after d: the testimony's placement anchor
    // is d, which the acknowledgement round must pin while the testimony
    // survives.
    let moved = act(&mut fabric, &mut fleet, 2, |replica, out| {
        replica.move_visible(3, Some(0), out)
    });
    assert_eq!(moved, Some(d(2, 2)), "the testimony dot");
    fabric.drain(&mut fleet);

    // Delete d (the anchor) and c (the target); both settle, with every
    // acknowledgement gathered.
    let deleted = act(&mut fabric, &mut fleet, 1, |replica, out| {
        replica.delete_visible(0, out)
    });
    assert_eq!(deleted, Some(d(1, 2)), "d is deleted");
    fabric.drain(&mut fleet);
    let deleted = act(&mut fabric, &mut fleet, 3, |replica, out| {
        replica.delete_visible(2, out)
    });
    assert_eq!(deleted, Some(d(3, 1)), "c is deleted");
    fabric.drain(&mut fleet);
    let baseline = fleet[&1].effective_order();
    assert_eq!(baseline.len(), 2, "a and b remain");

    // Staggered condense: replica 1 excises exactly the target's sterile
    // tombstone; the anchor d stays, pinned by the surviving testimony.
    // The reads are invariant.
    assert_eq!(
        condense_at(&mut fleet, 1),
        1,
        "c excises; d is pinned by the testimony's anchor"
    );
    assert_eq!(fleet[&1].effective_order(), baseline);
    assert_eq!(fleet[&1].text().store().skeleton_len(), 3);
    assert_eq!(fleet[&2].text().store().skeleton_len(), 4);

    // A duplicated weave re-learns the excised locus; the cycle is lawful
    // and the next condense excises it again.
    fabric.broadcast(3, &c_weave);
    fabric.drain(&mut fleet);
    assert_eq!(fleet[&1].text().store().skeleton_len(), 4, "c re-learned");
    assert_eq!(fleet[&1].effective_order(), baseline);
    assert_eq!(
        condense_at(&mut fleet, 1),
        1,
        "the re-learned tombstone excises again"
    );

    // The dotted orphan retirement: c's locus is gone at replica 1, so the
    // testimony targeting it retires under a minted marker, classifiable
    // at every future boundary. Everyone folds it, including the replicas
    // that still hold c's locus.
    let marker = act(&mut fabric, &mut fleet, 1, |replica, out| {
        replica.retire_orphans(out)
    });
    assert_eq!(marker, Some(d(1, 4)), "the retirement minted its marker");
    fabric.drain(&mut fleet);
    for replica in fleet.values() {
        assert_eq!(
            replica.moves().store().testimonies().count(),
            0,
            "replica {} folded the dotted retirement",
            replica.id()
        );
        assert_eq!(replica.effective_order(), baseline);
    }

    // The pin is released: the anchor's tombstone excises everywhere, the
    // laggards catch up on the target too, and the fleet reads one byte
    // sequence.
    assert_eq!(
        condense_at(&mut fleet, 1),
        1,
        "the unpinned anchor excises at the early condenser"
    );
    for &station in &[2u32, 3] {
        assert_eq!(
            condense_at(&mut fleet, station),
            2,
            "the laggard excises target and anchor together"
        );
    }
    assert_converged(&fleet);
    for replica in fleet.values() {
        assert_eq!(replica.text().store().skeleton_len(), 2);
    }

    // The boundary composes: the declarer's rebuild is the base plus the
    // covered log, insensitive to who condensed when, and the sealed
    // generation converges whole.
    let _ = act(&mut fabric, &mut fleet, 2, |replica, out| {
        replica.try_declare(out)
    })
    .expect("the settled watermark licenses the declaration");
    fabric.drain(&mut fleet);
    assert_converged(&fleet);
    for replica in fleet.values() {
        assert_eq!(replica.generation(), 2, "the fleet sealed");
        assert_eq!(replica.effective_order().len(), 2);
    }

    // The next generation's own round licenses its own hygiene.
    second_generation_round(&mut fabric, &mut fleet);
}