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
129
130
131
132
133
134
135
136
137
138
139

extern crate alloc;

use super::super::support::arb_vv;
use crate::metis::{Cut, Dot, DotSet, Stability, VersionVector};
use alloc::vec::Vec;
use proptest::prelude::*;

/// A report stream over the four-member roster `0..4`, small enough that members
/// overlap and the watermark moves.
fn arb_reports() -> impl Strategy<Value = Vec<(u32, VersionVector)>> {
    prop::collection::vec((0u32..4, arb_vv()), 0..12)
}

/// A receipt stream, so a witnessed cut can be built via `floor_of`.
fn arb_dots() -> impl Strategy<Value = Vec<(u32, u64)>> {
    prop::collection::vec((0u32..4, 0u64..8), 0..16)
}

/// Builds a witnessed cut from a receipt stream (the honest `floor_of` door).
fn cut_of(dots: &[(u32, u64)]) -> Cut {
    let mut have = DotSet::new();
    for &pair in dots {
        // The stream can name the non-dot counter zero; it never entered
        // the receipt set before and cannot be an identity now (R-91).
        if let Ok(dot) = Dot::try_from(pair) {
            let _ = have.insert(dot);
        }
    }
    Cut::floor_of(&have)
}

proptest! {
    /// The watermark is a lower bound of every member's vouched cut, and exactly
    /// their n-ary meet.
    #[test]
    fn prop_watermark_is_the_meet_of_vouched_cuts(reports in arb_reports()) {
        let mut tracker = Stability::new(0..4);
        // Re-derive each member's vouched cut independently of the tracker.
        let mut vouched: Vec<VersionVector> =
            (0..4).map(|_| VersionVector::new()).collect();
        for (station, cut) in &reports {
            tracker.report(*station, cut).unwrap();
            let slot = &mut vouched[usize::try_from(*station).unwrap()];
            *slot = slot.merge(cut);
        }
        let watermark = tracker.watermark();
        for cut in &vouched {
            prop_assert!(&watermark <= cut);
        }
        let expected = vouched[1..]
            .iter()
            .fold(vouched[0].clone(), |acc, cut| acc.meet(cut));
        prop_assert_eq!(watermark, expected);
    }

    /// Reporting is join-monotone: the watermark never regresses, whatever the
    /// order or staleness of the reports. The license for acting on it.
    #[test]
    fn prop_watermark_never_regresses(reports in arb_reports()) {
        let mut tracker = Stability::new(0..4);
        let mut last = tracker.watermark();
        for (station, cut) in &reports {
            tracker.report(*station, cut).unwrap();
            let next = tracker.watermark();
            prop_assert!(last <= next);
            last = next;
        }
    }

    /// `report_cut` and `report` agree on the watermark wherever both apply:
    /// feeding the same vectors through the witnessed and the bare door leaves
    /// the tracker computing the identical n-ary meet. The witness is the only
    /// difference.
    #[test]
    fn prop_report_cut_agrees_with_report(streams in prop::collection::vec((0u32..4, arb_dots()), 0..12)) {
        let mut witnessed = Stability::new(0..4);
        let mut bare = Stability::new(0..4);
        for (station, dots) in &streams {
            let cut = cut_of(dots);
            witnessed.report_cut(*station, &cut).unwrap();
            bare.report(*station, cut.as_vector()).unwrap();
        }
        // Same slots, same watermark; only the witnessed gate differs.
        prop_assert_eq!(witnessed.watermark(), bare.watermark());
        // The all-witnessed tracker hands back the watermark as a cut, equal in
        // value to the bare read.
        let branded = witnessed.watermark_cut().expect("all reports witnessed");
        prop_assert_eq!(branded.as_vector(), &witnessed.watermark());
        // The bare-door tracker withdraws its witness the moment any report lands.
        if !streams.is_empty() {
            prop_assert!(bare.watermark_cut().is_none());
        }
    }

    /// `watermark_cut` is `Some` exactly when every report was witnessed, and its
    /// value is always the bare `watermark`. A single bare `report` withdraws it.
    #[test]
    fn prop_watermark_cut_tracks_the_witnessed_gate(
        streams in prop::collection::vec((0u32..4, arb_dots(), any::<bool>()), 0..12),
    ) {
        let mut tracker = Stability::new(0..4);
        let mut all_witnessed = true;
        for (station, dots, witnessed) in &streams {
            let cut = cut_of(dots);
            if *witnessed {
                tracker.report_cut(*station, &cut).unwrap();
            } else {
                tracker.report(*station, cut.as_vector()).unwrap();
                all_witnessed = false;
            }
        }
        match tracker.watermark_cut() {
            Some(branded) => {
                prop_assert!(all_witnessed);
                prop_assert_eq!(branded.as_vector(), &tracker.watermark());
            }
            None => prop_assert!(!all_witnessed),
        }
    }

    /// The tracker is a fold of joins: report order does not matter, so gossip
    /// reordering cannot change what is considered stable.
    #[test]
    fn prop_reports_are_order_invariant(
        (reports, permuted) in arb_reports()
            .prop_flat_map(|r| (Just(r.clone()), Just(r).prop_shuffle())),
    ) {
        let mut a = Stability::new(0..4);
        for (station, cut) in &reports {
            a.report(*station, cut).unwrap();
        }
        let mut b = Stability::new(0..4);
        for (station, cut) in &permuted {
            b.report(*station, cut).unwrap();
        }
        prop_assert_eq!(&a, &b);
        prop_assert_eq!(a.watermark(), b.watermark());
    }
}