enrich keepalive content

Add ready and syncs_with_you flags as well as actual heads, so the peer
- ready: same meaning as in the sync message
- syncs_with_you: sender counts recipient as member and in same epoch
- heads: allows to tell who's behind or if both sides have changes

1 files changed, 169 insertions, 0 deletions

diff --git a/src/swarm/source.rs b/src/swarm/source.rs
new file mode 100644
index 0000000..6be62b7
--- /dev/null
+++ b/src/swarm/source.rs
@@ -0,0 +1,169 @@
+//! The catch-up source: while a node catches up (joining, or rejoining a newer epoch,
+//! until its initial sync is done), it syncs 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 suitable peer we would sync with becomes the source; the others wait
+//! until the initial sync is done, and then only exchange heads. Messages from waiting
+//! peers are still applied; we just don't answer them yet.
+//!
+//! A source must be ready and willing to sync with us, as far as we know (its sync
+//! messages and keepalives say so). It counts as alive while any message from it
+//! arrives, so a slow transfer of a large document never looks like a stall. When it
+//! is quiet for `Timing::catch_up_probe`, we PING it; after three unanswered probes, or
+//! when it says it isn't ready, or twice in a row that it wouldn't sync with us (the
+//! first time may be a handshake still under way), the next peer takes over.
+
+use std::sync::atomic::Ordering;
+use std::time::Duration;
+
+use tokio::time::Instant;
+use tracing::{debug, info};
+
+use crate::state::{AppState, CatchUpSource, PeerState};
+use crate::swarm::sync::SyncEvent;
+
+/// Probes without an answer before the source is given up.
+const UNANSWERED_PROBES: u32 = 3;
+
+/// Keepalives in a row saying "I wouldn't sync with you" before the source is given up.
+const REFUSALS: u32 = 2;
+
+fn catching_up(state: &AppState) -> bool {
+    !state.initial_sync_done.load(Ordering::Acquire)
+}
+
+fn lock(state: &AppState) -> std::sync::MutexGuard<'_, Option<CatchUpSource>> {
+    state
+        .catch_up_source
+        .lock()
+        .unwrap_or_else(|poisoned| poisoned.into_inner())
+}
+
+/// Returns how long to hold back sync with `peer`, if at all: while catching up, with
+/// everyone but the source. Picks the source if there is none (or it went quiet).
+pub(crate) fn defer(state: &AppState, peer: &str, peer_state: &PeerState) -> Option<Duration> {
+    if !catching_up(state) {
+        return None;
+    }
+    let stall = state.timing.catch_up_probe * UNANSWERED_PROBES;
+    if peer_state.reports_ready == Some(false) || peer_state.syncs_with_us == Some(false) {
+        // Not a candidate; its next message re-evaluates this, the timer is a fallback.
+        return Some(stall);
+    }
+
+    let now = Instant::now();
+    let mut source = lock(state);
+    let quiet = match &*source {
+        Some(current) if current.peer == peer => return None,
+        Some(current) if now < current.last_heard + stall => {
+            return Some(current.last_heard + stall - now);
+        }
+        Some(current) => Some(current.peer.clone()),
+        None => None,
+    };
+    match quiet {
+        Some(previous) => info!(
+            peer = %peer,
+            previous = %previous,
+            quiet_secs = stall.as_secs_f32(),
+            "Catch-up source stopped answering; catching up from another peer"
+        ),
+        None => debug!(peer = %peer, "Catching up from this peer first"),
+    }
+    *source = Some(CatchUpSource {
+        peer: peer.to_owned(),
+        last_heard: now,
+        last_probe: None,
+        refusals: 0,
+    });
+    None
+}
+
+/// Any message from `peer` arrived.
+pub(crate) fn heard_from(state: &AppState, peer: &str) {
+    if let Some(current) = lock(state).as_mut().filter(|current| current.peer == peer) {
+        current.last_heard = Instant::now();
+    }
+}
+
+/// What `peer` said about itself, in a sync message (`syncs_with_us: None`) or a
+/// keepalive. Gives up the source if that rules it out.
+pub(crate) fn reported(state: &AppState, peer: &str, ready: bool, syncs_with_us: Option<bool>) {
+    if !catching_up(state) {
+        return;
+    }
+    let reason = {
+        let mut source = lock(state);
+        let Some(current) = source.as_mut().filter(|current| current.peer == peer) else {
+            return;
+        };
+        let reason = if !ready {
+            Some("it isn't ready itself")
+        } else {
+            match syncs_with_us {
+                Some(false) => {
+                    current.refusals += 1;
+                    (current.refusals >= REFUSALS).then_some("it wouldn't sync with us")
+                }
+                Some(true) => {
+                    current.refusals = 0;
+                    None
+                }
+                None => None,
+            }
+        };
+        if reason.is_some() {
+            *source = None;
+        }
+        reason
+    };
+    if let Some(reason) = reason {
+        debug!(peer = %peer, reason, "Giving up the catch-up source; trying another peer");
+        resync_everyone(state);
+    }
+}
+
+/// Whether to PING `peer` now: it is the source and has been quiet for a probe interval.
+pub(crate) fn probe_due(state: &AppState, peer: &str, now: Instant) -> bool {
+    if !catching_up(state) {
+        return false;
+    }
+    let probe = state.timing.catch_up_probe;
+    let mut source = lock(state);
+    let Some(current) = source.as_mut().filter(|current| current.peer == peer) else {
+        return false;
+    };
+    let due = now >= current.last_heard + probe
+        && current.last_probe.is_none_or(|last| now >= last + probe);
+    if due {
+        current.last_probe = Some(now);
+    }
+    due
+}
+
+/// The initial sync is done: no source any more, and the peers we held back only need
+/// to learn our heads now.
+pub(crate) fn finished(state: &AppState, from: &str) {
+    *lock(state) = None;
+    let state = state.clone();
+    let from = from.to_owned();
+    tokio::spawn(async move {
+        let peers: Vec<String> = state.peer_states.read().await.keys().cloned().collect();
+        for other in peers.into_iter().filter(|other| *other != from) {
+            let _ = state.sync_events.send(SyncEvent::PeerNeedsSync(other));
+        }
+    });
+}
+
+/// After giving up a source: let every peer be considered again.
+fn resync_everyone(state: &AppState) {
+    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 _ = state.sync_events.send(SyncEvent::PeerNeedsSync(other));
+        }
+    });
+}