introduce sync session

SyncSession holds all fields that get reset by reset_sync()

1 files changed, 90 insertions, 81 deletions

diff --git a/src/state.rs b/src/state.rs
index 33bc9b5..24ae2fb 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -82,34 +82,33 @@ pub(crate) struct PendingOutbound {
     pub(crate) message: transport::EncodedMessage,
     pub(crate) retry_delay: Duration,
     pub(crate) next_attempt: Instant,
+    /// A sync pass was requested while this was pending. Once it is delivered, the peer
+    /// must be enqueued again so the deferred changes go out.
+    pub(crate) regenerate: bool,
 }
 
 #[derive(Clone)]
 pub(crate) struct PeerState {
     pub(crate) standing: Standing,
-    pub(crate) sync_state: sync::State,
     pub(crate) hello_state: HelloState,
 
     /// The peer sent HELLO; we still owe it a HELLO_REPLY (e.g. we had no swarm ID yet,
     /// or sending failed).
     pub(crate) reply_owed: bool,
 
-    /// Received from this peer, waiting to be processed.
+    /// Received from this peer before it became a member, waiting to be processed.
     pub(crate) pending_inbound_sync: Vec<transport::SyncEnvelope>,
 
-    /// Generated for this peer, waiting for delivery.
-    pub(crate) pending_outbound_sync: Option<PendingOutbound>,
-
-    /// A sync pass was requested while an envelope was pending. Once that envelope
-    /// is delivered, the peer must be re-enqueued so the deferred changes go out.
-    pub(crate) needs_generate: bool,
+    /// Everything about the current Automerge sync with this peer; a reset replaces it.
+    pub(crate) sync: SyncSession,
 
     pub(crate) liveness: Liveness,
     pub(crate) added_at: Instant,
     pub(crate) last_seen: Option<Instant>,
     pub(crate) last_ping_at: Option<Instant>,
 
-    /// Last time a sync message from this peer was applied.
+    /// Last time a sync message from this peer was applied (survives resets: the
+    /// keepalive heal looks at it).
     pub(crate) last_sync_rx: Option<Instant>,
 
     /// Last self-healing re-handshake (cooldown).
@@ -118,17 +117,6 @@ pub(crate) struct PeerState {
     /// Epoch the peer advertised in its most recent message.
     pub(crate) epoch: Option<Epoch>,
 
-    /// Consecutive sync exchanges that changed nothing (livelock detection).
-    pub(crate) stalled_exchanges: u32,
-
-    /// The answer we expect to our last sync message(s), if any.
-    pub(crate) sync_reply: Option<SyncReplyWait>,
-    /// How long to wait for it; doubles while answers stay out.
-    pub(crate) sync_reply_timeout: Duration,
-
-    /// When sync with this peer last started from scratch.
-    pub(crate) sync_reset_at: Instant,
-
     /// We switched epochs, and this peer, already on our new epoch, still has our old one
     /// on record: it won't sync with us until a message tells it otherwise. Our sync
     /// opening does that, so we open without waiting for the pair's leader, and the
@@ -136,6 +124,27 @@ pub(crate) struct PeerState {
     pub(crate) epoch_unannounced: bool,
 }
 
+/// One Automerge sync with a peer, from a reset to the next. Everything a reset must
+/// forget lives here, so a reset is `SyncSession::new()` and can't miss a field.
+#[derive(Clone)]
+pub(crate) struct SyncSession {
+    pub(crate) state: sync::State,
+
+    /// Generated for this peer, waiting for delivery.
+    pub(crate) outbound: Option<PendingOutbound>,
+
+    /// The answer we expect to our last sync message(s), if any.
+    pub(crate) reply: Option<SyncReplyWait>,
+    /// How long to wait for it; doubles while answers stay out.
+    pub(crate) reply_timeout: Duration,
+
+    /// When this session started.
+    pub(crate) started_at: Instant,
+
+    /// Consecutive exchanges that changed nothing (livelock detection).
+    pub(crate) stalled_exchanges: u32,
+}
+
 /// Sync messages we sent and expect an answer to. Automerge assumes reliable delivery:
 /// a lost message leaves both sides waiting (`in_flight`), and changes it carried are
 /// never sent again (`sent_hashes`). We resend after a timeout instead.
@@ -156,28 +165,72 @@ const SYNC_REPLY_TIMEOUT_MAX: Duration = Duration::from_secs(300);
 /// before opening itself (the other one may not have reset, or its message got lost).
 const SYNC_OPENING_GRACE: Duration = Duration::from_secs(3);
 
+impl SyncSession {
+    pub(crate) fn new() -> Self {
+        Self {
+            state: sync::State::new(),
+            outbound: None,
+            reply: None,
+            reply_timeout: SYNC_REPLY_TIMEOUT,
+            started_at: Instant::now(),
+            stalled_exchanges: 0,
+        }
+    }
+
+    /// A sync message went out; `sent` are the changes it carried for the first time.
+    pub(crate) fn expect_reply(&mut self, sent: Vec<automerge::ChangeHash>, now: Instant) {
+        match &mut self.reply {
+            // Still waiting for an earlier one: keep its deadline, remember these too.
+            Some(wait) => wait.sent.extend(sent),
+            None => {
+                self.reply = Some(SyncReplyWait {
+                    deadline: now + self.reply_timeout,
+                    sent,
+                });
+            }
+        }
+    }
+
+    /// No answer is due any more: the peer answered, or keepalives show both documents
+    /// are the same.
+    pub(crate) fn reply_settled(&mut self) {
+        self.reply = None;
+        self.reply_timeout = SYNC_REPLY_TIMEOUT;
+    }
+
+    /// If the answer is overdue: forgets that the unanswered changes were sent and that
+    /// a message is in flight, so the next sync pass sends them again. Returns how many
+    /// changes that concerns.
+    fn take_overdue_reply(&mut self, now: Instant) -> Option<usize> {
+        if self.reply.as_ref().is_none_or(|wait| now < wait.deadline) {
+            return None;
+        }
+        let wait = self.reply.take()?;
+        for hash in &wait.sent {
+            self.state.sent_hashes.remove(hash);
+        }
+        self.state.in_flight = false;
+        self.reply_timeout = (self.reply_timeout * 2).min(SYNC_REPLY_TIMEOUT_MAX);
+        Some(wait.sent.len())
+    }
+}
+
 impl Default for PeerState {
     fn default() -> Self {
         Self {
             standing: Standing::default(),
-            sync_state: sync::State::new(),
             hello_state: HelloState::default(),
             reply_owed: false,
             pending_inbound_sync: Vec::new(),
-            pending_outbound_sync: None,
-            needs_generate: false,
+            sync: SyncSession::new(),
             liveness: Liveness::Active,
             added_at: Instant::now(),
             last_seen: None,
             last_ping_at: None,
             last_sync_rx: None,
             last_resync_at: None,
-            stalled_exchanges: 0,
-            sync_reply: None,
-            sync_reply_timeout: SYNC_REPLY_TIMEOUT,
-            sync_reset_at: Instant::now(),
-            epoch_unannounced: false,
             epoch: None,
+            epoch_unannounced: false,
         }
     }
 }
@@ -186,78 +239,34 @@ impl PeerState {
     /// Start Automerge sync with this peer from scratch. Any envelope generated against
     /// the old state is discarded; a stale RetryDelivery event is then a no-op.
     pub(crate) fn reset_sync(&mut self) {
-        self.sync_state = sync::State::new();
-        self.pending_outbound_sync = None;
-        self.needs_generate = false;
-        self.sync_reply = None;
-        self.sync_reply_timeout = SYNC_REPLY_TIMEOUT;
-        self.sync_reset_at = Instant::now();
+        self.sync = SyncSession::new();
     }
 
     /// After a reset, usually both sides do it and both open the sync at once. The
     /// openings cross, and each side answers the other's: harmless between peers that
     /// share history, but an empty peer answers with empty heads, which Automerge takes
     /// for "lost everything", so the whole document goes out a second time. Only the
-    /// side that leads the pair (`swarm::pair`) opens; the other waits for its opening, up to
-    /// `SYNC_OPENING_GRACE`. Returns how much longer to wait, if at all.
+    /// side that leads the pair (`swarm::pair`) opens; the other waits for its opening,
+    /// up to `SYNC_OPENING_GRACE`. Returns how much longer to wait, if at all.
     ///
     /// After our epoch switch, we open anyway (`epoch_unannounced`): the peer can't, as
     /// it still thinks we are on another epoch.
     pub(crate) fn opening_wait(&self, we_open: bool, now: Instant) -> Option<Duration> {
-        if we_open || self.epoch_unannounced || self.sync_state.their_heads.is_some() {
+        if we_open || self.epoch_unannounced || self.sync.state.their_heads.is_some() {
             return None;
         }
         SYNC_OPENING_GRACE
-            .checked_sub(now.saturating_duration_since(self.sync_reset_at))
+            .checked_sub(now.saturating_duration_since(self.sync.started_at))
             .filter(|wait| !wait.is_zero())
     }
 
-    /// A sync message went out; `sent` are the changes it carried for the first time.
-    pub(crate) fn expect_sync_reply(&mut self, sent: Vec<automerge::ChangeHash>, now: Instant) {
-        match &mut self.sync_reply {
-            // Still waiting for an earlier one: keep its deadline, remember these too.
-            Some(wait) => wait.sent.extend(sent),
-            None => {
-                self.sync_reply = Some(SyncReplyWait {
-                    deadline: now + self.sync_reply_timeout,
-                    sent,
-                });
-            }
-        }
-    }
-
-    /// Keepalives show both documents are the same: stop waiting for an answer.
-    pub(crate) fn sync_in_step(&mut self) {
-        self.sync_reply = None;
-        self.sync_reply_timeout = SYNC_REPLY_TIMEOUT;
-    }
-
-    /// The peer answered.
-    pub(crate) fn sync_reply_received(&mut self) {
-        self.sync_reply = None;
-        self.sync_reply_timeout = SYNC_REPLY_TIMEOUT;
-    }
-
-    /// If the answer is overdue: forgets that the unanswered changes were sent and that
-    /// a message is in flight, so the next sync pass sends them again. Returns how many
-    /// changes that concerns.
+    /// An overdue sync answer from an established, active peer (see
+    /// `SyncSession::take_overdue_reply`).
     pub(crate) fn take_overdue_sync_reply(&mut self, now: Instant) -> Option<usize> {
-        if self.liveness != Liveness::Active
-            || self.hello_state != HelloState::Established
-            || self
-                .sync_reply
-                .as_ref()
-                .is_none_or(|wait| now < wait.deadline)
-        {
+        if self.liveness != Liveness::Active || self.hello_state != HelloState::Established {
             return None;
         }
-        let wait = self.sync_reply.take()?;
-        for hash in &wait.sent {
-            self.sync_state.sent_hashes.remove(hash);
-        }
-        self.sync_state.in_flight = false;
-        self.sync_reply_timeout = (self.sync_reply_timeout * 2).min(SYNC_REPLY_TIMEOUT_MAX);
-        Some(wait.sent.len())
+        self.sync.take_overdue_reply(now)
     }
 
     fn quiet_for(&self, now: Instant) -> Duration {