Make update readiness and history tests reliable (#2304)

## What and why Stabilise agent update-readiness and monitoring history-window tests without weakening their assertions. The failed v6.4.5-rc.4 exact preflight (packet 20260928T080527Z-release) exposed timing-sensitive failures that could block the #2282 updater repair from reaching a qualified release. This main-line batch also incorporates the already-published dropped-session dispatch repair from Pulse #2300 and container-console setup guidance from #2302 without rewriting either reviewed test commit. ## Included range Pulse `main` `75c72ef505b24e31214162b7b46f97e357528ae3..cbb9122bcc216e335867fa4fed98c4451b12bd27`: reviewed test candidates `bc3a7306` and `383e05ae`, followed by additive upstream integrations `894d7409` and `cbb9122b` (published main `ad15366f`). ## Validation Provenance, clean-tree and diff checks passed. Committed-source offline race proof `source-_ki_8c7c` passed the named failure tests for 20 repetitions. The composed v6.4 counterpart passed the full affected packages with race detection (`source-7g64cyz2`). Exact-head GitHub checks remain required before merge; source tests do not qualify or publish v6.4.5. [How Pulse handles triage](https://github.com/rcourtman/Pulse/blob/main/docs/AI_TRANSPARENCY.md)
2026-10-24 04:33:17 +00:00 · 2026-09-28 15:48:17 +00:00
parent ad15366f3d cbb9122bcc
commit 260337c951
2 changed files with 109 additions and 25 deletions
@@ -1901,11 +1901,13 @@ func (s *pendingUpdateSupervisorStub) Rollback(_ context.Context, activation age
 func testPendingUpdate(t *testing.T, stateDir string) *agentupdate.PendingPrivilegedUpdate {
 	t.Helper()
 	activation := agenthelper.UpdateResult{
-		Action:           "pending",
-		ActivationID:     "pulse-agent-0123456789abcdef0123456789abcdef:0123456789abcdef",
-		ActiveSHA256:     strings.Repeat("a", 64),
-		RollbackSHA256:   strings.Repeat("b", 64),
-		RollbackDeadline: time.Now().Add(2 * time.Second).UTC(),
+		Action:         "pending",
+		ActivationID:   "pulse-agent-0123456789abcdef0123456789abcdef:0123456789abcdef",
+		ActiveSHA256:   strings.Repeat("a", 64),
+		RollbackSHA256: strings.Repeat("b", 64),
+		// Leave enough time for the race-instrumented supervisor to be scheduled.
+		// The test below checks the health gates, not the expiry path.
+		RollbackDeadline: time.Now().Add(30 * time.Second).UTC(),
 	}
 	if err := agentupdate.PersistPendingPrivilegedUpdate(stateDir, "1.0.0", activation); err != nil {
 		t.Fatal(err)
@@ -1932,26 +1934,71 @@ func TestPendingPrivilegedUpdateCommitsOnlyAfterReadinessAndAcceptedReport(t *te
 	reportAccepted := make(chan struct{})
 	close(reportAccepted)
 	var ready atomic.Bool
+	notReadyChecks := make(chan struct{}, 2)
+	locallyReady := func() bool {
+		if ready.Load() {
+			return true
+		}
+		select {
+		case notReadyChecks <- struct{}{}:
+		default:
+		}
+		return false
+	}
 	result := make(chan error, 1)
 	go func() {
-		result <- supervisePendingPrivilegedUpdate(context.Background(), stub, pending, stateDir, pending.Activation.ActiveSHA256, ready.Load, reportAccepted, time.Millisecond, nil)
+		result <- supervisePendingPrivilegedUpdate(context.Background(), stub, pending, stateDir, pending.Activation.ActiveSHA256, locallyReady, reportAccepted, time.Millisecond, nil)
 	}()
+	// Observe the supervisor checking the false readiness state on two passes.
+	// A sleep cannot prove it started before we flip readiness under -race.
+	for range 2 {
+		select {
+		case <-notReadyChecks:
+		case <-stub.commitCalls:
+			t.Fatal("pending update committed before local readiness")
+		case err := <-result:
+			t.Fatalf("pending update supervisor stopped before local readiness: %v", err)
+		case <-time.After(10 * time.Second):
+			t.Fatal("pending update supervisor did not check local readiness")
+		}
+	}
 	select {
 	case <-stub.commitCalls:
 		t.Fatal("pending update committed before local readiness")
-	case <-time.After(20 * time.Millisecond):
+	default:
 	}
 	ready.Store(true)
+	var activation agenthelper.UpdateResult
+	finished := false
 	select {
-	case activation := <-stub.commitCalls:
-		if activation != pending.Activation {
-			t.Fatalf("commit activation = %#v", activation)
+	case activation = <-stub.commitCalls:
+	case err := <-result:
+		if err != nil {
+			t.Fatalf("pending update supervisor stopped before committing: %v", err)
 		}
-	case <-time.After(time.Second):
+		finished = true
+		// Commit sends its recorded call before the supervisor can return. If
+		// both channels are ready, selecting result first is still success.
+		select {
+		case activation = <-stub.commitCalls:
+		default:
+			t.Fatal("pending update supervisor returned without committing")
+		}
+	case <-time.After(10 * time.Second):
 		t.Fatal("pending update was not committed after both health signals")
 	}
-	if err := <-result; err != nil {
-		t.Fatal(err)
+	if activation != pending.Activation {
+		t.Fatalf("commit activation = %#v", activation)
+	}
+	if !finished {
+		select {
+		case err := <-result:
+			if err != nil {
+				t.Fatal(err)
+			}
+		case <-time.After(10 * time.Second):
+			t.Fatal("pending update supervisor did not finish after commit")
+		}
 	}
 	if loaded, err := agentupdate.LoadPendingPrivilegedUpdate(stateDir); err != nil || loaded != nil {
 		t.Fatalf("committed handoff = %#v, %v", loaded, err)
@@ -2,12 +2,35 @@ package monitoring

 import (
 	"strings"
+	"sync"
 	"testing"
 	"time"

 	"github.com/rcourtman/pulse-go-rewrite/internal/alerts"
+	"github.com/rcourtman/pulse-go-rewrite/internal/models"
+	"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
 )

+// This resolver blocks only after metricWindowPoints has captured its history.
+// It lets the test replace that history at an exact point without scheduling
+// sleeps or exposing a production-only hook.
+type blockingMetricTargetStore struct {
+	entered chan struct{}
+	resume  <-chan struct{}
+}
+
+func (*blockingMetricTargetStore) ShouldSkipAPIPolling(string) bool { return false }
+func (*blockingMetricTargetStore) GetPollingRecommendations() map[string]float64 {
+	return nil
+}
+func (*blockingMetricTargetStore) GetAll() []unifiedresources.Resource       { return nil }
+func (*blockingMetricTargetStore) PopulateFromSnapshot(models.StateSnapshot) {}
+func (s *blockingMetricTargetStore) MetricsTargetForResource(string) *unifiedresources.MetricsTarget {
+	s.entered <- struct{}{}
+	<-s.resume
+	return nil
+}
+
 func TestMetricWindowPointsUsesInMemoryMetricAlias(t *testing.T) {
 	now := time.Now().UTC()
 	history := NewMetricsHistory(32, time.Hour)
@@ -77,33 +100,47 @@ func TestMetricWindowPointsKeepsHistorySnapshotDuringReplacement(t *testing.T) {
 		},
 	}

-	history.metricWindowMu.Lock()
-	result := make(chan []alerts.MetricWindowPoint, 1)
+	resume := make(chan struct{})
+	var resumeOnce sync.Once
+	resumeRequest := func() { resumeOnce.Do(func() { close(resume) }) }
+	defer resumeRequest()
+	entered := make(chan struct{}, 1)
+	monitor.resourceStore = &blockingMetricTargetStore{entered: entered, resume: resume}
+	type requestResult struct {
+		points []alerts.MetricWindowPoint
+		err    error
+	}
+	result := make(chan requestResult, 1)
 	go func() {
-		points, _ := monitor.metricWindowPoints(alerts.MetricWindowRequest{
+		points, err := monitor.metricWindowPoints(alerts.MetricWindowRequest{
 			ResourceID:   "vm-1",
 			ResourceType: "vm",
 			Metric:       "cpu",
 			Start:        now.Add(-5 * time.Minute),
 			End:          now,
 		})
-		result <- points
+		result <- requestResult{points: points, err: err}
 	}()

-	// Give the request time to snapshot history and block on its cache mutex,
-	// matching the mock seed replacement that exposed the startup panic.
-	time.Sleep(20 * time.Millisecond)
+	select {
+	case <-entered:
+	case <-time.After(10 * time.Second):
+		t.Fatal("metric request did not reach target resolution after snapshot")
+	}
 	monitor.mu.Lock()
 	monitor.metricsHistory = NewMetricsHistory(32, time.Hour)
 	monitor.mu.Unlock()
-	history.metricWindowMu.Unlock()
+	resumeRequest()

 	select {
-	case points := <-result:
-		if len(points) != 1 || points[0].Value != 42 {
-			t.Fatalf("metricWindowPoints = %+v, want cached point from the request snapshot", points)
+	case got := <-result:
+		if got.err != nil {
+			t.Fatalf("metricWindowPoints returned error: %v", got.err)
 		}
-	case <-time.After(time.Second):
+		if len(got.points) != 1 || got.points[0].Value != 42 {
+			t.Fatalf("metricWindowPoints = %+v, want cached point from the request snapshot", got.points)
+		}
+	case <-time.After(10 * time.Second):
 		t.Fatal("metricWindowPoints did not finish after history replacement")
 	}
 }