diff options
Diffstat (limited to 'src/state.rs')
| -rw-r--r-- | src/state.rs | 147 |
1 files changed, 26 insertions, 121 deletions
diff --git a/src/state.rs b/src/state.rs index 64fa8ea..3abfd85 100644 --- a/src/state.rs +++ b/src/state.rs @@ -89,8 +89,8 @@ 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. + /// Whether we lead the pair (`swarm::pair`): we send the keepalive PINGs. Fixed for + /// the pair, so computed once. pub(crate) leads: bool, pub(crate) standing: Standing, @@ -138,79 +138,10 @@ pub(crate) struct SyncSession { /// How long to wait for it; doubles while answers stay out. pub(crate) reply_timeout: Duration, - /// Who sends the first message of this session. - pub(crate) opening: Opening, - - /// What started this session. - pub(crate) cause: SessionCause, - /// 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. - Ours, - /// We open, leader or not: 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). - /// Survives resets until the peer has answered, since only then do we know it - /// learned our epoch. - Announce, - /// 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 }, -} - -/// What started a sync session. Kept for the log: when the peer doesn't open within the -/// grace period, the cause usually tells why. -#[derive(Clone, Copy, Debug, PartialEq, Eq)] -pub(crate) enum SessionCause { - /// First session with a peer we just learned about. - NewPeer, - /// The peer sent HELLO: it (re)started its handshake with us. - PeerHello, - /// Its HELLO_REPLY completed our handshake. - HandshakeComplete, - /// The peer advertised another epoch than before. - PeerEpochChanged, - /// We switched epochs. - OwnEpochSwitch, - /// The peer went dormant. - PeerDormant, - /// The peer failed the swarm key check. - KeyRejected, -} - -impl SessionCause { - /// The likely reason the peer didn't open a session started this way. - pub(crate) fn missing_opening_hint(self) -> &'static str { - match self { - Self::NewPeer | Self::PeerHello | Self::HandshakeComplete => { - "the peer may not count us as a member yet (our HELLO_REPLY hasn't reached it), \ - or its opening was lost" - } - Self::PeerEpochChanged => { - "the peer switched epochs but doesn't know we're on its new one yet, or its \ - opening was lost" - } - Self::OwnEpochSwitch => { - "we switched epochs; the peer opens only once it learns that, and it may not \ - have yet" - } - Self::PeerDormant | Self::KeyRejected => { - "only our side reset, so the peer had no reason to open" - } - } - } -} - /// 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. @@ -227,47 +158,27 @@ 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); -/// How long the side that doesn't open a fresh sync waits for the other's opening -/// 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(leads: bool, cause: SessionCause) -> Self { + pub(crate) fn new() -> Self { Self { - cause, state: sync::State::new(), outbound: None, reply: None, reply_timeout: SYNC_REPLY_TIMEOUT, - 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 | Opening::Announce => None, - Opening::Theirs { until } => until - .checked_duration_since(now) - .filter(|wait| !wait.is_zero()), - } - } - - /// The peer should have opened by now but hasn't said anything: we open ourselves. - pub(crate) fn opening_overdue(&self, now: Instant) -> bool { - self.state.their_heads.is_none() - && matches!(self.opening, Opening::Theirs { until } if until <= now) + /// Whether we leave the first message of this session to the peer. + /// + /// After a reset, both sides usually open at once, and the openings cross. Between + /// peers that share history that costs nothing. But a peer with an empty document + /// answers our opening with empty heads, which Automerge takes for "lost + /// everything", so the whole document would go out a second time. So a ready node + /// never opens towards a peer that isn't ready (one that is still catching up); that + /// peer opens itself. + pub(crate) fn leave_opening(&self, we_ready: bool, peer_ready: Option<bool>) -> bool { + self.state.their_heads.is_none() && we_ready && peer_ready == Some(false) } /// A sync message went out; `sent` are the changes it carried for the first time. @@ -314,7 +225,7 @@ impl PeerState { leads, standing: Standing::default(), hello_state: HelloState::default(), - sync: SyncSession::new(leads, SessionCause::NewPeer), + sync: SyncSession::new(), liveness: Liveness::Active, added_at: Instant::now(), last_seen: None, @@ -330,13 +241,8 @@ 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, cause: SessionCause) { - let announce = - self.sync.opening == Opening::Announce && self.sync.state.their_heads.is_none(); - self.sync = SyncSession::new(self.leads, cause); - if announce { - self.sync.opening = Opening::Announce; - } + pub(crate) fn reset_sync(&mut self) { + self.sync = SyncSession::new(); } /// An overdue sync answer from an established, active peer (see @@ -374,7 +280,7 @@ impl PeerState { } self.liveness = Liveness::Dormant { since: now }; - self.reset_sync(SessionCause::PeerDormant); + self.reset_sync(); if !matches!(self.hello_state, HelloState::Sending { .. }) { self.hello_state = HelloState::AwaitingReply { @@ -782,22 +688,21 @@ impl AppState { self.publish_change(); // Tell peers about the new epoch now instead of with the next keepalive (up to a - // keepalive interval later). Peers already on it learn it from our sync opening, - // which goes out right away; everyone else gets a PING, so peers that never saw - // the commit start catching up (we don't sync with them, so nothing else would - // tell them). + // keepalive interval later). Ready peers already on it learn it from our sync + // opening, which goes out right away; everyone else gets a PING, so peers that + // never saw the commit start catching up (we don't sync with them, so nothing + // else would tell them). let local = self.local_epoch(); let mut announce = Vec::new(); for (id, peer_state_mutex) in &peers { let mut peer_state = peer_state_mutex.lock().await; // What it said about syncing with us was about our old epoch. peer_state.syncs_with_us = None; - let syncs_with_us = peer_state.liveness == Liveness::Active + let gets_our_opening = peer_state.liveness == Liveness::Active && matches!(peer_state.standing, Standing::Member { .. }) - && peer_state.epoch == Some(local); - if syncs_with_us { - peer_state.sync.opening = Opening::Announce; - } else { + && peer_state.epoch == Some(local) + && peer_state.reports_ready == Some(true); + if !gets_our_opening { announce.push(id); } } @@ -1067,7 +972,7 @@ impl Host for AppState { let now = Instant::now(); for peer in state.peer_states.blocking_read().values() { let mut peer = peer.blocking_lock(); - peer.reset_sync(SessionCause::OwnEpochSwitch); + peer.reset_sync(); // A fresh start, not an idle sync: no heal re-handshake for it. peer.last_sync_rx = Some(now); } |