From 05e31eadf03ffd25a11f6e3244b60f6ddf1462e5 Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Sun, 6 Sep 2026 19:46:16 +0100 Subject: [PATCH 1/2] test(monitoring): synchronise canonical token host fixture GetMonitor starts polling concurrently, so monitor.mu does not protect the fixture host slice from State.GetSnapshot. Use the state-owned UpsertHost setter to match the reader lock while retaining all canonical-token diagnostics assertions. Addresses the fixture race reported in PR1943 rest-1; no production behaviour changes. Change-source: pulse-maintainer --- internal/monitoring/canonical_guardrails_test.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/internal/monitoring/canonical_guardrails_test.go b/internal/monitoring/canonical_guardrails_test.go index ac7f80490..4398e9f8c 100644 --- a/internal/monitoring/canonical_guardrails_test.go +++ b/internal/monitoring/canonical_guardrails_test.go @@ -2683,8 +2683,9 @@ func TestDefaultOrgMonitorSharesCanonicalRuntimeTokenInventory(t *testing.T) { Scopes: []string{config.ScopeAgentExec}, }} config.Mu.Unlock() - monitor.mu.Lock() - monitor.state.Hosts = []models.Host{{ + // GetMonitor starts polling concurrently; host fixtures must use the + // state-owned lock, not monitor.mu, to synchronise with snapshots. + monitor.state.UpsertHost(models.Host{ ID: "agent-fresh-token", Hostname: "fresh-token-host", Status: "online", @@ -2692,8 +2693,7 @@ func TestDefaultOrgMonitorSharesCanonicalRuntimeTokenInventory(t *testing.T) { AgentVersion: "6.2.2", TokenID: "fresh-agent-token", CommandsEnabled: true, - }} - monitor.mu.Unlock() + }) diagnostics := monitor.GetAgentFleetDiagnostics("6.2.2", now) agent := requireAgentDiagnostic(t, diagnostics, "agent-agent-fresh-token") From 8033fa581a179fe152443cbf2863caa167478071 Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Sun, 6 Sep 2026 20:00:22 +0100 Subject: [PATCH 2/2] fix(monitoring): serialize delivery health alert projection Concurrent timer and queue callbacks could apply an older health snapshot after a newer one, hiding a new delivery failure or resurrecting a dismissed warning. Serialize the complete read/apply operation without holding the monitor or queue mutex across alert updates. Add isolated channel-controlled stale-clear and stale-raise regression cases. Removing the lock fails both final-state assertions; restored code passes 100 race-enabled focused repetitions. This does not qualify the integrated release candidate or clear unrelated adverse evidence. Change-source: pulse-maintainer Contract-Neutral: Restores monitoring contract extension point 21 immediate canonical delivery-warning reconciliation by serializing existing read/apply operations; no public API, verdict, throttle, alert identity, or agent-lifecycle contract changes. Existing isolated ordering regression tests cover both stale-clear and stale-raise outcomes. --- internal/monitoring/monitor.go | 3 +- internal/monitoring/system_alerts.go | 13 ++++- internal/monitoring/system_alerts_test.go | 69 +++++++++++++++++++++++ 3 files changed, 83 insertions(+), 2 deletions(-) diff --git a/internal/monitoring/monitor.go b/internal/monitoring/monitor.go index 11a370b76..31d60322f 100644 --- a/internal/monitoring/monitor.go +++ b/internal/monitoring/monitor.go @@ -1124,7 +1124,8 @@ type Monitor struct { deadManConfigMu sync.RWMutex deadManConfig notifications.DeadManConfig deadManConfigLoadErr error - lastDeliveryHealthCheck time.Time // throttles the notification-delivery system alert evaluation; guarded by mu + deliveryHealthProjectionMu sync.Mutex // serializes delivery-health reads and alert projection + lastDeliveryHealthCheck time.Time // throttles the notification-delivery system alert evaluation; guarded by mu configPersist *config.ConfigPersistence discoveryService *discovery.Service // Background discovery service activePollCount int32 // Number of active polling operations diff --git a/internal/monitoring/system_alerts.go b/internal/monitoring/system_alerts.go index f49b28f5e..4b6bdd1f9 100644 --- a/internal/monitoring/system_alerts.go +++ b/internal/monitoring/system_alerts.go @@ -57,7 +57,18 @@ func (m *Monitor) evaluateNotificationDeliveryAt(now time.Time, force bool) { return } - health := notificationMgr.DeliveryHealth() + m.projectNotificationDeliveryHealth(alertManager, notificationMgr.DeliveryHealth) +} + +// projectNotificationDeliveryHealth serializes the entire read/apply operation. +// Locking only the alert mutation lets a paused, older health read overwrite a +// newer reconciliation (either resurrecting a dismissed warning or hiding a +// new failure). Queue callbacks enter here after releasing the queue lock. +func (m *Monitor) projectNotificationDeliveryHealth(alertManager *alerts.Manager, readHealth func() notifications.DeliveryHealth) { + m.deliveryHealthProjectionMu.Lock() + defer m.deliveryHealthProjectionMu.Unlock() + + health := readHealth() if health.Healthy { alertManager.ClearSystemAlert(alerts.NotificationDeliveryAlertType) return diff --git a/internal/monitoring/system_alerts_test.go b/internal/monitoring/system_alerts_test.go index 435fd94c8..40d8168bc 100644 --- a/internal/monitoring/system_alerts_test.go +++ b/internal/monitoring/system_alerts_test.go @@ -5,6 +5,7 @@ import ( "testing" "time" + "github.com/rcourtman/pulse-go-rewrite/internal/alerts" "github.com/rcourtman/pulse-go-rewrite/internal/notifications" ) @@ -130,3 +131,71 @@ func TestEvaluateNotificationDeliveryIsSafeWithoutAMonitor(t *testing.T) { var m *Monitor m.evaluateNotificationDelivery(time.Now()) } + +// Hold the older snapshot between read and apply while a newer reconciliation +// tries to enter. Exercise both stale-clear and stale-raise failure modes +// without a database, queue workers, or notification destinations. +func TestProjectNotificationDeliveryHealthOrdersSnapshots(t *testing.T) { + healthy := notifications.ClassifyQueueHealth(map[string]int{}) + failed := notifications.ClassifyQueueHealth(map[string]int{string(notifications.QueueStatusDLQ): 1}) + for _, tc := range []struct { + name string + old, current notifications.DeliveryHealth + }{ + {"new_failure_survives_old_clear", healthy, failed}, + {"dismissal_survives_old_failure", failed, healthy}, + } { + t.Run(tc.name, func(t *testing.T) { + manager := alerts.NewManagerWithDataDir(t.TempDir()) + t.Cleanup(manager.Stop) + m := &Monitor{} + read := make(chan struct{}) + release := make(chan struct{}) + oldDone := make(chan struct{}) + go func() { + defer close(oldDone) + m.projectNotificationDeliveryHealth(manager, func() notifications.DeliveryHealth { + close(read) + <-release + return tc.old + }) + }() + <-read + // This assertion is independent of scheduling: the snapshot must + // already be protected before reading, not just when applying it. + if m.deliveryHealthProjectionMu.TryLock() { + m.deliveryHealthProjectionMu.Unlock() + t.Error("health read is not protected by the projection lock") + } + newRead := make(chan struct{}) + newDone := make(chan struct{}) + go func() { + defer close(newDone) + m.projectNotificationDeliveryHealth(manager, func() notifications.DeliveryHealth { + close(newRead) + return tc.current + }) + }() + select { + case <-newRead: + // Ensure the newer state applies before releasing the stale + // snapshot when checking the unprotected implementation. + <-newDone + t.Error("new health read overtook an unfinished projection") + case <-time.After(25 * time.Millisecond): + } + close(release) + <-oldDone + <-newDone + active := false + for _, alert := range manager.GetActiveAlerts() { + if alert.Type == alerts.NotificationDeliveryAlertType { + active = true + } + } + if active == tc.current.Healthy { + t.Errorf("delivery warning active = %v, latest health healthy = %v", active, tc.current.Healthy) + } + }) + } +}