From 3c8250388f72ff2617d38e24fd1ef647b201c35a Mon Sep 17 00:00:00 2001
From: Karim Radhouani <medkarimrdi@gmail.com>
Date: Mon, 10 Aug 2026 21:58:45 -0700
Subject: [PATCH 1/5] refactor: enhance scaling configuration and resource
 management in tests

- Updated test.mk to include new environment variables for scaling configuration: SCALE_REPLICAS, SCALE_CPU_REQUEST, SCALE_CPU_LIMIT, SCALE_MEMORY_REQUEST, and SCALE_MEMORY_LIMIT.
- Refactored integration tests to utilize new helper functions for managing subscription paths and outputs, improving test reliability and clarity.
- Adjusted baseline YAML to dynamically reference CPU and memory resource requests and limits, enhancing flexibility in scaling tests.
---
 test.mk                                       | 16 +++-
 .../004-subscriptions/subscriptions_test.go   | 74 ++++++++++++++++---
 .../suite/005-outputs/outputs_test.go         | 27 +++++--
 .../suite/013-scale/fixtures/baseline.yaml    | 10 +--
 .../integration/suite/013-scale/scale_test.go | 40 ++++++++--
 5 files changed, 136 insertions(+), 31 deletions(-)

diff --git a/test.mk b/test.mk
index b4a5c47..6f7b1bb 100644
--- a/test.mk
+++ b/test.mk
@@ -283,9 +283,21 @@ run-integration-tests-v2: ## Bring up the gnmi-gen suite env, run all suites, te
 	exit $$status
 
 # Nightly / local-only fleet suite. Not wired into CI.
-# SCALE_TARGETS defaults to 200; override e.g. SCALE_TARGETS=50 make integration-test-scale
+# Override e.g. SCALE_TARGETS=50 SCALE_REPLICAS=2 SCALE_CPU_LIMIT=1 make integration-test-scale
 SCALE_TARGETS ?= 200
+SCALE_REPLICAS ?= 4
+SCALE_CPU_REQUEST ?= 500m
+SCALE_CPU_LIMIT ?= 2
+SCALE_MEMORY_REQUEST ?= 512Mi
+SCALE_MEMORY_LIMIT ?= 2Gi
 .PHONY: integration-test-scale
 integration-test-scale: integration-env-check ## Run 013-scale (sets RUN_SCALE=1)
-	RUN_SCALE=1 SCALE_TARGETS=$(SCALE_TARGETS) go test -tags=integration -count=1 -timeout 45m -v ./$(IT_SUITE_DIR)/013-scale/...
+	RUN_SCALE=1 \
+	SCALE_TARGETS=$(SCALE_TARGETS) \
+	SCALE_REPLICAS=$(SCALE_REPLICAS) \
+	SCALE_CPU_REQUEST=$(SCALE_CPU_REQUEST) \
+	SCALE_CPU_LIMIT=$(SCALE_CPU_LIMIT) \
+	SCALE_MEMORY_REQUEST=$(SCALE_MEMORY_REQUEST) \
+	SCALE_MEMORY_LIMIT=$(SCALE_MEMORY_LIMIT) \
+	go test -tags=integration -count=1 -timeout 45m -v ./$(IT_SUITE_DIR)/013-scale/...
 
diff --git a/test/integration/suite/004-subscriptions/subscriptions_test.go b/test/integration/suite/004-subscriptions/subscriptions_test.go
index ece79fc..95d6738 100644
--- a/test/integration/suite/004-subscriptions/subscriptions_test.go
+++ b/test/integration/suite/004-subscriptions/subscriptions_test.go
@@ -103,7 +103,6 @@ func waitIdle(t *testing.T) {
 
 func applySubscription(t *testing.T, name string, vars map[string]any) {
 	t.Helper()
-	setPipelineEnabled(t, true)
 	base := map[string]any{
 		"Name":              name,
 		"Mode":              "STREAM/SAMPLE",
@@ -122,9 +121,13 @@ func applySubscription(t *testing.T, name string, vars map[string]any) {
 	if err != nil {
 		t.Fatalf("reading subscription fixture: %v", err)
 	}
+	// Create the Subscription before re-enabling the pipeline. Enabling first
+	// can race an apply with targets but no subscriptions, which gNMIc rejects
+	// and can leave the collector mid-reconnect with paths=[].
 	if _, err := s.K8s.ApplyYAMLNoCleanup(string(b), base); err != nil {
 		t.Fatalf("applying subscription %s: %v", name, err)
 	}
+	setPipelineEnabled(t, true)
 	waitClusterReady(t)
 }
 
@@ -167,6 +170,61 @@ func waitPathOnAll(t *testing.T, path string) {
 	}
 }
 
+// waitPathsOnAll waits until every target carries all want paths on the same
+// stream snapshot. Split waitPathOnAll calls race stream reconnects: the first
+// path can still be visible on the old stream, then paths go empty before the
+// next assertion.
+func waitPathsOnAll(t *testing.T, want ...string) {
+	t.Helper()
+	harness.Wait(t, harness.Medium, fmt.Sprintf("paths %v on all targets", want), func() (bool, string) {
+		for _, name := range allTargets {
+			paths := s.GnmiGen.Paths(name)
+			for _, w := range want {
+				found := false
+				for _, p := range paths {
+					if p == w {
+						found = true
+						break
+					}
+				}
+				if !found {
+					return false, fmt.Sprintf("%s paths=%v", name, paths)
+				}
+			}
+		}
+		return true, ""
+	})
+}
+
+// waitPathReplacedOnAll waits until every target shows have and not gone on the
+// same snapshot. Sequential present-then-absent waits can pass the first check
+// on a transitional stream (or see paths=[] mid-reconnect) and then burn the
+// full Medium timeout on the second.
+func waitPathReplacedOnAll(t *testing.T, have, gone string) {
+	t.Helper()
+	harness.Wait(t, harness.Medium, fmt.Sprintf("path %s replaces %s on all targets", have, gone), func() (bool, string) {
+		for _, name := range allTargets {
+			if n := s.GnmiGen.StreamCount(name); n != 1 {
+				return false, fmt.Sprintf("%s streams=%d paths=%v", name, n, s.GnmiGen.Paths(name))
+			}
+			paths := s.GnmiGen.Paths(name)
+			hasHave, hasGone := false, false
+			for _, p := range paths {
+				if p == have {
+					hasHave = true
+				}
+				if p == gone {
+					hasGone = true
+				}
+			}
+			if !hasHave || hasGone {
+				return false, fmt.Sprintf("%s paths=%v", name, paths)
+			}
+		}
+		return true, ""
+	})
+}
+
 func waitPathAbsentOnAll(t *testing.T, path string) {
 	t.Helper()
 	for _, name := range allTargets {
@@ -305,9 +363,9 @@ func TestSub001_PathChangeReplacesCollectedPaths(t *testing.T) {
 
 	restartsBefore := s.K8s.RestartCounts(t, cluster)
 	patchSubscription(t, "s1", map[string]any{"paths": []string{pathCPU}})
+	harness.WaitConfigApplied(t, s.K8s, cluster)
 
-	waitPathOnAll(t, pathCPU)
-	waitPathAbsentOnAll(t, pathIF)
+	waitPathReplacedOnAll(t, pathCPU, pathIF)
 	waitStreams(t, 1)
 	harness.AssertNoRestarts(t, restartsBefore, s.K8s.RestartCounts(t, cluster))
 
@@ -325,12 +383,7 @@ func TestSub002_AddingPathExtendsSubscription(t *testing.T) {
 	restartsBefore := s.K8s.RestartCounts(t, cluster)
 	patchSubscription(t, "s1", map[string]any{"paths": []string{pathIF, pathCPU}})
 
-	waitPathOnAll(t, pathIF)
-	waitPathOnAll(t, pathCPU)
-	harness.Wait(t, harness.Medium, "resolved path count is 2", func() (bool, string) {
-		n := len(s.GnmiGen.Paths(leaf1))
-		return n >= 2, fmt.Sprintf("paths=%v", s.GnmiGen.Paths(leaf1))
-	})
+	waitPathsOnAll(t, pathIF, pathCPU)
 	// Path edits re-establish the Subscribe stream, which resets
 	// notifications_sent. Prove data still flows on the new stream rather
 	// than comparing absolute counters across the teardown.
@@ -474,8 +527,7 @@ func TestSub009_DeletingSubscriptionStopsPaths(t *testing.T) {
 	applySubscription(t, "if", map[string]any{"Paths": []string{pathIF}})
 	applySubscription(t, "cpu", map[string]any{"Paths": []string{pathCPU}})
 	waitStreams(t, 2)
-	waitPathOnAll(t, pathIF)
-	waitPathOnAll(t, pathCPU)
+	waitPathsOnAll(t, pathIF, pathCPU)
 	harness.WaitClusterCounts(t, s.K8s, cluster, harness.ClusterCounts{
 		SubscriptionsCount: harness.I32(2),
 	})
diff --git a/test/integration/suite/005-outputs/outputs_test.go b/test/integration/suite/005-outputs/outputs_test.go
index bf7089d..f880252 100644
--- a/test/integration/suite/005-outputs/outputs_test.go
+++ b/test/integration/suite/005-outputs/outputs_test.go
@@ -275,6 +275,12 @@ func TestOut005_EditingOutputNoRestart(t *testing.T) {
 	waitIdle(t)
 	const oldFile = "/tmp/005-old.jsonl"
 	const newFile = "/tmp/005-new.jsonl"
+	// Prior runs leave these files in the pod; clear while no file output is
+	// mounted so "non-empty" cannot pass on stale content.
+	pod := s.K8s.FirstCollectorPod(t, cluster)
+	_ = s.K8s.Exec(t, pod, harness.CollectorContainer, "sh", "-c",
+		fmt.Sprintf("rm -f %q %q", oldFile, newFile))
+
 	applyFileOutput(t, "file1", oldFile)
 	applyPromOutput(t, "out1", map[string]any{"Expiration": "60s"})
 	applyPipeline(t, "p1", []string{"file1", "out1"}, "")
@@ -283,7 +289,6 @@ func TestOut005_EditingOutputNoRestart(t *testing.T) {
 	s.K8s.WaitClusterPrometheusSources(t, cluster, "p1", "out1", allTargets, harness.Medium)
 
 	restartsBefore := s.K8s.RestartCounts(t, cluster)
-	oldLen := len(s.K8s.ReadCollectorFile(t, cluster, oldFile))
 
 	out := &gnmicv1alpha1.Output{}
 	s.K8s.WaitExists(t, "file1", out)
@@ -294,14 +299,22 @@ func TestOut005_EditingOutputNoRestart(t *testing.T) {
 	s.K8s.Patch(t, prom, `{"spec":{"config":{"expiration":"120s"}}}`)
 
 	harness.WaitConfigApplied(t, s.K8s, cluster)
-	s.K8s.WaitCollectorFileNonEmpty(t, cluster, newFile)
+	newBody := s.K8s.WaitCollectorFileNonEmpty(t, cluster, newFile)
 	s.K8s.WaitClusterPrometheusSources(t, cluster, "p1", "out1", allTargets, harness.Medium)
 
-	// Old file should stop growing once the output points elsewhere.
-	time.Sleep(3 * time.Second)
-	if got := len(s.K8s.ReadCollectorFile(t, cluster, oldFile)); got > oldLen+64 {
-		t.Errorf("old file still growing: was %d now %d", oldLen, got)
-	}
+	// Mark after the switch: sampling before the patch counted apply-window
+	// writes as "growth" and flaked. Prove live traffic hits the new file,
+	// then that the old one stays put.
+	newFloor := len(newBody)
+	s.K8s.WaitCollectorFileGrows(t, cluster, newFile, newFloor)
+	oldLen := len(s.K8s.ReadCollectorFile(t, cluster, oldFile))
+	harness.Consistently(t, 5*time.Second, time.Second, "old file stopped growing", func() (bool, string) {
+		got := len(s.K8s.ReadCollectorFile(t, cluster, oldFile))
+		if got > oldLen+64 {
+			return false, fmt.Sprintf("was %d now %d", oldLen, got)
+		}
+		return true, ""
+	})
 	harness.AssertNoRestarts(t, restartsBefore, s.K8s.RestartCounts(t, cluster))
 }
 
diff --git a/test/integration/suite/013-scale/fixtures/baseline.yaml b/test/integration/suite/013-scale/fixtures/baseline.yaml
index bceb3b3..70a7e4f 100644
--- a/test/integration/suite/013-scale/fixtures/baseline.yaml
+++ b/test/integration/suite/013-scale/fixtures/baseline.yaml
@@ -1,5 +1,5 @@
 # Shared props for 013-scale. Target CRs are generated in AfterBaseline.
-# Cluster replicas come from SCALE_REPLICAS via BaselineVars.
+# Cluster replicas and collector resources come from SCALE_* via BaselineVars.
 apiVersion: v1
 kind: Secret
 metadata:
@@ -30,11 +30,11 @@ spec:
     restPort: 7890
   resources:
     requests:
-      cpu: 500m
-      memory: 512Mi
+      cpu: "{{ .CPURequest }}"
+      memory: "{{ .MemoryRequest }}"
     limits:
-      cpu: "2"
-      memory: 2Gi
+      cpu: "{{ .CPULimit }}"
+      memory: "{{ .MemoryLimit }}"
 ---
 apiVersion: operator.gnmic.dev/v1alpha1
 kind: Subscription
diff --git a/test/integration/suite/013-scale/scale_test.go b/test/integration/suite/013-scale/scale_test.go
index 20a6dd6..29d074f 100644
--- a/test/integration/suite/013-scale/scale_test.go
+++ b/test/integration/suite/013-scale/scale_test.go
@@ -4,8 +4,14 @@
 // single-target churn cost, sustained membership change, and mass reboot.
 //
 // Gated behind RUN_SCALE=1 (see TestMain). Not part of the default CI lane.
-// Fleet size defaults to 200 (SCALE_TARGETS) and collector pods to 4
-// (SCALE_REPLICAS).
+// Tunables (env vars, with defaults):
+//
+//	SCALE_TARGETS=200
+//	SCALE_REPLICAS=4
+//	SCALE_CPU_REQUEST=500m
+//	SCALE_CPU_LIMIT=2
+//	SCALE_MEMORY_REQUEST=512Mi
+//	SCALE_MEMORY_LIMIT=2Gi
 package scale
 
 import (
@@ -37,6 +43,10 @@ var (
 	targets   []string
 	fleetN    int
 	replicas  int
+	cpuReq    string
+	cpuLim    string
+	memReq    string
+	memLim    string
 	spareSim  string
 	sparePort int
 	minPerPod int
@@ -50,6 +60,10 @@ func TestMain(m *testing.M) {
 	}
 	fleetN = envInt("SCALE_TARGETS", 200)
 	replicas = envInt("SCALE_REPLICAS", 4)
+	cpuReq = envStr("SCALE_CPU_REQUEST", "500m")
+	cpuLim = envStr("SCALE_CPU_LIMIT", "2")
+	memReq = envStr("SCALE_MEMORY_REQUEST", "512Mi")
+	memLim = envStr("SCALE_MEMORY_LIMIT", "2Gi")
 	if replicas < 1 {
 		fmt.Fprintf(os.Stderr, "013-scale: SCALE_REPLICAS=%d too small (min 1)\n", replicas)
 		os.Exit(1)
@@ -65,8 +79,8 @@ func TestMain(m *testing.M) {
 	for i := 1; i <= fleetN; i++ {
 		targets[i-1] = fmt.Sprintf("dev-%d", i)
 	}
-	fmt.Fprintf(os.Stderr, "[harness] suite 013-scale: SCALE_TARGETS=%d SCALE_REPLICAS=%d spare=%s placement=%d..%d per pod\n",
-		fleetN, replicas, spareSim, minPerPod, maxPerPod)
+	fmt.Fprintf(os.Stderr, "[harness] suite 013-scale: SCALE_TARGETS=%d SCALE_REPLICAS=%d cpu=%s/%s mem=%s/%s spare=%s placement=%d..%d per pod\n",
+		fleetN, replicas, cpuReq, cpuLim, memReq, memLim, spareSim, minPerPod, maxPerPod)
 
 	require := append(append([]string{}, targets...), spareSim)
 	os.Exit(harness.RunSuite(m, harness.Options{
@@ -74,8 +88,14 @@ func TestMain(m *testing.M) {
 		GnmiGenConfigData: renderGnmiGenConfig(fleetN + 1),
 		RequireTargets:    require,
 		Baseline:          []string{"fixtures/baseline.yaml"},
-		BaselineVars:      map[string]any{"Replicas": replicas},
-		AfterBaseline:     applyFleetTargets,
+		BaselineVars: map[string]any{
+			"Replicas":      replicas,
+			"CPURequest":    cpuReq,
+			"CPULimit":      cpuLim,
+			"MemoryRequest": memReq,
+			"MemoryLimit":   memLim,
+		},
+		AfterBaseline: applyFleetTargets,
 	}, &s))
 }
 
@@ -91,6 +111,14 @@ func envInt(name string, def int) int {
 	return n
 }
 
+func envStr(name, def string) string {
+	v := strings.TrimSpace(os.Getenv(name))
+	if v == "" {
+		return def
+	}
+	return v
+}
+
 // placementBand is avg±20% with a floor that matches the design's 40–60 band
 // at the default 200/4 fleet.
 func placementBand(n, r int) (int, int) {

From 86fb3a0171a0e94fa3b3235cf6988c034ef25ef5 Mon Sep 17 00:00:00 2001
From: Karim Radhouani <medkarimrdi@gmail.com>
Date: Tue, 11 Aug 2026 18:30:58 -0700
Subject: [PATCH 2/5] fix: re-apply config once after ACK to recover missed
 Subscribe streams

gNMIc can ACK config/apply before every target re-subscribes; ApplyCache then
short-circuited the same plan for 5m and left paths=[] long enough to flake
Sub005/Sub010. Verify each hash with a second POST, and wait on fleet state
atomically in those tests.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 internal/controller/applycache.go             | 48 ++++++++--
 internal/controller/applycache_test.go        | 43 ++++++++-
 internal/controller/cluster_controller.go     | 37 +++++---
 .../004-subscriptions/subscriptions_test.go   | 92 +++++++++++++------
 4 files changed, 165 insertions(+), 55 deletions(-)

diff --git a/internal/controller/applycache.go b/internal/controller/applycache.go
index 6f5f861..23de37b 100644
--- a/internal/controller/applycache.go
+++ b/internal/controller/applycache.go
@@ -21,6 +21,12 @@ import (
 // to one apply per pod per interval.
 const applyRefreshInterval = 5 * time.Minute
 
+// applyVerifyDelay is how soon to re-POST the same payload after the first
+// successful apply. gNMIc's config/apply can ACK before every Subscribe stream
+// has re-established; without a second pass, ApplyCache would short-circuit
+// until applyRefreshInterval and leave targets with paths=[] for minutes.
+const applyVerifyDelay = 2 * time.Second
+
 // ApplyCache records what was last successfully applied to each collector pod,
 // so a reconcile that changes nothing does not re-POST the whole configuration
 // to every pod.
@@ -29,6 +35,10 @@ const applyRefreshInterval = 5 * time.Minute
 // it costs one redundant apply per pod, which is the safe direction to fail.
 // Persisting it would make a stale entry survive the restart that might have
 // been the only thing left to clear it.
+//
+// Each hash is applied twice before the short-circuit engages: the first POST
+// installs the plan, the second (after applyVerifyDelay) recovers targets that
+// missed the reconnect window. Only then is the record marked verified.
 type ApplyCache struct {
 	mu      sync.Mutex
 	entries map[string]applyRecord
@@ -38,8 +48,9 @@ type ApplyCache struct {
 }
 
 type applyRecord struct {
-	hash string
-	at   time.Time
+	hash     string
+	at       time.Time
+	verified bool
 }
 
 // NewApplyCache returns a cache using the default refresh interval.
@@ -54,8 +65,9 @@ func (c *ApplyCache) timeNow() time.Time {
 	return time.Now()
 }
 
-// Unchanged reports whether the pod already holds this exact configuration and
-// the record is still inside the refresh interval.
+// Unchanged reports whether the pod already holds this exact configuration,
+// the record has been verified by a second successful apply, and the record
+// is still inside the refresh interval.
 //
 // A nil cache always reports false, so a reconciler constructed without one
 // (tests, the API server) applies unconditionally rather than silently skipping.
@@ -66,22 +78,40 @@ func (c *ApplyCache) Unchanged(key, hash string) bool {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	rec, ok := c.entries[key]
-	if !ok || rec.hash != hash {
+	if !ok || rec.hash != hash || !rec.verified {
 		return false
 	}
 	return c.timeNow().Sub(rec.at) < c.ttl
 }
 
-// Record marks a configuration as successfully applied to a pod. It must only
-// be called after the POST succeeds: recording an attempt would make a failed
-// apply look applied until the plan changes again.
+// Record marks a configuration as successfully POSTed to a pod. The first
+// Record for a hash leaves the entry unverified so the next reconcile re-applies
+// once; the second Record of the same hash marks it verified and enables the
+// short-circuit. It must only be called after the POST succeeds.
 func (c *ApplyCache) Record(key, hash string) {
 	if c == nil {
 		return
 	}
 	c.mu.Lock()
 	defer c.mu.Unlock()
-	c.entries[key] = applyRecord{hash: hash, at: c.timeNow()}
+	now := c.timeNow()
+	if rec, ok := c.entries[key]; ok && rec.hash == hash && !rec.verified {
+		c.entries[key] = applyRecord{hash: hash, at: now, verified: true}
+		return
+	}
+	c.entries[key] = applyRecord{hash: hash, at: now, verified: false}
+}
+
+// NeedsVerify reports whether key holds an unverified record for hash, so the
+// reconciler should requeue for the follow-up apply.
+func (c *ApplyCache) NeedsVerify(key, hash string) bool {
+	if c == nil {
+		return false
+	}
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	rec, ok := c.entries[key]
+	return ok && rec.hash == hash && !rec.verified
 }
 
 // Invalidate drops one pod's record, forcing the next reconcile to re-apply to
diff --git a/internal/controller/applycache_test.go b/internal/controller/applycache_test.go
index da4ff4a..42b9f2b 100644
--- a/internal/controller/applycache_test.go
+++ b/internal/controller/applycache_test.go
@@ -20,8 +20,18 @@ func TestApplyCache_RecordThenUnchanged(t *testing.T) {
 		t.Fatal("empty cache reported unchanged; a pod we know nothing about must be applied to")
 	}
 	c.Record("ns/c1/0", "hashA")
+	if c.Unchanged("ns/c1/0", "hashA") {
+		t.Fatal("first record must stay unverified so a follow-up apply still runs")
+	}
+	if !c.NeedsVerify("ns/c1/0", "hashA") {
+		t.Fatal("first record should need verify")
+	}
+	c.Record("ns/c1/0", "hashA")
 	if !c.Unchanged("ns/c1/0", "hashA") {
-		t.Fatal("recorded hash reported changed")
+		t.Fatal("verified hash reported changed")
+	}
+	if c.NeedsVerify("ns/c1/0", "hashA") {
+		t.Fatal("verified record should not need verify")
 	}
 	if c.Unchanged("ns/c1/0", "hashB") {
 		t.Fatal("different hash reported unchanged")
@@ -37,6 +47,7 @@ func TestApplyCache_ExpiresAfterRefreshInterval(t *testing.T) {
 	now := time.Unix(1000, 0)
 	c := testCache(t, &now)
 	c.Record("ns/c1/0", "hashA")
+	c.Record("ns/c1/0", "hashA") // verify
 
 	now = now.Add(applyRefreshInterval - time.Second)
 	if !c.Unchanged("ns/c1/0", "hashA") {
@@ -55,6 +66,8 @@ func TestApplyCache_InvalidateIsPerPod(t *testing.T) {
 	now := time.Unix(1000, 0)
 	c := testCache(t, &now)
 	c.Record("ns/c1/0", "hashA")
+	c.Record("ns/c1/0", "hashA")
+	c.Record("ns/c1/1", "hashB")
 	c.Record("ns/c1/1", "hashB")
 
 	c.Invalidate("ns/c1/0")
@@ -69,10 +82,10 @@ func TestApplyCache_InvalidateIsPerPod(t *testing.T) {
 func TestApplyCache_InvalidateCluster(t *testing.T) {
 	now := time.Unix(1000, 0)
 	c := testCache(t, &now)
-	c.Record("ns/c1/0", "h")
-	c.Record("ns/c1/1", "h")
-	c.Record("ns/c10/0", "h") // prefix-adjacent, must survive
-	c.Record("ns/c2/0", "h")
+	for _, key := range []string{"ns/c1/0", "ns/c1/1", "ns/c10/0", "ns/c2/0"} {
+		c.Record(key, "h")
+		c.Record(key, "h")
+	}
 
 	c.InvalidateCluster("ns", "c1")
 	if c.Unchanged("ns/c1/0", "h") || c.Unchanged("ns/c1/1", "h") {
@@ -86,6 +99,23 @@ func TestApplyCache_InvalidateCluster(t *testing.T) {
 	}
 }
 
+func TestApplyCache_NewHashResetsVerification(t *testing.T) {
+	now := time.Unix(1000, 0)
+	c := testCache(t, &now)
+	c.Record("ns/c1/0", "hashA")
+	c.Record("ns/c1/0", "hashA")
+	if !c.Unchanged("ns/c1/0", "hashA") {
+		t.Fatal("setup: hashA should be verified")
+	}
+	c.Record("ns/c1/0", "hashB")
+	if c.Unchanged("ns/c1/0", "hashB") {
+		t.Fatal("new hash must not be verified after a single record")
+	}
+	if !c.NeedsVerify("ns/c1/0", "hashB") {
+		t.Fatal("new hash should need verify")
+	}
+}
+
 // A reconciler built without a cache must apply unconditionally rather than
 // silently skip, so the short-circuit can never be enabled by accident.
 func TestApplyCache_NilIsAlwaysChanged(t *testing.T) {
@@ -94,6 +124,9 @@ func TestApplyCache_NilIsAlwaysChanged(t *testing.T) {
 	if c.Unchanged("ns/c1/0", "hashA") {
 		t.Fatal("nil cache reported unchanged")
 	}
+	if c.NeedsVerify("ns/c1/0", "hashA") {
+		t.Fatal("nil cache reported needs verify")
+	}
 	c.Invalidate("ns/c1/0")
 	c.InvalidateCluster("ns", "c1")
 }
diff --git a/internal/controller/cluster_controller.go b/internal/controller/cluster_controller.go
index 26de113..25b0434 100644
--- a/internal/controller/cluster_controller.go
+++ b/internal/controller/cluster_controller.go
@@ -534,13 +534,15 @@ func (r *ClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 	configApplied := false
 	var configError error
 	var unassignedTargets int32
-	if unassigned, err := r.applyConfigToPods(ctx, &cluster, applyPlan, numPods); err != nil {
+	var verifyApply bool
+	if unassigned, needsVerify, err := r.applyConfigToPods(ctx, &cluster, applyPlan, numPods); err != nil {
 		logger.Error(err, "failed to apply config to gNMIc pods")
 		configError = err
 	} else {
 		configApplied = true
 		unassignedTargets = unassigned
-		logger.Info("successfully applied config to gNMIc cluster", "pods", numPods)
+		verifyApply = needsVerify
+		logger.Info("successfully applied config to gNMIc cluster", "pods", numPods, "verifyPending", needsVerify)
 	}
 
 	// calculate resource counts from pipelineDataMap
@@ -719,6 +721,11 @@ func (r *ClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 	if configApplied && len(pipelines) == 0 {
 		return ctrl.Result{RequeueAfter: time.Second}, nil
 	}
+	// First POST of a new plan is unverified: requeue so a second apply can
+	// recover targets that missed the Subscribe reconnect window.
+	if verifyApply {
+		return ctrl.Result{RequeueAfter: applyVerifyDelay}, nil
+	}
 
 	return ctrl.Result{}, nil
 }
@@ -750,7 +757,7 @@ func clusterStatusEqual(a, b gnmicv1alpha1.ClusterStatus) bool {
 
 // applyConfigToPods sends the apply plan to all gNMIc pods with distributed targets.
 // Returns the number of targets that could not be assigned due to capacity limits.
-func (r *ClusterReconciler) applyConfigToPods(ctx context.Context, cluster *gnmicv1alpha1.Cluster, plan *gnmic.ApplyPlan, numPods int) (int32, error) {
+func (r *ClusterReconciler) applyConfigToPods(ctx context.Context, cluster *gnmicv1alpha1.Cluster, plan *gnmic.ApplyPlan, numPods int) (unassigned int32, needsVerify bool, err error) {
 	logger := log.FromContext(ctx)
 
 	stsName := fmt.Sprintf("%s%s", resourcePrefix, cluster.Name)
@@ -762,7 +769,7 @@ func (r *ClusterReconciler) applyConfigToPods(ctx context.Context, cluster *gnmi
 	// create an HTTP client to send the apply plan to the gNMIc pods
 	httpClient, err := r.createHTTPClientForCluster(ctx, cluster)
 	if err != nil {
-		return 0, fmt.Errorf("failed to create HTTP client: %w", err)
+		return 0, false, fmt.Errorf("failed to create HTTP client: %w", err)
 	}
 	distResult := gnmic.DistributeTargets(plan, numPods, cluster.Spec.TargetDistribution)
 
@@ -796,7 +803,7 @@ func (r *ClusterReconciler) applyConfigToPods(ctx context.Context, cluster *gnmi
 		}
 		body, err := json.Marshal(podPlan)
 		if err != nil {
-			return 0, fmt.Errorf("failed to marshal apply plan for pod %d: %w", podIndex, err)
+			return 0, false, fmt.Errorf("failed to marshal apply plan for pod %d: %w", podIndex, err)
 		}
 		hash := fingerprint(body)
 		bodies[podIndex] = body
@@ -807,7 +814,7 @@ func (r *ClusterReconciler) applyConfigToPods(ctx context.Context, cluster *gnmi
 		}
 	}
 	if !anyChanged {
-		return int32(len(distResult.UnassignedTargets)), nil
+		return int32(len(distResult.UnassignedTargets)), false, nil
 	}
 
 	// Two-phase apply avoids double-collection when a target moves between pods.
@@ -848,7 +855,7 @@ func (r *ClusterReconciler) applyConfigToPods(ctx context.Context, cluster *gnmi
 			url := podURL(podIndex)
 			logger.Info("shrinking gNMIc pod targets before reassignment", "url", url, "targets", len(shrink.Targets))
 			if err := r.sendApplyRequest(ctx, url, shrink, httpClient); err != nil {
-				return 0, fmt.Errorf("failed to shrink config on pod %d: %w", podIndex, err)
+				return 0, false, fmt.Errorf("failed to shrink config on pod %d: %w", podIndex, err)
 			}
 		}
 		// gNMIc's config/apply returns before Subscribe streams are fully torn
@@ -857,7 +864,7 @@ func (r *ClusterReconciler) applyConfigToPods(ctx context.Context, cluster *gnmi
 		// double-collects.
 		select {
 		case <-ctx.Done():
-			return 0, ctx.Err()
+			return 0, false, ctx.Err()
 		case <-time.After(time.Second):
 		}
 	}
@@ -870,20 +877,26 @@ func (r *ClusterReconciler) applyConfigToPods(ctx context.Context, cluster *gnmi
 		url := podURL(podIndex)
 		logger.Info("sending config to gNMIc pod", "url", url)
 		if err := r.sendApplyBody(ctx, url, bodies[podIndex], httpClient); err != nil {
-			return 0, fmt.Errorf("failed to apply config to pod %d: %w", podIndex, err)
+			return 0, false, fmt.Errorf("failed to apply config to pod %d: %w", podIndex, err)
 		}
 		// Recorded only after the POST succeeds. Recording the attempt would
 		// make a failed apply look applied until the plan changes again.
-		r.Applied.Record(streamKey(cluster.Namespace, cluster.Name, podIndex), hashes[podIndex])
+		// The first Record leaves the entry unverified so the reconciler
+		// requeues for a follow-up POST (see NeedsVerify).
+		key := streamKey(cluster.Namespace, cluster.Name, podIndex)
+		r.Applied.Record(key, hashes[podIndex])
+		if r.Applied.NeedsVerify(key, hashes[podIndex]) {
+			needsVerify = true
+		}
 		logger.Info("config applied to pod", "pod", podIndex, "targets", len(podPlan.Targets))
 	}
 
-	unassigned := int32(len(distResult.UnassignedTargets))
+	unassigned = int32(len(distResult.UnassignedTargets))
 	if unassigned > 0 {
 		logger.Info("targets unassigned due to capacity limits", "count", unassigned)
 	}
 
-	return unassigned, nil
+	return unassigned, needsVerify, nil
 }
 
 func (r *ClusterReconciler) createHTTPClientForCluster(ctx context.Context, cluster *gnmicv1alpha1.Cluster) (*http.Client, error) {
diff --git a/test/integration/suite/004-subscriptions/subscriptions_test.go b/test/integration/suite/004-subscriptions/subscriptions_test.go
index 95d6738..f0c7895 100644
--- a/test/integration/suite/004-subscriptions/subscriptions_test.go
+++ b/test/integration/suite/004-subscriptions/subscriptions_test.go
@@ -158,9 +158,16 @@ func deleteSubscription(t *testing.T, name string) {
 
 func waitStreams(t *testing.T, want int) {
 	t.Helper()
-	for _, name := range allTargets {
-		s.GnmiGen.WaitStreams(t, name, want)
-	}
+	// One shared Medium budget across the fleet. Sequential per-target waits
+	// burn the full timeout on the first lagging target and hide siblings.
+	harness.Wait(t, harness.Medium, fmt.Sprintf("%d stream(s) on all targets", want), func() (bool, string) {
+		for _, name := range allTargets {
+			if n := s.GnmiGen.StreamCount(name); n != want {
+				return false, fmt.Sprintf("%s streams=%d", name, n)
+			}
+		}
+		return true, ""
+	})
 }
 
 func waitPathOnAll(t *testing.T, path string) {
@@ -202,23 +209,37 @@ func waitPathsOnAll(t *testing.T, want ...string) {
 // full Medium timeout on the second.
 func waitPathReplacedOnAll(t *testing.T, have, gone string) {
 	t.Helper()
-	harness.Wait(t, harness.Medium, fmt.Sprintf("path %s replaces %s on all targets", have, gone), func() (bool, string) {
+	waitExactPathsOnAll(t, []string{have}, []string{gone})
+}
+
+// waitExactPathsOnAll waits until every target has all want paths, none of the
+// gone paths, and exactly one stream — in a single snapshot per poll.
+func waitExactPathsOnAll(t *testing.T, want, gone []string) {
+	t.Helper()
+	harness.Wait(t, harness.Medium, fmt.Sprintf("paths want=%v gone=%v on all targets", want, gone), func() (bool, string) {
 		for _, name := range allTargets {
 			if n := s.GnmiGen.StreamCount(name); n != 1 {
 				return false, fmt.Sprintf("%s streams=%d paths=%v", name, n, s.GnmiGen.Paths(name))
 			}
 			paths := s.GnmiGen.Paths(name)
-			hasHave, hasGone := false, false
-			for _, p := range paths {
-				if p == have {
-					hasHave = true
+			for _, w := range want {
+				found := false
+				for _, p := range paths {
+					if p == w {
+						found = true
+						break
+					}
 				}
-				if p == gone {
-					hasGone = true
+				if !found {
+					return false, fmt.Sprintf("%s paths=%v", name, paths)
 				}
 			}
-			if !hasHave || hasGone {
-				return false, fmt.Sprintf("%s paths=%v", name, paths)
+			for _, g := range gone {
+				for _, p := range paths {
+					if p == g {
+						return false, fmt.Sprintf("%s still has %s: %v", name, g, paths)
+					}
+				}
 			}
 		}
 		return true, ""
@@ -303,6 +324,25 @@ func waitUpdatesOnly(t *testing.T, target string, want bool) {
 	})
 }
 
+func waitUpdatesOnlyOnAll(t *testing.T, want bool) {
+	t.Helper()
+	harness.Wait(t, harness.Medium, fmt.Sprintf("updates_only=%v on all targets", want), func() (bool, string) {
+		for _, name := range allTargets {
+			subs, err := s.GnmiGen.Subscriptions(name)
+			if err != nil {
+				return false, err.Error()
+			}
+			if len(subs) != 1 {
+				return false, fmt.Sprintf("%s streams=%d", name, len(subs))
+			}
+			if subs[0].UpdatesOnly != want {
+				return false, fmt.Sprintf("%s updates_only=%v", name, subs[0].UpdatesOnly)
+			}
+		}
+		return true, ""
+	})
+}
+
 func waitSuppressAndHeartbeat(t *testing.T, target string, suppress bool, heartbeat string) {
 	t.Helper()
 	harness.Wait(t, harness.Medium, fmt.Sprintf("suppress=%v heartbeat=%s on %s", suppress, heartbeat, target), func() (bool, string) {
@@ -439,22 +479,17 @@ func TestSub005_UpdatesOnlyReachesWire(t *testing.T) {
 		"UpdatesOnly": false,
 	})
 	waitStreams(t, 1)
-	waitUpdatesOnly(t, leaf1, false)
+	waitUpdatesOnlyOnAll(t, false)
 	syncBefore := firstStream(t, leaf1).SyncMessagesSent
 
 	restartsBefore := s.K8s.RestartCounts(t, cluster)
 	patchSubscription(t, "s1", map[string]any{"updatesOnly": true})
-	waitUpdatesOnly(t, leaf1, true)
-
-	// updatesOnly re-establishes the stream; sync pattern should differ from
-	// the initial full sync (typically sync_messages_sent stays 0 or much lower).
-	harness.Wait(t, harness.Medium, "updates_only stream settled", func() (bool, string) {
-		sub := firstStream(t, leaf1)
-		if !sub.UpdatesOnly {
-			return false, "updates_only still false"
-		}
-		return true, ""
-	})
+	harness.WaitConfigApplied(t, s.K8s, cluster)
+	// Assert the flag on every target in one snapshot. Waiting only on leaf1
+	// then waitStreams(leaf2) races a mid-reconnect gap where leaf2 has
+	// streams=0 after the apply ACK (ApplyCache verify re-POST recovers it).
+	waitUpdatesOnlyOnAll(t, true)
+
 	syncAfter := firstStream(t, leaf1).SyncMessagesSent
 	if syncBefore == 0 && syncAfter == 0 {
 		t.Log("sync_messages_sent was 0 both before and after; flag still asserted on wire")
@@ -462,7 +497,6 @@ func TestSub005_UpdatesOnlyReachesWire(t *testing.T) {
 		// Soft signal only — gnmi-gen may still emit a sync response.
 		t.Logf("sync_messages_sent before=%d after=%d", syncBefore, syncAfter)
 	}
-	waitStreams(t, 1)
 	harness.AssertNoRestarts(t, restartsBefore, s.K8s.RestartCounts(t, cluster))
 }
 
@@ -570,11 +604,11 @@ func TestSub010_RapidSuccessiveEditsConverge(t *testing.T) {
 	for _, p := range paths {
 		patchSubscription(t, "s1", map[string]any{"paths": p})
 	}
+	harness.WaitConfigApplied(t, s.K8s, cluster)
 
-	waitPathOnAll(t, pathIF)
-	waitPathAbsentOnAll(t, pathCPU)
-	waitPathAbsentOnAll(t, pathSys)
-	waitStreams(t, 1)
+	// Final state in one snapshot: pathIF present, prior paths gone, one stream.
+	// Split present/absent waits race reconnect gaps after the patch burst.
+	waitExactPathsOnAll(t, []string{pathIF}, []string{pathCPU, pathSys})
 	s.GnmiGen.ConsistentlyCollectedOnce(t, 15*time.Second, 1, allTargets...)
 	harness.AssertNoRestarts(t, restartsBefore, s.K8s.RestartCounts(t, cluster))
 	harness.AssertNoPanics(t, s.K8s.OperatorLogs(t, time.Since(since)+time.Minute))

From 71a12d566bdf6701e94d641453161dadd7ef5d09 Mon Sep 17 00:00:00 2001
From: Karim Radhouani <medkarimrdi@gmail.com>
Date: Tue, 11 Aug 2026 23:28:48 -0700
Subject: [PATCH 3/5] refactor: simplify ApplyCache logic and improve test
 coverage

---
 internal/controller/applycache.go         | 48 +++++------------------
 internal/controller/applycache_test.go    | 43 +++-----------------
 internal/controller/cluster_controller.go | 37 ++++++-----------
 3 files changed, 26 insertions(+), 102 deletions(-)

diff --git a/internal/controller/applycache.go b/internal/controller/applycache.go
index 23de37b..6f5f861 100644
--- a/internal/controller/applycache.go
+++ b/internal/controller/applycache.go
@@ -21,12 +21,6 @@ import (
 // to one apply per pod per interval.
 const applyRefreshInterval = 5 * time.Minute
 
-// applyVerifyDelay is how soon to re-POST the same payload after the first
-// successful apply. gNMIc's config/apply can ACK before every Subscribe stream
-// has re-established; without a second pass, ApplyCache would short-circuit
-// until applyRefreshInterval and leave targets with paths=[] for minutes.
-const applyVerifyDelay = 2 * time.Second
-
 // ApplyCache records what was last successfully applied to each collector pod,
 // so a reconcile that changes nothing does not re-POST the whole configuration
 // to every pod.
@@ -35,10 +29,6 @@ const applyVerifyDelay = 2 * time.Second
 // it costs one redundant apply per pod, which is the safe direction to fail.
 // Persisting it would make a stale entry survive the restart that might have
 // been the only thing left to clear it.
-//
-// Each hash is applied twice before the short-circuit engages: the first POST
-// installs the plan, the second (after applyVerifyDelay) recovers targets that
-// missed the reconnect window. Only then is the record marked verified.
 type ApplyCache struct {
 	mu      sync.Mutex
 	entries map[string]applyRecord
@@ -48,9 +38,8 @@ type ApplyCache struct {
 }
 
 type applyRecord struct {
-	hash     string
-	at       time.Time
-	verified bool
+	hash string
+	at   time.Time
 }
 
 // NewApplyCache returns a cache using the default refresh interval.
@@ -65,9 +54,8 @@ func (c *ApplyCache) timeNow() time.Time {
 	return time.Now()
 }
 
-// Unchanged reports whether the pod already holds this exact configuration,
-// the record has been verified by a second successful apply, and the record
-// is still inside the refresh interval.
+// Unchanged reports whether the pod already holds this exact configuration and
+// the record is still inside the refresh interval.
 //
 // A nil cache always reports false, so a reconciler constructed without one
 // (tests, the API server) applies unconditionally rather than silently skipping.
@@ -78,40 +66,22 @@ func (c *ApplyCache) Unchanged(key, hash string) bool {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	rec, ok := c.entries[key]
-	if !ok || rec.hash != hash || !rec.verified {
+	if !ok || rec.hash != hash {
 		return false
 	}
 	return c.timeNow().Sub(rec.at) < c.ttl
 }
 
-// Record marks a configuration as successfully POSTed to a pod. The first
-// Record for a hash leaves the entry unverified so the next reconcile re-applies
-// once; the second Record of the same hash marks it verified and enables the
-// short-circuit. It must only be called after the POST succeeds.
+// Record marks a configuration as successfully applied to a pod. It must only
+// be called after the POST succeeds: recording an attempt would make a failed
+// apply look applied until the plan changes again.
 func (c *ApplyCache) Record(key, hash string) {
 	if c == nil {
 		return
 	}
 	c.mu.Lock()
 	defer c.mu.Unlock()
-	now := c.timeNow()
-	if rec, ok := c.entries[key]; ok && rec.hash == hash && !rec.verified {
-		c.entries[key] = applyRecord{hash: hash, at: now, verified: true}
-		return
-	}
-	c.entries[key] = applyRecord{hash: hash, at: now, verified: false}
-}
-
-// NeedsVerify reports whether key holds an unverified record for hash, so the
-// reconciler should requeue for the follow-up apply.
-func (c *ApplyCache) NeedsVerify(key, hash string) bool {
-	if c == nil {
-		return false
-	}
-	c.mu.Lock()
-	defer c.mu.Unlock()
-	rec, ok := c.entries[key]
-	return ok && rec.hash == hash && !rec.verified
+	c.entries[key] = applyRecord{hash: hash, at: c.timeNow()}
 }
 
 // Invalidate drops one pod's record, forcing the next reconcile to re-apply to
diff --git a/internal/controller/applycache_test.go b/internal/controller/applycache_test.go
index 42b9f2b..da4ff4a 100644
--- a/internal/controller/applycache_test.go
+++ b/internal/controller/applycache_test.go
@@ -20,18 +20,8 @@ func TestApplyCache_RecordThenUnchanged(t *testing.T) {
 		t.Fatal("empty cache reported unchanged; a pod we know nothing about must be applied to")
 	}
 	c.Record("ns/c1/0", "hashA")
-	if c.Unchanged("ns/c1/0", "hashA") {
-		t.Fatal("first record must stay unverified so a follow-up apply still runs")
-	}
-	if !c.NeedsVerify("ns/c1/0", "hashA") {
-		t.Fatal("first record should need verify")
-	}
-	c.Record("ns/c1/0", "hashA")
 	if !c.Unchanged("ns/c1/0", "hashA") {
-		t.Fatal("verified hash reported changed")
-	}
-	if c.NeedsVerify("ns/c1/0", "hashA") {
-		t.Fatal("verified record should not need verify")
+		t.Fatal("recorded hash reported changed")
 	}
 	if c.Unchanged("ns/c1/0", "hashB") {
 		t.Fatal("different hash reported unchanged")
@@ -47,7 +37,6 @@ func TestApplyCache_ExpiresAfterRefreshInterval(t *testing.T) {
 	now := time.Unix(1000, 0)
 	c := testCache(t, &now)
 	c.Record("ns/c1/0", "hashA")
-	c.Record("ns/c1/0", "hashA") // verify
 
 	now = now.Add(applyRefreshInterval - time.Second)
 	if !c.Unchanged("ns/c1/0", "hashA") {
@@ -66,8 +55,6 @@ func TestApplyCache_InvalidateIsPerPod(t *testing.T) {
 	now := time.Unix(1000, 0)
 	c := testCache(t, &now)
 	c.Record("ns/c1/0", "hashA")
-	c.Record("ns/c1/0", "hashA")
-	c.Record("ns/c1/1", "hashB")
 	c.Record("ns/c1/1", "hashB")
 
 	c.Invalidate("ns/c1/0")
@@ -82,10 +69,10 @@ func TestApplyCache_InvalidateIsPerPod(t *testing.T) {
 func TestApplyCache_InvalidateCluster(t *testing.T) {
 	now := time.Unix(1000, 0)
 	c := testCache(t, &now)
-	for _, key := range []string{"ns/c1/0", "ns/c1/1", "ns/c10/0", "ns/c2/0"} {
-		c.Record(key, "h")
-		c.Record(key, "h")
-	}
+	c.Record("ns/c1/0", "h")
+	c.Record("ns/c1/1", "h")
+	c.Record("ns/c10/0", "h") // prefix-adjacent, must survive
+	c.Record("ns/c2/0", "h")
 
 	c.InvalidateCluster("ns", "c1")
 	if c.Unchanged("ns/c1/0", "h") || c.Unchanged("ns/c1/1", "h") {
@@ -99,23 +86,6 @@ func TestApplyCache_InvalidateCluster(t *testing.T) {
 	}
 }
 
-func TestApplyCache_NewHashResetsVerification(t *testing.T) {
-	now := time.Unix(1000, 0)
-	c := testCache(t, &now)
-	c.Record("ns/c1/0", "hashA")
-	c.Record("ns/c1/0", "hashA")
-	if !c.Unchanged("ns/c1/0", "hashA") {
-		t.Fatal("setup: hashA should be verified")
-	}
-	c.Record("ns/c1/0", "hashB")
-	if c.Unchanged("ns/c1/0", "hashB") {
-		t.Fatal("new hash must not be verified after a single record")
-	}
-	if !c.NeedsVerify("ns/c1/0", "hashB") {
-		t.Fatal("new hash should need verify")
-	}
-}
-
 // A reconciler built without a cache must apply unconditionally rather than
 // silently skip, so the short-circuit can never be enabled by accident.
 func TestApplyCache_NilIsAlwaysChanged(t *testing.T) {
@@ -124,9 +94,6 @@ func TestApplyCache_NilIsAlwaysChanged(t *testing.T) {
 	if c.Unchanged("ns/c1/0", "hashA") {
 		t.Fatal("nil cache reported unchanged")
 	}
-	if c.NeedsVerify("ns/c1/0", "hashA") {
-		t.Fatal("nil cache reported needs verify")
-	}
 	c.Invalidate("ns/c1/0")
 	c.InvalidateCluster("ns", "c1")
 }
diff --git a/internal/controller/cluster_controller.go b/internal/controller/cluster_controller.go
index 25b0434..26de113 100644
--- a/internal/controller/cluster_controller.go
+++ b/internal/controller/cluster_controller.go
@@ -534,15 +534,13 @@ func (r *ClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 	configApplied := false
 	var configError error
 	var unassignedTargets int32
-	var verifyApply bool
-	if unassigned, needsVerify, err := r.applyConfigToPods(ctx, &cluster, applyPlan, numPods); err != nil {
+	if unassigned, err := r.applyConfigToPods(ctx, &cluster, applyPlan, numPods); err != nil {
 		logger.Error(err, "failed to apply config to gNMIc pods")
 		configError = err
 	} else {
 		configApplied = true
 		unassignedTargets = unassigned
-		verifyApply = needsVerify
-		logger.Info("successfully applied config to gNMIc cluster", "pods", numPods, "verifyPending", needsVerify)
+		logger.Info("successfully applied config to gNMIc cluster", "pods", numPods)
 	}
 
 	// calculate resource counts from pipelineDataMap
@@ -721,11 +719,6 @@ func (r *ClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 	if configApplied && len(pipelines) == 0 {
 		return ctrl.Result{RequeueAfter: time.Second}, nil
 	}
-	// First POST of a new plan is unverified: requeue so a second apply can
-	// recover targets that missed the Subscribe reconnect window.
-	if verifyApply {
-		return ctrl.Result{RequeueAfter: applyVerifyDelay}, nil
-	}
 
 	return ctrl.Result{}, nil
 }
@@ -757,7 +750,7 @@ func clusterStatusEqual(a, b gnmicv1alpha1.ClusterStatus) bool {
 
 // applyConfigToPods sends the apply plan to all gNMIc pods with distributed targets.
 // Returns the number of targets that could not be assigned due to capacity limits.
-func (r *ClusterReconciler) applyConfigToPods(ctx context.Context, cluster *gnmicv1alpha1.Cluster, plan *gnmic.ApplyPlan, numPods int) (unassigned int32, needsVerify bool, err error) {
+func (r *ClusterReconciler) applyConfigToPods(ctx context.Context, cluster *gnmicv1alpha1.Cluster, plan *gnmic.ApplyPlan, numPods int) (int32, error) {
 	logger := log.FromContext(ctx)
 
 	stsName := fmt.Sprintf("%s%s", resourcePrefix, cluster.Name)
@@ -769,7 +762,7 @@ func (r *ClusterReconciler) applyConfigToPods(ctx context.Context, cluster *gnmi
 	// create an HTTP client to send the apply plan to the gNMIc pods
 	httpClient, err := r.createHTTPClientForCluster(ctx, cluster)
 	if err != nil {
-		return 0, false, fmt.Errorf("failed to create HTTP client: %w", err)
+		return 0, fmt.Errorf("failed to create HTTP client: %w", err)
 	}
 	distResult := gnmic.DistributeTargets(plan, numPods, cluster.Spec.TargetDistribution)
 
@@ -803,7 +796,7 @@ func (r *ClusterReconciler) applyConfigToPods(ctx context.Context, cluster *gnmi
 		}
 		body, err := json.Marshal(podPlan)
 		if err != nil {
-			return 0, false, fmt.Errorf("failed to marshal apply plan for pod %d: %w", podIndex, err)
+			return 0, fmt.Errorf("failed to marshal apply plan for pod %d: %w", podIndex, err)
 		}
 		hash := fingerprint(body)
 		bodies[podIndex] = body
@@ -814,7 +807,7 @@ func (r *ClusterReconciler) applyConfigToPods(ctx context.Context, cluster *gnmi
 		}
 	}
 	if !anyChanged {
-		return int32(len(distResult.UnassignedTargets)), false, nil
+		return int32(len(distResult.UnassignedTargets)), nil
 	}
 
 	// Two-phase apply avoids double-collection when a target moves between pods.
@@ -855,7 +848,7 @@ func (r *ClusterReconciler) applyConfigToPods(ctx context.Context, cluster *gnmi
 			url := podURL(podIndex)
 			logger.Info("shrinking gNMIc pod targets before reassignment", "url", url, "targets", len(shrink.Targets))
 			if err := r.sendApplyRequest(ctx, url, shrink, httpClient); err != nil {
-				return 0, false, fmt.Errorf("failed to shrink config on pod %d: %w", podIndex, err)
+				return 0, fmt.Errorf("failed to shrink config on pod %d: %w", podIndex, err)
 			}
 		}
 		// gNMIc's config/apply returns before Subscribe streams are fully torn
@@ -864,7 +857,7 @@ func (r *ClusterReconciler) applyConfigToPods(ctx context.Context, cluster *gnmi
 		// double-collects.
 		select {
 		case <-ctx.Done():
-			return 0, false, ctx.Err()
+			return 0, ctx.Err()
 		case <-time.After(time.Second):
 		}
 	}
@@ -877,26 +870,20 @@ func (r *ClusterReconciler) applyConfigToPods(ctx context.Context, cluster *gnmi
 		url := podURL(podIndex)
 		logger.Info("sending config to gNMIc pod", "url", url)
 		if err := r.sendApplyBody(ctx, url, bodies[podIndex], httpClient); err != nil {
-			return 0, false, fmt.Errorf("failed to apply config to pod %d: %w", podIndex, err)
+			return 0, fmt.Errorf("failed to apply config to pod %d: %w", podIndex, err)
 		}
 		// Recorded only after the POST succeeds. Recording the attempt would
 		// make a failed apply look applied until the plan changes again.
-		// The first Record leaves the entry unverified so the reconciler
-		// requeues for a follow-up POST (see NeedsVerify).
-		key := streamKey(cluster.Namespace, cluster.Name, podIndex)
-		r.Applied.Record(key, hashes[podIndex])
-		if r.Applied.NeedsVerify(key, hashes[podIndex]) {
-			needsVerify = true
-		}
+		r.Applied.Record(streamKey(cluster.Namespace, cluster.Name, podIndex), hashes[podIndex])
 		logger.Info("config applied to pod", "pod", podIndex, "targets", len(podPlan.Targets))
 	}
 
-	unassigned = int32(len(distResult.UnassignedTargets))
+	unassigned := int32(len(distResult.UnassignedTargets))
 	if unassigned > 0 {
 		logger.Info("targets unassigned due to capacity limits", "count", unassigned)
 	}
 
-	return unassigned, needsVerify, nil
+	return unassigned, nil
 }
 
 func (r *ClusterReconciler) createHTTPClientForCluster(ctx context.Context, cluster *gnmicv1alpha1.Cluster) (*http.Client, error) {

From 5505b38cf6a1842d3731d191a9801ab91c461011 Mon Sep 17 00:00:00 2001
From: Karim Radhouani <medkarimrdi@gmail.com>
Date: Wed, 12 Aug 2026 08:27:31 -0700
Subject: [PATCH 4/5] fix: settle streams/service before asserting they hold
 steady

TestOut006, TestTarget003, and TestOut003 sampled their invariant right
after triggering a config apply or waiting on Cluster status, both of
which can return before the actual gNMIc streams/Service exist. Wait for
the settled state first so the assertion window doesn't start mid-reconnect.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 test/integration/suite/003-targets/targets_test.go | 4 ++++
 test/integration/suite/005-outputs/outputs_test.go | 8 +++++++-
 2 files changed, 11 insertions(+), 1 deletion(-)

diff --git a/test/integration/suite/003-targets/targets_test.go b/test/integration/suite/003-targets/targets_test.go
index 363dfba..31e4d7d 100644
--- a/test/integration/suite/003-targets/targets_test.go
+++ b/test/integration/suite/003-targets/targets_test.go
@@ -241,6 +241,10 @@ spec:
 		return true, ""
 	})
 	s.GnmiGen.WaitStreams(t, leaf2, 0)
+	// Adding leaf2 re-applied the whole pod's config, briefly reloading
+	// leaf1's stream too; settle before proving it holds, or the window can
+	// start mid-reconnect and flake on a transient 0.
+	s.GnmiGen.WaitStreams(t, leaf1, 1)
 	s.GnmiGen.ConsistentlyCollectedOnce(t, 5*time.Second, 1, leaf1)
 
 	// Replace the Secret via YAML apply (avoids managedFields on a reused object),
diff --git a/test/integration/suite/005-outputs/outputs_test.go b/test/integration/suite/005-outputs/outputs_test.go
index f880252..99b532d 100644
--- a/test/integration/suite/005-outputs/outputs_test.go
+++ b/test/integration/suite/005-outputs/outputs_test.go
@@ -223,7 +223,9 @@ func TestOut003_ServiceTypeHonored(t *testing.T) {
 	applyPipeline(t, "p1", []string{"out1"}, "")
 	waitClusterReady(t)
 
-	svc := s.K8s.Service(t, harness.PromServiceName(cluster, "p1", "out1"))
+	// Cluster Ready can precede the per-pipeline Service reconcile; wait for
+	// the Service itself rather than racing it.
+	svc := waitPromService(t, "p1", "out1")
 	if svc.Spec.Type != corev1.ServiceTypeNodePort {
 		t.Fatalf("type=%s want NodePort", svc.Spec.Type)
 	}
@@ -336,6 +338,10 @@ func TestOut006_DeletingOutputRemovesService(t *testing.T) {
 	harness.WaitClusterCounts(t, s.K8s, cluster, harness.ClusterCounts{
 		OutputsCount: harness.I32(0),
 	})
+	// The delete's config/apply briefly reloads every stream on the pod;
+	// settle to 1-per-target before proving it holds, or the window can start
+	// mid-reconnect and flake on a transient 0.
+	s.GnmiGen.WaitFleetStreams(t, harness.Medium, 1, allTargets)
 	s.GnmiGen.ConsistentlyCollectedOnce(t, 10*time.Second, 1, allTargets...)
 	harness.AssertNoRestarts(t, restartsBefore, s.K8s.RestartCounts(t, cluster))
 }

From f3192b2e081cee00b146b2c68491e12a41a98959 Mon Sep 17 00:00:00 2001
From: Karim Radhouani <medkarimrdi@gmail.com>
Date: Wed, 12 Aug 2026 10:55:06 -0700
Subject: [PATCH 5/5] fix: safety-net re-POST for streams that never
 re-establish after apply

gNMIc's config/apply can ACK before every Subscribe stream has
re-established, and rarely one never comes back on its own, leaving a
target stuck with paths=[] indefinitely. Track whether an applied hash
has been confirmed by a later reconcile; if it's still unverified after
a 30s grace period, requeue and re-POST the same config once to kick
the stream loose.

The delay matters: firing the repost immediately (an earlier attempt)
doubled every apply back-to-back and made pods work harder to
reconnect right when they were already struggling, making CI flakes
worse instead of better. 30s clears both unrelated reconcile churn and
tests' own "stays stable" assertion windows.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 internal/controller/applycache.go         | 61 +++++++++++++++++---
 internal/controller/applycache_test.go    | 70 ++++++++++++++++++++++-
 internal/controller/cluster_controller.go | 54 ++++++++++++-----
 3 files changed, 162 insertions(+), 23 deletions(-)

diff --git a/internal/controller/applycache.go b/internal/controller/applycache.go
index 6f5f861..319868a 100644
--- a/internal/controller/applycache.go
+++ b/internal/controller/applycache.go
@@ -21,6 +21,19 @@ import (
 // to one apply per pod per interval.
 const applyRefreshInterval = 5 * time.Minute
 
+// applyVerifyDelay is the grace period before a safety-net re-POST of an
+// unverified hash. gNMIc's config/apply can ACK before every Subscribe stream
+// has re-established, and occasionally one never comes back on its own; a
+// second POST of the same config reliably kicks it loose. The delay has to
+// clear two things: every unrelated reconcile trigger that fires in the
+// meantime (Target status, Secret changes, periodic resync — all check
+// Unchanged and must not treat a still-settling apply as stale), and every
+// test's own tight "stays stable" window, so the safety net never lands mid
+// assertion. Firing it immediately (as a first attempt did) instead doubled
+// every apply back-to-back and made pods work harder to reconnect right when
+// they were already struggling, which was worse than the bug it targeted.
+const applyVerifyDelay = 30 * time.Second
+
 // ApplyCache records what was last successfully applied to each collector pod,
 // so a reconcile that changes nothing does not re-POST the whole configuration
 // to every pod.
@@ -38,8 +51,9 @@ type ApplyCache struct {
 }
 
 type applyRecord struct {
-	hash string
-	at   time.Time
+	hash     string
+	at       time.Time
+	verified bool
 }
 
 // NewApplyCache returns a cache using the default refresh interval.
@@ -55,7 +69,14 @@ func (c *ApplyCache) timeNow() time.Time {
 }
 
 // Unchanged reports whether the pod already holds this exact configuration and
-// the record is still inside the refresh interval.
+// no apply is due right now.
+//
+// An unverified hash still reports unchanged until applyVerifyDelay elapses —
+// otherwise every unrelated reconcile in that window would re-POST the same
+// config, compounding the very reconnect problem the verify pass exists to
+// fix. Once the grace period passes with no newer hash recorded, the next
+// reconcile (organic or the deliberate requeue below) is due its one
+// safety-net repost.
 //
 // A nil cache always reports false, so a reconciler constructed without one
 // (tests, the API server) applies unconditionally rather than silently skipping.
@@ -69,19 +90,43 @@ func (c *ApplyCache) Unchanged(key, hash string) bool {
 	if !ok || rec.hash != hash {
 		return false
 	}
-	return c.timeNow().Sub(rec.at) < c.ttl
+	age := c.timeNow().Sub(rec.at)
+	if !rec.verified && age >= applyVerifyDelay {
+		return false
+	}
+	return age < c.ttl
 }
 
-// Record marks a configuration as successfully applied to a pod. It must only
-// be called after the POST succeeds: recording an attempt would make a failed
-// apply look applied until the plan changes again.
+// Record marks a configuration as successfully POSTed to a pod. The first
+// Record for a hash leaves the entry unverified; if Unchanged later reports
+// changed because the grace period elapsed, the resulting repost's Record
+// call marks it verified so the short-circuit engages. It must only be called
+// after the POST succeeds: recording an attempt would make a failed apply
+// look applied until the plan changes again.
 func (c *ApplyCache) Record(key, hash string) {
 	if c == nil {
 		return
 	}
 	c.mu.Lock()
 	defer c.mu.Unlock()
-	c.entries[key] = applyRecord{hash: hash, at: c.timeNow()}
+	now := c.timeNow()
+	if rec, ok := c.entries[key]; ok && rec.hash == hash && !rec.verified {
+		c.entries[key] = applyRecord{hash: hash, at: now, verified: true}
+		return
+	}
+	c.entries[key] = applyRecord{hash: hash, at: now, verified: false}
+}
+
+// NeedsVerify reports whether key holds an unverified record for hash, so the
+// reconciler should requeue after applyVerifyDelay for the follow-up POST.
+func (c *ApplyCache) NeedsVerify(key, hash string) bool {
+	if c == nil {
+		return false
+	}
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	rec, ok := c.entries[key]
+	return ok && rec.hash == hash && !rec.verified
 }
 
 // Invalidate drops one pod's record, forcing the next reconcile to re-apply to
diff --git a/internal/controller/applycache_test.go b/internal/controller/applycache_test.go
index da4ff4a..d7420b9 100644
--- a/internal/controller/applycache_test.go
+++ b/internal/controller/applycache_test.go
@@ -32,11 +32,14 @@ func TestApplyCache_RecordThenUnchanged(t *testing.T) {
 }
 
 // The TTL is the backstop for a pod losing its config without dropping its SSE
-// stream, which nothing else can detect.
+// stream, which nothing else can detect. It is measured from verification,
+// since an unverified hash is on its own, shorter clock (applyVerifyDelay).
 func TestApplyCache_ExpiresAfterRefreshInterval(t *testing.T) {
 	now := time.Unix(1000, 0)
 	c := testCache(t, &now)
 	c.Record("ns/c1/0", "hashA")
+	now = now.Add(applyVerifyDelay)
+	c.Record("ns/c1/0", "hashA") // simulate the safety-net repost verifying it
 
 	now = now.Add(applyRefreshInterval - time.Second)
 	if !c.Unchanged("ns/c1/0", "hashA") {
@@ -98,6 +101,71 @@ func TestApplyCache_NilIsAlwaysChanged(t *testing.T) {
 	c.InvalidateCluster("ns", "c1")
 }
 
+// A freshly applied hash is unverified but must still read as unchanged
+// during the grace period: an unrelated reconcile arriving in that window
+// (e.g. a Target status update) must not trigger a redundant repost.
+func TestApplyCache_UnverifiedUnchangedDuringGrace(t *testing.T) {
+	now := time.Unix(1000, 0)
+	c := testCache(t, &now)
+	c.Record("ns/c1/0", "hashA")
+
+	now = now.Add(applyVerifyDelay - time.Second)
+	if !c.Unchanged("ns/c1/0", "hashA") {
+		t.Fatal("unverified hash reported changed before the grace period elapsed")
+	}
+	if !c.NeedsVerify("ns/c1/0", "hashA") {
+		t.Fatal("NeedsVerify should stay true until the follow-up Record")
+	}
+}
+
+// Once the grace period elapses with no newer hash recorded, the safety-net
+// repost is due: Unchanged must report changed so the reconciler re-POSTs.
+func TestApplyCache_UnverifiedNeedsRepostAfterGrace(t *testing.T) {
+	now := time.Unix(1000, 0)
+	c := testCache(t, &now)
+	c.Record("ns/c1/0", "hashA")
+
+	now = now.Add(applyVerifyDelay)
+	if c.Unchanged("ns/c1/0", "hashA") {
+		t.Fatal("unverified hash still reported unchanged after the grace period")
+	}
+}
+
+// The repost's Record call marks the hash verified, so it settles for good
+// (until the TTL) instead of triggering yet another repost.
+func TestApplyCache_SecondRecordVerifies(t *testing.T) {
+	now := time.Unix(1000, 0)
+	c := testCache(t, &now)
+	c.Record("ns/c1/0", "hashA")
+
+	now = now.Add(applyVerifyDelay)
+	c.Record("ns/c1/0", "hashA")
+	if c.NeedsVerify("ns/c1/0", "hashA") {
+		t.Fatal("hash still reported as needing verify after the follow-up Record")
+	}
+	now = now.Add(applyVerifyDelay)
+	if !c.Unchanged("ns/c1/0", "hashA") {
+		t.Fatal("verified hash reported changed")
+	}
+}
+
+// A real edit before the grace period elapses must reset verification for
+// the new hash rather than leave a stale repost pending for the old one.
+func TestApplyCache_NewHashResetsVerification(t *testing.T) {
+	now := time.Unix(1000, 0)
+	c := testCache(t, &now)
+	c.Record("ns/c1/0", "hashA")
+
+	now = now.Add(time.Second)
+	c.Record("ns/c1/0", "hashB")
+	if c.Unchanged("ns/c1/0", "hashA") {
+		t.Fatal("superseded hash still reported unchanged")
+	}
+	if !c.NeedsVerify("ns/c1/0", "hashB") {
+		t.Fatal("new hash should start its own unverified grace period")
+	}
+}
+
 // The fingerprint covers the bytes on the wire, so equal payloads compare equal
 // and any difference in what is sent is a difference here.
 func TestFingerprint(t *testing.T) {
diff --git a/internal/controller/cluster_controller.go b/internal/controller/cluster_controller.go
index 26de113..e74ca1f 100644
--- a/internal/controller/cluster_controller.go
+++ b/internal/controller/cluster_controller.go
@@ -532,14 +532,16 @@ func (r *ClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 	// distrubute to desired replicas only, this makes redistribution fast in case of scaling down.
 	numPods := int(desiredReplicas)
 	configApplied := false
+	verifyPending := false
 	var configError error
 	var unassignedTargets int32
-	if unassigned, err := r.applyConfigToPods(ctx, &cluster, applyPlan, numPods); err != nil {
+	if unassigned, needsVerify, err := r.applyConfigToPods(ctx, &cluster, applyPlan, numPods); err != nil {
 		logger.Error(err, "failed to apply config to gNMIc pods")
 		configError = err
 	} else {
 		configApplied = true
 		unassignedTargets = unassigned
+		verifyPending = needsVerify
 		logger.Info("successfully applied config to gNMIc cluster", "pods", numPods)
 	}
 
@@ -719,6 +721,13 @@ func (r *ClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 	if configApplied && len(pipelines) == 0 {
 		return ctrl.Result{RequeueAfter: time.Second}, nil
 	}
+	// A pod is holding an unverified hash: requeue for the safety-net repost
+	// once the grace period elapses (see applyVerifyDelay). If something else
+	// reconciles first with a newer plan, this pass is superseded harmlessly.
+	if verifyPending {
+		logger.Info("scheduling apply verify requeue", "cluster", cluster.Name, "namespace", cluster.Namespace, "after", applyVerifyDelay)
+		return ctrl.Result{RequeueAfter: applyVerifyDelay}, nil
+	}
 
 	return ctrl.Result{}, nil
 }
@@ -749,8 +758,11 @@ func clusterStatusEqual(a, b gnmicv1alpha1.ClusterStatus) bool {
 }
 
 // applyConfigToPods sends the apply plan to all gNMIc pods with distributed targets.
-// Returns the number of targets that could not be assigned due to capacity limits.
-func (r *ClusterReconciler) applyConfigToPods(ctx context.Context, cluster *gnmicv1alpha1.Cluster, plan *gnmic.ApplyPlan, numPods int) (int32, error) {
+// Returns the number of targets that could not be assigned due to capacity
+// limits, and whether any pod's apply is still unverified and needs a
+// follow-up POST once the ApplyCache grace period elapses (see
+// applyVerifyDelay).
+func (r *ClusterReconciler) applyConfigToPods(ctx context.Context, cluster *gnmicv1alpha1.Cluster, plan *gnmic.ApplyPlan, numPods int) (unassigned int32, needsVerify bool, err error) {
 	logger := log.FromContext(ctx)
 
 	stsName := fmt.Sprintf("%s%s", resourcePrefix, cluster.Name)
@@ -762,7 +774,7 @@ func (r *ClusterReconciler) applyConfigToPods(ctx context.Context, cluster *gnmi
 	// create an HTTP client to send the apply plan to the gNMIc pods
 	httpClient, err := r.createHTTPClientForCluster(ctx, cluster)
 	if err != nil {
-		return 0, fmt.Errorf("failed to create HTTP client: %w", err)
+		return 0, false, fmt.Errorf("failed to create HTTP client: %w", err)
 	}
 	distResult := gnmic.DistributeTargets(plan, numPods, cluster.Spec.TargetDistribution)
 
@@ -796,7 +808,7 @@ func (r *ClusterReconciler) applyConfigToPods(ctx context.Context, cluster *gnmi
 		}
 		body, err := json.Marshal(podPlan)
 		if err != nil {
-			return 0, fmt.Errorf("failed to marshal apply plan for pod %d: %w", podIndex, err)
+			return 0, false, fmt.Errorf("failed to marshal apply plan for pod %d: %w", podIndex, err)
 		}
 		hash := fingerprint(body)
 		bodies[podIndex] = body
@@ -806,8 +818,20 @@ func (r *ClusterReconciler) applyConfigToPods(ctx context.Context, cluster *gnmi
 			anyChanged = true
 		}
 	}
+	// Checked for every pod with a plan, not just those just (re)applied to:
+	// a pod can be sitting on an unverified hash from an earlier reconcile
+	// that this pass left untouched because Unchanged still held it inside
+	// the grace period.
+	pendingVerify := func() bool {
+		for podIndex := range hashes {
+			if r.Applied.NeedsVerify(streamKey(cluster.Namespace, cluster.Name, podIndex), hashes[podIndex]) {
+				return true
+			}
+		}
+		return false
+	}
 	if !anyChanged {
-		return int32(len(distResult.UnassignedTargets)), nil
+		return int32(len(distResult.UnassignedTargets)), pendingVerify(), nil
 	}
 
 	// Two-phase apply avoids double-collection when a target moves between pods.
@@ -848,7 +872,7 @@ func (r *ClusterReconciler) applyConfigToPods(ctx context.Context, cluster *gnmi
 			url := podURL(podIndex)
 			logger.Info("shrinking gNMIc pod targets before reassignment", "url", url, "targets", len(shrink.Targets))
 			if err := r.sendApplyRequest(ctx, url, shrink, httpClient); err != nil {
-				return 0, fmt.Errorf("failed to shrink config on pod %d: %w", podIndex, err)
+				return 0, false, fmt.Errorf("failed to shrink config on pod %d: %w", podIndex, err)
 			}
 		}
 		// gNMIc's config/apply returns before Subscribe streams are fully torn
@@ -857,7 +881,7 @@ func (r *ClusterReconciler) applyConfigToPods(ctx context.Context, cluster *gnmi
 		// double-collects.
 		select {
 		case <-ctx.Done():
-			return 0, ctx.Err()
+			return 0, false, ctx.Err()
 		case <-time.After(time.Second):
 		}
 	}
@@ -868,22 +892,24 @@ func (r *ClusterReconciler) applyConfigToPods(ctx context.Context, cluster *gnmi
 			continue
 		}
 		url := podURL(podIndex)
-		logger.Info("sending config to gNMIc pod", "url", url)
+		key := streamKey(cluster.Namespace, cluster.Name, podIndex)
+		isVerifyRepost := r.Applied.NeedsVerify(key, hashes[podIndex])
+		logger.Info("sending config to gNMIc pod", "url", url, "verifyRepost", isVerifyRepost)
 		if err := r.sendApplyBody(ctx, url, bodies[podIndex], httpClient); err != nil {
-			return 0, fmt.Errorf("failed to apply config to pod %d: %w", podIndex, err)
+			return 0, false, fmt.Errorf("failed to apply config to pod %d: %w", podIndex, err)
 		}
 		// Recorded only after the POST succeeds. Recording the attempt would
 		// make a failed apply look applied until the plan changes again.
-		r.Applied.Record(streamKey(cluster.Namespace, cluster.Name, podIndex), hashes[podIndex])
-		logger.Info("config applied to pod", "pod", podIndex, "targets", len(podPlan.Targets))
+		r.Applied.Record(key, hashes[podIndex])
+		logger.Info("config applied to pod", "pod", podIndex, "targets", len(podPlan.Targets), "verifyRepost", isVerifyRepost)
 	}
 
-	unassigned := int32(len(distResult.UnassignedTargets))
+	unassigned = int32(len(distResult.UnassignedTargets))
 	if unassigned > 0 {
 		logger.Info("targets unassigned due to capacity limits", "count", unassigned)
 	}
 
-	return unassigned, nil
+	return unassigned, pendingVerify(), nil
 }
 
 func (r *ClusterReconciler) createHTTPClientForCluster(ctx context.Context, cluster *gnmicv1alpha1.Cluster) (*http.Client, error) {