atomr-cluster 0.1.0

Membership, gossip, reachability, vector clocks, and split-brain resolvers for atomr clusters.
Documentation 
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

//! Gossip envelope. akka.net: `Cluster/Gossip.cs`, `GossipOverview.cs`.

use serde::{Deserialize, Serialize};

use crate::membership::MembershipState;
use crate::vector_clock::VectorClock;

#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct Gossip {
    pub version: VectorClock,
    pub state: MembershipState,
}

impl Gossip {
    pub fn new() -> Self {
        Self::default()
    }

    /// Increment our local clock entry for `node`.
    pub fn tick(&mut self, node: &str) {
        self.version.tick(node);
    }

    /// Merge with another gossip, taking vector-clock max and union of members.
    pub fn merge(&self, other: &Self) -> Self {
        let mut merged = Self { version: self.version.merge(&other.version), ..Self::default() };
        for m in self.state.members.iter().chain(other.state.members.iter()) {
            merged.state.add_or_update(m.clone());
        }
        for ((a, b), st) in
            self.state.reachability.records.iter().chain(other.state.reachability.records.iter())
        {
            merged.state.reachability.records.insert((a.clone(), b.clone()), *st);
        }
        merged
    }
}

/// Snapshot of seen-by + unreachable info. akka.net: `GossipOverview.cs`.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct GossipOverview {
    pub seen_by: Vec<String>,
    pub reachability: crate::reachability::Reachability,
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::member::Member;
    use atomr_core::actor::Address;

    #[test]
    fn merge_is_commutative_for_members() {
        let mut a = Gossip::new();
        a.tick("A");
        a.state.add_or_update(Member::new(Address::local("A"), vec![]));
        let mut b = Gossip::new();
        b.tick("B");
        b.state.add_or_update(Member::new(Address::local("B"), vec![]));
        let ab = a.merge(&b);
        let ba = b.merge(&a);
        assert_eq!(ab.state.member_count(), 2);
        assert_eq!(ab.state.member_count(), ba.state.member_count());
    }
}