diff --git a/cmd/dataprotection/main.go b/cmd/dataprotection/main.go
index 4d7cc2c09ca..c28c7004d86 100644
--- a/cmd/dataprotection/main.go
+++ b/cmd/dataprotection/main.go
@@ -340,6 +340,14 @@ func main() {
 		os.Exit(1)
 	}
 
+	if err = (&dpcontrollers.ClusterRestoreReconciler{
+		Client:   mgr.GetClient(),
+		Recorder: mgr.GetEventRecorderFor("cluster-restore-controller"),
+	}).SetupWithManager(mgr); err != nil {
+		setupLog.Error(err, "unable to create controller", "controller", "ClusterRestore")
+		os.Exit(1)
+	}
+
 	if err = (&dpcontrollers.BackupScheduleReconciler{
 		Client:   dputils.NewCompatClient(mgr.GetClient()),
 		Scheme:   mgr.GetScheme(),
diff --git a/controllers/dataprotection/cluster_restore_controller.go b/controllers/dataprotection/cluster_restore_controller.go
new file mode 100644
index 00000000000..304a7b4bc82
--- /dev/null
+++ b/controllers/dataprotection/cluster_restore_controller.go
@@ -0,0 +1,198 @@
+/*
+Copyright (C) 2022-2026 ApeCloud Co., Ltd
+
+This file is part of KubeBlocks project
+
+This program is free software: you can redistribute it and/or modify
+it under the terms of the GNU Affero General Public License as published by
+the Free Software Foundation, either version 3 of the License, or
+(at your option) any later version.
+
+This program is distributed in the hope that it will be useful,
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+GNU Affero General Public License for more details.
+
+You should have received a copy of the GNU Affero General Public License
+along with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+package dataprotection
+
+import (
+	"context"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/tools/record"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
+	"sigs.k8s.io/controller-runtime/pkg/handler"
+	"sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+
+	appsv1 "github.com/apecloud/kubeblocks/apis/apps/v1"
+	dpv1alpha1 "github.com/apecloud/kubeblocks/apis/dataprotection/v1alpha1"
+	"github.com/apecloud/kubeblocks/pkg/constant"
+	intctrlutil "github.com/apecloud/kubeblocks/pkg/controllerutil"
+	dprestore "github.com/apecloud/kubeblocks/pkg/dataprotection/restore"
+	dptypes "github.com/apecloud/kubeblocks/pkg/dataprotection/types"
+)
+
+// ClusterRestoreReconciler coordinates the Cluster-level restore lifecycle.
+type ClusterRestoreReconciler struct {
+	client.Client
+	Recorder record.EventRecorder
+}
+
+// +kubebuilder:rbac:groups=apps.kubeblocks.io,resources=clusters,verbs=get;list;watch;patch;update
+// +kubebuilder:rbac:groups=apps.kubeblocks.io,resources=clusters/finalizers,verbs=update;patch
+// +kubebuilder:rbac:groups=dataprotection.kubeblocks.io,resources=restores,verbs=get;list;watch
+// +kubebuilder:rbac:groups=core,resources=persistentvolumeclaims,verbs=get;list;watch
+
+func (r *ClusterRestoreReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
+	reqCtx := intctrlutil.RequestCtx{
+		Ctx: ctx, Req: req,
+		Log:      log.FromContext(ctx).WithValues("cluster-restore", req.NamespacedName),
+		Recorder: r.Recorder,
+	}
+	cluster := &appsv1.Cluster{}
+	if err := r.Client.Get(ctx, req.NamespacedName, cluster); err != nil {
+		return intctrlutil.CheckedRequeueWithError(err, reqCtx.Log, "")
+	}
+
+	restoring, err := r.isClusterRestoring(ctx, cluster)
+	if err != nil {
+		return intctrlutil.CheckedRequeueWithError(err, reqCtx.Log, "failed to determine Cluster restore state")
+	}
+	if !restoring {
+		return r.releaseClusterRestoreProtection(reqCtx, cluster)
+	}
+	if !isClusterRestoreProtected(cluster) {
+		// If deletion began without our finalizer, this controller cannot acquire
+		// protection and must not delay Cluster deletion.
+		if !cluster.DeletionTimestamp.IsZero() {
+			return intctrlutil.Reconciled()
+		}
+		return r.protectClusterRestore(reqCtx, cluster)
+	}
+	if !cluster.DeletionTimestamp.IsZero() {
+		return intctrlutil.RequeueAfter(reconcileInterval, reqCtx.Log,
+			"waiting for restore owners to finish Cluster termination")
+	}
+	return intctrlutil.Reconciled()
+}
+
+func (r *ClusterRestoreReconciler) SetupWithManager(mgr ctrl.Manager) error {
+	return intctrlutil.NewControllerManagedBy(mgr).
+		For(&appsv1.Cluster{}).
+		Watches(&dpv1alpha1.Restore{}, handler.EnqueueRequestsFromMapFunc(r.mapObjectToCluster)).
+		Watches(&corev1.PersistentVolumeClaim{}, handler.EnqueueRequestsFromMapFunc(r.mapObjectToCluster)).
+		Complete(r)
+}
+
+func (r *ClusterRestoreReconciler) mapObjectToCluster(_ context.Context, obj client.Object) []reconcile.Request {
+	clusterName := obj.GetLabels()[constant.AppInstanceLabelKey]
+	if clusterName == "" {
+		return nil
+	}
+	return []reconcile.Request{{NamespacedName: client.ObjectKey{Namespace: obj.GetNamespace(), Name: clusterName}}}
+}
+
+func (r *ClusterRestoreReconciler) isClusterRestoring(ctx context.Context,
+	cluster *appsv1.Cluster) (bool, error) {
+	if cluster.DeletionTimestamp.IsZero() && clusterAllowsRestoreProgress(cluster) {
+		return true, nil
+	}
+
+	// Inspect PVCs before Restores. VP releases temporary target protection only
+	// after observing postReady Restore, so this order closes the handoff window.
+	pvcs := &corev1.PersistentVolumeClaimList{}
+	if err := r.Client.List(ctx, pvcs, client.InNamespace(cluster.Namespace), client.MatchingLabels{
+		constant.AppInstanceLabelKey: cluster.Name,
+	}); err != nil {
+		return false, err
+	}
+	for i := range pvcs.Items {
+		pvc := &pvcs.Items[i]
+		if pvc.Labels[dptypes.ClusterUIDLabelKey] != string(cluster.UID) {
+			continue
+		}
+		if isClusterRestoreHelperPVC(pvc) ||
+			(isClusterRestoreTargetPVC(pvc) && controllerutil.ContainsFinalizer(pvc, dptypes.DataProtectionFinalizerName)) {
+			return true, nil
+		}
+	}
+
+	restores := &dpv1alpha1.RestoreList{}
+	if err := r.Client.List(ctx, restores, client.InNamespace(cluster.Namespace), client.MatchingLabels{
+		constant.AppInstanceLabelKey: cluster.Name,
+	}); err != nil {
+		return false, err
+	}
+	for i := range restores.Items {
+		restore := &restores.Items[i]
+		if restore.Labels[dprestore.DataProtectionRestoreLabelKey] != restore.Name {
+			continue
+		}
+		owned := restore.Labels[dptypes.ClusterUIDLabelKey] == string(cluster.UID)
+		terminal := restore.Status.Phase == dpv1alpha1.RestorePhaseCompleted ||
+			restore.Status.Phase == dpv1alpha1.RestorePhaseFailed
+		if owned && (!cluster.DeletionTimestamp.IsZero() || !terminal || !restore.DeletionTimestamp.IsZero()) {
+			return true, nil
+		}
+	}
+	return false, nil
+}
+
+func clusterAllowsRestoreProgress(cluster *appsv1.Cluster) bool {
+	if cluster.Spec.Restore == nil {
+		return false
+	}
+	condition := meta.FindStatusCondition(cluster.Status.Conditions, appsv1.ConditionTypeRestore)
+	// Restore=False is terminal for status aggregation, but the Cluster still
+	// has initial-restore intent. Keep the lifecycle active until deletion so
+	// PVC restores cannot lose protection while converging on the failure.
+	return condition == nil || condition.Status != metav1.ConditionTrue
+}
+
+func isClusterRestoreProtected(cluster *appsv1.Cluster) bool {
+	return controllerutil.ContainsFinalizer(cluster, dptypes.RestoreProtectionFinalizerName)
+}
+
+func (r *ClusterRestoreReconciler) protectClusterRestore(reqCtx intctrlutil.RequestCtx,
+	cluster *appsv1.Cluster) (ctrl.Result, error) {
+	if isClusterRestoreProtected(cluster) {
+		return intctrlutil.Reconciled()
+	}
+	patch := client.MergeFromWithOptions(cluster.DeepCopy(), client.MergeFromWithOptimisticLock{})
+	controllerutil.AddFinalizer(cluster, dptypes.RestoreProtectionFinalizerName)
+	if err := r.Client.Patch(reqCtx.Ctx, cluster, patch); err != nil {
+		return intctrlutil.CheckedRequeueWithError(err, reqCtx.Log, "failed to add Cluster restore-protection finalizer")
+	}
+	return intctrlutil.Reconciled()
+}
+
+func (r *ClusterRestoreReconciler) releaseClusterRestoreProtection(reqCtx intctrlutil.RequestCtx,
+	cluster *appsv1.Cluster) (ctrl.Result, error) {
+	if !isClusterRestoreProtected(cluster) {
+		return intctrlutil.Reconciled()
+	}
+	patch := client.MergeFromWithOptions(cluster.DeepCopy(), client.MergeFromWithOptimisticLock{})
+	controllerutil.RemoveFinalizer(cluster, dptypes.RestoreProtectionFinalizerName)
+	if err := r.Client.Patch(reqCtx.Ctx, cluster, patch); err != nil {
+		return intctrlutil.CheckedRequeueWithError(err, reqCtx.Log, "failed to remove Cluster restore-protection finalizer")
+	}
+	return intctrlutil.Reconciled()
+}
+
+func isClusterRestoreHelperPVC(pvc *corev1.PersistentVolumeClaim) bool {
+	return pvc.Labels[dprestore.DataProtectionPopulatePVCLabelKey] != ""
+}
+
+func isClusterRestoreTargetPVC(pvc *corev1.PersistentVolumeClaim) bool {
+	return pvc.Spec.DataSourceRef != nil && pvc.Spec.DataSourceRef.APIGroup != nil &&
+		*pvc.Spec.DataSourceRef.APIGroup == dptypes.DataprotectionAPIGroup
+}
diff --git a/controllers/dataprotection/cluster_restore_controller_test.go b/controllers/dataprotection/cluster_restore_controller_test.go
new file mode 100644
index 00000000000..051960f4eea
--- /dev/null
+++ b/controllers/dataprotection/cluster_restore_controller_test.go
@@ -0,0 +1,149 @@
+/*
+Copyright (C) 2022-2026 ApeCloud Co., Ltd
+
+This file is part of KubeBlocks project
+
+This program is free software: you can redistribute it and/or modify
+it under the terms of the GNU Affero General Public License as published by
+the Free Software Foundation, either version 3 of the License, or
+(at your option) any later version.
+
+This program is distributed in the hope that it will be useful,
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+GNU Affero General Public License for more details.
+
+You should have received a copy of the GNU Affero General Public License
+along with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+package dataprotection
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+
+	appsv1 "github.com/apecloud/kubeblocks/apis/apps/v1"
+	dpv1alpha1 "github.com/apecloud/kubeblocks/apis/dataprotection/v1alpha1"
+	dptypes "github.com/apecloud/kubeblocks/pkg/dataprotection/types"
+)
+
+func TestClusterRestoreProtectionLifecycle(t *testing.T) {
+	for _, tc := range []struct {
+		name      string
+		deleting  bool
+		protected bool
+		status    metav1.ConditionStatus
+		resource  string
+		wantKeep  bool
+		wantWait  bool
+	}{
+		{"initial intent", false, false, metav1.ConditionUnknown, "", true, false},
+		{"deletion without protection", true, false, metav1.ConditionUnknown, "target", false, false},
+		{"failed restore", false, true, metav1.ConditionFalse, "", true, false},
+		{"successful restore", false, true, metav1.ConditionTrue, "completed restore", false, false},
+		{"target still protected", false, true, metav1.ConditionTrue, "target", true, false},
+		{"helper remains", false, true, metav1.ConditionTrue, "helper", true, false},
+		{"execution still running", false, true, metav1.ConditionTrue, "running restore", true, false},
+		{"deletion waits for target", true, true, metav1.ConditionTrue, "target", true, true},
+		{"deletion waits for helper", true, true, metav1.ConditionTrue, "helper", true, true},
+		{"deletion waits for failed restore", true, true, metav1.ConditionTrue, "failed restore", true, true},
+		{"deletion waits for completed restore", true, true, metav1.ConditionTrue, "completed restore", true, true},
+		{"deletion cleanup finished", true, true, metav1.ConditionTrue, "", false, false},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx := context.Background()
+			scheme, cluster, _, _, target := parentRestoreObjects(t)
+			cluster.Finalizers = []string{"example.io/app-owner"}
+			if tc.protected {
+				cluster.Finalizers = append(cluster.Finalizers, dptypes.RestoreProtectionFinalizerName)
+			}
+			if tc.deleting {
+				now := metav1.Now()
+				cluster.DeletionTimestamp = &now
+			}
+			cluster.Status.Conditions = []metav1.Condition{{
+				Type: appsv1.ConditionTypeRestore, Status: tc.status,
+			}}
+			var resource client.Object
+			switch tc.resource {
+			case "target":
+				target.Finalizers = []string{dptypes.DataProtectionFinalizerName}
+				resource = target
+			case "helper":
+				resource = restoreHelperForTarget(target, cluster)
+			case "running restore", "failed restore", "completed restore":
+				restore := executionRestoreForTarget(target, cluster)
+				switch tc.resource {
+				case "failed restore":
+					restore.Status.Phase = dpv1alpha1.RestorePhaseFailed
+				case "completed restore":
+					restore.Status.Phase = dpv1alpha1.RestorePhaseCompleted
+				default:
+					restore.Status.Phase = dpv1alpha1.RestorePhaseRunning
+				}
+				restore.Finalizers = []string{dptypes.DataProtectionFinalizerName}
+				resource = restore
+			}
+			objects := []client.Object{cluster}
+			if resource != nil {
+				objects = append(objects, resource)
+			}
+			cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(objects...).Build()
+			if resource != nil {
+				require.NoError(t, cli.Get(ctx, client.ObjectKeyFromObject(resource), resource))
+			}
+			reconciler := &ClusterRestoreReconciler{Client: cli}
+
+			result, err := reconciler.Reconcile(ctx, ctrl.Request{NamespacedName: client.ObjectKeyFromObject(cluster)})
+			require.NoError(t, err)
+			require.Equal(t, tc.wantWait, result.RequeueAfter > 0)
+			require.NoError(t, cli.Get(ctx, client.ObjectKeyFromObject(cluster), cluster))
+			expected := []string{"example.io/app-owner"}
+			if tc.wantKeep {
+				expected = append(expected, dptypes.RestoreProtectionFinalizerName)
+			}
+			require.ElementsMatch(t, expected, cluster.Finalizers)
+			if resource != nil {
+				current := resource.DeepCopyObject().(client.Object)
+				require.NoError(t, cli.Get(ctx, client.ObjectKeyFromObject(resource), current))
+				require.Equal(t, resource, current)
+			}
+		})
+	}
+}
+
+func TestClusterRestoreControllerIgnoresResourcesWithoutExactClusterUID(t *testing.T) {
+	for _, uid := range []string{"", "another-cluster-uid"} {
+		t.Run("uid="+uid, func(t *testing.T) {
+			ctx := context.Background()
+			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}
+			helper := restoreHelperForTarget(target, cluster)
+			restore := executionRestoreForTarget(target, cluster)
+			for _, obj := range []client.Object{target, helper, restore} {
+				obj.GetLabels()[dptypes.ClusterUIDLabelKey] = uid
+			}
+			cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(cluster, target, helper, restore).Build()
+			reconciler := &ClusterRestoreReconciler{Client: cli}
+
+			_, err := reconciler.Reconcile(ctx, ctrl.Request{NamespacedName: client.ObjectKeyFromObject(cluster)})
+			require.NoError(t, err)
+			require.NoError(t, cli.Get(ctx, client.ObjectKeyFromObject(cluster), cluster))
+			require.Equal(t, []string{"example.io/app-owner"}, cluster.Finalizers)
+			require.NoError(t, cli.Get(ctx, client.ObjectKeyFromObject(target), &corev1.PersistentVolumeClaim{}))
+			require.NoError(t, cli.Get(ctx, client.ObjectKeyFromObject(helper), &corev1.PersistentVolumeClaim{}))
+			require.NoError(t, cli.Get(ctx, client.ObjectKeyFromObject(restore), &dpv1alpha1.Restore{}))
+		})
+	}
+}
diff --git a/controllers/dataprotection/suite_test.go b/controllers/dataprotection/suite_test.go
index 337d93d66f2..9da6f79fa2b 100644
--- a/controllers/dataprotection/suite_test.go
+++ b/controllers/dataprotection/suite_test.go
@@ -187,6 +187,12 @@ var _ = BeforeSuite(func() {
 	}).SetupWithManager(k8sManager)
 	Expect(err).ToNot(HaveOccurred())
 
+	err = (&ClusterRestoreReconciler{
+		Client:   k8sClient,
+		Recorder: k8sManager.GetEventRecorderFor("cluster-restore-controller"),
+	}).SetupWithManager(k8sManager)
+	Expect(err).ToNot(HaveOccurred())
+
 	err = (&BackupScheduleReconciler{
 		Client:   k8sClient,
 		Scheme:   k8sManager.GetScheme(),
diff --git a/controllers/dataprotection/volumepopulator_controller.go b/controllers/dataprotection/volumepopulator_controller.go
index e4811694246..27731830954 100644
--- a/controllers/dataprotection/volumepopulator_controller.go
+++ b/controllers/dataprotection/volumepopulator_controller.go
@@ -60,7 +60,7 @@ import (
 	viper "github.com/apecloud/kubeblocks/pkg/viperx"
 )
 
-// VolumePopulatorReconciler reconciles PVCs with Backup or Restore data sources.
+// VolumePopulatorReconciler coordinates data population and restore for PVCs.
 type VolumePopulatorReconciler struct {
 	client.Client
 	Scheme   *runtime.Scheme
@@ -88,7 +88,7 @@ type pvcRestoreDecision struct {
 
 // +kubebuilder:rbac:groups=core,resources=persistentvolumeclaims/status,verbs=get;update;patch
 // +kubebuilder:rbac:groups=core,resources=persistentvolumeclaims/finalizers,verbs=update
-// +kubebuilder:rbac:groups=dataprotection.kubeblocks.io,resources=restores,verbs=get;list;watch
+// +kubebuilder:rbac:groups=dataprotection.kubeblocks.io,resources=restores,verbs=get;list;watch;delete
 // +kubebuilder:rbac:groups=apps.kubeblocks.io,resources=clusters,verbs=get;list;watch
 // +kubebuilder:rbac:groups=apps.kubeblocks.io,resources=components,verbs=get;list;watch
 // +kubebuilder:rbac:groups=apps.kubeblocks.io,resources=componentdefinitions,verbs=get;list;watch
@@ -213,7 +213,7 @@ func (r *VolumePopulatorReconciler) mapClusterToPVCs(ctx context.Context, obj cl
 	}
 	return r.mapRestorePVCs(ctx, cluster.Namespace, client.MatchingLabels{
 		constant.AppInstanceLabelKey: cluster.Name,
-	}, string(cluster.UID), false)
+	}, string(cluster.UID), !cluster.DeletionTimestamp.IsZero())
 }
 
 func (r *VolumePopulatorReconciler) mapRestorePVCs(ctx context.Context, namespace string,
@@ -343,7 +343,9 @@ func clusterDependencyPredicate() predicate.Predicate {
 			oldCluster, oldOK := e.ObjectOld.(*appsv1.Cluster)
 			newCluster, newOK := e.ObjectNew.(*appsv1.Cluster)
 			return oldOK && newOK && (oldCluster.Status.Phase != newCluster.Status.Phase ||
-				!reflect.DeepEqual(oldCluster.DeletionTimestamp, newCluster.DeletionTimestamp))
+				!reflect.DeepEqual(oldCluster.DeletionTimestamp, newCluster.DeletionTimestamp) ||
+				controllerutil.ContainsFinalizer(oldCluster, dptypes.RestoreProtectionFinalizerName) !=
+					controllerutil.ContainsFinalizer(newCluster, dptypes.RestoreProtectionFinalizerName))
 		},
 	}
 }
@@ -394,6 +396,17 @@ func (r *VolumePopulatorReconciler) syncPVC(reqCtx intctrlutil.RequestCtx, pvc *
 	if !matched {
 		return nil
 	}
+	terminated, err := r.handleRestoreClusterLifecycle(reqCtx, pvc)
+	if err != nil || terminated {
+		return err
+	}
+	// Parent deletion is checked first because it authorizes cleanup even after
+	// target protection has been handed off. Target deletion alone remains a
+	// no-op, and Kubernetes does not allow acquiring a new finalizer here.
+	if !pvc.DeletionTimestamp.IsZero() &&
+		!controllerutil.ContainsFinalizer(pvc, dptypes.DataProtectionFinalizerName) {
+		return nil
+	}
 	// A non-deleting bound PVC with a terminal Restore condition does not need
 	// its source Backup/Restore. Populating can finish while postReady is pending.
 	if pvc.Spec.VolumeName != "" && pvc.DeletionTimestamp.IsZero() && pvcRestoreTerminal(pvc) {
@@ -418,6 +431,351 @@ 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,
+	pvc *corev1.PersistentVolumeClaim) (bool, error) {
+	clusterName := pvc.Labels[constant.AppInstanceLabelKey]
+	componentName := pvc.Labels[constant.KBAppComponentLabelKey]
+	if clusterName == "" || componentName == "" {
+		return false, nil
+	}
+	hasClusterIdentity := pvc.Annotations[constant.KBAppClusterUIDKey] != "" ||
+		pvc.Labels[dptypes.ClusterUIDLabelKey] != ""
+	// App labels alone do not establish Cluster restore identity; standalone DP
+	// restores may use the same labels.
+	if !hasClusterIdentity {
+		return false, nil
+	}
+
+	cluster := &appsv1.Cluster{}
+	clusterKey := types.NamespacedName{Namespace: pvc.Namespace, Name: clusterName}
+	if err := r.Client.Get(reqCtx.Ctx, clusterKey, cluster); err != nil {
+		if apierrors.IsNotFound(err) {
+			if !controllerutil.ContainsFinalizer(pvc, dptypes.DataProtectionFinalizerName) {
+				return true, nil
+			}
+			// Registration can lose the race with Cluster deletion, but no work
+			// starts until a later reconcile has rechecked the parent.
+			if releaseErr := r.releaseUnusedTargetFinalizer(reqCtx, pvc); releaseErr != nil {
+				return true, restoreParentRequeue(releaseErr)
+			}
+			return true, nil
+		}
+		return false, restoreParentRequeue(err)
+	}
+	for _, clusterUID := range []string{
+		pvc.Annotations[constant.KBAppClusterUIDKey],
+		pvc.Labels[dptypes.ClusterUIDLabelKey],
+	} {
+		if clusterUID != "" && clusterUID != string(cluster.UID) {
+			return false, restoreParentRequeue(fmt.Errorf(
+				"PVC %s/%s identifies Cluster %s/%s UID %s, not current UID %s",
+				pvc.Namespace, pvc.Name, cluster.Namespace, cluster.Name, clusterUID, cluster.UID))
+		}
+	}
+	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 {
+			return false, restoreParentRequeue(err)
+		}
+	}
+	// Aggregate restore status gates normal progression, not owner cleanup.
+	if !clusterAllowsRestoreProgress(cluster) && !pvcRestoreTerminal(pvc) {
+		return false, intctrlutil.NewRequeueError(reconcileInterval, "Cluster restore is no longer active")
+	}
+	if !committed {
+		if pvcRestoreTerminal(pvc) && !controllerutil.ContainsFinalizer(pvc, dptypes.DataProtectionFinalizerName) {
+			return false, nil
+		}
+		comp, err := r.validateClusterRestorePVCOwnership(reqCtx.Ctx, pvc, cluster)
+		if err != nil {
+			return false, restoreParentRequeue(err)
+		}
+		if err = r.registerVolumePopulation(reqCtx.Ctx, pvc, cluster, comp); err != nil {
+			return false, restoreParentRequeue(err)
+		}
+		return false, intctrlutil.NewRequeueError(reconcileInterval, "waiting for target PVC restore protection")
+	}
+	if !controllerutil.ContainsFinalizer(cluster, dptypes.RestoreProtectionFinalizerName) {
+		if pvcRestoreTerminal(pvc) && !controllerutil.ContainsFinalizer(pvc, dptypes.DataProtectionFinalizerName) {
+			return false, nil
+		}
+		return false, intctrlutil.NewRequeueError(reconcileInterval,
+			"waiting for Cluster restore-protection finalizer")
+	}
+	if !pvcPopulateReleased(pvc) && !pvcRestoreTerminal(pvc) {
+		if err := r.ensureTargetFinalizer(reqCtx, pvc); err != nil {
+			return false, err
+		}
+	}
+	return false, nil
+}
+
+// releaseUnusedTargetFinalizer rolls back an unused registration after its
+// Cluster disappears. Existing population or Restore resources retain it.
+func (r *VolumePopulatorReconciler) releaseUnusedTargetFinalizer(reqCtx intctrlutil.RequestCtx,
+	pvc *corev1.PersistentVolumeClaim) error {
+	clusterUID := clusterRestorePVCUID(pvc)
+	if clusterUID == "" || pvc.Labels[dptypes.ComponentUIDLabelKey] == "" {
+		return fmt.Errorf("restore PVC %s/%s has no committed parent identity", pvc.Namespace, pvc.Name)
+	}
+	if r.ContainPopulatingCondition(pvc) && !pvcPopulateReleased(pvc) {
+		return fmt.Errorf("cluster is missing after population started for PVC %s/%s", pvc.Namespace, pvc.Name)
+	}
+	key := types.NamespacedName{Namespace: pvc.Namespace, Name: getPopulatePVCName(pvc.UID)}
+	for _, obj := range []client.Object{&corev1.PersistentVolumeClaim{}, &dpv1alpha1.Restore{}} {
+		if err := r.Client.Get(reqCtx.Ctx, key, obj); !apierrors.IsNotFound(err) {
+			if err != nil {
+				return err
+			}
+			return fmt.Errorf("cluster is missing while restore resources for PVC %s/%s remain", pvc.Namespace, pvc.Name)
+		}
+	}
+
+	list := &dpv1alpha1.RestoreList{}
+	if err := r.Client.List(reqCtx.Ctx, list, client.InNamespace(pvc.Namespace), client.MatchingLabels{
+		constant.AppInstanceLabelKey: pvc.Labels[constant.AppInstanceLabelKey],
+		dptypes.ClusterUIDLabelKey:   clusterUID,
+	}); err != nil {
+		return err
+	}
+	for i := range list.Items {
+		if internalPostReadyRestoreOwner(&list.Items[i]) != nil {
+			return fmt.Errorf("cluster is missing while postReady Restore %s/%s remains",
+				list.Items[i].Namespace, list.Items[i].Name)
+		}
+	}
+	return r.releaseTargetPVC(reqCtx, pvc)
+}
+
+func volumePopulationIdentityCommitted(pvc *corev1.PersistentVolumeClaim, cluster *appsv1.Cluster) bool {
+	return pvc.Labels[dptypes.ClusterUIDLabelKey] == string(cluster.UID) &&
+		pvc.Labels[dptypes.ComponentUIDLabelKey] != ""
+}
+
+// committedVolumePopulationComponent resolves the Component identity recorded
+// before VP creates restore resources. Retention may detach the workload owner
+// chain, but it does not change this identity.
+func (r *VolumePopulatorReconciler) committedVolumePopulationComponent(ctx context.Context,
+	pvc *corev1.PersistentVolumeClaim, cluster *appsv1.Cluster) (*appsv1.Component, error) {
+	componentName := pvc.Labels[constant.KBAppComponentLabelKey]
+	comp := &appsv1.Component{}
+	key := types.NamespacedName{
+		Namespace: pvc.Namespace,
+		Name:      constant.GenerateClusterComponentName(cluster.Name, componentName),
+	}
+	if err := r.Client.Get(ctx, key, comp); err != nil {
+		return nil, err
+	}
+	if string(comp.UID) != pvc.Labels[dptypes.ComponentUIDLabelKey] {
+		return nil, fmt.Errorf("restore PVC %s/%s Component UID changed from %s to %s",
+			pvc.Namespace, pvc.Name, pvc.Labels[dptypes.ComponentUIDLabelKey], comp.UID)
+	}
+	return comp, nil
+}
+
+func restoreParentRequeue(err error) error {
+	return intctrlutil.NewRequeueError(reconcileInterval, err.Error())
+}
+
+func (r *VolumePopulatorReconciler) terminateClusterVolumePopulation(reqCtx intctrlutil.RequestCtx,
+	pvc *corev1.PersistentVolumeClaim, cluster *appsv1.Cluster) (bool, error) {
+	err := r.cleanupClusterVolumePopulation(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)
+	if err != nil {
+		return err
+	}
+	postReadyPending, err := r.deleteClusterPostReadyRestoresAndWait(reqCtx.Ctx, cluster)
+	if err != nil {
+		return err
+	}
+	if pending || postReadyPending {
+		return intctrlutil.NewRequeueError(reconcileInterval, "waiting for Restore owners to finish termination")
+	}
+	pending, err = r.deletePopulatePVCAndWait(reqCtx.Ctx, pvc, cluster)
+	if err != nil {
+		return err
+	}
+	if pending {
+		return intctrlutil.NewRequeueError(reconcileInterval, "waiting for helper PVC to disappear")
+	}
+	return r.releaseTargetPVC(reqCtx, pvc)
+}
+
+func (r *VolumePopulatorReconciler) deleteExecutionRestoreAndWait(ctx context.Context,
+	pvc *corev1.PersistentVolumeClaim, cluster *appsv1.Cluster) (bool, error) {
+	restore := &dpv1alpha1.Restore{}
+	key := types.NamespacedName{Namespace: pvc.Namespace, Name: getPopulatePVCName(pvc.UID)}
+	if err := r.Client.Get(ctx, key, restore); err != nil {
+		return false, client.IgnoreNotFound(err)
+	}
+	if restore.Labels[dptypes.ClusterUIDLabelKey] != string(cluster.UID) ||
+		restore.Labels[dprestore.DataProtectionRestoreLabelKey] != restore.Name ||
+		!hasExactOwnerReference(restore.OwnerReferences, corev1.SchemeGroupVersion.String(),
+			"PersistentVolumeClaim", pvc.Name, pvc.UID) {
+		return false, fmt.Errorf("refusing to delete execution Restore %s/%s without exact VP ownership",
+			restore.Namespace, restore.Name)
+	}
+	if restore.DeletionTimestamp.IsZero() {
+		if err := r.Client.Delete(ctx, restore); err != nil && !apierrors.IsNotFound(err) {
+			return false, err
+		}
+	}
+	return true, nil
+}
+
+func (r *VolumePopulatorReconciler) deleteClusterPostReadyRestoresAndWait(ctx context.Context,
+	cluster *appsv1.Cluster) (bool, error) {
+	list := &dpv1alpha1.RestoreList{}
+	if err := r.Client.List(ctx, list, client.InNamespace(cluster.Namespace), client.MatchingLabels{
+		constant.AppInstanceLabelKey: cluster.Name,
+		dptypes.ClusterUIDLabelKey:   string(cluster.UID),
+	}); err != nil {
+		return false, err
+	}
+	pending := false
+	for i := range list.Items {
+		restore := &list.Items[i]
+		if internalPostReadyRestoreOwner(restore) == nil {
+			continue
+		}
+		pending = true
+		if !restore.DeletionTimestamp.IsZero() {
+			continue
+		}
+		if err := r.Client.Delete(ctx, restore); err != nil && !apierrors.IsNotFound(err) {
+			return false, err
+		}
+	}
+	return pending, nil
+}
+
+func (r *VolumePopulatorReconciler) deletePopulatePVCAndWait(ctx context.Context,
+	pvc *corev1.PersistentVolumeClaim, cluster *appsv1.Cluster) (bool, error) {
+	helper := &corev1.PersistentVolumeClaim{}
+	key := types.NamespacedName{Namespace: pvc.Namespace, Name: getPopulatePVCName(pvc.UID)}
+	if err := r.Client.Get(ctx, key, helper); err != nil {
+		return false, client.IgnoreNotFound(err)
+	}
+	if helper.Labels[dptypes.ClusterUIDLabelKey] != string(cluster.UID) ||
+		helper.Labels[dprestore.DataProtectionPopulatePVCLabelKey] != helper.Name {
+		return false, fmt.Errorf("refusing to delete helper PVC %s/%s without exact VP identity",
+			helper.Namespace, helper.Name)
+	}
+	if helper.DeletionTimestamp.IsZero() {
+		if err := r.Client.Delete(ctx, helper); err != nil && !apierrors.IsNotFound(err) {
+			return false, err
+		}
+	}
+	return true, nil
+}
+
+func hasExactOwnerReference(refs []metav1.OwnerReference, apiVersion, kind, name string, uid types.UID) bool {
+	for i := range refs {
+		ref := refs[i]
+		if ref.APIVersion == apiVersion && ref.Kind == kind && ref.Name == name && ref.UID == uid {
+			return true
+		}
+	}
+	return false
+}
+
+func (r *VolumePopulatorReconciler) validateClusterRestorePVCOwnership(ctx context.Context,
+	pvc *corev1.PersistentVolumeClaim, cluster *appsv1.Cluster) (*appsv1.Component, error) {
+	owner := metav1.GetControllerOf(pvc)
+	if owner == nil || owner.APIVersion != workloads.GroupVersion.String() {
+		return nil, fmt.Errorf("restore PVC %s/%s has no supported workload controller owner", pvc.Namespace, pvc.Name)
+	}
+	var itsOwner *metav1.OwnerReference
+	switch owner.Kind {
+	case workloads.InstanceSetKind:
+		itsOwner = owner
+	case "Instance":
+		instance := &workloads.Instance{}
+		if err := r.Client.Get(ctx, client.ObjectKey{Namespace: pvc.Namespace, Name: owner.Name}, instance); err != nil {
+			return nil, err
+		}
+		if instance.UID != owner.UID {
+			return nil, fmt.Errorf("restore PVC %s/%s Instance owner UID does not match", pvc.Namespace, pvc.Name)
+		}
+		itsOwner = metav1.GetControllerOf(instance)
+		if itsOwner == nil || itsOwner.APIVersion != workloads.GroupVersion.String() ||
+			itsOwner.Kind != workloads.InstanceSetKind {
+			return nil, fmt.Errorf("instance %s/%s has no InstanceSet controller owner", instance.Namespace, instance.Name)
+		}
+	default:
+		return nil, fmt.Errorf("restore PVC %s/%s has unsupported workload owner kind %s",
+			pvc.Namespace, pvc.Name, owner.Kind)
+	}
+
+	its := &workloads.InstanceSet{}
+	if err := r.Client.Get(ctx, client.ObjectKey{Namespace: pvc.Namespace, Name: itsOwner.Name}, its); err != nil {
+		return nil, err
+	}
+	if its.UID != itsOwner.UID {
+		return nil, fmt.Errorf("restore PVC %s/%s InstanceSet owner UID does not match", pvc.Namespace, pvc.Name)
+	}
+	if its.Labels[constant.AppInstanceLabelKey] != cluster.Name ||
+		its.Labels[constant.KBAppComponentLabelKey] != pvc.Labels[constant.KBAppComponentLabelKey] {
+		return nil, fmt.Errorf("InstanceSet %s/%s does not match restore PVC parent identity", its.Namespace, its.Name)
+	}
+	componentOwner := metav1.GetControllerOf(its)
+	if componentOwner == nil || componentOwner.APIVersion != appsv1.GroupVersion.String() ||
+		componentOwner.Kind != appsv1.ComponentKind {
+		return nil, fmt.Errorf("InstanceSet %s/%s has no Component controller owner", its.Namespace, its.Name)
+	}
+	comp := &appsv1.Component{}
+	if err := r.Client.Get(ctx, client.ObjectKey{Namespace: pvc.Namespace, Name: componentOwner.Name}, comp); err != nil {
+		return nil, err
+	}
+	if comp.UID != componentOwner.UID || comp.Labels[constant.AppInstanceLabelKey] != cluster.Name ||
+		comp.Labels[constant.KBAppComponentLabelKey] != pvc.Labels[constant.KBAppComponentLabelKey] {
+		return nil, fmt.Errorf("InstanceSet %s/%s is not owned by the PVC Component in Cluster %s/%s",
+			its.Namespace, its.Name, cluster.Namespace, cluster.Name)
+	}
+	clusterOwner := metav1.GetControllerOf(comp)
+	if clusterOwner == nil || clusterOwner.APIVersion != appsv1.GroupVersion.String() ||
+		clusterOwner.Kind != appsv1.ClusterKind || clusterOwner.Name != cluster.Name || clusterOwner.UID != cluster.UID {
+		return nil, fmt.Errorf("component %s/%s is not owned by current Cluster UID %s",
+			comp.Namespace, comp.Name, cluster.UID)
+	}
+	return comp, nil
+}
+
+// registerVolumePopulation records verified App ownership and target protection
+// together. The caller returns so the shared cache observes it before work starts.
+func (r *VolumePopulatorReconciler) registerVolumePopulation(ctx context.Context,
+	pvc *corev1.PersistentVolumeClaim, cluster *appsv1.Cluster, comp *appsv1.Component) error {
+	if uid := pvc.Labels[dptypes.ComponentUIDLabelKey]; uid != "" && uid != string(comp.UID) {
+		return fmt.Errorf("restore PVC %s/%s Component UID %s does not match %s",
+			pvc.Namespace, pvc.Name, uid, comp.UID)
+	}
+	originalPVC := pvc.DeepCopy()
+	pvc.Labels[dptypes.ClusterUIDLabelKey] = string(cluster.UID)
+	pvc.Labels[dptypes.ComponentUIDLabelKey] = string(comp.UID)
+	controllerutil.AddFinalizer(pvc, dptypes.DataProtectionFinalizerName)
+	return r.Client.Patch(ctx, pvc,
+		client.MergeFromWithOptions(originalPVC, client.MergeFromWithOptimisticLock{}))
+}
+
 // dispatchUnboundPVC routes an unbound PVC to either Populate or ProvisionOnly.
 // When mode is RestoreData but PrepareDataBackupSets is empty, it checks
 // PostReadyBackupSets: if postReady actions exist, fall back to ProvisionOnly
@@ -939,6 +1297,7 @@ func internalRestoreLabels(pvc *corev1.PersistentVolumeClaim) map[string]string
 		constant.KBAppComponentLabelKey,
 		constant.KBAppShardingNameLabelKey,
 		constant.VolumeClaimTemplateNameLabelKey,
+		dptypes.ComponentUIDLabelKey,
 	} {
 		if value := pvc.Labels[key]; value != "" {
 			labels[key] = value
@@ -1132,13 +1491,8 @@ func (r *VolumePopulatorReconciler) Populate(reqCtx intctrlutil.RequestCtx, pvc
 	if err != nil || wait {
 		return err
 	}
-	// Make sure the PVC finalizer is present
-	if !slices.Contains(pvc.Finalizers, dptypes.DataProtectionFinalizerName) {
-		pvcPatch := client.MergeFrom(pvc.DeepCopy())
-		controllerutil.AddFinalizer(pvc, dptypes.DataProtectionFinalizerName)
-		if err = r.Client.Patch(reqCtx.Ctx, pvc, pvcPatch); err != nil {
-			return err
-		}
+	if err = r.ensureTargetFinalizer(reqCtx, pvc); err != nil {
+		return err
 	}
 	if err = r.UpdatePVCConditions(reqCtx, pvc, ReasonPopulatingProcessing, "Populator started"); err != nil {
 		return err
@@ -1194,12 +1548,8 @@ func (r *VolumePopulatorReconciler) ProvisionOnly(reqCtx intctrlutil.RequestCtx,
 	if err != nil || wait {
 		return err
 	}
-	if !slices.Contains(pvc.Finalizers, dptypes.DataProtectionFinalizerName) {
-		pvcPatch := client.MergeFrom(pvc.DeepCopy())
-		controllerutil.AddFinalizer(pvc, dptypes.DataProtectionFinalizerName)
-		if err = r.Client.Patch(reqCtx.Ctx, pvc, pvcPatch); err != nil {
-			return err
-		}
+	if err = r.ensureTargetFinalizer(reqCtx, pvc); err != nil {
+		return err
 	}
 	if err = r.UpdatePVCConditions(reqCtx, pvc, ReasonPopulatingProcessing, "Provisioning PVC without data restore"); err != nil {
 		return err
@@ -1221,6 +1571,24 @@ func (r *VolumePopulatorReconciler) ProvisionOnly(reqCtx intctrlutil.RequestCtx,
 	return r.completeBoundPVCIfNeeded(reqCtx, pvc, restoreCtx)
 }
 
+func (r *VolumePopulatorReconciler) ensureTargetFinalizer(
+	reqCtx intctrlutil.RequestCtx, pvc *corev1.PersistentVolumeClaim) error {
+	if slices.Contains(pvc.Finalizers, dptypes.DataProtectionFinalizerName) {
+		return nil
+	}
+	pvcPatch := client.MergeFromWithOptions(pvc.DeepCopy(), client.MergeFromWithOptimisticLock{})
+	controllerutil.AddFinalizer(pvc, dptypes.DataProtectionFinalizerName)
+	if err := r.Client.Patch(reqCtx.Ctx, pvc, pvcPatch); err != nil {
+		return err
+	}
+	if clusterRestorePVCUID(pvc) != "" {
+		// Reobserve the marker through the shared PVC cache before another
+		// informer can expose newly-created restore resources.
+		return intctrlutil.NewRequeueError(reconcileInterval, "waiting for target PVC restore protection")
+	}
+	return nil
+}
+
 func (r *VolumePopulatorReconciler) completeBoundPVCIfNeeded(reqCtx intctrlutil.RequestCtx,
 	pvc *corev1.PersistentVolumeClaim,
 	restoreCtx *pvcRestoreContext) error {
@@ -1245,6 +1613,9 @@ func (r *VolumePopulatorReconciler) completeBoundPVCIfNeeded(reqCtx intctrlutil.
 	if !postReadyCompleted {
 		return intctrlutil.NewRequeueError(reconcileInterval, "waiting for postReady restore")
 	}
+	if err := r.releaseTargetPVC(reqCtx, pvc); err != nil {
+		return err
+	}
 	return r.UpdatePVCConditions(reqCtx, pvc, reason, message)
 }
 
@@ -1477,6 +1848,11 @@ func (r *VolumePopulatorReconciler) ensurePostReadyRestoreCompleted(reqCtx intct
 		if !apierrors.IsNotFound(err) {
 			return false, err
 		}
+		if clusterRestorePVCUID(pvc) != "" {
+			if err = r.ensureTargetFinalizer(reqCtx, pvc); err != nil {
+				return false, err
+			}
+		}
 		if err = r.Client.Create(reqCtx.Ctx, postReadyRestore); err != nil && !apierrors.IsAlreadyExists(err) {
 			return false, err
 		}
@@ -1488,6 +1864,13 @@ func (r *VolumePopulatorReconciler) ensurePostReadyRestoreCompleted(reqCtx intct
 	if err = validatePostReadyRestore(existing, postReadyRestore, comp); err != nil {
 		return false, err
 	}
+	// The Restore is now visible to the Cluster lifecycle resource scan. The
+	// helper has already been released, so temporary target protection can go.
+	if pvcPopulateReleased(pvc) {
+		if err = r.releaseTargetPVC(reqCtx, pvc); err != nil {
+			return false, err
+		}
+	}
 	switch existing.Status.Phase {
 	case dpv1alpha1.RestorePhaseCompleted:
 		return true, nil
@@ -1920,7 +2303,7 @@ func (r *VolumePopulatorReconciler) deletePopulatePVC(reqCtx intctrlutil.Request
 
 func (r *VolumePopulatorReconciler) releaseTargetPVC(reqCtx intctrlutil.RequestCtx, pvc *corev1.PersistentVolumeClaim) error {
 	if slices.Contains(pvc.Finalizers, dptypes.DataProtectionFinalizerName) {
-		pvcPatch := client.MergeFrom(pvc.DeepCopy())
+		pvcPatch := client.MergeFromWithOptions(pvc.DeepCopy(), client.MergeFromWithOptimisticLock{})
 		controllerutil.RemoveFinalizer(pvc, dptypes.DataProtectionFinalizerName)
 		if err := r.Client.Patch(reqCtx.Ctx, pvc, pvcPatch); err != nil {
 			return client.IgnoreNotFound(err)
diff --git a/controllers/dataprotection/volumepopulator_controller_test.go b/controllers/dataprotection/volumepopulator_controller_test.go
index 8c5211fbb97..9ddc90e2f28 100644
--- a/controllers/dataprotection/volumepopulator_controller_test.go
+++ b/controllers/dataprotection/volumepopulator_controller_test.go
@@ -4221,6 +4221,415 @@ func dependencyRestorePVC(name, componentName string, uid types.UID) *corev1.Per
 	}
 }
 
+func TestClusterDeletionTerminatesVolumePopulationInOrder(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()
+			cluster.DeletionTimestamp = &now
+			cluster.Finalizers = append(cluster.Finalizers, "example.io/app-owner")
+			target.Finalizers = []string{dptypes.DataProtectionFinalizerName, "example.io/app-owner"}
+			target.Labels[dptypes.ComponentUIDLabelKey] = string(component.UID)
+			objects := []client.Object{cluster, 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")
+			for _, expected := range []*dpv1alpha1.Restore{execution, postReady} {
+				current := &dpv1alpha1.Restore{}
+				require.NoError(t, cli.Get(ctx, client.ObjectKeyFromObject(expected), current))
+				require.False(t, current.DeletionTimestamp.IsZero())
+				require.Equal(t, expected.Finalizers, current.Finalizers)
+				current.Finalizers = nil
+				require.NoError(t, cli.Update(ctx, current))
+			}
+			require.NoError(t, cli.Get(ctx, client.ObjectKeyFromObject(helper), &corev1.PersistentVolumeClaim{}))
+
+			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(cluster), cluster))
+			require.Contains(t, cluster.Finalizers, dptypes.RestoreProtectionFinalizerName)
+		})
+	}
+}
+
+func TestClusterLifecycleRegistersBeforeWaitingForProtection(t *testing.T) {
+	for _, existingFinalizer := range []bool{false, true} {
+		t.Run(fmt.Sprintf("existing-finalizer=%t", existingFinalizer), func(t *testing.T) {
+			ctx := context.Background()
+			scheme, cluster, component, its, target := parentRestoreObjects(t)
+			cluster.Finalizers = nil
+			delete(target.Labels, dptypes.ClusterUIDLabelKey)
+			if existingFinalizer {
+				target.Finalizers = []string{dptypes.DataProtectionFinalizerName}
+			}
+			cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(cluster, component, its, target).Build()
+			vp := &VolumePopulatorReconciler{Client: cli, Scheme: scheme}
+
+			require.ErrorContains(t, vp.syncPVC(intctrlutil.RequestCtx{Ctx: ctx}, target),
+				"waiting for target PVC restore protection")
+			require.NoError(t, cli.Get(ctx, client.ObjectKeyFromObject(target), target))
+			require.Equal(t, string(cluster.UID), target.Labels[dptypes.ClusterUIDLabelKey])
+			require.Equal(t, string(component.UID), target.Labels[dptypes.ComponentUIDLabelKey])
+			require.Contains(t, target.Finalizers, dptypes.DataProtectionFinalizerName)
+			require.Empty(t, target.Status.Conditions)
+
+			require.ErrorContains(t, vp.syncPVC(intctrlutil.RequestCtx{Ctx: ctx}, target),
+				"waiting for Cluster restore-protection finalizer")
+			require.NoError(t, cli.Get(ctx, client.ObjectKeyFromObject(cluster), cluster))
+			require.NotContains(t, cluster.Finalizers, dptypes.RestoreProtectionFinalizerName,
+				"VolumePopulator must not add the Cluster finalizer")
+			pvcs := &corev1.PersistentVolumeClaimList{}
+			require.NoError(t, cli.List(ctx, pvcs))
+			require.Len(t, pvcs.Items, 1)
+			restores := &dpv1alpha1.RestoreList{}
+			require.NoError(t, cli.List(ctx, restores))
+			require.Empty(t, restores.Items)
+		})
+	}
+}
+
+func TestClusterLifecycleRegistrationReturnsBeforeRestoreValidation(t *testing.T) {
+	ctx := context.Background()
+	scheme, cluster, component, its, target := parentRestoreObjects(t)
+	delete(target.Labels, dptypes.ClusterUIDLabelKey)
+	target.Spec.DataSourceRef.Kind = dptypes.RestoreKind
+	target.Spec.DataSourceRef.Name = "source"
+	target.Annotations[constant.RestoreSourceKindAnnotationKey] = dptypes.RestoreKind
+	target.Annotations[constant.RestoreSourceNameAnnotationKey] = "source"
+	base := fake.NewClientBuilder().WithScheme(scheme).WithObjects(cluster, component, its, target).Build()
+	validated := false
+	cli := interceptor.NewClient(base, interceptor.Funcs{Get: func(ctx context.Context, inner client.WithWatch,
+		key client.ObjectKey, obj client.Object, opts ...client.GetOption) error {
+		if _, ok := obj.(*dpv1alpha1.Restore); ok {
+			validated = true
+		}
+		return inner.Get(ctx, key, obj, opts...)
+	}})
+	vp := &VolumePopulatorReconciler{Client: cli, Scheme: scheme}
+
+	require.ErrorContains(t, vp.syncPVC(intctrlutil.RequestCtx{Ctx: ctx}, target),
+		"waiting for target PVC restore protection")
+	require.False(t, validated)
+	require.NoError(t, base.Get(ctx, client.ObjectKeyFromObject(target), target))
+	require.Contains(t, target.Finalizers, dptypes.DataProtectionFinalizerName)
+}
+
+func TestClusterLifecyclePostReadyProtectionHandoff(t *testing.T) {
+	ctx := context.Background()
+	scheme, cluster, component, _, target := parentRestoreObjects(t)
+	target.Labels[dptypes.ComponentUIDLabelKey] = string(component.UID)
+	target.Finalizers = nil
+	target.Spec.VolumeName = "target-pv"
+	target.Status.Conditions = []corev1.PersistentVolumeClaimCondition{{
+		Type: PersistentVolumeClaimPopulating, Status: corev1.ConditionTrue, Reason: ReasonPopulatingProvisioned,
+	}}
+	component.Status.Phase = kbappsv1.RunningComponentPhase
+	cluster.Status.Conditions = []metav1.Condition{{
+		Type: kbappsv1.ConditionTypeRestore, Status: metav1.ConditionTrue,
+	}}
+	backup, actionSet := restoreBackupObjects()
+	actionSet.Spec.Restore.PostReady = []dpv1alpha1.ActionSpec{{Job: &dpv1alpha1.JobActionSpec{}}}
+	worker := &corev1.ServiceAccount{ObjectMeta: metav1.ObjectMeta{Name: "worker", Namespace: target.Namespace}}
+	for key, value := range map[string]string{
+		dptypes.CfgKeyWorkerServiceAccountName: "worker",
+		dptypes.CfgKeyWorkerClusterRoleName:    "worker-role",
+	} {
+		previous := viper.Get(key)
+		viper.Set(key, value)
+		t.Cleanup(func() { viper.Set(key, previous) })
+	}
+	cli := fake.NewClientBuilder().WithScheme(scheme).WithStatusSubresource(target).
+		WithObjects(cluster, component, target, backup, actionSet, worker).Build()
+	vp := &VolumePopulatorReconciler{Client: cli, Scheme: scheme, Recorder: record.NewFakeRecorder(10)}
+	mgr := dprestore.NewRestoreManager(&dpv1alpha1.Restore{}, nil, scheme, cli)
+	mgr.PostReadyBackupSets = []dprestore.BackupActionSet{{Backup: backup}}
+	restoreCtx := &pvcRestoreContext{restoreMgr: mgr, mode: pvcRestoreModeProvisionOnly}
+	reqCtx := intctrlutil.RequestCtx{Ctx: ctx}
+
+	require.ErrorContains(t, vp.completeBoundPVCIfNeeded(reqCtx, target, restoreCtx),
+		"waiting for target PVC restore protection")
+	require.NoError(t, cli.Get(ctx, client.ObjectKeyFromObject(target), target))
+	require.Contains(t, target.Finalizers, dptypes.DataProtectionFinalizerName)
+	restores := &dpv1alpha1.RestoreList{}
+	require.NoError(t, cli.List(ctx, restores))
+	require.Empty(t, restores.Items)
+
+	require.ErrorContains(t, vp.completeBoundPVCIfNeeded(reqCtx, target, restoreCtx), "waiting for postReady restore")
+	require.NoError(t, cli.Get(ctx, client.ObjectKeyFromObject(target), target))
+	require.Contains(t, target.Finalizers, dptypes.DataProtectionFinalizerName)
+	require.NoError(t, cli.List(ctx, restores))
+	require.Len(t, restores.Items, 1)
+
+	require.ErrorContains(t, vp.completeBoundPVCIfNeeded(reqCtx, target, restoreCtx), "waiting for postReady restore")
+	require.NoError(t, cli.Get(ctx, client.ObjectKeyFromObject(target), target))
+	require.NotContains(t, target.Finalizers, dptypes.DataProtectionFinalizerName)
+	coordinator := &ClusterRestoreReconciler{Client: cli}
+	_, err := coordinator.Reconcile(ctx, reconcile.Request{NamespacedName: client.ObjectKeyFromObject(cluster)})
+	require.NoError(t, err)
+	require.NoError(t, cli.Get(ctx, client.ObjectKeyFromObject(cluster), cluster))
+	require.True(t, isClusterRestoreProtected(cluster))
+}
+
+func TestClusterLifecycleSafetyBoundaries(t *testing.T) {
+	t.Run("target deletion is not termination", func(t *testing.T) {
+		scheme, cluster, component, its, target := parentRestoreObjects(t)
+		now := metav1.Now()
+		target.DeletionTimestamp = &now
+		target.Finalizers = []string{dptypes.DataProtectionFinalizerName, "example.io/app-owner"}
+		target.Labels[dptypes.ComponentUIDLabelKey] = string(component.UID)
+		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)
+		require.NoError(t, err)
+		require.False(t, terminated)
+	})
+
+	t.Run("deleting target without VP finalizer is ignored", func(t *testing.T) {
+		scheme, cluster, _, _, target := parentRestoreObjects(t)
+		now := metav1.Now()
+		target.DeletionTimestamp = &now
+		target.Finalizers = []string{"kubernetes.io/pvc-protection"}
+		cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(cluster, 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.Equal(t, []string{"kubernetes.io/pvc-protection"}, target.Finalizers)
+		restores := &dpv1alpha1.RestoreList{}
+		require.NoError(t, cli.List(context.Background(), restores))
+		require.Empty(t, restores.Items)
+	})
+
+	t.Run("Cluster deletion cleans postReady after target protection handoff", func(t *testing.T) {
+		scheme, cluster, component, _, target := parentRestoreObjects(t)
+		now := metav1.Now()
+		cluster.DeletionTimestamp = &now
+		cluster.Finalizers = append(cluster.Finalizers, "example.io/app-owner")
+		target.DeletionTimestamp = &now
+		target.Finalizers = []string{"example.io/app-owner"}
+		postReady := postReadyRestoreForComponent(target, cluster, component)
+		postReady.Finalizers = []string{"example.io/restore-owner"}
+		cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(cluster, target, postReady).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(postReady), postReady))
+		require.False(t, postReady.DeletionTimestamp.IsZero())
+		require.NoError(t, cli.Get(context.Background(), client.ObjectKeyFromObject(target), target))
+		require.NotContains(t, target.Finalizers, dptypes.DataProtectionFinalizerName)
+	})
+
+	t.Run("missing Cluster does not authorize active cleanup", func(t *testing.T) {
+		scheme, cluster, component, _, target := parentRestoreObjects(t)
+		target.Labels[dptypes.ComponentUIDLabelKey] = string(component.UID)
+		target.Finalizers = []string{dptypes.DataProtectionFinalizerName}
+		helper := restoreHelperForTarget(target, cluster)
+		cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(target, helper).
+			WithInterceptorFuncs(interceptor.Funcs{
+				Delete: func(context.Context, client.WithWatch, client.Object, ...client.DeleteOption) error {
+					t.Fatal("missing Cluster must not authorize deletion")
+					return nil
+				},
+				Patch: func(context.Context, client.WithWatch, client.Object, client.Patch, ...client.PatchOption) error {
+					t.Fatal("active protection must remain")
+					return nil
+				},
+			}).Build()
+		vp := &VolumePopulatorReconciler{Client: cli, Scheme: scheme}
+		require.True(t, intctrlutil.IsRequeueError(vp.syncPVC(
+			intctrlutil.RequestCtx{Ctx: context.Background()}, target)))
+	})
+
+	t.Run("completed Cluster restore does not register", func(t *testing.T) {
+		scheme, cluster, component, its, target := parentRestoreObjects(t)
+		cluster.Status.Conditions = []metav1.Condition{{
+			Type: kbappsv1.ConditionTypeRestore, Status: metav1.ConditionTrue,
+		}}
+		delete(target.Labels, dptypes.ClusterUIDLabelKey)
+		cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(cluster, component, its, target).Build()
+		vp := &VolumePopulatorReconciler{Client: cli, Scheme: scheme}
+		err := vp.syncPVC(intctrlutil.RequestCtx{Ctx: context.Background()}, target)
+		require.ErrorContains(t, err, "Cluster restore is no longer active")
+		require.NoError(t, cli.Get(context.Background(), client.ObjectKeyFromObject(target), target))
+		require.NotContains(t, target.Finalizers, dptypes.DataProtectionFinalizerName)
+	})
+}
+
+func TestClusterLifecycleRefusesForeignExecutionRestore(t *testing.T) {
+	scheme, cluster, component, _, target := parentRestoreObjects(t)
+	now := metav1.Now()
+	cluster.DeletionTimestamp = &now
+	target.Finalizers = []string{dptypes.DataProtectionFinalizerName}
+	target.Labels[dptypes.ComponentUIDLabelKey] = string(component.UID)
+	helper := restoreHelperForTarget(target, cluster)
+	foreign := executionRestoreForTarget(target, cluster)
+	foreign.OwnerReferences[0].UID = "foreign-pvc"
+	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)
+	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{}))
+	require.NoError(t, cli.Get(context.Background(), client.ObjectKeyFromObject(helper), &corev1.PersistentVolumeClaim{}))
+}
+
+func TestValidateClusterRestorePVCOwnershipThroughInstance(t *testing.T) {
+	scheme, cluster, component, its, target := parentRestoreObjects(t)
+	controller := true
+	instance := &workloadsv1.Instance{ObjectMeta: metav1.ObjectMeta{
+		Namespace: target.Namespace, Name: "cluster-mysql-0", UID: "instance-uid",
+		OwnerReferences: []metav1.OwnerReference{{
+			APIVersion: workloadsv1.GroupVersion.String(), Kind: workloadsv1.InstanceSetKind,
+			Name: its.Name, UID: its.UID, Controller: &controller,
+		}},
+	}}
+	target.OwnerReferences = []metav1.OwnerReference{{
+		APIVersion: workloadsv1.GroupVersion.String(), Kind: "Instance",
+		Name: instance.Name, UID: instance.UID, Controller: &controller,
+	}}
+	cli := fake.NewClientBuilder().WithScheme(scheme).
+		WithObjects(cluster, component, its, instance, target).Build()
+	vp := &VolumePopulatorReconciler{Client: cli, Scheme: scheme}
+
+	actual, err := vp.validateClusterRestorePVCOwnership(context.Background(), target, cluster)
+	require.NoError(t, err)
+	require.Equal(t, component.UID, actual.UID)
+}
+
+func restoreBackupObjects() (*dpv1alpha1.Backup, *dpv1alpha1.ActionSet) {
+	backup := newBackupForRestoreDecision([]string{"data"}, nil)
+	backup.Status.Phase = dpv1alpha1.BackupPhaseCompleted
+	backup.Status.BackupMethod.ActionSetName = "full"
+	backup.Status.Target.PodSelector = &dpv1alpha1.PodSelector{Strategy: dpv1alpha1.PodSelectionStrategyAny}
+	actionSet := &dpv1alpha1.ActionSet{ObjectMeta: metav1.ObjectMeta{Name: "full"}, Spec: dpv1alpha1.ActionSetSpec{
+		BackupType: dpv1alpha1.BackupTypeFull,
+		Restore:    &dpv1alpha1.RestoreActionSpec{PrepareData: &dpv1alpha1.JobActionSpec{}},
+	}}
+	return backup, actionSet
+}
+
+func parentRestoreObjects(t *testing.T) (*runtime.Scheme, *kbappsv1.Cluster, *kbappsv1.Component,
+	*workloadsv1.InstanceSet, *corev1.PersistentVolumeClaim) {
+	t.Helper()
+	scheme := runtime.NewScheme()
+	require.NoError(t, corev1.AddToScheme(scheme))
+	require.NoError(t, kbappsv1.AddToScheme(scheme))
+	require.NoError(t, workloadsv1.AddToScheme(scheme))
+	require.NoError(t, dpv1alpha1.AddToScheme(scheme))
+	controller := true
+	cluster := &kbappsv1.Cluster{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: "default", Name: "cluster", UID: "cluster-uid",
+			Finalizers: []string{dptypes.RestoreProtectionFinalizerName},
+		},
+		Spec: kbappsv1.ClusterSpec{Restore: &kbappsv1.ClusterRestore{}},
+	}
+	component := &kbappsv1.Component{ObjectMeta: metav1.ObjectMeta{
+		Namespace: cluster.Namespace, Name: "cluster-mysql", UID: "component-uid",
+		Labels: map[string]string{
+			constant.AppInstanceLabelKey: cluster.Name, constant.KBAppComponentLabelKey: "mysql",
+		},
+		OwnerReferences: []metav1.OwnerReference{{
+			APIVersion: kbappsv1.GroupVersion.String(), Kind: kbappsv1.ClusterKind,
+			Name: cluster.Name, UID: cluster.UID, Controller: &controller,
+		}},
+	}}
+	its := &workloadsv1.InstanceSet{ObjectMeta: metav1.ObjectMeta{
+		Namespace: cluster.Namespace, Name: "cluster-mysql", UID: "its-uid",
+		Labels: map[string]string{
+			constant.AppInstanceLabelKey: cluster.Name, constant.KBAppComponentLabelKey: "mysql",
+		},
+		OwnerReferences: []metav1.OwnerReference{{
+			APIVersion: kbappsv1.GroupVersion.String(), Kind: kbappsv1.ComponentKind,
+			Name: component.Name, UID: component.UID, Controller: &controller,
+		}},
+	}}
+	target := dependencyRestorePVC("data-mysql-0", "mysql", "target-uid")
+	target.Annotations[constant.KBAppClusterUIDKey] = string(cluster.UID)
+	target.Labels[dptypes.ClusterUIDLabelKey] = string(cluster.UID)
+	target.OwnerReferences = []metav1.OwnerReference{{
+		APIVersion: workloadsv1.GroupVersion.String(), Kind: workloadsv1.InstanceSetKind,
+		Name: its.Name, UID: its.UID, Controller: &controller,
+	}}
+	return scheme, cluster, component, its, target
+}
+
+func restoreHelperForTarget(target *corev1.PersistentVolumeClaim,
+	cluster *kbappsv1.Cluster) *corev1.PersistentVolumeClaim {
+	name := getPopulatePVCName(target.UID)
+	return &corev1.PersistentVolumeClaim{ObjectMeta: metav1.ObjectMeta{
+		Namespace: target.Namespace, Name: name,
+		Labels: map[string]string{
+			constant.AppInstanceLabelKey:                     cluster.Name,
+			constant.KBAppComponentLabelKey:                  target.Labels[constant.KBAppComponentLabelKey],
+			dptypes.ClusterUIDLabelKey:                       string(cluster.UID),
+			dprestore.DataProtectionRestoreLabelKey:          name,
+			dprestore.DataProtectionRestoreNamespaceLabelKey: target.Namespace,
+			dprestore.DataProtectionPopulatePVCLabelKey:      name,
+		},
+	}}
+}
+
+func executionRestoreForTarget(target *corev1.PersistentVolumeClaim,
+	cluster *kbappsv1.Cluster) *dpv1alpha1.Restore {
+	name := getPopulatePVCName(target.UID)
+	return &dpv1alpha1.Restore{ObjectMeta: metav1.ObjectMeta{
+		Namespace: target.Namespace, Name: name,
+		Labels: map[string]string{
+			constant.AppInstanceLabelKey:            cluster.Name,
+			constant.KBAppComponentLabelKey:         target.Labels[constant.KBAppComponentLabelKey],
+			dptypes.ClusterUIDLabelKey:              string(cluster.UID),
+			dprestore.DataProtectionRestoreLabelKey: name,
+		},
+		OwnerReferences: []metav1.OwnerReference{{
+			APIVersion: corev1.SchemeGroupVersion.String(), Kind: "PersistentVolumeClaim",
+			Name: target.Name, UID: target.UID,
+		}},
+	}}
+}
+
+func postReadyRestoreForComponent(target *corev1.PersistentVolumeClaim, cluster *kbappsv1.Cluster,
+	component *kbappsv1.Component) *dpv1alpha1.Restore {
+	name := postReadyRestoreName(component.UID)
+	return &dpv1alpha1.Restore{ObjectMeta: metav1.ObjectMeta{
+		Namespace: target.Namespace, Name: name,
+		Labels: map[string]string{
+			constant.AppInstanceLabelKey:            cluster.Name,
+			constant.KBAppComponentLabelKey:         target.Labels[constant.KBAppComponentLabelKey],
+			dptypes.ClusterUIDLabelKey:              string(cluster.UID),
+			dptypes.ComponentUIDLabelKey:            string(component.UID),
+			dprestore.DataProtectionRestoreLabelKey: name,
+		},
+		OwnerReferences: []metav1.OwnerReference{{
+			APIVersion: kbappsv1.GroupVersion.String(), Kind: kbappsv1.ComponentKind,
+			Name: component.Name, UID: component.UID,
+		}},
+	}}
+}
+
 func TestEnsurePostReadyRestore_ShardingMissingTargetSkip_DoesNotRedirect(t *testing.T) {
 	scheme := runtime.NewScheme()
 	require.NoError(t, corev1.AddToScheme(scheme))
diff --git a/pkg/dataprotection/types/constant.go b/pkg/dataprotection/types/constant.go
index b653dc6e86a..810adca70ff 100644
--- a/pkg/dataprotection/types/constant.go
+++ b/pkg/dataprotection/types/constant.go
@@ -46,6 +46,9 @@ const (
 const (
 	// DataProtectionFinalizerName is the name of our custom finalizer
 	DataProtectionFinalizerName = "dataprotection.kubeblocks.io/finalizer"
+	// RestoreProtectionFinalizerName prevents Cluster deletion from completing
+	// before its restore resources have been cleaned up.
+	RestoreProtectionFinalizerName = "dataprotection.kubeblocks.io/restore-protection-finalizer"
 )
 
 // annotation keys