Merge pull request #1944 from rcourtman/maintainer/20260906T185414Z

Keep delivery warnings current and alert checks reliable

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")

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

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

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)
+			}
+		})
+	}
+}