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
|
//! This node's links, as it measures them, for the dashboard's map (`own_links`): its
//! own in the state, and to any page that subscribes to them
//! (`GET /admin/dashboard/links/own/events`, also from another member's page).
use crate::state::AppState;
use crate::swarm::topology;
use crate::transport::paths::{self, ClockOffset};
use crate::transport::{Connection, Destination};
/// Links listed at most (a node has about `--neighbours`, more if others prefer it).
const MAX_LINKS: usize = 256;
/// Other transports in a link at most.
const MAX_OTHERS: usize = 8;
/// One of this node's links.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct Link {
/// The peer's node ID.
pub(crate) peer: String,
/// The transport its messages go over now (the scheme of the path in use), if any.
pub(crate) transport: Option<String>,
/// Only through a relay (libp2p).
pub(crate) relayed: bool,
/// The path in use is down: none to the peer answers (`paths`), whatever its
/// connection was.
pub(crate) down: bool,
/// The round trip on that path, in microseconds, if measured.
pub(crate) rtt_us: Option<u32>,
/// Silent for long (`peers.md`, dormancy).
pub(crate) dormant: bool,
/// The peer's other transports (`MAX_OTHERS` at most): where else it proved
/// itself, those paths up.
pub(crate) others: Vec<OtherPath>,
/// How far the peer's clock is ahead of this node's (`paths::ClockOffset`), if
/// measured: with other members', the page shows whose clock is off.
pub(crate) clock: Option<ClockOffset>,
}
/// A transport of a link besides the one in use.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct OtherPath {
/// Its scheme (`mycelium`, `libp2p`, `tor`).
pub(crate) transport: String,
/// The round trip of its fastest path, in microseconds, if any is measured.
pub(crate) rtt_us: Option<u32>,
/// Only through a relay (libp2p): no direct connection at any of its paths.
pub(crate) relayed: bool,
}
/// This node's links: its neighbours, and the peers it holds dormant, as it measures
/// them. A neighbour that goes dormant is one no more (it is unreachable now), but its
/// link stays listed, dormant, until it is heard from again: left out, the link would
/// show only as that peer last reported it, from before.
pub(crate) async fn own_links(state: &AppState) -> Vec<Link> {
let mut links = Vec::new();
for (peer, peer_state) in state.peers_snapshot().await {
let dormant = peer_state.lock().await.liveness.is_dormant();
if !dormant && !topology::is_neighbour(state, &peer) {
continue;
}
let destination = state.transport.destination(&peer);
let transport = destination
.as_ref()
.and_then(|destination| scheme_of(&destination.address));
links.push(Link {
others: others(destination.as_ref(), transport.as_deref()),
transport,
relayed: destination
.as_ref()
.is_some_and(|destination| destination.connection == Some(Connection::Relayed)),
// Ranked up before down: the first is down only if all are.
down: destination
.as_ref()
.and_then(|destination| destination.paths.first())
.is_some_and(|path| path.down),
// The path in use's (`Transport::rtt`, the first of `destination.paths`).
rtt_us: state
.transport
.rtt(&peer)
.map(|rtt| u32::try_from(rtt.as_micros()).unwrap_or(u32::MAX)),
dormant,
clock: state.transport.clock_offset(&peer),
peer,
});
}
links.sort_by(|a, b| a.peer.cmp(&b.peer));
links.truncate(MAX_LINKS);
links
}
/// The scheme of `address` (`mycelium:…`): its transport (`paths::scheme_of`), if it
/// has one.
fn scheme_of(address: &str) -> Option<String> {
Some(paths::scheme_of(address))
.filter(|scheme| !scheme.is_empty())
.map(str::to_owned)
}
/// The transports besides `used` the peer proved itself at, with a path up: each with
/// the round trip of its fastest path, if any is measured, and whether all of them go
/// through a relay.
fn others(destination: Option<&Destination>, used: Option<&str>) -> Vec<OtherPath> {
let mut others: Vec<OtherPath> = Vec::new();
for path in destination.map_or(&[][..], |destination| &destination.paths[..]) {
let Some(transport) = scheme_of(&path.address) else {
continue;
};
if path.down || used == Some(transport.as_str()) {
continue;
}
let rtt_us = path
.rtt_ms
.filter(|ms| ms.is_finite() && *ms >= 0.0)
.map(|ms| (ms * 1000.0).min(f64::from(u32::MAX)) as u32);
let relayed = path.connection == Some(Connection::Relayed);
match others.iter_mut().find(|other| other.transport == transport) {
Some(other) => {
other.rtt_us = match (other.rtt_us, rtt_us) {
(Some(a), Some(b)) => Some(a.min(b)),
(a, b) => a.or(b),
};
other.relayed &= relayed;
}
None => others.push(OtherPath {
transport,
rtt_us,
relayed,
}),
}
}
others.sort_by(|a, b| a.transport.cmp(&b.transport));
others.truncate(MAX_OTHERS);
others
}
/// Whether `new` says something `old` didn't: another peer, transport, relay or
/// dormancy, a round trip measured or moved by more than a fifth (and 1 ms), or a clock
/// measured or moved beyond both bounds. Round trips and clock samples wobble; small
/// moves aren't news.
pub(crate) fn changed(old: &[Link], new: &[Link]) -> bool {
let clock_moved = |a: Option<ClockOffset>, b: Option<ClockOffset>| match (a, b) {
(Some(a), Some(b)) => {
a.offset_ms.abs_diff(b.offset_ms) > a.within_ms.saturating_add(b.within_ms)
}
(a, b) => a.is_some() != b.is_some(),
};
let moved = |a: Option<u32>, b: Option<u32>| match (a, b) {
(Some(a), Some(b)) => {
let by = a.abs_diff(b);
by > 1000 && by > a.max(b) / 5
}
(a, b) => a.is_some() != b.is_some(),
};
old.len() != new.len()
|| old.iter().zip(new).any(|(a, b)| {
a.peer != b.peer
|| a.transport != b.transport
|| a.relayed != b.relayed
|| a.down != b.down
|| a.dormant != b.dormant
|| moved(a.rtt_us, b.rtt_us)
|| clock_moved(a.clock, b.clock)
|| a.others.len() != b.others.len()
|| a.others.iter().zip(&b.others).any(|(x, y)| {
x.transport != y.transport
|| x.relayed != y.relayed
|| moved(x.rtt_us, y.rtt_us)
})
})
}
#[cfg(test)]
mod tests {
use super::*;
fn link(peer: &str, rtt_us: Option<u32>) -> Link {
Link {
peer: peer.into(),
transport: Some("libp2p".into()),
relayed: false,
down: false,
rtt_us,
dormant: false,
others: Vec::new(),
clock: None,
}
}
#[test]
fn only_news_is_a_change() {
let old = [link("a", Some(10_000))];
assert!(!changed(&old, &[link("a", Some(11_000))]));
assert!(changed(&old, &[link("a", Some(13_000))]));
assert!(changed(&old, &[link("a", None)]));
assert!(changed(&old, &[link("b", Some(10_000))]));
assert!(changed(&old, &[]));
let mut dormant = link("a", Some(10_000));
dormant.dormant = true;
assert!(changed(&old, &[dormant]));
let mut down = link("a", Some(10_000));
down.down = true;
assert!(changed(&old, &[down]));
// Small round trips: a millisecond's wobble isn't news.
assert!(!changed(&[link("a", Some(400))], &[link("a", Some(1_300))]));
// A clock: news when measured, and when it moved beyond both bounds.
let clocked = |offset_ms| Link {
clock: Some(ClockOffset {
offset_ms,
within_ms: 50,
}),
..link("a", Some(10_000))
};
assert!(changed(&old, &[clocked(0)]));
assert!(!changed(&[clocked(0)], &[clocked(80)]));
assert!(changed(&[clocked(0)], &[clocked(5_000)]));
}
}
|
