retry using fresh messages

Don't store encoded messages but retry using a newly generated one.
That way any accumulated changes can be delivered at once.

1 files changed, 127 insertions, 158 deletions

diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index be4b1e2..e05d275 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -4,7 +4,6 @@
 use std::collections::HashSet;
 use std::sync::Arc;
 use std::sync::atomic::Ordering;
-use std::time::Duration;
 
 use anyhow::{Context, Result, anyhow, bail};
 use automerge::sync::SyncDoc;
@@ -12,21 +11,17 @@ use tokio::sync::mpsc;
 use tokio::time::Instant;
 use tracing::{debug, info, trace, warn};
 
-use crate::state::{AppState, Liveness, PendingOutbound, Standing};
+use crate::state::{AppState, Liveness, Sending, 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);
-const MAX_SYNC_RETRY: Duration = Duration::from_secs(30);
-
 #[derive(Debug)]
 pub(crate) enum SyncEvent {
     PeerValidated(String),
     DocumentChanged,
     MessageReceived(String),
     PeerNeedsSync(String),
-    RetryDelivery(String),
 }
 
 /// Store paths touched by one batch of document changes.
@@ -35,7 +30,7 @@ pub(crate) struct ChangeSet {
     pub(crate) paths: Vec<Vec<String>>,
 }
 
-async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool) -> Result<()> {
+async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
     let Some(peer_state_mutex) = state.get_peer_state(peer).await else {
         trace!("Peer was forgotten; skipping synchronization");
         return Ok(());
@@ -65,31 +60,29 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool
             return Ok(());
         };
 
-        if !retry_delivery {
-            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 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));
-                });
-                return Ok(());
-            }
+        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 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));
+            });
+            return Ok(());
+        }
 
-            if peer_state
-                .sync
-                .leave_opening(we_ready, peer_state.reports_ready)
-            {
-                // Its own opening (or anything else from it) brings us back here.
-                trace!("Leaving the opening sync message to the peer, which isn't ready");
-                return Ok(());
-            }
+        if peer_state
+            .sync
+            .leave_opening(we_ready, peer_state.reports_ready)
+        {
+            // Its own opening (or anything else from it) brings us back here.
+            trace!("Leaving the opening sync message to the peer, which isn't ready");
+            return Ok(());
         }
 
         swarm_id.clone()
@@ -104,110 +97,109 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool
         bail!("Peer {peer} belongs to a different swarm");
     }
 
-    let existing = {
-        let mut guard = peer_state_mutex.lock().await;
-        let peer_state = &mut *guard;
-        match &mut peer_state.sync.outbound {
-            Some(entry) if retry_delivery => Some((entry.message.clone(), entry.next_attempt)),
-            Some(entry) => {
-                entry.regenerate = true;
-                return Ok(());
-            }
-            None => None,
+    {
+        let mut peer_state = peer_state_mutex.lock().await;
+        if let Some(sending) = &mut peer_state.sync.sending {
+            sending.regenerate = true;
+            return Ok(());
         }
-    };
-
-    let encoded = match existing {
-        Some((message, next_attempt)) => {
-            if Instant::now() < next_attempt {
-                return Ok(());
-            }
-            message
+        if peer_state
+            .sync
+            .retry_at
+            .is_some_and(|at| Instant::now() < at)
+        {
+            // The retry timer enqueues the peer once it's due.
+            return Ok(());
         }
-        None if retry_delivery => return Ok(()),
-        None => {
-            let doc = Arc::clone(&state.doc);
-            let psm = Arc::clone(&peer_state_mutex);
-            let transport = Arc::clone(&state.transport);
-            let peer_owned = peer.to_owned();
-            let ready = state.is_ready().await;
+    }
 
-            let generated = tokio::task::spawn_blocking(
-                move || -> Result<Option<transport::EncodedMessage>> {
-                    // Label the message with the epoch of the document it was generated
-                    // from, read under the same lock.
-                    let (epoch, message) = {
-                        let mut replica = doc.blocking_write();
-                        let mut peer_state = psm.blocking_lock();
-                        let epoch = replica.epoch;
-                        let sent_before = peer_state.sync.state.sent_hashes.clone();
-                        let message = {
-                            let sync_doc = replica.doc.sync();
-                            sync_doc.generate_sync_message(&mut peer_state.sync.state)
-                        };
-                        // A peer whose heads are ours already has everything; Automerge
-                        // answers only when it has something to say, so it will stay
-                        // silent. Waiting for an answer would only end in a pointless
-                        // resend.
-                        let in_step = peer_state.sync.state.their_heads.as_ref()
-                            == Some(&replica.doc.get_heads());
-                        if in_step {
-                            peer_state.sync.reply_settled();
-                        } else if message.is_some() {
-                            let sent = peer_state
-                                .sync
-                                .state
-                                .sent_hashes
-                                .difference(&sent_before)
-                                .copied()
-                                .collect();
-                            peer_state.sync.expect_reply(sent, Instant::now());
-                        }
-                        (epoch, message)
-                    };
-                    let Some(message) = message else {
-                        return Ok(None);
-                    };
-                    let envelope = transport::SyncEnvelope {
-                        swarm_id,
-                        epoch,
-                        message,
-                        ready,
+    let encoded = {
+        let doc = Arc::clone(&state.doc);
+        let psm = Arc::clone(&peer_state_mutex);
+        let transport = Arc::clone(&state.transport);
+        let peer_owned = peer.to_owned();
+        let ready = state.is_ready().await;
+
+        let generated =
+            tokio::task::spawn_blocking(move || -> Result<Option<transport::EncodedMessage>> {
+                // Label the message with the epoch of the document it was generated
+                // from, read under the same lock.
+                let (epoch, message) = {
+                    let mut replica = doc.blocking_write();
+                    let mut peer_state = psm.blocking_lock();
+                    let epoch = replica.epoch;
+                    let sent_before = peer_state.sync.state.sent_hashes.clone();
+                    let message = {
+                        let sync_doc = replica.doc.sync();
+                        sync_doc.generate_sync_message(&mut peer_state.sync.state)
                     };
-                    transport
-                        .encode(&peer_owned, &ParsedMessage::Sync(envelope))
-                        .map(Some)
-                },
-            )
+                    // A peer whose heads are ours already has everything; Automerge
+                    // answers only when it has something to say, so it will stay
+                    // silent. Waiting for an answer would only end in a pointless
+                    // resend.
+                    let in_step = peer_state.sync.state.their_heads.as_ref()
+                        == Some(&replica.doc.get_heads());
+                    let sent: Vec<_> = peer_state
+                        .sync
+                        .state
+                        .sent_hashes
+                        .difference(&sent_before)
+                        .copied()
+                        .collect();
+                    if in_step {
+                        peer_state.sync.reply_settled();
+                    } else if message.is_some() {
+                        peer_state.sync.expect_reply(sent.clone(), Instant::now());
+                    }
+                    if message.is_some() {
+                        // Marked under the same lock as the generation, so no second
+                        // message is generated while this one is handed over.
+                        peer_state.sync.sending = Some(Sending {
+                            sent,
+                            regenerate: false,
+                        });
+                    }
+                    (epoch, message)
+                };
+                let Some(message) = message else {
+                    return Ok(None);
+                };
+                let envelope = transport::SyncEnvelope {
+                    swarm_id,
+                    epoch,
+                    message,
+                    ready,
+                };
+                transport
+                    .encode(&peer_owned, &ParsedMessage::Sync(envelope))
+                    .map(Some)
+            })
             .await
-            .context("Sync generation task panicked")??;
-
-            let Some(encoded) = generated else {
-                return Ok(());
-            };
+            .context("Sync generation task panicked")?;
+        let generated = match generated {
+            Ok(generated) => generated,
+            Err(error) => {
+                // Not encodable, so nothing went out: take back what it would have carried.
+                let _ = peer_state_mutex
+                    .lock()
+                    .await
+                    .sync
+                    .send_failed(Instant::now());
+                return Err(error);
+            }
+        };
 
-            peer_state_mutex.lock().await.sync.outbound = Some(PendingOutbound {
-                message: encoded.clone(),
-                retry_delay: INITIAL_SYNC_RETRY,
-                next_attempt: Instant::now(),
-                regenerate: false,
-            });
+        let Some(encoded) = generated else {
+            return Ok(());
+        };
 
-            encoded
-        }
+        encoded
     };
 
     // No state locks are held across the send.
     match state.transport.send_encoded(&encoded).await {
         Ok(()) => {
-            let regenerate = {
-                let mut peer_state = peer_state_mutex.lock().await;
-                peer_state
-                    .sync
-                    .outbound
-                    .take()
-                    .is_some_and(|entry| entry.regenerate)
-            };
+            let regenerate = peer_state_mutex.lock().await.sync.send_succeeded();
 
             trace!(regenerate, "Outbound sync message accepted by Mycelium");
 
@@ -221,30 +213,27 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool
             Ok(())
         }
         Err(error) => {
-            let delay = {
-                let mut peer_state = peer_state_mutex.lock().await;
-
-                let Some(ref mut entry) = peer_state.sync.outbound else {
-                    debug!("Pending envelope discarded by a sync reset during delivery");
-                    return Ok(());
-                };
-
-                entry.retry_delay = entry.retry_delay.saturating_mul(2).min(MAX_SYNC_RETRY);
-                entry.next_attempt = Instant::now() + entry.retry_delay;
-                entry.retry_delay
+            let Some(delay) = peer_state_mutex
+                .lock()
+                .await
+                .sync
+                .send_failed(Instant::now())
+            else {
+                debug!("Sync message refused, but a reset already discarded it");
+                return Ok(());
             };
 
             warn!(
                 retry_ms = delay.as_millis(),
                 error = %format!("{error:#}"),
-                "Sync delivery failed; retaining the same outbound envelope"
+                "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::RetryDelivery(peer));
+                let _ = tx.send(SyncEvent::PeerNeedsSync(peer));
             });
 
             Ok(())
@@ -405,13 +394,12 @@ fn resend_if_needed(peer: &str, sync_state: &mut automerge::sync::State) {
 
 pub(crate) async fn run_sync_worker(state: AppState, mut rx: mpsc::UnboundedReceiver<SyncEvent>) {
     let mut dirty_peers = HashSet::<String>::new();
-    let mut retry_peers = HashSet::<String>::new();
     let mut document_dirty = false;
 
     info!("Starting synchronization worker");
 
     loop {
-        if dirty_peers.is_empty() && retry_peers.is_empty() && !document_dirty {
+        if dirty_peers.is_empty() && !document_dirty {
             match rx.recv().await {
                 Some(event) => match event {
                     SyncEvent::PeerValidated(peer)
@@ -419,9 +407,6 @@ pub(crate) async fn run_sync_worker(state: AppState, mut rx: mpsc::UnboundedRece
                     | SyncEvent::PeerNeedsSync(peer) => {
                         dirty_peers.insert(peer);
                     }
-                    SyncEvent::RetryDelivery(peer) => {
-                        retry_peers.insert(peer);
-                    }
                     SyncEvent::DocumentChanged => {
                         document_dirty = true;
                     }
@@ -441,9 +426,6 @@ pub(crate) async fn run_sync_worker(state: AppState, mut rx: mpsc::UnboundedRece
                 | SyncEvent::PeerNeedsSync(peer) => {
                     dirty_peers.insert(peer);
                 }
-                SyncEvent::RetryDelivery(peer) => {
-                    retry_peers.insert(peer);
-                }
                 SyncEvent::DocumentChanged => {
                     document_dirty = true;
                 }
@@ -469,23 +451,10 @@ pub(crate) async fn run_sync_worker(state: AppState, mut rx: mpsc::UnboundedRece
             }
         }
 
-        // Retry existing envelopes separately from generating new ones.
-        let retries: Vec<String> = retry_peers.drain().collect();
-
-        for peer in retries {
-            if let Err(error) = deliver_sync_to_peer(&state, &peer, true).await {
-                warn!(
-                    peer = %peer,
-                    error = %format!("{error:#}"),
-                    "Sync delivery retry failed"
-                );
-            }
-        }
-
         let peers: Vec<String> = dirty_peers.drain().collect();
 
         for peer in peers {
-            if let Err(error) = deliver_sync_to_peer(&state, &peer, false).await {
+            if let Err(error) = deliver_sync_to_peer(&state, &peer).await {
                 warn!(
                     peer = %peer,
                     error = %format!("{error:#}"),