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
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
|
//! Which members a node syncs with: its neighbours.
//!
//! In a full mesh every change reaches every node from every other node (Automerge
//! decides per pair, so nobody knows that a third peer already passed it on), so the
//! copies of one write grow with N². Above `k` members, each node keeps only a few
//! neighbours, and changes travel over several hops.
//!
//! Every node computes the same graph from the member list alone, with no coordination:
//! * a ring ordered by key hash: each member's predecessor and successor, which keeps
//! the graph connected;
//! * plus, for each member, the `k - 2` others with the lowest pair score (a hash of
//! both keys, so both ends agree on it), which keeps paths short.
//!
//! A link exists if either end chose it, so links are symmetric and a node has at least
//! `k` neighbours. With at most `k + 1` members the graph is the full mesh: small swarms
//! behave exactly as before. A member joining or leaving changes only links near it.
use std::collections::{BTreeMap, BTreeSet};
use sha2::{Digest, Sha256};
const RING_LABEL: &[u8] = b"continuum/topology-ring/1";
const PAIR_LABEL: &[u8] = b"continuum/topology-pair/1";
/// The neighbours of `local` among `members` (which may or may not contain `local`).
pub(crate) fn neighbours(members: &[String], local: &str, k: usize) -> BTreeSet<String> {
let mut all: Vec<String> = members.to_vec();
all.push(local.to_owned());
graph(&all, k).remove(local).unwrap_or_default()
}
/// Every member's neighbours. Costs about N² hashes (a few milliseconds for hundreds of
/// members); computed again only when the member list changes.
pub(crate) fn graph(members: &[String], k: usize) -> BTreeMap<String, BTreeSet<String>> {
let k = k.max(2);
let all: BTreeSet<&str> = members.iter().map(String::as_str).collect();
let mut links: BTreeMap<String, BTreeSet<String>> = all
.iter()
.map(|m| ((*m).to_owned(), BTreeSet::new()))
.collect();
let mut link = |a: &str, b: &str| {
if a != b {
links.get_mut(a).expect("member").insert(b.to_owned());
links.get_mut(b).expect("member").insert(a.to_owned());
}
};
if all.len() <= k + 1 {
for a in &all {
for b in &all {
link(a, b);
}
}
return links;
}
let ring = ring_links(&all);
for (member, [prev, next]) in &ring {
link(member, prev);
link(member, next);
// Its k - 2 partners with the lowest pair score, besides its ring links.
let mut candidates: Vec<([u8; 32], &str)> = all
.iter()
.copied()
.filter(|other| other != member && other != prev && other != next)
.map(|other| (pair_score(member, other), other))
.collect();
candidates.sort_unstable();
for (_, partner) in candidates.into_iter().take(k - 2) {
link(member, partner);
}
}
links
}
/// Ring predecessor and successor of every member, by key hash.
fn ring_links<'a>(all: &BTreeSet<&'a str>) -> BTreeMap<&'a str, [&'a str; 2]> {
let mut ring: Vec<([u8; 32], &'a str)> =
all.iter().map(|m| (hash(RING_LABEL, &[*m]), *m)).collect();
ring.sort_unstable();
let n = ring.len();
(0..n)
.map(|i| (ring[i].1, [ring[(i + n - 1) % n].1, ring[(i + 1) % n].1]))
.collect()
}
/// Symmetric: the same for (a, b) and (b, a).
fn pair_score(a: &str, b: &str) -> [u8; 32] {
let (low, high) = if a <= b { (a, b) } else { (b, a) };
hash(PAIR_LABEL, &[low, high])
}
fn hash(label: &[u8], fields: &[&str]) -> [u8; 32] {
let mut hasher = Sha256::new();
for field in [label]
.into_iter()
.chain(fields.iter().map(|f| f.as_bytes()))
{
hasher.update((field.len() as u32).to_be_bytes());
hasher.update(field);
}
hasher.finalize().into()
}
#[cfg(test)]
mod tests {
use super::*;
use std::collections::VecDeque;
fn members(n: usize) -> Vec<String> {
(0..n).map(|i| format!("{:064x}", i * 7919 + 13)).collect()
}
fn edges(graph: &BTreeMap<String, BTreeSet<String>>) -> BTreeSet<(String, String)> {
graph
.iter()
.flat_map(|(a, ns)| {
ns.iter().map(move |b| {
if a < b {
(a.clone(), b.clone())
} else {
(b.clone(), a.clone())
}
})
})
.collect()
}
#[test]
fn small_swarms_are_a_full_mesh() {
let ms = members(13);
for m in &ms {
assert_eq!(neighbours(&ms, m, 12).len(), 12);
}
}
#[test]
fn links_are_symmetric_and_degree_at_least_k() {
let ms = members(250);
let g = graph(&ms, 8);
for (a, ns) in &g {
assert!(ns.len() >= 8, "{a} has {} neighbours", ns.len());
assert!(!ns.contains(a));
for b in ns {
assert!(g[b].contains(a), "{a} -> {b} not symmetric");
}
}
}
#[test]
fn graph_is_connected() {
for (n, k) in [(250, 8), (250, 2), (60, 3)] {
let ms = members(n);
let g = graph(&ms, k);
let mut seen = BTreeSet::from([ms[0].clone()]);
let mut queue = VecDeque::from([ms[0].clone()]);
while let Some(m) = queue.pop_front() {
for next in &g[&m] {
if seen.insert(next.clone()) {
queue.push_back(next.clone());
}
}
}
assert_eq!(seen.len(), n, "n={n} k={k}");
}
}
#[test]
fn one_member_joining_changes_few_links() {
let ms = members(200);
let before = edges(&graph(&ms, 8));
let mut more = ms.clone();
more.push(format!("{:064x}", 999_999));
let after = edges(&graph(&more, 8));
let changed = before.symmetric_difference(&after).count();
assert!(changed <= 4 * 8, "{changed} links changed");
}
}
|
