1 files changed, 27 insertions, 6 deletions

diff --git a/src/swarm/source.rs b/src/swarm/source.rs
index d80e4d7..1288d86 100644
--- a/src/swarm/source.rs
+++ b/src/swarm/source.rs
@@ -37,7 +37,6 @@ use tracing::{debug, info, warn};
 
 use crate::locks::MutexExt;
 use crate::state::{AppState, CatchUpSource, PeerState};
-use crate::swarm::sync::SyncEvent;
 
 /// How long a clearly closer peer (by round-trip time) gets to claim the source role
 /// before whoever asks next takes it.
@@ -147,6 +146,7 @@ pub(crate) fn defer(state: &AppState, peer: &str, peer_state: &PeerState) -> Opt
         last_probe: None,
         refusals: 0,
     });
+    state.peers_due.by(now + state.timing.catch_up_probe);
     None
 }
 
@@ -286,18 +286,39 @@ pub(crate) fn probe_due(state: &AppState, peer: &str, now: Instant) -> bool {
     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) {
+/// When a STATUS to `peer` is next due (`probe_due`), if it is the source.
+pub(crate) fn probe_at(state: &AppState, peer: &str) -> Option<Instant> {
+    if !catching_up(state) {
+        return None;
+    }
+    let probe = state.timing.catch_up_probe;
+    let source = lock(state);
+    let current = source.as_ref().filter(|current| current.peer == peer)?;
+    let heard = current.last_heard + probe;
+    Some(match current.last_probe {
+        Some(last) => heard.max(last + probe),
+        None => heard,
+    })
+}
+
+/// No catch-up going on: no source, no head start, nobody avoided. When one ends, and
+/// before one starts (a rejoin: `Host::reset_initial_sync`).
+pub(crate) fn reset(state: &AppState) {
     *lock(state) = None;
     *lock_preference(state) = None;
     *lock_avoided(state) = None;
+}
+
+/// 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) {
+    reset(state);
     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));
+            state.request_sync(&other);
         }
     });
 }
@@ -308,7 +329,7 @@ pub(crate) fn resync_everyone(state: &AppState) {
     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));
+            state.request_sync(&other);
         }
     });
 }