farewell means dormant

10 files changed, 44 insertions, 10 deletions

diff --git a/src/simtest/identity.rs b/src/simtest/identity.rs
index e0dd6f2..6290ba7 100644
--- a/src/simtest/identity.rs
+++ b/src/simtest/identity.rs
@@ -109,6 +109,7 @@ fn status(node: &AppState) -> ParsedMessage {
         epoch_seq: 0,
         ready: false,
         syncs_with_you: false,
+        leaving: false,
     })
 }
 
diff --git a/src/simtest/timing.rs b/src/simtest/timing.rs
index 72c6c05..0752f2f 100644
--- a/src/simtest/timing.rs
+++ b/src/simtest/timing.rs
@@ -318,6 +318,7 @@ async fn late_epoch_does_not_flip_a_peers_epoch() {
             epoch_seq,
             ready: true,
             syncs_with_you: false,
+            leaving: false,
         })
     };
     let a = &a;
diff --git a/src/state/peer.rs b/src/state/peer.rs
index 923c6d2..dc1b0ec 100644
--- a/src/state/peer.rs
+++ b/src/state/peer.rs
@@ -304,9 +304,23 @@ impl PeerState {
         if self.liveness.is_dormant() || self.quiet_for(now) < timing.dormant_after {
             return false;
         }
+        self.incidents.dormancies += 1;
+        self.go_dormant(now);
+        true
+    }
+
+    /// 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 {
+        if self.liveness.is_dormant() {
+            return false;
+        }
+        self.go_dormant(now);
+        true
+    }
 
+    fn go_dormant(&mut self, now: Instant) {
         self.liveness = Liveness::Dormant { since: now };
-        self.incidents.dormancies += 1;
         self.unreachable_since.get_or_insert(now);
         self.reset_sync();
 
@@ -315,7 +329,6 @@ impl PeerState {
                 next_attempt: now + self.probe_backoff.next(),
             };
         }
-        true
     }
 
     /// The keepalive an established, quiet peer needs, if any (and records it): a PING,
diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs
index afc2b3d..367b1d9 100644
--- a/src/swarm/peer/keepalive.rs
+++ b/src/swarm/peer/keepalive.rs
@@ -121,6 +121,7 @@ pub(crate) fn announce(state: &AppState) {
 /// in time: our transport may be gone already.
 pub(crate) async fn farewell(state: &AppState) {
     state.stopping.store(true, Ordering::Release);
+    info!("Saying farewell to our neighbours");
     status_to_neighbours(state, Some(FAREWELL_TIMEOUT)).await;
 }
 
@@ -151,6 +152,7 @@ pub(crate) async fn local_status(state: &AppState, peer: &str) -> transport::Sta
         epoch_seq,
         ready: state.is_ready().await && !state.stopping.load(Ordering::Acquire),
         syncs_with_you,
+        leaving: state.stopping.load(Ordering::Acquire),
     }
 }
 
@@ -169,13 +171,19 @@ pub(super) async fn status_received(state: &AppState, peer: &str) {
 pub(super) async fn note_status(state: &AppState, peer: &str, remote: &transport::Status) {
     if let Some(peer_state) = state.get_peer_state(peer).await {
         let mut peer_state = peer_state.lock().await;
+        // Its farewell: dormant now, as if silent for long (a message from it, as its
+        // HELLO when it's back, makes it active again).
+        let left = remote.leaving && peer_state.say_farewell(Instant::now());
         let became_ready = remote.ready && peer_state.reports_ready != Some(true);
         peer_state.reports_ready = Some(remote.ready);
         let changed =
             peer_state.syncs_with_us.replace(remote.syncs_with_you) != Some(remote.syncs_with_you);
         peer_state.reported_heads = remote.heads.heads().map(<[_]>::len);
         drop(peer_state);
-        if changed {
+        if left {
+            info!(peer = %peer, "Peer said farewell; dormant until it is heard from again");
+        }
+        if changed || left {
             // A neighbour that only connects out stays one as long as it says so.
             state.topology_changed.notify_one();
         }
diff --git a/src/transport/libp2p.rs b/src/transport/libp2p.rs
index d023079..75792c5 100644
--- a/src/transport/libp2p.rs
+++ b/src/transport/libp2p.rs
@@ -2684,6 +2684,7 @@ mod tests {
             epoch_seq: 3,
             ready: true,
             syncs_with_you: false,
+            leaving: false,
         })
     }
 
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index d774402..7fbbd73 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -199,6 +199,9 @@ pub(crate) struct Status {
     /// The sender would sync with the recipient: counts it as an active member on its
     /// own epoch.
     pub(crate) syncs_with_you: bool,
+    /// The sender is shutting down: its farewell (`keepalive::farewell`). The
+    /// recipient holds it dormant at once, rather than after `dormant_after` of silence.
+    pub(crate) leaving: bool,
 }
 
 /// Cheap to clone: large parts are shared (`bytes::Bytes`).
diff --git a/src/transport/mycelium/mod.rs b/src/transport/mycelium/mod.rs
index 20abee8..04474fe 100644
--- a/src/transport/mycelium/mod.rs
+++ b/src/transport/mycelium/mod.rs
@@ -667,6 +667,7 @@ mod tests {
             epoch_seq: u32::MAX,
             ready: true,
             syncs_with_you: true,
+            leaving: false,
         };
         let payload = wire::encode(&ParsedMessage::StatusReply(status.clone())).unwrap();
         // Mycelium's message data: a length byte, the topic, then our payload.
diff --git a/src/transport/paths.rs b/src/transport/paths.rs
index 76d22b6..6d1f9b4 100644
--- a/src/transport/paths.rs
+++ b/src/transport/paths.rs
@@ -1251,6 +1251,7 @@ mod tests {
             epoch_seq: 0,
             ready: false,
             syncs_with_you: false,
+            leaving: false,
         })
     }
 
diff --git a/src/transport/sim.rs b/src/transport/sim.rs
index a558e9f..9267232 100644
--- a/src/transport/sim.rs
+++ b/src/transport/sim.rs
@@ -367,6 +367,7 @@ mod tests {
             epoch_seq: 0,
             ready: true,
             syncs_with_you: false,
+            leaving: false,
         })
     }
 
diff --git a/src/transport/wire.rs b/src/transport/wire.rs
index 817437a..dc94ab3 100644
--- a/src/transport/wire.rs
+++ b/src/transport/wire.rs
@@ -14,7 +14,7 @@ use super::{MessageKind, ParsedMessage};
 
 /// The wire format's version, the first byte of every payload. Bump it with every change
 /// to a message type or a codec (`golden_bytes` fails then).
-pub(crate) const VERSION: u8 = 9;
+pub(crate) const VERSION: u8 = 10;
 
 /// What the second byte of a payload says.
 fn tag(kind: MessageKind) -> u8 {
@@ -353,6 +353,7 @@ pub(crate) fn samples() -> Vec<ParsedMessage> {
         epoch_seq: 1,
         ready: true,
         syncs_with_you: true,
+        leaving: false,
     };
     let proof = IdentityProof {
         key: [8; 32],
@@ -515,6 +516,7 @@ mod tests {
             epoch_seq: 0,
             ready: true,
             syncs_with_you: false,
+            leaving: false,
         };
         let decoded =
             decode(&encode(&ParsedMessage::StatusReply(status.clone())).unwrap()).unwrap();
@@ -537,6 +539,7 @@ mod tests {
             epoch_seq: 0,
             ready: true,
             syncs_with_you: true,
+            leaving: false,
         };
         assert!(decode(&encode(&ParsedMessage::Status(status)).unwrap()).is_err());
     }
@@ -596,7 +599,7 @@ mod tests {
         });
         assert_eq!(
             hex(&encode(&hello).unwrap()),
-            format!("090097a5737761726d{epoch_hex}01c403010203c3{}07", run("04"))
+            format!("0a0097a5737761726d{epoch_hex}01c403010203c3{}07", run("04"))
         );
 
         let reply = ParsedMessage::HelloReply(HelloReply {
@@ -616,7 +619,7 @@ mod tests {
         assert_eq!(
             hex(&encode(&reply).unwrap()),
             format!(
-                "090199a5737761726d9192a17091a161{epoch_hex}01c400c2{}05{}",
+                "0a0199a5737761726d9192a17091a161{epoch_hex}01c400c2{}05{}",
                 run("04"),
                 run("06")
             )
@@ -628,10 +631,11 @@ mod tests {
             epoch_seq: 1,
             ready: true,
             syncs_with_you: false,
+            leaving: false,
         });
         assert_eq!(
             hex(&encode(&status).unwrap()),
-            format!("0903959290c400{epoch_hex}01c3c2")
+            format!("0a03969290c400{epoch_hex}01c3c2c2")
         );
 
         let who = ParsedMessage::Who(IdentityProof {
@@ -640,7 +644,7 @@ mod tests {
         });
         assert_eq!(
             hex(&encode(&who).unwrap()),
-            format!("090692c420{}c440{}", "08".repeat(32), "09".repeat(64))
+            format!("0a0692c420{}c440{}", "08".repeat(32), "09".repeat(64))
         );
         let ParsedMessage::Me(proof) = decode(
             &encode(&ParsedMessage::Me(IdentityProof {
@@ -654,10 +658,10 @@ mod tests {
         };
         assert_eq!((proof.key, proof.signature), ([8; 32], [9; 64]));
 
-        assert_eq!(hex(&encode(&ParsedMessage::Ping(7)).unwrap()), "090807");
+        assert_eq!(hex(&encode(&ParsedMessage::Ping(7)).unwrap()), "0a0807");
         assert_eq!(
             hex(&encode(&ParsedMessage::Pong(u64::MAX)).unwrap()),
-            "0909cfffffffffffffffff"
+            "0a09cfffffffffffffffff"
         );
         let ParsedMessage::Pong(nonce) = decode(&[VERSION, 9, 0x2a]).unwrap() else {
             panic!("not a PONG");