diff options
| -rw-r--r-- | src/state.rs | 67 | ||||
| -rw-r--r-- | src/swarm/peer.rs | 12 | ||||
| -rw-r--r-- | src/swarm/sync.rs | 17 |
3 files changed, 90 insertions, 6 deletions
diff --git a/src/state.rs b/src/state.rs index e24c882..81a3e6a 100644 --- a/src/state.rs +++ b/src/state.rs @@ -105,8 +105,29 @@ pub(crate) struct PeerState { /// Consecutive sync exchanges that changed nothing (livelock detection). pub(crate) stalled_exchanges: u32, + + /// The answer we expect to our last sync message(s), if any. + pub(crate) sync_reply: Option<SyncReplyWait>, + /// How long to wait for it; doubles while answers stay out. + pub(crate) sync_reply_timeout: Duration, +} + +/// Sync messages we sent and expect an answer to. Automerge assumes reliable delivery: +/// a lost message leaves both sides waiting (`in_flight`), and changes it carried are +/// never sent again (`sent_hashes`). We resend after a timeout instead. +#[derive(Clone, Debug)] +pub(crate) struct SyncReplyWait { + pub(crate) deadline: Instant, + /// Changes first sent in the unanswered messages. + pub(crate) sent: Vec<automerge::ChangeHash>, } +/// First wait for a sync answer; generous, since Mycelium delivers large messages in +/// many small packets. +const SYNC_REPLY_TIMEOUT: Duration = Duration::from_secs(10); +/// Upper bound (Mycelium gives up on a message after five minutes). +const SYNC_REPLY_TIMEOUT_MAX: Duration = Duration::from_secs(300); + impl Default for PeerState { fn default() -> Self { Self { @@ -124,6 +145,8 @@ impl Default for PeerState { last_sync_rx: None, last_resync_at: None, stalled_exchanges: 0, + sync_reply: None, + sync_reply_timeout: SYNC_REPLY_TIMEOUT, epoch: None, } } @@ -136,6 +159,50 @@ impl PeerState { self.sync_state = sync::State::new(); self.pending_outbound_sync = None; self.needs_generate = false; + self.sync_reply = None; + self.sync_reply_timeout = SYNC_REPLY_TIMEOUT; + } + + /// A sync message went out; `sent` are the changes it carried for the first time. + pub(crate) fn expect_sync_reply(&mut self, sent: Vec<automerge::ChangeHash>, now: Instant) { + match &mut self.sync_reply { + // Still waiting for an earlier one: keep its deadline, remember these too. + Some(wait) => wait.sent.extend(sent), + None => { + self.sync_reply = Some(SyncReplyWait { + deadline: now + self.sync_reply_timeout, + sent, + }); + } + } + } + + /// The peer answered. + pub(crate) fn sync_reply_received(&mut self) { + self.sync_reply = None; + self.sync_reply_timeout = SYNC_REPLY_TIMEOUT; + } + + /// If the answer is overdue: forgets that the unanswered changes were sent and that + /// a message is in flight, so the next sync pass sends them again. Returns how many + /// changes that concerns. + pub(crate) fn take_overdue_sync_reply(&mut self, now: Instant) -> Option<usize> { + if self.liveness != Liveness::Active + || self.hello_state != HelloState::Established + || self + .sync_reply + .as_ref() + .is_none_or(|wait| now < wait.deadline) + { + return None; + } + let wait = self.sync_reply.take()?; + for hash in &wait.sent { + self.sync_state.sent_hashes.remove(hash); + } + self.sync_state.in_flight = false; + self.sync_reply_timeout = (self.sync_reply_timeout * 2).min(SYNC_REPLY_TIMEOUT_MAX); + Some(wait.sent.len()) } fn quiet_for(&self, now: Instant) -> Duration { diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs index b092b03..6184427 100644 --- a/src/swarm/peer.rs +++ b/src/swarm/peer.rs @@ -387,13 +387,21 @@ pub(crate) async fn run_peer_worker(state: AppState) { for (peer, peer_state_mutex) in &peers { let now = Instant::now(); - let (became_dormant, ping_due) = { + let (became_dormant, ping_due, unanswered) = { let mut peer_state = peer_state_mutex.lock().await; let became_dormant = ready && peer_state.check_dormancy(&state.timing, now); let ping_due = peer_state.take_keepalive_due(&state.timing, now); - (became_dormant, ping_due) + let unanswered = peer_state.take_overdue_sync_reply(now); + (became_dormant, ping_due, unanswered) }; + if let Some(changes) = unanswered { + debug!(peer = %peer, changes, "No answer to our sync message; sending again"); + let _ = state + .sync_events + .send(SyncEvent::PeerNeedsSync(peer.clone())); + } + if became_dormant { warn!( peer = %peer, diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index 115d5c0..04d5b15 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -109,11 +109,19 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool 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 sync_doc = replica.doc.sync(); - ( - epoch, - sync_doc.generate_sync_message(&mut peer_state.sync_state), - ) + let message = sync_doc.generate_sync_message(&mut peer_state.sync_state); + if message.is_some() { + let sent = peer_state + .sync_state + .sent_hashes + .difference(&sent_before) + .copied() + .collect(); + peer_state.expect_sync_reply(sent, Instant::now()); + } + (epoch, message) }; let Some(message) = message else { return Ok(None); @@ -290,6 +298,7 @@ pub(crate) async fn handle_sync( .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(); let heads_after = replica.doc.get_heads(); let unproductive = heads_before == heads_after |