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

use crate::kairos::Kairos;
use crate::metis::{Event, Gate, Ideal, ReleaseDrivenGate, VersionVector};
use core::num::NonZeroUsize;

/// A test-only non-causal gate in the participation shape from PRD 0009.
pub(super) struct Quorum;

impl Gate for Quorum {
    type Dep = u64;
    type Progress = VersionVector;

    fn deliverable(progress: &VersionVector, _sender: u32, required: &u64) -> bool {
        u64::try_from(progress.iter().count()).unwrap_or(u64::MAX) >= *required
    }

    fn advance(progress: &mut VersionVector, sender: u32, _required: &u64) {
        progress.observe(sender, 1);
    }

    fn stale(_progress: &VersionVector, _sender: u32, _required: &u64) -> bool {
        false
    }
}

impl ReleaseDrivenGate for Quorum {}

#[test]
fn test_quorum_gate_releases_on_threshold() {
    // The same delivery machine unlocks these by threshold order, not by causal
    // contiguity.
    let mut buf: Ideal<&str, Quorum> = Ideal::default();
    buf.insert(Event {
        stamp: Kairos::new(30, 0, 3, 0u16),
        deps: 2,
        payload: "third",
    });
    buf.insert(Event {
        stamp: Kairos::new(20, 0, 2, 0u16),
        deps: 1,
        payload: "second",
    });
    buf.insert(Event {
        stamp: Kairos::new(10, 0, 1, 0u16),
        deps: 0,
        payload: "first",
    });

    assert_eq!(buf.pop_ready(), Some("first"));
    assert_eq!(buf.pop_ready(), Some("second"));
    assert_eq!(buf.pop_ready(), Some("third"));
    assert_eq!(buf.pop_ready(), None);
}

#[test]
fn test_try_insert_quorum_is_a_hard_cap() {
    // Quorum's `stale` is always false, so a full buffer of live-but-undeliverable
    // events can reclaim nothing.
    let cap = NonZeroUsize::new(1).unwrap();
    let mut buf: Ideal<&str, Quorum> = Ideal::default();
    let waiting = Event {
        stamp: Kairos::new(10, 0, 1, 0u16),
        deps: 2,
        payload: "a",
    };
    assert!(buf.try_insert(waiting, cap).is_ok());
    let before = buf.pending_len();
    let rejected = buf
        .try_insert(
            Event {
                stamp: Kairos::new(20, 0, 2, 0u16),
                deps: 2,
                payload: "b",
            },
            cap,
        )
        .unwrap_err();
    assert_eq!(rejected.event.payload, "b");
    assert_eq!(buf.pending_len(), before);
}