diff options
Diffstat (limited to 'src/swarm')
| -rw-r--r-- | src/swarm/peer.rs | 2 | ||||
| -rw-r--r-- | src/swarm/sync.rs | 58 |
2 files changed, 33 insertions, 27 deletions
diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs index 9de0ce0..adde444 100644 --- a/src/swarm/peer.rs +++ b/src/swarm/peer.rs @@ -139,7 +139,7 @@ async fn heal_if_diverged( // Same document: whatever answer we were waiting for, there is nothing left // to send. (Automerge answers only when it has something to say, so a sync // message to a peer that already has everything gets no reply at all.) - peer_state.sync_in_step(); + peer_state.sync.reply_settled(); } peer_state.needs_heal( &local.heads_digest, diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index 22d6600..91c6eee 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -94,10 +94,10 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool let existing = { let mut guard = peer_state_mutex.lock().await; let peer_state = &mut *guard; - match &peer_state.pending_outbound_sync { + match &mut peer_state.sync.outbound { Some(entry) if retry_delivery => Some((entry.message.clone(), entry.next_attempt)), - Some(_) => { - peer_state.needs_generate = true; + Some(entry) => { + entry.regenerate = true; return Ok(()); } None => None, @@ -125,29 +125,29 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool let (epoch, message) = { let mut replica = doc.blocking_write(); let mut peer_state = psm.blocking_lock(); - peer_state.needs_generate = false; let epoch = replica.epoch; - let sent_before = peer_state.sync_state.sent_hashes.clone(); + let sent_before = peer_state.sync.state.sent_hashes.clone(); let message = { let sync_doc = replica.doc.sync(); - sync_doc.generate_sync_message(&mut peer_state.sync_state) + sync_doc.generate_sync_message(&mut peer_state.sync.state) }; // A peer whose heads are ours already has everything; Automerge // answers only when it has something to say, so it will stay // silent. Waiting for an answer would only end in a pointless // resend. - let in_step = peer_state.sync_state.their_heads.as_ref() + let in_step = peer_state.sync.state.their_heads.as_ref() == Some(&replica.doc.get_heads()); if in_step { - peer_state.sync_in_step(); + peer_state.sync.reply_settled(); } else if message.is_some() { let sent = peer_state - .sync_state + .sync + .state .sent_hashes .difference(&sent_before) .copied() .collect(); - peer_state.expect_sync_reply(sent, Instant::now()); + peer_state.sync.expect_reply(sent, Instant::now()); } if message.is_some() { // It carries our epoch. @@ -175,10 +175,11 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool return Ok(()); }; - peer_state_mutex.lock().await.pending_outbound_sync = Some(PendingOutbound { + peer_state_mutex.lock().await.sync.outbound = Some(PendingOutbound { message: encoded.clone(), retry_delay: INITIAL_SYNC_RETRY, next_attempt: Instant::now(), + regenerate: false, }); encoded @@ -190,8 +191,11 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool Ok(()) => { let regenerate = { let mut peer_state = peer_state_mutex.lock().await; - peer_state.pending_outbound_sync = None; - std::mem::take(&mut peer_state.needs_generate) + peer_state + .sync + .outbound + .take() + .is_some_and(|entry| entry.regenerate) }; trace!(regenerate, "Outbound sync message accepted by Mycelium"); @@ -209,7 +213,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool let delay = { let mut peer_state = peer_state_mutex.lock().await; - let Some(ref mut entry) = peer_state.pending_outbound_sync else { + let Some(ref mut entry) = peer_state.sync.outbound else { debug!("Pending envelope discarded by a sync reset during delivery"); return Ok(()); }; @@ -322,26 +326,27 @@ pub(crate) async fn handle_sync( let mut peer_state = peer_state_mutex.blocking_lock(); let heads_before = replica.doc.get_heads(); - let their_heads_before = peer_state.sync_state.their_heads.clone(); + let their_heads_before = peer_state.sync.state.their_heads.clone(); replica .doc .sync() - .receive_sync_message(&mut peer_state.sync_state, message) + .receive_sync_message(&mut peer_state.sync.state, message) .context("Automerge synchronization failed")?; peer_state.last_sync_rx = Some(Instant::now()); - peer_state.sync_reply_received(); + peer_state.sync.reply_settled(); let heads_after = replica.doc.get_heads(); let unproductive = heads_before == heads_after - && peer_state.sync_state.their_heads == their_heads_before - && peer_state.sync_state.their_heads.as_ref() != Some(&heads_after); + && peer_state.sync.state.their_heads == their_heads_before + && peer_state.sync.state.their_heads.as_ref() != Some(&heads_after); break_sync_stall(&peer, &mut peer_state, unproductive); Ok(Some( check_caught_up && peer_state - .sync_state + .sync + .state .their_heads .as_ref() .is_some_and(|heads| { @@ -393,17 +398,18 @@ const STALL_RESET: u32 = 32; /// everything the peer's bloom filter lacks. /// * If that doesn't help either, reset both sides through a fresh handshake. fn break_sync_stall(peer: &str, peer_state: &mut PeerState, unproductive: bool) { + let session = &mut peer_state.sync; if !unproductive { - peer_state.stalled_exchanges = 0; + session.stalled_exchanges = 0; return; } - peer_state.stalled_exchanges += 1; + session.stalled_exchanges += 1; - let sync_state = &mut peer_state.sync_state; + let sync_state = &mut session.state; // Not on the first unproductive exchange: when messages cross, the peer asks for a // change that is merely still in flight, and resending it would only duplicate it. - if peer_state.stalled_exchanges >= STALL_RESEND_NEEDED { + if session.stalled_exchanges >= STALL_RESEND_NEEDED { if let Some(need) = &sync_state.their_need { let before = sync_state.sent_hashes.len(); sync_state.sent_hashes.retain(|hash| !need.contains(hash)); @@ -414,7 +420,7 @@ fn break_sync_stall(peer: &str, peer_state: &mut PeerState, unproductive: bool) } } - match peer_state.stalled_exchanges { + match session.stalled_exchanges { STALL_CLEAR_SENT => { warn!( peer = %peer, @@ -427,7 +433,7 @@ fn break_sync_stall(peer: &str, peer_state: &mut PeerState, unproductive: bool) } STALL_RESET => { warn!(peer = %peer, "Sync is still stalled; resetting it on both sides"); - peer_state.stalled_exchanges = 0; + session.stalled_exchanges = 0; peer_state.request_resync(); } _ => {} |