diff options
| -rw-r--r-- | docs/peers.md | 5 | ||||
| -rw-r--r-- | src/state/peer.rs | 52 | ||||
| -rw-r--r-- | src/swarm/peer/inbound.rs | 15 | ||||
| -rw-r--r-- | src/swarm/peer/keepalive.rs | 6 |
4 files changed, 75 insertions, 3 deletions
diff --git a/docs/peers.md b/docs/peers.md index be04b16..cacceb5 100644 --- a/docs/peers.md +++ b/docs/peers.md @@ -114,7 +114,10 @@ With a swarm key (`--swarm-key-file`), every HELLO and HELLO_REPLY carries until then everything else it sends is ignored, and a handshake without a valid proof gets no answer at all. -Retries: an unanswered HELLO is resent after 1 s, doubling up to 30 s. To an +Retries: an unanswered HELLO is resent after 1 s, doubling up to 30 s, but at once +(at most every second) when anything else comes from the peer meanwhile: it is up and +reaches us, so the HELLO or its reply was lost, and waiting out the backoff (or a +probe's) would only delay the handshake. To an unreachable peer (dormant, or three unanswered HELLOs) it is a probe: after `dormant_probe`, doubling up to 6 × `dormant_probe` (or `--dormant-probe-max`, which is every wait if shorter); members get probes even when they diff --git a/src/state/peer.rs b/src/state/peer.rs index ce7a3c3..32b35fc 100644 --- a/src/state/peer.rs +++ b/src/state/peer.rs @@ -165,6 +165,8 @@ pub(crate) struct PeerState { /// HELLOs sent in the current handshake attempt without hearing anything back. pub(crate) hello_attempts: u32, + /// When our last HELLO to it went out (`hello_again_now`). + last_hello_at: Option<Instant>, /// Since when we consider the peer unreachable (three unanswered HELLOs, or /// dormant). Only a message from it clears it (`mark_seen`): until then it is no /// neighbour candidate, and a member (by its HELLO_REPLY, or by the member list) gets @@ -288,6 +290,7 @@ impl PeerState { tree_writer: None, left: false, hello_attempts: 0, + last_hello_at: None, unreachable_since: None, incidents: Incidents::default(), whole_failed: None, @@ -377,12 +380,34 @@ impl PeerState { /// A HELLO went out (or failed to). After the third unanswered one, the peer counts /// as unreachable. pub(crate) fn hello_sent(&mut self, now: Instant) { + self.last_hello_at = Some(now); self.hello_attempts += 1; if self.hello_attempts >= UNREACHABLE_AFTER_HELLOS { self.unreachable_since.get_or_insert(now); } } + /// A message from it came while we wait for its HELLO_REPLY: it is up, and reaches + /// us, now. So our HELLO or its reply was lost: the HELLO goes again now (the peer + /// worker sends it), rather than when the backoff says, which after a few lost ones + /// is up to `HELLO_RETRY_MAX` (or a probe's wait) away; at most every `HELLO_RETRY`. + /// Not for a HELLO or HELLO_REPLY of its own, which the handshake handles (but for a + /// HELLO while we are joining, which we don't answer yet). Returns whether the HELLO + /// is due now. + pub(crate) fn hello_again_now(&mut self, now: Instant) -> bool { + let HelloState::AwaitingReply { next_attempt } = self.hello_state else { + return false; + }; + let lately = self + .last_hello_at + .is_some_and(|at| now.saturating_duration_since(at) < HELLO_RETRY); + if next_attempt <= now || lately { + return false; + } + self.hello_state = HelloState::AwaitingReply { next_attempt: now }; + true + } + /// Whether we sync with it on `epoch`: an active member on that epoch (what our /// STATUS tells it as `syncs_with_you`). pub(crate) fn syncs_on(&self, epoch: Epoch) -> bool { @@ -864,6 +889,33 @@ mod tests { assert!(peer.needs_heal(&heads(102), &heads(50), &timing, from + timing.resync_after)); } + /// A message from a peer whose HELLO_REPLY we await makes the HELLO due now, but not + /// within `HELLO_RETRY` of the last one, and not where none is awaited. + #[test] + fn hello_again_when_heard_from() { + let timing = Timing::from_keepalive(Duration::from_secs(60), Duration::from_secs(300)); + let mut peer = PeerState::new(true, &timing); + let start = peer.added_at; + assert!(!peer.hello_again_now(start), "no HELLO sent"); + peer.hello_sent(start); + let later = start + Duration::from_secs(30); + peer.hello_state = HelloState::AwaitingReply { + next_attempt: later, + }; + assert!(!peer.hello_again_now(start + HELLO_RETRY / 2), "just sent"); + let heard = start + HELLO_RETRY; + assert!(peer.hello_again_now(heard)); + assert_eq!( + peer.hello_state, + HelloState::AwaitingReply { + next_attempt: heard + } + ); + assert!(!peer.hello_again_now(heard), "due already"); + peer.hello_state = HelloState::Established; + assert!(!peer.hello_again_now(later)); + } + /// Keepalives are PINGs, every `status_refresh` a STATUS; only when quiet. #[test] fn keepalives_are_pings_and_now_and_then_a_status() { diff --git a/src/swarm/peer/inbound.rs b/src/swarm/peer/inbound.rs index 28066cb..cfafbd3 100644 --- a/src/swarm/peer/inbound.rs +++ b/src/swarm/peer/inbound.rs @@ -68,13 +68,19 @@ pub(crate) async fn handle_message( // one does). When we heard from it before tells a restarted peer how long it was // away (in our HELLO_REPLY). let peer_state = state.get_or_create_peer_state(&peer).await; + let mut hello_now = false; let (reactivated, previously_seen) = heard_from(state, &peer, &peer_state, Instant::now(), |peer_state| { - if let Some(run) = run { - peer_state.returned(run); + match run { + Some(run) => peer_state.returned(run), + // Up and reaching us, while we wait for its HELLO_REPLY. + None => hello_now = peer_state.hello_again_now(Instant::now()), } }) .await; + if hello_now { + state.peers_due.now(); + } let same_epoch = observe_peer_epoch(state, &peer, epoch, epoch_seq, run).await; match message { @@ -83,7 +89,12 @@ pub(crate) async fn handle_message( // restart. It becomes a member right away; our reply makes us one for it. let Some(swarm_id) = state.swarm_id().await else { // We are joining ourselves; the peer retries until we know our swarm. + // It is up and reaches us: if we wait for its HELLO_REPLY, our HELLO or + // the reply was lost, and ours goes again now. debug!(peer = %peer, "HELLO before we know our swarm; not answering yet"); + if peer_state.lock().await.hello_again_now(Instant::now()) { + state.peers_due.now(); + } return Ok(()); }; if let Some(theirs) = &hello.swarm_id diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs index 0f34eb8..aa34e70 100644 --- a/src/swarm/peer/keepalive.rs +++ b/src/swarm/peer/keepalive.rs @@ -84,7 +84,10 @@ pub(super) async fn note_alive(state: &AppState, peer: &str, message: &ParsedMes return; }; let ping = matches!(message, ParsedMessage::Ping(_)); + let mut hello_now = false; let (reactivated, _) = heard_from(state, peer, &peer_state, now, |peer_state| { + // Up and reaching us, while we wait for its HELLO_REPLY. + hello_now = peer_state.hello_again_now(now); if ping { // Its keepalive (or a path probe): the next is ours. peer_state.keepalive_received_at = Some(now); @@ -94,6 +97,9 @@ pub(super) async fn note_alive(state: &AppState, peer: &str, message: &ParsedMes } }) .await; + if hello_now { + state.peers_due.now(); + } if reactivated { ensure_handshake(state, peer).await; } |