drain queued messages on startup

1 files changed, 48 insertions, 2 deletions

diff --git a/src/transport/mycelium.rs b/src/transport/mycelium.rs
index 4a8a5a5..4c03839 100644
--- a/src/transport/mycelium.rs
+++ b/src/transport/mycelium.rs
@@ -456,14 +456,60 @@ impl MyceliumTransport {
             .await
             .context("Querying local Mycelium identity")?;
 
-        Ok(Self {
+        let transport = Self {
             setup: tokio::sync::Mutex::new(setup),
             client,
             poll_client,
             messages_endpoint: rest_url.join("/api/v1/messages")?,
             pubkey_basepath: rest_url.join("/api/v1/pubkey/")?,
             local_id: admin.node_pubkey,
-        })
+        };
+        transport.discard_backlog().await;
+        Ok(transport)
+    }
+
+    /// Drops what Mycelium queued for us before we started. A Mycelium that outlives us
+    /// (an external one, or across our restart) keeps accepting messages meant for us:
+    /// HELLO retries, keepalives, sync messages of sessions that ended with our previous
+    /// run. Processing them would only replay stale handshakes. Everything that still
+    /// matters is resent: peers retry their HELLOs, and sync starts over after the
+    /// handshake. Runs before we send anything, so no answer to us is dropped.
+    ///
+    /// Bounded in time, not count: the backlog is whatever piled up while we were away,
+    /// which a local loop drains in well under a second even for large swarms. Whatever
+    /// is still arriving after that is current traffic (peers that don't know we're
+    /// back), and it is processed normally, exactly as without this step.
+    async fn discard_backlog(&self) {
+        const DRAIN_FOR: Duration = Duration::from_secs(2);
+        let deadline = std::time::Instant::now() + DRAIN_FOR;
+        let mut discarded = 0u64;
+        while std::time::Instant::now() < deadline {
+            let response = self
+                .client
+                .get(self.messages_endpoint.clone())
+                .query(&[("timeout", 0)])
+                .query(&[("topic", encode_topic(TOPIC_PREFIX))])
+                .send()
+                .await;
+            match response {
+                Ok(response) if response.status() == StatusCode::NO_CONTENT => break,
+                Ok(response) if response.status().is_success() => discarded += 1,
+                Ok(response) => {
+                    tracing::warn!(status = %response.status(), "Couldn't drain the Mycelium inbox");
+                    break;
+                }
+                Err(error) => {
+                    tracing::warn!(error = %error, "Couldn't drain the Mycelium inbox");
+                    break;
+                }
+            }
+        }
+        if discarded > 0 {
+            tracing::info!(
+                discarded,
+                "Discarded messages Mycelium queued for us before we started"
+            );
+        }
     }
 
     async fn lookup_pubkey(&self, ip: &std::net::Ipv6Addr) -> Result<String> {