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
|
//! Keepalives (STATUS / STATUS_REPLY) and the self-healing re-handshake when they show our
//! documents differ while sync is idle.
use std::sync::atomic::Ordering;
use std::time::Duration;
use anyhow::Result;
use tokio::time::Instant;
use tracing::warn;
use super::handshake::ensure_handshake;
use crate::state::{AppState, Liveness, Standing};
use crate::swarm::sync::SyncEvent;
use crate::swarm::topology;
use crate::transport::{self, ParsedMessage};
/// Sends `peer` a STATUS with our status, noting the time for its round trip.
pub(crate) async fn send_status(state: &AppState, peer: &str) -> Result<()> {
let status = local_status(state, peer).await;
state.lock_rtt().status_sent(peer, Instant::now());
state
.transport
.send_message(peer, &ParsedMessage::Status(status))
.await
}
/// How long shutting down waits for the last STATUSes (`farewell`).
const FAREWELL_TIMEOUT: Duration = Duration::from_secs(2);
/// At shutdown: one last STATUS to each neighbour, saying we aren't ready. A neighbour
/// catching up from us picks another source now, instead of once we go dormant. Bounded
/// in time: our transport may be gone already.
pub(crate) async fn farewell(state: &AppState) {
state.stopping.store(true, Ordering::Release);
let mut statuses = tokio::task::JoinSet::new();
for (peer, _) in state.peers_snapshot().await {
if topology::is_neighbour(state, &peer) {
let state = state.clone();
statuses.spawn(async move { send_status(&state, &peer).await });
}
}
let all_sent = async { while statuses.join_next().await.is_some() {} };
let _ = tokio::time::timeout(FAREWELL_TIMEOUT, all_sent).await;
}
/// Our status for `peer`.
pub(crate) async fn local_status(state: &AppState, peer: &str) -> transport::Status {
// Heads and epoch are published together (`Replica::stamps`), so they always match,
// and reading them doesn't wait for a sync message being applied.
let (heads, epoch, epoch_seq) = {
let stamp = state.stamp.borrow();
(
transport::HeadsSummary::of(stamp.heads.to_vec()),
stamp.epoch,
stamp.epoch_seq,
)
};
let syncs_with_you = match state.get_peer_state(peer).await {
Some(peer_state) => {
let peer_state = peer_state.lock().await;
peer_state.liveness == Liveness::Active
&& matches!(peer_state.standing, Standing::Member { .. })
&& peer_state.epoch == Some(epoch)
}
None => false,
};
transport::Status {
heads,
epoch,
epoch_seq,
ready: state.is_ready().await && !state.stopping.load(Ordering::Acquire),
syncs_with_you,
}
}
/// What a peer's status says about it.
pub(super) async fn note_status(state: &AppState, peer: &str, remote: &transport::Status) {
if let Some(peer_state) = state.get_peer_state(peer).await {
let mut peer_state = peer_state.lock().await;
let became_ready = remote.ready && peer_state.reports_ready != Some(true);
peer_state.reports_ready = Some(remote.ready);
peer_state.syncs_with_us = Some(remote.syncs_with_you);
peer_state.reported_heads = remote.heads.heads().map(<[_]>::len);
drop(peer_state);
// Off the consumer: telling which heads we lack takes the document's lock.
{
let (state, peer, remote) = (state.clone(), peer.to_owned(), remote.clone());
tokio::spawn(async move {
crate::swarm::relay::pull_if_behind(&state, &peer, &remote).await;
});
}
if became_ready {
// We may have left the opening to it while it wasn't ready, and nothing else
// would make us look again.
let _ = state
.sync_events
.send(SyncEvent::PeerNeedsSync(peer.to_owned()));
}
}
crate::swarm::source::reported(state, peer, remote.ready, Some(remote.syncs_with_you));
}
pub(super) async fn heal_if_diverged(
state: &AppState,
peer: &str,
local: &transport::Status,
remote: &transport::Status,
) {
// Heads of documents from different epochs are never equal and never converge.
// And a peer that isn't our neighbour isn't ours to keep in step.
if local.epoch != remote.epoch || !topology::is_neighbour(state, peer) {
return;
}
let Some(peer_state_mutex) = state.get_peer_state(peer).await else {
return;
};
let heal = {
let mut peer_state = peer_state_mutex.lock().await;
if local.heads == remote.heads {
// Same document: whatever answer we were waiting for, there is nothing left
// to send. (Automerge answers only when it has something to say, so a sync
// message to a peer that already has everything gets no reply at all.)
peer_state.sync.reply_settled();
}
// While we catch up, we hold back sync with everyone but the source on purpose:
// differing heads and idle sync are expected then, not a reason to heal.
let catching_up = !state.initial_sync_done.load(Ordering::Acquire);
!catching_up
&& peer_state.needs_heal(&local.heads, &remote.heads, &state.timing, Instant::now())
};
if heal {
warn!(peer = %peer, "Documents differ and sync is idle; re-handshaking to restart sync");
ensure_handshake(state, peer).await;
}
}
|
