Skip to main content

max / makenotwork

pom: debounce status transitions (N-of-M) and seed state on restart The probe loops transitioned — paging + opening an incident — on the FIRST differing check, so one transient timeout flipped operational->unreachable, emailed an alert, and opened an incident that resolved next tick (alert fatigue + incident noise). And their in-memory transition state reset to None on restart, so a persistently-degraded target re-fired its alert on every restart (fuzz-2026-07-06). Add a seedable TransitionGate (new serve.confirmations, default 2): a change is reported only after the new status holds for N consecutive checks, and the gate is seeded from the last stored status so a restart doesn't re-fire. Health and ssh-banner loops route transitions through it; scan_pipeline and meta_alert seed their flags from the alert ledger. Raw snapshots are still stored every tick — only the alert/incident decision is debounced. Pure gate logic is unit-tested (blip absorbed, flap resets the run, restart no-refire). Also sets Delay on the health/scan_pipeline/meta_alert/peer loops (tick-storm). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-07 17:10 UTC
Commit: 025bd839720e52a216af6e058bb4c7491d5f43bf
Parent: 1a0b49e
8 files changed, +211 insertions, -36 deletions
M pom/src/cli/mod.rs +1
@@ -4,6 +4,7 @@ mod incident;
4 4 mod serve;
5 5 mod status;
6 6 mod tasks;
7 + mod transition;
7 8
8 9 pub(crate) use serve::cmd_serve;
9 10 pub(crate) use status::cmd_status;
M pom/src/cli/serve.rs +1 -1
@@ -63,7 +63,7 @@ pub(crate) async fn cmd_serve(
63 63 handles.extend(tasks::spawn_whois_tasks(config, pool, &token, &alerter));
64 64 handles.extend(tasks::spawn_cors_tasks(config, pool, &token, &alerter));
65 65 handles.extend(tasks::spawn_backup_tasks(config, pool, &token, &alerter));
66 - handles.extend(tasks::spawn_scan_pipeline_tasks(config, &token, &alerter));
66 + handles.extend(tasks::spawn_scan_pipeline_tasks(config, pool, &token, &alerter));
67 67 handles.push(tasks::spawn_prune_task(pool, prune_days, &token));
68 68
69 69 // Spawn peer heartbeat tasks
M pom/src/cli/tasks/health.rs +44 -31
@@ -8,6 +8,7 @@ use pom::db;
8 8 use pom::types::{HealthStatus, LatencyStats};
9 9
10 10 use super::super::incident::{incident_action, IncidentAction};
11 + use super::super::transition::TransitionGate;
11 12
12 13 pub(crate) fn spawn_health_tasks(
13 14 config: &Config,
@@ -16,6 +17,7 @@ pub(crate) fn spawn_health_tasks(
16 17 alerter: &Option<Alerter>,
17 18 ) -> Vec<JoinHandle<()>> {
18 19 let default_interval = config.serve.interval_secs;
20 + let confirmations = config.serve.confirmations;
19 21 let mut handles = Vec::new();
20 22
21 23 for name in config.target_names() {
@@ -35,47 +37,53 @@ pub(crate) fn spawn_health_tasks(
35 37 let mut interval = tokio::time::interval(
36 38 std::time::Duration::from_secs(interval_secs),
37 39 );
40 + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
38 41 let expect = health_config.expect.as_ref();
39 42 let mut in_drift = false;
43 + // Seed the N-of-M transition gate from the last stored status so a
44 + // restart of a persistently-degraded target doesn't re-fire, and a
45 + // transient blip is debounced before it pages/opens an incident.
46 + let seed = db::get_latest_health(&pool, &name)
47 + .await
48 + .ok()
49 + .flatten()
50 + .map(|s| s.status);
51 + let mut gate = TransitionGate::seeded(seed, confirmations);
40 52 interval.tick().await; // consume immediate first tick
41 53 loop {
42 54 tokio::select! {
43 55 _ = cancel.cancelled() => break,
44 56 _ = interval.tick() => {}
45 57 }
46 - let previous = db::get_latest_health(&pool, &name).await.ok().flatten();
47 58 let snapshot = http::check_health(&name, &health_config, expect).await;
48 59 info!("{}: {} ({}ms)", name, snapshot.status, snapshot.response_time_ms);
49 60 if let Err(e) = db::insert_health_check(&pool, &snapshot).await {
50 61 tracing::error!("{name}: failed to store health check: {e}");
51 62 }
52 63
53 - // Fire alerts on status transitions
54 - if let Some(ref alerter) = alerter
55 - && let Some(ref prev) = previous
56 - && prev.status != snapshot.status
57 - {
58 - let from = prev.status.to_string();
59 - let to = snapshot.status.to_string();
60 - if snapshot.status == HealthStatus::Operational {
61 - alerter.send_health_recovery(&name, &label, &from).await;
62 - } else {
63 - alerter.send_health_alert(
64 - &name,
65 - &label,
66 - &from,
67 - &to,
68 - snapshot.error.as_deref(),
69 - ).await;
64 + // Alert AND open/close incidents only on a *confirmed*
65 + // transition (debounced + restart-seeded above).
66 + if let Some((from_status, to_status)) = gate.observe(snapshot.status) {
67 + if let Some(ref alerter) = alerter {
68 + let from = from_status.to_string();
69 + let to = to_status.to_string();
70 + if to_status == HealthStatus::Operational {
71 + alerter.send_health_recovery(&name, &label, &from).await;
72 + } else {
73 + alerter.send_health_alert(
74 + &name,
75 + &label,
76 + &from,
77 + &to,
78 + snapshot.error.as_deref(),
79 + ).await;
80 + }
70 81 }
71 - }
72 82
73 - // Track incidents on status transitions
74 - if let Some(ref prev) = previous {
75 - match incident_action(prev.status, snapshot.status) {
83 + match incident_action(from_status, to_status) {
76 84 IncidentAction::None => {}
77 85 IncidentAction::Open => {
78 - if let Err(e) = db::insert_incident(&pool, &name, &prev.status.to_string(), &snapshot.status.to_string()).await {
86 + if let Err(e) = db::insert_incident(&pool, &name, &from_status.to_string(), &to_status.to_string()).await {
79 87 tracing::error!("{name}: failed to open incident: {e}");
80 88 }
81 89 }
@@ -88,7 +96,7 @@ pub(crate) fn spawn_health_tasks(
88 96 if let Err(e) = db::close_open_incidents(&pool, &name).await {
89 97 tracing::error!("{name}: failed to close incidents: {e}");
90 98 }
91 - if let Err(e) = db::insert_incident(&pool, &name, &prev.status.to_string(), &snapshot.status.to_string()).await {
99 + if let Err(e) = db::insert_incident(&pool, &name, &from_status.to_string(), &to_status.to_string()).await {
92 100 tracing::error!("{name}: failed to open incident: {e}");
93 101 }
94 102 }
@@ -154,26 +162,31 @@ pub(crate) fn spawn_health_tasks(
154 162 let mut interval = tokio::time::interval(
155 163 std::time::Duration::from_secs(interval_secs),
156 164 );
165 + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
166 + let seed = db::get_latest_health(&pool, &check_name)
167 + .await
168 + .ok()
169 + .flatten()
170 + .map(|s| s.status);
171 + let mut gate = TransitionGate::seeded(seed, confirmations);
157 172 interval.tick().await;
158 173 loop {
159 174 tokio::select! {
160 175 _ = cancel.cancelled() => break,
161 176 _ = interval.tick() => {}
162 177 }
163 - let previous = db::get_latest_health(&pool, &check_name).await.ok().flatten();
164 178 let snapshot = ssh_banner::check_ssh_banner(&check_name, &ssh_config).await;
165 179 info!("{}: {} ({}ms)", check_name, snapshot.status, snapshot.response_time_ms);
166 180 if let Err(e) = db::insert_health_check(&pool, &snapshot).await {
167 181 tracing::error!("{check_name}: failed to store SSH banner check: {e}");
168 182 }
169 183
170 - if let Some(ref alerter) = alerter
171 - && let Some(ref prev) = previous
172 - && prev.status != snapshot.status
184 + if let Some((from_status, to_status)) = gate.observe(snapshot.status)
185 + && let Some(ref alerter) = alerter
173 186 {
174 - let from = prev.status.to_string();
175 - let to = snapshot.status.to_string();
176 - if snapshot.status == HealthStatus::Operational {
187 + let from = from_status.to_string();
188 + let to = to_status.to_string();
189 + if to_status == HealthStatus::Operational {
177 190 alerter.send_health_recovery(&check_name, &label, &from).await;
178 191 } else {
179 192 alerter.send_health_alert(
M pom/src/cli/tasks/meta_alert.rs +9 -1
@@ -33,8 +33,16 @@ pub(crate) fn spawn_meta_alert_task(
33 33 let mut interval = tokio::time::interval(
34 34 std::time::Duration::from_secs(meta_interval_secs),
35 35 );
36 + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
36 37 interval.tick().await; // consume immediate first tick
37 - let mut was_all_down = false;
38 + // Seed from the ledger so a restart while monitoring is already-offline
39 + // doesn't re-fire the alert: was-down iff the latest monitoring alert is
40 + // the offline (not the recovery) one.
41 + let mut was_all_down = db::get_latest_alert_matching(&pool, "monitoring:self", "monitoring_%")
42 + .await
43 + .ok()
44 + .flatten()
45 + .is_some_and(|a| a.alert_type == "monitoring_offline");
38 46
39 47 // A snapshot older than this is treated as unreachable, not its stale
40 48 // status. Without it a probe task that has died (stopped inserting rows)
M pom/src/cli/tasks/scan_pipeline.rs +15 -3
@@ -8,9 +8,11 @@ use tracing::{info, warn};
8 8 use pom::alerts::Alerter;
9 9 use pom::checks::scan_pipeline;
10 10 use pom::config::Config;
11 + use pom::db;
11 12
12 13 pub(crate) fn spawn_scan_pipeline_tasks(
13 14 config: &Config,
15 + pool: &sqlx::SqlitePool,
14 16 cancel: &tokio_util::sync::CancellationToken,
15 17 alerter: &Option<Alerter>,
16 18 ) -> Vec<JoinHandle<()>> {
@@ -23,6 +25,7 @@ pub(crate) fn spawn_scan_pipeline_tasks(
23 25 let name = name.clone();
24 26 let label = target_config.label.clone();
25 27 let alerter = alerter.clone();
28 + let pool = pool.clone();
26 29 let cancel = cancel.clone();
27 30
28 31 info!(
@@ -34,11 +37,20 @@ pub(crate) fn spawn_scan_pipeline_tasks(
34 37 let mut interval = tokio::time::interval(
35 38 std::time::Duration::from_secs(sp_config.interval_secs),
36 39 );
40 + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
37 41 interval.tick().await; // consume immediate first tick
38 42
39 - // Track previous status so we only fire alerts on transitions.
40 - // None means we haven't observed yet.
41 - let mut previous_status: Option<String> = None;
43 + // Track previous status so we only fire alerts on transitions. Seed
44 + // from the ledger so a restart while the pipeline is already-degraded
45 + // doesn't re-fire: if the latest scan-pipeline alert is the degraded
46 + // (not the recovery) one, start from that non-operational status.
47 + let mut previous_status: Option<String> =
48 + db::get_latest_alert_matching(&pool, &format!("scan_pipeline:{name}"), "scan_pipeline_%")
49 + .await
50 + .ok()
51 + .flatten()
52 + .filter(|a| a.alert_type == "scan_pipeline_degraded")
53 + .and_then(|a| a.to_status);
42 54
43 55 loop {
44 56 tokio::select! {
A pom/src/cli/transition.rs +129
@@ -0,0 +1,129 @@
1 + //! N-of-M status-transition debounce, shared by the health and ssh-banner loops.
2 + //!
3 + //! The probe loops insert a health snapshot every tick (the raw history), but a
4 + //! *transition* — the thing that pages an operator and opens an incident — must
5 + //! not fire on the first differing check: one transient timeout would flip
6 + //! `Operational -> Unreachable`, email an alert, and open an incident that
7 + //! resolves on the next probe (fuzz-2026-07-06: alert fatigue + incident noise).
8 + //!
9 + //! [`TransitionGate`] only reports a transition once a new status has been
10 + //! observed for `confirmations` consecutive checks, and it is **seeded from the
11 + //! last stored status on startup**, so a restart of a persistently-degraded
12 + //! target does not re-fire its alert.
13 +
14 + use pom::types::HealthStatus;
15 +
16 + /// Tracks the confirmed status and a pending candidate change.
17 + #[derive(Debug, Clone)]
18 + pub(crate) struct TransitionGate {
19 + /// The status we have committed to (alerted/incident-tracked on).
20 + confirmed: Option<HealthStatus>,
21 + /// A differing status observed but not yet confirmed: (candidate, run length).
22 + pending: Option<(HealthStatus, u32)>,
23 + /// Consecutive matching observations required to confirm a change.
24 + confirmations: u32,
25 + }
26 +
27 + impl TransitionGate {
28 + /// Seed from the last stored status (`None` if the target has no history).
29 + /// `confirmations` is clamped to at least 1 (1 = alert on first difference).
30 + pub(crate) fn seeded(confirmed: Option<HealthStatus>, confirmations: u32) -> Self {
31 + Self {
32 + confirmed,
33 + pending: None,
34 + confirmations: confirmations.max(1),
35 + }
36 + }
37 +
38 + /// Feed one observed status. Returns `Some((from, to))` only when a change is
39 + /// *confirmed* — observed for `confirmations` consecutive checks. Returns
40 + /// `None` while a change is still pending, when the status is unchanged, or on
41 + /// the very first observation of a target with no prior confirmed status.
42 + pub(crate) fn observe(
43 + &mut self,
44 + observed: HealthStatus,
45 + ) -> Option<(HealthStatus, HealthStatus)> {
46 + match self.confirmed {
47 + // No prior status (fresh target with no history): adopt silently — the
48 + // old code likewise needs a `previous` row before it alerts.
49 + None => {
50 + self.confirmed = Some(observed);
51 + self.pending = None;
52 + None
53 + }
54 + // Back to (or still at) the confirmed status: any pending change was a
55 + // blip; drop it.
56 + Some(cur) if observed == cur => {
57 + self.pending = None;
58 + None
59 + }
60 + // Differs from confirmed: accumulate consecutive agreement.
61 + Some(cur) => {
62 + let count = match self.pending {
63 + Some((p, c)) if p == observed => c + 1,
64 + _ => 1,
65 + };
66 + if count >= self.confirmations {
67 + self.pending = None;
68 + self.confirmed = Some(observed);
69 + Some((cur, observed))
70 + } else {
71 + self.pending = Some((observed, count));
72 + None
73 + }
74 + }
75 + }
76 + }
77 + }
78 +
79 + #[cfg(test)]
80 + mod tests {
81 + use super::*;
82 + use pom::types::HealthStatus::*;
83 +
84 + #[test]
85 + fn confirmations_1_fires_on_first_difference() {
86 + let mut g = TransitionGate::seeded(Some(Operational), 1);
87 + assert_eq!(g.observe(Error), Some((Operational, Error)));
88 + assert_eq!(g.observe(Error), None); // now confirmed
89 + }
90 +
91 + #[test]
92 + fn single_blip_is_absorbed_at_confirmations_2() {
93 + let mut g = TransitionGate::seeded(Some(Operational), 2);
94 + // One differing check → pending, no transition.
95 + assert_eq!(g.observe(Error), None);
96 + // Back to operational → the blip is dropped.
97 + assert_eq!(g.observe(Operational), None);
98 + // A later genuine outage still needs two in a row.
99 + assert_eq!(g.observe(Error), None);
100 + assert_eq!(g.observe(Error), Some((Operational, Error)));
101 + }
102 +
103 + #[test]
104 + fn flapping_candidate_resets_the_run() {
105 + let mut g = TransitionGate::seeded(Some(Operational), 3);
106 + assert_eq!(g.observe(Error), None); // Error x1
107 + assert_eq!(g.observe(Unreachable), None); // candidate changed → Unreachable x1
108 + assert_eq!(g.observe(Unreachable), None); // Unreachable x2
109 + assert_eq!(g.observe(Unreachable), Some((Operational, Unreachable))); // x3 confirms
110 + }
111 +
112 + #[test]
113 + fn seeded_status_does_not_refire_on_restart() {
114 + // Restart: last stored status was Error. Observing Error again must NOT
115 + // produce a transition (that was the re-fire-on-restart bug).
116 + let mut g = TransitionGate::seeded(Some(Error), 2);
117 + assert_eq!(g.observe(Error), None);
118 + assert_eq!(g.observe(Error), None);
119 + }
120 +
121 + #[test]
122 + fn first_ever_observation_is_silent() {
123 + let mut g = TransitionGate::seeded(None, 2);
124 + assert_eq!(g.observe(Operational), None);
125 + // Now Operational is confirmed; a change debounces normally.
126 + assert_eq!(g.observe(Error), None);
127 + assert_eq!(g.observe(Error), Some((Operational, Error)));
128 + }
129 + }
M pom/src/config.rs +11
@@ -99,6 +99,16 @@ pub struct ServeConfig {
99 99 /// Enable the HTML dashboard at `GET /`. Disabled by default.
100 100 #[serde(default)]
101 101 pub dashboard: bool,
102 + /// Consecutive checks that must agree on a new status before a health/ssh
103 + /// transition fires an alert or opens an incident (N-of-M debounce). Default
104 + /// 2: a single transient blip no longer pages or logs a false incident. Set 1
105 + /// to alert on the first differing check (the pre-debounce behavior).
106 + #[serde(default = "default_confirmations")]
107 + pub confirmations: u32,
108 + }
109 +
110 + fn default_confirmations() -> u32 {
111 + 2
102 112 }
103 113
104 114 impl Default for ServeConfig {
@@ -115,6 +125,7 @@ impl Default for ServeConfig {
115 125 whois_check_interval_secs: 86400,
116 126 api_token: None,
117 127 dashboard: false,
128 + confirmations: default_confirmations(),
118 129 }
119 130 }
120 131 }
M pom/src/peer.rs +1
@@ -205,6 +205,7 @@ async fn heartbeat_loop(
205 205 cancel: tokio_util::sync::CancellationToken,
206 206 ) {
207 207 let mut interval = tokio::time::interval(std::time::Duration::from_secs(interval_secs));
208 + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
208 209 // Skip the first immediate tick — give peers time to start up
209 210 interval.tick().await;
210 211