drop pre-membership sync messages

The handshake resets the session anyway, and the sender's reply timeout
resends what still matters.

3 files changed, 10 insertions, 43 deletions

diff --git a/src/swarm/bootstrap.rs b/src/swarm/bootstrap.rs
index edadce2..27ffdd8 100644
--- a/src/swarm/bootstrap.rs
+++ b/src/swarm/bootstrap.rs
@@ -70,10 +70,8 @@ pub(crate) async fn bootstrap(state: &AppState, seeds: &[String]) -> Result<()>
                 let peer_state = peer_state_mutex.lock().await;
                 let validated_swarm = peer_state.standing.swarm_id();
                 peer_diagnostics.push(format!(
-                    "{peer}: validated_swarm={:?}, epoch={:?}, pending_inbound_sync={}",
-                    validated_swarm,
-                    peer_state.epoch,
-                    peer_state.pending_inbound_sync.len()
+                    "{peer}: validated_swarm={:?}, epoch={:?}",
+                    validated_swarm, peer_state.epoch,
                 ))
             }
             let local_swarm = state.swarm_id().await;
diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs
index 368c969..4c07d7d 100644
--- a/src/swarm/peer.rs
+++ b/src/swarm/peer.rs
@@ -473,11 +473,7 @@ pub(crate) async fn run_peer_worker(state: AppState) {
 
 // --- peer membership ---------------------------------------------------------
 
-async fn establish_peer(
-    state: &AppState,
-    peer: &str,
-    swarm_id: String,
-) -> Result<Vec<transport::SyncEnvelope>> {
+async fn establish_peer(state: &AppState, peer: &str, swarm_id: String) -> Result<()> {
     let expected_swarm = state.swarm_id().await;
 
     if let Some(expected) = &expected_swarm {
@@ -522,10 +518,8 @@ async fn establish_peer(
         info!(peer = %peer, "Handshake complete; sync state reset");
     }
 
-    let pending = std::mem::take(&mut peer_state.pending_inbound_sync);
-
-    info!(peer = %peer, swarm_id = %swarm_id, pending_messages = pending.len(), "Peer membership established");
-    Ok(pending)
+    info!(peer = %peer, swarm_id = %swarm_id, "Peer membership established");
+    Ok(())
 }
 
 #[instrument(skip(state), fields(peer = %peer))]
@@ -726,26 +720,13 @@ pub(crate) async fn handle_message(
                 .send(SyncEvent::PeerNeedsSync(peer.clone()));
         }
         transport::ParsedMessage::HelloReply(reply) => {
-            let pending = establish_peer(state, &peer, reply.swarm_id.clone()).await?;
+            establish_peer(state, &peer, reply.swarm_id.clone()).await?;
 
             discover_peers(state, &peer, &reply.peers).await;
 
             // Our swarm ID may have just become known: pay any replies we owe.
             flush_owed_hello_replies(state).await;
 
-            if !pending.is_empty() {
-                debug!(
-                    to = %peer,
-                    pending_messages = pending.len(),
-                    "Replaying pending sync messages"
-                );
-                for envelope in pending {
-                    if let Err(error) = handle_sync(state, &peer, envelope).await {
-                        warn!(peer = %peer, error = %format!("{error:#}"), "Discarding invalid pending sync envelope");
-                    }
-                }
-            }
-
             state
                 .sync_events
                 .send(SyncEvent::PeerValidated(peer.clone()))
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 04a174c..04cb31c 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -274,25 +274,13 @@ pub(crate) async fn handle_sync(
     let peer_state_mutex = state.get_or_create_peer_state(peer).await;
 
     let needs_hello = {
-        let mut peer_state = peer_state_mutex.lock().await;
+        let peer_state = peer_state_mutex.lock().await;
 
         match &peer_state.standing {
             Standing::Stranger | Standing::Admitted => {
-                let duplicate = peer_state.pending_inbound_sync.iter().any(|pending| {
-                    pending.message == envelope.message
-                        && pending.swarm_id == envelope.swarm_id
-                        && pending.epoch == envelope.epoch
-                });
-                if !duplicate {
-                    peer_state.pending_inbound_sync.push(envelope.clone());
-                    info!(
-                        peer = %peer,
-                        pending_count = peer_state.pending_inbound_sync.len(),
-                        "Queued sync envelope pending peer membership"
-                    );
-                } else {
-                    trace!(peer = %peer, "Ignoring duplicate pending sync envelope");
-                }
+                // Not processed: our handshake reply resets the session anyway, and the
+                // sender's reply timeout resends what still matters.
+                debug!(peer = %peer, "Dropping sync message from a peer that isn't a member yet");
                 true
             }
             Standing::Member { swarm_id } if swarm_id != &envelope.swarm_id => {