drop pre-membership sync messages

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

1 files changed, 4 insertions, 23 deletions

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()))