fix: release target cluster state only from its owning pod

internal/controller/targetstate_controller.go

@@ -314,7 +314,7 @@ func (r *TargetStateReconciler) pollAndSync(
 	podKey := podStateKey(namespace, clusterName, podName)
 	for _, name := range r.swapReported(podKey, reportedTargets) {
 		logger.Info("poll: releasing stale cluster state", "target", name, "cluster", clusterName, "pod", podName)
-		r.removeClusterState(ctx, types.NamespacedName{Name: name, Namespace: namespace}, clusterName, logger)
+		r.removeClusterState(ctx, types.NamespacedName{Name: name, Namespace: namespace}, clusterName, podName, logger)
 	}
 
 	if !r.dueForSweep(podKey) {
@@ -336,7 +336,7 @@ func (r *TargetStateReconciler) pollAndSync(
 			continue
 		}
 		logger.Info("sweep: removing stale cluster state", "target", target.Name, "cluster", clusterName, "pod", podName)
-		r.removeClusterState(ctx, types.NamespacedName{Name: target.Name, Namespace: target.Namespace}, clusterName, logger)
+		r.removeClusterState(ctx, types.NamespacedName{Name: target.Name, Namespace: target.Namespace}, clusterName, podName, logger)
 	}
 }
 
@@ -361,7 +361,7 @@ func (r *TargetStateReconciler) handleEvent(ctx context.Context, event gnmic.SSE
 	targetNN := types.NamespacedName{Name: targetName, Namespace: targetNamespace}
 
 	if event.EventType == gnmic.SSEEventDelete {
-		r.removeClusterState(ctx, targetNN, clusterName, logger)
+		r.removeClusterState(ctx, targetNN, clusterName, podName, logger)
 		return
 	}
 

internal/controller/targetstate_ownership_test.go

@@ -0,0 +1,88 @@
+package controller
+
+import (
+	"context"
+	"testing"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+	logf "sigs.k8s.io/controller-runtime/pkg/log"
+
+	gnmicv1alpha1 "github.com/gnmic/operator/api/v1alpha1"
+)
+
+func targetOwnedBy(pod string) *gnmicv1alpha1.Target {
+	return &gnmicv1alpha1.Target{
+		ObjectMeta: metav1.ObjectMeta{Name: "leaf1", Namespace: "default"},
+		Status: gnmicv1alpha1.TargetStatus{
+			Clusters: 1,
+			ClusterStates: map[string]gnmicv1alpha1.ClusterTargetState{
+				"c1": {Pod: pod, State: "running", ConnectionState: "READY"},
+			},
+		},
+	}
+}
+
+func targetStateReconcilerWith(t *testing.T, objs ...client.Object) (*TargetStateReconciler, client.Client) {
+	t.Helper()
+	scheme := secretWatchScheme(t)
+	cl := fake.NewClientBuilder().WithScheme(scheme).
+		WithObjects(objs...).WithStatusSubresource(&gnmicv1alpha1.Target{}).Build()
+	return &TargetStateReconciler{Client: cl, Scheme: scheme}, cl
+}
+
+// A target moving from pod 0 to pod 1 is reported by pod 1 immediately, but pod 0 only
+// notices it is gone on its next poll — up to a poll interval later. Releasing
+// unconditionally then deleted the entry pod 1 had already written.
+func TestRemoveClusterStateLeavesTheNewOwnerAlone(t *testing.T) {
+	r, cl := targetStateReconcilerWith(t, targetOwnedBy("gnmic-c1-1"))
+	nn := types.NamespacedName{Name: "leaf1", Namespace: "default"}
+
+	// pod 0 releases a target it no longer holds
+	r.removeClusterState(context.Background(), nn, "c1", "gnmic-c1-0", logf.Log)
+
+	var got gnmicv1alpha1.Target
+	if err := cl.Get(context.Background(), nn, &got); err != nil {
+		t.Fatal(err)
+	}
+	state, ok := got.Status.ClusterStates["c1"]
+	if !ok {
+		t.Fatal("pod 0's release deleted the entry pod 1 owns")
+	}
+	if state.Pod != "gnmic-c1-1" {
+		t.Fatalf("owner = %q, want gnmic-c1-1", state.Pod)
+	}
+	if got.Status.Clusters != 1 {
+		t.Errorf("clusters = %d, want 1", got.Status.Clusters)
+	}
+}
+
+// The owning pod still releases its own entry.
+func TestRemoveClusterStateReleasesItsOwnEntry(t *testing.T) {
+	r, cl := targetStateReconcilerWith(t, targetOwnedBy("gnmic-c1-0"))
+	nn := types.NamespacedName{Name: "leaf1", Namespace: "default"}
+
+	r.removeClusterState(context.Background(), nn, "c1", "gnmic-c1-0", logf.Log)
+
+	var got gnmicv1alpha1.Target
+	_ = cl.Get(context.Background(), nn, &got)
+	if _, ok := got.Status.ClusterStates["c1"]; ok {
+		t.Fatal("the owning pod's release was ignored")
+	}
+}
+
+// An empty pod name means "whoever holds it", for cluster-wide cleanup.
+func TestRemoveClusterStateEmptyPodReleasesAnyOwner(t *testing.T) {
+	r, cl := targetStateReconcilerWith(t, targetOwnedBy("gnmic-c1-7"))
+	nn := types.NamespacedName{Name: "leaf1", Namespace: "default"}
+
+	r.removeClusterState(context.Background(), nn, "c1", "", logf.Log)
+
+	var got gnmicv1alpha1.Target
+	_ = cl.Get(context.Background(), nn, &got)
+	if _, ok := got.Status.ClusterStates["c1"]; ok {
+		t.Fatal("an unconditional release did nothing")
+	}
+}

internal/controller/targetstate_status.go

@@ -65,15 +65,32 @@ func (r *TargetStateReconciler) applyClusterState(
 	})
 }
 
-// removeClusterState drops one cluster's entry from a Target's status.
+// removeClusterState drops one cluster's entry from a Target's status, but only when
+// the entry still names the pod releasing it.
+//
+// Releasing unconditionally meant a pod that had lost a target deleted the entry the
+// target's *new* owner had just written. During redistribution the old pod only
+// notices the target missing from its own poll up to a poll interval later, so the
+// delete lands after the new owner has already reported itself. That is not cosmetic:
+// PlanBuilder reads this field back to build CurrentTargetAssignment, so wiping it
+// makes the target look unassigned, re-hashes it on the next reconcile, and empties
+// the assignment that the two-phase shrink depends on.
+//
+// An empty podName releases whoever holds it, for callers cleaning up a cluster
+// wholesale rather than a single pod's view.
 func (r *TargetStateReconciler) removeClusterState(
 	ctx context.Context,
 	targetNN types.NamespacedName,
-	clusterName string,
+	clusterName, podName string,
 	logger logr.Logger,
 ) {
 	r.mutateStatus(ctx, targetNN, logger, func(target *gnmicv1alpha1.Target) bool {
-		if _, ok := target.Status.ClusterStates[clusterName]; !ok {
+		current, ok := target.Status.ClusterStates[clusterName]
+		if !ok {
+			return false
+		}
+		if podName != "" && current.Pod != podName {
+			// Someone else owns it now; their entry is the current one.
 			return false
 		}
 		delete(target.Status.ClusterStates, clusterName)

internal/controller/targetstate_status_test.go

@@ -162,7 +162,7 @@ func TestRemoveClusterState(t *testing.T) {
 	r := &TargetStateReconciler{Client: cl, Scheme: scheme}
 	nn := types.NamespacedName{Name: "leaf1", Namespace: "default"}
 
-	r.removeClusterState(context.Background(), nn, "c1", logf.Log)
+	r.removeClusterState(context.Background(), nn, "c1", "gnmic-c1-0", logf.Log)
 
 	var got gnmicv1alpha1.Target
 	if err := cl.Get(context.Background(), nn, &got); err != nil {
@@ -180,7 +180,7 @@ func TestRemoveClusterState(t *testing.T) {
 
 	// Removing what is not there must not write.
 	before := readVersion(t, cl, nn)
-	r.removeClusterState(context.Background(), nn, "c1", logf.Log)
+	r.removeClusterState(context.Background(), nn, "c1", "gnmic-c1-0", logf.Log)
 	if after := readVersion(t, cl, nn); after != before {
 		t.Fatalf("removing an absent entry wrote: %s -> %s", before, after)
 	}