handshakes counted

5 files changed, 74 insertions, 2 deletions

diff --git a/docs/measurements.md b/docs/measurements.md
index f57db9b..f49f4ff 100644
--- a/docs/measurements.md
+++ b/docs/measurements.md
@@ -266,6 +266,7 @@ forgotten peer's are gone).
 | `incidents.askedAgain` | `peer_asked_again_total` | Times nothing answered what we sent (a keepalive, or a sync message that awaits an answer) for half the time it may before the peer counts as unreached (`unreached_after`, twice the first sync reply timeout), so a PING asked again ([`peers.md`](peers.md)). Each is a PING and its PONG on top of the keepalives. |
 | `incidents.dormancies` | `peer_dormancies_total` | Times the peer went dormant (nothing received for `dormant_after`). |
 | `incidents.dormantMs` | `peer_dormant_seconds_total` | Time it spent dormant, up to its last return. |
+| `incidents.handshakes` | `peer_handshakes_total` | Handshakes the peer started with us (its HELLO, not a retry of one), each resetting our session with it: one per start of the peer, and one when it comes back from dormancy or heals. More than four within a minute are logged as a warning (once a minute): a restart loop, or handshakes that keep failing on its side. |
 
 ## Per peer
 
diff --git a/docs/openapi.yaml b/docs/openapi.yaml
index 5ca41b0..da89b6c 100644
--- a/docs/openapi.yaml
+++ b/docs/openapi.yaml
@@ -1747,7 +1747,7 @@ components:
 
     Incidents:
       type: object
-      required: [replyTimeouts, heals, askedAgain, dormancies, dormantMs]
+      required: [replyTimeouts, heals, askedAgain, dormancies, dormantMs, handshakes]
       properties:
         replyTimeouts:
           type: integer
@@ -1763,6 +1763,9 @@ components:
         dormantMs:
           type: integer
           description: Time spent dormant up to the last return (the current dormancy isn't included).
+        handshakes:
+          type: integer
+          description: Handshakes the peer started with us (its HELLO, not a retry), each resetting our session with it.
 
     ClockOffset:
       type: object
diff --git a/src/api/stats/prometheus.rs b/src/api/stats/prometheus.rs
index 8b8ca1c..94525cb 100644
--- a/src/api/stats/prometheus.rs
+++ b/src/api/stats/prometheus.rs
@@ -1080,7 +1080,7 @@ pub(super) fn render(stats: &StatsResponse, list: &PeerListResponse) -> String {
         }
     }
 
-    let incidents: [IncidentSeries; 5] = [
+    let incidents: [IncidentSeries; 6] = [
         (
             "peer_reply_timeouts_total",
             "Sync messages unanswered until the reply timeout.",
@@ -1092,6 +1092,11 @@ pub(super) fn render(stats: &StatsResponse, list: &PeerListResponse) -> String {
             |i| i.heals as f64,
         ),
         (
+            "peer_handshakes_total",
+            "Handshakes the peer started with us, each resetting our session with it.",
+            |i| i.handshakes as f64,
+        ),
+        (
             "peer_asked_again_total",
             "Times nothing answered for a while, so a PING asked again.",
             |i| i.asked_again as f64,
diff --git a/src/state/peer.rs b/src/state/peer.rs
index 2f027ba..7000386 100644
--- a/src/state/peer.rs
+++ b/src/state/peer.rs
@@ -188,6 +188,10 @@ pub(crate) struct PeerState {
 
     /// What went wrong with this peer so far (counted only, for /peers and /admin/stats).
     pub(crate) incidents: Incidents,
+    /// The handshakes it started within the current `HANDSHAKES_WINDOW`, from when it
+    /// began; and when we last warned of them.
+    handshakes_lately: Option<(Instant, u32)>,
+    handshakes_warned_at: Option<Instant>,
     /// Changes it asked for that we had counted as its, as another parent in their
     /// writer's tree brings them (`SyncSession::resend_if_needed`): per asker, for the
     /// bench (`simtest::bench`).
@@ -254,6 +258,9 @@ pub(crate) struct Incidents {
     /// Time spent dormant, in milliseconds, up to its last return (the current
     /// dormancy, if any, isn't included).
     pub(crate) dormant_ms: u64,
+    /// Handshakes the peer started with us (its HELLO, not a retry of one): each resets
+    /// our session with it (`PeerState::handshake_started`).
+    pub(crate) handshakes: u64,
 }
 
 impl Incidents {
@@ -263,6 +270,7 @@ impl Incidents {
         self.asked_again += other.asked_again;
         self.dormancies += other.dormancies;
         self.dormant_ms += other.dormant_ms;
+        self.handshakes += other.handshakes;
     }
 }
 
@@ -278,6 +286,11 @@ pub(crate) enum Keepalive {
 /// Unanswered HELLOs after which a peer counts as unreachable (1 + 2 + 4 s apart).
 const UNREACHABLE_AFTER_HELLOS: u32 = 3;
 
+/// Handshakes a peer may start with us within `HANDSHAKES_WINDOW` before we warn
+/// (`PeerState::handshake_started`): a restart now and then is fine, a loop isn't.
+const HANDSHAKES_WARN: u32 = 4;
+const HANDSHAKES_WINDOW: Duration = Duration::from_secs(60);
+
 impl PeerState {
     pub(crate) fn new(leads: bool, timing: &Timing) -> Self {
         Self {
@@ -319,11 +332,36 @@ impl PeerState {
             last_hello_at: None,
             unreachable_since: None,
             incidents: Incidents::default(),
+            handshakes_lately: None,
+            handshakes_warned_at: None,
             asked_known: 0,
             whole_failed: None,
         }
     }
 
+    /// The peer started a handshake with us (`swarm::peer::handshake::establish_peer`):
+    /// counted (`Incidents::handshakes`). Returns how many it started within the
+    /// current window when more than `HANDSHAKES_WARN`, at most once a window: a peer
+    /// in a restart loop, or one whose handshakes keep failing on its side.
+    pub(crate) fn handshake_started(&mut self, now: Instant) -> Option<u32> {
+        self.incidents.handshakes += 1;
+        let (since, count) = match self.handshakes_lately {
+            Some((since, count)) if now.saturating_duration_since(since) < HANDSHAKES_WINDOW => {
+                (since, count + 1)
+            }
+            _ => (now, 1),
+        };
+        self.handshakes_lately = Some((since, count));
+        let warned_lately = self
+            .handshakes_warned_at
+            .is_some_and(|at| now.saturating_duration_since(at) < HANDSHAKES_WINDOW);
+        if count > HANDSHAKES_WARN && !warned_lately {
+            self.handshakes_warned_at = Some(now);
+            return Some(count);
+        }
+        None
+    }
+
     /// Start Automerge sync with this peer from scratch. A message still being handed to
     /// the transport was generated against the old state; its outcome is then ignored.
     pub(crate) fn reset_sync(&mut self) {
@@ -742,6 +780,24 @@ impl PeerState {
 mod tests {
     use super::*;
 
+    /// Handshakes are counted; more than `HANDSHAKES_WARN` within a window are warned
+    /// of, once a window.
+    #[test]
+    fn handshakes_are_counted_and_a_loop_warned_of() {
+        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;
+        let at = |secs: u64| start + Duration::from_secs(secs);
+        for i in 0..HANDSHAKES_WARN {
+            assert_eq!(peer.handshake_started(at(u64::from(i))), None);
+        }
+        assert_eq!(peer.handshake_started(at(10)), Some(HANDSHAKES_WARN + 1));
+        assert_eq!(peer.handshake_started(at(11)), None, "once a window");
+        // A new window: counted from one again.
+        assert_eq!(peer.handshake_started(at(70)), None);
+        assert_eq!(peer.incidents.handshakes, u64::from(HANDSHAKES_WARN) + 3);
+    }
+
     /// Unknown until heard from; silent long enough, dormant like an active peer.
     #[test]
     fn unknown_until_heard_from() {
diff --git a/src/swarm/peer/handshake.rs b/src/swarm/peer/handshake.rs
index 92216fd..b964064 100644
--- a/src/swarm/peer/handshake.rs
+++ b/src/swarm/peer/handshake.rs
@@ -440,6 +440,13 @@ pub(super) async fn establish_peer(
     // resets: it starts over with us. Its resent HELLOs and duplicate replies (caused by
     // our own retries) don't.
     peer_state.reports_ready = Some(ready);
+    if from_hello && let Some(count) = peer_state.handshake_started(tokio::time::Instant::now()) {
+        warn!(
+            peer = %peer,
+            handshakes = count,
+            "Peer started many handshakes within a minute (a restart loop?); each resets our session with it"
+        );
+    }
     if from_hello || peer_state.hello_state != HelloState::Established {
         peer_state.hello_state = HelloState::Established;
         peer_state.reset_sync();