introduce sync session

SyncSession holds all fields that get reset by reset_sync()

1 files changed, 32 insertions, 26 deletions

diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 22d6600..91c6eee 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -94,10 +94,10 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool
     let existing = {
         let mut guard = peer_state_mutex.lock().await;
         let peer_state = &mut *guard;
-        match &peer_state.pending_outbound_sync {
+        match &mut peer_state.sync.outbound {
             Some(entry) if retry_delivery => Some((entry.message.clone(), entry.next_attempt)),
-            Some(_) => {
-                peer_state.needs_generate = true;
+            Some(entry) => {
+                entry.regenerate = true;
                 return Ok(());
             }
             None => None,
@@ -125,29 +125,29 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool
                     let (epoch, message) = {
                         let mut replica = doc.blocking_write();
                         let mut peer_state = psm.blocking_lock();
-                        peer_state.needs_generate = false;
                         let epoch = replica.epoch;
-                        let sent_before = peer_state.sync_state.sent_hashes.clone();
+                        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)
+                            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()
+                        let in_step = peer_state.sync.state.their_heads.as_ref()
                             == Some(&replica.doc.get_heads());
                         if in_step {
-                            peer_state.sync_in_step();
+                            peer_state.sync.reply_settled();
                         } else if message.is_some() {
                             let sent = peer_state
-                                .sync_state
+                                .sync
+                                .state
                                 .sent_hashes
                                 .difference(&sent_before)
                                 .copied()
                                 .collect();
-                            peer_state.expect_sync_reply(sent, Instant::now());
+                            peer_state.sync.expect_reply(sent, Instant::now());
                         }
                         if message.is_some() {
                             // It carries our epoch.
@@ -175,10 +175,11 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool
                 return Ok(());
             };
 
-            peer_state_mutex.lock().await.pending_outbound_sync = Some(PendingOutbound {
+            peer_state_mutex.lock().await.sync.outbound = Some(PendingOutbound {
                 message: encoded.clone(),
                 retry_delay: INITIAL_SYNC_RETRY,
                 next_attempt: Instant::now(),
+                regenerate: false,
             });
 
             encoded
@@ -190,8 +191,11 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool
         Ok(()) => {
             let regenerate = {
                 let mut peer_state = peer_state_mutex.lock().await;
-                peer_state.pending_outbound_sync = None;
-                std::mem::take(&mut peer_state.needs_generate)
+                peer_state
+                    .sync
+                    .outbound
+                    .take()
+                    .is_some_and(|entry| entry.regenerate)
             };
 
             trace!(regenerate, "Outbound sync message accepted by Mycelium");
@@ -209,7 +213,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool
             let delay = {
                 let mut peer_state = peer_state_mutex.lock().await;
 
-                let Some(ref mut entry) = peer_state.pending_outbound_sync else {
+                let Some(ref mut entry) = peer_state.sync.outbound else {
                     debug!("Pending envelope discarded by a sync reset during delivery");
                     return Ok(());
                 };
@@ -322,26 +326,27 @@ pub(crate) async fn handle_sync(
             let mut peer_state = peer_state_mutex.blocking_lock();
 
             let heads_before = replica.doc.get_heads();
-            let their_heads_before = peer_state.sync_state.their_heads.clone();
+            let their_heads_before = peer_state.sync.state.their_heads.clone();
 
             replica
                 .doc
                 .sync()
-                .receive_sync_message(&mut peer_state.sync_state, message)
+                .receive_sync_message(&mut peer_state.sync.state, message)
                 .context("Automerge synchronization failed")?;
             peer_state.last_sync_rx = Some(Instant::now());
-            peer_state.sync_reply_received();
+            peer_state.sync.reply_settled();
 
             let heads_after = replica.doc.get_heads();
             let unproductive = heads_before == heads_after
-                && peer_state.sync_state.their_heads == their_heads_before
-                && peer_state.sync_state.their_heads.as_ref() != Some(&heads_after);
+                && peer_state.sync.state.their_heads == their_heads_before
+                && peer_state.sync.state.their_heads.as_ref() != Some(&heads_after);
             break_sync_stall(&peer, &mut peer_state, unproductive);
 
             Ok(Some(
                 check_caught_up
                     && peer_state
-                        .sync_state
+                        .sync
+                        .state
                         .their_heads
                         .as_ref()
                         .is_some_and(|heads| {
@@ -393,17 +398,18 @@ const STALL_RESET: u32 = 32;
 ///   everything the peer's bloom filter lacks.
 /// * If that doesn't help either, reset both sides through a fresh handshake.
 fn break_sync_stall(peer: &str, peer_state: &mut PeerState, unproductive: bool) {
+    let session = &mut peer_state.sync;
     if !unproductive {
-        peer_state.stalled_exchanges = 0;
+        session.stalled_exchanges = 0;
         return;
     }
-    peer_state.stalled_exchanges += 1;
+    session.stalled_exchanges += 1;
 
-    let sync_state = &mut peer_state.sync_state;
+    let sync_state = &mut session.state;
 
     // Not on the first unproductive exchange: when messages cross, the peer asks for a
     // change that is merely still in flight, and resending it would only duplicate it.
-    if peer_state.stalled_exchanges >= STALL_RESEND_NEEDED {
+    if session.stalled_exchanges >= STALL_RESEND_NEEDED {
         if let Some(need) = &sync_state.their_need {
             let before = sync_state.sent_hashes.len();
             sync_state.sent_hashes.retain(|hash| !need.contains(hash));
@@ -414,7 +420,7 @@ fn break_sync_stall(peer: &str, peer_state: &mut PeerState, unproductive: bool)
         }
     }
 
-    match peer_state.stalled_exchanges {
+    match session.stalled_exchanges {
         STALL_CLEAR_SENT => {
             warn!(
                 peer = %peer,
@@ -427,7 +433,7 @@ fn break_sync_stall(peer: &str, peer_state: &mut PeerState, unproductive: bool)
         }
         STALL_RESET => {
             warn!(peer = %peer, "Sync is still stalled; resetting it on both sides");
-            peer_state.stalled_exchanges = 0;
+            session.stalled_exchanges = 0;
             peer_state.request_resync();
         }
         _ => {}