From 8f42497b8f3bc2a4f46dc5d9f1d9cd867e731298 Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Sat, 3 Oct 2026 01:16:42 +0100 Subject: [PATCH 1/4] Keep delayed alert recovery scoped to its resolved occurrence Capture the resolved alert before asynchronous callbacks and carry its start/end identity through monitor recovery. Cancel only that occurrence in grouping, cooldown and persistent firing rows, retaining newer incidents, operational links, destination receipts and failed attempts. Keep the explicit identifier-wide cancellation API and legacy callback IDs. Add snapshot, restart and connected queue-to-HTTP regression controls. Change-source: pulse-maintainer --- .../v6/internal/subsystems/agent-lifecycle.md | 11 ++ .../v6/internal/subsystems/alerts.md | 27 ++++ .../v6/internal/subsystems/monitoring.md | 21 +++ .../v6/internal/subsystems/notifications.md | 19 +++ internal/alerts/active_lifecycle.go | 6 +- internal/alerts/alerts_test.go | 14 +- internal/alerts/callbacks.go | 71 ++++++--- internal/alerts/canonical_lifecycle.go | 4 +- internal/alerts/canonical_metric.go | 2 +- internal/alerts/concurrency_test.go | 2 +- internal/alerts/config_runtime.go | 2 +- internal/alerts/guest.go | 2 +- internal/alerts/metric_runtime.go | 2 +- internal/alerts/operational_contract_test.go | 52 +++++++ internal/monitoring/monitor.go | 4 +- .../monitoring/monitor_alert_handling_test.go | 136 ++++++++++++++++++ internal/monitoring/monitor_alerts.go | 47 ++++-- .../monitor_notification_startup_test.go | 39 +++++ internal/notifications/notifications.go | 32 ++++- internal/notifications/queue.go | 31 +++- .../queue_occurrence_cancellation_test.go | 110 ++++++++++++++ 21 files changed, 570 insertions(+), 64 deletions(-) create mode 100644 internal/notifications/queue_occurrence_cancellation_test.go diff --git a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md index 4c18ec8692..4754a53d0d 100644 --- a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md +++ b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md @@ -8433,3 +8433,14 @@ update commands and credential authority unchanged. `internal/api/onboarding_handlers.go` changed only the human-readable pairing readiness messages so they name Settings > Pulse Mobile. No agent registration, lifecycle or install path changed. + +### Monitor recovery callback ownership + +The monitor's external alert wiring uses a captured resolved-occurrence +snapshot for notification cancellation. Re-enrolled or refiring resources keep +their new occurrence's queued notification even when an older recovery callback +arrives late under the same ID. Agent registration, deletion/tombstones, +authorisation and continuity evidence are unchanged. The connected callback +wiring control in `monitor_notification_startup_test.go` verifies the shared +monitor boundary against real alert evaluation and local HTTP delivery; it does +not claim host re-enrolment or installed-agent acceptance. diff --git a/docs/release-control/v6/internal/subsystems/alerts.md b/docs/release-control/v6/internal/subsystems/alerts.md index 4b002b9e4b..14bdd0b99f 100644 --- a/docs/release-control/v6/internal/subsystems/alerts.md +++ b/docs/release-control/v6/internal/subsystems/alerts.md @@ -3203,3 +3203,30 @@ upgrade prompt, and never load or save relay push settings. Licensed instances keep the minimum-severity control and the link to Pulse Mobile settings, with the 31 March 2027 retirement date in the copy. Webhook, email, Apprise and dead-man destinations are unaffected. + +### Occurrence-qualified asynchronous recovery delivery + +Every resolution funnel passes the actual `ResolvedAlert` to the callback bus. +`SetResolvedAlertCallback` captures a deep-cloned alert plus its resolution time +before asynchronous execution, rather than reconstructing an occurrence from a +reusable ID later. Legacy ID-only subscribers retain their public IDs; a panic +in one recovery consumer does not prevent the other consumers from running. + +The live monitor uses this snapshot for AI recovery and notification delivery. +`CancelResolvedAlert` removes only matching ID/start-time grouping entries and +cooldown state. `CancelByAlertOccurrence` applies the same match to persistent +pending, sending, failed and dead-letter firing rows, preserving newer +occurrences, unrelated grouped members and their operational links. Missing +start-time identity is a no-op, never a request for identifier-wide cancellation. +The existing explicit ID-wide cancellation API remains available. Delivery +gates, pending-only suppression counts, destination receipts, attempt history, +quiet hours and best-effort treatment of already-started sends are unchanged. + +`operational_contract_test.go` verifies snapshot isolation across a +blocked callback and changed ID-indexed history, legacy identity and panic +isolation. `queue_occurrence_cancellation_test.go` verifies nanosecond-distinct +recurrences, grouping/cooldown preservation, missing identity and restart +readback for all four cancellable row states without deleting failed audits or +new-occurrence destination receipts. The monitor's connected tests use real PBS +evaluation, normal callbacks, the persistent queue and a local HTTP destination; +these receipts are not native provider or installed-release acceptance. diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index 6265060bec..6f5caeb5a0 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -4634,3 +4634,24 @@ proof lives in `internal/truenas/provider_test.go` (`TestMaxTrueNASSystemTemperaturePrefersCoresOverSynthesisedAggregate`). This is source-level proof, not reporter acceptance or appliance confirmation (#2122). + +### Recovery callbacks retain the resolved occurrence + +`wireExternalAlertCallbacks` binds the occurrence-qualified recovery callback, +not the legacy ID-only callback. The monitor uses its immutable alert/start/end +snapshot for AI recovery and for exact notification cancellation, so a delayed +recovery cannot remove a newer firing admitted under the same reusable ID. +AI receives a separate clone. An ID-only websocket removal is omitted when the +manager already exposes that ID as active; the regular state stream continues +to own current alert truth. The legacy local ID adapter resolves a snapshot once +and fails closed when no resolved record exists; it is not the live delivery +binding. No browser payload, retention or notification routing contract changes. + +`monitor_alert_handling_test.go` and +`monitor_notification_startup_test.go` covers a delayed ID-adapter cancellation +with queue restart/readback, and a real PBS breach/recovery/new breach held at +the callback boundary. Both require the replacement firing's exact ID/start to +reach a loopback HTTP destination with one sent row and one cancelled old row. +No clock, reducer map or configured cooldown is altered. This establishes local +source-connected delivery, not the outstanding natural flapping cooldown check, +a native appliance result or containing stable availability. diff --git a/docs/release-control/v6/internal/subsystems/notifications.md b/docs/release-control/v6/internal/subsystems/notifications.md index 83ced5ecfe..5b5b782ab2 100644 --- a/docs/release-control/v6/internal/subsystems/notifications.md +++ b/docs/release-control/v6/internal/subsystems/notifications.md @@ -1054,3 +1054,22 @@ regression establish frontend control preservation, not provider delivery. Formatting-only follow-up retains this warning-level contract. The production-component browser matrix was rerun after formatting at desktop and narrow widths, including all/critical/warning save/reload, cancel, and warning webhook creation. + +### Resolution cancellation is occurrence-qualified + +The monitor's delayed recovery path calls `CancelResolvedAlert` with the +captured ID/start-time snapshot. It removes only that occurrence's pending +in-memory grouping members and cooldown record. Persistent +`CancelByAlertOccurrence` shares the existing per-ID delivery gates and +cancellation machinery, but matches both ID and start time: pending, sending, +failed and dead-letter firing rows are cancelled or rewritten without removing +a recurring incident or unrelated members/operational links in a mixed row. +Missing occurrence identity is a no-op. The explicit identifier-wide API, audit +history, destination receipts, retry budget, recovery rows and best-effort +in-flight semantics remain unchanged. + +`queue_occurrence_cancellation_test.go` verifies all four row states after a +fresh database reopen, nanosecond-distinct occurrences, operational-link and +failed-attempt preservation, missing-identity safety, grouping/cooldown retention +and the unchanged ID-wide API. Monitor connected controls additionally require +HTTP acceptance of the surviving occurrence through the autonomous queue. diff --git a/internal/alerts/active_lifecycle.go b/internal/alerts/active_lifecycle.go index 6b90e6ec6d..3771f98205 100644 --- a/internal/alerts/active_lifecycle.go +++ b/internal/alerts/active_lifecycle.go @@ -115,7 +115,7 @@ func (m *Manager) clearAlert(alertID string) { m.addRecentlyResolvedUnlocked(resolvedAlert) - m.safeCallResolvedAlertCallback(alert, publicID, false) + m.safeCallResolvedAlertCallback(resolvedAlert, publicID, false) log.Info(). Str("alertID", publicID). @@ -738,7 +738,7 @@ func (m *Manager) resolveDiscreteRecoveryNoLock(resourceID, specKey, alertID str resolvedAlert := m.newResolvedAlert(alert, time.Now(), nil) m.addRecentlyResolvedWithPrimaryLock(resolvedAlert) - m.safeCallResolvedAlertCallback(alert, alertID, true) + m.safeCallResolvedAlertCallback(resolvedAlert, alertID, true) log.Info(). Str(strings.ToLower(resourceKind), resourceName). @@ -790,7 +790,7 @@ func (m *Manager) clearAlertNoLock(alertID string) { m.addRecentlyResolvedWithPrimaryLock(resolvedAlert) - m.safeCallResolvedAlertCallback(alert, publicID, true) + m.safeCallResolvedAlertCallback(resolvedAlert, publicID, true) log.Info(). Str("alertID", publicID). diff --git a/internal/alerts/alerts_test.go b/internal/alerts/alerts_test.go index ccafc38ed5..68daa1b65c 100644 --- a/internal/alerts/alerts_test.go +++ b/internal/alerts/alerts_test.go @@ -7032,7 +7032,7 @@ func TestSafeCallResolvedAlertCallback(t *testing.T) { receivedID = alertID }) - m.safeCallResolvedAlertCallback(&Alert{ID: "test-alert-123"}, "test-alert-123", false) + m.safeCallResolvedAlertCallback(&ResolvedAlert{Alert: &Alert{ID: "test-alert-123"}}, "test-alert-123", false) if receivedID != "test-alert-123" { t.Errorf("expected alert ID 'test-alert-123', got %q", receivedID) @@ -7051,7 +7051,7 @@ func TestSafeCallResolvedAlertCallback(t *testing.T) { close(done) }) - m.safeCallResolvedAlertCallback(&Alert{ID: "async-alert"}, "async-alert", true) + m.safeCallResolvedAlertCallback(&ResolvedAlert{Alert: &Alert{ID: "async-alert"}}, "async-alert", true) select { case <-done: @@ -7078,7 +7078,7 @@ func TestSafeCallResolvedAlertCallback(t *testing.T) { CanonicalState: buildCanonicalStateID("pve1:node1:101", "powered-state:powered-off"), } - m.safeCallResolvedAlertCallback(alert, alert.CanonicalState, false) + m.safeCallResolvedAlertCallback(&ResolvedAlert{Alert: alert}, alert.CanonicalState, false) if receivedID != alert.CanonicalState { t.Errorf("expected canonical alert ID %q, got %q", alert.CanonicalState, receivedID) @@ -7091,8 +7091,8 @@ func TestSafeCallResolvedAlertCallback(t *testing.T) { // No callback set // Should not panic - m.safeCallResolvedAlertCallback(&Alert{ID: "test-alert"}, "test-alert", false) - m.safeCallResolvedAlertCallback(&Alert{ID: "test-alert"}, "test-alert", true) + m.safeCallResolvedAlertCallback(&ResolvedAlert{Alert: &Alert{ID: "test-alert"}}, "test-alert", false) + m.safeCallResolvedAlertCallback(&ResolvedAlert{Alert: &Alert{ID: "test-alert"}}, "test-alert", true) }) t.Run("recovers from panic in sync callback", func(t *testing.T) { @@ -7104,7 +7104,7 @@ func TestSafeCallResolvedAlertCallback(t *testing.T) { }) // Should not panic the caller - m.safeCallResolvedAlertCallback(&Alert{ID: "panic-test"}, "panic-test", false) + m.safeCallResolvedAlertCallback(&ResolvedAlert{Alert: &Alert{ID: "panic-test"}}, "panic-test", false) }) t.Run("recovers from panic in async callback", func(t *testing.T) { @@ -7117,7 +7117,7 @@ func TestSafeCallResolvedAlertCallback(t *testing.T) { panic("async panic") }) - m.safeCallResolvedAlertCallback(&Alert{ID: "async-panic"}, "async-panic", true) + m.safeCallResolvedAlertCallback(&ResolvedAlert{Alert: &Alert{ID: "async-panic"}}, "async-panic", true) select { case <-done: diff --git a/internal/alerts/callbacks.go b/internal/alerts/callbacks.go index e2378ecb0e..4a6a7a9d05 100644 --- a/internal/alerts/callbacks.go +++ b/internal/alerts/callbacks.go @@ -23,13 +23,14 @@ type LifecycleEvent struct { type callbackBus struct { mu sync.RWMutex - onAlert func(alert *Alert) - alertSubs map[int]func(alert *Alert) - onAlertForAI func(alert *Alert) - alertForAISubs map[int]func(alert *Alert) - onResolved func(alertID string) - resolvedSubs map[int]func(alertID string) - lifecycleSubs map[int]func(event LifecycleEvent) + onAlert func(alert *Alert) + alertSubs map[int]func(alert *Alert) + onAlertForAI func(alert *Alert) + alertForAISubs map[int]func(alert *Alert) + onResolved func(alertID string) + resolvedSubs map[int]func(alertID string) + onResolvedAlert func(resolved *ResolvedAlert) + lifecycleSubs map[int]func(event LifecycleEvent) onAcknowledged func(alert *Alert, user string) onUnacknowledged func(alert *Alert, user string) @@ -146,6 +147,18 @@ func (b *callbackBus) setResolvedCallback(cb func(alertID string)) { b.onResolved = cb } +func (b *callbackBus) setResolvedAlertCallback(cb func(resolved *ResolvedAlert)) { + b.mu.Lock() + defer b.mu.Unlock() + b.onResolvedAlert = cb +} + +func (b *callbackBus) resolvedAlertCallback() func(resolved *ResolvedAlert) { + b.mu.RLock() + defer b.mu.RUnlock() + return b.onResolvedAlert +} + func (b *callbackBus) subscribeResolvedCallback(cb func(alertID string)) func() { if cb == nil { return func() {} @@ -324,6 +337,14 @@ func (m *Manager) SetResolvedCallback(cb func(alertID string)) { m.callbacks.setResolvedCallback(cb) } +// SetResolvedAlertCallback installs the occurrence-qualified recovery consumer. +// The snapshot is captured before asynchronous dispatch; consumers must not +// look up a reusable alert ID to reconstruct the resolved occurrence later. +// Legacy ID-only callbacks remain independent and retain their public IDs. +func (m *Manager) SetResolvedAlertCallback(cb func(resolved *ResolvedAlert)) { + m.callbacks.setResolvedAlertCallback(cb) +} + // SubscribeResolvedCallback registers an additional resolved-alert callback // without replacing the legacy single callback slot. The returned function // removes the subscription when called. @@ -402,31 +423,43 @@ func (m *Manager) getEscalateCallback() func(alert *Alert, level int) { // safeCallResolvedAlertCallback invokes onResolved with panic recovery while // preserving canonical state as the internal identity and emitting the public // alert ID to external callbacks for compatibility. -func (m *Manager) safeCallResolvedAlertCallback(alert *Alert, fallbackID string, async bool) { +func (m *Manager) safeCallResolvedAlertCallback(resolved *ResolvedAlert, fallbackID string, async bool) { + var alert *Alert + if resolved != nil { + alert = resolved.Alert + } // Record the resolution before the callback guard: this funnel is the // resolve seam for every lifecycle path, with or without subscribers. m.recordAlertEvent(eventlog.TypeResolved, alert, fallbackID, "", "Alert resolved.", nil) callbacks := m.getResolvedCallbacks() - if len(callbacks) == 0 { + resolvedCallback := m.callbacks.resolvedAlertCallback() + if len(callbacks) == 0 && resolvedCallback == nil { return } publicID := exportedAlertID(alert, fallbackID) trackingKey := canonicalTrackingKeyForAlert(alert) + var snapshot *ResolvedAlert + if resolved != nil && alert != nil { + snapshot = &ResolvedAlert{Alert: cloneAlertForOutput(alert), ResolvedTime: resolved.ResolvedTime} + } callbackFunc := func() { - defer func() { - if r := recover(); r != nil { - log.Error(). - Interface("panic", r). - Str("alertID", publicID). - Str("trackingKey", trackingKey). - Msg("Panic in onResolved callback") - } - }() + callSafely := func(callback func()) { + defer func() { + if r := recover(); r != nil { + log.Error().Interface("panic", r).Str("alertID", publicID). + Str("trackingKey", trackingKey).Msg("Panic in onResolved callback") + } + }() + callback() + } for _, callback := range callbacks { - callback(publicID) + callSafely(func() { callback(publicID) }) + } + if resolvedCallback != nil && snapshot != nil { + callSafely(func() { resolvedCallback(snapshot) }) } } diff --git a/internal/alerts/canonical_lifecycle.go b/internal/alerts/canonical_lifecycle.go index d1c8e738b5..b6f8386273 100644 --- a/internal/alerts/canonical_lifecycle.go +++ b/internal/alerts/canonical_lifecycle.go @@ -603,7 +603,7 @@ func (m *Manager) evaluateCanonicalLifecycleAlert(params canonicalLifecycleAlert recoveryEvidenceRef, ) m.addRecentlyResolvedWithPrimaryLock(resolvedAlert) - m.safeCallResolvedAlertCallback(existing, storageKey, true) + m.safeCallResolvedAlertCallback(resolvedAlert, storageKey, true) return result, true } @@ -863,7 +863,7 @@ func (m *Manager) evaluateCanonicalStatefulAlert(params canonicalStatefulAlertPa recoveryEvidenceRef, ) m.addRecentlyResolvedWithPrimaryLock(resolvedAlert) - m.safeCallResolvedAlertCallback(existing, storageKey, true) + m.safeCallResolvedAlertCallback(resolvedAlert, storageKey, true) return result, true } } diff --git a/internal/alerts/canonical_metric.go b/internal/alerts/canonical_metric.go index 1cceeaa0de..fcfa75baa5 100644 --- a/internal/alerts/canonical_metric.go +++ b/internal/alerts/canonical_metric.go @@ -407,7 +407,7 @@ func (m *Manager) evaluateCanonicalMetricAlert(spec alertspecs.ResourceAlertSpec m.removeActiveAlertNoLock(storageKey) m.saveActiveAlertsAsync("canonical metric resolution") m.addRecentlyResolvedWithPrimaryLock(resolvedAlert) - m.safeCallResolvedAlertCallback(existingAlert, storageKey, true) + m.safeCallResolvedAlertCallback(resolvedAlert, storageKey, true) } } diff --git a/internal/alerts/concurrency_test.go b/internal/alerts/concurrency_test.go index 7f9cc69bda..7d88d2db33 100644 --- a/internal/alerts/concurrency_test.go +++ b/internal/alerts/concurrency_test.go @@ -96,7 +96,7 @@ func TestManagerCallbacksConcurrentSetAndUse(t *testing.T) { defer wg.Done() <-start for i := 0; i < iterations; i++ { - manager.safeCallResolvedAlertCallback(&Alert{ID: "concurrency-alert"}, "concurrency-alert", false) + manager.safeCallResolvedAlertCallback(&ResolvedAlert{Alert: &Alert{ID: "concurrency-alert"}}, "concurrency-alert", false) manager.safeCallAcknowledgedCallback(alert, "test-user") manager.safeCallUnacknowledgedCallback(alert, "test-user") manager.safeCallEscalateCallback(alert, 1) diff --git a/internal/alerts/config_runtime.go b/internal/alerts/config_runtime.go index 73b9fe2891..17d0ecf11e 100644 --- a/internal/alerts/config_runtime.go +++ b/internal/alerts/config_runtime.go @@ -608,7 +608,7 @@ func (m *Manager) reevaluateActiveAlertsLocked() { Str("alertID", alertID). Msg("Alert auto-resolved after configuration change") - m.safeCallResolvedAlertCallback(resolvedAlert.Alert, alertID, true) + m.safeCallResolvedAlertCallback(resolvedAlert, alertID, true) } } diff --git a/internal/alerts/guest.go b/internal/alerts/guest.go index 5eb502e55c..b185ea1d30 100644 --- a/internal/alerts/guest.go +++ b/internal/alerts/guest.go @@ -551,7 +551,7 @@ func (m *Manager) clearGuestPoweredOffAlert(guestID, name string) { // Send recovery notification (async to avoid deadlock because callback acquires m.mu.RLock // via ShouldSuppressResolvedNotification, and we currently hold m.mu.Lock) - m.safeCallResolvedAlertCallback(alert, effectiveAlertID(alert, alertID), true) + m.safeCallResolvedAlertCallback(resolvedAlert, effectiveAlertID(alert, alertID), true) // Log recovery log.Info(). diff --git a/internal/alerts/metric_runtime.go b/internal/alerts/metric_runtime.go index 6000fad5d8..6d1ad3cba2 100644 --- a/internal/alerts/metric_runtime.go +++ b/internal/alerts/metric_runtime.go @@ -612,7 +612,7 @@ func (m *Manager) checkMetric(resourceID, resourceName, node, instance, resource Float64("value", value). Bool("wasAcknowledged", existingAlert.Acknowledged). Msg("Alert resolved with hysteresis") - m.safeCallResolvedAlertCallback(existingAlert, alertID, true) + m.safeCallResolvedAlertCallback(resolvedAlert, alertID, true) } func sanitizeAlertKey(label string) string { diff --git a/internal/alerts/operational_contract_test.go b/internal/alerts/operational_contract_test.go index e0e5c68a31..444cb7aba3 100644 --- a/internal/alerts/operational_contract_test.go +++ b/internal/alerts/operational_contract_test.go @@ -692,3 +692,55 @@ func TestMemoryUnavailablePreservesActiveAlertsUntilTrustedRecovery(t *testing.T }) } } + +func TestResolvedOccurrenceCallbackCapturesBeforeDispatch(t *testing.T) { + m := newTestManager(t) + start := time.Now().Add(-time.Hour) + end := start.Add(time.Minute) + resolved := &ResolvedAlert{Alert: &Alert{ID: "reusable", StartTime: start, + Metadata: map[string]interface{}{"generation": "old"}}, ResolvedTime: end} + entered, release := make(chan struct{}), make(chan struct{}) + defer close(release) + m.SetResolvedCallback(func(id string) { + if id != "reusable" { + t.Errorf("legacy ID = %s", id) + } + close(entered) + <-release + }) + got := make(chan *ResolvedAlert, 1) + m.SetResolvedAlertCallback(func(r *ResolvedAlert) { got <- r }) + m.safeCallResolvedAlertCallback(resolved, resolved.Alert.ID, true) + select { + case <-entered: + case <-time.After(time.Second): + t.Fatal("legacy callback did not start") + } + // Both the source object and the ID-indexed history can change while the + // consumer waits. Neither is a source for the already-admitted snapshot. + resolved.Alert.StartTime = start.Add(time.Hour) + resolved.Alert.Metadata["generation"] = "new" + resolved.ResolvedTime = end.Add(time.Hour) + m.addRecentlyResolvedUnlocked(resolved) + // Unblock without closing twice in cleanup. + release <- struct{}{} + select { + case r := <-got: + if !r.Alert.StartTime.Equal(start) || !r.ResolvedTime.Equal(end) || r.Alert.Metadata["generation"] != "old" { + t.Fatalf("callback reconstructed the wrong occurrence: %+v", r) + } + case <-time.After(time.Second): + t.Fatal("no occurrence callback") + } +} + +func TestResolvedOccurrenceCallbackSurvivesLegacyPanic(t *testing.T) { + m := newTestManager(t) + m.SetResolvedCallback(func(string) { panic("legacy consumer failure") }) + called := false + m.SetResolvedAlertCallback(func(r *ResolvedAlert) { called = r.Alert.ID == "reusable" }) + m.safeCallResolvedAlertCallback(&ResolvedAlert{Alert: &Alert{ID: "reusable"}}, "reusable", false) + if !called { + t.Fatal("legacy panic prevented occurrence-qualified recovery") + } +} diff --git a/internal/monitoring/monitor.go b/internal/monitoring/monitor.go index e2cf45886b..4d97279cda 100644 --- a/internal/monitoring/monitor.go +++ b/internal/monitoring/monitor.go @@ -2248,8 +2248,8 @@ func (m *Monitor) wireExternalAlertCallbacks(wsHub *websocket.Hub) { callback(alert) } }) - m.alertManager.SetResolvedCallback(func(alertID string) { - m.handleAlertResolved(alertID) + m.alertManager.SetResolvedAlertCallback(func(resolved *alerts.ResolvedAlert) { + m.handleResolvedAlert(resolved) // Don't broadcast full state here - it causes a cascade with many guests. // The frontend gets updated alerts through the regular broadcast ticker. }) diff --git a/internal/monitoring/monitor_alert_handling_test.go b/internal/monitoring/monitor_alert_handling_test.go index c0b8b27947..312a2d89f0 100644 --- a/internal/monitoring/monitor_alert_handling_test.go +++ b/internal/monitoring/monitor_alert_handling_test.go @@ -1205,3 +1205,139 @@ func TestMonitorLifecycleRefireReopensRetainedOccurrence(t *testing.T) { }) } } + +type occurrenceHTTPReceipt struct { + Alerts []*alerts.Alert `json:"alerts"` +} + +func occurrenceNotifier(t *testing.T, dir string, endpoint string) *notifications.NotificationManager { + t.Helper() + n := notifications.NewNotificationManagerWithDeferredQueue("", dir) + t.Cleanup(n.Stop) + if err := n.UpdateAllowedPrivateCIDRs("127.0.0.1/32,::1/128"); err != nil { + t.Fatal(err) + } + n.AddWebhook(notifications.WebhookConfig{ID: "ops", Enabled: true, URL: endpoint, Service: "generic"}) + n.SetGroupingWindow(0) + n.SetNotifyOnResolve(false) + return n +} + +func occurrenceManager(t *testing.T) (*alerts.Manager, models.PBSInstance) { + t.Helper() + a := alerts.NewManagerWithDataDir(t.TempDir()) + t.Cleanup(a.Stop) + cfg := a.GetConfig() + cfg.Enabled, cfg.ActivationState, cfg.FlappingEnabled = true, alerts.ActivationActive, false + cfg.Schedule.QuietHours.Enabled = false + cfg.TimeThresholds["pbs"] = 0 + cfg.PBSDefaults.CPU = &alerts.HysteresisThreshold{Trigger: 80, Clear: 75} + a.UpdateConfig(cfg) + return a, models.PBSInstance{ID: "recurring-pbs", Name: "backup", Host: "pbs.invalid", Status: "online", ConnectionHealth: "healthy", CPU: 99} +} + +func occurrenceEndpoint(t *testing.T) (string, <-chan occurrenceHTTPReceipt) { + t.Helper() + receipts := make(chan occurrenceHTTPReceipt, 8) + s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + defer r.Body.Close() + var receipt occurrenceHTTPReceipt + if err := json.NewDecoder(r.Body).Decode(&receipt); err != nil { + t.Errorf("decode HTTP acceptance: %v", err) + w.WriteHeader(http.StatusBadRequest) + return + } + receipts <- receipt + w.WriteHeader(http.StatusOK) + })) + t.Cleanup(s.Close) + return s.URL, receipts +} + +func awaitOccurrenceRows(t *testing.T, n *notifications.NotificationManager, count int) []*notifications.QueuedNotification { + t.Helper() + deadline := time.Now().Add(3 * time.Second) + for { + rows, err := n.GetQueue().GetPending(10) + if err != nil { + t.Fatal(err) + } + if len(rows) == count { + return rows + } + if time.Now().After(deadline) { + t.Fatalf("pending rows = %d, want %d", len(rows), count) + } + time.Sleep(time.Millisecond) + } +} + +func assertOccurrenceHTTP(t *testing.T, n *notifications.NotificationManager, receipts <-chan occurrenceHTTPReceipt, current *alerts.Alert) { + t.Helper() + n.StartQueueProcessing() + select { + case got := <-receipts: + if len(got.Alerts) != 1 || got.Alerts[0].ID != current.ID || !got.Alerts[0].StartTime.Equal(current.StartTime) { + t.Fatalf("wrong HTTP occurrence: %+v", got) + } + t.Logf("HTTP 200 accepted current occurrence %s at %s", current.ID, current.StartTime.Format(time.RFC3339Nano)) + case <-time.After(12 * time.Second): + t.Fatal("current firing did not reach the local HTTP destination") + } + deadline := time.Now().Add(3 * time.Second) + for { + stats, err := n.GetQueueStats() + if err != nil { + t.Fatal(err) + } + if stats["sent"] == 1 && stats["cancelled"] == 1 && stats["pending"] == 0 && stats["sending"] == 0 { + t.Logf("persistent queue completion: %v", stats) + break + } + if time.Now().After(deadline) { + t.Fatalf("queue completion = %v", stats) + } + time.Sleep(time.Millisecond) + } + select { + case extra := <-receipts: + t.Fatalf("obsolete or duplicate delivery: %+v", extra) + case <-time.After(200 * time.Millisecond): + } +} + +// This existing-interface control also runs against the exact predecessor. +// It models a delayed ID callback after the next occurrence has entered the +// real persistent queue. It does not assert native timing or cause a refire. +func TestMonitorDelayedIDResolutionPreservesNewQueuedOccurrence(t *testing.T) { + endpoint, receipts := occurrenceEndpoint(t) + dir := t.TempDir() + n := occurrenceNotifier(t, dir, endpoint) + a, pbs := occurrenceManager(t) + a.CheckPBS(pbs) + active := a.GetActiveAlerts() + if len(active) != 1 { + t.Fatalf("PBS firing = %+v", active) + } + old := active[0].Clone() + pbs.CPU = 0 + a.CheckPBS(pbs) + if r := a.GetResolvedAlert(old.ID); r == nil || !r.Alert.StartTime.Equal(old.StartTime) { + t.Fatal("missing resolved PBS snapshot") + } + current := old.Clone() + current.StartTime = current.StartTime.Add(time.Nanosecond) + n.SendAlert(old) + n.SendAlert(current) + awaitOccurrenceRows(t, n, 2) + m := &Monitor{alertManager: a, notificationMgr: n} + m.handleAlertResolved(old.ID) + rows, err := n.GetQueue().GetPending(10) + if err != nil || len(rows) != 1 || len(rows[0].Alerts) != 1 || !rows[0].Alerts[0].StartTime.Equal(current.StartTime) { + t.Fatalf("delayed recovery cancelled the replacement occurrence: rows=%+v err=%v", rows, err) + } + // Resume the autonomous processor only after reopening the same disk state. + n.Stop() + n = occurrenceNotifier(t, dir, endpoint) + assertOccurrenceHTTP(t, n, receipts, current) +} diff --git a/internal/monitoring/monitor_alerts.go b/internal/monitoring/monitor_alerts.go index bb9a7af975..b84a4e2489 100644 --- a/internal/monitoring/monitor_alerts.go +++ b/internal/monitoring/monitor_alerts.go @@ -139,7 +139,9 @@ func (m *Monitor) SetAlertResolvedAICallback(callback func(*alerts.Alert)) { if m.alertManager == nil { return } + m.mu.Lock() m.alertResolvedAICallback = callback + m.mu.Unlock() log.Info().Msg("alert-resolved AI callback registered") } @@ -220,41 +222,56 @@ func (m *Monitor) handleAlertFired(alert *alerts.Alert) { } func (m *Monitor) handleAlertResolved(alertID string) { - var resolvedAlert *alerts.ResolvedAlert + // Compatibility for ID-only callers. The live monitor is wired to the + // occurrence snapshot callback, not a lookup after asynchronous dispatch. + if m.alertManager == nil { + return + } + m.handleResolvedAlert(m.alertManager.GetResolvedAlert(alertID)) +} + +func (m *Monitor) handleResolvedAlert(resolvedAlert *alerts.ResolvedAlert) { + if resolvedAlert == nil || resolvedAlert.Alert == nil { + return + } + alertID := resolvedAlert.Alert.ID if m.wsHub != nil { - m.wsHub.BroadcastAlertResolvedToTenant(m.GetOrgID(), alertID) + // The legacy websocket message has only an ID. Do not remove a current + // firing occurrence that is already visible under that reusable ID. + active := false + if m.alertManager != nil { + _, active = m.alertManager.DiagnoseAlertDelivery(alertID) + } + if !active { + m.wsHub.BroadcastAlertResolvedToTenant(m.GetOrgID(), alertID) + } } // Always trigger AI callback, regardless of notification suppression. - if m.alertResolvedAICallback != nil { - if resolvedAlert == nil { - resolvedAlert = m.alertManager.GetResolvedAlert(alertID) - } - if resolvedAlert != nil && resolvedAlert.Alert != nil { - go m.alertResolvedAICallback(resolvedAlert.Alert) - } + m.mu.RLock() + aiCallback := m.alertResolvedAICallback + m.mu.RUnlock() + if aiCallback != nil { + go aiCallback(resolvedAlert.Alert.Clone()) } // Handle notifications — recovery notifications respect quiet hours. // If the original alert would have been suppressed during quiet hours, // the recovery notification is also suppressed to avoid noise. if m.notificationMgr != nil { - firingNeverDelivered := m.notificationMgr.CancelAlert(alertID) + firingNeverDelivered := m.notificationMgr.CancelResolvedAlert(resolvedAlert.Alert) if m.notificationMgr.GetNotifyOnResolve() { - if resolvedAlert == nil { - resolvedAlert = m.alertManager.GetResolvedAlert(alertID) - } if resolvedAlert != nil && resolvedAlert.Alert != nil { if firingNeverDelivered { // The firing notification was still in the grouping window // or waiting in the queue (e.g. quiet-hours replay) when the - // alert resolved, and CancelAlert just cancelled it. A + // alert resolved, and exact cancellation just cancelled it. A // recovery for an alert the user never saw fire is noise. log.Info(). Str("alertID", alertID). Msg("Resolved notification suppressed because the firing notification was cancelled before delivery") - } else if m.alertManager.ShouldSuppressResolvedNotification(resolvedAlert.Alert) { + } else if m.alertManager != nil && m.alertManager.ShouldSuppressResolvedNotification(resolvedAlert.Alert) { log.Info(). Str("alertID", alertID). Msg("Resolved notification suppressed during quiet hours") diff --git a/internal/monitoring/monitor_notification_startup_test.go b/internal/monitoring/monitor_notification_startup_test.go index 34897b605a..0dc557f8a4 100644 --- a/internal/monitoring/monitor_notification_startup_test.go +++ b/internal/monitoring/monitor_notification_startup_test.go @@ -5,6 +5,7 @@ import ( "net/http" "net/http/httptest" "reflect" + "sync" "sync/atomic" "testing" "time" @@ -179,3 +180,41 @@ func TestNewRestoresSavedNotificationChoices(t *testing.T) { }) } } + +// Real PBS breach -> recovery -> new breach, with one callback deliberately +// held at its asynchronous consumer boundary. No clocks or manager maps change. +func TestMonitorResolvedSnapshotKeepsRefiringPBSDelivery(t *testing.T) { + endpoint, receipts := occurrenceEndpoint(t) + n := occurrenceNotifier(t, t.TempDir(), endpoint) + a, pbs := occurrenceManager(t) + m := &Monitor{alertManager: a, notificationMgr: n} + m.wireExternalAlertCallbacks(nil) + entered, release := make(chan struct{}), make(chan struct{}) + var releaseOnce sync.Once + t.Cleanup(func() { releaseOnce.Do(func() { close(release) }) }) + unsubscribe := a.SubscribeResolvedCallback(func(string) { close(entered); <-release }) + t.Cleanup(unsubscribe) + a.CheckPBS(pbs) + first := awaitOccurrenceRows(t, n, 1)[0].Alerts[0] + pbs.CPU = 0 + a.CheckPBS(pbs) + select { + case <-entered: + case <-time.After(3 * time.Second): + t.Fatal("resolution callback did not start") + } + pbs.CPU = 95 + a.CheckPBS(pbs) + awaitOccurrenceRows(t, n, 2) + active := a.GetActiveAlerts() + if len(active) != 1 || active[0].ID != first.ID || active[0].StartTime.Equal(first.StartTime) { + t.Fatalf("not a new recurring PBS occurrence: %+v", active) + } + current := active[0].Clone() + releaseOnce.Do(func() { close(release) }) + remaining := awaitOccurrenceRows(t, n, 1) + if !remaining[0].Alerts[0].StartTime.Equal(current.StartTime) { + t.Fatal("delayed resolution kept the wrong occurrence") + } + assertOccurrenceHTTP(t, n, receipts, current) +} diff --git a/internal/notifications/notifications.go b/internal/notifications/notifications.go index 17189a5043..7b93dbf590 100644 --- a/internal/notifications/notifications.go +++ b/internal/notifications/notifications.go @@ -1580,9 +1580,24 @@ func (n *NotificationManager) dispatchResolvedAlerts(alertList []*alerts.Alert) // the grouping window or waiting in the queue), so sending a recovery // notification would reference an alert that was never announced. func (n *NotificationManager) CancelAlert(alertID string) bool { + return n.cancelAlert(alertID, nil) +} + +// CancelResolvedAlert cancels firing work for one resolved occurrence, without +// deleting a newer occurrence's grouping entries or delivery cooldown. +// ID-wide cancellation remains available for explicit administrative callers. +func (n *NotificationManager) CancelResolvedAlert(alert *alerts.Alert) bool { + if alert == nil || alert.ID == "" || alert.StartTime.IsZero() { + return false + } + return n.cancelAlert(alert.ID, &alert.StartTime) +} + +func (n *NotificationManager) cancelAlert(alertID string, start *time.Time) bool { n.mu.Lock() queue := n.queue - _, firingDelivered := n.lastNotified[alertID] + record, firingDelivered := n.lastNotified[alertID] + firingDelivered = firingDelivered && (start == nil || record.alertStart.Equal(*start)) removed := 0 if len(n.pendingAlerts) > 0 { @@ -1591,7 +1606,7 @@ func (n *NotificationManager) CancelAlert(alertID string) bool { if pending == nil { continue } - if pending.ID == alertID { + if pending.ID == alertID && (start == nil || pending.StartTime.Equal(*start)) { removed++ continue } @@ -1613,14 +1628,21 @@ func (n *NotificationManager) CancelAlert(alertID string) bool { } // Clean up cooldown record for resolved alert - delete(n.lastNotified, alertID) + if start == nil || record.alertStart.Equal(*start) { + delete(n.lastNotified, alertID) + } + remaining := len(n.pendingAlerts) n.mu.Unlock() // Cancel any queued notifications containing this alert cancelledPending := 0 if queue != nil { var err error - cancelledPending, err = queue.CancelByAlertIdentifiers([]string{alertID}) + if start == nil { + cancelledPending, err = queue.CancelByAlertIdentifiers([]string{alertID}) + } else { + cancelledPending, err = queue.CancelByAlertOccurrence(alertID, *start) + } if err != nil { log.Error().Err(err).Str("alertID", alertID).Msg("failed to cancel queued notifications") } @@ -1630,7 +1652,7 @@ func (n *NotificationManager) CancelAlert(alertID string) bool { log.Debug(). Str("alertID", alertID). - Int("remaining", len(n.pendingAlerts)). + Int("remaining", remaining). Bool("firingNeverDelivered", firingNeverDelivered). Msg("removed resolved alert from pending notifications and cooldown map") diff --git a/internal/notifications/queue.go b/internal/notifications/queue.go index 47c28fda14..9835972871 100644 --- a/internal/notifications/queue.go +++ b/internal/notifications/queue.go @@ -2150,6 +2150,30 @@ func calculateBackoff(attempt int) time.Duration { // these do not contribute to the pending-only return count. func (nq *NotificationQueue) CancelByAlertIdentifiers(alertIdentifiers []string) (int, error) { alertIdentifiers = normalizeAlertIdentifiers(alertIdentifiers) + identifiers := make(map[string]struct{}, len(alertIdentifiers)) + for _, id := range alertIdentifiers { + identifiers[id] = struct{}{} + } + return nq.cancelAlertNotifications(alertIdentifiers, func(alert *alerts.Alert) bool { + _, matched := identifiers[alert.ID] + return matched + }) +} + +// CancelByAlertOccurrence retires only the resolved occurrence's firing work. +// Missing occurrence identity must never widen into ID-wide cancellation. +// The same delivery gates, terminal-row handling and audit retention apply. +func (nq *NotificationQueue) CancelByAlertOccurrence(alertID string, start time.Time) (int, error) { + alertID = strings.TrimSpace(alertID) + if alertID == "" || start.IsZero() { + return 0, nil + } + return nq.cancelAlertNotifications([]string{alertID}, func(alert *alerts.Alert) bool { + return alert.ID == alertID && alert.StartTime.Equal(start) + }) +} + +func (nq *NotificationQueue) cancelAlertNotifications(alertIdentifiers []string, matches func(*alerts.Alert) bool) (int, error) { if len(alertIdentifiers) == 0 { return 0, nil } @@ -2175,11 +2199,6 @@ func (nq *NotificationQueue) CancelByAlertIdentifiers(alertIdentifiers []string) return 0, fmt.Errorf("failed to query notifications for cancellation: %w", err) } - alertIdentifierSet := make(map[string]struct{}) - for _, id := range alertIdentifiers { - alertIdentifierSet[id] = struct{}{} - } - type queuedAlertCancellation struct { notificationID string remaining []*alerts.Alert @@ -2244,7 +2263,7 @@ func (nq *NotificationQueue) CancelByAlertIdentifiers(alertIdentifiers []string) remainingAlerts = append(remainingAlerts, alert) continue } - if _, exists := alertIdentifierSet[alert.ID]; exists { + if matches(alert) { matchedAlertCount++ if alert.OperationalRecord != nil && alert.LatestTransition != nil { removedLinkKeys[alert.OperationalRecord.ID+"\x00"+alert.LatestTransition.ID] = struct{}{} diff --git a/internal/notifications/queue_occurrence_cancellation_test.go b/internal/notifications/queue_occurrence_cancellation_test.go new file mode 100644 index 0000000000..713dbe0e38 --- /dev/null +++ b/internal/notifications/queue_occurrence_cancellation_test.go @@ -0,0 +1,110 @@ +package notifications + +import ( + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/alerts" +) + +func TestQueueOccurrenceCancellationPreservesRecurrenceAcrossRestart(t *testing.T) { + for _, status := range []NotificationQueueStatus{QueueStatusPending, QueueStatusSending, QueueStatusFailed, QueueStatusDLQ} { + t.Run(string(status), func(t *testing.T) { + dir := t.TempDir() + q, err := NewNotificationQueue(dir) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = q.Stop() }) + start := time.Now().UTC().Add(-time.Hour) + old := quietReplayAlert("reusable", alerts.AlertLevelWarning, start) + current := quietReplayAlert("reusable", alerts.AlertLevelCritical, start.Add(time.Nanosecond)) + current.OperationalRecord.ID += "-new" + current.LatestTransition.ID += "-new" + peer := quietReplayAlert("peer", alerts.AlertLevelWarning, start) + for _, row := range []*QueuedNotification{ + {ID: "old", Type: "webhook", Alerts: []*alerts.Alert{old}}, + {ID: "new", Type: "webhook", Alerts: []*alerts.Alert{current}}, + {ID: "mixed", Type: "webhook", Alerts: []*alerts.Alert{old, current, peer}}, + {ID: "recovery", Type: "webhook_resolved", Alerts: []*alerts.Alert{old}}, + {ID: "unknown", Type: "webhook", Alerts: []*alerts.Alert{{ID: old.ID}}}, + } { + row.Config = []byte("{}") + row.DestinationID = "webhook:ops" + row.MaxAttempts = 3 + if err := q.Enqueue(row); err != nil { + t.Fatal(err) + } + if err := q.UpdateStatus(row.ID, status, "retained-attempt"); err != nil { + t.Fatal(err) + } + if err := q.RecordAudit(row, false, "retained-attempt"); err != nil { + t.Fatal(err) + } + } + if err := q.RecordDeliveryReceipt(current.ID, current.StartTime, "webhook:ops", time.Now()); err != nil { + t.Fatal(err) + } + for _, identity := range []struct { + id string + start time.Time + }{{"", start}, {old.ID, time.Time{}}} { + if count, err := q.CancelByAlertOccurrence(identity.id, identity.start); err != nil || count != 0 { + t.Fatalf("missing identity widened cancellation: %d, %v", count, err) + } + } + count, err := q.CancelByAlertOccurrence(old.ID, old.StartTime) + wantCount := 0 + if status == QueueStatusPending { + wantCount = 2 + } + if err != nil || count != wantCount { + t.Fatalf("cancel count = %d, %v, want %d", count, err, wantCount) + } + if err := q.Stop(); err != nil { + t.Fatal(err) + } + q, err = NewNotificationQueue(dir) + if err != nil { + t.Fatal(err) + } + oldRow, mixed := loadQuietReplay(t, q, "old"), loadQuietReplay(t, q, "mixed") + if oldRow.Status != QueueStatusCancelled || mixed.Status != status || len(mixed.Alerts) != 2 || + !mixed.Alerts[0].StartTime.Equal(current.StartTime) || mixed.Alerts[1].ID != peer.ID || len(mixed.Links) != 2 || + mixed.Links[0].OperationalRecordID != current.OperationalRecord.ID || mixed.Links[1].OperationalRecordID != peer.OperationalRecord.ID { + t.Fatalf("wrong retained rows/links: old=%+v mixed=%+v", oldRow, mixed) + } + for _, id := range []string{"new", "recovery", "unknown"} { + if row := loadQuietReplay(t, q, id); row.Status != status || len(row.Alerts) != 1 { + t.Fatalf("unrelated row %s changed: %+v", id, row) + } + } + if has, err := q.HasDeliveryReceipt(current.ID, current.StartTime, "webhook:ops"); err != nil || !has { + t.Fatalf("current occurrence lost its receipt: %t, %v", has, err) + } + var audits int + if err := q.db.QueryRow(`SELECT count(*) FROM notification_audit WHERE success = 0`).Scan(&audits); err != nil || audits != 5 { + t.Fatalf("retained failed attempts = %d, %v", audits, err) + } + }) + } +} + +func TestCancelResolvedAlertPreservesNewGroupingAndCooldown(t *testing.T) { + start := time.Now().Add(-time.Minute) + old := &alerts.Alert{ID: "reusable", StartTime: start} + current := &alerts.Alert{ID: old.ID, StartTime: start.Add(time.Nanosecond)} + record := notificationRecord{alertStart: current.StartTime, lastSent: time.Now(), level: alerts.AlertLevelCritical} + n := &NotificationManager{pendingAlerts: []*alerts.Alert{old, current}, lastNotified: map[string]notificationRecord{old.ID: record}} + if n.CancelResolvedAlert(&alerts.Alert{ID: old.ID}) || len(n.pendingAlerts) != 2 { + t.Fatal("missing occurrence cancelled pending work") + } + if !n.CancelResolvedAlert(old) || len(n.pendingAlerts) != 1 || n.pendingAlerts[0] != current || n.lastNotified[current.ID] != record { + t.Fatal("delayed recovery removed current grouping or cooldown state") + } + // Keep the explicit, identifier-wide cancellation contract for its callers. + n.CancelAlert(current.ID) + if len(n.pendingAlerts) != 0 || len(n.lastNotified) != 0 { + t.Fatal("ID-wide cancellation regressed") + } +} From 2c57a54ec4cba0ffd108465b74a582bf61b35bec Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Sat, 3 Oct 2026 01:20:39 +0100 Subject: [PATCH 2/4] Preserve sending-row crash recovery in the occurrence fixture Assert surviving row status before restart, then expect the existing sending-to-pending crash recovery after reopening the queue. The initial source-dtamnmrn fixture failure is retained; runtime source and delivery policy are unchanged. Change-source: pulse-maintainer --- .../queue_occurrence_cancellation_test.go | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/internal/notifications/queue_occurrence_cancellation_test.go b/internal/notifications/queue_occurrence_cancellation_test.go index 713dbe0e38..e2e5c7495b 100644 --- a/internal/notifications/queue_occurrence_cancellation_test.go +++ b/internal/notifications/queue_occurrence_cancellation_test.go @@ -61,6 +61,9 @@ func TestQueueOccurrenceCancellationPreservesRecurrenceAcrossRestart(t *testing. if err != nil || count != wantCount { t.Fatalf("cancel count = %d, %v, want %d", count, err, wantCount) } + if row := loadQuietReplay(t, q, "mixed"); row.Status != status || len(row.Alerts) != 2 { + t.Fatalf("cancellation changed surviving row status: %+v", row) + } if err := q.Stop(); err != nil { t.Fatal(err) } @@ -68,14 +71,20 @@ func TestQueueOccurrenceCancellationPreservesRecurrenceAcrossRestart(t *testing. if err != nil { t.Fatal(err) } + reopenStatus := status + if status == QueueStatusSending { + // Existing crash recovery requeues interrupted sends. Exact + // cancellation must preserve that behaviour for survivors. + reopenStatus = QueueStatusPending + } oldRow, mixed := loadQuietReplay(t, q, "old"), loadQuietReplay(t, q, "mixed") - if oldRow.Status != QueueStatusCancelled || mixed.Status != status || len(mixed.Alerts) != 2 || + if oldRow.Status != QueueStatusCancelled || mixed.Status != reopenStatus || len(mixed.Alerts) != 2 || !mixed.Alerts[0].StartTime.Equal(current.StartTime) || mixed.Alerts[1].ID != peer.ID || len(mixed.Links) != 2 || mixed.Links[0].OperationalRecordID != current.OperationalRecord.ID || mixed.Links[1].OperationalRecordID != peer.OperationalRecord.ID { t.Fatalf("wrong retained rows/links: old=%+v mixed=%+v", oldRow, mixed) } for _, id := range []string{"new", "recovery", "unknown"} { - if row := loadQuietReplay(t, q, id); row.Status != status || len(row.Alerts) != 1 { + if row := loadQuietReplay(t, q, id); row.Status != reopenStatus || len(row.Alerts) != 1 { t.Fatalf("unrelated row %s changed: %+v", id, row) } } From 1c9440546afbb1d17432b10814d078ddb7a07aee Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Sat, 3 Oct 2026 01:29:51 +0100 Subject: [PATCH 3/4] Use destination receipts for delayed partial-delivery recovery A cancelled retry and a lost RAM cooldown marker do not mean that no destination received the old firing. Keep quiet-hours and recovery controls, then let persistent occurrence/destination receipts select the actual recipients. Add an HTTP 200/503, restart and recurrence control so the newer firing survives and only the previously informed destination receives the old recovery. Change-source: pulse-maintainer --- .../v6/internal/subsystems/alerts.md | 6 + .../v6/internal/subsystems/monitoring.md | 9 ++ .../v6/internal/subsystems/notifications.md | 5 + .../monitoring/monitor_alert_handling_test.go | 116 ++++++++++++++++++ internal/monitoring/monitor_alerts.go | 25 ++-- 5 files changed, 144 insertions(+), 17 deletions(-) diff --git a/docs/release-control/v6/internal/subsystems/alerts.md b/docs/release-control/v6/internal/subsystems/alerts.md index 14bdd0b99f..fefabf9828 100644 --- a/docs/release-control/v6/internal/subsystems/alerts.md +++ b/docs/release-control/v6/internal/subsystems/alerts.md @@ -3230,3 +3230,9 @@ readback for all four cancellable row states without deleting failed audits or new-occurrence destination receipts. The monitor's connected tests use real PBS evaluation, normal callbacks, the persistent queue and a local HTTP destination; these receipts are not native provider or installed-release acceptance. + +Pending cancellation is not proof that every destination missed the firing. +The monitor delegates recovery eligibility to persisted occurrence/destination +receipts even when an old retry remains or restart has lost RAM cooldown state. +The partial-destination monitor control establishes a real local HTTP 200/503 +split and preserves the newer firing and the old recipient's recovery. diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index 6f5caeb5a0..9660e69b59 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -4655,3 +4655,12 @@ reach a loopback HTTP destination with one sent row and one cancelled old row. No clock, reducer map or configured cooldown is altered. This establishes local source-connected delivery, not the outstanding natural flapping cooldown check, a native appliance result or containing stable availability. + +Recovery eligibility is destination-receipt-owned, not the cancellation helper's +pending-work count or the RAM cooldown marker. After partial delivery/restart, +a cancelled retry at one destination cannot suppress a recovery to another +that actually accepted the same occurrence. The connected +`TestMonitorDelayedPartialResolutionKeepsDestinationRecovery` retains a real +HTTP 200/503 split, reopens the persistent queue, and requires the old accepted +destination's recovery plus the new firing, with no recovery to the unannounced +destination. Quiet hours and disabled recovery controls still apply. diff --git a/docs/release-control/v6/internal/subsystems/notifications.md b/docs/release-control/v6/internal/subsystems/notifications.md index 5b5b782ab2..068c67b93a 100644 --- a/docs/release-control/v6/internal/subsystems/notifications.md +++ b/docs/release-control/v6/internal/subsystems/notifications.md @@ -1073,3 +1073,8 @@ fresh database reopen, nanosecond-distinct occurrences, operational-link and failed-attempt preservation, missing-identity safety, grouping/cooldown retention and the unchanged ID-wide API. Monitor connected controls additionally require HTTP acceptance of the surviving occurrence through the autonomous queue. + +The live monitor does not interpret pending-only cancellation counts as +recipient acceptance. Per-occurrence/per-destination receipts remain the sole +recovery admission proof, including partial delivery and lost RAM markers after +restart; an unannounced destination receives no recovery. diff --git a/internal/monitoring/monitor_alert_handling_test.go b/internal/monitoring/monitor_alert_handling_test.go index 312a2d89f0..33c98887b5 100644 --- a/internal/monitoring/monitor_alert_handling_test.go +++ b/internal/monitoring/monitor_alert_handling_test.go @@ -1207,6 +1207,7 @@ func TestMonitorLifecycleRefireReopensRetainedOccurrence(t *testing.T) { } type occurrenceHTTPReceipt struct { + Event string `json:"event"` Alerts []*alerts.Alert `json:"alerts"` } @@ -1341,3 +1342,118 @@ func TestMonitorDelayedIDResolutionPreservesNewQueuedOccurrence(t *testing.T) { n = occurrenceNotifier(t, dir, endpoint) assertOccurrenceHTTP(t, n, receipts, current) } + +// One destination accepted the old firing and another still has a retry. After +// restart, that pending work (and the lost RAM cooldown marker) is not evidence +// that no recipient saw it. Recovery must use exact persisted receipts. +func TestMonitorDelayedPartialResolutionKeepsDestinationRecovery(t *testing.T) { + endpoint, receipts := occurrenceEndpoint(t) + var failedFirings, unwantedRecoveries atomic.Int32 + broken := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + defer r.Body.Close() + var receipt occurrenceHTTPReceipt + if err := json.NewDecoder(r.Body).Decode(&receipt); err != nil { + t.Errorf("decode failing destination: %v", err) + } + if receipt.Event == "resolved" { + unwantedRecoveries.Add(1) + } else { + failedFirings.Add(1) + } + w.WriteHeader(http.StatusServiceUnavailable) + })) + t.Cleanup(broken.Close) + dir := t.TempDir() + open := func() *notifications.NotificationManager { + n := occurrenceNotifier(t, dir, endpoint) + n.SetNotifyOnResolve(true) + n.AddWebhook(notifications.WebhookConfig{ID: "unannounced", Enabled: true, URL: broken.URL, Service: "generic"}) + return n + } + n := open() + a, pbs := occurrenceManager(t) + a.CheckPBS(pbs) + old := a.GetActiveAlerts()[0].Clone() + n.SendAlert(old) + n.StartQueueProcessing() + select { + case got := <-receipts: + if got.Event != "" || len(got.Alerts) != 1 || !got.Alerts[0].StartTime.Equal(old.StartTime) { + t.Fatalf("wrong initial firing: %+v", got) + } + case <-time.After(12 * time.Second): + t.Fatal("initial firing did not reach accepting destination") + } + deadline := time.Now().Add(3 * time.Second) + for { + stats, err := n.GetQueueStats() + if err != nil { + t.Fatal(err) + } + if stats["sent"] == 1 && stats["pending"] == 1 && failedFirings.Load() > 0 { + break + } + if time.Now().After(deadline) { + t.Fatalf("partial delivery not retained: %v", stats) + } + time.Sleep(time.Millisecond) + } + n.Stop() + pbs.CPU = 0 + a.CheckPBS(pbs) + n = open() // deferred processor; prior destination receipts are now on disk only + pbs.CPU = 95 + a.CheckPBS(pbs) + active := a.GetActiveAlerts() + if len(active) != 1 || active[0].ID != old.ID || active[0].StartTime.Equal(old.StartTime) { + t.Fatalf("not a new occurrence: %+v", active) + } + current := active[0].Clone() + n.SendAlert(current) + m := &Monitor{alertManager: a, notificationMgr: n} + m.handleAlertResolved(old.ID) + n.StartQueueProcessing() + seenFiring, seenRecovery := false, false + for i := 0; i < 2; i++ { + select { + case got := <-receipts: + if len(got.Alerts) != 1 || got.Alerts[0].ID != old.ID { + t.Fatalf("wrong delivery: %+v", got) + } + switch got.Event { + case "resolved": + if seenRecovery || !got.Alerts[0].StartTime.Equal(old.StartTime) { + t.Fatalf("wrong recovery occurrence: %+v", got) + } + seenRecovery = true + case "": + if seenFiring || !got.Alerts[0].StartTime.Equal(current.StartTime) { + t.Fatalf("wrong firing occurrence: %+v", got) + } + seenFiring = true + default: + t.Fatalf("unknown event: %+v", got) + } + case <-time.After(12 * time.Second): + t.Fatal("lost current firing or old accepted destination's recovery after restart") + } + } + deadline = time.Now().Add(3 * time.Second) + for { + stats, err := n.GetQueueStats() + if err != nil { + t.Fatal(err) + } + if stats["sent"] == 3 && stats["cancelled"] == 1 && stats["pending"]+stats["sending"]+stats["failed"]+stats["dlq"] == 1 { + t.Logf("partial destination recovery retained exact old receipt and new firing: %v", stats) + break + } + if time.Now().After(deadline) { + t.Fatalf("partial recovery queue completion: %v", stats) + } + time.Sleep(time.Millisecond) + } + if unwantedRecoveries.Load() != 0 { + t.Fatal("unannounced destination received a recovery") + } +} diff --git a/internal/monitoring/monitor_alerts.go b/internal/monitoring/monitor_alerts.go index b84a4e2489..826f9971b7 100644 --- a/internal/monitoring/monitor_alerts.go +++ b/internal/monitoring/monitor_alerts.go @@ -260,24 +260,15 @@ func (m *Monitor) handleResolvedAlert(resolvedAlert *alerts.ResolvedAlert) { // If the original alert would have been suppressed during quiet hours, // the recovery notification is also suppressed to avoid noise. if m.notificationMgr != nil { - firingNeverDelivered := m.notificationMgr.CancelResolvedAlert(resolvedAlert.Alert) + m.notificationMgr.CancelResolvedAlert(resolvedAlert.Alert) if m.notificationMgr.GetNotifyOnResolve() { - if resolvedAlert != nil && resolvedAlert.Alert != nil { - if firingNeverDelivered { - // The firing notification was still in the grouping window - // or waiting in the queue (e.g. quiet-hours replay) when the - // alert resolved, and exact cancellation just cancelled it. A - // recovery for an alert the user never saw fire is noise. - log.Info(). - Str("alertID", alertID). - Msg("Resolved notification suppressed because the firing notification was cancelled before delivery") - } else if m.alertManager != nil && m.alertManager.ShouldSuppressResolvedNotification(resolvedAlert.Alert) { - log.Info(). - Str("alertID", alertID). - Msg("Resolved notification suppressed during quiet hours") - } else { - go m.notificationMgr.SendResolvedAlert(resolvedAlert) - } + if m.alertManager != nil && m.alertManager.ShouldSuppressResolvedNotification(resolvedAlert.Alert) { + log.Info().Str("alertID", alertID).Msg("Resolved notification suppressed during quiet hours") + } else { + // Only occurrence/destination receipts establish who saw the + // firing. A pending retry or lost in-memory cooldown marker + // after restart must not suppress recovery for other recipients. + go m.notificationMgr.SendResolvedAlert(resolvedAlert) } } else { log.Info(). From d0d621a7c2d65c46dcf876986c30a20264fe9b08 Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Sat, 3 Oct 2026 01:38:16 +0100 Subject: [PATCH 4/4] Exercise detector admission in partial-delivery recovery proof Drive firings through the existing monitor callback instead of sending them directly through the notifier. This records the detector notification state required by the unchanged recovery policy; keep that policy and all failed attempts intact. Change-source: pulse-maintainer --- .../monitoring/monitor_alert_handling_test.go | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/internal/monitoring/monitor_alert_handling_test.go b/internal/monitoring/monitor_alert_handling_test.go index 33c98887b5..498bd35b8d 100644 --- a/internal/monitoring/monitor_alert_handling_test.go +++ b/internal/monitoring/monitor_alert_handling_test.go @@ -1372,9 +1372,17 @@ func TestMonitorDelayedPartialResolutionKeepsDestinationRecovery(t *testing.T) { } n := open() a, pbs := occurrenceManager(t) + m := &Monitor{alertManager: a, notificationMgr: n} + // Use the live firing callback so the detector records notification policy + // state. Direct notifier calls would not establish LastNotified, and existing + // recovery policy would correctly suppress that unannounced detector alert. + a.SetAlertCallback(m.handleAlertFired) a.CheckPBS(pbs) old := a.GetActiveAlerts()[0].Clone() - n.SendAlert(old) + if old.LastNotified == nil { + t.Fatal("detector did not admit the firing notification") + } + awaitOccurrenceRows(t, n, 2) n.StartQueueProcessing() select { case got := <-receipts: @@ -1402,6 +1410,8 @@ func TestMonitorDelayedPartialResolutionKeepsDestinationRecovery(t *testing.T) { pbs.CPU = 0 a.CheckPBS(pbs) n = open() // deferred processor; prior destination receipts are now on disk only + m = &Monitor{alertManager: a, notificationMgr: n} + a.SetAlertCallback(m.handleAlertFired) pbs.CPU = 95 a.CheckPBS(pbs) active := a.GetActiveAlerts() @@ -1409,8 +1419,7 @@ func TestMonitorDelayedPartialResolutionKeepsDestinationRecovery(t *testing.T) { t.Fatalf("not a new occurrence: %+v", active) } current := active[0].Clone() - n.SendAlert(current) - m := &Monitor{alertManager: a, notificationMgr: n} + awaitOccurrenceRows(t, n, 2) m.handleAlertResolved(old.ID) n.StartQueueProcessing() seenFiring, seenRecovery := false, false