aboutsummaryrefslogtreecommitdiffstats
path: root/src/swarm/tree.rs
blob: 20df2ec94750de3aff65011d7e3f271919d152b1 (plain)
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
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
//! Passing a writer's changes on along its tree: every member gets them once, over a
//! link that works, and as soon as the member before it has them.
//!
//! The tree is the same on every node, computed from what they share: the graph the
//! member list gives (`topology::member_list_graph`: every member's neighbours, as if
//! all reached all) and what each member tells its neighbours it reaches
//! (`swarm::reach`). Every member's distance from the writer over that graph orders the
//! members: by distance, then whether it reaches a member a hop closer, then pair key;
//! a member's parent is the neighbour it reaches that comes before it, the closest to
//! the writer first, then one that reaches closer, then by pair score
//! (`topology::pair_score`). Coming before rules out cycles, and taking only a neighbour
//! it reaches routes around links that don't work: where a writer behind NAT reaches a
//! single member, that one comes first and the others take it.
//!
//! Whose a change is, its author says (every node writes as itself: its node ID, by
//! pair key here): whoever applies changes passes them on to its own children in each of
//! their writers' trees (`forward`), however they came; so does the writer, to all its
//! neighbours, with its own writes. A sync message along a tree also says whose
//! (`SyncEnvelope::writer`), for one that had them already (`swarm::sync`); like any
//! sync message, it needs no answer where the child had nothing the parent lacks
//! (`SyncSession::assume_delivered`).
//! A member no member it reaches comes before (or whose parent didn't get the change)
//! gets it from the announcements (`swarm::relay`), as before.
//!
//! The graph is computed once per member list (tens of milliseconds for thousands of
//! members), the distances once per writer and graph (`MAX_WRITERS` kept).

use std::collections::{BTreeMap, BTreeSet, HashMap, VecDeque};
use std::sync::Arc;

use tracing::trace;

use crate::locks::MutexExt;
use crate::state::AppState;
use crate::swarm::{reach, topology};

/// The writers whose distances are kept at most.
const MAX_WRITERS: usize = 256;

type Graph = BTreeMap<String, BTreeSet<String>>;

/// The distances from each writer, over the graph they were computed for.
#[derive(Default)]
pub(crate) struct Trees {
    graph: Option<Arc<Graph>>,
    /// The members by pair key, of that graph.
    by_key: HashMap<u64, String>,
    distances: HashMap<u64, Arc<HashMap<String, u32>>>,
}

/// The graph, and every member's distance from `writer` (by pair key) over it; None
/// while the member list isn't known, or `writer` isn't in it.
fn distances(state: &AppState, writer: u64) -> Option<(Arc<Graph>, Arc<HashMap<String, u32>>)> {
    let graph = topology::member_list_graph(state)?;
    let mut trees = state.trees.lock_or_recover();
    if !trees
        .graph
        .as_ref()
        .is_some_and(|known| Arc::ptr_eq(known, &graph))
    {
        trees.by_key = graph
            .keys()
            .map(|member| (topology::pair_key(member), member.clone()))
            .collect();
        trees.distances.clear();
        trees.graph = Some(Arc::clone(&graph));
    }
    if let Some(known) = trees.distances.get(&writer) {
        return Some((graph, Arc::clone(known)));
    }
    let from = trees.by_key.get(&writer)?.clone();
    let mut distance = HashMap::from([(from.clone(), 0u32)]);
    let mut queue = VecDeque::from([from]);
    while let Some(member) = queue.pop_front() {
        let next = distance[&member] + 1;
        for neighbour in graph.get(&member).into_iter().flatten() {
            if !distance.contains_key(neighbour) {
                distance.insert(neighbour.clone(), next);
                queue.push_back(neighbour.clone());
            }
        }
    }
    if trees.distances.len() >= MAX_WRITERS {
        trees.distances.clear();
    }
    let distance = Arc::new(distance);
    trees.distances.insert(writer, Arc::clone(&distance));
    Some((graph, distance))
}

/// Our children in `writer`'s tree: our neighbours in the graph whose parent we are,
/// and whom we reach.
///
/// Members are ordered by distance from the writer, then whether they reach a member a
/// hop closer (as they say: else, while it hasn't the change from its own side, it is
/// as good as farther), then pair key; a member's parent is the neighbour it reaches
/// that comes before it, by the same, then pair score. So where the writer reaches few
/// (behind NAT), those it reaches come first among the members as far from it, and the
/// others take them as parents. What a member reaches we know of ourselves and our
/// neighbours, as they said; of others, the graph is taken as right (if we're wrong,
/// a change goes twice or by announcement).
#[cfg(test)]
pub(crate) async fn children(state: &AppState, writer: u64) -> Vec<String> {
    plan(state, writer)
        .await
        .map(|plan| plan.children)
        .unwrap_or_default()
}

/// Where `member` comes in the writer's tree (by `distance` from it over `graph`):
/// its distance, then whether it reaches a member a hop closer (`reached(by, other)`:
/// whether `by` reaches `other`, as known; the writer does), then its pair key. None if
/// the graph doesn't have it.
fn rank(
    graph: &Graph,
    distance: &HashMap<String, u32>,
    reached: &dyn Fn(&str, &str) -> bool,
    member: &str,
) -> Option<(u32, bool, u64)> {
    let d = *distance.get(member)?;
    let closer =
        d == 0
            || graph.get(member).into_iter().flatten().any(|other| {
                distance.get(other.as_str()) == Some(&(d - 1)) && reached(member, other)
            });
    Some((d, !closer, topology::pair_key(member)))
}

/// `member`'s parent in the writer's tree (`rank`): the neighbour it reaches that comes
/// before it, the closest to the writer first, then one that reaches closer, then by
/// pair score; None if none does (or it is the writer).
fn parent<'g>(
    graph: &'g Graph,
    distance: &HashMap<String, u32>,
    reached: &dyn Fn(&str, &str) -> bool,
    member: &str,
) -> Option<&'g str> {
    let its_rank = rank(graph, distance, reached, member)?;
    graph
        .get(member)
        .into_iter()
        .flatten()
        .map(String::as_str)
        .filter(|candidate| {
            rank(graph, distance, reached, candidate).is_some_and(|r| r < its_rank)
                && reached(member, candidate)
        })
        .min_by_key(|candidate| {
            let (d, far, _) = rank(graph, distance, reached, candidate).expect("ranked");
            (d, far, topology::pair_score(candidate, member))
        })
}

/// `writer`'s whole tree, as far as `reached` knows whom each member reaches (the
/// dashboard's map, from the links members report): every member's parent, None for
/// those without one (they get the writer's changes by announcement). None while the
/// member list isn't known, or the writer isn't in it.
#[cfg_attr(not(feature = "dashboard"), allow(dead_code))]
pub(crate) fn whole(
    state: &AppState,
    writer: &str,
    reached: &dyn Fn(&str, &str) -> bool,
) -> Option<BTreeMap<String, Option<String>>> {
    let (graph, distance) = distances(state, topology::pair_key(writer))?;
    Some(
        graph
            .keys()
            .filter(|member| distance.get(member.as_str()).is_some_and(|d| *d > 0))
            .map(|member| {
                let parent = parent(&graph, &distance, reached, member).map(str::to_owned);
                (member.clone(), parent)
            })
            .collect(),
    )
}

/// Whom we and our neighbours in `graph` reach, by pair key, as far as we know (our
/// own, and what our neighbours said).
async fn known_reaches<'g>(state: &AppState, graph: &'g Graph) -> HashMap<&'g str, Vec<u64>> {
    let me = state.local_id();
    let mut reaches: HashMap<&str, Vec<u64>> = HashMap::new();
    if let Some((me, _)) = graph.get_key_value(me) {
        reaches.insert(me.as_str(), reach::published(state));
    }
    for member in graph.get(me).into_iter().flatten() {
        if let Some(peer_state) = state.get_peer_state(member).await
            && let Some(keys) = reach::of(peer_state.lock().await.reaches.as_ref())
        {
            reaches.insert(member.as_str(), keys);
        }
    }
    reaches
}

/// Our own tree, as our writes go along it (`Plan`, as every member computes its part).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct Shape {
    /// The other members in it.
    pub(crate) members: usize,
    /// The most hops from us to one of them.
    pub(crate) depth: u32,
    /// The hops to each, summed (for the mean).
    pub(crate) hops: u64,
    /// Members it doesn't reach: without a parent there (no member they reach comes
    /// before them), below one without, or not connected to us in the member list's
    /// graph. They get our writes by announcement.
    pub(crate) off: usize,
}

/// The shape of our own tree; None while the member list isn't known, or we aren't in
/// it. What members other than us and our neighbours reach is taken from the graph, as
/// in `plan`.
pub(crate) async fn own_shape(state: &AppState) -> Option<Shape> {
    let me = state.local_id();
    let (graph, distance) = distances(state, topology::pair_key(me))?;
    let reaches = known_reaches(state, &graph).await;
    let reached = |by: &str, member: &str| {
        reaches
            .get(by)
            .is_none_or(|keys| keys.contains(&topology::pair_key(member)))
    };
    Some(shape(&graph, &distance, &reached, me))
}

/// `own_shape` over `graph`, from `me` at distance 0.
fn shape(
    graph: &Graph,
    distance: &HashMap<String, u32>,
    reached: &dyn Fn(&str, &str) -> bool,
    me: &str,
) -> Shape {
    let parents: HashMap<&str, Option<&str>> = graph
        .keys()
        .filter(|member| distance.get(member.as_str()).is_some_and(|d| *d > 0))
        .map(|member| (member.as_str(), parent(graph, distance, reached, member)))
        .collect();
    // Hops from us along parents; None where the chain ends without us. Parents come
    // before their children, so chains end.
    let mut hops: HashMap<&str, Option<u32>> = HashMap::from([(me, Some(0))]);
    fn along<'g>(
        member: &'g str,
        parents: &HashMap<&'g str, Option<&'g str>>,
        hops: &mut HashMap<&'g str, Option<u32>>,
    ) -> Option<u32> {
        if let Some(known) = hops.get(member) {
            return *known;
        }
        let found = parents
            .get(member)
            .copied()
            .flatten()
            .and_then(|parent| along(parent, parents, hops))
            .map(|hops| hops + 1);
        hops.insert(member, found);
        found
    }
    let mut shape = Shape {
        members: 0,
        depth: 0,
        hops: 0,
        off: graph.len() - 1 - parents.len(),
    };
    for member in parents.keys() {
        match along(member, &parents, &mut hops) {
            Some(h) => {
                shape.members += 1;
                shape.depth = shape.depth.max(h);
                shape.hops += u64::from(h);
            }
            None => shape.off += 1,
        }
    }
    shape
}

/// What `writer`'s tree means for us: our children in it, and our neighbours it leaves
/// out (`uncovered`).
pub(crate) struct Plan {
    pub(crate) children: Vec<String>,
    /// Those of our children that reach no member but us, as they said: nobody else's
    /// slow round tells them of a change, should our message to them be lost (it is
    /// taken as delivered, `SyncSession::assume_delivered`), so ours does.
    pub(crate) alone: Vec<String>,
    /// Our neighbours that no member they reach comes before (the tree reaches them only
    /// by announcement), and those not in the member list's graph (whom no tree has).
    pub(crate) uncovered: Vec<String>,
    /// Our neighbours that get the writer's changes from another parent, and the writer
    /// itself: they have them, or will, without us (`forward`).
    pub(crate) covered: Vec<String>,
}

/// Our part in `writer`'s tree (`children`, `Plan`); None while the member list isn't
/// known, or the writer or we aren't in it.
pub(crate) async fn plan(state: &AppState, writer: u64) -> Option<Plan> {
    let (graph, distance) = distances(state, writer)?;
    let me = state.local_id();
    if !distance.contains_key(me) {
        return None;
    }
    let reaches = known_reaches(state, &graph).await;
    let reached = |by: &str, member: &str| {
        reaches
            .get(by)
            .is_none_or(|keys| keys.contains(&topology::pair_key(member)))
    };
    let my_rank = rank(&graph, &distance, &reached, me)?;
    let my_key = topology::pair_key(me);
    let mut children = Vec::new();
    let mut alone = Vec::new();
    let mut uncovered = Vec::new();
    let mut covered = Vec::new();
    for member in graph.get(me).into_iter().flatten() {
        let Some(its_rank) = rank(&graph, &distance, &reached, member) else {
            continue;
        };
        if distance[member.as_str()] == 0 {
            covered.push(member.clone());
            continue;
        }
        match parent(&graph, &distance, &reached, member) {
            None => uncovered.push(member.clone()),
            Some(parent)
                if parent == me && its_rank > my_rank && topology::reaches(state, member).await =>
            {
                if reaches
                    .get(member.as_str())
                    .is_some_and(|keys| keys.iter().all(|key| *key == my_key))
                {
                    alone.push(member.clone());
                }
                children.push(member.clone());
            }
            Some(parent) if parent != me => covered.push(member.clone()),
            // Ours by the graph, but we don't reach it, or it comes before us.
            Some(_) => {}
        }
    }
    // Neighbours the member list's graph doesn't give us (not listed yet, or only
    // connecting out): no tree has them.
    for (peer, _) in state.peers_snapshot().await {
        if topology::is_neighbour(state, &peer)
            && !graph.get(me).is_some_and(|ours| ours.contains(&peer))
        {
            uncovered.push(peer);
        }
    }
    Some(Plan {
        children,
        alone,
        uncovered,
        covered,
    })
}

/// We applied `changes` of `writer`'s that `from` passed on (none if we had them
/// already): to our children in its tree (but `from`), a sync message each, saying
/// whose. Returns our part in the tree (`Plan`), None if we have none (announcements
/// then do it all).
///
/// Our neighbours that get them from another parent (`Plan::covered`) count as having
/// them (`SyncSession::known_to_peer`): else, with writers in turn, each change coming
/// along another tree, the next message we pass on to one would carry all we got since
/// our last to it, mostly what its other parents brought it. Should one lack them after
/// all, it asks, and gets them (`resend_if_needed`).
pub(crate) async fn forward(
    state: &AppState,
    writer: u64,
    from: &str,
    changes: &[crate::document::ChangeHash],
) -> Option<Plan> {
    let mut plan = plan(state, writer).await?;
    if !changes.is_empty() {
        for neighbour in &plan.covered {
            if let Some(peer_state) = state.get_peer_state(neighbour).await {
                peer_state.lock().await.sync.known_to_peer(changes);
            }
        }
    }
    plan.children.retain(|child| child != from);
    let children = &plan.children;
    for child in children {
        if let Some(peer_state) = state.get_peer_state(child).await {
            peer_state.lock().await.tree_writer = Some(writer);
        }
        state.stats.tree_forward();
        state.request_sync(child);
    }
    if !children.is_empty() {
        trace!(
            writer,
            children = children.len(),
            "Passing a change on along the writer's tree"
        );
    }
    Some(plan)
}

#[cfg(test)]
mod tests {
    use super::*;

    /// Hops along parents from us; members without a parent, below one without, or
    /// not connected count as off the tree.
    #[test]
    fn our_tree_in_numbers() {
        let link = |graph: &mut Graph, a: &str, b: &str| {
            graph.entry(a.into()).or_default().insert(b.into());
            graph.entry(b.into()).or_default().insert(a.into());
        };
        let mut graph = Graph::new();
        link(&mut graph, "me", "a");
        link(&mut graph, "a", "b");
        link(&mut graph, "me", "c");
        link(&mut graph, "c", "e");
        graph.insert("d".into(), BTreeSet::new());
        let distance: HashMap<String, u32> = [("me", 0), ("a", 1), ("c", 1), ("b", 2), ("e", 2)]
            .into_iter()
            .map(|(member, d)| (member.to_owned(), d))
            .collect();
        // c doesn't reach us: no parent, nor e below it.
        let reached = |by: &str, member: &str| !(by == "c" && member == "me");
        assert_eq!(
            shape(&graph, &distance, &reached, "me"),
            Shape {
                members: 2,
                depth: 2,
                hops: 3,
                off: 3,
            }
        );
    }
}