diff --git a/docs/release-control/v6/internal/subsystems/notifications.md b/docs/release-control/v6/internal/subsystems/notifications.md index e4f103a1c8..44d11627a7 100644 --- a/docs/release-control/v6/internal/subsystems/notifications.md +++ b/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 diff --git a/frontend-modern/browser-tests/pbs-host-history-correlation.tsx b/frontend-modern/browser-tests/pbs-host-history-correlation.tsx index cbb60a02aa..72e46eb0a3 100644 --- a/frontend-modern/browser-tests/pbs-host-history-correlation.tsx +++ b/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; diff --git a/internal/notifications/notifications.go b/internal/notifications/notifications.go index 032a3e6d22..1ec9051bd7 100644 --- a/internal/notifications/notifications.go +++ b/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) diff --git a/internal/notifications/queue.go b/internal/notifications/queue.go index 3141e7a3c3..e365ec2b0a 100644 --- a/internal/notifications/queue.go +++ b/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 diff --git a/internal/notifications/resolved_grouping_contract_test.go b/internal/notifications/resolved_grouping_contract_test.go index 3fd37bd36d..0e0241a3a9 100644 --- a/internal/notifications/resolved_grouping_contract_test.go +++ b/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() diff --git a/scripts/check-pbs-host-history-correlation.cjs b/scripts/check-pbs-host-history-correlation.cjs index 6b48f45a61..1d0cab8afc 100644 --- a/scripts/check-pbs-host-history-correlation.cjs +++ b/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, }); }