minerva 0.2.0

//! Crash rehydration, journal repair, and bounded-record scenarios.

extern crate alloc;

use alloc::collections::BTreeMap;
use alloc::vec::Vec;

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

/// The rehydration recipe end to end, on the honest path: a replica that
/// crashes mid-window rebuilds by replaying its journal through the
/// ordinary inbound door, re-earns its adoption witness by re-running
/// `adopt` over the rebuilt record (the witness is deliberately never
/// persisted), takes the repair lane for whatever the crash lost from
/// its volatile buffer, and the fleet seals and converges as if nobody
/// had died. A second crash straddling the seal pins the checkpoint
/// half: the consigned base, the sealed lineage, and the allocator
/// basis reopen exactly on the consigned plane.
#[test]
fn a_mid_window_crash_rehydrates_and_the_fleet_seals() {
    let depth = horizon(2);
    let mut fleet = fleet_of(&ROSTER, depth);
    let mut fabric = Fabric::new(0xC4A5_84E5, &ROSTER, 8);

    // Seed traffic from every station, settle, declare.
    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, out| {
        replica.try_declare(out)
    })
    .expect("a settled watermark self-supports the declaration");

    // Step until the declarer adopts: the window is open and natively
    // live at station 1 while the others may still lag.
    let mut guard = 0;
    while !fleet[&1].adopted() {
        assert!(fabric.step(&mut fleet), "the fabric must keep moving");
        guard += 1;
        assert!(guard < 10_000, "adoption must arrive under a live fabric");
    }

    // Window traffic on both sides of the boundary, partially delivered.
    let _ = act(&mut fabric, &mut fleet, 1, |replica, out| {
        replica.insert_visible(1, out)
    });
    let _ = act(&mut fabric, &mut fleet, 3, |replica, out| {
        replica.insert_visible(0, out)
    });
    for _ in 0..4 {
        let _ = fabric.step(&mut fleet);
    }

    // The mid-window crash: the adopted replica dies and rehydrates from
    // its journal alone.
    crash(&mut fleet, &ROSTER, depth, 1);
    assert!(
        fleet[&1].adopted(),
        "adoption re-derives from the journal: the witness is re-earned by \
         re-running adopt over the rebuilt record, never restored from disk"
    );

    // The repair lane re-serves what the volatile buffer lost, and the
    // lifecycle completes everywhere.
    resync(&mut fabric, &fleet, 1);
    fabric.drain(&mut fleet);
    for replica in fleet.values() {
        assert_eq!(
            replica.generation(),
            2,
            "replica {} seals the window it crashed inside",
            replica.id()
        );
    }
    assert_converged(&fleet);

    // The seal-straddling crash: kill the freshly sealed replica before
    // it says anything in the new generation. The checkpoint carries the
    // consigned base pair, the sealed lineage, and the allocator basis,
    // so the restart opens exactly on the consigned plane.
    crash(&mut fleet, &ROSTER, depth, 1);
    resync(&mut fabric, &fleet, 1);
    fabric.drain(&mut fleet);
    assert_converged(&fleet);

    // And the new plane is live: fresh edits from the restarted replica
    // converge like anyone's.
    let _ = act(&mut fabric, &mut fleet, 1, |replica, out| {
        replica.insert_visible(0, out)
    });
    fabric.drain(&mut fleet);
    assert_converged(&fleet);
}

/// The persisted-but-unsent gap (the review-caught P1): a crash can land
/// between the fence and the transport post, leaving a durably minted
/// note that no peer ever received and that no retransmission will ever
/// carry, because the outbox died with the process. Left unrepaired, the
/// fleet wedges: the minter's floor covers a dot no peer can ever
/// deliver, so its own adoption cut would enter a sealed join whose
/// window no peer's shadow can complete. The repair lane is therefore
/// bidirectional: after a crash the journal, not the dead outbox, is the
/// source of record for what the replica has said, and the restarted
/// replica re-serves its own durable logs and round positions to every
/// peer.
#[test]
fn a_persisted_but_unsent_mint_reaches_the_fleet_through_repair() {
    let depth = horizon(2);
    let mut fleet = fleet_of(&ROSTER, depth);
    let mut fabric = Fabric::new(0x0DD5_E47D, &ROSTER, 0);

    // Settle shared content first.
    let _ = act(&mut fabric, &mut fleet, 1, |replica, out| {
        replica.insert_visible(0, out)
    });
    fabric.drain(&mut fleet);

    // Station 2 mints; the fence runs inside the emission; the process
    // dies before the outbox reaches the fabric. The journal holds a
    // note the network never saw.
    let unsent = {
        let replica = fleet.get_mut(&2).expect("roster member");
        let mut dead_outbox = Vec::new();
        replica.insert_visible(1, &mut dead_outbox)
        // The crash eats `dead_outbox`: nothing was posted.
    };
    crash(&mut fleet, &ROSTER, depth, 2);

    // The restarted replica still holds its own words (the mint duty
    // fenced them in the same stroke that would have emitted them)...
    assert!(
        fleet[&2].text().store().locus(unsent).is_some(),
        "the journal is the source of record for the unsent mint"
    );
    // ...and no peer does, yet.
    for station in [1, 3] {
        assert!(
            fleet[&station].text().store().locus(unsent).is_none(),
            "the fabric never carried the unsent mint to station {station}"
        );
    }

    // The bidirectional repair lane re-serves it from the journal side.
    resync(&mut fabric, &fleet, 2);
    fabric.drain(&mut fleet);
    assert_converged(&fleet);

    // And the boundary composes on top: the once-unsent dot rides the
    // watermark, the window opens, seals, and carries it across.
    let _ = act(&mut fabric, &mut fleet, 1, |replica, out| {
        replica.try_declare(out)
    })
    .expect("the repaired watermark self-supports the declaration");
    fabric.drain(&mut fleet);
    for replica in fleet.values() {
        assert_eq!(replica.generation(), 2, "the repaired fleet seals");
    }
    assert_converged(&fleet);
}

/// Asserts every replica's record sits at exactly `unique` data notes:
/// one log entry and one journaled data note per unique event, nothing
/// data-shaped unfenced (the bounded-record read).
fn assert_record_exact(fleet: &BTreeMap<u32, Replica>, unique: usize, context: &str) {
    for replica in fleet.values() {
        assert_eq!(
            replica.log_len(),
            unique,
            "{context}: replica {} log never compounds",
            replica.id()
        );
        assert_eq!(
            replica.journal_data_notes(),
            unique,
            "{context}: replica {} journal data never compounds",
            replica.id()
        );
        assert_eq!(
            replica.volatile_data_notes(),
            0,
            "{context}: replica {} fences every data note by drain",
            replica.id()
        );
    }
}

/// The bounded-record pin (the review-caught P2): redelivery, fabric
/// duplication, and repeated crash-repair cycles must not compound the
/// record. A pure duplicate folds to nothing, so the log door absorbs it
/// without appending and the journal never fences it; across alternating
/// restarts the log and the journal's data notes stay exactly one entry
/// per unique event.
#[test]
fn alternating_restarts_keep_the_record_bounded() {
    let depth = horizon(2);
    let mut fleet = fleet_of(&ROSTER, depth);
    // Aggressive duplication: roughly one delivery in three re-enqueues.
    let mut fabric = Fabric::new(0xB07D_ED42, &ROSTER, 3);

    let mut mints = 0usize;
    for round in 0..3 {
        for station in ROSTER {
            let _ = act(&mut fabric, &mut fleet, station, |replica, out| {
                replica.insert_visible(round, out)
            });
            mints += 1;
        }
    }
    let deleted = act(&mut fabric, &mut fleet, 1, |replica, out| {
        replica.delete_visible(0, out)
    });
    assert!(deleted.is_some(), "the plane has an element to delete");
    mints += 1;
    fabric.drain(&mut fleet);

    // Duplicated delivery never compounded anything: one log entry and
    // one journaled data note per unique event, at every replica.
    assert_record_exact(&fleet, mints, "post-tape");

    // Alternating crash-restart-repair cycles with no new traffic: the
    // record stays exact however many times the repair lane re-serves
    // the same logs and round positions in both directions. Fencing is
    // effect-based, so once the first cycle has repaired the pre-crash
    // volatile tail, a repair round adds *nothing*: total journal
    // lengths (protocol restatements included) are frozen from the
    // first post-crash settle onward, and no data note ever sits
    // unfenced at a drained fleet (protocol tails may, the bounded loss
    // the repair lane re-states).
    let mut frozen: Option<BTreeMap<u32, usize>> = None;
    for cycle in 0..4usize {
        let station = ROSTER[cycle % ROSTER.len()];
        crash(&mut fleet, &ROSTER, depth, station);
        resync(&mut fabric, &fleet, station);
        fabric.drain(&mut fleet);
        assert_record_exact(&fleet, mints, "crash cycle");
        let totals: BTreeMap<u32, usize> = fleet
            .iter()
            .map(|(&id, replica)| (id, replica.journal_notes_len()))
            .collect();
        if let Some(frozen) = &frozen {
            assert_eq!(
                frozen, &totals,
                "cycle {cycle}: a no-effect repair round fences nothing at all"
            );
        }
        frozen = Some(totals);
    }
    assert_converged(&fleet);

    // The window composes with the bound: declare, let duplicated native
    // traffic cross the pre-adoption gate at the laggards (the gate holds
    // one copy per event), seal, and the fresh generation's record is
    // exactly its unique native mints.
    let _ = act(&mut fabric, &mut fleet, 1, |replica, out| {
        replica.try_declare(out)
    })
    .expect("a settled watermark self-supports the declaration");
    // The first adopter fleet-wide cannot seal in the stroke that adopts
    // it (the seal still needs the others' adoption reports), so it is
    // always observable between steps; a fixed station is not, since a
    // late last confirmation can carry a replica from unadopted through
    // adoption to seal inside one delivery.
    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 native_mints = 2usize;
    let _ = act(&mut fabric, &mut fleet, adopter, |replica, out| {
        replica.insert_visible(0, out)
    });
    let _ = act(&mut fabric, &mut fleet, adopter, |replica, out| {
        replica.insert_visible(1, out)
    });
    fabric.drain(&mut fleet);
    for replica in fleet.values() {
        assert_eq!(replica.generation(), 2, "the window sealed everywhere");
    }
    assert_record_exact(&fleet, native_mints, "post-seal");
    assert_converged(&fleet);
}

/// The stamp-monotonicity pin (the review-caught round-five gap): every
/// stamp a replica speaks after a restart must strictly exceed every
/// stamp it spoke before. Two spoken stamps live in no plane a restart
/// could walk: a native delete's stamp rides only its note, and a
/// sealed window's stamps are re-based away by the re-foundation (the
/// consignment re-mints ranks under the chain law). The journal covers
/// both: rehydration observes every journaled note's carried stamp, and
/// the checkpoint carries a clock watermark minted at the seal, which
/// dominates everything the sealed window spoke.
#[test]
fn a_restart_never_undercuts_a_spoken_stamp() {
    let depth = horizon(2);
    let mut fleet = fleet_of(&ROSTER, depth);
    let mut fabric = Fabric::new(0x57A4_9CA5, &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, out| {
        replica.try_declare(out)
    })
    .expect("a settled watermark self-supports the declaration");
    let mut adopter = None;
    while adopter.is_none() {
        assert!(fabric.step(&mut fleet), "the rounds must complete");
        adopter = ROSTER.iter().copied().find(|id| fleet[id].adopted());
    }
    let adopter = adopter.expect("just found");

    let stamp_of = |outbox: &[Note]| {
        outbox
            .iter()
            .find_map(|note| match note {
                Note::Native { stamp, .. } => Some(*stamp),
                _ => None,
            })
            .expect("a native op emits a stamped note")
    };

    // A run of native inserts, then a longer run of native deletes. The
    // deletes are the corner: each speaks a stamp (the clock ticks) while
    // leaving no walkable locus (the marker is born dead), so the highest
    // spoken stamps live only on the notes and a plane walk alone cannot
    // re-derive them, however long it is.
    let mut insert_stamp = None;
    for _ in 0..12 {
        let mut outbox = Vec::new();
        let _ = fleet
            .get_mut(&adopter)
            .expect("roster member")
            .insert_visible(0, &mut outbox);
        insert_stamp = Some(stamp_of(&outbox));
        fabric.post(adopter, outbox);
    }
    let insert_stamp = insert_stamp.expect("the native inserts ran");
    let mut delete_stamp = insert_stamp;
    for _ in 0..12 {
        let mut outbox = Vec::new();
        let _ = fleet
            .get_mut(&adopter)
            .expect("roster member")
            .delete_visible(0, &mut outbox);
        delete_stamp = stamp_of(&outbox);
        fabric.post(adopter, outbox);
    }
    assert!(delete_stamp > insert_stamp, "the clock is monotone live");

    // The repair lane is a pure read: re-served native notes carry
    // their original stamps, never fresh mints. A freshly minted repair
    // stamp would be spoken but unfenced, and a crash right after would
    // let the next mint re-speak it (the round-six review finding).
    let mut served = Vec::new();
    fleet[&adopter].reserve_logs(&mut served);
    for note in &served {
        if let Note::Native { stamp, .. } = note {
            assert!(
                *stamp <= delete_stamp,
                "a re-served stamp is an already-spoken stamp"
            );
        }
    }

    // Mid-window crash: the first post-restart stamp strictly exceeds
    // the plane-less delete stamp, because rehydration observed the
    // journaled note's carried stamp.
    crash(&mut fleet, &ROSTER, depth, adopter);
    let mut outbox = Vec::new();
    let _ = fleet
        .get_mut(&adopter)
        .expect("roster member")
        .insert_visible(0, &mut outbox);
    let post_restart = stamp_of(&outbox);
    fabric.post(adopter, outbox);
    assert!(
        post_restart > delete_stamp,
        "the journaled stamps re-derive the clock past the note-only stamp"
    );

    // Seal, crash again: the sealed window's stamps survive only in the
    // checkpoint's clock watermark, and the next spoken rank still
    // strictly exceeds them all.
    resync(&mut fabric, &fleet, adopter);
    fabric.drain(&mut fleet);
    for replica in fleet.values() {
        assert_eq!(replica.generation(), 2, "the window sealed");
    }
    crash(&mut fleet, &ROSTER, depth, adopter);
    let dot = {
        let mut outbox = Vec::new();
        let dot = fleet
            .get_mut(&adopter)
            .expect("roster member")
            .insert_visible(0, &mut outbox);
        fabric.post(adopter, outbox);
        dot
    };
    let fresh = fleet[&adopter]
        .text()
        .store()
        .locus(dot)
        .expect("just woven")
        .rank;
    assert!(
        fresh > post_restart && fresh > delete_stamp,
        "the checkpoint watermark dominates the re-based window's stamps"
    );
    resync(&mut fabric, &fleet, adopter);
    fabric.drain(&mut fleet);
    assert_converged(&fleet);
}