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
|
//! 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.
//!
//! In a full mesh the writer reaches everyone itself, so nothing is announced.
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);
/// After changes from `from` were applied and our heads changed: tell our other
/// neighbours.
pub(crate) async fn announce(state: &AppState, from: &str) {
if !state.initial_sync_done.load(Ordering::Acquire) || topology::is_full_mesh(state) {
return;
}
let neighbours: Vec<String> = state
.peer_states
.read()
.await
.keys()
.filter(|peer| peer.as_str() != from && 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;
}
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;
}
}
|
