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
|
//! The epochs peers advertise: recorded per peer, late messages aside, and compared
//! with ours.
use tokio::time::Instant;
use tracing::{debug, info, warn};
use crate::document::replica::{Epoch, EpochRelation};
use crate::state::AppState;
use crate::swarm::sync::SyncEvent;
use crate::transport;
/// Records the epoch a peer advertised and compares it with ours. Returns true when the
/// message's epoch is ours, the only case in which its documents may be synced.
///
/// `seq` is the peer's `epoch_seq` with it. A message with a lower one than we have was
/// sent before the peer's last epoch change and delivered late: its epoch isn't
/// recorded, so it can't flip our view of the peer back. The count starts over when the
/// peer restarts, which its HELLO or HELLO_REPLY tells by a new run (`run`).
pub(super) async fn observe_peer_epoch(
state: &AppState,
peer: &str,
remote: Epoch,
seq: u32,
run: Option<transport::RunId>,
) -> bool {
let local = state.local_epoch();
let relation = local.relation_to(&remote);
let Some(peer_state_mutex) = state.get_peer_state(peer).await else {
return relation == EpochRelation::Same;
};
let previous = {
let mut peer_state = peer_state_mutex.lock().await;
if let Some(run) = run.filter(|run| peer_state.run_id != Some(*run)) {
peer_state.run_id = Some(run);
peer_state.epoch_seq = None;
}
if peer_state.epoch_seq.is_some_and(|newest| seq < newest) {
debug!(
peer = %peer,
epoch = %remote,
seq,
newest = ?peer_state.epoch_seq,
"Ignoring a late message's epoch"
);
return relation == EpochRelation::Same;
}
peer_state.epoch_seq = Some(seq);
let previous = peer_state.epoch.replace(remote);
if previous.is_some_and(|previous| previous != remote) {
// The peer switched documents; our sync state describes its old one. Count
// the restart as sync activity, so the keepalive heal doesn't re-handshake
// (resetting both sides again) while this fresh sync is under way.
peer_state.reset_sync();
peer_state.last_sync_rx = Some(Instant::now());
// Whether it would sync with us was said about its old epoch.
peer_state.syncs_with_us = None;
}
previous
};
// Log transitions only, not every keepalive.
if previous != Some(remote) {
match relation {
EpochRelation::Same => {
if previous.is_some() {
info!(peer = %peer, epoch = %remote, "Peer is on our epoch again; resuming sync");
let _ = state
.sync_events
.send(SyncEvent::PeerNeedsSync(peer.to_owned()));
}
}
// Expected after a remake: either side switches or catches up.
EpochRelation::PeerAhead | EpochRelation::PeerBehind => {
info!(
peer = %peer,
local_epoch = %local,
peer_epoch = %remote,
relation = ?relation,
"Peer is on another epoch; not syncing with it until both are on the same"
);
}
EpochRelation::Diverged => {
warn!(
peer = %peer,
local_epoch = %local,
peer_epoch = %remote,
"Peer is on a different epoch of the same generation; not syncing with it"
);
}
}
}
relation == EpochRelation::Same
}
|
