diff options
Diffstat (limited to 'src/swarm')
| -rw-r--r-- | src/swarm/pair.rs | 4 | ||||
| -rw-r--r-- | src/swarm/peer.rs | 39 | ||||
| -rw-r--r-- | src/swarm/sync.rs | 27 |
3 files changed, 28 insertions, 42 deletions
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() |