1 files changed, 155 insertions, 0 deletions
diff --git a/src/state/session.rs b/src/state/session.rs
new file mode 100644
index 0000000..94f9d71
--- /dev/null
+++ b/src/state/session.rs
@@ -0,0 +1,155 @@
+//! One Automerge sync with a peer, from a reset to the next.
+
+use std::time::Duration;
+
+use automerge::sync;
+use tokio::time::Instant;
+
+use crate::backoff::Backoff;
+use crate::swarm::throughput::SyncBytes;
+
+/// A sync message on its way to our Mycelium (the HTTP request is in progress).
+#[derive(Clone)]
+pub(crate) struct Sending {
+    /// Changes it carries for the first time; taken back if the send fails.
+    pub(crate) sent: Vec<automerge::ChangeHash>,
+    /// A sync pass was requested meanwhile. Once this one is handed over, the peer must
+    /// be enqueued again so the deferred changes go out.
+    pub(crate) regenerate: bool,
+}
+
+/// First wait before generating again after our Mycelium refused a message; doubles
+/// while it keeps refusing.
+const SEND_RETRY: Duration = Duration::from_millis(500);
+const SEND_RETRY_MAX: Duration = Duration::from_secs(30);
+
+/// One Automerge sync with a peer, from a reset to the next. Everything a reset must
+/// forget lives here, so a reset replaces it whole and can't miss a field.
+#[derive(Clone)]
+pub(crate) struct SyncSession {
+    pub(crate) state: sync::State,
+
+    /// A message being handed to our Mycelium.
+    pub(crate) sending: Option<Sending>,
+
+    /// After our Mycelium refused a message: no new one before this. The refused one is
+    /// not kept; the next one is generated afresh and carries everything pending.
+    pub(crate) retry_at: Option<Instant>,
+    /// The wait after the next refusal.
+    pub(crate) send_backoff: Backoff,
+
+    /// 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_backoff: Backoff,
+}
+
+/// 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.
+#[derive(Clone, Debug)]
+pub(crate) struct SyncReplyWait {
+    pub(crate) deadline: Instant,
+    /// When the first of the unanswered messages was generated, and the sync bytes
+    /// exchanged with the peer by then (`swarm::throughput`).
+    pub(crate) started: (Instant, SyncBytes),
+    /// Changes first sent in the unanswered messages.
+    pub(crate) sent: Vec<automerge::ChangeHash>,
+}
+
+impl SyncSession {
+    /// `reply_backoff`: waits for sync answers (`Timing::sync_reply_backoff`).
+    pub(crate) fn new(reply_backoff: Backoff) -> Self {
+        Self {
+            state: sync::State::new(),
+            sending: None,
+            retry_at: None,
+            send_backoff: Backoff::new(SEND_RETRY, SEND_RETRY_MAX),
+            reply: None,
+            reply_backoff,
+        }
+    }
+
+    /// Whether we leave the first message of this session to the peer.
+    ///
+    /// After a reset, both sides usually open at once, and the openings cross. Between
+    /// peers that share history that costs nothing. But a peer with an empty document
+    /// answers our opening with empty heads, which Automerge takes for "lost
+    /// everything", so the whole document would go out a second time. So a ready node
+    /// never opens towards a peer that isn't ready (one that is still catching up); that
+    /// peer opens itself.
+    pub(crate) fn leave_opening(&self, we_ready: bool, peer_ready: Option<bool>) -> bool {
+        self.state.their_heads.is_none() && we_ready && peer_ready == Some(false)
+    }
+
+    /// Our Mycelium took the message.
+    pub(crate) fn send_succeeded(&mut self) -> bool {
+        self.retry_at = None;
+        self.send_backoff.reset();
+        self.sending
+            .take()
+            .is_some_and(|sending| sending.regenerate)
+    }
+
+    /// Our Mycelium refused the message: as far as the peer is concerned, it was never
+    /// sent. Takes back its changes (they go out with the next message), clears
+    /// `in_flight`, and returns how long to wait before generating again. None if a reset
+    /// already discarded what was being sent.
+    pub(crate) fn send_failed(&mut self, now: Instant) -> Option<Duration> {
+        let sending = self.sending.take()?;
+        for hash in &sending.sent {
+            self.state.sent_hashes.remove(hash);
+        }
+        if let Some(wait) = &mut self.reply {
+            wait.sent.retain(|hash| !sending.sent.contains(hash));
+        }
+        self.state.in_flight = false;
+        let delay = self.send_backoff.next();
+        self.retry_at = Some(now + delay);
+        Some(delay)
+    }
+
+    /// A sync message went out; `sent` are the changes it carried for the first time.
+    /// `bytes` are the sync bytes exchanged with the peer before it.
+    pub(crate) fn expect_reply(
+        &mut self,
+        sent: Vec<automerge::ChangeHash>,
+        now: Instant,
+        bytes: SyncBytes,
+    ) {
+        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_backoff.current(),
+                    started: (now, bytes),
+                    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_backoff.reset();
+    }
+
+    /// 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.
+    pub(super) 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_backoff.grow();
+        Some(wait.sent.len())
+    }
+}