aboutsummaryrefslogtreecommitdiffstats
path: root/src/swarm/topology/mod.rs
blob: c23d2b28ab15d045f9b6ed4753dfea67362dd5f7 (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
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
//! 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 graph from the member list and which members it can reach,
//! with no coordination:
//! * a ring ordered by key hash: each member's nearest reachable member in both
//!   directions, which keeps the reachable members connected;
//! * plus, for each member, the `k - 2` reachable 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` reachable members the graph is the full mesh:
//! small swarms behave exactly as before. A member joining, leaving or going offline
//! changes only links near it.
//!
//! Members on several transports (Mycelium, libp2p) get such a graph per transport,
//! among the members on it (as their addresses in the member list say), and a node's
//! neighbours are the union of its neighbours on each transport it is on
//! (`neighbours_across`). Each of those graphs is connected (its ring), and a member on
//! two transports is in both: as long as members on several transports link the
//! transports, the swarm stays connected, and such a member has about `k` neighbours per
//! transport. With one transport, it is the one graph as before.

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

use tracing::info;

use crate::locks::RwLockExt;
use crate::state::{AppState, PeerState};
use crate::swarm::members;
use crate::transport::ParsedMessage;

mod graph;

use graph::*;
pub(crate) use graph::{RingAndPairs, pair_key, pair_score};

/// How members' neighbours are chosen: the graph above (`RingAndPairs`) is one way;
/// others can be tried against it in the benchmark (`docs/future-work.md`, Swappable
/// topology and dissemination). One per swarm: a node's neighbours must count it as
/// theirs.
///
/// A topology either computes a graph every node shares (`graph`: the same inputs give
/// every node the same), or keeps its own state from its own messages (HyParView's
/// views), and has none. What needs the shared graph does nothing without one: writers'
/// trees (`member_list_graph`, `swarm::tree`), the links members record where it is
/// wrong (`working_graph`, `swarm::working_links`), indirect probes through a member's
/// neighbours, the dashboard's map of the intended links (`intended_graph`). A
/// dissemination that needs none goes with such a topology (flooding, PlumTree).
#[async_trait::async_trait]
pub(crate) trait Topology: Send + Sync + 'static {
    /// Its name (in logs, and once chosen by the swarm, in the document).
    fn name(&self) -> &'static str;

    /// Every reachable member's neighbours among `members` (those on one transport),
    /// each keeping about `k`, skipping those `unreachable`, preferring those `stable`
    /// as it does. Pure: the same inputs give every node the same graph. None for a
    /// topology without a shared graph.
    fn graph(
        &self,
        members: &[String],
        k: usize,
        unreachable: &dyn Fn(&str) -> bool,
        stable: &dyn Fn(&str) -> bool,
    ) -> Option<BTreeMap<String, BTreeSet<String>>>;

    /// Our neighbours, from what they are chosen from (`Inputs`), each time those may
    /// have changed (`refresh`). By default our part of the shared graph, on each of our
    /// transports (`neighbours_across`); a topology with state of its own keeps them
    /// there.
    fn neighbours(&self, local: &str, inputs: &Inputs<'_>) -> BTreeSet<String> {
        neighbours_across(
            self,
            inputs.transports,
            local,
            inputs.local_transports,
            inputs.k,
            inputs.unreachable,
            inputs.stable,
        )
    }

    /// With every recomputation (`refresh`, at least every 5 s), before `neighbours`:
    /// what a topology with state of its own does about it (asking members to be
    /// neighbours, giving up on those that don't answer). By default nothing.
    async fn maintain(&self, state: &AppState, inputs: &Inputs<'_>) {
        let _ = (state, inputs);
    }

    /// A message of the topology's own from `peer` (HyParView's NEIGHBOUR,
    /// NEIGHBOUR_REPLY, DISCONNECT). By default ignored.
    async fn received(&self, state: &AppState, peer: &str, message: &ParsedMessage) {
        let _ = (state, peer, message);
    }

    /// Changes whenever the topology's own state changed what `neighbours` gives: the
    /// neighbours are recomputed then, even though nothing they are chosen from did. By
    /// default always the same.
    fn version(&self) -> u64 {
        0
    }

    /// Taking over from another topology, whose neighbours were `neighbours` (the swarm
    /// switched, `swarm_ways`): a topology with state of its own starts from them. By
    /// default nothing.
    fn seed(&self, neighbours: &BTreeSet<String>) {
        let _ = neighbours;
    }

    /// The neighbours each agreed to, for a topology that keeps such (HyParView's active
    /// view), for the tests and the benchmark to look at. By default none.
    #[cfg(test)]
    fn active(&self) -> Option<BTreeSet<String>> {
        None
    }
}

/// The topology and the dissemination in use (`AppState::ways`), and which they are.
pub(crate) struct Ways {
    pub(crate) kinds: (Kind, crate::swarm::dissemination::Kind),
    pub(crate) topology: Arc<dyn Topology>,
    pub(crate) dissemination: Arc<dyn crate::swarm::dissemination::Dissemination>,
}

impl Ways {
    pub(crate) fn new(topology: Kind, dissemination: crate::swarm::dissemination::Kind) -> Self {
        Self {
            kinds: (topology, dissemination),
            topology: topology.make(),
            dissemination: dissemination.make(),
        }
    }
}

/// Where the swarm records its topology and its dissemination (`swarm_ways`); the
/// genesis node writes its `--topology` and `--dissemination` there
/// (`swarm::bootstrap`).
pub(crate) fn topology_path() -> Vec<String> {
    crate::document::meta::path(&[crate::document::meta::TOPOLOGY])
}

pub(crate) fn dissemination_path() -> Vec<String> {
    crate::document::meta::path(&[crate::document::meta::DISSEMINATION])
}

/// The topology and the dissemination the swarm records: its genesis node writes its
/// own (`topology_path`, `dissemination_path`), a protocol writer may change them (every
/// node then swaps to them, `refresh`). Where the document has none (a swarm from
/// before they were recorded), or one this node doesn't know, this node's own (`own`).
pub(crate) fn swarm_ways(
    doc: &crate::document::Document,
    own: (Kind, crate::swarm::dissemination::Kind),
) -> (Kind, crate::swarm::dissemination::Kind) {
    fn recorded<T: serde::de::DeserializeOwned>(
        doc: &crate::document::Document,
        path: &[String],
        what: &str,
        own: T,
    ) -> T {
        match crate::document::store::read_path(doc, path) {
            Ok(Some(value)) => match serde_json::from_value(value.clone()) {
                Ok(recorded) => recorded,
                Err(_) => {
                    warn_once(&format!(
                        "The swarm's {what} ({value}) isn't one this node knows; using --{what}"
                    ));
                    own
                }
            },
            _ => own,
        }
    }
    (
        recorded(doc, &topology_path(), "topology", own.0),
        recorded(doc, &dissemination_path(), "dissemination", own.1),
    )
}

/// Swaps to the topology and the dissemination `kinds` if they aren't the ones in use:
/// fresh, the new topology starting from our neighbours (`Topology::seed`).
fn switch_ways(state: &AppState, kinds: (Kind, crate::swarm::dissemination::Kind)) {
    let neighbours: BTreeSet<String> = our_neighbours(state).into_iter().collect();
    let mut ways = state.ways.write_or_recover();
    if ways.kinds == kinds {
        return;
    }
    let before = (ways.topology.name(), ways.dissemination.name());
    if ways.kinds.0 != kinds.0 {
        ways.topology = kinds.0.make();
        ways.topology.seed(&neighbours);
    }
    if ways.kinds.1 != kinds.1 {
        ways.dissemination = kinds.1.make();
    }
    ways.kinds = kinds;
    info!(
        from_topology = before.0,
        from_dissemination = before.1,
        topology = ways.topology.name(),
        dissemination = ways.dissemination.name(),
        "The swarm switched how neighbours are chosen, or how changes pass on"
    );
    if ways
        .topology
        .graph(&[], 2, &|_| false, &|_| false)
        .is_none()
        && kinds.1 == crate::swarm::dissemination::Kind::WritersTrees
    {
        warn_once(
            "Writers' trees need a graph every node shares, which this topology has none of: changes pass on by announcements alone (a protocol writer may record dissemination \"flood\")",
        );
    }
}

/// The topologies to choose from (`--topology`); one per swarm.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "kebab-case")]
pub(crate) enum Kind {
    /// The member list's graph (`RingAndPairs`).
    #[default]
    RingAndPairs,
    /// HyParView's active view, slimmed down (`swarm::hyparview`).
    Hyparview,
}

impl Kind {
    pub(crate) fn make(self) -> Arc<dyn Topology> {
        match self {
            Kind::RingAndPairs => Arc::new(RingAndPairs),
            Kind::Hyparview => Arc::new(crate::swarm::hyparview::HyParView::default()),
        }
    }
}

/// What a node's neighbours are chosen from (`Topology::neighbours`): the members, by
/// the transports each is on (as the member list gives their addresses), ours, the
/// neighbours each keeps, whom we judge unreachable, who is stable.
pub(crate) struct Inputs<'a> {
    pub(crate) transports: &'a BTreeMap<String, BTreeSet<String>>,
    pub(crate) local_transports: &'a BTreeSet<String>,
    pub(crate) k: usize,
    pub(crate) unreachable: &'a (dyn Fn(&str) -> bool + Sync),
    pub(crate) stable: &'a (dyn Fn(&str) -> bool + Sync),
}

/// This node's current neighbours, and the member list they were computed from.
#[derive(Debug, Default)]
pub(crate) struct NeighbourView {
    members: Vec<String>,
    k: usize,
    unreachable: BTreeSet<String>,
    stable: BTreeSet<String>,
    /// The transports (address schemes) of each member, and ours.
    transports: BTreeMap<String, BTreeSet<String>>,
    local_transports: BTreeSet<String>,
    /// Peers that proved they are of our swarm but aren't in our member list (yet): their
    /// registration hasn't reached us. They are neighbours too (`refresh`).
    unlisted: BTreeSet<String>,
    /// Members the member list gives no address for on our transports, that reach us
    /// anyway: they only connect out (no listen address). They are neighbours too while
    /// they say they sync with us (`refresh`).
    outbound: BTreeSet<String>,
    neighbours: BTreeSet<String>,
    /// The topology they were computed with, and its version (`Topology::version`).
    version: (&'static str, u64),
    /// Every member not judged unreachable is a neighbour (the tests wait for it,
    /// `is_full_mesh`). Not the swarm being a full mesh: members on a transport we are
    /// not on, or that we never reached, don't count against it.
    #[cfg(test)]
    full_mesh: bool,
    /// The graph the member list gives, as if every member reached every other
    /// (`member_list_graph`): computed when first asked for, kept while the member list
    /// stays the same. None for a topology without a shared graph.
    whole: std::sync::OnceLock<Option<Arc<BTreeMap<String, BTreeSet<String>>>>>,
    /// Where members recorded their links differ from that graph (`swarm::working_links`),
    /// those with a record.
    recorded: BTreeMap<String, members::RecordedLinks>,
    /// That graph with the links members recorded (`working_graph`): computed when first
    /// asked for, kept while the member list and the records stay the same.
    working: std::sync::OnceLock<Option<Arc<BTreeMap<String, BTreeSet<String>>>>>,
}

/// Every member's neighbours as the member list gives them, as if every member reached
/// every other: the same on every node that knows the same member list, whatever each
/// reaches (`swarm::tree` builds writers' trees on it). On each transport among the
/// members on it, all together. None until the view is known, and for a topology
/// without a shared graph (`Topology::graph`).
pub(crate) fn member_list_graph(
    state: &AppState,
) -> Option<Arc<BTreeMap<String, BTreeSet<String>>>> {
    let view = state.neighbours.read_or_recover();
    let view = view.as_ref()?;
    view.whole
        .get_or_init(|| {
            let mut whole: BTreeMap<String, BTreeSet<String>> = view
                .members
                .iter()
                .map(|member| (member.clone(), BTreeSet::new()))
                .collect();
            let transports: BTreeSet<&String> = view.transports.values().flatten().collect();
            for transport in transports {
                let on_it: Vec<String> = view
                    .transports
                    .iter()
                    .filter(|(_, on)| on.contains(transport))
                    .map(|(member, _)| member.clone())
                    .collect();
                let graph = state
                    .topology()
                    .graph(&on_it, view.k, &|_| false, &|m| view.stable.contains(m))?;
                for (member, neighbours) in graph {
                    whole.entry(member).or_default().extend(neighbours);
                }
            }
            Some(Arc::new(whole))
        })
        .clone()
}

/// The member list's graph (`member_list_graph`) with the links members recorded
/// (`swarm::working_links`): without those either end recorded as unreached, with those
/// one recorded as reached (`with_recorded`). The same on every node that has the same
/// document; what the writers' trees are built on (`swarm::tree`). None until the view
/// is known, and for a topology without a shared graph.
pub(crate) fn working_graph(state: &AppState) -> Option<Arc<BTreeMap<String, BTreeSet<String>>>> {
    let whole = member_list_graph(state)?;
    let view = state.neighbours.read_or_recover();
    let view = view.as_ref()?;
    view.working
        .get_or_init(|| {
            Some(if view.recorded.is_empty() {
                whole
            } else {
                Arc::new(with_recorded(&whole, &view.recorded))
            })
        })
        .clone()
}

/// `graph` with the links members recorded (`recorded`): those one recorded as reached
/// added, between members `graph` has; then those either end recorded as unreached
/// taken out (a link one end reaches and the other doesn't counts as down).
pub(crate) fn with_recorded(
    graph: &BTreeMap<String, BTreeSet<String>>,
    recorded: &BTreeMap<String, members::RecordedLinks>,
) -> BTreeMap<String, BTreeSet<String>> {
    let mut graph = graph.clone();
    for (member, links) in recorded {
        for other in &links.reached {
            if member != other && graph.contains_key(member) && graph.contains_key(other) {
                graph
                    .entry(member.clone())
                    .or_default()
                    .insert(other.clone());
                graph
                    .entry(other.clone())
                    .or_default()
                    .insert(member.clone());
            }
        }
    }
    for (member, links) in recorded {
        for other in &links.unreached {
            if let Some(neighbours) = graph.get_mut(member) {
                neighbours.remove(other);
            }
            if let Some(neighbours) = graph.get_mut(other) {
                neighbours.remove(member);
            }
        }
    }
    graph
}

/// Whether `peer` is in the member list, as the view was last computed from it; None
/// while the view isn't known.
pub(crate) fn listed(state: &AppState, peer: &str) -> Option<bool> {
    state
        .neighbours
        .read_or_recover()
        .as_ref()
        .map(|view| view.members.iter().any(|member| member == peer))
}

/// Whether the neighbours are computed from the member list yet (before, everyone
/// counts as a neighbour).
pub(crate) fn view_known(state: &AppState) -> bool {
    state.neighbours.read_or_recover().is_some()
}

/// Whether every reachable member is our neighbour (also while the view is unknown).
#[cfg(test)]
pub(crate) fn is_full_mesh(state: &AppState) -> bool {
    state
        .neighbours
        .read_or_recover()
        .as_ref()
        .is_none_or(|view| view.full_mesh)
}

/// The current view in numbers, once known.
#[derive(Debug, Clone)]
pub(crate) struct Summary {
    /// Registered members, us included.
    pub(crate) members: usize,
    /// Members other than us not currently judged unreachable.
    pub(crate) reachable: usize,
    pub(crate) stable: usize,
    pub(crate) neighbours: usize,
    pub(crate) stable_neighbours: usize,
    /// For each of our transports (address scheme), the neighbours the member list
    /// gives an address on it: one on two transports counts on both, one it gives none
    /// for (not listed yet, or only connecting out) on none.
    pub(crate) neighbours_by_transport: BTreeMap<String, usize>,
}

pub(crate) fn summary(state: &AppState) -> Option<Summary> {
    let local = state.local_id();
    state
        .neighbours
        .read_or_recover()
        .as_ref()
        .map(|view| Summary {
            members: view.members.len(),
            reachable: view
                .members
                .iter()
                .filter(|m| m.as_str() != local && !view.unreachable.contains(*m))
                .count(),
            stable: view.stable.len(),
            neighbours: view.neighbours.len(),
            stable_neighbours: view
                .neighbours
                .iter()
                .filter(|m| view.stable.contains(*m))
                .count(),
            neighbours_by_transport: view
                .local_transports
                .iter()
                .map(|transport| {
                    let on_it = view
                        .neighbours
                        .iter()
                        .filter(|m| {
                            view.transports
                                .get(*m)
                                .is_some_and(|theirs| theirs.contains(transport))
                        })
                        .count();
                    (transport.clone(), on_it)
                })
                .collect(),
        })
}

/// Whether `peer` is in the member list the neighbours were computed from (false until
/// the view is known).
pub(crate) fn is_listed(state: &AppState, peer: &str) -> bool {
    state
        .neighbours
        .read_or_recover()
        .as_ref()
        .is_some_and(|view| view.members.iter().any(|member| member == peer))
}

/// Whether we initiate contact with `peer`: HELLOs, keepalives, dormancy checks, sync
/// openings. Everyone is a neighbour until the view is known (while joining, or before
/// any member registered), so joining works as in a full mesh. We always *answer*
/// anyone who contacts us.
/// Our neighbours, as the view was last computed; none while it isn't known (when
/// everyone counts: `is_neighbour`).
pub(crate) fn our_neighbours(state: &AppState) -> Vec<String> {
    state
        .neighbours
        .read_or_recover()
        .as_ref()
        .map(|view| view.neighbours.iter().cloned().collect())
        .unwrap_or_default()
}

pub(crate) fn is_neighbour(state: &AppState, peer: &str) -> bool {
    match &*state.neighbours.read_or_recover() {
        Some(view) => view.neighbours.contains(peer),
        None => true,
    }
}

/// Every member's neighbours as the member list makes them, as this node sees it (whom
/// it can reach, who is stable): the graph each node computes its own part of. Ours as
/// they are (with members that only connect out, or aren't listed yet). For a map of
/// the swarm (the dashboard); None until the view is known. A link exists if either end
/// chose it: draw one if either lists the other. About `members²` work: compute it when
/// the member list may have changed, not on every look. None too for a topology without
/// a shared graph.
#[cfg_attr(not(feature = "dashboard"), allow(dead_code))]
pub(crate) fn intended_graph(state: &AppState) -> Option<BTreeMap<String, BTreeSet<String>>> {
    member_list_graph(state)?;
    let topology = state.topology();
    let view = state.neighbours.read_or_recover();
    let view = view.as_ref()?;
    let graph = view
        .members
        .iter()
        .map(|member| {
            (
                member.clone(),
                view.neighbours_of(&*topology, state.local_id(), member),
            )
        })
        .collect();
    Some(graph)
}

/// Whether we reach `peer` ourselves, as choosing neighbours judges it: on one of our
/// transports, and not unanswered (three HELLOs, or dormant: `PeerState::is_unreachable`).
/// One we have no word of yet counts as reached until it fails to answer.
pub(crate) async fn reaches(state: &AppState, peer: &str) -> bool {
    if !state.transport.can_reach(peer) {
        return false;
    }
    match state.get_peer_state(peer).await {
        Some(peer_state) => !peer_state.lock().await.is_unreachable(),
        None => true,
    }
}

/// The transports `member` is on, by the member list (the schemes of its addresses);
/// ours as we are on them. None known: empty.
pub(crate) fn transports_of(state: &AppState, member: &str) -> BTreeSet<String> {
    let view = state.neighbours.read_or_recover();
    match view.as_ref() {
        Some(view) if member == state.local_id() => view.local_transports.clone(),
        Some(view) => view.transports.get(member).cloned().unwrap_or_default(),
        None if member == state.local_id() => state.transport.transports().into_iter().collect(),
        None => BTreeSet::new(),
    }
}

impl NeighbourView {
    fn neighbours_of(
        &self,
        topology: &dyn Topology,
        local: &str,
        member: &str,
    ) -> BTreeSet<String> {
        if member == local {
            return self.neighbours.clone();
        }
        let own = self.transports.get(member).cloned().unwrap_or_default();
        neighbours_across(
            topology,
            &self.transports,
            member,
            &own,
            self.k,
            &|member| self.unreachable.contains(member),
            &|member| self.stable.contains(member),
        )
    }
}

/// How the neighbours changed.
pub(crate) struct NeighbourChange {
    pub(crate) added: Vec<String>,
    pub(crate) removed: Vec<String>,
}

/// Runs `change` on a peer's state, and has the neighbours recomputed (`run_worker`) if
/// it changed what they are chosen from (`PeerState::neighbour_inputs`). Changes to a
/// peer that may matter to the neighbours go through here.
pub(crate) fn change_peer<T>(
    state: &AppState,
    peer_state: &mut PeerState,
    change: impl FnOnce(&mut PeerState) -> T,
) -> T {
    let before = peer_state.neighbour_inputs();
    let result = change(peer_state);
    if peer_state.neighbour_inputs() != before {
        state.topology_changed.notify_one();
    }
    result
}

/// Recomputes the neighbours whenever something they depend on may have changed
/// (`AppState::topology_changed`): the member list (`swarm::changes::run_change_publisher`),
/// a peer's state (`change_peer`), and every 5 s as a fallback (`swarm::peer`). A burst
/// of notifications coalesces into one recomputation.
pub(crate) async fn run_worker(state: AppState) {
    info!(
        topology = state.topology().name(),
        dissemination = state.dissemination().name(),
        "Choosing neighbours"
    );
    loop {
        state.topology_changed.notified().await;
        if let Some(change) = refresh(&state).await {
            crate::swarm::peer::apply_neighbour_change(&state, change).await;
        }
    }
}

/// Whether a change at `path` may change the neighbours: one in the member list, or the
/// root (the whole document replaced, as after an epoch switch).
pub(crate) fn affects_neighbours(path: &[String]) -> bool {
    path.is_empty()
        || path.starts_with(&members::members_dir())
        || path.starts_with(&neighbours_path())
        || path.starts_with(&topology_path())
        || path.starts_with(&dissemination_path())
}

/// The most neighbours a swarm may ask each member to keep (`swarm_neighbours`).
const MAX_NEIGHBOURS: u64 = 256;

/// Where the swarm records its neighbours (`swarm_neighbours`); the genesis node writes
/// its `--neighbours` there (`swarm::bootstrap`).
pub(crate) fn neighbours_path() -> Vec<String> {
    crate::document::meta::path(&[crate::document::meta::NEIGHBOURS])
}

/// The neighbours each member keeps, as the swarm records it: its genesis node writes
/// its `--neighbours` (`neighbours_path`), a protocol writer may change it (every
/// node then recomputes its neighbours, and the writers' trees with them). The same on
/// every node, as the member list's graph must be (`member_list_graph`). Where the
/// document has none (a swarm from before it was recorded), or one that isn't a number
/// from 2 to `MAX_NEIGHBOURS`, this node's own (`own`).
pub(crate) fn swarm_neighbours(doc: &crate::document::Document, own: usize) -> usize {
    match crate::document::store::read_path(doc, &neighbours_path()) {
        Ok(Some(value)) => match value.as_u64() {
            Some(k) if (2..=MAX_NEIGHBOURS).contains(&k) => k as usize,
            _ => {
                warn_once(&format!(
                    "The swarm's neighbours ({value}) isn't a number from 2 to {MAX_NEIGHBOURS}; using --neighbours ({own})"
                ));
                own
            }
        },
        _ => own,
    }
}

/// Logs `message` at info level, unless it was the last one logged so.
fn info_once(message: &str) {
    static LAST: std::sync::Mutex<String> = std::sync::Mutex::new(String::new());
    let mut last = LAST.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
    if *last != message {
        info!("{message}");
        *last = message.to_owned();
    }
}

/// Logs `message` at warn level, unless it was the last one logged so.
fn warn_once(message: &str) {
    static LAST: std::sync::Mutex<String> = std::sync::Mutex::new(String::new());
    let mut last = LAST.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
    if *last != message {
        tracing::warn!("{message}");
        *last = message.to_owned();
    }
}

/// The neighbours each member keeps, as the view was last computed with
/// (`swarm_neighbours`); this node's `--neighbours` before.
pub(crate) fn k(state: &AppState) -> usize {
    state
        .neighbours
        .read_or_recover()
        .as_ref()
        .map_or(state.neighbours_k, |view| view.k)
}

/// Recomputes the neighbours once ready, if the member list, the reachable members or
/// `k` changed.
pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> {
    if !state.is_ready().await {
        return None;
    }
    let (current, stable, transports, alive, k, recorded) = {
        let replica = state.doc.read().await;
        members::apply(state, &replica.doc);
        let k = swarm_neighbours(&replica.doc, state.neighbours_k);
        switch_ways(state, swarm_ways(&replica.doc, state.configured_ways));
        let current = members::members(&replica.doc);
        // Those that said they are alive within the electorate window.
        let alive: BTreeSet<String> = members::electorate(
            &replica.doc,
            state.clock.now_millis().0,
            state.electorate_window,
        )
        .into_iter()
        .collect();
        let stable: BTreeSet<String> = current
            .iter()
            .filter(|m| members::tier(&replica.doc, m) == members::Tier::Stable)
            .cloned()
            .collect();
        let transports: BTreeMap<String, BTreeSet<String>> = current
            .iter()
            .map(|m| (m.clone(), schemes(&members::addresses(&replica.doc, m))))
            .collect();
        let recorded: BTreeMap<String, members::RecordedLinks> = current
            .iter()
            .map(|m| (m.clone(), members::links(&replica.doc, m)))
            .filter(|(_, links)| *links != members::RecordedLinks::default())
            .collect();
        (current, stable, transports, alive, k, recorded)
    };
    if k != state.neighbours_k {
        info_once(&format!(
            "The swarm keeps {k} neighbours per member (recorded by its genesis node, or a protocol writer); --neighbours ({}) counts only on a genesis node",
            state.neighbours_k
        ));
    }
    let local_transports: BTreeSet<String> = state.transport.transports().into_iter().collect();
    // Those that don't answer, those on none of our transports, and those that haven't
    // said they are alive for longer than the electorate window, unless we heard from
    // them since we started (one that comes back says so to us first: its HELLO).
    let mut unreachable = BTreeSet::new();
    for member in &current {
        if member == state.local_id() {
            continue;
        }
        if !reaches(state, member).await {
            unreachable.insert(member.clone());
            continue;
        }
        let heard = match state.get_peer_state(member).await {
            Some(peer_state) => peer_state.lock().await.last_seen.is_some(),
            None => false,
        };
        if !heard && !alive.contains(member) {
            unreachable.insert(member.clone());
        }
    }
    // A node that joined while the genesis node hadn't registered yet has the genesis
    // document without the genesis node in it, and registers itself there: then neither
    // lists the other, and with each the other's neighbour only by the member list,
    // neither would ever sync with the other, and their registrations never pass. So a
    // peer that proved itself a member of our swarm counts until the list says otherwise:
    // syncing brings each the other's registration.
    let mut unlisted = BTreeSet::new();
    for (peer, peer_state) in state.peers_snapshot().await {
        if current.contains(&peer) || peer == state.local_id() {
            continue;
        }
        let peer_state = peer_state.lock().await;
        if matches!(peer_state.standing, crate::state::Standing::Member { .. })
            && !peer_state.is_unreachable()
        {
            unlisted.insert(peer);
        }
    }
    // A member that lists no address on our transports, but reached us anyway, only
    // connects out: it isn't in the graph on our side (it's on none of our transports, as
    // far as the list goes), while on its side we are, so it would sync with us while we
    // never sent it our changes (only answers to its). It picks its neighbours; those it
    // picked keep it, as it says (STATUS).
    let mut outbound = BTreeSet::new();
    for member in &current {
        if member == state.local_id()
            || unreachable.contains(member)
            || transports
                .get(member)
                .is_some_and(|on| !on.is_disjoint(&local_transports))
        {
            continue;
        }
        if let Some(peer_state) = state.get_peer_state(member).await {
            let peer_state = peer_state.lock().await;
            if matches!(peer_state.standing, crate::state::Standing::Member { .. })
                && peer_state.syncs_with_us != Some(false)
            {
                outbound.insert(member.clone());
            }
        }
    }
    let unreachable_of = |m: &str| unreachable.contains(m);
    let stable_of = |m: &str| stable.contains(m);
    let inputs = Inputs {
        transports: &transports,
        local_transports: &local_transports,
        k,
        unreachable: &unreachable_of,
        stable: &stable_of,
    };
    let topology = state.topology();
    topology.maintain(state, &inputs).await;
    let version = (topology.name(), topology.version());
    let previous = {
        let view = state.neighbours.read_or_recover();
        if view.as_ref().is_some_and(|view| {
            view.version == version
                && view.members == current
                && view.unlisted == unlisted
                && view.outbound == outbound
                && view.k == k
                && view.unreachable == unreachable
                && view.stable == stable
                && view.transports == transports
                && view.local_transports == local_transports
                && view.recorded == recorded
        }) {
            return None;
        }
        view.as_ref().map(|view| {
            // The member list's graph stays while the member list does; with the
            // records, while they do too.
            let same_list = view.members == current
                && view.k == k
                && view.stable == stable
                && view.transports == transports;
            let whole = if same_list {
                view.whole.clone()
            } else {
                Default::default()
            };
            let working = if same_list && view.recorded == recorded {
                view.working.clone()
            } else {
                Default::default()
            };
            (view.neighbours.clone(), whole, working)
        })
    };
    let (previous, whole, working) = match previous {
        Some((previous, whole, working)) => (Some(previous), whole, working),
        None => (None, Default::default(), Default::default()),
    };

    // Before the first view everyone known counted as a neighbour.
    let previous = match previous {
        Some(previous) => previous,
        None => state.peer_states.read().await.keys().cloned().collect(),
    };
    let mut computed = topology.neighbours(state.local_id(), &inputs);
    computed.extend(outbound.iter().cloned());
    #[cfg(test)]
    let listed_neighbours = computed.len();
    computed.extend(unlisted.iter().cloned());
    let change = NeighbourChange {
        added: computed
            .iter()
            .filter(|m| !previous.contains(*m))
            .cloned()
            .collect(),
        removed: previous
            .iter()
            .filter(|m| !computed.contains(*m))
            .cloned()
            .collect(),
    };
    info!(
        members = current.len(),
        unreachable = unreachable.len(),
        stable = stable.len(),
        k,
        neighbours = computed.len(),
        unlisted = unlisted.len(),
        outbound = outbound.len(),
        added = change.added.len(),
        removed = change.removed.len(),
        "Neighbours computed from the member list"
    );
    #[cfg(test)]
    let reachable_others = current
        .iter()
        .filter(|m| m.as_str() != state.local_id() && !unreachable.contains(*m))
        .count();
    *state.neighbours.write_or_recover() = Some(NeighbourView {
        #[cfg(test)]
        full_mesh: listed_neighbours >= reachable_others,
        members: current,
        k,
        unreachable,
        stable,
        transports,
        local_transports,
        unlisted,
        outbound,
        neighbours: computed,
        version,
        whole,
        recorded,
        working,
    });
    // New neighbours are kept alive and judged.
    state.peers_due.now();
    Some(change)
}

#[cfg(test)]
mod tests;