Let flap dampening engage again after an episode lapses

Carries the maintenance line's fix onto this branch's flap dampener,
which was extracted into its own component with an injected monotonic
clock and so could not take the change as written.

The arming check tested whether a dampening deadline had ever been set
rather than whether one was still in effect, so the first episode
disarmed the mechanism permanently. A node in a second flap storm went
on switching parents under hold-down alone, and neither the
flap_dampened counter nor the warning fired again, so the storm was
invisible to anyone watching that counter.

Retire a lapsed episode explicitly, clearing both the deadline and the
switch counter, so a second episode requires a fresh threshold of
switches within one window rather than re-engaging on the first switch
after lapse. Hold-down was unaffected throughout and continued to limit
discretionary switching, which is why the practical effect at shipped
settings was lost visibility and a lost escalation tier rather than
unrestrained flapping.

Cap node.tree.flap_dampening_secs at one year. The seconds-to-
milliseconds conversion saturates at u64::MAX and the deadline
arithmetic then overflowed when dampening engaged; a year is
indistinguishable from permanent for this mechanism.

Report every engagement. Parent-loss recovery ran inside the tree state
where no metrics handle is reachable, so it returns the fact of
engagement to the caller that has one, through a crate-internal recovery
entry point that leaves handle_parent_lost's published bool return
alone. The warning now names which path armed the episode and how long
discretionary switching stays suppressed, using the same trigger values
as the parent-switch logs beside it.

Five tests, each verified by constructing the defect rather than by
reading the fix: restoring the deadline-only gate reds the re-arm test,
retiring the deadline without the counter reds it at a different
assertion, removing the clamp panics the overflow test, and making
recovery report no engagement reds the recovery test. The six existing
flap tests stay green under all four, which is why none of them was
holding this.

The ancestry-update path takes the reporting call for completeness but
cannot reach it: that arm is selected only when the declared parent is
unchanged, so no episode can be armed there. Same on the maintenance
line.

src/node/handlers/mmp.rs

@@ -246,8 +246,7 @@ impl Node {
                     "Parent switched after first RTT measurement"
                 );
                 if flap_dampened {
-                    self.metrics().tree.flap_dampened.inc();
-                    warn!("Flap dampening engaged: excessive parent switches detected");
+                    self.note_flap("first-rtt");
                 }
                 self.send_tree_announce_to_all().await;
                 let all_peers: Vec<crate::NodeAddr> = self.peers.keys().copied().collect();

src/node/tree.rs

@@ -15,6 +15,20 @@ use super::reject::TreeReject;
 use super::{Node, NodeError};
 use tracing::{debug, info, trace, warn};
 
+impl Node {
+    /// Report a flap-dampening engagement: one counter tick and one warning
+    /// naming which path armed the episode and how long discretionary parent
+    /// switching stays suppressed.
+    pub(super) fn note_flap(&self, trigger: &str) {
+        self.metrics().tree.flap_dampened.inc();
+        warn!(
+            trigger = trigger,
+            dampening_secs = self.tree_state.dampening_secs(),
+            "Flap dampening engaged, discretionary parent switching suppressed"
+        );
+    }
+}
+
 /// Sign a node's own tree declaration, writing the 64-byte signature back into
 /// it. The key-crypto boundary (§6): `proto::stp` owns the declaration data and
 /// the pure `signing_bytes()` serialization; the shell owns the `secp256k1`
@@ -383,8 +397,7 @@ impl Node {
                     "Parent switched, invalidated downstream coord cache entries, announcing to all peers"
                 );
                 if flap_dampened {
-                    self.metrics().tree.flap_dampened.inc();
-                    warn!("Flap dampening engaged: excessive parent switches detected");
+                    self.note_flap("announce");
                 }
 
                 self.send_tree_announce_to_all().await;
@@ -436,10 +449,8 @@ impl Node {
                     .duration_since(std::time::UNIX_EPOCH)
                     .map(|d| d.as_secs())
                     .unwrap_or(0);
-                if self
-                    .tree_state
-                    .handle_parent_lost(&peer_costs, timestamp, mono_now_ms)
-                {
+                let outcome = self.tree_state.recover(&peer_costs, timestamp, mono_now_ms);
+                if outcome.changed {
                     // Clone identity up front to avoid a split borrow against the
                     // &mut self.tree_state / &mut self.coord_cache calls below (cold path).
                     let our_identity = self.identity().clone();
@@ -460,6 +471,9 @@ impl Node {
                         .invalidate_other_roots(self.tree_state.root());
                     self.reset_lookup_backoff();
                     self.send_tree_announce_to_all().await;
+                    if outcome.dampened {
+                        self.note_flap("loop-detected");
+                    }
                 }
             }
             TreeDecision::AncestryUpdate { parent, new_seq } => {
@@ -487,8 +501,17 @@ impl Node {
                 // Clone identity up front to avoid a split borrow against the
                 // &mut self.tree_state / &mut self.coord_cache calls below (cold path).
                 let our_identity = self.identity().clone();
-                self.tree_state
-                    .set_parent(parent, new_seq, timestamp, mono_now_ms);
+                let flap_dampened =
+                    self.tree_state
+                        .set_parent(parent, new_seq, timestamp, mono_now_ms);
+                // Defensive rather than reachable: this arm is selected only
+                // when the declared parent is unchanged, so `set_parent`'s
+                // `parent_changed` is false and no episode can be armed here.
+                // Kept so the reporting stays complete if the classify core's
+                // guard ever admits a different parent on this path.
+                if flap_dampened {
+                    self.note_flap("ancestry-update");
+                }
                 self.tree_state.recompute_coords();
                 if let Err(e) =
                     sign_declaration(self.tree_state.my_declaration_mut(), &our_identity)
@@ -636,8 +659,7 @@ impl Node {
                     "Parent switched via periodic cost re-evaluation"
                 );
                 if flap_dampened {
-                    self.metrics().tree.flap_dampened.inc();
-                    warn!("Flap dampening engaged: excessive parent switches detected");
+                    self.note_flap("periodic");
                 }
 
                 self.send_tree_announce_to_all().await;
@@ -748,10 +770,8 @@ impl Node {
         // Removal is not a pure classify: `handle_parent_lost` is a &mut mutator
         // whose returned `changed` bool IS the decision. Drive it and map the
         // outcome onto the TreeDecision vocabulary.
-        let decision = if self
-            .tree_state
-            .handle_parent_lost(&peer_costs, now_secs, mono_now_ms)
-        {
+        let outcome = self.tree_state.recover(&peer_costs, now_secs, mono_now_ms);
+        let decision = if outcome.changed {
             TreeDecision::ParentLost
         } else {
             TreeDecision::NoChange
@@ -785,6 +805,9 @@ impl Node {
                     is_root = self.tree_state.is_root(),
                     "Tree state updated after parent loss"
                 );
+                if outcome.dampened {
+                    self.note_flap("parent-loss");
+                }
                 true
             }
             TreeDecision::NoChange => false,

src/proto/stp/limits.rs

@@ -16,6 +16,12 @@
 //! setters accept seconds (matching the node config) and store the value scaled
 //! to milliseconds so the comparisons stay in one unit.
 
+/// Longest dampening episode representable on the monotonic clock, in
+/// milliseconds. A year is indistinguishable from permanent for this
+/// mechanism; the bound is what keeps a hostile `flap_dampening_secs` from
+/// overflowing the stamp.
+const MAX_FLAP_DAMPENING_MS: u64 = 365 * 24 * 60 * 60 * 1000;
+
 /// Flap-dampening / hold-down state for a node's parent selection.
 ///
 /// Groups the flap-detection timers behind a single struct so the tree
@@ -70,7 +76,17 @@ impl FlapDampener {
     ) {
         self.flap_threshold = threshold;
         self.flap_window = window_secs.saturating_mul(1000);
-        self.flap_dampening_duration = dampening_secs.saturating_mul(1000);
+        self.flap_dampening_duration = dampening_secs
+            .saturating_mul(1000)
+            .min(MAX_FLAP_DAMPENING_MS);
+    }
+
+    /// How long a dampening episode suppresses discretionary parent
+    /// switching, in seconds, after the configured value is clamped.
+    ///
+    /// Feeds the log field on the engagement warning.
+    pub(crate) fn dampening_secs(&self) -> u64 {
+        self.flap_dampening_duration / 1000
     }
 
     /// Stamp the time of a parent switch (called on every `set_parent`,
@@ -84,6 +100,17 @@ impl FlapDampener {
     /// Returns true if dampening was just engaged. `now_ms` is the injected
     /// monotonic time in milliseconds.
     pub(crate) fn record_parent_switch(&mut self, now_ms: u64) -> bool {
+        // Retire a lapsed episode here rather than lazily. Clearing the
+        // deadline and the counter together is what makes each episode cost a
+        // fresh threshold of switches inside one window: switches taken during
+        // an episode (mandatory ones bypass the veto) would otherwise carry
+        // into the next window and re-engage on a single switch after lapse.
+        if self.flap_dampening_until.is_some() && !self.is_flap_dampened(now_ms) {
+            self.flap_dampening_until = None;
+            self.flap_count = 0;
+            self.flap_window_start = None;
+        }
+
         // Reset window if expired or not started
         match self.flap_window_start {
             Some(start) if now_ms.saturating_sub(start) < self.flap_window => {
@@ -95,9 +122,11 @@ impl FlapDampener {
             }
         }
 
-        // Check threshold
-        if self.flap_count >= self.flap_threshold && self.flap_dampening_until.is_none() {
-            self.flap_dampening_until = Some(now_ms + self.flap_dampening_duration);
+        // Check threshold. The dampening test is redundant with the retirement
+        // above in this control flow; it is kept so the gate reads correctly on
+        // its own and survives an edit that moves the retirement.
+        if self.flap_count >= self.flap_threshold && !self.is_flap_dampened(now_ms) {
+            self.flap_dampening_until = Some(now_ms.saturating_add(self.flap_dampening_duration));
             return true;
         }
         false

src/proto/stp/state.rs

@@ -8,6 +8,20 @@ use super::limits::FlapDampener;
 use super::{CoordEntry, ParentDeclaration, TreeCoordinate};
 use crate::NodeAddr;
 
+/// What a parent-loss recovery did: whether the tree state changed, and
+/// whether the recovery switch was the one that armed a dampening episode.
+///
+/// Crate-internal on purpose. The published entry point is
+/// [`TreeState::handle_parent_lost`], whose `bool` return this type must not
+/// displace.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) struct ParentLoss {
+    /// Whether the tree state changed and the caller should re-announce.
+    pub(crate) changed: bool,
+    /// Whether the recovery switch armed a flap dampening episode.
+    pub(crate) dampened: bool,
+}
+
 /// Local spanning tree state for a node.
 ///
 /// Contains this node's declaration, coordinates, and view of peers'
@@ -311,6 +325,14 @@ impl TreeState {
             .set_flap_dampening(threshold, window_secs, dampening_secs);
     }
 
+    /// How long a dampening episode suppresses discretionary parent switching,
+    /// after the configured value is clamped.
+    ///
+    /// Crate-internal: it feeds a log field on the engagement warning.
+    pub(crate) fn dampening_secs(&self) -> u64 {
+        self.flap.dampening_secs()
+    }
+
     /// Check if flap dampening is currently active. `now_ms` is the injected
     /// monotonic time in milliseconds.
     pub fn is_flap_dampened(&self, now_ms: u64) -> bool {
@@ -499,6 +521,22 @@ impl TreeState {
         now_secs: u64,
         now_ms: u64,
     ) -> bool {
+        self.recover(peer_costs, now_secs, now_ms).changed
+    }
+
+    /// Handle loss of current parent, reporting whether the recovery switch
+    /// itself armed a flap dampening episode.
+    ///
+    /// Same recovery as [`TreeState::handle_parent_lost`], which delegates
+    /// here. A caller holding a metrics handle uses this one so the
+    /// engagement can be counted and logged; the published signature stays
+    /// `bool`.
+    pub(crate) fn recover(
+        &mut self,
+        peer_costs: &BTreeMap<NodeAddr, f64>,
+        now_secs: u64,
+        now_ms: u64,
+    ) -> ParentLoss {
         // Try to find an alternative parent. The veto is computed at the edge and
         // applied only to a discretionary result; a mandatory switch bypasses it.
         let suppressed = self.is_switch_suppressed(now_ms);
@@ -509,16 +547,23 @@ impl TreeState {
         };
         if let Some(new_parent) = alt {
             let new_seq = self.my_declaration.sequence() + 1;
-            self.set_parent(new_parent, new_seq, now_secs, now_ms);
+            let dampened = self.set_parent(new_parent, new_seq, now_secs, now_ms);
             self.recompute_coords();
-            return true;
+            return ParentLoss {
+                changed: true,
+                dampened,
+            };
         }
 
-        // No alternative: become own root
+        // No alternative: become own root. This branch never calls
+        // `set_parent`, so it cannot arm a dampening episode.
         let new_seq = self.my_declaration.sequence() + 1;
         self.my_declaration = ParentDeclaration::self_root(self.my_node_addr, new_seq, now_secs);
         self.recompute_coords();
-        true
+        ParentLoss {
+            changed: true,
+            dampened: false,
+        }
     }
 
     /// Mutable access to this node's declaration.

src/proto/stp/tests/limits.rs

@@ -3,6 +3,7 @@
 use alloc::collections::{BTreeMap, BTreeSet};
 
 use super::util::{make_coords, make_costs, make_node_addr};
+use crate::NodeAddr;
 use crate::proto::stp::{ParentDeclaration, ParentEval, TreeState};
 
 #[test]
@@ -251,3 +252,193 @@ fn test_flap_dampening_same_parent_no_count() {
     // Should NOT be dampened since only the first was a real switch
     assert!(!state.is_flap_dampened(3000));
 }
+
+#[test]
+fn test_flap_dampening_engages_a_second_time_after_first_episode_lapses() {
+    // A lapsed episode must re-arm the mechanism: a second flap storm has to
+    // engage dampening again, and must cost a fresh threshold of switches
+    // rather than re-engaging on the first switch after the lapse. The
+    // injected clock is advanced past the deadline instead of using a
+    // zero-length episode, so the retirement is driven by a genuinely expired
+    // episode.
+    let my_node = make_node_addr(5);
+    let mut state = TreeState::new(my_node, 1000);
+    state.set_flap_dampening(3, 60, 10);
+    state.set_hold_down(0);
+
+    let peer_a = make_node_addr(1);
+    let peer_b = make_node_addr(2);
+    let root = make_node_addr(0);
+
+    state.update_peer(
+        ParentDeclaration::new(peer_a, root, 1, 1000),
+        make_coords(&[1, 0]),
+    );
+    state.update_peer(
+        ParentDeclaration::new(peer_b, root, 1, 1000),
+        make_coords(&[2, 0]),
+    );
+
+    // First episode: three switches inside the window reach the threshold.
+    let first = state.set_parent(peer_a, 1, 1000, 1000);
+    state.recompute_coords();
+    let second = state.set_parent(peer_b, 2, 2000, 2000);
+    state.recompute_coords();
+    let third = state.set_parent(peer_a, 3, 3000, 3000);
+    state.recompute_coords();
+
+    assert!(!first);
+    assert!(!second);
+    assert!(third, "first episode must engage at threshold");
+    assert!(state.is_flap_dampened(3000));
+
+    // The episode was stamped at 3000 for 10s, so it has lapsed by 14000.
+    assert!(!state.is_flap_dampened(14_000));
+
+    // Second episode: a fresh threshold of switches is required, so the first
+    // two switches after the lapse must not re-engage.
+    let fourth = state.set_parent(peer_b, 4, 4000, 14_000);
+    state.recompute_coords();
+    let fifth = state.set_parent(peer_a, 5, 5000, 15_000);
+    state.recompute_coords();
+    let sixth = state.set_parent(peer_b, 6, 6000, 16_000);
+    state.recompute_coords();
+
+    assert!(!fourth, "a lapsed episode must not re-engage on one switch");
+    assert!(
+        !fifth,
+        "a lapsed episode must not re-engage below threshold"
+    );
+    assert!(sixth, "a second episode must engage after the first lapses");
+    assert!(state.is_flap_dampened(16_000));
+}
+
+#[test]
+fn test_flap_dampening_duration_at_u64_max_does_not_panic() {
+    // A hostile node.tree.flap_dampening_secs must be clamped rather than
+    // overflow the monotonic stamp when an episode engages. Unclamped, the
+    // seconds-to-milliseconds conversion saturates at u64::MAX and the
+    // deadline arithmetic then overflows on the switch that engages.
+    let my_node = make_node_addr(5);
+    let mut state = TreeState::new(my_node, 1000);
+    state.set_flap_dampening(3, 60, u64::MAX);
+    state.set_hold_down(0);
+
+    let peer_a = make_node_addr(1);
+    let peer_b = make_node_addr(2);
+    let root = make_node_addr(0);
+
+    state.update_peer(
+        ParentDeclaration::new(peer_a, root, 1, 1000),
+        make_coords(&[1, 0]),
+    );
+    state.update_peer(
+        ParentDeclaration::new(peer_b, root, 1, 1000),
+        make_coords(&[2, 0]),
+    );
+
+    state.set_parent(peer_a, 1, 1000, 1000);
+    state.recompute_coords();
+    state.set_parent(peer_b, 2, 2000, 2000);
+    state.recompute_coords();
+    let dampened = state.set_parent(peer_a, 3, 3000, 3000);
+    state.recompute_coords();
+
+    assert!(dampened, "the threshold switch must still engage dampening");
+    assert!(state.is_flap_dampened(3000));
+    // Clamped to a year, which is what the episode reports to its log field.
+    assert_eq!(state.dampening_secs(), 365 * 24 * 60 * 60);
+}
+
+#[test]
+fn test_parent_loss_recovery_reports_the_engagement_that_arms_dampening() {
+    // A parent-loss storm engages dampening on a mandatory recovery switch,
+    // which bypasses the veto but still feeds the flap counter. The recovery
+    // must report that engagement so the shell, which is the side holding a
+    // metrics handle, can surface it.
+    let my_node = make_node_addr(5);
+    let mut state = TreeState::new(my_node, 1000);
+    state.set_flap_dampening(2, 60, 120);
+    state.set_hold_down(0);
+
+    let peer_a = make_node_addr(1);
+    let peer_b = make_node_addr(2);
+    let root = make_node_addr(0);
+
+    state.update_peer(
+        ParentDeclaration::new(peer_a, root, 1, 1000),
+        make_coords(&[1, 0]),
+    );
+    state.update_peer(
+        ParentDeclaration::new(peer_b, root, 1, 1000),
+        make_coords(&[2, 0]),
+    );
+
+    // One switch short of the threshold.
+    let first = state.set_parent(peer_a, 1, 1000, 1000);
+    state.recompute_coords();
+    assert!(!first, "one switch is below the threshold");
+
+    // Parent disappears; recovery picks peer_b and crosses the threshold.
+    state.remove_peer(&peer_a);
+    let outcome = state.recover(&BTreeMap::new(), 2000, 2000);
+
+    assert!(outcome.changed);
+    assert_eq!(state.my_declaration().parent_id(), &peer_b);
+    assert!(
+        outcome.dampened,
+        "parent-loss recovery must report the engagement it armed"
+    );
+    assert!(state.is_flap_dampened(2000));
+}
+
+#[test]
+fn test_parent_loss_recovery_to_self_root_reports_no_engagement() {
+    // The self-root fallthrough takes no parent switch, so it can never arm
+    // an episode and must never report one.
+    let my_node = make_node_addr(5);
+    let mut state = TreeState::new(my_node, 1000);
+    state.set_flap_dampening(1, 60, 120);
+    state.set_hold_down(0);
+
+    let peer_a = make_node_addr(1);
+    let root = make_node_addr(0);
+
+    state.update_peer(
+        ParentDeclaration::new(peer_a, root, 1, 1000),
+        make_coords(&[1, 0]),
+    );
+    state.set_parent(peer_a, 1, 1000, 1000);
+    state.recompute_coords();
+
+    state.remove_peer(&peer_a);
+    let outcome = state.recover(&BTreeMap::new(), 2000, 2000);
+
+    assert!(outcome.changed);
+    assert!(state.is_root());
+    assert!(!outcome.dampened, "self-root recovery arms no episode");
+}
+
+#[test]
+fn test_handle_parent_lost_keeps_its_published_bool_return() {
+    // `TreeState` is published, so the return type of this entry point is part
+    // of the API and the richer `ParentLoss` must not displace it. Binding the
+    // method to an explicitly typed function pointer is the assertion: any
+    // other return type fails to compile. Calling through the pointer keeps
+    // the binding live.
+    let published: fn(&mut TreeState, &BTreeMap<NodeAddr, f64>, u64, u64) -> bool =
+        TreeState::handle_parent_lost;
+
+    let mut state = TreeState::new(make_node_addr(5), 1000);
+    let peer = make_node_addr(1);
+    state.update_peer(
+        ParentDeclaration::new(peer, make_node_addr(0), 1, 1000),
+        make_coords(&[1, 0]),
+    );
+    state.set_parent(peer, 1, 1000, 1000);
+    state.recompute_coords();
+
+    state.remove_peer(&peer);
+    assert!(published(&mut state, &BTreeMap::new(), 2000, 2000));
+    assert!(state.is_root());
+}