minerva 0.2.0

//! Crafted collation cost and occupancy-boundary cases.

use super::*;
use crate::metis::dot::RawDot;

/// Wide sibling buckets must charge child-plane shifts to the replay budget.
#[test]
fn test_a_wide_bucket_move_storm_stays_exact() {
    let clk = clock(1);
    let mut text = Rhapsody::new();
    let anchor_dot = d(3, 1);
    assert!(text.weave(
        anchor_dot,
        Locus {
            anchor: Anchor::Origin,
            rank: clk.now(0u16),
        }
    ));
    for index in 1..=96u64 {
        assert!(text.weave(
            d(1, index),
            Locus {
                anchor: Anchor::After(anchor_dot.into()),
                rank: clk.now(0u16),
            }
        ));
    }
    let mut view = text.recension(&Metatheses::new());
    let mut record = Metatheses::new();
    let rebuilds_before = view.collation_rebuilds();
    let minted: Vec<(Dot, Metathesis)> = (0..24u64)
        .map(|step| {
            let target = (1, (step % 96) + 1);
            let anchor = if step % 2 == 0 {
                Anchor::Origin
            } else {
                Anchor::After(anchor_dot.into())
            };
            (
                d(1, 1000 + step),
                Metathesis {
                    target: target.into(),
                    to: locus_at(&clk, anchor),
                },
            )
        })
        .collect();
    for (witness, testimony) in minted.into_iter().rev() {
        assert!(record.insert(witness, testimony));
        view.collate(&text, &record);
        view.check_collation(&text.recension(&record));
    }
    assert!(
        view.collation_rebuilds() > rebuilds_before,
        "the width-charged budget trips the eager-parity cap under the storm",
    );
}

/// Descending ranks exercise the eager-parity cap's worst replay order.
#[test]
fn test_a_descending_rank_storm_stays_exact() {
    let clk = clock(1);
    let mut text = Rhapsody::new();
    let dots = typed_chain(&mut text, &clk, 1, 1, 24);
    let minted: Vec<(Dot, Metathesis)> = (0..16u64)
        .map(|i| {
            let target = dots[usize::try_from(i).unwrap() % 8 + 8].into();
            let anchor = Anchor::After(dots[usize::try_from(i).unwrap() % 4].into());
            (
                d(1, 200 + i),
                Metathesis {
                    target,
                    to: locus_at(&clk, anchor),
                },
            )
        })
        .collect();
    let mut view = text.recension(&Metatheses::new());
    let mut record = Metatheses::new();
    for (witness, testimony) in minted.into_iter().rev() {
        assert!(record.insert(witness, testimony));
        view.collate(&text, &record);
        view.check_collation(&text.recension(&record));
    }
}

/// A visibility flip above dot 64 exercises nonzero occupancy pages.
#[test]
fn test_a_high_page_flip_collates_exactly() {
    let clk = clock(1);
    let mut text = Text::new(1);
    let mut caret = None;
    let mut dots = Vec::new();
    for _ in 0..100 {
        let dot = type_after(&mut text, &clk, caret);
        caret = Some(dot.into());
        dots.push(dot);
    }
    let moves = Metatheses::new();
    let mut view = text.state().store().recension(&moves);
    let mut gone = DotSet::new();
    let _ = gone.insert(dots[69]);
    let _ = text.retract(gone);
    view.collate(text.state().store(), &moves);
    view.check_collation(&text.state().store().recension(&moves));
    assert!(!view.is_visible(dots[69]), "the high-page flip landed");
    assert!(view.is_visible(dots[70]), "its neighbors did not move");
}

/// A bulk retirement exercises whole-page removal in one collation.
#[test]
fn test_a_bulk_retirement_collates_exactly() {
    let clk = clock(1);
    let mut text = Text::new(1);
    let mut caret = None;
    let mut dots = Vec::new();
    for _ in 0..512 {
        let dot = type_after(&mut text, &clk, caret);
        caret = Some(dot.into());
        dots.push(dot);
    }
    let moves = Metatheses::new();
    let mut view = text.state().store().recension(&moves);
    view.collate(text.state().store(), &moves);
    let mut gone = DotSet::new();
    for &dot in &dots[128..384] {
        let _ = gone.insert(dot);
    }
    let _ = text.retract(gone);
    view.collate(text.state().store(), &moves);
    view.check_collation(&text.state().store().recension(&moves));
    assert_eq!(view.order_len(), 256, "the middle half retired");
}

/// Fragment-grade re-placement keeps a suffix storm below the rebuild cap.
#[test]
fn test_a_suffix_move_storm_stays_fragment_grade() {
    let clk = clock(1);
    let mut text = Rhapsody::new();
    let dots = typed_chain(&mut text, &clk, 1, 1, 128);
    let minted: Vec<(Dot, Metathesis)> = (0..6u64)
        .map(|i| {
            (
                d(1, 500 + i),
                Metathesis {
                    target: dots[8].into(),
                    to: locus_at(
                        &clk,
                        Anchor::After(dots[usize::try_from(i).unwrap()].into()),
                    ),
                },
            )
        })
        .collect();
    let mut view = text.recension(&Metatheses::new());
    let mut record = Metatheses::new();
    for (witness, testimony) in minted.into_iter().rev() {
        assert!(record.insert(witness, testimony));
        view.collate(&text, &record);
        view.check_collation(&text.recension(&record));
    }
    assert!(view.collation_replays() >= 6, "every arrival replays");
    assert_eq!(
        view.collation_rebuilds(),
        0,
        "fragment-grade work keeps the whole storm under the parity cap"
    );
}

/// A mass retraction of parked testimonies leaves the pending vector in
/// one ordered pass, not one `Vec::remove` per witness. The law is
/// unchanged agreement with a fresh eager replay, so the batching cannot
/// change semantics; the rounds pin what the pass must get right. The
/// parked witnesses sit in two blocks with a RUN of verdict-earning
/// witnesses between them, so the middle round makes the pass skip
/// several non-pending departures in a row and drop an alternating
/// selection, and the last round retracts everything at once.
#[test]
fn test_a_mass_pending_retraction_stays_exact() {
    let clk = clock(1);
    let mut text = Rhapsody::new();
    let dots = typed_chain(&mut text, &clk, 1, 1, 6);
    let block = 100u64;
    // Testimonies naming dots this replica has not woven: every one
    // parks, so none of them earns a verdict. Two blocks, either side of
    // the verdict-earning run below.
    let low: Vec<Dot> = (0..block).map(|step| d(1, 1000 + step)).collect();
    let high: Vec<Dot> = (0..block).map(|step| d(1, 1200 + step)).collect();
    let mut record = Metatheses::new();
    for (step, &witness) in low.iter().chain(high.iter()).enumerate() {
        assert!(record.insert(
            witness,
            Metathesis {
                target: RawDot::new(2, u64::try_from(step).unwrap() + 1),
                to: locus_at(&clk, Anchor::After(dots[0].into())),
            }
        ));
    }
    // Testimonies that DO earn verdicts, in a contiguous run BETWEEN the
    // two parked blocks: retracting them together forces the ordered
    // pass past several non-pending departures before it meets the next
    // pending one.
    let earning: Vec<Dot> = (0..4u64).map(|step| d(1, 1100 + step)).collect();
    for (step, &witness) in earning.iter().enumerate() {
        assert!(record.insert(
            witness,
            Metathesis {
                target: dots[step + 2].into(),
                to: locus_at(&clk, Anchor::After(dots[0].into())),
            }
        ));
    }
    let mut view = text.recension(&Metatheses::new());
    view.collate(&text, &record);
    view.check_collation(&text.recension(&record));
    assert_eq!(
        view.pending().len(),
        usize::try_from(2 * block).unwrap(),
        "every outrun move parks"
    );
    assert_ne!(
        view.order(),
        text.order(),
        "the earning moves moved something"
    );

    // One collation retracts the whole high block, the whole earning
    // run, and every odd witness of the low block: the pass skips a run
    // of non-pending departures and drops an alternating selection, and
    // the survivors must come out in order.
    let mut thinned = Metatheses::new();
    let survivors: Vec<Dot> = low.iter().copied().step_by(2).collect();
    for &witness in &survivors {
        assert!(thinned.insert(
            witness,
            *record.get(witness).expect("the record carries it")
        ));
    }
    let touched = view.collation_compacted();
    view.collate(&text, &thinned);
    view.check_collation(&text.recension(&thinned));
    assert_eq!(
        view.pending(),
        survivors.as_slice(),
        "the survivors keep the sorted order"
    );
    assert_eq!(
        view.order(),
        text.order(),
        "the earning moves retracted with the rest"
    );
    // The bound, not merely the result: 150 of the 200 parked witnesses
    // retract here (every odd one of the low block, and the whole high
    // block), the first of them sitting at index 1, so the compaction
    // walks the 199-slot tail from that index exactly once. The count is
    // pinned exactly, not bounded: a pass starting from the head, or a
    // second pass, or a splice per departure all read differently here,
    // and nothing else the view surfaces can tell them apart.
    assert_eq!(
        view.collation_compacted() - touched,
        2 * usize::try_from(block).unwrap() - 1,
        "one compaction over the tail from the first parked departure"
    );

    // The survivors retract together: the parked list empties in one
    // pass and the view still equals its eager twin.
    let undone = Metatheses::new();
    view.collate(&text, &undone);
    view.check_collation(&text.recension(&undone));
    assert!(
        view.pending().is_empty(),
        "the mass retraction empties the parked list"
    );
}

/// The sparse corner of the same seam: a collation whose departures are
/// all UNPARKED must not pay for the parked backlog it is not touching.
/// The classification stays a probe per departure, so the pending vector
/// is read and never walked, and a large backlog costs the retraction
/// nothing.
#[test]
fn test_an_unparked_departure_never_walks_the_parked_backlog() {
    let clk = clock(1);
    let mut text = Rhapsody::new();
    let dots = typed_chain(&mut text, &clk, 1, 1, 4);
    let mut record = Metatheses::new();
    // A large parked backlog: testimonies naming dots this replica has
    // not woven.
    for step in 0..512u64 {
        assert!(record.insert(
            d(1, 1000 + step),
            Metathesis {
                target: RawDot::new(2, step + 1),
                to: locus_at(&clk, Anchor::After(dots[0].into())),
            }
        ));
    }
    // One testimony that earns a verdict, and nothing else changes.
    let earning = d(1, 900);
    assert!(record.insert(
        earning,
        Metathesis {
            target: dots[3].into(),
            to: locus_at(&clk, Anchor::After(dots[0].into())),
        }
    ));
    let mut view = text.recension(&Metatheses::new());
    view.collate(&text, &record);
    view.check_collation(&text.recension(&record));
    assert_eq!(view.pending().len(), 512, "the backlog parks");

    // Retract the earning testimony alone: the backlog stands, so the
    // collation has no parked departure to compact and must leave the
    // vector alone.
    let touched = view.collation_compacted();
    let mut kept = Metatheses::new();
    for step in 0..512u64 {
        assert!(
            kept.insert(
                d(1, 1000 + step),
                *record
                    .get(d(1, 1000 + step))
                    .expect("the record carries it")
            )
        );
    }
    view.collate(&text, &kept);
    view.check_collation(&text.recension(&kept));
    assert_eq!(view.pending().len(), 512, "the backlog is untouched");
    assert_eq!(
        view.collation_compacted(),
        touched,
        "an unparked departure never walks the parked vector"
    );
    assert_eq!(view.order(), text.order(), "the earning move retracted");
}