Merge reviewed occurrence-qualified alert recovery

Retain exact Core candidate d0d621a7c2 and source proof; preserve current main frontend content.

Change-source: pulse-maintainer
This commit is contained in:
pulse-triage[bot]
2026-10-03 02:13:17 +01:00
21 changed files with 729 additions and 78 deletions
@@ -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.
@@ -3203,3 +3203,36 @@ 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.
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.
@@ -4634,3 +4634,33 @@ 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.
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.
@@ -1054,3 +1054,27 @@ 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.
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.
+3 -3
View File
@@ -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).
+7 -7
View File
@@ -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:
+52 -19
View File
@@ -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) })
}
}
+2 -2
View File
@@ -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
}
}
+1 -1
View File
@@ -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)
}
}
+1 -1
View File
@@ -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)
+1 -1
View File
@@ -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)
}
}
+1 -1
View File
@@ -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().
+1 -1
View File
@@ -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 {
@@ -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")
}
}
+2 -2
View File
@@ -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.
})
@@ -1205,3 +1205,264 @@ func TestMonitorLifecycleRefireReopensRetainedOccurrence(t *testing.T) {
})
}
}
type occurrenceHTTPReceipt struct {
Event string `json:"event"`
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)
}
// 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)
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()
if old.LastNotified == nil {
t.Fatal("detector did not admit the firing notification")
}
awaitOccurrenceRows(t, n, 2)
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
m = &Monitor{alertManager: a, notificationMgr: n}
a.SetAlertCallback(m.handleAlertFired)
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()
awaitOccurrenceRows(t, n, 2)
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")
}
}
+37 -29
View File
@@ -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,47 +222,53 @@ 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)
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
// 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) {
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().
@@ -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)
}
+27 -5
View File
@@ -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")
+25 -6
View File
@@ -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{}{}
@@ -0,0 +1,119 @@
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 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)
}
q, err = NewNotificationQueue(dir)
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 != 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 != reopenStatus || 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")
}
}