diff options
| -rw-r--r-- | src/state.rs | 147 | ||||
| -rw-r--r-- | src/swarm/pair.rs | 4 | ||||
| -rw-r--r-- | src/swarm/peer.rs | 39 | ||||
| -rw-r--r-- | src/swarm/sync.rs | 27 | ||||
| -rw-r--r-- | src/transport/mod.rs | 4 | ||||
| -rw-r--r-- | src/transport/mycelium.rs | 6 |
6 files changed, 64 insertions, 163 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); } diff --git a/src/swarm/pair.rs b/src/swarm/pair.rs index 7200a91..c1a196b 100644 --- a/src/swarm/pair.rs +++ b/src/swarm/pair.rs @@ -1,5 +1,5 @@ -//! Which side of a pair of nodes leads, for duties only one side needs to do (opening -//! the sync after a reset, sending keepalive PINGs). +//! Which side of a pair of nodes leads, for duties only one side needs to do (sending +//! keepalive PINGs). //! //! Both sides decide on their own, from the two keys alone, and agree. The leader is //! chosen per pair by one bit of a hash of both keys, not by comparing the keys: a diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs index bf552e9..0ef42bf 100644 --- a/src/swarm/peer.rs +++ b/src/swarm/peer.rs @@ -12,7 +12,7 @@ use tracing::{debug, info, instrument, trace, warn}; use crate::document::replica::{Epoch, EpochRelation}; use crate::document::store::{self, StoreError}; -use crate::state::{AppState, HelloState, Liveness, PeerState, SessionCause, Standing}; +use crate::state::{AppState, HelloState, Liveness, PeerState, Standing}; use crate::swarm::sync::{SyncEvent, handle_sync}; use crate::transport::{self, ParsedMessage}; @@ -267,6 +267,7 @@ async fn transmit_hello(state: &AppState, peer: &str, delay: Duration) -> Result swarm_id, epoch: state.local_epoch(), auth, + ready: state.is_ready().await, }), ) .await; @@ -327,15 +328,15 @@ async fn send_hello_reply(state: &AppState, peer: &str) -> Result<()> { .as_ref() .map(|key| key.hello_reply_tag(&swarm_id, state.local_id(), peer)) .unwrap_or_default(); + let ready = state.is_ready().await; let reply = transport::HelloReply { swarm_id, peers, epoch: state.local_epoch(), auth, + ready, }; - let ready = state.is_ready().await; - debug!(to = %peer, ready, "Sending HELLO_REPLY"); state @@ -445,15 +446,16 @@ pub(crate) async fn run_peer_worker(state: AppState) { // --- peer membership --------------------------------------------------------- -/// Makes `peer` a member of `swarm_id`, after its valid HELLO (`cause: PeerHello`) or -/// HELLO_REPLY (`HandshakeComplete`). Either completes the handshake: one round trip -/// proves both sides to each other (the HELLO the sender's key, the reply the swarm ID -/// and the responder's key), so neither side needs a HELLO of its own. +/// Makes `peer` a member of `swarm_id`, after its valid HELLO (`from_hello`) or +/// HELLO_REPLY. Either completes the handshake: one round trip proves both sides to +/// each other (the HELLO the sender's key, the reply the swarm ID and the responder's +/// key), so neither side needs a HELLO of its own. `ready` is what the message said. async fn establish_peer( state: &AppState, peer: &str, swarm_id: String, - cause: SessionCause, + from_hello: bool, + ready: bool, ) -> Result<()> { let expected_swarm = state.swarm_id().await; @@ -493,10 +495,11 @@ async fn establish_peer( // Stop our retries and sync from a clean slate. A HELLO always resets: the peer // starts over with us. Duplicate replies (caused by our own retries) don't. - if cause == SessionCause::PeerHello || peer_state.hello_state != HelloState::Established { + peer_state.reports_ready = Some(ready); + if from_hello || peer_state.hello_state != HelloState::Established { peer_state.hello_state = HelloState::Established; - peer_state.reset_sync(cause); - info!(peer = %peer, ?cause, "Handshake complete; sync state reset"); + peer_state.reset_sync(); + info!(peer = %peer, from_hello, "Handshake complete; sync state reset"); } info!(peer = %peer, swarm_id = %swarm_id, "Peer membership established"); @@ -529,7 +532,7 @@ async fn observe_peer_epoch(state: &AppState, peer: &str, remote: Epoch) -> bool // The peer switched documents; our sync state describes its old one. Count // the restart as sync activity, so the keepalive heal doesn't re-handshake // (resetting both sides again) while this fresh sync is under way. - peer_state.reset_sync(SessionCause::PeerEpochChanged); + peer_state.reset_sync(); peer_state.last_sync_rx = Some(Instant::now()); // Whether it would sync with us was said about its old epoch. peer_state.syncs_with_us = None; @@ -620,7 +623,7 @@ async fn admit(state: &AppState, peer: &str, message: &ParsedMessage) -> bool { } peer_state.standing = Standing::Stranger; peer_state.hello_state = HelloState::NotSent; - peer_state.reset_sync(SessionCause::KeyRejected); + peer_state.reset_sync(); } // The one place an operator can see why a node can't join: rate limited, // since anyone can send us handshakes. @@ -695,7 +698,7 @@ pub(crate) async fn handle_message( bail!("Incompatible swarm at {peer}: ours is {swarm_id}, it claims {theirs}"); } } - establish_peer(state, &peer, swarm_id, SessionCause::PeerHello).await?; + establish_peer(state, &peer, swarm_id, true, hello.ready).await?; if let Err(error) = send_hello_reply(state, &peer).await { // The peer retries its HELLO. @@ -707,13 +710,7 @@ pub(crate) async fn handle_message( .send(SyncEvent::PeerValidated(peer.clone())); } transport::ParsedMessage::HelloReply(reply) => { - establish_peer( - state, - &peer, - reply.swarm_id.clone(), - SessionCause::HandshakeComplete, - ) - .await?; + establish_peer(state, &peer, reply.swarm_id.clone(), false, reply.ready).await?; discover_peers(state, &peer, &reply.peers).await; diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index 8a85b8c..c918fe8 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -41,6 +41,8 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool return Ok(()); }; + let we_ready = state.is_ready().await; + let validated_swarm = { let peer_state = peer_state_mutex.lock().await; @@ -80,27 +82,14 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool return Ok(()); } - let now = Instant::now(); - if let Some(wait) = peer_state.sync.opening_wait(now) { - trace!( - wait_ms = wait.as_millis(), - "Leaving the opening sync message to the peer" - ); - let tx = state.sync_events.clone(); - let peer = peer.to_owned(); - tokio::spawn(async move { - tokio::time::sleep(wait).await; - let _ = tx.send(SyncEvent::PeerNeedsSync(peer)); - }); + if peer_state + .sync + .leave_opening(we_ready, peer_state.reports_ready) + { + // Its own opening (or anything else from it) brings us back here. + trace!("Leaving the opening sync message to the peer, which isn't ready"); return Ok(()); } - if peer_state.sync.opening_overdue(now) { - debug!( - cause = ?peer_state.sync.cause, - likely = peer_state.sync.cause.missing_opening_hint(), - "The peer didn't open the sync within the grace period; opening ourselves" - ); - } } swarm_id.clone() diff --git a/src/transport/mod.rs b/src/transport/mod.rs index d96b059..83c4e8d 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -118,6 +118,8 @@ pub struct HelloRequest { pub epoch: Epoch, /// Proof of the swarm key (`swarm::auth`); empty without one. pub auth: Vec<u8>, + /// The sender is ready (same meaning as `SyncEnvelope::ready`). + pub ready: bool, } #[derive(Debug, Clone)] @@ -127,6 +129,8 @@ pub struct HelloReply { pub epoch: Epoch, /// Proof of the swarm key (`swarm::auth`); empty without one. pub auth: Vec<u8>, + /// The sender is ready. + pub ready: bool, } #[derive(Debug, Clone)] diff --git a/src/transport/mycelium.rs b/src/transport/mycelium.rs index 2406312..4a8a5a5 100644 --- a/src/transport/mycelium.rs +++ b/src/transport/mycelium.rs @@ -830,6 +830,7 @@ pub struct HelloRequestShadow { swarm_id: Option<String>, epoch: EpochShadow, auth: ByteBuf, + ready: bool, } #[derive(Serialize, Deserialize)] @@ -842,6 +843,7 @@ pub struct HelloReplyShadow<'a> { epoch: EpochShadow, #[serde(default)] auth: ByteBuf, + ready: bool, } #[derive(Serialize, Deserialize)] @@ -891,6 +893,7 @@ pub fn hello_request_to_bytes(msg: &HelloRequest) -> Result<Vec<u8>, rmp_serde:: swarm_id: msg.swarm_id.clone(), epoch: epoch_to_shadow(&msg.epoch), auth: ByteBuf::from(msg.auth.clone()), + ready: msg.ready, }) } pub fn bytes_to_hello_request(bytes: &[u8]) -> Result<HelloRequest, TransportError> { @@ -899,6 +902,7 @@ pub fn bytes_to_hello_request(bytes: &[u8]) -> Result<HelloRequest, TransportErr swarm_id: shadow.swarm_id, epoch: epoch_from_shadow(shadow.epoch)?, auth: shadow.auth.into_vec(), + ready: shadow.ready, }) } @@ -909,6 +913,7 @@ pub fn hello_reply_to_bytes(msg: &HelloReply) -> Result<Vec<u8>, rmp_serde::enco peers: msg.peers.clone(), // Vec must be cloned for serialization wrapper matching epoch: epoch_to_shadow(&msg.epoch), auth: ByteBuf::from(msg.auth.clone()), + ready: msg.ready, }; rmp_serde::to_vec(&shadow) } @@ -919,6 +924,7 @@ pub fn bytes_to_hello_reply(bytes: &[u8]) -> Result<HelloReply, TransportError> peers: shadow.peers, epoch: epoch_from_shadow(shadow.epoch)?, auth: shadow.auth.into_vec(), + ready: shadow.ready, }) } |