minerva 0.2.0

//! Departure: the family narrows, and nothing else moves (S336).
//!
//! Ruling R-83 left liveness under a member that will not report to
//! membership or to nothing. `Stability::abandon` takes the first half of
//! that: it narrows the family the *watermark* meets over. The second half
//! --- narrowing the epoch rounds --- is measured here as a refusal rather
//! than shipped, because doing it *on a locally computed bound* seals a
//! record the consignment door then rejects, one door after the lineage has
//! advanced.
//!
//! So the suite has two halves: what the tracker's doors do, and why the
//! rounds deliberately do not follow them. Both still hold exactly as
//! written. S339 did not weaken the refusal; it built the object the
//! refusal named as its price, and `tests/departure.rs` is that suite ---
//! every fixture here stays the control it compares against.
//!
//! The end-to-end companion is
//! `fleet::byzantine::abandoning_the_withholder_releases_the_meet_and_cures_this_freeze`;
//! the argument behind both is `docs/metis-membership-departure.adoc`.

extern crate alloc;

use core::num::NonZeroUsize;

use crate::kairos::Kairos;
use crate::metis::{
    AbandonRefusal, Cut, Dot, EpochLedgerRehydrateError, EpochRefusal, Epochs, SealedEpoch,
    Stability, VersionVector, Vouched,
};

/// Builds an identity literal. Panics on the non-dot counter zero (R-91).
#[track_caller]
fn d(station: u32, counter: u64) -> Dot {
    Dot::from_parts(station, counter).expect("test literal names the non-dot counter zero")
}

const ROSTER: [u32; 3] = [1, 2, 3];
const SURVIVORS: [u32; 2] = [1, 2];
/// The member that departs.
const GONE: u32 = 3;
/// The declaring station's dot, one above the base cut's ceiling for it.
const DECLARATION: Dot = match Dot::from_parts(1, 2) {
    Ok(dot) => dot,
    Err(_) => unreachable!(),
};

fn horizon() -> NonZeroUsize {
    NonZeroUsize::new(2).expect("positive")
}

fn vector(pairs: &[(u32, u64)]) -> VersionVector {
    let mut vector = VersionVector::new();
    for &(station, counter) in pairs {
        vector.observe(station, counter);
    }
    vector
}

fn base() -> Cut {
    Cut::from_witnessed(vector(&[(1, 1), (2, 1), (3, 1)]))
}

/// The cut the survivors reach once the declaration is delivered.
fn delivered() -> Cut {
    Cut::from_witnessed(vector(&[(1, 2), (2, 1), (3, 1)]))
}

fn rank() -> Kairos {
    Kairos::new(2, 0, 1, 0u16)
}

/// A settled tracker over `roster`: every member has vouched for the base.
fn settled(roster: &[u32]) -> Stability {
    let mut stability = Stability::new(roster.iter().copied());
    for &station in roster {
        stability.report_cut(station, &base()).expect("on roster");
    }
    stability
}

// -- The tracker's own doors ------------------------------------------------

/// A silent member pins the watermark at its last report; abandoning it is
/// what releases it.
///
/// The freeze at its source. Crash, partition, and malice reach the same
/// state, which is why the cure is membership rather than fault detection.
#[test]
fn a_silent_member_pins_the_watermark_until_it_is_abandoned() {
    let mut stability = settled(&ROSTER);
    for &station in &SURVIVORS {
        stability
            .report_cut(station, &delivered())
            .expect("on roster");
    }

    assert_eq!(
        stability.watermark(),
        base().as_vector().clone(),
        "the meet is pinned at the silent member's last report"
    );

    let abandoned = stability.abandon(GONE).expect("a surviving family remains");
    assert_eq!(abandoned.station(), GONE);
    assert_eq!(
        stability.watermark(),
        delivered().as_vector().clone(),
        "the meet now ranges over the surviving family alone"
    );
    assert!(stability.is_abandoned(GONE));
    assert_eq!(
        stability.abandoned().collect::<alloc::vec::Vec<_>>(),
        [GONE]
    );
}

/// The bound reads what the *survivors* kept, never the departing member's
/// own claim: only what somebody else can still serve survives a
/// departure.
#[test]
fn the_abandonment_bound_reads_the_survivors_and_not_the_departing_claim() {
    let mut stability = Stability::new(ROSTER);
    // The departing member vouches for a great deal of its own minting...
    stability
        .report_cut(GONE, &Cut::from_witnessed(vector(&[(GONE, 9)])))
        .expect("on roster");
    // ...but only this much of it reached anyone else, and the survivors
    // hold plenty of their own minting besides, which the bound must not
    // sweep in: it answers about one coordinate.
    stability
        .report_cut(1, &Cut::from_witnessed(vector(&[(1, 8), (GONE, 4)])))
        .expect("on roster");
    stability
        .report_cut(2, &Cut::from_witnessed(vector(&[(2, 6), (GONE, 2)])))
        .expect("on roster");

    let abandoned = stability.abandon(GONE).expect("a surviving family remains");
    assert_eq!(
        abandoned.counter(),
        4,
        "the bound is the survivors' join at the departing coordinate"
    );
    assert_eq!(
        abandoned.bound(),
        &vector(&[(GONE, 4)]),
        "restricted to that one coordinate and nothing else"
    );
    assert_eq!(
        stability.abandonment_bound(GONE),
        vector(&[(GONE, 4)]),
        "the read answers the same before and after the decision"
    );
}

/// The bound is readable in advance and only rises: it prices the eviction
/// before the decision.
#[test]
fn the_abandonment_bound_is_readable_in_advance_and_only_rises() {
    let mut stability = Stability::new(ROSTER);
    assert_eq!(
        stability.abandonment_bound(GONE),
        VersionVector::new(),
        "nothing vouched for yet: the whole of a departure would be loss"
    );
    stability
        .report_cut(1, &Cut::from_witnessed(vector(&[(GONE, 3)])))
        .expect("on roster");
    assert_eq!(stability.abandonment_bound(GONE), vector(&[(GONE, 3)]));
    stability
        .report_cut(2, &Cut::from_witnessed(vector(&[(GONE, 5)])))
        .expect("on roster");
    assert_eq!(
        stability.abandonment_bound(GONE),
        vector(&[(GONE, 5)]),
        "a second survivor's report can only raise what survives"
    );
    // A reported cut ranges over whatever dots its sender delivered, so a
    // survivor's vector can carry an off-roster coordinate. Pricing a
    // departure that cannot happen would be a fabrication.
    stability
        .report_cut(1, &Cut::from_witnessed(vector(&[(GONE, 3), (404, 7)])))
        .expect("on roster");
    assert_eq!(
        stability.abandonment_bound(404),
        VersionVector::new(),
        "a station off the roster is priced at bottom even when survivors hold its dots"
    );
}

/// A fleet does not abandon its way to zero, nor abandon what was never a
/// member: an empty family meets to the lattice top `watermark` refuses.
#[test]
fn abandonment_refuses_to_empty_the_family_or_to_invent_one() {
    let mut stability = Stability::new(SURVIVORS);
    assert_eq!(
        stability.abandon(404),
        Err(AbandonRefusal::UnknownStation { station: 404 }),
        "abandoning a non-member would install the belief that it was one"
    );

    let _ = stability.abandon(2).expect("one survivor remains");
    assert_eq!(
        stability.abandon(1),
        Err(AbandonRefusal::LastSurvivor { station: 1 }),
        "the last survivor cannot leave"
    );
    assert!(
        !stability.is_abandoned(1),
        "the refusal left the tracker unchanged"
    );
    assert_eq!(
        stability.watermark(),
        VersionVector::new(),
        "a family of one still answers that one member's cut"
    );
}

/// Abandoning is idempotent and the watermark is monotone across it.
///
/// Removing a floor can only raise the meet, which is why there is no
/// inverse. The absence is pinned here rather than merely documented.
#[test]
fn abandonment_is_idempotent_and_raises_the_watermark_monotonically() {
    let mut stability = settled(&ROSTER);
    for &station in &SURVIVORS {
        stability
            .report_cut(station, &delivered())
            .expect("on roster");
    }
    let before = stability.watermark();
    let first = stability.abandon(GONE).expect("a surviving family remains");
    let once = stability.watermark();
    let second = stability.abandon(GONE).expect("idempotent");
    let twice = stability.watermark();

    assert_eq!(first, second, "the witness is the same both times");
    assert_eq!(once, twice, "the second call changes nothing");
    assert!(
        before <= once,
        "abandoning removes a floor, so the meet can only rise"
    );
    assert_eq!(
        stability.abandoned().collect::<alloc::vec::Vec<_>>(),
        [GONE],
        "and the departed member is named once"
    );
}

/// An abandoned member that speaks again is **inert but visible**: the fold
/// lands in a slot nothing reads, and the premise no one can certify is
/// caught being false at the one read that can catch it.
#[test]
fn an_abandoned_member_that_speaks_again_is_inert_but_visible() {
    let mut stability = settled(&ROSTER);
    for &station in &SURVIVORS {
        stability
            .report_cut(station, &delivered())
            .expect("on roster");
    }
    let _ = stability.abandon(GONE).expect("a surviving family remains");
    let raised = stability.watermark();
    assert!(
        stability.resurgent().next().is_none(),
        "nothing has falsified the premise yet"
    );

    stability
        .report_cut(
            GONE,
            &Cut::from_witnessed(vector(&[(1, 2), (2, 1), (3, 7)])),
        )
        .expect("the door stays total for an abandoned member");

    assert_eq!(
        stability.watermark(),
        raised,
        "the report is inert: the meet no longer ranges over that member"
    );
    assert_eq!(
        stability.resurgent().collect::<alloc::vec::Vec<_>>(),
        [GONE],
        "but the premise's falsifier is visible"
    );
    assert!(
        stability.is_abandoned(GONE),
        "and speaking again does not un-abandon: the meet has no inverse"
    );
}

/// What `resurgent` cannot see, pinned in both directions.
///
/// The read observes one thing --- a stability report absorbed after local
/// abandonment --- and the two ways that falls short of a verdict are worth
/// pinning, because the temptation to read it as one is strong.
/// *False positive*: a report already in flight when the operator decided
/// arrives afterwards and is indistinguishable from a live member.
/// *False negative*: an abandoned member's declaration is still accepted by
/// the epoch machine, which never touches this tracker, so the loudest
/// possible evidence of life leaves no mark here.
#[test]
fn resurgence_over_reports_neither_proves_life_nor_catches_all_of_it() {
    let mut stability = settled(&ROSTER);
    // Minted before the decision; delivered after it.
    let in_flight = Cut::from_witnessed(vector(&[(1, 1), (2, 1), (3, 1)]));
    let _ = stability.abandon(GONE).expect("a surviving family remains");
    stability.report_cut(GONE, &in_flight).expect("total");
    assert_eq!(
        stability.resurgent().collect::<alloc::vec::Vec<_>>(),
        [GONE],
        "a report in flight at the decision is a false positive by construction"
    );

    let mut quiet = settled(&ROSTER);
    let mut epochs = Epochs::new(ROSTER, horizon());
    let _ = quiet.abandon(GONE).expect("a surviving family remains");
    let mut rival = Epochs::new(ROSTER, horizon());
    let declaration = rival
        .declare(d(GONE, 2), rank(), &quiet, &Cut::bottom())
        .expect("licensed");
    epochs
        .deliver(declaration, &quiet, &Cut::bottom())
        .expect("an abandoned member's declaration is still a lawful candidate");
    assert_eq!(epochs.candidates().count(), 1);
    assert!(
        quiet.resurgent().next().is_none(),
        "and it leaves no mark: the tracker never sees that channel"
    );
}

/// The *witness* narrows with the family too.
///
/// A bare charter `report` withdraws `watermark_cut`'s gap-freedom brand for
/// good, and the epoch rounds read only that branded form --- so a witness
/// tracked as one global flag would leave a fleet frozen after evicting the
/// very member whose bare report withdrew it. The closure theorem asks only
/// about the slots the meet ranges over, so the flag is per station.
#[test]
fn abandoning_the_unwitnessed_reporter_restores_the_branded_meet() {
    let mut stability = Stability::new(ROSTER);
    for &station in &SURVIVORS {
        stability
            .report_cut(station, &delivered())
            .expect("on roster");
    }
    stability
        .report(GONE, base().as_vector())
        .expect("the bare charter door");
    assert!(
        stability.watermark_cut().is_none(),
        "a mixed family's meet cannot be branded"
    );

    let _ = stability.abandon(GONE).expect("a surviving family remains");
    assert_eq!(
        stability.watermark_cut().map(|cut| cut.as_vector().clone()),
        Some(delivered().as_vector().clone()),
        "with the unwitnessed slot outside the family, the meet is a cut by closure again"
    );
}

/// Departure is configuration, not machine state: a rebuilt tracker starts
/// with a whole family again.
///
/// The consumer duty this states is the seal-time one --- the epoch recipe
/// builds a fresh tracker per generation, so a departure not re-declared
/// re-freezes the next generation on the member the last one evicted. The
/// tracker cannot carry it for the caller: the family refines a roster the
/// caller declares, and neither is machine state.
#[test]
fn a_rebuilt_tracker_starts_with_a_whole_family_again() {
    let mut stability = settled(&ROSTER);
    for &station in &SURVIVORS {
        stability
            .report_cut(station, &delivered())
            .expect("on roster");
    }
    let _ = stability.abandon(GONE).expect("a surviving family remains");
    assert_eq!(stability.watermark(), delivered().as_vector().clone());

    let mut next = Stability::new(ROSTER);
    assert!(
        !next.is_abandoned(GONE),
        "the new tracker knows nothing of the departure"
    );
    for &station in &SURVIVORS {
        next.report_cut(station, &delivered()).expect("on roster");
    }
    assert_eq!(
        next.watermark(),
        VersionVector::new(),
        "so it freezes again on the member the last generation evicted"
    );

    let _ = next.abandon(GONE).expect("re-declared");
    assert_eq!(
        next.watermark(),
        delivered().as_vector().clone(),
        "and only re-declaring cures it"
    );
}

// -- Where departure stops ---------------------------------------------------

/// Drives one honest round over `family` to whatever it seals, with no
/// departure anywhere: the control the negative results below compare to.
fn honest_round(family: &[u32]) -> Option<SealedEpoch> {
    let mut stability = settled(family);
    let mut epochs = Epochs::new(family.iter().copied(), horizon());
    let declaration = epochs
        .declare(DECLARATION, rank(), &stability, &Cut::bottom())
        .expect("the settled watermark licenses the declaration");
    let epoch = declaration.address();
    for &station in family {
        stability
            .report_cut(station, &delivered())
            .expect("on roster");
        epochs
            .confirm(epoch, &Vouched::trust(station, delivered()))
            .expect("a confirmation for the delivered declaration");
    }
    let _ = epochs
        .adopt(1, delivered().as_vector().get(1), &stability)
        .expect("the confirmation round is complete under the watermark");
    for &station in family {
        if station == 1 {
            continue;
        }
        epochs
            .adopt_report(
                epoch,
                &Vouched::trust(station, delivered().as_vector().get(station)),
            )
            .expect("an adoption report for a delivered candidate");
    }
    epochs.try_seal(&stability).cloned()
}

/// **The refusal.** Abandoning cures the watermark and deliberately leaves
/// the epoch rounds frozen.
///
/// The rounds could be narrowed --- it is three lines --- and doing so
/// completes them and seals a record whose join carries no coordinate for
/// the departed member, while the survivors' shadows still hold deltas that
/// member minted above the declared cut. The consignment door then refuses,
/// one door *after* the lineage has advanced, which is the durable wedge
/// S334 measured: a recoverable stall traded for an unrecoverable halt.
///
/// So the stall stays, and this pins that it stays *recoverably*: nothing is
/// retired, the window is intact, and a departure that carries an agreed and
/// enforced bound completes it --- which S339 built, so the completion is now
/// itself pinned, one fixture over
/// (`departure::an_attested_departure_completes_the_rounds_a_bare_one_freezes`).
/// This test is that one's control: the same window, the same silence, and
/// the only difference is whether the departure was attested.
#[test]
fn abandoning_cures_the_watermark_and_leaves_the_rounds_frozen() {
    let mut stability = settled(&ROSTER);
    let mut epochs = Epochs::new(ROSTER, horizon());
    let declaration = epochs
        .declare(DECLARATION, rank(), &stability, &Cut::bottom())
        .expect("licensed");
    let epoch = declaration.address();
    for &station in &SURVIVORS {
        stability
            .report_cut(station, &delivered())
            .expect("on roster");
        epochs
            .confirm(epoch, &Vouched::trust(station, delivered()))
            .expect("confirm");
    }
    assert!(stability.watermark() < delivered().as_vector().clone());

    let _ = stability.abandon(GONE).expect("a surviving family remains");
    assert_eq!(
        stability.watermark(),
        delivered().as_vector().clone(),
        "the meet ranges over the surviving family at once"
    );
    assert_eq!(
        epochs.adopt(1, 2, &stability),
        Err(EpochRefusal::Unconfirmed),
        "the confirmation round still wants every roster member"
    );
    assert!(epochs.try_seal(&stability).is_none());
    assert!(
        epochs.sealed().next().is_none() && epochs.candidates().count() == 1,
        "and the stall stays recoverable: nothing retired, the window intact"
    );
}

/// The join-neutral substitute is **refused at the confirmation door**.
///
/// R-35 framed membership as substitute testimony; the confirmation one
/// cannot be supplied. Bottom is the only join-neutral report and the
/// coverage check that shipped in S218 refuses it. Not an obstacle to work
/// around --- it is one of the reasons a departure cannot be expressed as
/// testimony on the departed member's behalf.
#[test]
fn a_join_neutral_substitute_is_refused_at_the_confirmation_door() {
    let stability = settled(&ROSTER);
    let mut epochs = Epochs::new(ROSTER, horizon());
    let declaration = epochs
        .declare(DECLARATION, rank(), &stability, &Cut::bottom())
        .expect("licensed");
    let epoch = declaration.address();

    assert_eq!(
        epochs.confirm(epoch, &Vouched::trust(GONE, Cut::bottom())),
        Err(EpochRefusal::ConfirmationDoesNotCover {
            epoch,
            station: GONE,
            covered: 0,
        }),
        "the only join-neutral report is refused by the coverage check"
    );
}

/// A substitute large enough to pass that check **pollutes the seal
/// record**: its adoption counter enters the sealed join at the departed
/// member's coordinate, so a certificate binds over a record naming a
/// member that is gone.
#[test]
fn a_substitute_large_enough_to_pass_pollutes_the_seal_record() {
    let substituted = honest_round(&ROSTER).expect("seals");
    let surviving = honest_round(&SURVIVORS).expect("seals");

    assert_ne!(
        substituted, surviving,
        "testimony supplied for the departed member leaves it in the record"
    );
    assert_eq!(
        substituted.sealed_join().get(GONE),
        delivered().as_vector().get(GONE),
        "at exactly its own coordinate"
    );
    assert_eq!(
        surviving.sealed_join().get(GONE),
        0,
        "which the surviving family alone would never have written"
    );
}

/// No shipped door carries a lineage across a roster change: the second
/// reason the family narrows rather than the roster. The sealed joins
/// already held name the departed member, whose history stays lawful after
/// it stops being a member.
#[test]
fn no_shipped_door_carries_a_lineage_across_a_roster_change() {
    let mut stability = settled(&ROSTER);
    let mut epochs = Epochs::new(ROSTER, horizon());
    let declaration = epochs
        .declare(DECLARATION, rank(), &stability, &Cut::bottom())
        .expect("licensed");
    let epoch = declaration.address();
    for &station in &ROSTER {
        stability
            .report_cut(station, &delivered())
            .expect("on roster");
        epochs
            .confirm(epoch, &Vouched::trust(station, delivered()))
            .expect("confirm");
    }
    let _ = epochs
        .adopt(1, delivered().as_vector().get(1), &stability)
        .expect("adopt");
    for &station in &[2, GONE] {
        epochs
            .adopt_report(
                epoch,
                &Vouched::trust(station, delivered().as_vector().get(station)),
            )
            .expect("peer adoption");
    }
    let sealed = epochs.try_seal(&stability).cloned().expect("seals");
    assert!(
        sealed.sealed_join().get(GONE) > 0,
        "the lineage's support names the departed member"
    );

    assert_eq!(
        Epochs::rehydrate(&epochs.snapshot(), SURVIVORS, horizon()).err(),
        Some(EpochLedgerRehydrateError::RosterMismatch),
        "so the checkpoint door refuses to hand the lineage to a smaller roster"
    );
}

/// Departure does not move the candidate set: a departed member's delivered
/// declaration stays a lawful candidate.
///
/// A requirement, not a courtesy --- candidacy must stay a function of what
/// was delivered, or replicas abandoning at different moments would compute
/// different antichains. It is also the reason the eventual departure
/// mechanism cannot simply withdraw a departed member's traffic.
#[test]
fn departure_does_not_move_the_candidate_set() {
    let mut stability = settled(&ROSTER);
    let mut epochs = Epochs::new(ROSTER, horizon());
    let _ = epochs
        .declare(DECLARATION, rank(), &stability, &Cut::bottom())
        .expect("licensed");

    let mut rival = Epochs::new(ROSTER, horizon());
    let rival_declaration = rival
        .declare(
            d(GONE, 2),
            Kairos::new(9, 0, 1, 0u16),
            &stability,
            &Cut::bottom(),
        )
        .expect("licensed at the same settled watermark");
    epochs
        .deliver(rival_declaration.clone(), &stability, &Cut::bottom())
        .expect("a concurrent candidate");

    let _ = stability.abandon(GONE).expect("a surviving family remains");
    assert_eq!(
        epochs.candidates().count(),
        2,
        "abandoning the minter does not withdraw its candidacy"
    );
    assert!(
        epochs
            .candidates()
            .any(|candidate| candidate.address() == rival_declaration.address()),
        "and the departed member's declaration is still one of them"
    );
}