minerva 0.2.0

//! The fleet's generated law: under any op tape and any seeded adversarial
//! schedule (reordering and duplication throughout), two successive epoch
//! lifecycles seal everywhere and every replica converges to the same
//! record, the same pairs, the same bytes, and the same seal projections.

extern crate alloc;

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

use crate::metis::Dot;
use proptest::prelude::*;

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

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

/// One tape entry: which station acts, what it does, and where. Kinds:
/// insert, delete, move (before a window), or insert/delete (inside one).
type Op = (usize, usize, usize, usize);

fn station_of(op: Op) -> u32 {
    ROSTER[op.0 % ROSTER.len()]
}

/// Applies one pre-window op and lets the schedule interleave deliveries.
fn play(fabric: &mut Fabric, fleet: &mut BTreeMap<u32, Replica>, op: Op) {
    let (_, kind, at, to) = op;
    let station = station_of(op);
    match kind % 3 {
        0 => {
            let _ = act(fabric, fleet, station, |replica, out| {
                replica.insert_visible(at, out)
            });
        }
        1 => {
            let _ = act(fabric, fleet, station, |replica, out| {
                replica.delete_visible(at, out)
            });
        }
        _ => {
            let _ = act(fabric, fleet, station, |replica, out| {
                replica.move_visible(at, Some(to), out)
            });
        }
    }
    let steps = fabric.schedule().below(4);
    for _ in 0..steps {
        let _ = fabric.step(fleet);
    }
}

/// Applies one window op. An adopted replica edits natively (insert or
/// delete); a laggard keeps editing the old plane with the full alphabet
/// (insert, delete, move), traffic the shadow judges in the old topology
/// and the seal consignment carries whole into the next generation's base
/// (the stage-four document-plane carry, ruling R-50).
fn play_window(fabric: &mut Fabric, fleet: &mut BTreeMap<u32, Replica>, op: Op) {
    let (_, kind, at, to) = op;
    let station = station_of(op);
    if fleet[&station].adopted() {
        if kind % 2 == 0 {
            let _ = act(fabric, fleet, station, |replica, out| {
                replica.insert_visible(at, out)
            });
        } else {
            let _ = act(fabric, fleet, station, |replica, out| {
                replica.delete_visible(at, out)
            });
        }
    } else {
        match kind % 3 {
            0 => {
                let _ = act(fabric, fleet, station, |replica, out| {
                    replica.insert_visible(at, out)
                });
            }
            1 => {
                let _ = act(fabric, fleet, station, |replica, out| {
                    replica.delete_visible(at, out)
                });
            }
            _ => {
                let _ = act(fabric, fleet, station, |replica, out| {
                    replica.move_visible(at, Some(to), out)
                });
            }
        }
    }
    let steps = fabric.schedule().below(4);
    for _ in 0..steps {
        let _ = fabric.step(fleet);
    }
}

/// One full lifecycle: settle the tape, declare (possibly twice,
/// concurrently), run window traffic, drain to the seal.
fn cross_boundary(
    fabric: &mut Fabric,
    fleet: &mut BTreeMap<u32, Replica>,
    declarers: &[u32],
    window: &[Op],
    expected_generation: u64,
) {
    fabric.drain(fleet);
    for &declarer in declarers {
        let _ = act(fabric, fleet, declarer, |replica, out| {
            replica.try_declare(out)
        })
        .expect("a settled watermark self-supports the declaration");
    }
    // Roughly half the runs sever one member for the whole window: the
    // seal must wait for it (the R8 liveness honesty) and fire on heal.
    let severed = if fabric.schedule().one_in(2) {
        let station = ROSTER[fabric.schedule().below(ROSTER.len())];
        fabric.sever(&[station]);
        Some(station)
    } else {
        None
    };
    for &op in window {
        play_window(fabric, fleet, op);
    }
    if severed.is_some() {
        fabric.drain(fleet);
        assert!(
            fleet.values().all(|replica| replica.seals.len()
                < usize::try_from(expected_generation).expect("small") - 1),
            "no seal may fire while a member is silent"
        );
        fabric.heal();
    }
    fabric.drain(fleet);
    for replica in fleet.values() {
        assert_eq!(
            replica.generation(),
            expected_generation,
            "replica {} did not seal generation {}",
            replica.id(),
            expected_generation - 1
        );
    }
}

proptest! {
    #![proptest_config(ProptestConfig::with_cases(64))]

    /// The end-to-end convergence law across two re-foundations.
    #[test]
    fn any_schedule_seals_two_epochs_and_converges(
        seed in any::<u64>(),
        first_tape in prop::collection::vec((0usize..3, 0usize..3, 0usize..8, 0usize..8), 1..16),
        first_window in prop::collection::vec((0usize..3, 0usize..3, 0usize..8, 0usize..8), 0..6),
        second_tape in prop::collection::vec((0usize..3, 0usize..3, 0usize..8, 0usize..8), 1..10),
        second_window in prop::collection::vec((0usize..3, 0usize..3, 0usize..8, 0usize..8), 0..4),
        declarers in prop::sample::subsequence(ROSTER.to_vec(), 1..=2),
    ) {
        let mut fleet = fleet_of(&ROSTER, NonZeroUsize::new(2).expect("positive"));
        let mut fabric = Fabric::new(seed, &ROSTER, 7);

        for &op in &first_tape {
            play(&mut fabric, &mut fleet, op);
        }
        cross_boundary(&mut fabric, &mut fleet, &declarers, &first_window, 2);
        assert_converged(&fleet);

        for &op in &second_tape {
            play(&mut fabric, &mut fleet, op);
        }
        let second_declarer = [ROSTER[usize::try_from(seed % 3).expect("small")]];
        cross_boundary(&mut fabric, &mut fleet, &second_declarer, &second_window, 3);
        assert_converged(&fleet);

        for replica in fleet.values() {
            prop_assert_eq!(replica.seals.len(), 2);
        }
    }
}

/// Applies one op from the hygiene alphabet: the pre-window edits plus
/// the dotted orphan retirement, which is an ordinary old-plane write and
/// safe at any moment. Condense is deliberately NOT in this alphabet: a
/// condense claim is licensed for further writing only after every
/// replica has excised it (the racing-weave counterexample in
/// `boundary/writer.rs`), so the property fires it fleet-wide at settled points
/// instead, standing in for the consumer's apply round. The disciplined
/// alphabet below lifts exactly this restriction through the writer
/// discipline (the S224 arm).
fn play_hygiene(fabric: &mut Fabric, fleet: &mut BTreeMap<u32, Replica>, op: Op) {
    let (_, kind, at, to) = op;
    let station = station_of(op);
    match kind % 4 {
        0 => {
            let _ = act(fabric, fleet, station, |replica, out| {
                replica.insert_visible(at, out)
            });
        }
        1 => {
            let _ = act(fabric, fleet, station, |replica, out| {
                replica.delete_visible(at, out)
            });
        }
        2 => {
            let _ = act(fabric, fleet, station, |replica, out| {
                replica.move_visible(at, Some(to), out)
            });
        }
        _ => {
            let _ = act(fabric, fleet, station, |replica, out| {
                replica.retire_orphans(out)
            });
        }
    }
    let steps = fabric.schedule().below(4);
    for _ in 0..steps {
        let _ = fabric.step(fleet);
    }
}

/// Settles the fleet and applies one condense claim whole: every replica
/// excises against the same witness before any further write mints. The
/// per-replica excision counts must agree (same skeleton, same witness),
/// and the fleet must read one byte sequence afterwards.
fn condense_point(fabric: &mut Fabric, fleet: &mut BTreeMap<u32, Replica>) {
    fabric.drain(fleet);
    let mut counts = Vec::new();
    for &station in &ROSTER {
        counts.push(fleet.get_mut(&station).expect("roster member").condense());
    }
    assert!(
        counts.windows(2).all(|pair| pair[0] == pair[1]),
        "a settled fleet condenses identically everywhere: {counts:?}"
    );
    assert_converged(fleet);
}

proptest! {
    #![proptest_config(ProptestConfig::with_cases(64))]

    /// Condense and the dotted hygiene compose with the epoch lifecycle
    /// under any tape and schedule: fleet-applied condense claims between
    /// edit bursts never diverge a read, the settled fleet is
    /// byte-identical after every claim, and the boundary seals and
    /// converges regardless of what the claims excised.
    #[test]
    fn any_schedule_with_condense_and_hygiene_converges(
        seed in any::<u64>(),
        first_tape in prop::collection::vec((0usize..3, 0usize..4, 0usize..8, 0usize..8), 1..12),
        second_tape in prop::collection::vec((0usize..3, 0usize..4, 0usize..8, 0usize..8), 1..10),
        third_tape in prop::collection::vec((0usize..3, 0usize..4, 0usize..8, 0usize..8), 1..8),
        declarer in 0usize..3,
    ) {
        let mut fleet = fleet_of(&ROSTER, NonZeroUsize::new(2).expect("positive"));
        let mut fabric = Fabric::new(seed, &ROSTER, 7);

        for &op in &first_tape {
            play_hygiene(&mut fabric, &mut fleet, op);
        }
        condense_point(&mut fabric, &mut fleet);

        for &op in &second_tape {
            play_hygiene(&mut fabric, &mut fleet, op);
        }
        condense_point(&mut fabric, &mut fleet);

        // The boundary composes with everything the claims excised: the
        // declarer's rebuild is the base plus the covered log, insensitive
        // to what the live pairs condensed away.
        cross_boundary(&mut fabric, &mut fleet, &[ROSTER[declarer]], &[], 2);
        assert_converged(&fleet);

        // The next generation's own round runs the same alphabet over the
        // re-founded plane and its fresh retirement tracker.
        for &op in &third_tape {
            play_hygiene(&mut fabric, &mut fleet, op);
        }
        condense_point(&mut fabric, &mut fleet);
        for replica in fleet.values() {
            prop_assert_eq!(replica.seals.len(), 1);
        }
    }
}

/// Applies one op from the disciplined alphabet: DISCIPLINED inserts,
/// deletes, moves, the dotted orphan retirement, and condense on local
/// evidence at whatever moment the tape says, with no settled points and
/// no fleet-applied claim. This is the coordination-free license the
/// writer discipline buys (the S224 arm): the cut gate delivers every
/// pre-oath anchoring weave before an acknowledgement counts, the oath
/// rebinds every post-oath weave, and the testimony anchor pins ride the
/// acknowledgements themselves, so a replica may excise whenever its own
/// gathered meet covers a dot, mid-traffic, on local evidence alone.
fn play_disciplined(
    fabric: &mut Fabric,
    fleet: &mut BTreeMap<u32, Replica>,
    op: Op,
    mints: &mut Vec<Dot>,
) {
    let (_, kind, at, to) = op;
    let station = station_of(op);
    match kind % 5 {
        0 => {
            let minted = act(fabric, fleet, station, |replica, out| {
                replica.insert_disciplined(at, out)
            });
            mints.push(minted);
        }
        1 => {
            let _ = act(fabric, fleet, station, |replica, out| {
                replica.delete_visible(at, out)
            });
        }
        2 => {
            let _ = act(fabric, fleet, station, |replica, out| {
                replica.move_visible(at, Some(to), out)
            });
        }
        3 => {
            let _ = act(fabric, fleet, station, |replica, out| {
                replica.retire_orphans(out)
            });
        }
        _ => {
            let _ = fleet.get_mut(&station).expect("roster member").condense();
        }
    }
    let steps = fabric.schedule().below(4);
    for _ in 0..steps {
        let _ = fabric.step(fleet);
    }
}

proptest! {
    #![proptest_config(ProptestConfig::with_cases(64))]

    /// The coordination-free condense law: under the writer discipline,
    /// per-replica condense on local evidence composes with any tape and
    /// any adversarial schedule. No weave ever strands (every delivered
    /// locus is reachable once the tape drains), and one final
    /// equal-witness condense brings the fleet back to byte identity,
    /// however different the mid-tape excision states were. This is the
    /// R-41 counterweight's liveness cost bought back: the fleet-applied
    /// apply round is sufficient but not necessary once writers are
    /// disciplined.
    #[test]
    fn a_disciplined_fleet_condenses_on_local_evidence_and_converges(
        seed in any::<u64>(),
        tape in prop::collection::vec((0usize..3, 0usize..5, 0usize..8, 0usize..8), 1..24),
    ) {
        let mut fleet = fleet_of(&ROSTER, NonZeroUsize::new(2).expect("positive"));
        let mut fabric = Fabric::new(seed, &ROSTER, 7);
        let mut mints = Vec::new();

        for &op in &tape {
            play_disciplined(&mut fabric, &mut fleet, op, &mut mints);
        }
        fabric.drain(&mut fleet);

        // The no-strand law: a stranded weave is a delivered locus whose
        // anchor was excised before arrival, present but unreachable
        // forever. Under the discipline none exists at any replica.
        for replica in fleet.values() {
            let store = replica.text().store();
            for &dot in &mints {
                if store.locus(dot).is_some() {
                    prop_assert!(
                        store.is_reachable(dot),
                        "a disciplined weave never strands: {dot:?} \
                         is woven but unreachable at replica {}",
                        replica.id()
                    );
                }
            }
        }

        // One equal-witness condense apiece and the fleet is
        // byte-identical again, mid-tape excision histories notwithstanding.
        for replica in fleet.values_mut() {
            let _ = replica.condense();
        }
        assert_converged(&fleet);
    }
}

/// One crash point: rebuild the station from its journal, run the repair
/// lane, and let the schedule interleave a few deliveries on top.
fn crash_point(fabric: &mut Fabric, fleet: &mut BTreeMap<u32, Replica>, station: u32) {
    crash(
        fleet,
        &ROSTER,
        NonZeroUsize::new(2).expect("positive"),
        station,
    );
    resync(fabric, fleet, station);
    let steps = fabric.schedule().below(6);
    for _ in 0..steps {
        let _ = fabric.step(fleet);
    }
}

proptest! {
    #![proptest_config(ProptestConfig::with_cases(64))]

    /// The kill-restart law: under any op tape, any adversarial schedule
    /// (reordering and duplication throughout), and any crash schedule,
    /// replicas that follow the rehydration recipe (fence before emit,
    /// journal the mints, checkpoint at the seal, resync on restart)
    /// rejoin without un-saying anything: the boundary seals exactly
    /// once and the fleet converges to one record, one byte sequence,
    /// one seal projection. Crashes land anywhere in the pre-window tape
    /// and once inside the open window, so adopted replicas re-earn the
    /// witness and laggards rejoin behind it.
    #[test]
    fn any_schedule_with_crashes_rehydrates_and_converges(
        seed in any::<u64>(),
        tape in prop::collection::vec((0usize..3, 0usize..3, 0usize..8, 0usize..8), 1..12),
        crashes in prop::collection::vec((0usize..3, 0usize..12), 1..4),
        window in prop::collection::vec((0usize..3, 0usize..3, 0usize..8, 0usize..8), 0..4),
        window_crash in 0usize..3,
        declarer in 0usize..3,
    ) {
        let mut fleet = fleet_of(&ROSTER, NonZeroUsize::new(2).expect("positive"));
        let mut fabric = Fabric::new(seed, &ROSTER, 7);

        for (index, &op) in tape.iter().enumerate() {
            for &(station, position) in &crashes {
                if position == index {
                    crash_point(&mut fabric, &mut fleet, ROSTER[station]);
                }
            }
            play(&mut fabric, &mut fleet, op);
        }
        fabric.drain(&mut fleet);
        let _ = act(&mut fabric, &mut fleet, ROSTER[declarer], |replica, out| {
            replica.try_declare(out)
        })
        .expect("a settled watermark self-supports the declaration");
        for &op in &window {
            play_window(&mut fabric, &mut fleet, op);
        }
        crash_point(&mut fabric, &mut fleet, ROSTER[window_crash]);
        fabric.drain(&mut fleet);
        assert_converged(&fleet);
        for replica in fleet.values() {
            prop_assert_eq!(replica.generation(), 2);
            prop_assert_eq!(replica.seals.len(), 1);
        }
    }
}