diff options
| -rw-r--r-- | src/state.rs | 17 | ||||
| -rw-r--r-- | src/swarm/mod.rs | 1 | ||||
| -rw-r--r-- | src/swarm/peer.rs | 8 | ||||
| -rw-r--r-- | src/swarm/sync.rs | 111 |
4 files changed, 28 insertions, 109 deletions
diff --git a/src/state.rs b/src/state.rs index 72013b9..8ff8b9d 100644 --- a/src/state.rs +++ b/src/state.rs @@ -478,6 +478,8 @@ pub(crate) struct Timing { pub(crate) dormant_after: Duration, pub(crate) resync_after: Duration, pub(crate) dormant_probe: Duration, + /// While catching up: PING the source after this much silence (`swarm::source`). + pub(crate) catch_up_probe: Duration, } impl Timing { @@ -494,6 +496,9 @@ impl Timing { dormant_after: keepalive * 4, // dormant_after >= 4* keepalive to tolerate three lost PINGs in a row resync_after: keepalive * 2, // resync_after should be >= 2* keepalive to guarantee at least one digest exchange while sync is idle dormant_probe: keepalive * 10, // >= 10*: dormant peers cost almost nothing, but still rejoin within minutes + // Seconds, not a keepalive interval: a dead source should cost little time. + // Only the source gets these, and only while catching up. + catch_up_probe: (keepalive / 30).clamp(Duration::from_secs(1), Duration::from_secs(5)), } } } @@ -541,12 +546,16 @@ pub(crate) struct AppState { pub(crate) catch_up_source: Arc<std::sync::Mutex<Option<CatchUpSource>>>, } -/// The peer a catching-up node syncs with first (`swarm::sync::defer_for_catch_up`). +/// The peer a catching-up node syncs with first (`swarm::source`). #[derive(Clone, Debug)] pub(crate) struct CatchUpSource { pub(crate) peer: String, - /// When it was chosen or last sent us something. - pub(crate) last_progress: Instant, + /// When it was chosen or last sent us anything. + pub(crate) last_heard: Instant, + /// Our last probe PING to it. + pub(crate) last_probe: Option<Instant>, + /// Its keepalives in a row saying it wouldn't sync with us. + pub(crate) refusals: u32, } impl AppState { @@ -786,6 +795,8 @@ impl AppState { let mut announce = Vec::new(); for (id, peer_state_mutex) in &peers { let mut peer_state = peer_state_mutex.lock().await; + // What it said about syncing with us was about our old epoch. + peer_state.syncs_with_us = None; let syncs_with_us = peer_state.liveness == Liveness::Active && matches!(peer_state.standing, Standing::Member { .. }) && peer_state.epoch == Some(local); diff --git a/src/swarm/mod.rs b/src/swarm/mod.rs index 7b53afc..71e77ad 100644 --- a/src/swarm/mod.rs +++ b/src/swarm/mod.rs @@ -5,4 +5,5 @@ pub(crate) mod auth; pub(crate) mod bootstrap; pub(crate) mod pair; pub(crate) mod peer; +pub(crate) mod source; pub(crate) mod sync; diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs index f98bf17..9fe9b07 100644 --- a/src/swarm/peer.rs +++ b/src/swarm/peer.rs @@ -132,6 +132,7 @@ async fn note_keepalive(state: &AppState, peer: &str, remote: &transport::Keepal peer_state.reports_ready = Some(remote.ready); peer_state.syncs_with_us = Some(remote.syncs_with_you); } + crate::swarm::source::reported(state, peer, remote.ready, Some(remote.syncs_with_you)); } async fn heal_if_diverged( @@ -442,7 +443,9 @@ pub(crate) async fn run_peer_worker(state: AppState) { ); } - if ping_due { + let probe_due = + !ping_due && crate::swarm::source::probe_due(&state, peer, Instant::now()); + if ping_due || probe_due { let local = local_keepalive(&state, peer).await; if let Err(error) = state .transport @@ -552,6 +555,8 @@ async fn observe_peer_epoch(state: &AppState, peer: &str, remote: Epoch) -> bool // (resetting both sides again) while this fresh sync is under way. peer_state.reset_sync(SessionCause::PeerEpochChanged); 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 }; @@ -685,6 +690,7 @@ pub(crate) async fn handle_message( .lock() .await .mark_seen(Instant::now()); + crate::swarm::source::heard_from(state, &peer); if reactivated { info!(peer = %peer, "Dormant peer is active again; re-handshaking"); diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index f1715a6..04a174c 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -12,8 +12,9 @@ use tokio::sync::mpsc; use tokio::time::Instant; use tracing::{debug, info, trace, warn}; -use crate::state::{AppState, CatchUpSource, Liveness, PeerState, PendingOutbound, Standing}; +use crate::state::{AppState, Liveness, PeerState, PendingOutbound, Standing}; use crate::swarm::peer::{peer_swarm, send_hello_once}; +use crate::swarm::source; use crate::transport::{self, ParsedMessage}; const INITIAL_SYNC_RETRY: Duration = Duration::from_millis(500); @@ -34,98 +35,6 @@ pub(crate) struct ChangeSet { pub(crate) paths: Vec<Vec<String>>, } -/// A catch-up source that sent nothing for this long is given up for the next peer. -const CATCH_UP_STALL: Duration = Duration::from_secs(30); - -/// While catching up (joining, or rejoining a newer epoch), sync with one peer only. -/// -/// Every peer would otherwise send us what we lack, all at once: a node with an empty -/// document gets the whole document from each of its peers, and a node that is merely -/// behind gets the missing changes from each. Automerge can't deduplicate across peers. -/// So the first peer we would sync with becomes the source; the others wait until the -/// initial sync is done, and then only exchange heads. If the source sends nothing for -/// `CATCH_UP_STALL`, the next peer takes over. -/// -/// Only a peer that is ready can be the source; one that said it isn't (another node -/// still catching up) waits like the rest, until it says otherwise. -/// -/// Returns how long to hold back sync with `peer`, if at all. Messages from waiting -/// peers are still applied; we just don't answer them yet. -fn defer_for_catch_up( - state: &AppState, - peer: &str, - reports_ready: Option<bool>, -) -> Option<Duration> { - if state.initial_sync_done.load(Ordering::Acquire) { - return None; - } - if reports_ready == Some(false) { - // Its next message re-evaluates this; the timer is only a fallback. - return Some(CATCH_UP_STALL); - } - let now = Instant::now(); - let mut source = state - .catch_up_source - .lock() - .unwrap_or_else(|poisoned| poisoned.into_inner()); - let stalled = match &*source { - Some(current) if current.peer == peer => return None, - Some(current) if now < current.last_progress + CATCH_UP_STALL => { - return Some(current.last_progress + CATCH_UP_STALL - now); - } - Some(current) => Some(current.peer.clone()), - None => None, - }; - match stalled { - Some(previous) => info!( - peer = %peer, - previous = %previous, - stalled_secs = CATCH_UP_STALL.as_secs(), - "Catch-up source sent nothing; catching up from another peer" - ), - None => debug!(peer = %peer, "Catching up from this peer first"), - } - *source = Some(CatchUpSource { - peer: peer.to_owned(), - last_progress: now, - }); - None -} - -/// A sync message from `peer` was applied. From the catch-up source, that's progress, -/// unless the source turns out not to be ready: then it's dropped, and every peer is -/// enqueued so the next ready one takes over at once. -fn note_catch_up_progress(state: &AppState, peer: &str, ready: bool) { - let dropped = { - let mut source = state - .catch_up_source - .lock() - .unwrap_or_else(|poisoned| poisoned.into_inner()); - match source.as_mut().filter(|current| current.peer == peer) { - Some(current) if ready => { - current.last_progress = Instant::now(); - false - } - Some(_) => { - *source = None; - true - } - None => false, - } - }; - if dropped { - debug!(peer = %peer, "Catch-up source isn't ready itself; catching up from another peer"); - let tx = state.sync_events.clone(); - let state = state.clone(); - tokio::spawn(async move { - let peers: Vec<String> = state.peer_states.read().await.keys().cloned().collect(); - for other in peers { - let _ = tx.send(SyncEvent::PeerNeedsSync(other)); - } - }); - } -} - async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool) -> Result<()> { let Some(peer_state_mutex) = state.get_peer_state(peer).await else { trace!("Peer was forgotten; skipping synchronization"); @@ -155,13 +64,13 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool }; if !retry_delivery { - if let Some(wait) = defer_for_catch_up(state, peer, peer_state.reports_ready) { + if let Some(wait) = source::defer(state, peer, &peer_state) { trace!( wait_ms = wait.as_millis(), "Catching up from another peer first; not syncing with this one yet" ); // Done earlier when the initial sync completes (every peer is enqueued - // then); this only matters if the source stalls. + // then); this only matters if the source goes quiet. let tx = state.sync_events.clone(); let peer = peer.to_owned(); tokio::spawn(async move { @@ -484,19 +393,11 @@ pub(crate) async fn handle_sync( return Ok(()); }; - note_catch_up_progress(state, peer, from_ready); + source::reported(state, peer, from_ready, None); if caught_up && !state.initial_sync_done.swap(true, Ordering::AcqRel) { info!(peer = %peer, "Initial sync complete"); - *state - .catch_up_source - .lock() - .unwrap_or_else(|poisoned| poisoned.into_inner()) = None; - // The peers we held back: now they only need to learn our heads. - let peers: Vec<String> = state.peer_states.read().await.keys().cloned().collect(); - for other in peers.into_iter().filter(|other| other != peer) { - let _ = state.sync_events.send(SyncEvent::PeerNeedsSync(other)); - } + source::finished(state, peer); } state.notify_store_changed(); |