diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/state/peer.rs | 39 | ||||
| -rw-r--r-- | src/swarm/links.rs | 15 | ||||
| -rw-r--r-- | src/swarm/peer/inbound.rs | 18 |
3 files changed, 59 insertions, 13 deletions
diff --git a/src/state/peer.rs b/src/state/peer.rs index 99f912b..f6e8786 100644 --- a/src/state/peer.rs +++ b/src/state/peer.rs @@ -137,6 +137,10 @@ pub(crate) struct PeerState { /// How many heads its most recent keepalive listed (None if it sent a digest). pub(crate) reported_heads: Option<usize>, + /// It said farewell (`say_farewell`): what else that run of it still sends while it + /// stops doesn't make it active again; a message from another run does (`returned`). + pub(crate) left: bool, + /// HELLOs sent in the current handshake attempt without hearing anything back. pub(crate) hello_attempts: u32, /// Since when we consider the peer unreachable (three unanswered HELLOs, or @@ -212,6 +216,7 @@ impl PeerState { reports_ready: None, syncs_with_us: None, reported_heads: None, + left: false, hello_attempts: 0, unreachable_since: None, incidents: Incidents::default(), @@ -244,9 +249,13 @@ impl PeerState { now.saturating_duration_since(self.last_seen.unwrap_or(self.added_at)) } - /// Records inbound activity. Returns true if the peer was dormant. + /// Records inbound activity. Returns true if the peer was dormant. From a peer that + /// said farewell (`left`), only that it was heard from. pub(crate) fn mark_seen(&mut self, now: Instant) -> bool { self.last_seen = Some(now); + if self.left { + return false; + } self.hello_attempts = 0; self.unreachable_since = None; self.probe_backoff.reset(); @@ -333,6 +342,7 @@ impl PeerState { /// The peer said farewell (it is shutting down): Dormant now, without waiting for /// its silence, and not counted as an incident. Returns true on the transition. pub(crate) fn say_farewell(&mut self, now: Instant) -> bool { + self.left = true; if self.liveness.is_dormant() { return false; } @@ -340,6 +350,14 @@ impl PeerState { true } + /// A message from `run` of the peer (a HELLO's or HELLO_REPLY's): a run other than + /// the one that said farewell is back (`left`). + pub(crate) fn returned(&mut self, run: crate::transport::RunId) { + if self.run_id != Some(run) { + self.left = false; + } + } + fn go_dormant(&mut self, now: Instant) { self.liveness = Liveness::Dormant { since: now }; self.unreachable_since.get_or_insert(now); @@ -488,6 +506,25 @@ mod tests { assert!(silent.liveness.is_dormant()); } + /// After its farewell, a peer stays dormant whatever its stopping run still sends; + /// its next run makes it active again. + #[test] + fn dormant_after_farewell_until_back() { + let timing = Timing::from_keepalive(Duration::from_secs(60), Duration::from_secs(300)); + let mut peer = PeerState::new(true, &timing); + let now = peer.added_at; + peer.run_id = Some([1; 16]); + peer.mark_seen(now); + assert!(peer.say_farewell(now)); + assert!(!peer.mark_seen(now), "a straggler from the same run"); + assert!(peer.liveness.is_dormant() && peer.is_unreachable()); + peer.returned([1; 16]); + assert!(!peer.mark_seen(now), "the same run again"); + peer.returned([2; 16]); + assert!(peer.mark_seen(now), "its next run"); + assert!(!peer.liveness.is_dormant() && !peer.is_unreachable()); + } + /// The two sides of a pair take turns: each sends every other interval. #[test] fn keepalives_take_turns() { diff --git a/src/swarm/links.rs b/src/swarm/links.rs index 18b2bdd..82cd991 100644 --- a/src/swarm/links.rs +++ b/src/swarm/links.rs @@ -1,8 +1,8 @@ //! How members are linked, for a map of the swarm (the dashboard, `api::dashboard`): a //! node asks a member with LINKS, and the member answers with a LINKS_REPLY listing its -//! neighbours, each with the transport its messages go over, whether only through a -//! relay, the round trip and whether the neighbour is dormant: as the member measures -//! them, which no other node can. +//! neighbours and the peers it holds dormant (`own_links`), each with the transport its +//! messages go over, whether only through a relay, the round trip and whether it is +//! dormant: as the member measures them, which no other node can. //! //! Every node answers, whatever its build: one with a dashboard asks any member, which //! may have none. Only members (in the member list) are answered, as the links name @@ -101,14 +101,17 @@ impl Links { } } -/// This node's links: its neighbours, as it measures them. +/// This node's links: its neighbours, and the peers it holds dormant, as it measures +/// them. A neighbour that goes dormant is one no more (it is unreachable now), but its +/// link stays listed, dormant, until it is heard from again: left out, the link would +/// show only as that peer last reported it, from before. pub(crate) async fn own_links(state: &AppState) -> Vec<Link> { let mut links = Vec::new(); for (peer, peer_state) in state.peers_snapshot().await { - if !topology::is_neighbour(state, &peer) { + let dormant = peer_state.lock().await.liveness.is_dormant(); + if !dormant && !topology::is_neighbour(state, &peer) { continue; } - let dormant = peer_state.lock().await.liveness.is_dormant(); let destination = state.transport.destination(&peer); let transport = destination .as_ref() diff --git a/src/swarm/peer/inbound.rs b/src/swarm/peer/inbound.rs index b4a96e4..b57feaf 100644 --- a/src/swarm/peer/inbound.rs +++ b/src/swarm/peer/inbound.rs @@ -58,17 +58,23 @@ pub(crate) async fn handle_message( remember_peer(state, &peer).await; - // Any inbound message proves liveness. 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 (reactivated, previously_seen) = - heard_from(state, &peer, &peer_state, Instant::now(), |_| {}).await; - let run = match &message { transport::ParsedMessage::Hello(hello) => Some(hello.run_id), transport::ParsedMessage::HelloReply(reply) => Some(reply.sender_run_id), _ => None, }; + + // Any inbound message proves liveness (but from a run that said farewell: its next + // 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 (reactivated, previously_seen) = + heard_from(state, &peer, &peer_state, Instant::now(), |peer_state| { + if let Some(run) = run { + peer_state.returned(run); + } + }) + .await; let same_epoch = observe_peer_epoch(state, &peer, epoch, epoch_seq, run).await; match message { |