minerva 0.2.0

//! Concurrent declaration races and lifecycle ordering.

use crate::metis::Vouched;
use alloc::collections::VecDeque;
use alloc::vec::Vec;

use super::*;

/// The concurrent pair: `a` from station 1, `b` from station 2 with the
/// higher stamp, so `b` is the fixed rule's winner everywhere.
fn concurrent_declarations() -> (Declaration, Declaration) {
    // Declarations are minted through the licensed door against per-station
    // trackers whose watermark is each declarer's local view of stability.
    let mut one = tracker();
    for station in ROSTER {
        one.report_cut(station, &cut(&[(1, 4), (2, 0), (3, 2)]))
            .unwrap();
    }
    let a = machine()
        .declare(d(1, 5), Kairos::new(5, 0, 1, 0u16), &one, &Cut::bottom())
        .expect("station 1's watermark licenses a declaration");

    let mut two = tracker();
    for station in ROSTER {
        two.report_cut(station, &cut(&[(1, 0), (2, 4), (3, 1)]))
            .unwrap();
    }
    let b = machine()
        .declare(d(2, 5), Kairos::new(6, 0, 2, 0u16), &two, &Cut::bottom())
        .expect("station 2's watermark licenses a declaration");

    (a, b)
}

/// One race replica with honest cut reports and retried confirmations.
struct Fabric {
    replica: u32,
    epochs: Epochs,
    stability: Stability,
    delivered_a: bool,
    delivered_b: bool,
    pending: VecDeque<Event>,
}

#[derive(Clone, Debug)]
enum Event {
    DeliverA,
    DeliverB,
    Confirm {
        station: u32,
        epoch: EpochAddress,
        delivered: Cut,
    },
    /// Advances the other members to the terminal race cut.
    SyncOthers,
}

impl Fabric {
    fn new(replica: u32) -> Self {
        Self {
            replica,
            epochs: machine(),
            stability: tracker(),
            delivered_a: false,
            delivered_b: false,
            pending: VecDeque::new(),
        }
    }

    /// Applies one event, then retries confirmations that outran their declarations.
    fn apply(&mut self, event: &Event, a: &Declaration, b: &Declaration) {
        match event {
            Event::DeliverA => {
                self.epochs
                    .deliver(a.clone(), &self.stability, &Cut::bottom())
                    .expect("an honest interleaving never refuses a candidate");
                self.delivered_a = true;
            }
            Event::DeliverB => {
                self.epochs
                    .deliver(b.clone(), &self.stability, &Cut::bottom())
                    .expect("an honest interleaving never refuses a candidate");
                self.delivered_b = true;
            }
            Event::Confirm {
                station,
                epoch,
                delivered,
            } => {
                self.stability.report_cut(*station, delivered).unwrap();
                if self
                    .epochs
                    .confirm(*epoch, &Vouched::trust(*station, delivered.clone()))
                    .is_err()
                {
                    self.pending.push_back(event.clone());
                }
            }
            Event::SyncOthers => {
                for station in ROSTER {
                    if station != self.replica {
                        self.stability.report_cut(station, &top()).unwrap();
                    }
                }
            }
        }
        // Stability reports are emitted only for cuts this replica holds.
        if self.delivered_a && self.delivered_b {
            self.stability.report_cut(self.replica, &top()).unwrap();
        }
        for _ in 0..self.pending.len() {
            let Some(Event::Confirm {
                station,
                epoch,
                delivered,
            }) = self.pending.pop_front()
            else {
                unreachable!("only confirmations park");
            };
            if self
                .epochs
                .confirm(epoch, &Vouched::trust(station, delivered.clone()))
                .is_err()
            {
                self.pending.push_back(Event::Confirm {
                    station,
                    epoch,
                    delivered,
                });
            }
        }
    }
}

fn permutations(events: &[Event]) -> Vec<Vec<Event>> {
    if events.is_empty() {
        return alloc::vec![Vec::new()];
    }
    let mut all = Vec::new();
    for (index, event) in events.iter().enumerate() {
        let mut rest = events.to_vec();
        let _ = rest.remove(index);
        for mut tail in permutations(&rest) {
            tail.insert(0, event.clone());
            all.push(tail);
        }
    }
    all
}

#[test]
fn the_concurrent_declaration_race_fixes_one_winner_everywhere() {
    // Exhaust all 6! lifecycle orders per replica. Winner choice and the
    // sealed lineage must be independent of schedule and replica identity.
    let (a, b) = concurrent_declarations();
    let events = [
        Event::DeliverA,
        Event::DeliverB,
        Event::Confirm {
            station: 1,
            epoch: a.address(),
            delivered: cut(&[(1, 5), (2, 0), (3, 2)]),
        },
        Event::Confirm {
            station: 2,
            epoch: b.address(),
            delivered: cut(&[(1, 0), (2, 5), (3, 1)]),
        },
        Event::Confirm {
            station: 3,
            epoch: b.address(),
            delivered: cut(&[(1, 5), (2, 5), (3, 3)]),
        },
        Event::SyncOthers,
    ];

    let mut sealed_records = Vec::new();
    for order in permutations(&events) {
        for replica in ROSTER {
            let mut fabric = Fabric::new(replica);
            for event in &order {
                // Once fixed, only the deterministic winner may remain fixed.
                let before = fabric.epochs.fixed().map(Declaration::dot);
                assert!(before.is_none() || before == Some(b.dot()));
                fabric.apply(event, &a, &b);
            }
            assert!(fabric.pending.is_empty(), "every report lands");

            let winner = fabric
                .epochs
                .adopt(replica, 9, &fabric.stability)
                .expect("the confirmation watermark has covered the join");
            assert_eq!(winner.dot(), b.dot(), "the fixed rule's winner");
            assert_eq!(winner.rank(), b.rank());
            assert_eq!(fabric.epochs.candidates().count(), 2);
            assert!(fabric.epochs.adopted());

            for station in ROSTER {
                if station != replica {
                    fabric
                        .epochs
                        .adopt_report(b.address(), &Vouched::trust(station, 9))
                        .unwrap();
                }
            }
            let sealed = fabric
                .epochs
                .try_seal(&fabric.stability)
                .expect("all adoptions reported under a covering watermark")
                .clone();
            assert_eq!(sealed.declaration(), b.address());
            sealed_records.push(sealed);
        }
    }
    for sealed in &sealed_records {
        assert_eq!(sealed, &sealed_records[0]);
    }
}

#[test]
fn adoption_waits_for_the_confirmation_watermark() {
    // Confirmation possession alone does not establish a stable decision.
    let (a, b) = concurrent_declarations();
    let mut fabric = Fabric::new(3);
    fabric.apply(&Event::DeliverA, &a, &b);
    fabric.apply(&Event::DeliverB, &a, &b);
    for (station, epoch, delivered) in [
        (1, a.address(), cut(&[(1, 5), (2, 0), (3, 2)])),
        (2, b.address(), cut(&[(1, 0), (2, 5), (3, 1)])),
        (3, b.address(), cut(&[(1, 5), (2, 5), (3, 3)])),
    ] {
        fabric.apply(
            &Event::Confirm {
                station,
                epoch,
                delivered,
            },
            &a,
            &b,
        );
    }
    assert_eq!(
        fabric
            .epochs
            .adopt(3, 9, &fabric.stability)
            .expect_err("the join is not covered yet"),
        EpochRefusal::Unconfirmed
    );
    assert!(fabric.epochs.fixed().is_none());

    fabric.apply(&Event::SyncOthers, &a, &b);
    let winner = fabric
        .epochs
        .adopt(3, 9, &fabric.stability)
        .expect("covered now");
    assert_eq!(winner.dot(), b.dot());
}

#[test]
fn only_winner_adoptions_seal_and_loser_replays_absorb() {
    let (a, b) = concurrent_declarations();
    let mut fabric = Fabric::new(3);
    for event in [
        Event::DeliverA,
        Event::DeliverB,
        Event::Confirm {
            station: 1,
            epoch: a.address(),
            delivered: cut(&[(1, 5), (2, 0), (3, 2)]),
        },
        Event::Confirm {
            station: 2,
            epoch: b.address(),
            delivered: cut(&[(1, 0), (2, 5), (3, 1)]),
        },
        Event::Confirm {
            station: 3,
            epoch: b.address(),
            delivered: cut(&[(1, 5), (2, 5), (3, 3)]),
        },
        Event::SyncOthers,
    ] {
        fabric.apply(&event, &a, &b);
    }
    let adopted = fabric.epochs.adopt(3, 9, &fabric.stability).unwrap();
    assert_eq!(adopted.dot(), b.dot());

    fabric
        .epochs
        .adopt_report(a.address(), &Vouched::trust(1, 9))
        .unwrap();
    fabric
        .epochs
        .adopt_report(b.address(), &Vouched::trust(2, 9))
        .unwrap();
    assert!(
        fabric.epochs.try_seal(&fabric.stability).is_none(),
        "a loser-addressed report cannot complete the winner's round"
    );
    fabric
        .epochs
        .adopt_report(b.address(), &Vouched::trust(1, 9))
        .unwrap();
    let sealed = fabric.epochs.try_seal(&fabric.stability).unwrap();
    assert!(sealed.contains(a.address()));
    assert!(sealed.contains(b.address()));

    fabric
        .epochs
        .deliver(a.clone(), &fabric.stability, &Cut::bottom())
        .unwrap();
    assert_eq!(
        fabric
            .epochs
            .confirm(a.address(), &Vouched::trust(1, top())),
        Ok(())
    );
    assert_eq!(
        fabric
            .epochs
            .adopt_report(a.address(), &Vouched::trust(1, 9)),
        Ok(())
    );
    assert_eq!(fabric.epochs.candidates().count(), 0);
    assert!(
        fabric
            .epochs
            .declare(
                d(3, 10),
                Kairos::new(10, 0, 3, 0u16),
                &fabric.stability,
                &Cut::bottom()
            )
            .is_ok(),
        "a loser replay must not reopen or block the next epoch"
    );
}

#[test]
fn a_causally_later_declaration_is_refused_while_the_window_is_open() {
    let (a, _) = concurrent_declarations();
    let mut epochs = machine();
    let mut stability = tracker();
    for station in ROSTER {
        stability.report_cut(station, &top()).unwrap();
    }
    epochs
        .deliver(a.clone(), &stability, &Cut::bottom())
        .unwrap();

    let refused = epochs
        .declare(
            d(3, 1),
            Kairos::new(9, 0, 3, 0u16),
            &stability,
            &Cut::bottom(),
        )
        .expect_err("the window is open");
    assert_eq!(refused, EpochRefusal::WindowOpen { open: a.address() });
}

#[test]
fn a_peer_declaration_causally_after_the_window_is_never_a_candidate() {
    let mut early_tracker = tracker();
    let early_cut = cut(&[(1, 1), (2, 1), (3, 1)]);
    for station in ROSTER {
        early_tracker.report_cut(station, &early_cut).unwrap();
    }
    let early = machine()
        .declare(
            d(1, 2),
            Kairos::new(2, 0, 1, 0u16),
            &early_tracker,
            &Cut::bottom(),
        )
        .unwrap();

    let mut later_tracker = tracker();
    let later_cut = cut(&[(1, 2), (2, 1), (3, 1)]);
    for station in ROSTER {
        later_tracker.report_cut(station, &later_cut).unwrap();
    }
    let later = machine()
        .declare(
            d(2, 2),
            Kairos::new(9, 0, 2, 0u16),
            &later_tracker,
            &Cut::bottom(),
        )
        .unwrap();

    let mut stability = tracker();
    for station in ROSTER {
        stability.report_cut(station, &top()).unwrap();
    }

    let mut before_fix = machine();
    before_fix
        .deliver(early.clone(), &stability, &Cut::bottom())
        .unwrap();
    assert_eq!(
        before_fix.deliver(later.clone(), &stability, &Cut::bottom()),
        Err(EpochRefusal::WindowOpen {
            open: early.address()
        })
    );

    let mut reverse = machine();
    reverse
        .deliver(later.clone(), &stability, &Cut::bottom())
        .unwrap();
    reverse
        .deliver(early.clone(), &stability, &Cut::bottom())
        .unwrap();
    assert_eq!(
        reverse
            .candidates()
            .map(Declaration::dot)
            .collect::<Vec<_>>(),
        [early.dot()]
    );

    let mut after_fix = machine();
    after_fix
        .deliver(early.clone(), &stability, &Cut::bottom())
        .unwrap();
    for station in ROSTER {
        after_fix
            .confirm(early.address(), &Vouched::trust(station, later_cut.clone()))
            .unwrap();
    }
    let adopted = after_fix.adopt(1, 9, &stability).unwrap();
    assert_eq!(adopted.dot(), early.dot());
    assert_eq!(
        after_fix.deliver(later, &stability, &Cut::bottom()),
        Err(EpochRefusal::WindowOpen {
            open: early.address()
        })
    );
}

#[test]
fn peer_reports_cannot_seal_before_local_adoption() {
    let mut stability = Stability::new([1]);
    stability.report_cut(1, &top()).unwrap();
    let mut epochs = Epochs::new([1], NonZeroUsize::new(1).unwrap());
    let declaration = epochs
        .declare(
            d(1, 10),
            Kairos::new(5, 0, 1, 0u16),
            &stability,
            &Cut::bottom(),
        )
        .unwrap();
    let delivered = cut(&[(1, 10), (2, 9), (3, 9)]);
    stability.report_cut(1, &delivered).unwrap();
    epochs
        .confirm(declaration.address(), &Vouched::trust(1, delivered.clone()))
        .unwrap();
    epochs
        .adopt_report(declaration.address(), &Vouched::trust(1, 10))
        .unwrap();

    assert!(
        epochs.try_seal(&stability).is_none(),
        "peer reports cannot erase the local adoption capability"
    );
    let adopted = epochs.adopt(1, 10, &stability).unwrap();
    assert_eq!(adopted.dot(), declaration.dot());
    assert!(epochs.try_seal(&stability).is_some());
}