diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/api/peers.rs | 2 | ||||
| -rw-r--r-- | src/state.rs | 171 | ||||
| -rw-r--r-- | src/swarm/peer.rs | 2 | ||||
| -rw-r--r-- | src/swarm/sync.rs | 58 |
4 files changed, 124 insertions, 109 deletions
diff --git a/src/api/peers.rs b/src/api/peers.rs index 6637da4..ce41776 100644 --- a/src/api/peers.rs +++ b/src/api/peers.rs @@ -119,7 +119,7 @@ pub(crate) async fn list_peers( .last_seen .map(|t| now.saturating_duration_since(t).as_secs()), dormant_for_secs, - has_pending_delivery: ps.pending_outbound_sync.is_some(), + has_pending_delivery: ps.sync.outbound.is_some(), epoch: ps.epoch.map(EpochInfo::from), same_epoch: ps.epoch.map(|epoch| epoch == local_epoch), }); diff --git a/src/state.rs b/src/state.rs index 33bc9b5..24ae2fb 100644 --- a/src/state.rs +++ b/src/state.rs @@ -82,34 +82,33 @@ pub(crate) struct PendingOutbound { pub(crate) message: transport::EncodedMessage, pub(crate) retry_delay: Duration, pub(crate) next_attempt: Instant, + /// A sync pass was requested while this was pending. Once it is delivered, the peer + /// must be enqueued again so the deferred changes go out. + pub(crate) regenerate: bool, } #[derive(Clone)] pub(crate) struct PeerState { pub(crate) standing: Standing, - pub(crate) sync_state: sync::State, pub(crate) hello_state: HelloState, /// The peer sent HELLO; we still owe it a HELLO_REPLY (e.g. we had no swarm ID yet, /// or sending failed). pub(crate) reply_owed: bool, - /// Received from this peer, waiting to be processed. + /// Received from this peer before it became a member, waiting to be processed. pub(crate) pending_inbound_sync: Vec<transport::SyncEnvelope>, - /// Generated for this peer, waiting for delivery. - pub(crate) pending_outbound_sync: Option<PendingOutbound>, - - /// A sync pass was requested while an envelope was pending. Once that envelope - /// is delivered, the peer must be re-enqueued so the deferred changes go out. - pub(crate) needs_generate: bool, + /// Everything about the current Automerge sync with this peer; a reset replaces it. + pub(crate) sync: SyncSession, pub(crate) liveness: Liveness, pub(crate) added_at: Instant, pub(crate) last_seen: Option<Instant>, pub(crate) last_ping_at: Option<Instant>, - /// Last time a sync message from this peer was applied. + /// Last time a sync message from this peer was applied (survives resets: the + /// keepalive heal looks at it). pub(crate) last_sync_rx: Option<Instant>, /// Last self-healing re-handshake (cooldown). @@ -118,17 +117,6 @@ pub(crate) struct PeerState { /// Epoch the peer advertised in its most recent message. pub(crate) epoch: Option<Epoch>, - /// 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, - - /// When sync with this peer last started from scratch. - pub(crate) sync_reset_at: Instant, - /// We switched epochs, and this peer, already on our new epoch, still has our old one /// on record: it won't sync with us until a message tells it otherwise. Our sync /// opening does that, so we open without waiting for the pair's leader, and the @@ -136,6 +124,27 @@ pub(crate) struct PeerState { pub(crate) epoch_unannounced: bool, } +/// One Automerge sync with a peer, from a reset to the next. Everything a reset must +/// forget lives here, so a reset is `SyncSession::new()` and can't miss a field. +#[derive(Clone)] +pub(crate) struct SyncSession { + pub(crate) state: sync::State, + + /// Generated for this peer, waiting for delivery. + pub(crate) outbound: Option<PendingOutbound>, + + /// The answer we expect to our last sync message(s), if any. + pub(crate) reply: Option<SyncReplyWait>, + /// How long to wait for it; doubles while answers stay out. + pub(crate) reply_timeout: Duration, + + /// When this session started. + pub(crate) started_at: Instant, + + /// Consecutive exchanges that changed nothing (livelock detection). + pub(crate) stalled_exchanges: u32, +} + /// 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. @@ -156,28 +165,72 @@ const SYNC_REPLY_TIMEOUT_MAX: Duration = Duration::from_secs(300); /// before opening itself (the other one may not have reset, or its message got lost). const SYNC_OPENING_GRACE: Duration = Duration::from_secs(3); +impl SyncSession { + pub(crate) fn new() -> Self { + Self { + state: sync::State::new(), + outbound: None, + reply: None, + reply_timeout: SYNC_REPLY_TIMEOUT, + started_at: Instant::now(), + stalled_exchanges: 0, + } + } + + /// A sync message went out; `sent` are the changes it carried for the first time. + pub(crate) fn expect_reply(&mut self, sent: Vec<automerge::ChangeHash>, now: Instant) { + match &mut self.reply { + // Still waiting for an earlier one: keep its deadline, remember these too. + Some(wait) => wait.sent.extend(sent), + None => { + self.reply = Some(SyncReplyWait { + deadline: now + self.reply_timeout, + sent, + }); + } + } + } + + /// No answer is due any more: the peer answered, or keepalives show both documents + /// are the same. + pub(crate) fn reply_settled(&mut self) { + self.reply = None; + self.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. + fn take_overdue_reply(&mut self, now: Instant) -> Option<usize> { + if self.reply.as_ref().is_none_or(|wait| now < wait.deadline) { + return None; + } + let wait = self.reply.take()?; + for hash in &wait.sent { + self.state.sent_hashes.remove(hash); + } + self.state.in_flight = false; + self.reply_timeout = (self.reply_timeout * 2).min(SYNC_REPLY_TIMEOUT_MAX); + Some(wait.sent.len()) + } +} + impl Default for PeerState { fn default() -> Self { Self { standing: Standing::default(), - sync_state: sync::State::new(), hello_state: HelloState::default(), reply_owed: false, pending_inbound_sync: Vec::new(), - pending_outbound_sync: None, - needs_generate: false, + sync: SyncSession::new(), liveness: Liveness::Active, added_at: Instant::now(), last_seen: None, last_ping_at: None, last_sync_rx: None, last_resync_at: None, - stalled_exchanges: 0, - sync_reply: None, - sync_reply_timeout: SYNC_REPLY_TIMEOUT, - sync_reset_at: Instant::now(), - epoch_unannounced: false, epoch: None, + epoch_unannounced: false, } } } @@ -186,78 +239,34 @@ impl PeerState { /// Start Automerge sync with this peer from scratch. Any envelope generated against /// the old state is discarded; a stale RetryDelivery event is then a no-op. pub(crate) fn reset_sync(&mut self) { - 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; - self.sync_reset_at = Instant::now(); + self.sync = SyncSession::new(); } /// After a reset, usually both sides do it and both open the sync at once. The /// openings cross, and each side answers the other's: harmless between peers that /// share history, but an empty peer answers with empty heads, which Automerge takes /// for "lost everything", so the whole document goes out a second time. Only the - /// side that leads the pair (`swarm::pair`) opens; the other waits for its opening, up to - /// `SYNC_OPENING_GRACE`. Returns how much longer to wait, if at all. + /// side that leads the pair (`swarm::pair`) opens; the other waits for its opening, + /// up to `SYNC_OPENING_GRACE`. Returns how much longer to wait, if at all. /// /// After our epoch switch, we open anyway (`epoch_unannounced`): the peer can't, as /// it still thinks we are on another epoch. pub(crate) fn opening_wait(&self, we_open: bool, now: Instant) -> Option<Duration> { - if we_open || self.epoch_unannounced || self.sync_state.their_heads.is_some() { + if we_open || self.epoch_unannounced || self.sync.state.their_heads.is_some() { return None; } SYNC_OPENING_GRACE - .checked_sub(now.saturating_duration_since(self.sync_reset_at)) + .checked_sub(now.saturating_duration_since(self.sync.started_at)) .filter(|wait| !wait.is_zero()) } - /// 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, - }); - } - } - } - - /// Keepalives show both documents are the same: stop waiting for an answer. - pub(crate) fn sync_in_step(&mut self) { - self.sync_reply = None; - self.sync_reply_timeout = SYNC_REPLY_TIMEOUT; - } - - /// 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. + /// An overdue sync answer from an established, active peer (see + /// `SyncSession::take_overdue_reply`). 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) - { + if self.liveness != Liveness::Active || self.hello_state != HelloState::Established { 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()) + self.sync.take_overdue_reply(now) } fn quiet_for(&self, now: Instant) -> Duration { 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(); } _ => {} |