minerva 0.2.0

Causal ordering for distributed systems
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128

extern crate alloc;

use super::super::support::vv;
use crate::metis::{Cut, Dot, DotSet, Stability, UnknownStation, VersionVector};

/// 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")
}

/// A witnessed cut over the honest `floor_of` door, built from an explicit dot
/// list so the test controls the vector exactly.
fn cut(dots: &[(u32, u64)]) -> Cut {
    let mut have = DotSet::new();
    // Insert `1..=count` per station so the floor is exactly `count`.
    for &(station, count) in dots {
        for dot in 1..=count {
            let _ = have.insert(d(station, dot));
        }
    }
    Cut::floor_of(&have)
}

#[test]
fn test_watermark_is_common_knowledge() {
    let mut tracker = Stability::new([1, 2, 3]);
    tracker.report(1, &vv(&[(1, 3), (2, 1)])).unwrap();
    tracker.report(2, &vv(&[(1, 2), (2, 4), (3, 1)])).unwrap();
    tracker.report(3, &vv(&[(1, 5), (2, 2)])).unwrap();
    // Pointwise min across the roster; station 3 (vouched by one member only)
    // drops out entirely.
    assert_eq!(tracker.watermark(), vv(&[(1, 2), (2, 1)]));
}

#[test]
fn test_watermark_requires_every_report() {
    let mut tracker = Stability::new([1, 2]);
    tracker.report(1, &vv(&[(1, 9), (2, 9)])).unwrap();
    // Member 2 has never reported: it vouches for nothing, so nothing is stable.
    assert_eq!(tracker.watermark(), VersionVector::new());
}

#[test]
fn test_unknown_station_is_refused_unchanged() {
    let mut tracker = Stability::new([1, 2]);
    tracker.report(1, &vv(&[(1, 1)])).unwrap();
    let before = tracker.clone();
    let refused = tracker.report(7, &vv(&[(1, 5)]));
    assert_eq!(refused, Err(UnknownStation { station: 7 }));
    assert_eq!(tracker, before);
}

#[test]
fn test_stale_report_is_absorbed() {
    let mut tracker = Stability::new([1, 2]);
    tracker.report(1, &vv(&[(1, 4)])).unwrap();
    tracker.report(2, &vv(&[(1, 4)])).unwrap();
    let high = tracker.watermark();
    assert_eq!(high, vv(&[(1, 4)]));
    // A late, older report is join-absorbed: the watermark never regresses.
    tracker.report(1, &vv(&[(1, 2)])).unwrap();
    assert_eq!(tracker.watermark(), high);
}

#[test]
fn test_empty_roster_watermark_is_bottom() {
    // The empty family's mathematical meet is the lattice top ("everything is
    // stable", vacuously): the maximally unsafe fabrication for a safe-to-forget
    // quantity, so the degenerate answer is bottom.
    let tracker = Stability::new(core::iter::empty::<u32>());
    assert_eq!(tracker.watermark(), VersionVector::new());
}

#[test]
fn test_report_cut_and_report_agree_on_the_watermark() {
    // The witnessed door and the bare door fold the same vectors, so the
    // watermark is identical; only watermark_cut's honesty differs.
    let mut witnessed = Stability::new([1, 2]);
    witnessed.report_cut(1, &cut(&[(1, 3), (2, 1)])).unwrap();
    witnessed.report_cut(2, &cut(&[(1, 2), (2, 4)])).unwrap();

    let mut bare = Stability::new([1, 2]);
    bare.report(1, &vv(&[(1, 3), (2, 1)])).unwrap();
    bare.report(2, &vv(&[(1, 2), (2, 4)])).unwrap();

    assert_eq!(witnessed.watermark(), bare.watermark());
    assert_eq!(witnessed.watermark(), vv(&[(1, 2), (2, 1)]));
}

#[test]
fn test_watermark_cut_is_witnessed_only_when_every_report_was() {
    let mut tracker = Stability::new([1, 2]);
    tracker.report_cut(1, &cut(&[(1, 4), (2, 2)])).unwrap();
    tracker.report_cut(2, &cut(&[(1, 3), (2, 5)])).unwrap();
    // Every report witnessed: the meet is a cut by closure, handed back branded.
    let watermark = tracker.watermark_cut().expect("all reports witnessed");
    assert_eq!(watermark.as_vector(), &vv(&[(1, 3), (2, 2)]));
    assert_eq!(watermark.as_vector(), &tracker.watermark());

    // One bare report withdraws the witness for good, even a later witnessed one.
    tracker.report(1, &vv(&[(1, 9)])).unwrap();
    assert!(tracker.watermark_cut().is_none());
    tracker.report_cut(2, &cut(&[(2, 9)])).unwrap();
    assert!(tracker.watermark_cut().is_none());
}

#[test]
fn test_watermark_cut_fresh_and_empty_are_bottom() {
    // A fresh tracker is witnessed (every member at bottom, a genuine cut).
    let tracker = Stability::new([1, 2]);
    assert_eq!(tracker.watermark_cut(), Some(Cut::bottom()));
    // The empty roster is witnessed and bottom, never the vacuous top.
    let empty = Stability::new(core::iter::empty::<u32>());
    assert_eq!(empty.watermark_cut(), Some(Cut::bottom()));
}

#[test]
fn test_reports_reader_exposes_the_straggler() {
    let mut tracker = Stability::new([2, 1]);
    tracker.report(1, &vv(&[(1, 6)])).unwrap();
    // Ascending station order, unreported members at bottom: the member pinning
    // the watermark is visible.
    let slots: alloc::vec::Vec<_> = tracker.reports().collect();
    assert_eq!(slots.len(), 2);
    assert_eq!(slots[0], (1, &vv(&[(1, 6)])));
    assert_eq!(slots[1], (2, &VersionVector::new()));
}