diff options
| -rw-r--r-- | src/state.rs | 111 | ||||
| -rw-r--r-- | src/swarm/peer.rs | 5 | ||||
| -rw-r--r-- | src/swarm/sync.rs | 7 |
3 files changed, 67 insertions, 56 deletions
diff --git a/src/state.rs b/src/state.rs index 24ae2fb..eaa6562 100644 --- a/src/state.rs +++ b/src/state.rs @@ -89,6 +89,10 @@ pub(crate) struct PendingOutbound { #[derive(Clone)] pub(crate) struct PeerState { + /// Whether we lead the pair (`swarm::pair`): we open sync after a reset and send the + /// keepalive PINGs. Fixed for the pair, so computed once. + pub(crate) leads: bool, + pub(crate) standing: Standing, pub(crate) hello_state: HelloState, @@ -116,16 +120,10 @@ pub(crate) struct PeerState { /// Epoch the peer advertised in its most recent message. pub(crate) epoch: Option<Epoch>, - - /// 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 - /// epoch announcement PING is skipped. Cleared once a sync message is generated. - 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. +/// forget lives here, so a reset replaces it whole and can't miss a field. #[derive(Clone)] pub(crate) struct SyncSession { pub(crate) state: sync::State, @@ -138,13 +136,30 @@ pub(crate) struct SyncSession { /// 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, + /// Who sends the first message of this session. + pub(crate) opening: Opening, /// Consecutive exchanges that changed nothing (livelock detection). pub(crate) stalled_exchanges: u32, } +/// Who opens a sync session. After a reset, usually both sides do it and both would +/// open 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. So only +/// the side that leads the pair opens. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum Opening { + /// We open: we lead the pair, or we just switched epochs and the 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, and our opening does (no epoch announcement PING needed). A + /// reset before the opening goes out falls back to the grace period below. + Ours, + /// The peer opens. If nothing came by `until`, we open anyway: the peer may not have + /// reset, or its opening got lost. + Theirs { until: Instant }, +} + /// 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. @@ -166,17 +181,37 @@ const SYNC_REPLY_TIMEOUT_MAX: Duration = Duration::from_secs(300); const SYNC_OPENING_GRACE: Duration = Duration::from_secs(3); impl SyncSession { - pub(crate) fn new() -> Self { + pub(crate) fn new(leads: bool) -> Self { Self { state: sync::State::new(), outbound: None, reply: None, reply_timeout: SYNC_REPLY_TIMEOUT, - started_at: Instant::now(), + opening: if leads { + Opening::Ours + } else { + Opening::Theirs { + until: Instant::now() + SYNC_OPENING_GRACE, + } + }, stalled_exchanges: 0, } } + /// How much longer to hold back our opening, if at all. Only the first message + /// counts: once the peer has said anything, we answer as usual. + pub(crate) fn opening_wait(&self, now: Instant) -> Option<Duration> { + if self.state.their_heads.is_some() { + return None; + } + match self.opening { + Opening::Ours => None, + Opening::Theirs { until } => until + .checked_duration_since(now) + .filter(|wait| !wait.is_zero()), + } + } + /// 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 { @@ -215,14 +250,15 @@ impl SyncSession { } } -impl Default for PeerState { - fn default() -> Self { +impl PeerState { + pub(crate) fn new(leads: bool) -> Self { Self { + leads, standing: Standing::default(), hello_state: HelloState::default(), reply_owed: false, pending_inbound_sync: Vec::new(), - sync: SyncSession::new(), + sync: SyncSession::new(leads), liveness: Liveness::Active, added_at: Instant::now(), last_seen: None, @@ -230,34 +266,13 @@ impl Default for PeerState { last_sync_rx: None, last_resync_at: None, epoch: None, - epoch_unannounced: false, } } -} -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 = 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. - /// - /// 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() { - return None; - } - SYNC_OPENING_GRACE - .checked_sub(now.saturating_duration_since(self.sync.started_at)) - .filter(|wait| !wait.is_zero()) + self.sync = SyncSession::new(self.leads); } /// An overdue sync answer from an established, active peer (see @@ -309,19 +324,14 @@ impl PeerState { /// True (and records the ping) if an established, quiet peer needs a keepalive. /// /// One PING and its PONG tell both sides that the other is alive, so only one side of - /// a pair needs to ping: the one leading the pair (`we_ping`, `swarm::pair`). Each of its PINGs + /// a pair needs to ping: the one leading the pair (`leads`). Each of its PINGs /// makes the other side's peer look recently heard from, so the other side never /// pings while it does. If the pings stop, the other side pings once it has heard /// nothing for half an interval more. Without that, both sides would get due at about /// the same time (each one's quiet time starts with the same exchange) and their /// PINGs would often cross. - pub(crate) fn take_keepalive_due( - &mut self, - timing: &Timing, - we_ping: bool, - now: Instant, - ) -> bool { - let quiet_enough = if we_ping { + pub(crate) fn take_keepalive_due(&mut self, timing: &Timing, now: Instant) -> bool { + let quiet_enough = if self.leads { timing.keepalive } else { timing.keepalive + timing.keepalive / 2 @@ -572,10 +582,16 @@ impl AppState { Arc::clone( peers .entry(pk.to_owned()) - .or_insert_with(|| Arc::new(Mutex::new(PeerState::default()))), + .or_insert_with(|| self.new_peer_state(pk)), ) } + /// State for a peer we haven't known so far (not inserted). + pub(crate) fn new_peer_state(&self, pk: &str) -> Arc<Mutex<PeerState>> { + let leads = crate::swarm::pair::leads(self.local_id(), pk); + Arc::new(Mutex::new(PeerState::new(leads))) + } + pub(crate) fn notify_store_changed(&self) { self.change_tx.send_modify(|version| { *version = version.wrapping_add(1); @@ -693,8 +709,9 @@ impl AppState { let syncs_with_us = peer_state.liveness == Liveness::Active && matches!(peer_state.standing, Standing::Member { .. }) && peer_state.epoch == Some(local); - peer_state.epoch_unannounced = syncs_with_us; - if !syncs_with_us { + if syncs_with_us { + peer_state.sync.opening = Opening::Ours; + } else { announce.push(id); } } diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs index adde444..836a967 100644 --- a/src/swarm/peer.rs +++ b/src/swarm/peer.rs @@ -171,7 +171,7 @@ pub(crate) async fn remember_peer(state: &AppState, peer: &str) { let mut peer_states = state.peer_states.write().await; if !peer_states.contains_key(peer) { - peer_states.entry(peer.to_owned()).or_default(); + peer_states.insert(peer.to_owned(), state.new_peer_state(peer)); info!( peer = %peer, peer_count = peer_states.len(), @@ -410,8 +410,7 @@ pub(crate) async fn run_peer_worker(state: AppState) { 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 we_ping = crate::swarm::pair::leads(state.local_id(), peer); - let ping_due = peer_state.take_keepalive_due(&state.timing, we_ping, now); + let ping_due = peer_state.take_keepalive_due(&state.timing, now); let unanswered = peer_state.take_overdue_sync_reply(now); (became_dormant, ping_due, unanswered) }; diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index 91c6eee..1335361 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -63,8 +63,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool }; if !retry_delivery { - let we_open = crate::swarm::pair::leads(state.local_id(), peer); - if let Some(wait) = peer_state.opening_wait(we_open, Instant::now()) { + if let Some(wait) = peer_state.sync.opening_wait(Instant::now()) { trace!( wait_ms = wait.as_millis(), "Leaving the opening sync message to the peer" @@ -149,10 +148,6 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool .collect(); peer_state.sync.expect_reply(sent, Instant::now()); } - if message.is_some() { - // It carries our epoch. - peer_state.epoch_unannounced = false; - } (epoch, message) }; let Some(message) = message else { |