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

extern crate alloc;

use super::{Doc, Op, SCRIBES, arb_ops, observed};
use crate::metis::{Composer, DotMap, DotSet, Dotted};
use alloc::vec::Vec;
use proptest::prelude::*;

proptest! {
    #[test]
    fn prop_delta_exchange_converges_like_full_states(ops in arb_ops()) {
        let mut composers: Vec<Composer<Doc>> = (0..SCRIBES)
            .map(|i| Composer::new(u32::try_from(i).unwrap()))
            .collect();
        let mut states: Vec<Dotted<Doc>> = (0..SCRIBES).map(|_| Dotted::new()).collect();

        for op in &ops {
            match *op {
                Op::Add { composer, key } => {
                    let _ = composers[composer].compose(move |assigned| {
                        let mut held = DotSet::new();
                        let _ = held.insert(assigned);
                        (DotMap::singleton(key, held), DotSet::new())
                    });
                    let station = u32::try_from(composer).unwrap();
                    let assigned = states[composer].next_dot(station);
                    let mut held = DotSet::new();
                    let _ = held.insert(assigned);
                    let delta = Dotted::from_store(DotMap::singleton(key, held));
                    states[composer] = states[composer].merge(&delta);
                }
                Op::Remove { composer, key } => {
                    let seen = observed(composers[composer].state(), key);
                    let _ = composers[composer].retract(seen);
                    let obs = observed(&states[composer], key);
                    states[composer] = states[composer].merge(&Dotted::from_context(obs));
                }
                Op::Gossip { from, to } => {
                    let delta = composers[from].owed_to(composers[to].state().context());
                    let _ = composers[to].absorb(u32::try_from(from).unwrap(), &delta);
                    let full = states[from].clone();
                    states[to] = states[to].merge(&full);
                }
            }
        }

        let full_fold = states.iter().fold(Dotted::new(), |acc, s| acc.merge(s));
        let snapshots: Vec<Dotted<Doc>> = composers.iter().map(|s| s.state().clone()).collect();
        let mut converged = composers;
        for (to, composer) in converged.iter_mut().enumerate() {
            for (from, snapshot) in snapshots.iter().enumerate() {
                if from == to {
                    continue;
                }
                let delta = Composer::adopt(u32::try_from(from).unwrap(), snapshot.clone())
                    .owed_to(composer.state().context());
                let _ = composer.absorb(u32::try_from(from).unwrap(), &delta);
            }
        }
        for composer in &converged {
            prop_assert_eq!(composer.state(), &full_fold);
        }
    }
}