diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/state.rs | 4 | ||||
| -rw-r--r-- | src/swarm/bootstrap.rs | 6 | ||||
| -rw-r--r-- | src/swarm/peer.rs | 27 | ||||
| -rw-r--r-- | src/swarm/sync.rs | 20 |
4 files changed, 10 insertions, 47 deletions
diff --git a/src/state.rs b/src/state.rs index 4d82150..542cadb 100644 --- a/src/state.rs +++ b/src/state.rs @@ -100,9 +100,6 @@ pub(crate) struct PeerState { /// or sending failed). pub(crate) reply_owed: bool, - /// Received from this peer before it became a member, waiting to be processed. - pub(crate) pending_inbound_sync: Vec<transport::SyncEnvelope>, - /// Everything about the current Automerge sync with this peer; a reset replaces it. pub(crate) sync: SyncSession, @@ -322,7 +319,6 @@ impl PeerState { standing: Standing::default(), hello_state: HelloState::default(), reply_owed: false, - pending_inbound_sync: Vec::new(), sync: SyncSession::new(leads, SessionCause::NewPeer), liveness: Liveness::Active, added_at: Instant::now(), 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 => { |