fix(shard): always orphan PVCs on scale-down instead of hard-deleting

pkg/cluster-handler/controller/multigrescluster/multigrescluster_controller.go

@@ -391,8 +391,16 @@ func (r *MultigresClusterReconciler) ensureClusterFinalizer(
 	return nil
 }
 
-// handleDeletion deletes cluster children and retained topology PVCs before
-// releasing the cluster cleanup finalizer.
+// childDeletionRequeueDelay is how long handleDeletion waits before
+// rechecking for remaining children. The cluster controller does not watch
+// Shards, so it must poll rather than wait for a watch event.
+const childDeletionRequeueDelay = 2 * time.Second
+
+// handleDeletion deletes cluster children and retained topology PVCs, then
+// holds the cluster cleanup finalizer until all Shards, TableGroups, and
+// Cells are gone. That keeps the cluster's DeletionTimestamp visible so the
+// Shard controller's clusterIsChurning check can detect the teardown and
+// hard-delete PVCs instead of orphaning them.
 func (r *MultigresClusterReconciler) handleDeletion(
 	ctx context.Context,
 	cluster *multigresv1alpha1.MultigresCluster,
@@ -465,6 +473,20 @@ func (r *MultigresClusterReconciler) handleDeletion(
 		}
 	}
 
+	shards := &multigresv1alpha1.ShardList{}
+	if err := r.List(ctx, shards, ns, clusterLabels); err != nil {
+		return ctrl.Result{}, fmt.Errorf("failed to list shards: %w", err)
+	}
+
+	if len(shards.Items) > 0 || len(tableGroups.Items) > 0 || len(cells.Items) > 0 {
+		l.Info("Waiting for child resources to finish deleting",
+			"shards", len(shards.Items),
+			"tablegroups", len(tableGroups.Items),
+			"cells", len(cells.Items),
+		)
+		return ctrl.Result{RequeueAfter: childDeletionRequeueDelay}, nil
+	}
+
 	if slices.Contains(cluster.Finalizers, multigresv1alpha1.FinalizerClusterCleanup) {
 		before := cluster.DeepCopy()
 
@@ -495,7 +517,7 @@ func (r *MultigresClusterReconciler) handleDeletion(
 	}
 
 	l.Info("Cluster cleanup complete")
-	r.Recorder.Event(cluster, "Normal", "CleanupComplete", "Initiated deletion of child resources")
+	r.Recorder.Event(cluster, "Normal", "CleanupComplete", "All child resources deleted")
 	return ctrl.Result{}, nil
 }
 

pkg/cluster-handler/controller/multigrescluster/multigrescluster_controller_test.go

@@ -409,6 +409,149 @@ func TestHandleDeletionReleasesPoolerClient(t *testing.T) {
 	})
 }
 
+func TestHandleDeletionWaitsForChildren(t *testing.T) {
+	clusterKey := types.NamespacedName{Name: "test-cluster", Namespace: "test-ns"}
+	deletionTimestamp := metav1.Now()
+
+	newCluster := func() *multigresv1alpha1.MultigresCluster {
+		return &multigresv1alpha1.MultigresCluster{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:              clusterKey.Name,
+				Namespace:         clusterKey.Namespace,
+				DeletionTimestamp: &deletionTimestamp,
+				Finalizers:        []string{multigresv1alpha1.FinalizerClusterCleanup},
+			},
+		}
+	}
+	childLabels := map[string]string{metadata.LabelMultigresCluster: clusterKey.Name}
+
+	tests := []struct {
+		name  string
+		child client.Object
+	}{
+		{
+			name: "shard still present",
+			child: &multigresv1alpha1.Shard{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-shard",
+					Namespace: clusterKey.Namespace,
+					Labels:    childLabels,
+				},
+			},
+		},
+		{
+			name: "tablegroup still present",
+			child: &multigresv1alpha1.TableGroup{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:       "test-tablegroup",
+					Namespace:  clusterKey.Namespace,
+					Labels:     childLabels,
+					Finalizers: []string{"test/hold"},
+				},
+			},
+		},
+		{
+			name: "cell still present",
+			child: &multigresv1alpha1.Cell{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:       "test-cell",
+					Namespace:  clusterKey.Namespace,
+					Labels:     childLabels,
+					Finalizers: []string{"test/hold"},
+				},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			cluster := newCluster()
+			forgotten := false
+			r := &MultigresClusterReconciler{
+				Client: fake.NewClientBuilder().
+					WithScheme(setupScheme()).
+					WithObjects(cluster, tt.child).
+					Build(),
+				Recorder: record.NewFakeRecorder(1),
+				PoolerClientCache: poolerClientCacheFunc(func(types.NamespacedName) {
+					forgotten = true
+				}),
+			}
+
+			result, err := r.handleDeletion(t.Context(), cluster.DeepCopy())
+			if err != nil {
+				t.Fatalf("handleDeletion() error = %v", err)
+			}
+			if result.RequeueAfter != childDeletionRequeueDelay {
+				t.Fatalf(
+					"RequeueAfter = %v, want %v",
+					result.RequeueAfter,
+					childDeletionRequeueDelay,
+				)
+			}
+			if forgotten {
+				t.Fatal("pooler client cache notified while children remain")
+			}
+
+			got := &multigresv1alpha1.MultigresCluster{}
+			if err := r.Get(t.Context(), clusterKey, got); err != nil {
+				t.Fatalf("failed to get cluster: %v", err)
+			}
+			if !slices.Contains(got.Finalizers, multigresv1alpha1.FinalizerClusterCleanup) {
+				t.Fatal("cluster cleanup finalizer removed while children remain")
+			}
+		})
+	}
+
+	t.Run("no children", func(t *testing.T) {
+		cluster := newCluster()
+		r := &MultigresClusterReconciler{
+			Client: fake.NewClientBuilder().
+				WithScheme(setupScheme()).
+				WithObjects(cluster).
+				Build(),
+			Recorder: record.NewFakeRecorder(1),
+		}
+		forgotten := make(chan types.NamespacedName, 1)
+		r.PoolerClientCache = poolerClientCacheFunc(func(key types.NamespacedName) {
+			forgotten <- key
+		})
+
+		fetched := &multigresv1alpha1.MultigresCluster{}
+		if err := r.Get(t.Context(), clusterKey, fetched); err != nil {
+			t.Fatalf("failed to get cluster: %v", err)
+		}
+
+		result, err := r.handleDeletion(t.Context(), fetched)
+		if err != nil {
+			t.Fatalf("handleDeletion() error = %v", err)
+		}
+		if result.RequeueAfter != 0 {
+			t.Fatalf("RequeueAfter = %v, want 0", result.RequeueAfter)
+		}
+		select {
+		case got := <-forgotten:
+			if got != clusterKey {
+				t.Fatalf("forgot cluster %v, want %v", got, clusterKey)
+			}
+		default:
+			t.Fatal("pooler client cache was not notified")
+		}
+
+		got := &multigresv1alpha1.MultigresCluster{}
+		err = r.Get(t.Context(), clusterKey, got)
+		switch {
+		case err == nil:
+			if slices.Contains(got.Finalizers, multigresv1alpha1.FinalizerClusterCleanup) {
+				t.Fatal("cluster cleanup finalizer not removed")
+			}
+		case apierrors.IsNotFound(err):
+		default:
+			t.Fatalf("failed to get cluster: %v", err)
+		}
+	})
+}
+
 func TestReconcileNotFoundReleasesPoolerClient(t *testing.T) {
 	key := types.NamespacedName{Name: "missing", Namespace: "test-ns"}
 	forgotten := make(chan types.NamespacedName, 1)

pkg/resource-handler/controller/shard/reconcile_deletion.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"fmt"
 	"slices"
-	"strings"
 	"time"
 
 	"github.com/multigres/multigres/go/common/topoclient"
@@ -123,10 +122,10 @@ func (r *ShardReconciler) handleDeletion(
 		return ctrl.Result{RequeueAfter: podTerminationRequeueDelay}, nil
 	}
 
-	// All pods gone. Clean up PVCs whose policy resolves to Delete. Small
-	// shards (<= pvcOrphanReplicasThreshold replicas) defer the work to
-	// multigres-gc by labelling the PVC with multigres.com/orphan-since=<now>,
-	// larger shards are deleted in-line.
+	// All pods gone. Clean up PVCs whose policy resolves to Delete. Unless the
+	// owning MultigresCluster is being torn down, this defers to multigres-gc
+	// by labelling the PVC with multigres.com/orphan-since=<now> instead of
+	// deleting in-line.
 	if err := r.cleanupShardPVCs(ctx, shard); err != nil {
 		return ctrl.Result{}, err
 	}
@@ -149,14 +148,23 @@ func (r *ShardReconciler) handleDeletion(
 
 // cleanupShardPVCs handles per-PVC cleanup when a Shard is being deleted.
 // Only PVCs whose effective WhenDeleted policy is Delete are touched.
+//
+// If the owning MultigresCluster is confirmed still present and being
+// deleted, PVCs are deleted in line. The cluster is going away, so there is
+// nothing left to roll a scale down back to. The cluster controller holds
+// its own finalizer until Shards are gone, so this is the normal path for a
+// full cluster teardown. Otherwise the Shard is being individually removed
+// (e.g. a shard count scale down) while the cluster stays up, so PVCs are
+// orphaned instead.
 func (r *ShardReconciler) cleanupShardPVCs(
 	ctx context.Context,
 	shard *multigresv1alpha1.Shard,
 ) error {
 	logger := log.FromContext(ctx)
 
+	clusterName := shard.Labels[metadata.LabelMultigresCluster]
 	selector := map[string]string{
-		metadata.LabelMultigresCluster:    shard.Labels[metadata.LabelMultigresCluster],
+		metadata.LabelMultigresCluster:    clusterName,
 		metadata.LabelMultigresDatabase:   string(shard.Spec.DatabaseName),
 		metadata.LabelMultigresTableGroup: string(shard.Spec.TableGroupName),
 		metadata.LabelMultigresShard:      string(shard.Spec.ShardName),
@@ -172,42 +180,28 @@ func (r *ShardReconciler) cleanupShardPVCs(
 		return fmt.Errorf("failed to list PVCs for cleanup: %w", err)
 	}
 
-	// Group eligible PVCs by pool+cell so the keep-threshold decision is made
-	// per group (matching replicasPerCell semantics).
-	groups := map[string][]*corev1.PersistentVolumeClaim{}
+	churning, err := r.clusterIsChurning(ctx, shard.Namespace, clusterName)
+	if err != nil {
+		return fmt.Errorf("failed to determine MultigresCluster deletion state: %w", err)
+	}
+
+	now := time.Now()
 	for i := range pvcList.Items {
 		pvc := &pvcList.Items[i]
 		if !shardPVCShouldBeCleaned(shard, pvc) {
 			continue
 		}
-		key := pvc.Labels[metadata.LabelMultigresPool] + "/" + pvc.Labels[metadata.LabelMultigresCell]
-		groups[key] = append(groups[key], pvc)
-	}
-
-	now := time.Now()
-	for _, group := range groups {
-		// Sort descending by name so the highest-ordinal PVCs are visited first
-		// and become the deleted excess, the lowest are kept as orphans.
-		slices.SortFunc(group, func(a, b *corev1.PersistentVolumeClaim) int {
-			return strings.Compare(b.Name, a.Name)
-		})
-		live := len(group)
-		for _, pvc := range group {
-			_, hasIndex := resolvePodIndex(pvc.Name)
-			if !hasIndex || orphanByRemainingCount(live) {
-				if err := pvcutil.MarkOrphan(ctx, r.Client, pvc, shard.GetUID(), now); err != nil {
-					return fmt.Errorf("failed to mark PVC %s orphan: %w", pvc.Name, err)
-				}
-				logger.Info("Marked PVC orphan on Shard deletion", "pvc", pvc.Name)
-				live--
-				continue
-			}
+		if churning {
 			if err := r.Delete(ctx, pvc); err != nil && !errors.IsNotFound(err) {
 				return fmt.Errorf("failed to delete PVC %s: %w", pvc.Name, err)
 			}
-			logger.Info("Deleted PVC on Shard deletion", "pvc", pvc.Name)
-			live--
+			logger.Info("Deleted PVC on MultigresCluster deletion", "pvc", pvc.Name)
+			continue
+		}
+		if err := pvcutil.MarkOrphan(ctx, r.Client, pvc, shard.GetUID(), now); err != nil {
+			return fmt.Errorf("failed to mark PVC %s orphan: %w", pvc.Name, err)
 		}
+		logger.Info("Marked PVC orphan on Shard deletion", "pvc", pvc.Name)
 	}
 	return nil
 }

pkg/resource-handler/controller/shard/reconcile_deletion_test.go

@@ -8,6 +8,7 @@ import (
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	policyv1 "k8s.io/api/policy/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
@@ -363,6 +364,63 @@ func TestHandleDeletion(t *testing.T) {
 		}
 	})
 
+	t.Run(
+		"MultigresCluster being deleted hard-deletes PVC instead of orphaning",
+		func(t *testing.T) {
+			t.Parallel()
+
+			shard := baseShard.DeepCopy()
+			shard.Spec.PVCDeletionPolicy = &multigresv1alpha1.PVCDeletionPolicy{
+				WhenDeleted: multigresv1alpha1.DeletePVCRetentionPolicy,
+			}
+			pvc := &corev1.PersistentVolumeClaim{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "data-pvc-churn",
+					Namespace: "default",
+					Labels:    shardLabels,
+				},
+			}
+			cluster := &multigresv1alpha1.MultigresCluster{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:              "test-cluster",
+					Namespace:         "default",
+					DeletionTimestamp: &metav1.Time{Time: metav1.Now().Time},
+					Finalizers:        []string{multigresv1alpha1.FinalizerClusterCleanup},
+				},
+			}
+
+			c := fake.NewClientBuilder().
+				WithScheme(scheme).
+				WithObjects(shard, pvc, cluster).
+				WithStatusSubresource(&multigresv1alpha1.Shard{}).
+				Build()
+
+			r := &ShardReconciler{
+				Client:          c,
+				Scheme:          scheme,
+				Recorder:        record.NewFakeRecorder(10),
+				CreateTopoStore: newMemoryTopoFactory(),
+			}
+
+			result, err := r.handleDeletion(context.Background(), shard)
+			if err != nil {
+				t.Fatalf("handleDeletion returned error: %v", err)
+			}
+			if result.RequeueAfter != 0 {
+				t.Errorf("Expected no requeue once pods are gone, got %v", result.RequeueAfter)
+			}
+			got := &corev1.PersistentVolumeClaim{}
+			err = c.Get(context.Background(),
+				types.NamespacedName{Name: "data-pvc-churn", Namespace: "default"}, got)
+			if !apierrors.IsNotFound(err) {
+				t.Errorf(
+					"PVC should be hard-deleted while cluster is being deleted, got err=%v",
+					err,
+				)
+			}
+		},
+	)
+
 	t.Run("pod stuck terminating past timeout does not block PVC cleanup", func(t *testing.T) {
 		t.Parallel()