unify request_sync logic

9 files changed, 31 insertions, 46 deletions

diff --git a/src/state/host.rs b/src/state/host.rs
index a165637..9d77f35 100644
--- a/src/state/host.rs
+++ b/src/state/host.rs
@@ -15,7 +15,6 @@ use super::{AppState, HelloState, NodeLifecycle, StaleView, Standing};
 use crate::document::replica::Epoch;
 use crate::document::store::ChangeSet;
 use crate::remake::{Host, PeerSummary, RemakeState};
-use crate::swarm::sync::SyncEvent;
 
 impl AppState {
     /// After the document was replaced: every peer's sync state refers to the old
@@ -56,7 +55,7 @@ impl AppState {
         }
 
         for (id, _) in peers {
-            let _ = self.sync_events.send(SyncEvent::PeerNeedsSync(id));
+            self.request_sync(&id);
         }
     }
 }
diff --git a/src/state/mod.rs b/src/state/mod.rs
index e8975e2..8f3a841 100644
--- a/src/state/mod.rs
+++ b/src/state/mod.rs
@@ -180,6 +180,24 @@ pub(crate) struct CatchUpSource {
 }
 
 impl AppState {
+    /// Asks the sync worker to look at `peer` (`SyncEvent::PeerNeedsSync`): whether to
+    /// send it a sync message now. Nothing if the worker is gone (the node stops).
+    pub(crate) fn request_sync(&self, peer: &str) {
+        let _ = self
+            .sync_events
+            .send(SyncEvent::PeerNeedsSync(peer.to_owned()));
+    }
+
+    /// `request_sync`, after `delay`.
+    pub(crate) fn request_sync_after(&self, peer: &str, delay: std::time::Duration) {
+        let events = self.sync_events.clone();
+        let peer = peer.to_owned();
+        tokio::spawn(async move {
+            tokio::time::sleep(delay).await;
+            let _ = events.send(SyncEvent::PeerNeedsSync(peer));
+        });
+    }
+
     /// Whether the node shuts down (`closing`).
     pub(crate) fn is_closing(&self) -> bool {
         *self.closing.borrow()
diff --git a/src/swarm/peer/epoch.rs b/src/swarm/peer/epoch.rs
index 56711b1..d97db0f 100644
--- a/src/swarm/peer/epoch.rs
+++ b/src/swarm/peer/epoch.rs
@@ -6,7 +6,6 @@ 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
@@ -66,9 +65,7 @@ pub(super) async fn observe_peer_epoch(
             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()));
+                    state.request_sync(peer);
                 }
             }
             // Expected after a remake: either side switches or catches up.
diff --git a/src/swarm/peer/handshake.rs b/src/swarm/peer/handshake.rs
index a3290fa..225a7c9 100644
--- a/src/swarm/peer/handshake.rs
+++ b/src/swarm/peer/handshake.rs
@@ -10,7 +10,6 @@ use tokio::time::Instant;
 use tracing::{debug, info, instrument, trace, warn};
 
 use crate::state::{AppState, HelloState, PeerState, Standing};
-use crate::swarm::sync::SyncEvent;
 use crate::swarm::topology;
 use crate::transport::{self, ParsedMessage};
 
@@ -347,9 +346,7 @@ pub(crate) async fn apply_neighbour_change(state: &AppState, change: topology::N
         }
         // Skipped there unless the handshake is complete; otherwise its reply restarts
         // sync anyway.
-        let _ = state
-            .sync_events
-            .send(SyncEvent::PeerNeedsSync(peer.clone()));
+        state.request_sync(peer);
     }
     for peer in &change.removed {
         if let Some(peer_state) = state.get_peer_state(peer).await {
diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs
index 0e80980..40b8ed3 100644
--- a/src/swarm/peer/keepalive.rs
+++ b/src/swarm/peer/keepalive.rs
@@ -17,7 +17,6 @@ use tracing::{debug, info, warn};
 
 use super::handshake::ensure_handshake;
 use crate::state::{AppState, PeerState};
-use crate::swarm::sync::SyncEvent;
 use crate::swarm::topology;
 use crate::transport::{self, ParsedMessage};
 
@@ -214,9 +213,7 @@ pub(super) async fn note_status(state: &AppState, peer: &str, remote: &transport
         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()));
+            state.request_sync(peer);
         }
     }
     crate::swarm::source::reported(state, peer, remote.ready, Some(remote.syncs_with_you));
diff --git a/src/swarm/peer/worker.rs b/src/swarm/peer/worker.rs
index f3d0e84..87d2b3c 100644
--- a/src/swarm/peer/worker.rs
+++ b/src/swarm/peer/worker.rs
@@ -13,7 +13,6 @@ use super::handshake::retry_hello_if_due;
 use super::keepalive::{send_ping, send_status};
 use crate::locks::MutexExt;
 use crate::state::{AppState, Keepalive};
-use crate::swarm::sync::SyncEvent;
 use crate::swarm::topology;
 
 const PEER_TICK: Duration = Duration::from_millis(500);
@@ -69,9 +68,7 @@ pub(crate) async fn run_peer_worker(state: AppState) {
             if let Some(changes) = unanswered {
                 debug!(peer = %peer, changes, "No answer to our sync message; sending again");
                 crate::swarm::source::reply_timed_out(&state, peer);
-                let _ = state
-                    .sync_events
-                    .send(SyncEvent::PeerNeedsSync(peer.clone()));
+                state.request_sync(peer);
             }
 
             if became_dormant {
diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs
index 6172f67..00043f2 100644
--- a/src/swarm/relay.rs
+++ b/src/swarm/relay.rs
@@ -28,7 +28,6 @@ use tracing::debug;
 use crate::locks::MutexExt;
 use crate::state::{AppState, Standing};
 use crate::swarm::peer::send_status;
-use crate::swarm::sync::SyncEvent;
 use crate::swarm::topology;
 use crate::transport::{self, HeadsSummary};
 
@@ -183,9 +182,7 @@ pub(crate) async fn pull_if_behind(
     peer_state.lock().await.sync.pull_from(theirs);
     state.stats.pull_started();
     debug!(peer = %peer, "Neighbour has changes we lack; pulling");
-    let _ = state
-        .sync_events
-        .send(SyncEvent::PeerNeedsSync(peer.to_owned()));
+    state.request_sync(peer);
     true
 }
 
diff --git a/src/swarm/source.rs b/src/swarm/source.rs
index 5e428f2..e098dc1 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.
@@ -303,7 +302,7 @@ pub(crate) fn finished(state: &AppState, from: &str) {
     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);
         }
     });
 }
@@ -314,7 +313,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);
         }
     });
 }
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index e89c2e2..b34c20b 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -68,12 +68,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
             );
             // Done earlier when the initial sync completes (every peer is enqueued
             // then); this only matters if the source goes quiet.
-            let tx = state.sync_events.clone();
-            let peer = peer.to_owned();
-            tokio::spawn(async move {
-                tokio::time::sleep(wait).await;
-                let _ = tx.send(SyncEvent::PeerNeedsSync(peer));
-            });
+            state.request_sync_after(peer, wait);
             return Ok(());
         }
 
@@ -241,10 +236,7 @@ async fn send_generated(
             );
 
             if regenerate {
-                state
-                    .sync_events
-                    .send(SyncEvent::PeerNeedsSync(peer.to_owned()))
-                    .map_err(|error| anyhow!("Failed to notify sync worker: {error}"))?;
+                state.request_sync(peer);
             }
 
             Ok(())
@@ -258,10 +250,7 @@ async fn send_generated(
             else {
                 // The new session may not have opened yet.
                 debug!("Sync message refused, but a reset already discarded it");
-                state
-                    .sync_events
-                    .send(SyncEvent::PeerNeedsSync(peer.to_owned()))
-                    .map_err(|error| anyhow!("Failed to notify sync worker: {error}"))?;
+                state.request_sync(peer);
                 return Ok(());
             };
 
@@ -271,12 +260,7 @@ async fn send_generated(
                 "Sync delivery failed; generating a fresh message later"
             );
 
-            let tx = state.sync_events.clone();
-            let peer = peer.to_owned();
-            tokio::spawn(async move {
-                tokio::time::sleep(delay).await;
-                let _ = tx.send(SyncEvent::PeerNeedsSync(peer));
-            });
+            state.request_sync_after(peer, delay);
 
             Ok(())
         }
@@ -486,7 +470,7 @@ fn fetch_delta(state: &AppState, peer: &str, epoch: transport::Epoch, spec: tran
             }
         }
         // Our next message says what we have now; the peer answers with the rest.
-        let _ = state.sync_events.send(SyncEvent::PeerNeedsSync(peer));
+        state.request_sync(&peer);
     });
 }