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

extern crate alloc;

use alloc::collections::BTreeMap;

mod iter;
mod lattice;
#[cfg(kani)]
mod proofs;
mod relative;
mod wire;

pub use iter::Iter;
pub use relative::Disagreement;
pub use wire::DecodeError;

/// A version vector: for each station, the count of that station's events this
/// vector has observed.
///
/// The structure a [`Kairos`](crate::kairos::Kairos) scalar cannot be: it decides
/// whether two events are causally ordered or *concurrent*. The partial order
/// `<=` is pointwise `<=` and is exactly happens-before; two vectors that each
/// lead the other on some station are concurrent, which a single timestamp cannot
/// express. [`merge`](Self::merge) is the pointwise maximum, the least upper bound
/// of the lattice.
///
/// # Counters, not the `logical` field
///
/// A station's counter here is a lifetime-monotonic `u64` event count. It is
/// deliberately *not* [`Kairos::logical`](crate::kairos::Kairos::logical), the
/// 16-bit Lamport tiebreak that resets whenever physical time advances. The two
/// are complementary: keep the stamp for the total order you sort and ship, keep
/// the vector for the causal questions the total order cannot answer.
///
/// # Canonical form
///
/// A station at count `0` is held *absent* from the map, so "absent equals zero"
/// and equal vectors compare and hash equally regardless of how they were built.
/// Every operation preserves this: none ever stores a zero counter.
///
/// The type implements `PartialOrd` and deliberately omits `Ord`. Concurrent
/// vectors are genuinely incomparable, so a total order would report an order
/// that does not exist. Contrast the total [`Ord`] on `Kairos`.
#[derive(Clone, Default, PartialEq, Eq, Hash, Debug)]
pub struct VersionVector {
    counters: BTreeMap<u32, u64>,
}

impl VersionVector {
    /// Creates an empty version vector: the bottom of the lattice, every station
    /// at `0`.
    #[must_use]
    pub const fn new() -> Self {
        Self {
            counters: BTreeMap::new(),
        }
    }

    /// Returns the observed event count for `station_id`. A station never observed
    /// reads as `0`.
    #[must_use]
    pub fn get(&self, station_id: u32) -> u64 {
        self.counters.get(&station_id).copied().unwrap_or(0)
    }

    /// Records a local event at `station_id`, advancing its counter by one and
    /// returning the new value.
    ///
    /// Below the ceiling this always moves the vector strictly up the
    /// order; [`observe`](Self::observe) and [`merge`](Self::merge) move it
    /// up only where the argument leads. Counters *saturate* at [`u64::MAX`] rather than
    /// wrapping, so the move is strict only while `get(station_id) <
    /// u64::MAX`; at the ceiling it is a no-op. A station that minted
    /// `u64::MAX` events would re-issue its top dot. Successive events would
    /// then share that dot, and a buffer's [`Causal`](crate::metis::Causal)
    /// gate would treat each later event as a duplicate. The horizon is
    /// unreachable in practice: `2^64` events from one station. Saturation is
    /// the crate's deliberate panic-free choice. The guarantee is that a
    /// counter never wraps and never panics, not that it increases without
    /// bound.
    pub fn increment(&mut self, station_id: u32) -> u64 {
        let next = self.get(station_id).saturating_add(1);
        // next >= 1, so this never writes a zero entry (canonical form holds).
        let _ = self.counters.insert(station_id, next);
        next
    }

    /// Raises `station_id`'s counter to at least `counter`, recording that this
    /// many of its events have been observed.
    ///
    /// A no-op if the station is already at or beyond `counter`; in particular
    /// `observe(s, 0)` is always a no-op, so canonical form holds.
    pub fn observe(&mut self, station_id: u32, counter: u64) {
        if counter > self.get(station_id) {
            // counter > current >= 0, so counter >= 1: no zero entry is written.
            let _ = self.counters.insert(station_id, counter);
        }
    }

    /// Iterates the observed `(station_id, counter)` entries in ascending station
    /// order.
    ///
    /// Canonical form (a station at `0` is absent) means every yielded counter is
    /// `>= 1`; a station never observed is simply not produced. This is the read
    /// side a consumer walks, for example the delivery gate in
    /// [`Ideal`](crate::metis::Ideal).
    #[must_use]
    pub fn iter(&self) -> Iter<'_> {
        Iter::new(self.counters.iter())
    }
}

impl<'a> IntoIterator for &'a VersionVector {
    type Item = (u32, u64);
    type IntoIter = Iter<'a>;

    fn into_iter(self) -> Self::IntoIter {
        self.iter()
    }
}