mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-10-23 20:25:56 +00:00
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
This commit is contained in:
@@ -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.
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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().
|
||||
|
||||
Reference in New Issue
Block a user