Guard resolved notification durability across abrupt process exit

Exercise pending, interrupted, failed and dead-lettered grouped alerts without closing SQLite before restart. Verify that resolution survives and retry preserves only live members and genuine recovery. Also fix the disabled-delivery test race found by repeated race testing: queue construction already starts workers, so use the locked processor setter and await reconciliation.

Change-source: pulse-maintainer

internal/notifications/queue_disabled_delivery_test.go

@@ -1,7 +1,9 @@
 package notifications
 
 import (
+	"sync/atomic"
 	"testing"
+	"time"
 
 	"github.com/rcourtman/pulse-go-rewrite/internal/alerts"
 	"github.com/rcourtman/pulse-go-rewrite/internal/operationaltrust"
@@ -35,18 +37,26 @@ func TestQueueDisabledDeliveryIsCancelledNotSent(t *testing.T) {
 					if err := q.Enqueue(n); err != nil {
 						t.Fatal(err)
 					}
-					// Do not start background workers; process one persisted row synchronously.
-					q.processor = nm.ProcessQueuedNotification
-					callbacks := 0
+					// Queue construction starts workers. Configure via the locked setter
+					// and wait for reconciliation even if a worker wins the claim.
+					var callbacks atomic.Int32
+					reconciled := make(chan struct{}, 2)
 					q.SetDeliveryHealthChangedCallback(func() {
-						callbacks++
+						callbacks.Add(1)
 						release := q.acquireAlertDeliveryGates([]string{"incident"}, true)
 						defer release()
 						if _, err := q.GetQueueStats(); err != nil {
 							t.Error(err)
 						}
+						reconciled <- struct{}{}
 					})
+					q.SetProcessor(nm.ProcessQueuedNotification)
 					q.processNotification(n)
+					select {
+					case <-reconciled:
+					case <-time.After(3 * time.Second):
+						t.Fatal("disabled delivery did not reconcile")
+					}
 					var status string
 					var completed *int64
 					if err := q.db.QueryRow(`SELECT status, completed_at FROM notification_queue WHERE id = ?`, n.ID).Scan(&status, &completed); err != nil {
@@ -71,8 +81,8 @@ func TestQueueDisabledDeliveryIsCancelledNotSent(t *testing.T) {
 							t.Errorf("%s has %d rows for a policy skip", table, count)
 						}
 					}
-					if callbacks != 1 {
-						t.Errorf("health callbacks = %d, want 1", callbacks)
+					if count := callbacks.Load(); count != 1 {
+						t.Errorf("health callbacks = %d, want 1", count)
 					}
 					if count, err := q.RetryTerminalFailures(); err != nil || count != 0 {
 						t.Errorf("retry = %d, %v; want no replay", count, err)

internal/notifications/queue_resolution_crash_test.go

@@ -0,0 +1,129 @@
+package notifications
+
+import (
+	"context"
+	"os"
+	"os/exec"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/rcourtman/pulse-go-rewrite/internal/alerts"
+)
+
+// Exit without Stop or database Close: graceful reopen tests cannot establish
+// that resolution rewrites and cancellations survive a process exit.
+// This is queue durability evidence, not an installed provider receipt or a
+// power-loss simulation; an interrupted provider send remains at-least-once.
+func TestQueueResolutionSurvivesAbruptProcessExit(t *testing.T) {
+	const helperEnv = "PULSE_TEST_QUEUE_RESOLUTION_EXIT_DIR"
+	const exitCode = 23
+	states := []NotificationQueueStatus{QueueStatusPending, QueueStatusSending, QueueStatusFailed, QueueStatusDLQ}
+	if dir := os.Getenv(helperEnv); dir != "" {
+		q, err := NewNotificationQueue(dir)
+		if err != nil {
+			t.Fatal(err)
+		}
+		enqueue := func(id, kind string, members ...string) *QueuedNotification {
+			t.Helper()
+			n := &QueuedNotification{ID: id, Type: kind, Status: QueueStatusPending, Config: []byte(`{}`), MaxAttempts: 3}
+			for _, member := range members {
+				n.Alerts = append(n.Alerts, &alerts.Alert{ID: member})
+			}
+			if err := q.Enqueue(n); err != nil {
+				t.Fatal(err)
+			}
+			return n
+		}
+		for _, status := range states {
+			for _, grouped := range []bool{false, true} {
+				id := string(status)
+				members := []string{"recovered"}
+				if grouped {
+					id += "-group"
+					members = append(members, "still-firing")
+				}
+				enqueue(id, "webhook", members...)
+				if err := q.UpdateStatus(id, status, "destination unavailable"); err != nil {
+					t.Fatal(err)
+				}
+			}
+		}
+		enqueue("recovery", "webhook_resolved", "recovered")
+		if count, err := q.CancelByAlertIdentifiers([]string{"recovered"}); err != nil || count != 2 {
+			t.Fatalf("resolution = %d, %v; want two pending members", count, err)
+		}
+		os.Exit(exitCode)
+	}
+
+	dir := t.TempDir()
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	cmd := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestQueueResolutionSurvivesAbruptProcessExit$")
+	cmd.Env = append(os.Environ(), helperEnv+"="+dir)
+	output, err := cmd.CombinedOutput()
+	if ctx.Err() != nil {
+		t.Fatalf("child timed out: %v\n%s", ctx.Err(), output)
+	}
+	if exited, ok := err.(*exec.ExitError); !ok || exited.ExitCode() != exitCode {
+		t.Fatalf("child did not reach deliberate exit: %v\n%s", err, output)
+	}
+	q, err := NewNotificationQueue(dir)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer q.Stop()
+	if count, err := q.RetryTerminalFailures(); err != nil || count != 2 {
+		t.Fatalf("terminal retry = %d, %v; want only failed/dlq surviving groups", count, err)
+	}
+	var mu sync.Mutex
+	delivered := make(map[string][]string)
+	// Exercise ordinary dispatch as well as repeated batch selection.
+	q.SetProcessor(func(n *QueuedNotification) error {
+		mu.Lock()
+		defer mu.Unlock()
+		for _, a := range n.Alerts {
+			delivered[n.ID] = append(delivered[n.ID], a.ID)
+		}
+		return nil
+	})
+	q.processBatch()
+	q.processBatch()
+	deadline := time.Now().Add(3 * time.Second)
+	for {
+		stats, err := q.GetQueueStats()
+		if err != nil {
+			t.Fatal(err)
+		}
+		if stats["pending"]+stats["sending"] == 0 {
+			break
+		}
+		if time.Now().After(deadline) {
+			t.Fatal("restarted queue did not drain")
+		}
+		time.Sleep(time.Millisecond)
+	}
+	mu.Lock()
+	defer mu.Unlock()
+	if len(delivered) != 5 {
+		t.Fatalf("delivered = %v; want four surviving groups and recovery", delivered)
+	}
+	for _, id := range []string{"pending-group", "sending-group", "failed-group", "dlq-group", "recovery"} {
+		want := "still-firing"
+		if id == "recovery" {
+			want = "recovered"
+		}
+		if got := delivered[id]; len(got) != 1 || got[0] != want {
+			t.Errorf("%s delivered %v, want [%s] exactly once in this run", id, got, want)
+		}
+	}
+	for _, id := range []string{"pending", "sending", "failed", "dlq"} {
+		var status string
+		if err := q.db.QueryRow("SELECT status FROM notification_queue WHERE id = ?", id).Scan(&status); err != nil {
+			t.Fatal(err)
+		}
+		if status != string(QueueStatusCancelled) {
+			t.Errorf("%s status = %s, want cancelled", id, status)
+		}
+	}
+}