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
|
//! Passing changes on in a partial mesh (`swarm::topology`).
//!
//! A node pushes its own changes to its neighbours, but nothing it receives goes any
//! further by itself. So after applying changes from one neighbour, a node *announces*
//! its new heads to its other neighbours with a PING (a keepalive: heads, about 100
//! bytes, not the changes). A neighbour that sees heads it lacks *pulls*: it enters the
//! announced heads as the peer's heads in its sync state, Automerge asks for what's
//! missing (`need`), and the announcer answers with exactly those changes. Each node
//! receives a change about once; the rest is announcements. Regular keepalives work the
//! same way, so any gap is noticed within a keepalive interval.
//!
//! Announcements are coalesced: the first applied change schedules one announcement
//! round `ANNOUNCE_DELAY` later, and every change applied until then rides along. A
//! round costs one PING and one PONG per neighbour (plus the pulls), whatever the
//! write rate; the delay is the price per hop. Without coalescing, every applied sync
//! message would announce to every neighbour, and announcements would outnumber the
//! data they announce (about 15 to 1 at 250 nodes).
//!
//! In a full mesh the writer reaches everyone itself, so nothing is announced.
use std::collections::HashSet;
use std::sync::atomic::Ordering;
use std::time::Duration;
use tokio::time::Instant;
use tracing::debug;
use crate::state::{AppState, Standing};
use crate::swarm::peer::send_ping;
use crate::swarm::sync::SyncEvent;
use crate::swarm::topology;
use crate::transport::{self, HeadsSummary};
/// A pull that brought nothing for this long no longer blocks pulling from another peer.
const PULL_TIMEOUT: Duration = Duration::from_secs(15);
/// How long an announcement waits for further changes to ride along. Added once per
/// hop to propagation; well below a round trip over Mycelium.
const ANNOUNCE_DELAY: Duration = Duration::from_millis(200);
/// After changes from `from` were applied and our heads changed: tell our other
/// neighbours, in the next announcement round.
pub(crate) async fn announce(state: &AppState, from: &str) {
if !state.initial_sync_done.load(Ordering::Acquire) || topology::is_full_mesh(state) {
return;
}
let schedule = {
let mut pending = state
.announce_pending
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
match pending.as_mut() {
Some(senders) => {
senders.insert(from.to_owned());
false
}
None => {
*pending = Some(HashSet::from([from.to_owned()]));
true
}
}
};
if !schedule {
return;
}
let state = state.clone();
tokio::spawn(async move {
tokio::time::sleep(ANNOUNCE_DELAY).await;
let senders = state
.announce_pending
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.take()
.unwrap_or_default();
announce_now(&state, &senders).await;
});
}
/// One announcement round. The peer whose changes we announce has them already, so a
/// round with a single sender skips it; with several, each lacks the others' changes,
/// so every neighbour is told.
async fn announce_now(state: &AppState, senders: &HashSet<String>) {
let skip = match senders.len() {
1 => senders.iter().next().cloned(),
_ => None,
};
let neighbours: Vec<String> = state
.peer_states
.read()
.await
.keys()
.filter(|peer| {
skip.as_deref() != Some(peer.as_str()) && topology::is_neighbour(state, peer)
})
.cloned()
.collect();
for peer in neighbours {
if let Err(error) = send_ping(state, &peer).await {
debug!(peer = %peer, error = %format!("{error:#}"), "Announcing our heads failed");
}
}
}
/// A keepalive from `peer`: if it has heads we lack, pull them (one peer at a time).
pub(crate) async fn pull_if_behind(state: &AppState, peer: &str, remote: &transport::Keepalive) {
if remote.epoch != state.local_epoch() || !topology::is_neighbour(state, peer) {
return;
}
// The common case, the same heads, needs only the cached heads (read lock).
if HeadsSummary::of(state.doc.read().await.heads().to_vec()) == remote.heads {
return;
}
let missing = match &remote.heads {
HeadsSummary::Heads(heads) => {
// AutoCommit's lookups take &mut self.
let mut replica = state.doc.write().await;
let missing = heads
.iter()
.any(|head| replica.doc.get_change_by_hash(head).is_none());
if !missing {
return;
}
Some(heads.clone())
}
// Can't tell which heads are missing; the digest differs from ours, though.
HeadsSummary::Digest(_) => None,
};
let now = Instant::now();
{
let mut pulling = state
.pulling
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
let busy = matches!(
&*pulling,
Some((other, since)) if other != peer && now < *since + PULL_TIMEOUT
);
if busy {
return;
}
*pulling = Some((peer.to_owned(), now));
}
let Some(peer_state) = state.get_peer_state(peer).await else {
return;
};
{
let mut peer_state = peer_state.lock().await;
if !matches!(peer_state.standing, Standing::Member { .. }) {
return;
}
match missing {
Some(heads) => {
// Automerge asks for whatever these heads need that we don't have.
peer_state.sync.state.their_heads = Some(heads);
peer_state.sync.state.in_flight = false;
}
// A fresh session's opening carries our heads and bloom filter; the answer
// brings what we lack.
None => peer_state.reset_sync(),
}
}
debug!(peer = %peer, "Neighbour has changes we lack; pulling");
let _ = state
.sync_events
.send(SyncEvent::PeerNeedsSync(peer.to_owned()));
}
/// A sync message from `peer` was applied: a pull from it is answered.
pub(crate) fn pulled(state: &AppState, peer: &str) {
let mut pulling = state
.pulling
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
if pulling.as_ref().is_some_and(|(p, _)| p == peer) {
*pulling = None;
}
}
|
