fix(dataprotection): terminate restores for deleting source components

controllers/dataprotection/volumepopulator_controller.go

@@ -198,12 +198,17 @@ func (r *VolumePopulatorReconciler) mapComponentToPVCs(ctx context.Context, obj
 	if clusterOwner == nil || clusterOwner.Name != clusterName {
 		return nil
 	}
+	labels := client.MatchingLabels{constant.AppInstanceLabelKey: clusterName}
+	includeTerminal := !comp.DeletionTimestamp.IsZero()
+	if includeTerminal {
+		// Component deletion is a Component-scoped termination signal. Include
+		// terminal PVCs because they may still own restore resources.
+		labels[constant.KBAppComponentLabelKey] = componentName
+	}
 	// A PVC can depend on another Component through redirected postReady. The
 	// dependency is not represented on the Component, so normal Component
 	// changes fan out to unfinished restore PVCs in the exact Cluster instance.
-	return r.mapRestorePVCs(ctx, comp.Namespace, client.MatchingLabels{
-		constant.AppInstanceLabelKey: clusterName,
-	}, string(clusterOwner.UID), false)
+	return r.mapRestorePVCs(ctx, comp.Namespace, labels, string(clusterOwner.UID), includeTerminal)
 }
 
 func (r *VolumePopulatorReconciler) mapClusterToPVCs(ctx context.Context, obj client.Object) []reconcile.Request {
@@ -396,7 +401,7 @@ func (r *VolumePopulatorReconciler) syncPVC(reqCtx intctrlutil.RequestCtx, pvc *
 	if !matched {
 		return nil
 	}
-	terminated, err := r.handleRestoreClusterLifecycle(reqCtx, pvc)
+	terminated, err := r.handleRestoreParentLifecycle(reqCtx, pvc)
 	if err != nil || terminated {
 		return err
 	}
@@ -431,10 +436,11 @@ func (r *VolumePopulatorReconciler) syncPVC(reqCtx intctrlutil.RequestCtx, pvc *
 	return nil
 }
 
-// handleRestoreClusterLifecycle validates the Cluster identity and protection
-// before restore work starts, and initiates owner-driven cleanup when the
-// Cluster is deleting. Target PVC deletion alone is not a termination signal.
-func (r *VolumePopulatorReconciler) handleRestoreClusterLifecycle(reqCtx intctrlutil.RequestCtx,
+// handleRestoreParentLifecycle validates the recorded parent identity and
+// Cluster protection before restore work starts, and initiates owner-driven
+// cleanup when a supported parent is deleting. Target PVC deletion alone is
+// not a termination signal.
+func (r *VolumePopulatorReconciler) handleRestoreParentLifecycle(reqCtx intctrlutil.RequestCtx,
 	pvc *corev1.PersistentVolumeClaim) (bool, error) {
 	clusterName := pvc.Labels[constant.AppInstanceLabelKey]
 	componentName := pvc.Labels[constant.KBAppComponentLabelKey]
@@ -478,16 +484,20 @@ func (r *VolumePopulatorReconciler) handleRestoreClusterLifecycle(reqCtx intctrl
 	if !cluster.DeletionTimestamp.IsZero() {
 		return r.terminateClusterVolumePopulation(reqCtx, pvc, cluster)
 	}
-	if !pvc.DeletionTimestamp.IsZero() &&
-		!controllerutil.ContainsFinalizer(pvc, dptypes.DataProtectionFinalizerName) {
-		return true, nil
-	}
 
 	committed := volumePopulationIdentityCommitted(pvc, cluster)
 	if committed {
-		if _, err := r.committedVolumePopulationComponent(reqCtx.Ctx, pvc, cluster); err != nil {
+		comp, err := r.committedVolumePopulationComponent(reqCtx.Ctx, pvc, cluster)
+		if err != nil {
 			return false, restoreParentRequeue(err)
 		}
+		if !comp.DeletionTimestamp.IsZero() {
+			return r.terminateSourceComponentVolumePopulation(reqCtx, pvc, cluster)
+		}
+	}
+	if !pvc.DeletionTimestamp.IsZero() &&
+		!controllerutil.ContainsFinalizer(pvc, dptypes.DataProtectionFinalizerName) {
+		return true, nil
 	}
 	// Aggregate restore status gates normal progression, not owner cleanup.
 	if !clusterAllowsRestoreProgress(cluster) && !pvcRestoreTerminal(pvc) {
@@ -501,6 +511,11 @@ func (r *VolumePopulatorReconciler) handleRestoreClusterLifecycle(reqCtx intctrl
 		if err != nil {
 			return false, restoreParentRequeue(err)
 		}
+		// A deleting source Component terminates this PVC's restore before VP
+		// registers protection or creates any restore resources.
+		if !comp.DeletionTimestamp.IsZero() {
+			return true, nil
+		}
 		if err = r.registerVolumePopulation(reqCtx.Ctx, pvc, cluster, comp); err != nil {
 			return false, restoreParentRequeue(err)
 		}
@@ -597,6 +612,15 @@ func (r *VolumePopulatorReconciler) terminateClusterVolumePopulation(reqCtx intc
 	return true, err
 }
 
+func (r *VolumePopulatorReconciler) terminateSourceComponentVolumePopulation(reqCtx intctrlutil.RequestCtx,
+	pvc *corev1.PersistentVolumeClaim, cluster *appsv1.Cluster) (bool, error) {
+	err := r.cleanupSourceComponentVolumePopulation(reqCtx, pvc, cluster)
+	if err != nil && !intctrlutil.IsRequeueError(err) {
+		err = restoreParentRequeue(err)
+	}
+	return true, err
+}
+
 func (r *VolumePopulatorReconciler) cleanupClusterVolumePopulation(reqCtx intctrlutil.RequestCtx,
 	pvc *corev1.PersistentVolumeClaim, cluster *appsv1.Cluster) error {
 	pending, err := r.deleteExecutionRestoreAndWait(reqCtx.Ctx, pvc, cluster)
@@ -607,10 +631,26 @@ func (r *VolumePopulatorReconciler) cleanupClusterVolumePopulation(reqCtx intctr
 	if err != nil {
 		return err
 	}
-	if pending || postReadyPending {
+	return r.finishVolumePopulationTermination(reqCtx, pvc, cluster, pending || postReadyPending)
+}
+
+func (r *VolumePopulatorReconciler) cleanupSourceComponentVolumePopulation(reqCtx intctrlutil.RequestCtx,
+	pvc *corev1.PersistentVolumeClaim, cluster *appsv1.Cluster) error {
+	pending, err := r.deleteExecutionRestoreAndWait(reqCtx.Ctx, pvc, cluster)
+	if err != nil {
+		return err
+	}
+	// Source Component cleanup leaves shared postReady Restores to their target
+	// Component ownerReferences. Cluster cleanup deletes them at Cluster scope.
+	return r.finishVolumePopulationTermination(reqCtx, pvc, cluster, pending)
+}
+
+func (r *VolumePopulatorReconciler) finishVolumePopulationTermination(reqCtx intctrlutil.RequestCtx,
+	pvc *corev1.PersistentVolumeClaim, cluster *appsv1.Cluster, restoresPending bool) error {
+	if restoresPending {
 		return intctrlutil.NewRequeueError(reconcileInterval, "waiting for Restore owners to finish termination")
 	}
-	pending, err = r.deletePopulatePVCAndWait(reqCtx.Ctx, pvc, cluster)
+	pending, err := r.deletePopulatePVCAndWait(reqCtx.Ctx, pvc, cluster)
 	if err != nil {
 		return err
 	}

controllers/dataprotection/volumepopulator_controller_test.go

@@ -4091,6 +4091,13 @@ func TestMapComponentAndClusterDependencies(t *testing.T) {
 		{NamespacedName: client.ObjectKeyFromObject(mysql)},
 		{NamespacedName: client.ObjectKeyFromObject(postgresql)},
 	}, reconciler.mapComponentToPVCs(context.Background(), comp))
+	deleting := comp.DeepCopy()
+	now := metav1.Now()
+	deleting.DeletionTimestamp = &now
+	require.ElementsMatch(t, []reconcile.Request{
+		{NamespacedName: client.ObjectKeyFromObject(mysql)},
+		{NamespacedName: client.ObjectKeyFromObject(terminal)},
+	}, reconciler.mapComponentToPVCs(context.Background(), deleting))
 	require.ElementsMatch(t, []reconcile.Request{
 		{NamespacedName: client.ObjectKeyFromObject(mysql)},
 		{NamespacedName: client.ObjectKeyFromObject(postgresql)},
@@ -4272,6 +4279,167 @@ func TestClusterDeletionTerminatesVolumePopulationInOrder(t *testing.T) {
 	}
 }
 
+func TestComponentDeletionTerminatesVolumePopulationInOrder(t *testing.T) {
+	for _, retained := range []bool{false, true} {
+		t.Run(fmt.Sprintf("retained=%t", retained), func(t *testing.T) {
+			ctx := context.Background()
+			scheme, cluster, component, its, target := parentRestoreObjects(t)
+			now := metav1.Now()
+			component.DeletionTimestamp = &now
+			component.Finalizers = []string{"example.io/app-owner"}
+			cluster.Status.Conditions = []metav1.Condition{{
+				Type: kbappsv1.ConditionTypeRestore, Status: metav1.ConditionTrue,
+			}}
+			target.Finalizers = []string{dptypes.DataProtectionFinalizerName, "example.io/app-owner"}
+			target.Labels[dptypes.ComponentUIDLabelKey] = string(component.UID)
+			objects := []client.Object{cluster, component, target}
+			if retained {
+				target.OwnerReferences = nil
+			} else {
+				objects = append(objects, its)
+			}
+			helper := restoreHelperForTarget(target, cluster)
+			execution := executionRestoreForTarget(target, cluster)
+			execution.Finalizers = []string{"example.io/restore-owner"}
+			postReady := postReadyRestoreForComponent(target, cluster, component)
+			postReady.Finalizers = []string{"example.io/restore-owner"}
+			objects = append(objects, helper, execution, postReady)
+			cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(objects...).Build()
+			vp := &VolumePopulatorReconciler{Client: cli, Scheme: scheme}
+			reqCtx := intctrlutil.RequestCtx{Ctx: ctx}
+
+			require.ErrorContains(t, vp.syncPVC(reqCtx, target), "waiting for Restore owners")
+			require.NoError(t, cli.Get(ctx, client.ObjectKeyFromObject(execution), execution))
+			require.False(t, execution.DeletionTimestamp.IsZero())
+			execution.Finalizers = nil
+			require.NoError(t, cli.Update(ctx, execution))
+			require.NoError(t, cli.Get(ctx, client.ObjectKeyFromObject(postReady), postReady))
+			require.True(t, postReady.DeletionTimestamp.IsZero())
+
+			require.NoError(t, cli.Get(ctx, client.ObjectKeyFromObject(target), target))
+			require.ErrorContains(t, vp.syncPVC(reqCtx, target), "waiting for helper PVC to disappear")
+			require.True(t, apierrors.IsNotFound(cli.Get(ctx, client.ObjectKeyFromObject(helper), helper)))
+
+			require.NoError(t, cli.Get(ctx, client.ObjectKeyFromObject(target), target))
+			require.NoError(t, vp.syncPVC(reqCtx, target))
+			require.NoError(t, cli.Get(ctx, client.ObjectKeyFromObject(target), target))
+			require.Equal(t, []string{"example.io/app-owner"}, target.Finalizers)
+			require.NoError(t, cli.Get(ctx, client.ObjectKeyFromObject(postReady), postReady))
+			require.True(t, postReady.DeletionTimestamp.IsZero())
+		})
+	}
+}
+
+func TestSourceComponentDeletionPreservesSharedPostReadyRestore(t *testing.T) {
+	for _, tc := range []struct {
+		name    string
+		handoff bool
+	}{
+		{"protected terminal target", false},
+		{"deleting target after protection handoff", true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx := context.Background()
+			scheme, cluster, component, _, target := parentRestoreObjects(t)
+			now := metav1.Now()
+			component.DeletionTimestamp = &now
+			component.Finalizers = []string{"example.io/app-owner"}
+			target.Finalizers = []string{"example.io/app-owner"}
+			if tc.handoff {
+				target.DeletionTimestamp = &now
+			} else {
+				target.Finalizers = append(target.Finalizers, dptypes.DataProtectionFinalizerName)
+			}
+			target.Labels[dptypes.ComponentUIDLabelKey] = string(component.UID)
+			target.Status.Conditions = []corev1.PersistentVolumeClaimCondition{{
+				Type: corev1.PersistentVolumeClaimConditionType(kbappsv1.ConditionTypeRestore), Status: corev1.ConditionTrue,
+			}}
+			otherSource := dependencyRestorePVC("data-tikv-0", "tikv", "other-source-pvc-uid")
+			postReady := postReadyRestoreForComponent(target, cluster, component)
+			postReady.Finalizers = []string{"example.io/restore-owner"}
+			cli := fake.NewClientBuilder().WithScheme(scheme).
+				WithObjects(cluster, component, target, otherSource, postReady).Build()
+			vp := &VolumePopulatorReconciler{Client: cli, Scheme: scheme}
+
+			require.Equal(t, []reconcile.Request{{NamespacedName: client.ObjectKeyFromObject(otherSource)}},
+				vp.mapRestoreToPVCs(ctx, postReady))
+			requests := vp.mapComponentToPVCs(ctx, component)
+			require.Equal(t, []reconcile.Request{{NamespacedName: client.ObjectKeyFromObject(target)}}, requests)
+			require.NoError(t, cli.Get(ctx, requests[0].NamespacedName, target))
+			require.NoError(t, vp.syncPVC(intctrlutil.RequestCtx{Ctx: ctx}, target))
+
+			require.NoError(t, cli.Get(ctx, client.ObjectKeyFromObject(postReady), postReady))
+			require.True(t, postReady.DeletionTimestamp.IsZero())
+			require.Equal(t, []string{"example.io/restore-owner"}, postReady.Finalizers)
+			require.NoError(t, cli.Get(ctx, client.ObjectKeyFromObject(target), target))
+			require.Equal(t, []string{"example.io/app-owner"}, target.Finalizers)
+		})
+	}
+}
+
+func TestComponentIdentityMismatchDoesNotAuthorizeTermination(t *testing.T) {
+	scheme, cluster, component, _, target := parentRestoreObjects(t)
+	component.UID = "replacement-component-uid"
+	now := metav1.Now()
+	component.DeletionTimestamp = &now
+	component.Finalizers = []string{"example.io/app-owner"}
+	target.Finalizers = []string{dptypes.DataProtectionFinalizerName}
+	target.Labels[dptypes.ComponentUIDLabelKey] = "original-component-uid"
+	execution := executionRestoreForTarget(target, cluster)
+	cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(cluster, component, target, execution).Build()
+	vp := &VolumePopulatorReconciler{Client: cli, Scheme: scheme}
+
+	terminated, err := vp.handleRestoreParentLifecycle(
+		intctrlutil.RequestCtx{Ctx: context.Background()}, target)
+
+	require.False(t, terminated)
+	require.ErrorContains(t, err, "Component UID changed")
+	require.NoError(t, cli.Get(context.Background(), client.ObjectKeyFromObject(execution), &dpv1alpha1.Restore{}))
+}
+
+func TestClusterDeletionPrecedesComponentIdentityValidation(t *testing.T) {
+	scheme, cluster, _, _, target := parentRestoreObjects(t)
+	now := metav1.Now()
+	cluster.DeletionTimestamp = &now
+	cluster.Finalizers = append(cluster.Finalizers, "example.io/app-owner")
+	target.Finalizers = []string{dptypes.DataProtectionFinalizerName}
+	target.Labels[dptypes.ComponentUIDLabelKey] = "component-that-no-longer-exists"
+	execution := executionRestoreForTarget(target, cluster)
+	execution.Finalizers = []string{"example.io/restore-owner"}
+	cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(cluster, target, execution).Build()
+	vp := &VolumePopulatorReconciler{Client: cli, Scheme: scheme}
+
+	err := vp.syncPVC(intctrlutil.RequestCtx{Ctx: context.Background()}, target)
+
+	require.ErrorContains(t, err, "waiting for Restore owners")
+	require.NoError(t, cli.Get(context.Background(), client.ObjectKeyFromObject(execution), execution))
+	require.False(t, execution.DeletionTimestamp.IsZero())
+}
+
+func TestUncommittedSourceComponentDeletionDoesNotStartRestore(t *testing.T) {
+	scheme, cluster, component, its, target := parentRestoreObjects(t)
+	now := metav1.Now()
+	component.DeletionTimestamp = &now
+	component.Finalizers = []string{"example.io/app-owner"}
+	delete(target.Labels, dptypes.ComponentUIDLabelKey)
+	target.Finalizers = nil
+	cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(cluster, component, its, target).Build()
+	vp := &VolumePopulatorReconciler{Client: cli, Scheme: scheme}
+
+	require.NoError(t, vp.syncPVC(intctrlutil.RequestCtx{Ctx: context.Background()}, target))
+
+	require.NoError(t, cli.Get(context.Background(), client.ObjectKeyFromObject(target), target))
+	require.Empty(t, target.Labels[dptypes.ComponentUIDLabelKey])
+	require.NotContains(t, target.Finalizers, dptypes.DataProtectionFinalizerName)
+	restores := &dpv1alpha1.RestoreList{}
+	require.NoError(t, cli.List(context.Background(), restores))
+	require.Empty(t, restores.Items)
+	helper := &corev1.PersistentVolumeClaim{}
+	require.True(t, apierrors.IsNotFound(cli.Get(context.Background(), types.NamespacedName{
+		Namespace: target.Namespace, Name: getPopulatePVCName(target.UID),
+	}, helper)))
+}
+
 func TestClusterLifecycleRegistersBeforeWaitingForProtection(t *testing.T) {
 	for _, existingFinalizer := range []bool{false, true} {
 		t.Run(fmt.Sprintf("existing-finalizer=%t", existingFinalizer), func(t *testing.T) {
@@ -4400,7 +4568,7 @@ func TestClusterLifecycleSafetyBoundaries(t *testing.T) {
 		cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(cluster, component, its, target).Build()
 		vp := &VolumePopulatorReconciler{Client: cli, Scheme: scheme}
 
-		terminated, err := vp.handleRestoreClusterLifecycle(intctrlutil.RequestCtx{Ctx: context.Background()}, target)
+		terminated, err := vp.handleRestoreParentLifecycle(intctrlutil.RequestCtx{Ctx: context.Background()}, target)
 		require.NoError(t, err)
 		require.False(t, terminated)
 	})
@@ -4489,7 +4657,7 @@ func TestClusterLifecycleRefusesForeignExecutionRestore(t *testing.T) {
 	cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(cluster, target, helper, foreign).Build()
 	vp := &VolumePopulatorReconciler{Client: cli, Scheme: scheme}
 
-	terminated, err := vp.handleRestoreClusterLifecycle(intctrlutil.RequestCtx{Ctx: context.Background()}, target)
+	terminated, err := vp.handleRestoreParentLifecycle(intctrlutil.RequestCtx{Ctx: context.Background()}, target)
 	require.True(t, terminated)
 	require.ErrorContains(t, err, "refusing to delete execution Restore")
 	require.NoError(t, cli.Get(context.Background(), client.ObjectKeyFromObject(foreign), &dpv1alpha1.Restore{}))