Skip to main content

max / makenotwork

pom: persist-and-retry undelivered alerts; fix recovery cooldowns CRITICAL #1 — a failed alert send consumed the status transition. Alerts fire only on transition (the snapshot is inserted every tick, so the DB status has already moved), so a transient WAM/Postmark failure at the operational->error edge meant silence for the whole outage, not just the cooldown. Gating record_alert on delivery alone can't fix it — the transition never repeats. Close it with a durable queue that outlives the transition: a new pending_alerts table (migration 10), a single dispatch choke point in the Alerter that records on success and enqueues on failure, and a retry drainer task in serve that re-delivers due alerts (exp backoff, dead-letter after 8 tries) and records them in the ledger once they land. All ~24 send sites route through dispatch_wam/ dispatch_email; the retry path reuses the same senders. Folds in the same-root HIGH items: recoveries now have their own cooldown (a flapping target can't spam one recovery email per flap), and a recorded recovery clears the failure cooldown so a genuine fail->recover->fail fires the second failure (the old behavior was enshrined by a test that encoded the bug — rewritten to assert the fix). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-07 16:45 UTC
Commit: bfb496292d7d66d6a5df017a0a7894ab3dd605b3
Parent: 7335fec
4 files changed, +406 insertions, -79 deletions
M pom/src/alerts.rs +212 -66
@@ -53,6 +53,28 @@ pub struct Alerter {
53 53 wam_url: Option<String>,
54 54 }
55 55
56 + /// The record-context an alert carries: what to write to the `alerts` ledger on
57 + /// successful delivery, and what to persist for retry on failure.
58 + struct AlertMeta<'a> {
59 + key: &'a str,
60 + category: AlertCategory,
61 + from: Option<&'a str>,
62 + to: Option<&'a str>,
63 + error: Option<&'a str>,
64 + }
65 +
66 + /// `true` if `sent_at` (rfc3339) is within `cooldown_secs` of now. An unparseable
67 + /// timestamp is treated as expired (not in cooldown) so a bad row can never wedge
68 + /// alerting shut.
69 + fn within_cooldown_secs(sent_at: &str, cooldown_secs: u64) -> bool {
70 + match chrono::DateTime::parse_from_rfc3339(sent_at) {
71 + Ok(dt) => {
72 + chrono::Utc::now().signed_duration_since(dt).num_seconds() < cooldown_secs as i64
73 + }
74 + Err(_) => false,
75 + }
76 + }
77 +
56 78 impl Alerter {
57 79 pub fn new(config: AlertConfig, pool: SqlitePool, instance_name: String) -> Self {
58 80 let client = reqwest::Client::builder()
@@ -99,8 +121,8 @@ impl Alerter {
99 121 body.push_str("\n- PoM");
100 122
101 123 let priority = health_status_priority(to_status);
102 - self.wam_ticket(&subject, &body, priority, "pom-health", Some(target)).await;
103 - self.record_alert(&alert_key, AlertCategory::Health, Some(from_status), Some(to_status), error).await;
124 + self.dispatch_wam(&subject, &body, priority, "pom-health", Some(target),
125 + AlertMeta { key: &alert_key, category: AlertCategory::Health, from: Some(from_status), to: Some(to_status), error }).await;
104 126 }
105 127
106 128 #[instrument(skip_all)]
@@ -111,7 +133,12 @@ impl Alerter {
111 133 from_status: &str,
112 134 ) {
113 135 let alert_key = format!("health:{target}");
114 - // No cooldown on recovery — always send
136 + // Recoveries are throttled by their own cooldown so a flapping target
137 + // can't spam one recovery email per flap (fuzz-2026-07-06).
138 + if self.is_recovery_within_cooldown(&alert_key).await {
139 + info!("recovery cooldown active for {alert_key}, skipping");
140 + return;
141 + }
115 142 let subject = format!("[PoM] {target}: recovered");
116 143 let body = format!(
117 144 "Target: {label} ({target})\n\
@@ -123,8 +150,8 @@ impl Alerter {
123 150 chrono::Utc::now().to_rfc3339(),
124 151 );
125 152
126 - self.send_email(&subject, &body).await;
127 - self.record_alert(&alert_key, AlertCategory::Recovery, Some(from_status), Some("operational"), None).await;
153 + self.dispatch_email(&subject, &body,
154 + AlertMeta { key: &alert_key, category: AlertCategory::Recovery, from: Some(from_status), to: Some("operational"), error: None }).await;
128 155 }
129 156
130 157 #[instrument(skip_all)]
@@ -155,8 +182,8 @@ impl Alerter {
155 182 );
156 183
157 184 let priority = tls_expiry_priority(days_remaining);
158 - self.wam_ticket(&subject, &body, priority, "pom-tls", Some(&format!("{target}:{host}"))).await;
159 - self.record_alert(&alert_key, AlertCategory::TlsExpiry, None, None, None).await;
185 + self.dispatch_wam(&subject, &body, priority, "pom-tls", Some(&format!("{target}:{host}")),
186 + AlertMeta { key: &alert_key, category: AlertCategory::TlsExpiry, from: None, to: None, error: None }).await;
160 187 }
161 188
162 189 #[instrument(skip_all)]
@@ -184,8 +211,8 @@ impl Alerter {
184 211 chrono::Utc::now().to_rfc3339(),
185 212 );
186 213
187 - self.wam_ticket(&subject, &body, "high", "pom-tls", Some(&format!("{target}:{host}"))).await;
188 - self.record_alert(&alert_key, AlertCategory::TlsError, None, None, Some(error)).await;
214 + self.dispatch_wam(&subject, &body, "high", "pom-tls", Some(&format!("{target}:{host}")),
215 + AlertMeta { key: &alert_key, category: AlertCategory::TlsError, from: None, to: None, error: Some(error) }).await;
189 216 }
190 217
191 218 #[instrument(skip_all)]
@@ -196,7 +223,12 @@ impl Alerter {
196 223 days_remaining: i64,
197 224 ) {
198 225 let alert_key = format!("tls:{target}");
199 - // No cooldown on recovery — always send
226 + // Recoveries are throttled by their own cooldown so a flapping target
227 + // can't spam one recovery email per flap (fuzz-2026-07-06).
228 + if self.is_recovery_within_cooldown(&alert_key).await {
229 + info!("recovery cooldown active for {alert_key}, skipping");
230 + return;
231 + }
200 232 let subject = format!("[PoM] {target}: TLS cert renewed");
201 233 let body = format!(
202 234 "Target: {label} ({target})\n\
@@ -208,8 +240,8 @@ impl Alerter {
208 240 chrono::Utc::now().to_rfc3339(),
209 241 );
210 242
211 - self.send_email(&subject, &body).await;
212 - self.record_alert(&alert_key, AlertCategory::TlsRecovery, None, None, None).await;
243 + self.dispatch_email(&subject, &body,
244 + AlertMeta { key: &alert_key, category: AlertCategory::TlsRecovery, from: None, to: None, error: None }).await;
213 245 }
214 246
215 247 #[instrument(skip_all)]
@@ -237,8 +269,8 @@ impl Alerter {
237 269 chrono::Utc::now().to_rfc3339(),
238 270 );
239 271
240 - self.wam_ticket(&subject, &body, "high", "pom-peer", Some(peer_name)).await;
241 - self.record_alert(&alert_key, AlertCategory::PeerMissing, None, None, None).await;
272 + self.dispatch_wam(&subject, &body, "high", "pom-peer", Some(peer_name),
273 + AlertMeta { key: &alert_key, category: AlertCategory::PeerMissing, from: None, to: None, error: None }).await;
242 274 }
243 275
244 276 #[instrument(skip_all)]
@@ -259,8 +291,12 @@ impl Alerter {
259 291 );
260 292
261 293 let alert_key = format!("peer:{peer_name}");
262 - self.send_email(&subject, &body).await;
263 - self.record_alert(&alert_key, AlertCategory::PeerRecovery, None, None, None).await;
294 + if self.is_recovery_within_cooldown(&alert_key).await {
295 + info!("recovery cooldown active for {alert_key}, skipping");
296 + return;
297 + }
298 + self.dispatch_email(&subject, &body,
299 + AlertMeta { key: &alert_key, category: AlertCategory::PeerRecovery, from: None, to: None, error: None }).await;
264 300 }
265 301
266 302 #[instrument(skip_all)]
@@ -289,8 +325,8 @@ impl Alerter {
289 325 chrono::Utc::now().to_rfc3339(),
290 326 );
291 327
292 - self.wam_ticket(&subject, &body, "high", "pom-routes", Some(target)).await;
293 - self.record_alert(&alert_key, AlertCategory::RouteFailure, None, None, None).await;
328 + self.dispatch_wam(&subject, &body, "high", "pom-routes", Some(target),
329 + AlertMeta { key: &alert_key, category: AlertCategory::RouteFailure, from: None, to: None, error: None }).await;
294 330 }
295 331
296 332 #[instrument(skip_all)]
@@ -300,8 +336,13 @@ impl Alerter {
300 336 label: &str,
301 337 recovered_paths: &[String],
302 338 ) {
303 - // No cooldown on recovery — always send
304 339 let alert_key = format!("route:{target}");
340 + // Recoveries are throttled by their own cooldown so a flapping target
341 + // can't spam one recovery email per flap (fuzz-2026-07-06).
342 + if self.is_recovery_within_cooldown(&alert_key).await {
343 + info!("recovery cooldown active for {alert_key}, skipping");
344 + return;
345 + }
305 346 let subject = format!("[PoM] {label}: routes recovered");
306 347 let body = format!(
307 348 "Target: {label} ({target})\n\
@@ -314,8 +355,8 @@ impl Alerter {
314 355 chrono::Utc::now().to_rfc3339(),
315 356 );
316 357
317 - self.send_email(&subject, &body).await;
318 - self.record_alert(&alert_key, AlertCategory::RouteRecovery, None, None, None).await;
358 + self.dispatch_email(&subject, &body,
359 + AlertMeta { key: &alert_key, category: AlertCategory::RouteRecovery, from: None, to: None, error: None }).await;
319 360 }
320 361
321 362 #[instrument(skip_all)]
@@ -357,8 +398,8 @@ impl Alerter {
357 398 chrono::Utc::now().to_rfc3339(),
358 399 );
359 400
360 - self.wam_ticket(&subject, &body, "high", "pom-dns", Some(target)).await;
361 - self.record_alert(&alert_key, AlertCategory::DnsMismatch, None, None, None).await;
401 + self.dispatch_wam(&subject, &body, "high", "pom-dns", Some(target),
402 + AlertMeta { key: &alert_key, category: AlertCategory::DnsMismatch, from: None, to: None, error: None }).await;
362 403 }
363 404
364 405 #[instrument(skip_all)]
@@ -367,8 +408,13 @@ impl Alerter {
367 408 target: &str,
368 409 label: &str,
369 410 ) {
370 - // No cooldown on recovery — always send
371 411 let alert_key = format!("dns:{target}");
412 + // Recoveries are throttled by their own cooldown so a flapping target
413 + // can't spam one recovery email per flap (fuzz-2026-07-06).
414 + if self.is_recovery_within_cooldown(&alert_key).await {
415 + info!("recovery cooldown active for {alert_key}, skipping");
416 + return;
417 + }
372 418 let subject = format!("[PoM] {label}: DNS records recovered");
373 419 let body = format!(
374 420 "Target: {label} ({target})\n\
@@ -380,8 +426,8 @@ impl Alerter {
380 426 chrono::Utc::now().to_rfc3339(),
381 427 );
382 428
383 - self.send_email(&subject, &body).await;
384 - self.record_alert(&alert_key, AlertCategory::DnsRecovery, None, None, None).await;
429 + self.dispatch_email(&subject, &body,
430 + AlertMeta { key: &alert_key, category: AlertCategory::DnsRecovery, from: None, to: None, error: None }).await;
385 431 }
386 432
387 433 #[instrument(skip_all)]
@@ -411,8 +457,8 @@ impl Alerter {
411 457 );
412 458
413 459 let priority = whois_expiry_priority(days_remaining);
414 - self.wam_ticket(&subject, &body, priority, "pom-whois", Some(&format!("{target}:{domain}"))).await;
415 - self.record_alert(&alert_key, AlertCategory::WhoisExpiry, None, None, None).await;
460 + self.dispatch_wam(&subject, &body, priority, "pom-whois", Some(&format!("{target}:{domain}")),
461 + AlertMeta { key: &alert_key, category: AlertCategory::WhoisExpiry, from: None, to: None, error: None }).await;
416 462 }
417 463
418 464 #[instrument(skip_all)]
@@ -441,8 +487,8 @@ impl Alerter {
441 487 chrono::Utc::now().to_rfc3339(),
442 488 );
443 489
444 - self.wam_ticket(&subject, &body, "high", "pom-whois", Some(&format!("{target}:{domain}"))).await;
445 - self.record_alert(&alert_key, AlertCategory::WhoisError, None, None, Some(error)).await;
490 + self.dispatch_wam(&subject, &body, "high", "pom-whois", Some(&format!("{target}:{domain}")),
491 + AlertMeta { key: &alert_key, category: AlertCategory::WhoisError, from: None, to: None, error: Some(error) }).await;
446 492 }
447 493
448 494 #[instrument(skip_all)]
@@ -482,8 +528,8 @@ impl Alerter {
482 528 chrono::Utc::now().to_rfc3339(),
483 529 );
484 530
485 - self.wam_ticket(&subject, &body, "high", "pom-cors", Some(target)).await;
486 - self.record_alert(&alert_key, AlertCategory::CorsFailure, None, None, None).await;
531 + self.dispatch_wam(&subject, &body, "high", "pom-cors", Some(target),
532 + AlertMeta { key: &alert_key, category: AlertCategory::CorsFailure, from: None, to: None, error: None }).await;
487 533 }
488 534
489 535 #[instrument(skip_all)]
@@ -492,8 +538,13 @@ impl Alerter {
492 538 target: &str,
493 539 label: &str,
494 540 ) {
495 - // No cooldown on recovery — always send
496 541 let alert_key = format!("cors:{target}");
542 + // Recoveries are throttled by their own cooldown so a flapping target
543 + // can't spam one recovery email per flap (fuzz-2026-07-06).
544 + if self.is_recovery_within_cooldown(&alert_key).await {
545 + info!("recovery cooldown active for {alert_key}, skipping");
546 + return;
547 + }
497 548 let subject = format!("[PoM] {label}: CORS preflights recovered");
498 549 let body = format!(
499 550 "Target: {label} ({target})\n\
@@ -505,8 +556,8 @@ impl Alerter {
505 556 chrono::Utc::now().to_rfc3339(),
506 557 );
507 558
508 - self.send_email(&subject, &body).await;
509 - self.record_alert(&alert_key, AlertCategory::CorsRecovery, None, None, None).await;
559 + self.dispatch_email(&subject, &body,
560 + AlertMeta { key: &alert_key, category: AlertCategory::CorsRecovery, from: None, to: None, error: None }).await;
510 561 }
511 562
512 563 #[instrument(skip_all)]
@@ -533,8 +584,8 @@ impl Alerter {
533 584 chrono::Utc::now().to_rfc3339(),
534 585 );
535 586
536 - self.wam_ticket(&subject, &body, "medium", "pom-latency", Some(target)).await;
537 - self.record_alert(&alert_key, AlertCategory::LatencyDrift, None, None, Some(drift_message)).await;
587 + self.dispatch_wam(&subject, &body, "medium", "pom-latency", Some(target),
588 + AlertMeta { key: &alert_key, category: AlertCategory::LatencyDrift, from: None, to: None, error: Some(drift_message) }).await;
538 589 }
539 590
540 591 #[instrument(skip_all)]
@@ -543,8 +594,13 @@ impl Alerter {
543 594 target: &str,
544 595 label: &str,
545 596 ) {
546 - // No cooldown on recovery — always send
547 597 let alert_key = format!("latency:{target}");
598 + // Recoveries are throttled by their own cooldown so a flapping target
599 + // can't spam one recovery email per flap (fuzz-2026-07-06).
600 + if self.is_recovery_within_cooldown(&alert_key).await {
601 + info!("recovery cooldown active for {alert_key}, skipping");
602 + return;
603 + }
548 604 let subject = format!("[PoM] {target}: latency recovered");
549 605 let body = format!(
550 606 "Target: {label} ({target})\n\
@@ -556,8 +612,8 @@ impl Alerter {
556 612 chrono::Utc::now().to_rfc3339(),
557 613 );
558 614
559 - self.send_email(&subject, &body).await;
560 - self.record_alert(&alert_key, AlertCategory::LatencyRecovery, None, None, None).await;
615 + self.dispatch_email(&subject, &body,
616 + AlertMeta { key: &alert_key, category: AlertCategory::LatencyRecovery, from: None, to: None, error: None }).await;
561 617 }
562 618
563 619 #[instrument(skip_all)]
@@ -584,8 +640,8 @@ impl Alerter {
584 640 chrono::Utc::now().to_rfc3339(),
585 641 );
586 642
587 - self.wam_ticket(&subject, &body, "medium", "pom-test-duration", Some(target)).await;
588 - self.record_alert(&alert_key, AlertCategory::TestDurationDrift, None, None, Some(drift_message)).await;
643 + self.dispatch_wam(&subject, &body, "medium", "pom-test-duration", Some(target),
644 + AlertMeta { key: &alert_key, category: AlertCategory::TestDurationDrift, from: None, to: None, error: Some(drift_message) }).await;
589 645 }
590 646
591 647 #[instrument(skip_all)]
@@ -619,8 +675,8 @@ impl Alerter {
619 675 );
620 676
621 677 let priority = backup_status_priority(status);
622 - self.wam_ticket(&subject, &body, priority, "pom-backup", Some(&format!("{target}:{database}"))).await;
623 - self.record_alert(&alert_key, AlertCategory::BackupStale, None, Some(status), None).await;
678 + self.dispatch_wam(&subject, &body, priority, "pom-backup", Some(&format!("{target}:{database}")),
679 + AlertMeta { key: &alert_key, category: AlertCategory::BackupStale, from: None, to: Some(status), error: None }).await;
624 680 }
625 681
626 682 #[instrument(skip_all)]
@@ -630,8 +686,13 @@ impl Alerter {
630 686 label: &str,
631 687 database: &str,
632 688 ) {
633 - // No cooldown on recovery — always send
634 689 let alert_key = format!("backup:{target}:{database}");
690 + // Recoveries are throttled by their own cooldown so a flapping target
691 + // can't spam one recovery email per flap (fuzz-2026-07-06).
692 + if self.is_recovery_within_cooldown(&alert_key).await {
693 + info!("recovery cooldown active for {alert_key}, skipping");
694 + return;
695 + }
635 696 let subject = format!("[PoM] {label}: {database} backup recovered");
636 697 let body = format!(
637 698 "Target: {label} ({target})\n\
@@ -644,8 +705,8 @@ impl Alerter {
644 705 chrono::Utc::now().to_rfc3339(),
645 706 );
646 707
647 - self.send_email(&subject, &body).await;
648 - self.record_alert(&alert_key, AlertCategory::BackupRecovery, None, Some("ok"), None).await;
708 + self.dispatch_email(&subject, &body,
709 + AlertMeta { key: &alert_key, category: AlertCategory::BackupRecovery, from: None, to: Some("ok"), error: None }).await;
649 710 }
650 711
651 712 /// Fire when the scan pipeline transitions from operational → degraded or
@@ -680,14 +741,18 @@ impl Alerter {
680 741 );
681 742
682 743 let priority = if status == "unreachable" { "high" } else { "medium" };
683 - self.wam_ticket(&subject, &body, priority, "pom-scan-pipeline", Some(target)).await;
684 - self.record_alert(&alert_key, AlertCategory::ScanPipelineDegraded, None, Some(status), None).await;
744 + self.dispatch_wam(&subject, &body, priority, "pom-scan-pipeline", Some(target),
745 + AlertMeta { key: &alert_key, category: AlertCategory::ScanPipelineDegraded, from: None, to: Some(status), error: None }).await;
685 746 }
686 747
687 748 /// Fire on recovery from a degraded / unreachable scan-pipeline state.
688 749 #[instrument(skip_all)]
689 750 pub async fn send_scan_pipeline_recovery(&self, target: &str, label: &str) {
690 751 let alert_key = format!("scan_pipeline:{target}");
752 + if self.is_recovery_within_cooldown(&alert_key).await {
753 + info!("recovery cooldown active for {alert_key}, skipping");
754 + return;
755 + }
691 756 let subject = format!("[PoM] {label}: scan pipeline recovered");
692 757 let body = format!(
693 758 "Target: {label} ({target})\n\
@@ -699,8 +764,8 @@ impl Alerter {
699 764 chrono::Utc::now().to_rfc3339(),
700 765 );
701 766
702 - self.send_email(&subject, &body).await;
703 - self.record_alert(&alert_key, AlertCategory::ScanPipelineRecovery, None, Some("operational"), None).await;
767 + self.dispatch_email(&subject, &body,
768 + AlertMeta { key: &alert_key, category: AlertCategory::ScanPipelineRecovery, from: None, to: Some("operational"), error: None }).await;
704 769 }
705 770
706 771 /// All monitored targets are unreachable — likely a network issue with PoM itself.
@@ -724,14 +789,18 @@ impl Alerter {
724 789 chrono::Utc::now().to_rfc3339(),
725 790 );
726 791
727 - self.wam_ticket(&subject, &body, "critical", "pom-monitoring", Some("self")).await;
728 - self.record_alert(alert_key, AlertCategory::MonitoringOffline, None, None, None).await;
792 + self.dispatch_wam(&subject, &body, "critical", "pom-monitoring", Some("self"),
793 + AlertMeta { key: alert_key, category: AlertCategory::MonitoringOffline, from: None, to: None, error: None }).await;
729 794 }
730 795
731 796 /// At least one target is reachable again after a monitoring-offline event.
732 797 #[instrument(skip_all)]
733 798 pub async fn send_monitoring_recovery(&self) {
734 799 let alert_key = "monitoring:self";
800 + if self.is_recovery_within_cooldown(alert_key).await {
801 + info!("recovery cooldown active for {alert_key}, skipping");
802 + return;
803 + }
735 804 let subject = "[PoM] monitoring recovered".to_string();
736 805 let body = format!(
737 806 "At least one target is reachable again.\n\
@@ -742,23 +811,46 @@ impl Alerter {
742 811 chrono::Utc::now().to_rfc3339(),
743 812 );
744 813
745 - self.send_email(&subject, &body).await;
746 - self.record_alert(alert_key, AlertCategory::MonitoringRecovery, None, None, None).await;
814 + self.dispatch_email(&subject, &body,
815 + AlertMeta { key: alert_key, category: AlertCategory::MonitoringRecovery, from: None, to: None, error: None }).await;
747 816 }
748 817
818 + /// Whether a FAILURE alert for `target` is within its cooldown.
819 + ///
820 + /// A recorded recovery *newer* than the last failure clears the cooldown: the
821 + /// target came back, so a fresh failure is a genuine new outage and must not
822 + /// be suppressed by the prior failure's window (fuzz-2026-07-06 fail→recover→
823 + /// fail suppression).
749 824 async fn is_within_cooldown(&self, target: &str) -> bool {
750 - let latest = match db::get_latest_alert_for_target(&self.pool, target).await {
751 - Ok(Some(row)) => row,
752 - _ => return false,
825 + let Ok(Some(latest_fail)) = db::get_latest_alert_for_target(&self.pool, target).await else {
826 + return false; // never alerted → not in cooldown
753 827 };
828 + // A recovery after the last failure voids the cooldown.
829 + if let Ok(Some(rec)) =
830 + db::get_latest_alert_matching(&self.pool, target, "%recovery%").await
831 + && rec.id > latest_fail.id
832 + {
833 + return false;
834 + }
835 + within_cooldown_secs(&latest_fail.sent_at, self.config.cooldown_secs)
836 + }
754 837
755 - let sent_at = match chrono::DateTime::parse_from_rfc3339(&latest.sent_at) {
756 - Ok(dt) => dt,
757 - Err(_) => return false,
838 + /// Whether a RECOVERY for `target` is within its own cooldown — throttles a
839 + /// flapping target so it can't emit one recovery email per flap
840 + /// (fuzz-2026-07-06 un-throttled recoveries). A failure newer than the last
841 + /// recovery clears it, so a genuine recover-after-a-new-outage still sends.
842 + async fn is_recovery_within_cooldown(&self, target: &str) -> bool {
843 + let Ok(Some(latest_rec)) =
844 + db::get_latest_alert_matching(&self.pool, target, "%recovery%").await
845 + else {
846 + return false;
758 847 };
759 -
760 - let elapsed = chrono::Utc::now().signed_duration_since(sent_at);
761 - elapsed.num_seconds() < self.config.cooldown_secs as i64
848 + if let Ok(Some(fail)) = db::get_latest_alert_for_target(&self.pool, target).await
849 + && fail.id > latest_rec.id
850 + {
851 + return false;
852 + }
853 + within_cooldown_secs(&latest_rec.sent_at, self.config.cooldown_secs)
762 854 }
763 855
764 856 /// Send an email via Postmark. Returns `true` if the message was accepted
@@ -818,12 +910,118 @@ impl Alerter {
818 910 to_status: Option<&str>,
819 911 error: Option<&str>,
820 912 ) {
821 - let alert_type_str = alert_type.to_string();
822 - if let Err(e) = db::insert_alert(&self.pool, target, &alert_type_str, from_status, to_status, error).await {
913 + self.record_alert_str(target, &alert_type.to_string(), from_status, to_status, error).await;
914 + }
915 +
916 + /// String-typed variant so the retry task can record a queued alert straight
917 + /// from its stored category without round-tripping through the enum.
918 + async fn record_alert_str(
919 + &self,
920 + target: &str,
921 + alert_type: &str,
922 + from_status: Option<&str>,
923 + to_status: Option<&str>,
924 + error: Option<&str>,
925 + ) {
926 + if let Err(e) = db::insert_alert(&self.pool, target, alert_type, from_status, to_status, error).await {
823 927 warn!("failed to record alert: {e}");
824 928 }
825 929 }
826 930
931 + /// Dispatch a WAM failure alert: attempt delivery (WAM, falling back to
932 + /// email), record it on success, or persist it for retry on failure. Gating
933 + /// the ledger write on delivery — and queueing the miss — is what stops a
934 + /// transient send failure at a status transition from silencing the whole
935 + /// outage: the transition fires once, but the queued alert is retried until it
936 + /// lands (fuzz-2026-07-06 CRITICAL #1).
937 + async fn dispatch_wam(
938 + &self,
939 + subject: &str,
940 + body: &str,
941 + priority: &str,
942 + source: &str,
943 + source_ref: Option<&str>,
944 + meta: AlertMeta<'_>,
945 + ) {
946 + if self.wam_ticket(subject, body, priority, source, source_ref).await {
947 + self.record_alert(meta.key, meta.category, meta.from, meta.to, meta.error).await;
948 + } else {
949 + self.enqueue_undelivered(subject, body, "wam", Some(priority), Some(source), source_ref, &meta).await;
950 + }
951 + }
952 +
953 + /// Dispatch an email alert (recoveries, and failure alerts that already chose
954 + /// email): record on success, persist for retry on failure.
955 + async fn dispatch_email(&self, subject: &str, body: &str, meta: AlertMeta<'_>) {
956 + if self.send_email(subject, body).await {
957 + self.record_alert(meta.key, meta.category, meta.from, meta.to, meta.error).await;
958 + } else {
959 + self.enqueue_undelivered(subject, body, "email", None, None, None, &meta).await;
960 + }
961 + }
962 +
963 + #[allow(clippy::too_many_arguments)]
964 + async fn enqueue_undelivered(
965 + &self,
966 + subject: &str,
967 + body: &str,
968 + channel: &str,
969 + priority: Option<&str>,
970 + source: Option<&str>,
971 + source_ref: Option<&str>,
972 + meta: &AlertMeta<'_>,
Lines truncated
M pom/src/cli/serve.rs +37
@@ -85,6 +85,43 @@ pub(crate) async fn cmd_serve(
85 85 handles.push(handle);
86 86 }
87 87
88 + // Spawn the pending-alert retry drainer. When an alert's original send failed
89 + // it was queued (the status transition fires only once, so without this the
90 + // outage would go silent); re-deliver it here until it lands. This is the
91 + // durable half of the fix for the transition-once alert-loss CRITICAL.
92 + if let Some(retry_alerter) = alerter.clone() {
93 + let pool = pool.clone();
94 + let cancel = token.clone();
95 + handles.push(tokio::spawn(async move {
96 + const MAX_ATTEMPTS: i64 = 8; // ~ several hours of backoff before dead-letter
97 + let mut interval = tokio::time::interval(std::time::Duration::from_secs(30));
98 + interval.tick().await; // consume the immediate first tick
99 + loop {
100 + tokio::select! {
101 + _ = cancel.cancelled() => break,
102 + _ = interval.tick() => {}
103 + }
104 + let due = db::due_pending_alerts(&pool, 50).await.unwrap_or_default();
105 + for p in due {
106 + if retry_alerter.retry_pending(&p).await {
107 + let _ = db::delete_pending_alert(&pool, p.id).await;
108 + info!("delivered previously-queued alert to {}", p.alert_key);
109 + } else if p.attempts + 1 >= MAX_ATTEMPTS {
110 + // Give up so the queue can't grow without bound; losing the
111 + // row is logged loudly rather than silently.
112 + tracing::error!("alert to {} undeliverable after {} attempts; dropping", p.alert_key, p.attempts + 1);
113 + let _ = db::delete_pending_alert(&pool, p.id).await;
114 + } else {
115 + // Exponential backoff, capped at 1h.
116 + let backoff = std::cmp::min(30 * 2i64.pow((p.attempts.min(7)) as u32), 3600);
117 + let next = (chrono::Utc::now() + chrono::Duration::seconds(backoff)).to_rfc3339();
118 + let _ = db::bump_pending_alert(&pool, p.id, &next).await;
119 + }
120 + }
121 + }
122 + }));
123 + }
124 +
88 125 // Start HTTP API server
89 126 let api_app = pom::api::router(pool.clone(), config.clone(), Some(mesh.clone()));
90 127 let api_listener = tokio::net::TcpListener::bind(&listen_addr).await?;
M pom/src/db.rs +142
@@ -174,6 +174,26 @@ const MIGRATIONS: &[(i64, &str, &str)] = &[
174 174 );
175 175 CREATE INDEX IF NOT EXISTS idx_backup_checks_target ON backup_checks(target, id DESC);
176 176 "#),
177 + (10, "add pending_alerts retry queue", r#"
178 + CREATE TABLE IF NOT EXISTS pending_alerts (
179 + id INTEGER PRIMARY KEY AUTOINCREMENT,
180 + alert_key TEXT NOT NULL,
181 + category TEXT NOT NULL,
182 + channel TEXT NOT NULL, -- 'wam' or 'email'
183 + subject TEXT NOT NULL,
184 + body TEXT NOT NULL,
185 + priority TEXT, -- WAM only
186 + source TEXT, -- WAM only
187 + source_ref TEXT, -- WAM only
188 + from_status TEXT,
189 + to_status TEXT,
190 + error TEXT,
191 + attempts INTEGER NOT NULL DEFAULT 0,
192 + created_at TEXT NOT NULL,
193 + next_retry_at TEXT NOT NULL
194 + );
195 + CREATE INDEX IF NOT EXISTS idx_pending_alerts_due ON pending_alerts(next_retry_at, id);
196 + "#),
177 197 ];
178 198
179 199 #[instrument(skip_all)]
@@ -690,6 +710,128 @@ pub async fn get_latest_alert_for_target(
690 710 .await?)
691 711 }
692 712
713 + /// The most recent alert row for a target of a specific type-set. Unlike
714 + /// [`get_latest_alert_for_target`] (which excludes recoveries), this can target
715 + /// recovery rows so recoveries get their own cooldown, and so a recorded recovery
716 + /// can clear a prior failure's cooldown (fail→recover→fail). `like` is a SQL
717 + /// LIKE pattern matched against `alert_type` (e.g. `'%recovery%'`).
718 + #[instrument(skip_all)]
719 + pub async fn get_latest_alert_matching(
720 + pool: &SqlitePool,
721 + target: &str,
722 + like: &str,
723 + ) -> Result<Option<AlertRow>> {
724 + Ok(sqlx::query_as::<_, AlertRow>(
725 + "SELECT id, target, alert_type, from_status, to_status, sent_at, error
726 + FROM alerts WHERE target = ? AND alert_type LIKE ?
727 + ORDER BY id DESC LIMIT 1",
728 + )
729 + .bind(target)
730 + .bind(like)
731 + .fetch_optional(pool)
732 + .await?)
733 + }
734 +
735 + // --- Pending-alert retry queue ---
736 +
737 + /// A delivery that failed and must be retried, so a transient WAM/Postmark error
738 + /// at a status transition can't silence the whole outage (the transition fires
739 + /// once; this queue is the durable state that outlives it).
740 + #[derive(Debug, Clone, sqlx::FromRow)]
741 + pub struct PendingAlertRow {
742 + pub id: i64,
743 + pub alert_key: String,
744 + pub category: String,
745 + pub channel: String,
746 + pub subject: String,
747 + pub body: String,
748 + pub priority: Option<String>,
749 + pub source: Option<String>,
750 + pub source_ref: Option<String>,
751 + pub from_status: Option<String>,
752 + pub to_status: Option<String>,
753 + pub error: Option<String>,
754 + pub attempts: i64,
755 + }
756 +
757 + /// Fields needed to enqueue an undelivered alert for retry.
758 + #[derive(Debug, Clone)]
759 + pub struct NewPendingAlert<'a> {
760 + pub alert_key: &'a str,
761 + pub category: &'a str,
762 + pub channel: &'a str,
763 + pub subject: &'a str,
764 + pub body: &'a str,
765 + pub priority: Option<&'a str>,
766 + pub source: Option<&'a str>,
767 + pub source_ref: Option<&'a str>,
768 + pub from_status: Option<&'a str>,
769 + pub to_status: Option<&'a str>,
770 + pub error: Option<&'a str>,
771 + }
772 +
773 + #[instrument(skip_all)]
774 + pub async fn enqueue_pending_alert(pool: &SqlitePool, a: &NewPendingAlert<'_>) -> Result<()> {
775 + let now = chrono::Utc::now().to_rfc3339();
776 + sqlx::query(
777 + "INSERT INTO pending_alerts
778 + (alert_key, category, channel, subject, body, priority, source, source_ref,
779 + from_status, to_status, error, attempts, created_at, next_retry_at)
780 + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, ?, ?)",
781 + )
782 + .bind(a.alert_key)
783 + .bind(a.category)
784 + .bind(a.channel)
785 + .bind(a.subject)
786 + .bind(a.body)
787 + .bind(a.priority)
788 + .bind(a.source)
789 + .bind(a.source_ref)
790 + .bind(a.from_status)
791 + .bind(a.to_status)
792 + .bind(a.error)
793 + .bind(&now)
794 + .bind(&now) // due immediately for the first retry
795 + .execute(pool)
796 + .await?;
797 + Ok(())
798 + }
799 +
800 + /// Pending alerts whose `next_retry_at` is due (<= now), oldest first.
801 + #[instrument(skip_all)]
802 + pub async fn due_pending_alerts(pool: &SqlitePool, limit: i64) -> Result<Vec<PendingAlertRow>> {
803 + let now = chrono::Utc::now().to_rfc3339();
804 + Ok(sqlx::query_as::<_, PendingAlertRow>(
805 + "SELECT id, alert_key, category, channel, subject, body, priority, source, source_ref,
806 + from_status, to_status, error, attempts
807 + FROM pending_alerts WHERE next_retry_at <= ? ORDER BY id ASC LIMIT ?",
808 + )
809 + .bind(&now)
810 + .bind(limit)
811 + .fetch_all(pool)
812 + .await?)
813 + }
814 +
815 + #[instrument(skip_all)]
816 + pub async fn delete_pending_alert(pool: &SqlitePool, id: i64) -> Result<()> {
817 + sqlx::query("DELETE FROM pending_alerts WHERE id = ?")
818 + .bind(id)
819 + .execute(pool)
820 + .await?;
821 + Ok(())
822 + }
823 +
824 + /// Record a failed retry: bump the attempt count and push `next_retry_at` out.
825 + #[instrument(skip_all)]
826 + pub async fn bump_pending_alert(pool: &SqlitePool, id: i64, next_retry_at: &str) -> Result<()> {
827 + sqlx::query("UPDATE pending_alerts SET attempts = attempts + 1, next_retry_at = ? WHERE id = ?")
828 + .bind(next_retry_at)
829 + .bind(id)
830 + .execute(pool)
831 + .await?;
832 + Ok(())
833 + }
834 +
693 835 // --- TLS check queries ---
694 836
695 837 #[derive(Debug, sqlx::FromRow, serde::Serialize)]
M pom/tests/integration.rs +15 -13
@@ -382,7 +382,7 @@ async fn migration_fresh_db_reaches_latest_version() {
382 382 // A fresh in-memory DB should run all migrations and reach the latest version.
383 383 let pool = db::connect_in_memory().await.unwrap();
384 384 let version = db::get_schema_version(&pool).await.unwrap();
385 - assert_eq!(version, 9);
385 + assert_eq!(version, 10);
386 386
387 387 // Verify the schema_version table has entries for each migration
388 388 let rows = sqlx::query_as::<_, (i64, String)>(
@@ -391,7 +391,7 @@ async fn migration_fresh_db_reaches_latest_version() {
391 391 .fetch_all(&pool)
392 392 .await
393 393 .unwrap();
394 - assert_eq!(rows.len(), 9);
394 + assert_eq!(rows.len(), 10);
395 395 assert_eq!(rows[0].0, 1);
396 396 assert_eq!(rows[0].1, "initial schema");
397 397 assert_eq!(rows[1].0, 2);
@@ -410,6 +410,8 @@ async fn migration_fresh_db_reaches_latest_version() {
410 410 assert_eq!(rows[7].1, "add cors_checks table");
411 411 assert_eq!(rows[8].0, 9);
412 412 assert_eq!(rows[8].1, "add backup_checks table");
413 + assert_eq!(rows[9].0, 10);
414 + assert_eq!(rows[9].1, "add pending_alerts retry queue");
413 415
414 416 // Verify actual tables were created by inserting data
415 417 let snapshot = HealthSnapshot {
@@ -429,18 +431,18 @@ async fn migration_fresh_db_reaches_latest_version() {
429 431 async fn migration_already_current_is_idempotent() {
430 432 // Running migrations on an already-migrated DB should be a no-op.
431 433 let pool = db::connect_in_memory().await.unwrap();
432 - assert_eq!(db::get_schema_version(&pool).await.unwrap(), 9);
434 + assert_eq!(db::get_schema_version(&pool).await.unwrap(), 10);
433 435
434 436 // Run migrations again
435 437 db::run_migrations(&pool).await.unwrap();
436 - assert_eq!(db::get_schema_version(&pool).await.unwrap(), 9);
438 + assert_eq!(db::get_schema_version(&pool).await.unwrap(), 10);
437 439
438 - // schema_version should still have exactly nine entries (not duplicated)
440 + // schema_version should still have exactly ten entries (not duplicated)
439 441 let count = sqlx::query_as::<_, (i64,)>("SELECT COUNT(*) FROM schema_version")
440 442 .fetch_one(&pool)
441 443 .await
442 444 .unwrap();
443 - assert_eq!(count.0, 9);
445 + assert_eq!(count.0, 10);
444 446 }
445 447
446 448 #[tokio::test]
@@ -499,8 +501,8 @@ async fn migration_detects_pre_migration_database() {
499 501 // Now run migrations — should detect existing tables, stamp as v1, then run v2+v3+v4+v5+v6
500 502 db::run_migrations(&pool).await.unwrap();
501 503
502 - // Version should be 9 (stamped v1 + ran v2..v9)
503 - assert_eq!(db::get_schema_version(&pool).await.unwrap(), 9);
504 + // Version should be 10 (stamped v1 + ran v2..v10)
505 + assert_eq!(db::get_schema_version(&pool).await.unwrap(), 10);
504 506
505 507 // Description should indicate pre-existing
506 508 let row = sqlx::query_as::<_, (String,)>(
@@ -801,7 +803,7 @@ async fn tool_run_tests_no_test_config() {
801 803 async fn migration_v2_creates_alerts_table() {
802 804 let pool = db::connect_in_memory().await.unwrap();
803 805 let version = db::get_schema_version(&pool).await.unwrap();
804 - assert_eq!(version, 9);
806 + assert_eq!(version, 10);
805 807
806 808 // Verify alerts table exists by inserting
807 809 let id = db::insert_alert(&pool, "mnw", "health", Some("operational"), Some("error"), None)
@@ -869,7 +871,7 @@ async fn prune_removes_old_alerts() {
869 871 async fn migration_v3_creates_tls_checks_table() {
870 872 let pool = db::connect_in_memory().await.unwrap();
871 873 let version = db::get_schema_version(&pool).await.unwrap();
872 - assert_eq!(version, 9);
874 + assert_eq!(version, 10);
873 875
874 876 // Verify tls_checks table exists by inserting
875 877 let status = pom::types::TlsStatus {
@@ -1084,7 +1086,7 @@ cooldown_secs = 120
1084 1086 async fn migration_v4_creates_incidents_table() {
1085 1087 let pool = db::connect_in_memory().await.unwrap();
1086 1088 let version = db::get_schema_version(&pool).await.unwrap();
1087 - assert_eq!(version, 9);
1089 + assert_eq!(version, 10);
1088 1090
1089 1091 // Verify incidents table exists by inserting
1090 1092 let id = db::insert_incident(&pool, "mnw", "operational", "degraded")
@@ -1213,7 +1215,7 @@ async fn api_status_no_incidents_omits_fields() {
1213 1215 async fn migration_v5_creates_route_checks_table() {
1214 1216 let pool = db::connect_in_memory().await.unwrap();
1215 1217 let version = db::get_schema_version(&pool).await.unwrap();
1216 - assert_eq!(version, 9);
1218 + assert_eq!(version, 10);
1217 1219
1218 1220 // Verify route_checks table exists by inserting
1219 1221 let result = pom::checks::routes::RouteCheckResult {
@@ -2219,7 +2221,7 @@ async fn peer_uuid_mismatch_updates_db_identity() {
2219 2221 async fn migration_v6_creates_dns_and_whois_tables() {
2220 2222 let pool = db::connect_in_memory().await.unwrap();
2221 2223 let version = db::get_schema_version(&pool).await.unwrap();
2222 - assert_eq!(version, 9);
2224 + assert_eq!(version, 10);
2223 2225
2224 2226 // Verify dns_checks table exists
2225 2227 let dns_result = DnsCheckResult {