minerva 0.2.0

//! Writer-discipline counterexamples and placement cost.

extern crate alloc;

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

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

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

/// A weave racing a partially applied condense claim strands its anchor.
///
/// The retirement meet, cut-gated, proves every member *applied* the
/// removal; it cannot promise nobody *writes against the tombstone again*,
/// because the visual-insert rule anchors through tombstones by design (a
/// deleted element's followers stay in place). A weave minted after its
/// station's acknowledgement lawfully anchors the tombstone at any replica
/// that has not excised it yet, and arrives dangling at any replica that
/// has: text placed at one member and unplaced at another, the
/// unrecoverable direction. This is `Retired`'s "every replica that will
/// ever write again" clause made executable: the claim is licensed for
/// further writing only once every replica has excised it, so the fleet
/// condenses whole at settled points (standing in for the consumer's
/// apply round), and a lone replica's early condense is lawful only while
/// no new weave mints, which is exactly what the lifecycle scenario
/// exercises.
#[test]
fn a_weave_racing_a_partial_condense_strands_its_anchor() {
    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);

    // A visible element, then a tombstone at the front: station 1 weaves
    // A, station 3 weaves T before it, station 1 deletes T. Everything
    // floods; every acknowledgement is honestly gathered and counted.
    let mut round = Vec::new();
    let _ = one.insert_visible(0, &mut round);
    while !round.is_empty() {
        let mut next = Vec::new();
        for note in &round {
            for replica in [&mut one, &mut two, &mut three] {
                let mut out = Vec::new();
                replica.handle(note, &mut out);
                next.extend(out);
            }
        }
        round = next;
    }
    let mut round = Vec::new();
    let _ = three.insert_visible(0, &mut round);
    while !round.is_empty() {
        let mut next = Vec::new();
        for note in &round {
            for replica in [&mut one, &mut two, &mut three] {
                let mut out = Vec::new();
                replica.handle(note, &mut out);
                next.extend(out);
            }
        }
        round = next;
    }
    let mut round = Vec::new();
    let deleted = one.delete_visible(0, &mut round);
    assert_eq!(deleted, Some(d(3, 1)), "the front element is the tombstone");
    while !round.is_empty() {
        let mut next = Vec::new();
        for note in &round {
            for replica in [&mut one, &mut two, &mut three] {
                let mut out = Vec::new();
                replica.handle(note, &mut out);
                next.extend(out);
            }
        }
        round = next;
    }

    // Replica 2 condenses alone: the witness is complete and honest, the
    // tombstone is sterile there, and the excision is locally lawful.
    assert_eq!(
        two.condense(),
        1,
        "the settled witness licenses the excision"
    );

    // Station 3 now inserts at the front. Its own acknowledgement is long
    // since counted everywhere, yet the fresh weave lawfully anchors the
    // tombstone still present in its skeleton.
    let mut round = Vec::new();
    let minted = three.insert_visible(0, &mut round);
    while !round.is_empty() {
        let mut next = Vec::new();
        for note in &round {
            for replica in [&mut one, &mut two, &mut three] {
                let mut out = Vec::new();
                replica.handle(note, &mut out);
                next.extend(out);
            }
        }
        round = next;
    }

    // The fleet has diverged: the weave placed at the replicas that kept
    // the tombstone and dangles at the one that excised it.
    assert!(
        one.effective_order().contains(&minted),
        "the uncondensed replica places the weave"
    );
    assert!(
        three.effective_order().contains(&minted),
        "the writer places its own weave"
    );
    assert!(
        !two.effective_order().contains(&minted),
        "the condensed replica strands the weave: the claim was not \
         fleet-applied before writing resumed"
    );
}

/// Floods `round` and every note it provokes to all of `replicas`, in
/// rounds, until nothing more is provoked. The hand-rolled twin of
/// `Fabric::drain` for the standalone-replica pins, which need selective
/// routing around it.
fn flood(mut round: Vec<Note>, replicas: &mut [&mut Replica]) {
    while !round.is_empty() {
        let mut next = Vec::new();
        for note in &round {
            for replica in replicas.iter_mut() {
                let mut out = Vec::new();
                replica.handle(note, &mut out);
                next.extend(out);
            }
        }
        round = next;
    }
}

/// The writer discipline unstrands the racing weave: the coordination-free
/// license's positive twin of the strand pin above.
///
/// Same schedule as `a_weave_racing_a_partial_condense_strands_its_anchor`
/// up to the race: A visible, T a settled tombstone, every acknowledgement
/// honestly counted, replica 2 condensed alone. Station 3 now inserts at
/// the front through the DISCIPLINED weave: its published acknowledgement
/// is a quiescence oath (the QSBR shape), so the anchor is chosen in the
/// oath plane, where T is already enacted away, and lands `Before(A)`,
/// which every skeleton holds, condensed or not. The weave places at the
/// early condenser and the uncondensed replicas alike, and the late
/// condenses agree with the early one: under the discipline the cut-gated
/// meet alone licenses excision, and no apply round is owed. The pin
/// kills the discipline-dropped mutant (anchor read off the real plane)
/// at the replica-2 placement assertion.
#[test]
fn the_writer_discipline_unstrands_the_racing_weave() {
    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);

    // A visible element, then a tombstone at the front, exactly as in the
    // strand pin: station 1 weaves A, station 3 weaves T before it,
    // station 1 deletes T; everything floods and every acknowledgement is
    // honestly gathered and counted.
    let mut round = Vec::new();
    let _ = one.insert_visible(0, &mut round);
    flood(round, &mut [&mut one, &mut two, &mut three]);
    let mut round = Vec::new();
    let _ = three.insert_visible(0, &mut round);
    flood(round, &mut [&mut one, &mut two, &mut three]);
    let mut round = Vec::new();
    let deleted = one.delete_visible(0, &mut round);
    assert_eq!(deleted, Some(d(3, 1)), "the front element is the tombstone");
    flood(round, &mut [&mut one, &mut two, &mut three]);

    // Replica 2 condenses alone, exactly as before.
    assert_eq!(
        two.condense(),
        1,
        "the settled witness licenses the excision"
    );

    // Station 3 inserts at the front under the discipline. Its oath plane
    // has already enacted its own acknowledgement of T, so the descent
    // lands where replica 2's own placement rule would: Before(A).
    let mut round = Vec::new();
    let minted = three.insert_disciplined(0, &mut round);
    flood(round, &mut [&mut one, &mut two, &mut three]);

    // No strand anywhere: the weave places at the condensed replica too,
    // and every replica reads the same order.
    for (replica, name) in [
        (&one, "uncondensed"),
        (&two, "condensed"),
        (&three, "writer"),
    ] {
        assert!(
            replica.effective_order().contains(&minted),
            "the {name} replica places the disciplined weave"
        );
    }
    assert_eq!(one.effective_order(), two.effective_order());
    assert_eq!(two.effective_order(), three.effective_order());

    // The late condenses agree with the early one, and the fleet is
    // byte-identical afterwards: local evidence sufficed end to end.
    assert_eq!(one.condense(), 1, "the laggard's own meet licenses it late");
    assert_eq!(three.condense(), 1);
    assert_eq!(one.text().store().to_bytes(), two.text().store().to_bytes());
    assert_eq!(
        two.text().store().to_bytes(),
        three.text().store().to_bytes()
    );
}

/// The discipline binds on the station's own published acknowledgement,
/// never on the gathered meet.
///
/// Quiescence is declared, not observed: the meet lags at exactly the
/// wrong moment. Here station 3 has acknowledged T's removal, but the
/// other members' acknowledgements are still in flight toward it, so
/// 3's gathered meet is bottom while replica 2 holds the full meet and
/// excises. A weave bound by the meet would keep anchoring T and strand
/// at 2; the oath-bound weave rebinds to `Before(A)` and places
/// everywhere. The pin kills the binding-set mutant (oath built from
/// `retirement.retired()` instead of `acked`) at the replica-2 placement
/// assertion.
#[test]
fn the_discipline_binds_on_own_acknowledgement_not_the_meet() {
    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);

    // A and T settle everywhere, as in the strand pin.
    let mut round = Vec::new();
    let _ = one.insert_visible(0, &mut round);
    flood(round, &mut [&mut one, &mut two, &mut three]);
    let mut round = Vec::new();
    let _ = three.insert_visible(0, &mut round);
    flood(round, &mut [&mut one, &mut two, &mut three]);

    // Station 2 deletes T, and nothing reaches station 1 for now: 2 and 3
    // apply and acknowledge between themselves, so 3's own oath is
    // published while its gathered meet still misses member 1 entirely.
    let mut round = Vec::new();
    let deleted = two.delete_visible(0, &mut round);
    assert_eq!(deleted, Some(d(3, 1)), "the front element is the tombstone");
    let held_for_one = round.clone();
    flood(round, &mut [&mut two, &mut three]);

    // Station 3 weaves at the front under the discipline. Its own
    // acknowledgement binds (T rebound away); its meet, still bottom,
    // would not. Hold the weave back while the race completes.
    let mut weave_notes = Vec::new();
    let minted = three.insert_disciplined(0, &mut weave_notes);

    // Bring member 1 in and complete replica 2's meet: 1 applies the
    // removal and its acknowledgement (whose carried floor does NOT hold
    // the in-flight weave) flows to 2 and 3, cut-gated and counted.
    let mut from_one = Vec::new();
    for note in &held_for_one {
        one.handle(note, &mut from_one);
    }
    flood(from_one, &mut [&mut two, &mut three]);

    // Replica 2 holds the full meet and excises before the weave lands.
    assert_eq!(two.condense(), 1, "the full meet licenses the excision");

    // The weave arrives everywhere, after the excision at 2.
    flood(weave_notes, &mut [&mut one, &mut two, &mut three]);
    for (replica, name) in [(&one, "late"), (&two, "condensed"), (&three, "writer")] {
        assert!(
            replica.effective_order().contains(&minted),
            "the {name} replica places the oath-bound weave"
        );
    }
    assert_eq!(one.effective_order(), two.effective_order());
    assert_eq!(two.effective_order(), three.effective_order());
}

/// The discipline prices placement, never convergence: the H2 corner of
/// the excision brief made executable.
///
/// One schedule, two placement regimes. Station 2 deletes the front
/// tombstone T and the removal reaches station 3 but not yet station 1;
/// both then insert at the front concurrently. Under the plain rule both
/// weaves descend to `Before(T)` and run the sibling rank contest in one
/// bucket, where the lower concurrent rank reads frontmost. Under the
/// discipline, station 3 has sworn off T (its oath enacted the
/// acknowledgement) and rebinds to `Before(A)`, LEAVING the tombstone's
/// subtree: the rank contest is forfeited, the rebound weave reads after
/// the whole `Before(T)` region, and the two regimes converge to
/// DIFFERENT orders of the same two elements. Both worlds are internally
/// convergent; the divergence is between worlds, the discipline's
/// placement tax, paid whether or not anyone ever condenses. This is the
/// residual the intake records as unpriced by the literature (the
/// Fugue-grade guarantee is stated for one placement rule, not for a
/// mid-document rebinding of it).
#[test]
fn the_discipline_prices_placement_not_convergence() {
    fn race(disciplined: bool) -> (Dot, Dot, Vec<Dot>) {
        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);

        // A visible, T the front tombstone-to-be, and one tail element to
        // advance station 1's clock past station 3's, so the plain-rule
        // rank contest has a deterministic winner.
        let mut round = Vec::new();
        let _ = one.insert_visible(0, &mut round);
        flood(round, &mut [&mut one, &mut two, &mut three]);
        let mut round = Vec::new();
        let _ = three.insert_visible(0, &mut round);
        flood(round, &mut [&mut one, &mut two, &mut three]);
        let mut round = Vec::new();
        let _ = one.insert_visible(2, &mut round);
        flood(round, &mut [&mut one, &mut two, &mut three]);

        // Station 2 deletes T; the removal reaches 3 (which applies and
        // publishes its oath) but is held from 1.
        let mut round = Vec::new();
        let deleted = two.delete_visible(0, &mut round);
        assert_eq!(deleted, Some(d(3, 1)), "the front element is the tombstone");
        let held_for_one = round.clone();
        flood(round, &mut [&mut two, &mut three]);

        // The concurrent race: both stations insert at the front of their
        // own view, through the same weave door per regime.
        let mut notes_one = Vec::new();
        let mut notes_three = Vec::new();
        let (n1, n3) = if disciplined {
            (
                one.insert_disciplined(0, &mut notes_one),
                three.insert_disciplined(0, &mut notes_three),
            )
        } else {
            (
                one.insert_visible(0, &mut notes_one),
                three.insert_visible(0, &mut notes_three),
            )
        };

        // Everything settles everywhere.
        let mut from_one = Vec::new();
        for note in &held_for_one {
            one.handle(note, &mut from_one);
        }
        flood(from_one, &mut [&mut one, &mut two, &mut three]);
        flood(notes_one, &mut [&mut one, &mut two, &mut three]);
        flood(notes_three, &mut [&mut one, &mut two, &mut three]);

        // Each world converges internally.
        assert_eq!(one.effective_order(), two.effective_order());
        assert_eq!(two.effective_order(), three.effective_order());
        (n1, n3, one.effective_order())
    }

    let (d1, d3, disciplined) = race(true);
    let (p1, p3, plain) = race(false);
    assert_eq!(d1, p1, "the same schedule mints the same dots");
    assert_eq!(d3, p3, "the same schedule mints the same dots");

    let position = |order: &[Dot], dot: Dot| {
        order
            .iter()
            .position(|&placed| placed == dot)
            .expect("placed")
    };

    // Under the plain rule both weaves share the Before(T) bucket, where
    // the lower concurrent rank reads frontmost, so station 3's mint
    // leads; under the discipline station 3's rebound weave has left that
    // bucket and reads after the whole tombstone region, so station 1's
    // mint leads. Same schedule, opposite orders.
    assert!(
        position(&plain, p3) < position(&plain, p1),
        "plain rule: the shared-bucket rank contest reads station 3's \
         mint first: {plain:?}"
    );
    assert!(
        position(&disciplined, d1) < position(&disciplined, d3),
        "discipline: the rebound weave forfeits the contest and reads \
         after the Before(T) region: {disciplined:?}"
    );
}