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

extern crate alloc;

use crate::kairos::{Clock, TickCounter};
use crate::metis::{CausalIdeal, Dot, Event, Producer, VersionVector};
use alloc::vec::Vec;
use proptest::prelude::*;

/// Builds an identity literal. Panics on the non-dot counter zero, so a
/// test that writes `dot(1, 0)` fails loudly instead of silently modelling
/// a state the type forbids (ruling R-91).
#[track_caller]
pub(super) fn dot(station: u32, counter: u64) -> Dot {
    Dot::from_parts(station, counter).expect("test literal names the non-dot counter zero")
}

/// Builds a vector from `(station, count)` pairs via `observe`, so test
/// vectors stay canonical and read declaratively.
pub(super) fn vv(pairs: &[(u32, u64)]) -> VersionVector {
    let mut v = VersionVector::new();
    for &(station, count) in pairs {
        v.observe(station, count);
    }
    v
}

/// Small station domain and counts so generated vectors overlap, making
/// happens-before, equality, and concurrency all frequent.
pub(super) fn arb_vv() -> impl Strategy<Value = VersionVector> {
    prop::collection::vec((0u32..6, 0u64..8), 0..8).prop_map(|pairs| vv(&pairs))
}

/// Each generated event carries its index in the history as its payload, so a
/// released payload identifies the event and lets a test recover its stamp/deps.
pub(super) type Payload = u32;

/// Drains a buffer to exhaustion, returning released payloads in release order.
pub(super) fn drain(buf: &mut CausalIdeal<Payload>) -> Vec<Payload> {
    let mut released = Vec::new();
    while let Some(payload) = buf.pop_ready() {
        released.push(payload);
    }
    released
}

/// The deliverability gate, re-derived independently of the buffer so the
/// properties test the spec rather than the implementation against itself:
/// contiguity for the sender plus causal completeness for every other station.
pub(super) fn deliverable(event: &Event<Payload>, delivered: &VersionVector) -> bool {
    let sender = event.stamp.station_id();
    if event.deps.get(sender) != delivered.get(sender).saturating_add(1) {
        return false;
    }
    event
        .deps
        .iter()
        .all(|(station, count)| station == sender || count <= delivered.get(station))
}

/// Builds a valid causal history by simulating real [`Producer`]s. Each `step` is
/// `(producer, learn_from)`: the producing station first
/// [`observe`](Producer::observe)s every event so far from a station listed in
/// `learn_from` (modelling causal receipt of those peers' broadcasts), then
/// [`produce`](Producer::produce)s one event.
///
/// Because the history is generated through the shipped send step rather than a
/// parallel re-implementation, the buffer suite below is an end-to-end test of
/// [`Producer`] output: a single implementation of the send rule, exercised by both
/// the producer properties and the buffer properties. The result is a real causal
/// execution: every event's `deps` is causally closed and per-station contiguous,
/// the full causal closure of every event is present, and `a -> b` implies
/// `a.stamp < b.stamp`. `producer` and the entries of `learn_from` are 0-based
/// station indices; the matching `Clock` `station_id` is `index + 1` (`0` is not a
/// valid station). Payload is the event's index in the returned history.
pub(super) fn build_history(
    num_stations: usize,
    steps: &[(usize, Vec<usize>)],
) -> Vec<Event<Payload>> {
    let mut producers: Vec<Producer<Clock<TickCounter>>> = (1..=num_stations)
        .map(|id| {
            let station_id = u32::try_from(id).unwrap();
            let clock = Clock::with_default_config(TickCounter::new(), station_id).unwrap();
            Producer::new(clock)
        })
        .collect();
    let mut history: Vec<Event<Payload>> = Vec::new();

    for &(producer, ref learn_from) in steps {
        // Deliver, in arrival order, every peer event the producer did not author.
        // `observe` is order-robust, so any arrival order yields a valid causal cut.
        for event in &history {
            let source = event.stamp.station_id() as usize - 1;
            if source != producer && learn_from.contains(&source) {
                producers[producer].observe(event);
            }
        }
        let payload = u32::try_from(history.len()).unwrap();
        history.push(producers[producer].produce(0u16, payload));
    }
    history
}

/// Valid causal histories over 1..=4 stations and 1..14 events.
pub(super) fn arb_history() -> impl Strategy<Value = Vec<Event<Payload>>> {
    (1usize..=4).prop_flat_map(|num_stations| {
        let step = (
            0..num_stations,
            prop::collection::vec(0..num_stations, 0..num_stations),
        );
        prop::collection::vec(step, 1..14)
            .prop_map(move |steps| build_history(num_stations, &steps))
    })
}

/// A shuffled permutation of `0..len`, used as an arrival order over a history.
pub(super) fn perm_of(len: usize) -> impl Strategy<Value = Vec<usize>> {
    Just((0..len).collect::<Vec<usize>>()).prop_shuffle()
}