fix: prevent duplicate CNClaim binding to same CN Pod during migration

Add podClaimedByOthers() check to selectCN() and Finalize() to verify
no other CNClaim references a Pod via spec.podName before binding or
reclaiming it. This prevents the race condition where migration causes
Pod labels to become stale while spec.podName still holds the true
binding relationship.

Fixes #581, #582, #585

Author: xzxiong

pkg/controllers/cnclaim/controller.go

@@ -161,6 +161,13 @@ func (r *Actor) selectCN(ctx *recon.Context[*v1alpha1.CNClaim], orphans []corev1
 	sortCNByPriority(c, idleCNs)
 	for i := range idleCNs {
 		pod := &idleCNs[i]
+		// skip pod already referenced by another CNClaim's spec.podName
+		if claimed, err := podClaimedByOthers(ctx, c.Namespace, pod.Name, c.Name); err != nil {
+			return nil, errors.WrapPrefix(err, "error checking pod claims", 0)
+		} else if claimed {
+			ctx.Log.Info("skip pod claimed by other CNClaim", "podName", pod.Name)
+			continue
+		}
 		if err := r.ensureOwnership(ctx, pod); err != nil {
 			if apierrors.IsConflict(err) {
 				ctx.Log.Info("CN pod is not up to date, try next", "podName", pod.Name)
@@ -342,6 +349,13 @@ func (r *Actor) Finalize(ctx *recon.Context[*v1alpha1.CNClaim]) (bool, error) {
 	}
 	for i := range ownedCNs {
 		cn := ownedCNs[i]
+		// skip reclaim if another CNClaim still references this pod via spec.podName
+		if claimed, err := podClaimedByOthers(ctx, c.Namespace, cn.Name, c.Name); err != nil {
+			return false, errors.WrapPrefix(err, "error checking pod claims", 0)
+		} else if claimed {
+			ctx.Log.Info("skip reclaim, pod still claimed by other CNClaim", "pod", cn.Name)
+			continue
+		}
 		ctx.Log.Info("finalize CNClaim, reclaim bound CN", "cn", cn.Name)
 		if err := r.reclaimCN(ctx, &cn); err != nil {
 			return false, err
@@ -350,6 +364,25 @@ func (r *Actor) Finalize(ctx *recon.Context[*v1alpha1.CNClaim]) (bool, error) {
 	return false, nil
 }
 
+// podClaimedByOthers checks if the given pod is referenced by any CNClaim's
+// spec.podName other than excludeClaim in the same namespace.
+func podClaimedByOthers(cli recon.KubeClient, namespace, podName, excludeClaim string) (bool, error) {
+	claimList := &v1alpha1.CNClaimList{}
+	if err := cli.List(claimList, client.InNamespace(namespace)); err != nil {
+		return false, err
+	}
+	for i := range claimList.Items {
+		claim := &claimList.Items[i]
+		if claim.Name == excludeClaim {
+			continue
+		}
+		if claim.Spec.PodName == podName {
+			return true, nil
+		}
+	}
+	return false, nil
+}
+
 func (r *Actor) patchStore(ctx *recon.Context[*v1alpha1.CNClaim], pod *corev1.Pod, req logpb.CNStateLabel) (*metadata.CNService, error) {
 	cs, err := common.ResolveCNSet(ctx, pod)
 	if err != nil {

pkg/controllers/cnclaim/controller_test.go

@@ -15,17 +15,144 @@
 package cnclaim
 
 import (
+	"context"
 	"math/rand"
 	"testing"
 
 	"github.com/matrixorigin/matrixone-operator/api/core/v1alpha1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/utils/pointer"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 
 	. "github.com/onsi/gomega"
 )
 
+func newFakeClient(objs ...client.Object) client.Client {
+	scheme := runtime.NewScheme()
+	_ = v1alpha1.SchemeBuilder.AddToScheme(scheme)
+	_ = corev1.AddToScheme(scheme)
+	return fake.NewClientBuilder().WithScheme(scheme).WithObjects(objs...).Build()
+}
+
+// fakeKubeClient adapts client.Client to recon.KubeClient for testing.
+type fakeKubeClient struct {
+	client.Client
+}
+
+func (f *fakeKubeClient) Create(obj client.Object, opts ...client.CreateOption) error {
+	return f.Client.Create(context.TODO(), obj, opts...)
+}
+func (f *fakeKubeClient) CreateOwned(obj client.Object, opts ...client.CreateOption) error {
+	return f.Client.Create(context.TODO(), obj, opts...)
+}
+func (f *fakeKubeClient) Get(objKey client.ObjectKey, obj client.Object) error {
+	return f.Client.Get(context.TODO(), objKey, obj)
+}
+func (f *fakeKubeClient) Update(obj client.Object, opts ...client.UpdateOption) error {
+	return f.Client.Update(context.TODO(), obj, opts...)
+}
+func (f *fakeKubeClient) UpdateStatus(obj client.Object, opts ...client.SubResourceUpdateOption) error {
+	return f.Client.Status().Update(context.TODO(), obj, opts...)
+}
+func (f *fakeKubeClient) Delete(obj client.Object, opts ...client.DeleteOption) error {
+	return f.Client.Delete(context.TODO(), obj, opts...)
+}
+func (f *fakeKubeClient) List(objList client.ObjectList, opts ...client.ListOption) error {
+	return f.Client.List(context.TODO(), objList, opts...)
+}
+func (f *fakeKubeClient) Patch(obj client.Object, mutateFn func() error, opts ...client.PatchOption) error {
+	return mutateFn()
+}
+func (f *fakeKubeClient) Exist(objKey client.ObjectKey, kind client.Object) (bool, error) {
+	err := f.Client.Get(context.TODO(), objKey, kind)
+	if err != nil {
+		return false, client.IgnoreNotFound(err)
+	}
+	return true, nil
+}
+
+func Test_podClaimedByOthers(t *testing.T) {
+	tests := []struct {
+		name         string
+		claims       []client.Object
+		podName      string
+		excludeClaim string
+		want         bool
+	}{
+		{
+			name:         "no claims exist",
+			claims:       nil,
+			podName:      "pod-1",
+			excludeClaim: "claim-a",
+			want:         false,
+		},
+		{
+			name: "pod only claimed by self",
+			claims: []client.Object{
+				&v1alpha1.CNClaim{
+					ObjectMeta: metav1.ObjectMeta{Name: "claim-a", Namespace: "ns"},
+					Spec:       v1alpha1.CNClaimSpec{ClaimPodRef: v1alpha1.ClaimPodRef{PodName: "pod-1"}},
+				},
+			},
+			podName:      "pod-1",
+			excludeClaim: "claim-a",
+			want:         false,
+		},
+		{
+			name: "pod claimed by another claim",
+			claims: []client.Object{
+				&v1alpha1.CNClaim{
+					ObjectMeta: metav1.ObjectMeta{Name: "claim-a", Namespace: "ns"},
+					Spec:       v1alpha1.CNClaimSpec{ClaimPodRef: v1alpha1.ClaimPodRef{PodName: "pod-1"}},
+				},
+				&v1alpha1.CNClaim{
+					ObjectMeta: metav1.ObjectMeta{Name: "claim-b", Namespace: "ns"},
+					Spec:       v1alpha1.CNClaimSpec{ClaimPodRef: v1alpha1.ClaimPodRef{PodName: "pod-1"}},
+				},
+			},
+			podName:      "pod-1",
+			excludeClaim: "claim-a",
+			want:         true,
+		},
+		{
+			name: "pod not claimed by anyone",
+			claims: []client.Object{
+				&v1alpha1.CNClaim{
+					ObjectMeta: metav1.ObjectMeta{Name: "claim-a", Namespace: "ns"},
+					Spec:       v1alpha1.CNClaimSpec{ClaimPodRef: v1alpha1.ClaimPodRef{PodName: "pod-2"}},
+				},
+			},
+			podName:      "pod-1",
+			excludeClaim: "claim-a",
+			want:         false,
+		},
+		{
+			name: "claim with empty podName is ignored",
+			claims: []client.Object{
+				&v1alpha1.CNClaim{
+					ObjectMeta: metav1.ObjectMeta{Name: "claim-pending", Namespace: "ns"},
+					Spec:       v1alpha1.CNClaimSpec{},
+				},
+			},
+			podName:      "pod-1",
+			excludeClaim: "claim-a",
+			want:         false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			g := NewGomegaWithT(t)
+			cli := newFakeClient(tt.claims...)
+			got, err := podClaimedByOthers(&fakeKubeClient{cli}, "ns", tt.podName, tt.excludeClaim)
+			g.Expect(err).NotTo(HaveOccurred())
+			g.Expect(got).To(Equal(tt.want))
+		})
+	}
+}
+
 func Test_sortCNByPriority(t *testing.T) {
 	tests := []struct {
 		name  string