Merge pull request #2175 from rcourtman/maintainer/20260922T232553Z

Deliver persisted alert notifications that are due at startup

docs/release-control/v6/internal/subsystems/notifications.md

@@ -428,9 +428,22 @@ occurrence.
 That same queue boundary also owns processor attachment semantics. The
 canonical queue may persist pending notifications before a delivery processor is
 configured, but it must not mark those entries sending, failed, or sent until a
-processor exists. When a processor is attached, the queue owner must wake the
-pending backlog through the same canonical batch path instead of relying on a
-separate direct-send shortcut or waiting for an unrelated timer tick.
+processor exists. An explicit processor attachment must wake the pending backlog
+through the same canonical batch path instead of relying on a separate
+direct-send shortcut or waiting for an unrelated timer tick.
+
+Manager construction is the one attachment that must not wake the worker.
+Construction runs before the owner has applied saved destination configuration
+(webhooks, email, Apprise), so an already-due persisted job would be evaluated
+against a still-empty destination list and terminally cancelled as
+`ErrNotificationDeliverySkipped`. Construction records the processor without
+waking it; the worker's own ticker and any later enqueue begin delivery once
+configuration is present. This preserves the disabled-destination cancellation
+policy at processing time without letting startup order decide it.
+`TestRestartKeepsDueDeliveryPendingUntilConfigured` in
+`resolved_grouping_contract_test.go` seeds a due persisted delivery, constructs
+the manager before restoring its webhook, and proves the row stays `pending`
+until configuration is applied and then delivers.
 Alert delivery cooldown is also owned at this boundary. Normal alert delivery
 must suppress duplicate sends for the same active alert occurrence when
 cooldown is disabled or still active. A manager-admitted increase above the

frontend-modern/browser-tests/pbs-host-history-correlation.tsx

@@ -39,6 +39,7 @@ const pbs = {
   },
   // The PBS service target: correct for the service, but it has no host series.
   metricsTarget: { resourceType: 'agent', resourceId: 'pbs-1' },
+  discoveryTarget: { resourceType: 'agent', resourceId: 'pbs-1' },
   platformData: {
     sources: ['pbs'],
     pbs: { instanceId: 'proxback', hostname: 'proxback-vm', datastoreCount: 2 },
@@ -60,6 +61,7 @@ const guest = {
   memory: { current: 16.9, total: 8000, used: 1352, free: 6648 },
   agent: sharedAgent,
   metricsTarget: { resourceType: 'vm', resourceId: 'proxmox:100' },
+  discoveryTarget: { resourceType: 'agent', resourceId: 'proxback-vm' },
   platformData: { sources: ['proxmox', 'agent'], agent: sharedAgent },
 } as unknown as Resource;
 
@@ -76,6 +78,7 @@ const standalone = {
   lastSeen: Date.now(),
   agent: sharedAgent,
   metricsTarget: { resourceType: 'agent', resourceId: 'agent-proxback' },
+  discoveryTarget: { resourceType: 'agent', resourceId: 'proxback-vm' },
   platformData: { sources: ['agent', 'pbs'], agent: sharedAgent },
 } as unknown as Resource;
 

internal/notifications/notifications.go

@@ -755,9 +755,14 @@ func NewNotificationManagerWithDataDir(publicURL string, dataDir string) *Notifi
 	// Create webhook client after NotificationManager is initialized
 	nm.webhookClient = nm.createSecureWebhookClient(WebhookTimeout)
 
-	// Wire up queue processor if queue is available
+	// Wire up the queue processor without waking the worker. Construction runs
+	// before the owner applies saved destination configuration (webhooks,
+	// email, Apprise); waking here would let an already-due persisted job be
+	// processed while the destination list is still empty and be terminally
+	// cancelled as "delivery disabled". The worker's ticker and later enqueues
+	// begin delivery once configuration is present.
 	if queue != nil {
-		queue.SetProcessor(nm.ProcessQueuedNotification)
+		queue.installProcessor(nm.ProcessQueuedNotification)
 	}
 
 	// Start periodic cleanup of old lastNotified entries (every 1 hour)

internal/notifications/queue.go

@@ -1682,6 +1682,19 @@ func (nq *NotificationQueue) SetProcessor(processor func(*QueuedNotification) er
 	}
 }
 
+// installProcessor records the notification processor without waking the worker.
+// Manager construction uses this instead of SetProcessor because construction
+// runs before the owner applies saved destination configuration (webhooks,
+// email, Apprise). Waking here would let an already-due persisted job be
+// processed while the destination list is still empty and be terminally
+// cancelled as "delivery disabled". The worker's ticker and any later enqueue
+// begin delivery once configuration is present.
+func (nq *NotificationQueue) installProcessor(processor func(*QueuedNotification) error) {
+	nq.mu.Lock()
+	nq.processor = processor
+	nq.mu.Unlock()
+}
+
 // processBatch processes a batch of pending notifications concurrently
 func (nq *NotificationQueue) processBatch() {
 	const batchLimit = 20

internal/notifications/resolved_grouping_contract_test.go

@@ -102,7 +102,9 @@ func TestResolvedGroupingOrdinaryRestart(t *testing.T) {
 			var recovery payload
 			select {
 			case recovery = <-received:
-			case <-time.After(5 * time.Second):
+			case <-time.After(12 * time.Second):
+				// The queue processes on its own ticker; the wait must exceed that
+				// interval so a grouped recovery is not raced by the poll period.
 				t.Fatal("no grouped recovery")
 			}
 			if recovery.Event != "resolved" || len(recovery.Alerts) != 15 {
@@ -230,6 +232,77 @@ func TestResolvedGroupingDisabledBurstRetainsRateLimit(t *testing.T) {
 	}
 }
 
+// A restart must not terminally cancel a due persisted delivery before the
+// owner has applied saved destination configuration. Manager construction
+// happens before saved webhooks/email/Apprise are loaded in the monitor
+// startup path; the worker must not treat the still-empty destination list as
+// a permanent policy decision.
+func TestRestartKeepsDueDeliveryPendingUntilConfigured(t *testing.T) {
+	received := make(chan struct{}, 4)
+	server := newResolvedContractServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		received <- struct{}{}
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer server.Close()
+
+	dir := t.TempDir()
+	hook := WebhookConfig{ID: "ops", Name: "ops", URL: server.URL, Enabled: true}
+	configJSON, err := json.Marshal(hook)
+	if err != nil {
+		t.Fatal(err)
+	}
+	seed, err := NewNotificationQueue(dir)
+	if err != nil {
+		t.Fatal(err)
+	}
+	due := time.Now().Add(-time.Second)
+	if err := seed.Enqueue(&QueuedNotification{
+		ID:          "restart-due",
+		Type:        "webhook_resolved",
+		Status:      QueueStatusPending,
+		Config:      configJSON,
+		Alerts:      []*alerts.Alert{{ID: "restart-due", ResourceName: "restart-due", StartTime: time.Now().Add(-time.Minute)}},
+		MaxAttempts: 3,
+		NextRetryAt: &due,
+	}); err != nil {
+		t.Fatal(err)
+	}
+	if err := seed.Stop(); err != nil {
+		t.Fatal(err)
+	}
+
+	// Production order: construct the manager, then apply saved configuration.
+	m := NewNotificationManagerWithDataDir("", dir)
+	defer m.Stop()
+	m.webhookClient = server.Client()
+	if err := m.UpdateAllowedPrivateCIDRs("127.0.0.1/32"); err != nil {
+		t.Fatal(err)
+	}
+	// A premature startup wake must not consume the due job before the
+	// destination configuration is restored.
+	time.Sleep(250 * time.Millisecond)
+	queue := m.GetQueue()
+	if queue == nil {
+		t.Fatal("queue unavailable")
+	}
+	var status string
+	if err := queue.db.QueryRow(`SELECT status FROM notification_queue WHERE id = 'restart-due'`).Scan(&status); err != nil {
+		t.Fatal(err)
+	}
+	if status != string(QueueStatusPending) {
+		t.Fatalf("due delivery status before configuration = %s, want pending", status)
+	}
+
+	m.AddWebhook(hook)
+	queue.processBatch()
+	select {
+	case <-received:
+	case <-time.After(5 * time.Second):
+		stats, _ := queue.GetQueueStats()
+		t.Fatalf("configured due delivery not sent: %v", stats)
+	}
+}
+
 func TestResolvedGroupingPagerDutyKeepsIndividualKeys(t *testing.T) {
 	hook := WebhookConfig{ID: "pd", Service: "pagerduty", Enabled: true}
 	batch := contractGroupedAlerts()

scripts/check-pbs-host-history-correlation.cjs

@@ -38,6 +38,23 @@ const launchOptions = {
   const failures = [];
   const topologies = ['pbs-only', 'side-by-side', 'guest'];
   const wrong = 'agent/pbs-1';
+  // Reporter avsdev-cw sees the Identity card's "Discovery" and "Metrics
+  // Target" rows flick between a service id and the correlated host id while
+  // the table sits idle. Read the rendered rows so a snapshot-driven identity
+  // change fails the check instead of only being visible to a human.
+  const readIdentityRows = (detail) =>
+    detail
+      .locator('[data-testid="resource-identity-section"] tr')
+      .evaluateAll((rows) =>
+        rows
+          .map((row) => {
+            const cells = row.querySelectorAll('td');
+            return cells.length >= 2
+              ? [cells[0].textContent.trim(), cells[1].textContent.trim()]
+              : null;
+          })
+          .filter((row) => row && row[0]),
+      );
   try {
     await server.listen();
     browser = await chromium.launch(launchOptions);
@@ -170,6 +187,16 @@ const launchOptions = {
                 .locator('[data-testid="guest-history-plot"] path')
                 .count();
               assert.ok(pathsBefore > 0);
+              const identityBefore = await readIdentityRows(detail);
+              const identityLabels = identityBefore.map(([label]) => label);
+              assert.ok(
+                identityLabels.includes('Discovery'),
+                'Discovery identity row missing from the drawer',
+              );
+              assert.ok(
+                identityLabels.includes('Metrics Target'),
+                'Metrics Target identity row missing from the drawer',
+              );
               await page
                 .getByRole('button', { name: 'Schedule automatic snapshot' })
                 .click();
@@ -205,9 +232,16 @@ const launchOptions = {
                   .count(),
                 pathsBefore,
               );
+              const identityAfter = await readIdentityRows(detail);
+              assert.deepEqual(
+                identityAfter,
+                identityBefore,
+                'Identity rows changed after a refreshed snapshot',
+              );
               observations.push({
                 datastore,
                 paths: pathsBefore,
+                identity: identityAfter,
                 refreshPreservedHistory: true,
               });
             }