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
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163

extern crate alloc;

use alloc::vec::Vec;

use crate::kairos::Kairos;
use crate::metis::{Anchor, Composer, DotSet, Dotted, Locus, Retired, Rhapsody};

use super::reference::reference_order;
use super::wire::round_trip_rhapsody;
use super::{Dot, Fleet, REPLICAS, station_of};

/// The parent-with-side to insert-at-position translation law (S288, the
/// lit-review deliverable): under the tape's freshest-rank mint (the global
/// tick makes every new rank the strict maximum, so the new element heads
/// its sibling bucket), a weave realizes *exact index insertion*: the
/// post-op visible order is the pre-op order with the minted dot at exactly
/// the intended index. This is the translation that puts the anchor-based
/// surface inside the list-specification vocabulary; without it the strong
/// list ledger would be checking sequences with no stated relation to
/// `ins(a, k)`.
fn assert_index_insertion(pre: &[Dot], post: &[Dot], minted: Dot, index: usize) {
    let mut expected: Vec<Dot> = pre.to_vec();
    expected.insert(index, minted);
    assert_eq!(
        post, expected,
        "a freshest-rank weave must land at its intended index",
    );
}

/// Weave: insert a fresh element after a visible dot, or at the origin.
pub(super) fn weave(fleet: &mut Fleet, replica: usize, argument: u8, tick: &mut u64) {
    let station = station_of(replica);
    let visible = fleet.live[replica].state().store().order();
    let (anchor, intended) = if visible.is_empty() {
        (Anchor::Origin, 0)
    } else {
        let slot = usize::from(argument) % visible.len();
        (Anchor::After(visible[slot].into()), slot + 1)
    };
    let rank = Kairos::new(*tick, 0u16, station, 0u16);
    *tick += 1;
    let (minted, delta) = fleet.live[replica].compose(|assigned| {
        let mut rhapsody = Rhapsody::new();
        let _ = rhapsody.weave(assigned, Locus { anchor, rank });
        (rhapsody, DotSet::new())
    });
    round_trip_rhapsody(delta.store());
    round_trip_rhapsody(fleet.live[replica].state().store());
    let post = fleet.live[replica].state().store().order();
    assert_index_insertion(&visible, &post, minted, intended);
    let _ = fleet.ledger.observe(&post);
}

/// `WeaveBefore`: insert before a visible dot through the visual-insert seam.
pub(super) fn weave_before(fleet: &mut Fleet, replica: usize, argument: u8, tick: &mut u64) {
    let station = station_of(replica);
    let visible = fleet.live[replica].state().store().order();
    let intended = if visible.is_empty() {
        0
    } else {
        usize::from(argument) % visible.len()
    };
    let after = if intended == 0 {
        None
    } else {
        Some(visible[intended - 1])
    };
    let anchor = fleet.live[replica]
        .state()
        .store()
        .anchor_for_visual_insert(after.map(Into::into));
    let rank = Kairos::new(*tick, 0u16, station, 0u16);
    *tick += 1;
    let (minted, delta) = fleet.live[replica].compose(|assigned| {
        let mut rhapsody = Rhapsody::new();
        let _ = rhapsody.weave(assigned, Locus { anchor, rank });
        (rhapsody, DotSet::new())
    });
    round_trip_rhapsody(delta.store());
    round_trip_rhapsody(fleet.live[replica].state().store());
    let post = fleet.live[replica].state().store().order();
    assert_index_insertion(&visible, &post, minted, intended);
    let _ = fleet.ledger.observe(&post);
}

/// Retract one visible dot and record it in the removed oracle.
pub(super) fn retract(fleet: &mut Fleet, replica: usize, argument: u8) {
    let visible = fleet.live[replica].state().store().order();
    if visible.is_empty() {
        return;
    }
    let victim = visible[usize::from(argument) % visible.len()];
    let _ = fleet.ever_removed.insert(victim);
    let mut superseded = DotSet::new();
    let _ = superseded.insert(victim);
    let _ = fleet.live[replica].retract(superseded);
    round_trip_rhapsody(fleet.live[replica].state().store());
    let post = fleet.live[replica].state().store().order();
    let _ = fleet.ledger.observe(&post);
}

/// `GossipDelta`: merge exactly what `other` owes `replica`.
pub(super) fn gossip_delta(fleet: &mut Fleet, replica: usize, other: usize) {
    if replica == other {
        return;
    }
    let peer_context = fleet.live[replica].state().context().clone();
    let delta = fleet.live[other].owed_to(&peer_context);
    round_trip_rhapsody(delta.store());
    let _ = fleet.live[replica].absorb(station_of(other), &delta);
    round_trip_rhapsody(fleet.live[replica].state().store());
    let post = fleet.live[replica].state().store().order();
    let _ = fleet.ledger.observe(&post);
}

/// `GossipFull`: absorb `other`'s full state into `replica`.
pub(super) fn gossip_full(fleet: &mut Fleet, replica: usize, other: usize) {
    if replica == other {
        return;
    }
    let full = fleet.live[other].state().clone();
    round_trip_rhapsody(full.store());
    let _ = fleet.live[replica].absorb(station_of(other), &full);
    let post = fleet.live[replica].state().store().order();
    let _ = fleet.ledger.observe(&post);
}

/// Condense every replica under one honest removed-everywhere claim.
pub(super) fn condense_all(fleet: &mut Fleet) {
    let mut retired = DotSet::new();
    for &dot in &fleet.ever_removed {
        let invisible_everywhere = fleet
            .live
            .iter()
            .all(|r| !r.state().store().is_visible(dot));
        if invisible_everywhere {
            let _ = retired.insert(dot);
        }
    }
    let retired = Retired::trust(retired);

    for i in 0..REPLICAS {
        let mut rhapsody = fleet.live[i].state().store().clone();
        let _ = rhapsody.condense(&retired);
        let context = fleet.live[i].state().context().clone();
        let rebuilt =
            Dotted::try_new(rhapsody, context).unwrap_or_else(|_| fleet.live[i].state().clone());
        let before = fleet.live[i].state().store().order();
        let after = rebuilt.store().order();
        assert_eq!(before, after, "condense preserves order at every replica");
        fleet.live[i] = Composer::adopt(station_of(i), rebuilt);
        let _ = fleet.ledger.observe(&after);
    }
}

/// Read order through the maintained path and the independent reference walk.
pub(super) fn order_read(fleet: &mut Fleet, replica: usize) {
    let rhapsody = fleet.live[replica].state().store();
    let got = rhapsody.order();
    let reference = reference_order(rhapsody);
    assert_eq!(got, reference, "order() matches the independent reference");
    let _ = fleet.ledger.observe(&got);
}