1 files changed, 52 insertions, 11 deletions

diff --git a/src/state.rs b/src/state.rs
index 6d95bb5..49bc1a8 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -77,16 +77,21 @@ pub(crate) enum HelloState {
     Established,
 }
 
+/// A sync message on its way to our Mycelium (the HTTP request is in progress).
 #[derive(Clone)]
-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) 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);
+
 #[derive(Clone)]
 pub(crate) struct PeerState {
     /// Whether we lead the pair (`swarm::pair`): we send the keepalive PINGs. Fixed for
@@ -134,8 +139,14 @@ pub(crate) struct PeerState {
 pub(crate) struct SyncSession {
     pub(crate) state: sync::State,
 
-    /// Generated for this peer, waiting for delivery.
-    pub(crate) outbound: Option<PendingOutbound>,
+    /// 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) retry_delay: Duration,
 
     /// The answer we expect to our last sync message(s), if any.
     pub(crate) reply: Option<SyncReplyWait>,
@@ -163,7 +174,9 @@ impl SyncSession {
     pub(crate) fn new() -> Self {
         Self {
             state: sync::State::new(),
-            outbound: None,
+            sending: None,
+            retry_at: None,
+            retry_delay: SEND_RETRY,
             reply: None,
             reply_timeout: SYNC_REPLY_TIMEOUT,
         }
@@ -181,6 +194,34 @@ impl SyncSession {
         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.retry_delay = SEND_RETRY;
+        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.retry_delay;
+        self.retry_at = Some(now + delay);
+        self.retry_delay = (delay * 2).min(SEND_RETRY_MAX);
+        Some(delay)
+    }
+
     /// 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 {
@@ -240,8 +281,8 @@ 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.
+    /// Start Automerge sync with this peer from scratch. A message still being handed to
+    /// Mycelium was generated against the old state; its outcome is then ignored.
     pub(crate) fn reset_sync(&mut self) {
         self.sync = SyncSession::new();
     }