unify request_sync logic

1 files changed, 5 insertions, 21 deletions

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);
     });
 }