diff --git a/docs/controller-protocol.md b/docs/controller-protocol.md
index bf25950b..43103b29 100644
--- a/docs/controller-protocol.md
+++ b/docs/controller-protocol.md
@@ -8,6 +8,8 @@
     - [CreateOrUpdateMantleBackupResponse](#proto-CreateOrUpdateMantleBackupResponse)
     - [CreateOrUpdatePVCRequest](#proto-CreateOrUpdatePVCRequest)
     - [CreateOrUpdatePVCResponse](#proto-CreateOrUpdatePVCResponse)
+    - [ListMantleBackupRequest](#proto-ListMantleBackupRequest)
+    - [ListMantleBackupResponse](#proto-ListMantleBackupResponse)
   
     - [MantleService](#proto-MantleService)
   
@@ -78,6 +80,37 @@ CreateOrUpdatePVCResponse is a response message for CreateOrUpdatePVC RPC.
 
 
 
+
+<a name="proto-ListMantleBackupRequest"></a>
+
+### ListMantleBackupRequest
+ListMantleBackupRequest is a request message for ListMantleBackup RPC.
+
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| pvcUID | [string](#string) |  |  |
+| namespace | [string](#string) |  |  |
+
+
+
+
+
+
+<a name="proto-ListMantleBackupResponse"></a>
+
+### ListMantleBackupResponse
+ListMantleBackupResponse is a response message for ListMantleBackup RPC.
+
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| mantleBackupList | [bytes](#bytes) |  |  |
+
+
+
+
+
  
 
  
@@ -94,6 +127,7 @@ CreateOrUpdatePVCResponse is a response message for CreateOrUpdatePVC RPC.
 | ----------- | ------------ | ------------- | ------------|
 | CreateOrUpdatePVC | [CreateOrUpdatePVCRequest](#proto-CreateOrUpdatePVCRequest) | [CreateOrUpdatePVCResponse](#proto-CreateOrUpdatePVCResponse) |  |
 | CreateOrUpdateMantleBackup | [CreateOrUpdateMantleBackupRequest](#proto-CreateOrUpdateMantleBackupRequest) | [CreateOrUpdateMantleBackupResponse](#proto-CreateOrUpdateMantleBackupResponse) |  |
+| ListMantleBackup | [ListMantleBackupRequest](#proto-ListMantleBackupRequest) | [ListMantleBackupResponse](#proto-ListMantleBackupResponse) |  |
 
  
 
diff --git a/internal/controller/internal/testutil/resources.go b/internal/controller/internal/testutil/resources.go
index da62a9c1..f9517aa7 100644
--- a/internal/controller/internal/testutil/resources.go
+++ b/internal/controller/internal/testutil/resources.go
@@ -199,6 +199,15 @@ func (r *ResourceManager) WaitForBackupReady(ctx context.Context, backup *mantle
 	}).WithContext(ctx).Should(Succeed())
 }
 
+func (r *ResourceManager) WaitForBackupSyncedToRemote(ctx context.Context, backup *mantlev1.MantleBackup) {
+	EventuallyWithOffset(1, func(g Gomega, ctx context.Context) {
+		err := r.client.Get(ctx, types.NamespacedName{Name: backup.Name, Namespace: backup.Namespace}, backup)
+		g.Expect(err).NotTo(HaveOccurred())
+
+		g.Expect(meta.IsStatusConditionTrue(backup.Status.Conditions, mantlev1.BackupConditionSyncedToRemote)).Should(BeTrue())
+	}).WithContext(ctx).Should(Succeed())
+}
+
 // cf. https://go.googlesource.com/proposal/+/refs/heads/master/design/43651-type-parameters.md#pointer-method-example
 type ObjectConstraint[T any] interface {
 	client.Object
diff --git a/internal/controller/mantlebackup_controller.go b/internal/controller/mantlebackup_controller.go
index 6e8147cb..89a4971e 100644
--- a/internal/controller/mantlebackup_controller.go
+++ b/internal/controller/mantlebackup_controller.go
@@ -34,8 +34,13 @@ const (
 	labelLocalBackupTargetPVCUID  = "mantle.cybozu.io/local-backup-target-pvc-uid"
 	labelRemoteBackupTargetPVCUID = "mantle.cybozu.io/remote-backup-target-pvc-uid"
 	annotRemoteUID                = "mantle.cybozu.io/remote-uid"
+	annotDiffFrom                 = "mantle.cybozu.io/diff-from"
 	annotDiffTo                   = "mantle.cybozu.io/diff-to"
 	annotRetainIfExpired          = "mantle.cybozu.io/retain-if-expired"
+	annotSyncMode                 = "mantle.cybozu.io/sync-mode"
+
+	syncModeFull        = "full"
+	syncModeIncremental = "incremental"
 )
 
 // MantleBackupReconciler reconciles a MantleBackup object
@@ -397,10 +402,14 @@ func (r *MantleBackupReconciler) replicate(
 	if err != nil || result != (ctrl.Result{}) {
 		return result, err
 	}
-	prepareResult, result, err := r.prepareForDataSynchronization(ctx, backup, r.primarySettings.Client)
-	if err != nil || result != (ctrl.Result{}) {
-		return result, err
+	prepareResult, err := r.prepareForDataSynchronization(ctx, backup, r.primarySettings.Client)
+	if err != nil {
+		return ctrl.Result{}, err
 	}
+
+	// FIXME: Delete this code after implementing export().
+	prepareResult.isSecondaryMantleBackupReadyToUse = true
+
 	if prepareResult.isSecondaryMantleBackupReadyToUse {
 		return r.primaryCleanup(ctx, backup)
 	}
@@ -602,21 +611,146 @@ func (r *MantleBackupReconciler) finalize(
 }
 
 type dataSyncPrepareResult struct {
-	isIncremental                     bool
+	isIncremental                     bool // NOTE: The value is forcibly set to false if isSecondaryMantleBackupReadyToUse is true.
 	isSecondaryMantleBackupReadyToUse bool
 	diffFrom                          *mantlev1.MantleBackup // non-nil value iff isIncremental is true.
 }
 
 func (r *MantleBackupReconciler) prepareForDataSynchronization(
-	_ context.Context,
-	_ *mantlev1.MantleBackup,
-	_ proto.MantleServiceClient,
-) (*dataSyncPrepareResult, ctrl.Result, error) { //nolint:unparam
+	ctx context.Context,
+	backup *mantlev1.MantleBackup,
+	msc proto.MantleServiceClient,
+) (*dataSyncPrepareResult, error) {
+	exportTargetPVCUID, ok := backup.GetLabels()[labelLocalBackupTargetPVCUID]
+	if !ok {
+		return nil, fmt.Errorf(`"%s" label is missing`, labelLocalBackupTargetPVCUID)
+	}
+	resp, err := msc.ListMantleBackup(
+		ctx,
+		&proto.ListMantleBackupRequest{
+			PvcUID:    exportTargetPVCUID,
+			Namespace: backup.GetNamespace(),
+		},
+	)
+	if err != nil {
+		return nil, err
+	}
+	secondaryBackups := make([]mantlev1.MantleBackup, 0)
+	err = json.Unmarshal(resp.MantleBackupList, &secondaryBackups)
+	if err != nil {
+		return nil, err
+	}
+	secondaryBackupMap := convertToMap(secondaryBackups)
+
+	isSecondaryMantleBackupReadyToUse := false
+	secondaryBackup, ok := secondaryBackupMap[backup.GetName()]
+	if !ok {
+		return nil, fmt.Errorf("secondary MantleBackup not found: %s, %s",
+			backup.GetName(), backup.GetNamespace())
+	}
+	isSecondaryMantleBackupReadyToUse = meta.IsStatusConditionTrue(
+		secondaryBackup.Status.Conditions,
+		mantlev1.BackupConditionReadyToUse,
+	)
+
+	if isSecondaryMantleBackupReadyToUse {
+		return &dataSyncPrepareResult{
+			isIncremental:                     false,
+			isSecondaryMantleBackupReadyToUse: true,
+			diffFrom:                          nil,
+		}, nil
+	}
+
+	if syncMode, ok := backup.GetAnnotations()[annotSyncMode]; ok {
+		switch syncMode {
+		case syncModeFull:
+			return &dataSyncPrepareResult{
+				isIncremental:                     false,
+				isSecondaryMantleBackupReadyToUse: isSecondaryMantleBackupReadyToUse,
+				diffFrom:                          nil,
+			}, nil
+		case syncModeIncremental:
+			diffFromName, ok := backup.GetAnnotations()[annotDiffFrom]
+			if !ok {
+				return nil, fmt.Errorf(`"%s" annotation is missing`, annotDiffFrom)
+			}
+
+			var diffFrom mantlev1.MantleBackup
+			err = r.Client.Get(ctx, types.NamespacedName{
+				Name:      diffFromName,
+				Namespace: backup.GetNamespace(),
+			}, &diffFrom)
+			if err != nil {
+				return nil, err
+			}
+
+			return &dataSyncPrepareResult{
+				isIncremental:                     true,
+				isSecondaryMantleBackupReadyToUse: isSecondaryMantleBackupReadyToUse,
+				diffFrom:                          &diffFrom,
+			}, nil
+		default:
+			return nil, fmt.Errorf("unknown sync mode: %s", syncMode)
+		}
+	}
+
+	var primaryBackupList mantlev1.MantleBackupList
+	// TODO: Perhaps, we may have to use the client without cache.
+	err = r.Client.List(ctx, &primaryBackupList, &client.ListOptions{
+		LabelSelector: labels.SelectorFromSet(map[string]string{labelLocalBackupTargetPVCUID: exportTargetPVCUID}),
+		Namespace:     backup.GetNamespace(),
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	diffFrom := searchForDiffOriginMantleBackup(backup, primaryBackupList.Items, secondaryBackupMap)
+	isIncremental := (diffFrom != nil)
+
 	return &dataSyncPrepareResult{
-		isIncremental:                     false,
-		isSecondaryMantleBackupReadyToUse: true,
-		diffFrom:                          nil,
-	}, ctrl.Result{}, nil
+		isIncremental:                     isIncremental,
+		isSecondaryMantleBackupReadyToUse: isSecondaryMantleBackupReadyToUse,
+		diffFrom:                          diffFrom,
+	}, nil
+}
+
+func convertToMap(mantleBackups []mantlev1.MantleBackup) map[string]*mantlev1.MantleBackup {
+	m := make(map[string]*mantlev1.MantleBackup)
+	for _, mantleBackup := range mantleBackups {
+		mantleBackup := mantleBackup
+		m[mantleBackup.GetName()] = &mantleBackup
+	}
+	return m
+}
+
+func searchForDiffOriginMantleBackup(
+	backup *mantlev1.MantleBackup,
+	primaryBackups []mantlev1.MantleBackup,
+	secondaryBackupMap map[string]*mantlev1.MantleBackup,
+) *mantlev1.MantleBackup {
+	var diffOrigin *mantlev1.MantleBackup
+	for _, primaryBackup := range primaryBackups {
+		primaryBackup := primaryBackup
+		secondaryBackup, ok := secondaryBackupMap[primaryBackup.Name]
+		if !ok {
+			continue
+		}
+		if !meta.IsStatusConditionTrue(primaryBackup.Status.Conditions, mantlev1.BackupConditionReadyToUse) ||
+			!meta.IsStatusConditionTrue(secondaryBackup.Status.Conditions, mantlev1.BackupConditionReadyToUse) {
+			continue
+		}
+		if !primaryBackup.DeletionTimestamp.IsZero() || !secondaryBackup.DeletionTimestamp.IsZero() {
+			continue
+		}
+		if *backup.Status.SnapID <= *primaryBackup.Status.SnapID {
+			continue
+		}
+		if diffOrigin == nil || *diffOrigin.Status.SnapID < *primaryBackup.Status.SnapID {
+			diffOrigin = &primaryBackup
+		}
+	}
+
+	return diffOrigin
 }
 
 func (r *MantleBackupReconciler) export(
diff --git a/internal/controller/mantlebackup_controller_test.go b/internal/controller/mantlebackup_controller_test.go
index b87f27ee..fd31a793 100644
--- a/internal/controller/mantlebackup_controller_test.go
+++ b/internal/controller/mantlebackup_controller_test.go
@@ -3,13 +3,16 @@ package controller
 import (
 	"context"
 	"encoding/json"
+	"slices"
 	"sync"
 	"time"
 
 	mantlev1 "github.com/cybozu-go/mantle/api/v1"
 	"github.com/cybozu-go/mantle/internal/controller/internal/testutil"
+	"github.com/cybozu-go/mantle/pkg/controller/proto"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
+	"google.golang.org/grpc"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -89,137 +92,171 @@ var _ = Describe("MantleBackup controller", func() {
 		}()
 	}
 
-	BeforeEach(func() {
-		mgrUtil = testutil.NewManagerUtil(context.Background(), cfg, scheme.Scheme)
-
-		reconciler = NewMantleBackupReconciler(
-			mgrUtil.GetManager().GetClient(),
-			mgrUtil.GetManager().GetScheme(),
-			resMgr.ClusterID,
-			RoleStandalone,
-			nil,
-		)
-		reconciler.ceph = testutil.NewFakeRBD()
-		err := reconciler.SetupWithManager(mgrUtil.GetManager())
+	AfterEach(func() {
+		err := mgrUtil.Stop()
 		Expect(err).NotTo(HaveOccurred())
+	})
 
-		setupExpireQueueSniffer()
+	Context("when the role is `standalone`", func() {
+		BeforeEach(func() {
+			mgrUtil = testutil.NewManagerUtil(context.Background(), cfg, scheme.Scheme)
+
+			reconciler = NewMantleBackupReconciler(
+				mgrUtil.GetManager().GetClient(),
+				mgrUtil.GetManager().GetScheme(),
+				resMgr.ClusterID,
+				RoleStandalone,
+				nil,
+			)
+			reconciler.ceph = testutil.NewFakeRBD()
+			err := reconciler.SetupWithManager(mgrUtil.GetManager())
+			Expect(err).NotTo(HaveOccurred())
 
-		mgrUtil.Start()
-		time.Sleep(100 * time.Millisecond)
+			setupExpireQueueSniffer()
 
-		ns = resMgr.CreateNamespace()
-	})
+			mgrUtil.Start()
+			time.Sleep(100 * time.Millisecond)
 
-	AfterEach(func() {
-		err := mgrUtil.Stop()
-		Expect(err).NotTo(HaveOccurred())
-	})
+			ns = resMgr.CreateNamespace()
+		})
 
-	It("should be ready to use", func(ctx SpecContext) {
-		pv, pvc, err := resMgr.CreateUniquePVAndPVC(ctx, ns)
-		Expect(err).NotTo(HaveOccurred())
+		It("should be ready to use", func(ctx SpecContext) {
+			pv, pvc, err := resMgr.CreateUniquePVAndPVC(ctx, ns)
+			Expect(err).NotTo(HaveOccurred())
 
-		backup, err := resMgr.CreateUniqueBackupFor(ctx, pvc)
-		Expect(err).NotTo(HaveOccurred())
+			backup, err := resMgr.CreateUniqueBackupFor(ctx, pvc)
+			Expect(err).NotTo(HaveOccurred())
 
-		waitForHavingFinalizer(ctx, backup)
-		resMgr.WaitForBackupReady(ctx, backup)
+			waitForHavingFinalizer(ctx, backup)
+			resMgr.WaitForBackupReady(ctx, backup)
 
-		pvcJS := backup.Status.PVCManifest
-		Expect(pvcJS).NotTo(BeEmpty())
-		pvcStored := corev1.PersistentVolumeClaim{}
-		err = json.Unmarshal([]byte(pvcJS), &pvcStored)
-		Expect(err).NotTo(HaveOccurred())
-		Expect(pvcStored.Name).To(Equal(pvc.Name))
-		Expect(pvcStored.Namespace).To(Equal(pvc.Namespace))
+			pvcJS := backup.Status.PVCManifest
+			Expect(pvcJS).NotTo(BeEmpty())
+			pvcStored := corev1.PersistentVolumeClaim{}
+			err = json.Unmarshal([]byte(pvcJS), &pvcStored)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(pvcStored.Name).To(Equal(pvc.Name))
+			Expect(pvcStored.Namespace).To(Equal(pvc.Namespace))
 
-		pvJS := backup.Status.PVManifest
-		Expect(pvJS).NotTo(BeEmpty())
-		pvStored := corev1.PersistentVolume{}
-		err = json.Unmarshal([]byte(pvJS), &pvStored)
-		Expect(err).NotTo(HaveOccurred())
-		Expect(pvStored.Name).To(Equal(pv.Name))
+			pvJS := backup.Status.PVManifest
+			Expect(pvJS).NotTo(BeEmpty())
+			pvStored := corev1.PersistentVolume{}
+			err = json.Unmarshal([]byte(pvJS), &pvStored)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(pvStored.Name).To(Equal(pv.Name))
 
-		snaps, err := reconciler.ceph.RBDSnapLs(resMgr.PoolName, pv.Spec.CSI.VolumeAttributes["imageName"])
-		Expect(err).NotTo(HaveOccurred())
-		Expect(snaps).To(HaveLen(1))
-		snapID := backup.Status.SnapID
-		Expect(snapID).To(Equal(&snaps[0].Id))
+			snaps, err := reconciler.ceph.RBDSnapLs(resMgr.PoolName, pv.Spec.CSI.VolumeAttributes["imageName"])
+			Expect(err).NotTo(HaveOccurred())
+			Expect(snaps).To(HaveLen(1))
+			snapID := backup.Status.SnapID
+			Expect(snapID).To(Equal(&snaps[0].Id))
 
-		err = k8sClient.Delete(ctx, backup)
-		Expect(err).NotTo(HaveOccurred())
+			err = k8sClient.Delete(ctx, backup)
+			Expect(err).NotTo(HaveOccurred())
 
-		testutil.CheckDeletedEventually[mantlev1.MantleBackup](ctx, k8sClient, backup.Name, backup.Namespace)
-	})
+			testutil.CheckDeletedEventually[mantlev1.MantleBackup](ctx, k8sClient, backup.Name, backup.Namespace)
+		})
 
-	It("should still be ready to use even if the PVC lost", func(ctx SpecContext) {
-		_, pvc, err := resMgr.CreateUniquePVAndPVC(ctx, ns)
-		Expect(err).NotTo(HaveOccurred())
+		It("should still be ready to use even if the PVC lost", func(ctx SpecContext) {
+			_, pvc, err := resMgr.CreateUniquePVAndPVC(ctx, ns)
+			Expect(err).NotTo(HaveOccurred())
 
-		backup, err := resMgr.CreateUniqueBackupFor(ctx, pvc)
-		Expect(err).NotTo(HaveOccurred())
+			backup, err := resMgr.CreateUniqueBackupFor(ctx, pvc)
+			Expect(err).NotTo(HaveOccurred())
 
-		waitForHavingFinalizer(ctx, backup)
-		resMgr.WaitForBackupReady(ctx, backup)
+			waitForHavingFinalizer(ctx, backup)
+			resMgr.WaitForBackupReady(ctx, backup)
 
-		pvc.Status.Phase = corev1.ClaimLost // simulate lost PVC
-		err = k8sClient.Status().Update(ctx, pvc)
-		Expect(err).NotTo(HaveOccurred())
+			pvc.Status.Phase = corev1.ClaimLost // simulate lost PVC
+			err = k8sClient.Status().Update(ctx, pvc)
+			Expect(err).NotTo(HaveOccurred())
 
-		resMgr.WaitForBackupReady(ctx, backup)
-	})
+			resMgr.WaitForBackupReady(ctx, backup)
+		})
+
+		DescribeTable("MantleBackup with correct expiration",
+			func(ctx SpecContext, expire string) {
+				_, pvc, err := resMgr.CreateUniquePVAndPVC(ctx, ns)
+				Expect(err).NotTo(HaveOccurred())
+
+				_, err = resMgr.CreateUniqueBackupFor(ctx, pvc, func(backup *mantlev1.MantleBackup) {
+					backup.Spec.Expire = expire
+				})
+				Expect(err).NotTo(HaveOccurred())
+			},
+			Entry("min expire", "1d"),
+			Entry("max expire", "15d"),
+			Entry("complex expire", "1w2d3h4m5s"),
+		)
 
-	DescribeTable("MantleBackup with correct expiration",
-		func(ctx SpecContext, expire string) {
+		DescribeTable("MantleBackup with incorrect expiration",
+			func(ctx SpecContext, expire string) {
+				_, pvc, err := resMgr.CreateUniquePVAndPVC(ctx, ns)
+				Expect(err).NotTo(HaveOccurred())
+
+				_, err = resMgr.CreateUniqueBackupFor(ctx, pvc, func(backup *mantlev1.MantleBackup) {
+					backup.Spec.Expire = expire
+				})
+				Expect(err).To(Or(
+					MatchError(ContainSubstring("expire must be")),
+					MatchError(ContainSubstring("body must be of type duration")),
+				))
+			},
+			Entry("invalid short expire", "23h"),
+			Entry("invalid long expire", "15d1s"),
+			Entry("invalid duration", "foo"),
+		)
+
+		It("Should reject updating the expire field", func(ctx SpecContext) {
 			_, pvc, err := resMgr.CreateUniquePVAndPVC(ctx, ns)
 			Expect(err).NotTo(HaveOccurred())
 
-			_, err = resMgr.CreateUniqueBackupFor(ctx, pvc, func(backup *mantlev1.MantleBackup) {
-				backup.Spec.Expire = expire
-			})
+			backup, err := resMgr.CreateUniqueBackupFor(ctx, pvc)
 			Expect(err).NotTo(HaveOccurred())
-		},
-		Entry("min expire", "1d"),
-		Entry("max expire", "15d"),
-		Entry("complex expire", "1w2d3h4m5s"),
-	)
 
-	DescribeTable("MantleBackup with incorrect expiration",
-		func(ctx SpecContext, expire string) {
-			_, pvc, err := resMgr.CreateUniquePVAndPVC(ctx, ns)
+			expire, err := strfmt.ParseDuration(backup.Spec.Expire)
 			Expect(err).NotTo(HaveOccurred())
+			expire += time.Hour
+			backup.Spec.Expire = expire.String()
+			err = k8sClient.Update(ctx, backup)
+			Expect(err).To(MatchError(ContainSubstring("spec.expire is immutable")))
+		})
 
-			_, err = resMgr.CreateUniqueBackupFor(ctx, pvc, func(backup *mantlev1.MantleBackup) {
-				backup.Spec.Expire = expire
-			})
-			Expect(err).To(Or(
-				MatchError(ContainSubstring("expire must be")),
-				MatchError(ContainSubstring("body must be of type duration")),
-			))
-		},
-		Entry("invalid short expire", "23h"),
-		Entry("invalid long expire", "15d1s"),
-		Entry("invalid duration", "foo"),
-	)
+		DescribeTable("MantleBackup expiration",
+			func(ctx SpecContext, offset time.Duration) {
+				_, pvc, err := resMgr.CreateUniquePVAndPVC(ctx, ns)
+				Expect(err).NotTo(HaveOccurred())
 
-	It("Should reject updating the expire field", func(ctx SpecContext) {
-		_, pvc, err := resMgr.CreateUniquePVAndPVC(ctx, ns)
-		Expect(err).NotTo(HaveOccurred())
+				backup, err := resMgr.CreateUniqueBackupFor(ctx, pvc)
+				Expect(err).NotTo(HaveOccurred())
 
-		backup, err := resMgr.CreateUniqueBackupFor(ctx, pvc)
-		Expect(err).NotTo(HaveOccurred())
+				By("waiting for the backup to be ready")
+				resMgr.WaitForBackupReady(ctx, backup)
 
-		expire, err := strfmt.ParseDuration(backup.Spec.Expire)
-		Expect(err).NotTo(HaveOccurred())
-		expire += time.Hour
-		backup.Spec.Expire = expire.String()
-		err = k8sClient.Update(ctx, backup)
-		Expect(err).To(MatchError(ContainSubstring("spec.expire is immutable")))
-	})
+				expectCreatedAt := backup.Status.CreatedAt.Time
+
+				By("simulate backup expiration")
+				newCreatedAt := simulateExpire(ctx, backup, offset)
+
+				By("wait for the backup to be deleted")
+				testutil.CheckDeletedEventually[mantlev1.MantleBackup](ctx, k8sClient, backup.Name, backup.Namespace)
 
-	DescribeTable("MantleBackup expiration",
-		func(ctx SpecContext, offset time.Duration) {
+				By("check the queued backup has the correct createdAt")
+				// If expiration is deferred, the backup with the new createdAt is queued.
+				// Otherwise, the backup is not queued after updating the createdAt, so the backup has the original createdAt.
+				if offset > 0 {
+					expectCreatedAt = newCreatedAt
+				}
+				v, ok := lastExpireQueuedBackups.Load(types.NamespacedName{Namespace: backup.Namespace, Name: backup.Name})
+				Expect(ok).To(BeTrue())
+				createdAt := v.(*mantlev1.MantleBackup).Status.CreatedAt.Time
+				Expect(createdAt).To(BeTemporally("==", expectCreatedAt))
+			},
+			Entry("an already expired backup should be deleted immediately", -time.Hour),
+			Entry("a near expiring backup should be deleted after expiration", 10*time.Second),
+		)
+
+		It("should retain the backup if it has the retain-if-expired annotation", func(ctx SpecContext) {
 			_, pvc, err := resMgr.CreateUniquePVAndPVC(ctx, ns)
 			Expect(err).NotTo(HaveOccurred())
 
@@ -229,84 +266,284 @@ var _ = Describe("MantleBackup controller", func() {
 			By("waiting for the backup to be ready")
 			resMgr.WaitForBackupReady(ctx, backup)
 
-			expectCreatedAt := backup.Status.CreatedAt.Time
-
 			By("simulate backup expiration")
-			newCreatedAt := simulateExpire(ctx, backup, offset)
+			simulateExpire(ctx, backup, -time.Hour)
 
-			By("wait for the backup to be deleted")
-			testutil.CheckDeletedEventually[mantlev1.MantleBackup](ctx, k8sClient, backup.Name, backup.Namespace)
+			By("checking the backup is not deleted")
+			err = k8sClient.Get(ctx, types.NamespacedName{Name: backup.Name, Namespace: backup.Namespace}, backup)
+			Expect(err).NotTo(HaveOccurred())
+		})
 
-			By("check the queued backup has the correct createdAt")
-			// If expiration is deferred, the backup with the new createdAt is queued.
-			// Otherwise, the backup is not queued after updating the createdAt, so the backup has the original createdAt.
-			if offset > 0 {
-				expectCreatedAt = newCreatedAt
-			}
-			v, ok := lastExpireQueuedBackups.Load(types.NamespacedName{Namespace: backup.Namespace, Name: backup.Name})
-			Expect(ok).To(BeTrue())
-			createdAt := v.(*mantlev1.MantleBackup).Status.CreatedAt.Time
-			Expect(createdAt).To(BeTemporally("==", expectCreatedAt))
-		},
-		Entry("an already expired backup should be deleted immediately", -time.Hour),
-		Entry("a near expiring backup should be deleted after expiration", 10*time.Second),
-	)
+		It("should not be ready to use if the PVC is the lost state from the beginning", func(ctx SpecContext) {
+			pv, pvc, err := resMgr.CreateUniquePVAndPVC(ctx, ns)
+			Expect(err).NotTo(HaveOccurred())
+			pv.Status.Phase = corev1.VolumeAvailable
+			err = k8sClient.Status().Update(ctx, pv)
+			Expect(err).NotTo(HaveOccurred())
+			pvc.Status.Phase = corev1.ClaimLost
+			err = k8sClient.Status().Update(ctx, pvc)
+			Expect(err).NotTo(HaveOccurred())
 
-	It("should retain the backup if it has the retain-if-expired annotation", func(ctx SpecContext) {
-		_, pvc, err := resMgr.CreateUniquePVAndPVC(ctx, ns)
-		Expect(err).NotTo(HaveOccurred())
+			backup, err := resMgr.CreateUniqueBackupFor(ctx, pvc)
+			Expect(err).NotTo(HaveOccurred())
 
-		backup, err := resMgr.CreateUniqueBackupFor(ctx, pvc)
-		Expect(err).NotTo(HaveOccurred())
+			waitForBackupNotReady(ctx, backup)
+		})
 
-		By("waiting for the backup to be ready")
-		resMgr.WaitForBackupReady(ctx, backup)
+		It("should not be ready to use if specified non-existent PVC name", func(ctx SpecContext) {
+			var err error
+			backup, err := resMgr.CreateUniqueBackupFor(ctx, &corev1.PersistentVolumeClaim{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "non-existent-pvc",
+					Namespace: ns,
+				},
+			})
+			Expect(err).NotTo(HaveOccurred())
 
-		By("simulate backup expiration")
-		simulateExpire(ctx, backup, -time.Hour)
+			waitForBackupNotReady(ctx, backup)
+		})
 
-		By("checking the backup is not deleted")
-		err = k8sClient.Get(ctx, types.NamespacedName{Name: backup.Name, Namespace: backup.Namespace}, backup)
-		Expect(err).NotTo(HaveOccurred())
+		It("should fail the resource creation the second time if the same MantleBackup is created twice", func(ctx SpecContext) {
+			_, pvc, err := resMgr.CreateUniquePVAndPVC(ctx, ns)
+			Expect(err).NotTo(HaveOccurred())
+
+			backup, err := resMgr.CreateUniqueBackupFor(ctx, pvc)
+			Expect(err).NotTo(HaveOccurred())
+
+			err = k8sClient.Create(ctx, backup)
+			Expect(err).To(HaveOccurred())
+		})
 	})
 
-	It("should not be ready to use if the PVC is the lost state from the beginning", func(ctx SpecContext) {
-		pv, pvc, err := resMgr.CreateUniquePVAndPVC(ctx, ns)
-		Expect(err).NotTo(HaveOccurred())
-		pv.Status.Phase = corev1.VolumeAvailable
-		err = k8sClient.Status().Update(ctx, pv)
-		Expect(err).NotTo(HaveOccurred())
-		pvc.Status.Phase = corev1.ClaimLost
-		err = k8sClient.Status().Update(ctx, pvc)
-		Expect(err).NotTo(HaveOccurred())
+	Context("when the role is `primary`", func() {
+		BeforeEach(func() {
+			mgrUtil = testutil.NewManagerUtil(context.Background(), cfg, scheme.Scheme)
+
+			reconciler = NewMantleBackupReconciler(
+				mgrUtil.GetManager().GetClient(),
+				mgrUtil.GetManager().GetScheme(),
+				resMgr.ClusterID,
+				RolePrimary,
+				&PrimarySettings{
+					Client: &mockGRPCClient{},
+				},
+			)
+			reconciler.ceph = testutil.NewFakeRBD()
+			err := reconciler.SetupWithManager(mgrUtil.GetManager())
+			Expect(err).NotTo(HaveOccurred())
 
-		backup, err := resMgr.CreateUniqueBackupFor(ctx, pvc)
-		Expect(err).NotTo(HaveOccurred())
+			setupExpireQueueSniffer()
 
-		waitForBackupNotReady(ctx, backup)
-	})
+			mgrUtil.Start()
+			time.Sleep(100 * time.Millisecond)
 
-	It("should not be ready to use if specified non-existent PVC name", func(ctx SpecContext) {
-		var err error
-		backup, err := resMgr.CreateUniqueBackupFor(ctx, &corev1.PersistentVolumeClaim{
-			ObjectMeta: metav1.ObjectMeta{
-				Name:      "non-existent-pvc",
-				Namespace: ns,
-			},
+			ns = resMgr.CreateNamespace()
 		})
-		Expect(err).NotTo(HaveOccurred())
 
-		waitForBackupNotReady(ctx, backup)
-	})
+		It("should be synced to remote", func(ctx SpecContext) {
+			pv, pvc, err := resMgr.CreateUniquePVAndPVC(ctx, ns)
+			Expect(err).NotTo(HaveOccurred())
 
-	It("should fail the resource creation the second time if the same MantleBackup is created twice", func(ctx SpecContext) {
-		_, pvc, err := resMgr.CreateUniquePVAndPVC(ctx, ns)
-		Expect(err).NotTo(HaveOccurred())
+			backup, err := resMgr.CreateUniqueBackupFor(ctx, pvc)
+			Expect(err).NotTo(HaveOccurred())
 
-		backup, err := resMgr.CreateUniqueBackupFor(ctx, pvc)
-		Expect(err).NotTo(HaveOccurred())
+			waitForHavingFinalizer(ctx, backup)
+			resMgr.WaitForBackupReady(ctx, backup)
+			resMgr.WaitForBackupSyncedToRemote(ctx, backup)
+
+			pvcJS := backup.Status.PVCManifest
+			Expect(pvcJS).NotTo(BeEmpty())
+			pvcStored := corev1.PersistentVolumeClaim{}
+			err = json.Unmarshal([]byte(pvcJS), &pvcStored)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(pvcStored.Name).To(Equal(pvc.Name))
+			Expect(pvcStored.Namespace).To(Equal(pvc.Namespace))
+
+			pvJS := backup.Status.PVManifest
+			Expect(pvJS).NotTo(BeEmpty())
+			pvStored := corev1.PersistentVolume{}
+			err = json.Unmarshal([]byte(pvJS), &pvStored)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(pvStored.Name).To(Equal(pv.Name))
 
-		err = k8sClient.Create(ctx, backup)
-		Expect(err).To(HaveOccurred())
+			snaps, err := reconciler.ceph.RBDSnapLs(resMgr.PoolName, pv.Spec.CSI.VolumeAttributes["imageName"])
+			Expect(err).NotTo(HaveOccurred())
+			Expect(snaps).To(HaveLen(1))
+			snapID := backup.Status.SnapID
+			Expect(snapID).To(Equal(&snaps[0].Id))
+
+			// TODO: Currently, there is no way to check if the annotations are set correctly.
+			// After implementing export() function, the annotations check should be added
+			// for various conditions.
+
+			err = k8sClient.Delete(ctx, backup)
+			Expect(err).NotTo(HaveOccurred())
+
+			testutil.CheckDeletedEventually[mantlev1.MantleBackup](ctx, k8sClient, backup.Name, backup.Namespace)
+		})
 	})
 })
+
+type mockGRPCClient struct {
+	backup mantlev1.MantleBackup
+}
+
+var _ proto.MantleServiceClient = (*mockGRPCClient)(nil)
+
+func (m *mockGRPCClient) CreateOrUpdatePVC(
+	ctx context.Context,
+	req *proto.CreateOrUpdatePVCRequest,
+	opts ...grpc.CallOption,
+) (*proto.CreateOrUpdatePVCResponse, error) {
+	return &proto.CreateOrUpdatePVCResponse{}, nil
+}
+
+func (m *mockGRPCClient) CreateOrUpdateMantleBackup(
+	ctx context.Context,
+	req *proto.CreateOrUpdateMantleBackupRequest,
+	opts ...grpc.CallOption,
+) (*proto.CreateOrUpdateMantleBackupResponse, error) {
+	err := json.Unmarshal([]byte(req.MantleBackup), &m.backup)
+	if err != nil {
+		return nil, err
+	}
+	return &proto.CreateOrUpdateMantleBackupResponse{}, nil
+}
+
+func (m *mockGRPCClient) ListMantleBackup(
+	ctx context.Context,
+	req *proto.ListMantleBackupRequest,
+	opts ...grpc.CallOption,
+) (*proto.ListMantleBackupResponse, error) {
+	backups := []mantlev1.MantleBackup{m.backup}
+	data, err := json.Marshal(backups)
+	if err != nil {
+		return nil, err
+	}
+	return &proto.ListMantleBackupResponse{
+		MantleBackupList: data,
+	}, nil
+}
+
+func int2Ptr(i int) *int {
+	return &i
+}
+
+var _ = Describe("searchDiffOriginMantleBackup", func() {
+	testMantleBackup := mantlev1.MantleBackup{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "test0",
+		},
+		Status: mantlev1.MantleBackupStatus{
+			SnapID: int2Ptr(5),
+		},
+	}
+
+	basePrimaryBackups := []mantlev1.MantleBackup{
+		{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "test1",
+			},
+			Status: mantlev1.MantleBackupStatus{
+				Conditions: []metav1.Condition{
+					{
+						Type:   mantlev1.BackupConditionReadyToUse,
+						Status: metav1.ConditionTrue,
+					},
+				},
+				SnapID: int2Ptr(1),
+			},
+		},
+		{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "test2",
+			},
+			Status: mantlev1.MantleBackupStatus{
+				Conditions: []metav1.Condition{
+					{
+						Type:   mantlev1.BackupConditionReadyToUse,
+						Status: metav1.ConditionTrue,
+					},
+				},
+				SnapID: int2Ptr(6),
+			},
+		},
+		{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "test3",
+			},
+			Status: mantlev1.MantleBackupStatus{
+				Conditions: []metav1.Condition{
+					{
+						Type:   mantlev1.BackupConditionReadyToUse,
+						Status: metav1.ConditionTrue,
+					},
+				},
+				SnapID: int2Ptr(3),
+			},
+		},
+		{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "test4",
+			},
+			Status: mantlev1.MantleBackupStatus{
+				Conditions: []metav1.Condition{
+					{
+						Type:   mantlev1.BackupConditionReadyToUse,
+						Status: metav1.ConditionTrue,
+					},
+				},
+				SnapID: int2Ptr(4),
+			},
+		},
+	}
+	// Note that slices.Clone() does the shallow copy.
+	// ref. https://pkg.go.dev/slices#Clone
+	primaryBackupsWithConditionFalse := slices.Clone(basePrimaryBackups)
+	primaryBackupsWithConditionFalse[2] = *basePrimaryBackups[2].DeepCopy()
+	meta.SetStatusCondition(&primaryBackupsWithConditionFalse[2].Status.Conditions,
+		metav1.Condition{
+			Type:   mantlev1.BackupConditionReadyToUse,
+			Status: metav1.ConditionFalse,
+		})
+	primaryBackupsWithDeletionTimestamp := slices.Clone(basePrimaryBackups)
+	primaryBackupsWithDeletionTimestamp[2] = *basePrimaryBackups[2].DeepCopy()
+	now := metav1.Now()
+	primaryBackupsWithDeletionTimestamp[2].SetDeletionTimestamp(&now)
+
+	testSecondaryMantleBackups := map[string]*mantlev1.MantleBackup{
+		"test1": basePrimaryBackups[0].DeepCopy(),
+		// "test2" cannot exist on the secondary cluster
+		// because it has a higher snapID than "test0".
+		"test3": basePrimaryBackups[2].DeepCopy(),
+		// "test4" is intentionally omitted.
+	}
+
+	DescribeTable("Search for the MantleBackup which is used for the diff origin",
+		func(backup *mantlev1.MantleBackup,
+			primaryBackups []mantlev1.MantleBackup,
+			secondaryBackupMap map[string]*mantlev1.MantleBackup,
+			shouldFindBackup bool,
+			expectedBackupName string) {
+			foundBackup := searchForDiffOriginMantleBackup(backup, primaryBackups, secondaryBackupMap)
+			if shouldFindBackup {
+				Expect(foundBackup).NotTo(BeNil())
+				Expect(foundBackup.GetName()).To(Equal(expectedBackupName))
+			} else {
+				Expect(foundBackup).To(BeNil())
+			}
+		},
+		Entry("should return nil when no MantleBackup found on the secondary cluster",
+			&testMantleBackup, basePrimaryBackups, make(map[string]*mantlev1.MantleBackup),
+			false, ""),
+		Entry("should find the correct MantleBackup",
+			&testMantleBackup, basePrimaryBackups, testSecondaryMantleBackups,
+			true, "test3"),
+		Entry("should skip the not-ready MantleBackup",
+			&testMantleBackup, primaryBackupsWithConditionFalse, testSecondaryMantleBackups,
+			true, "test1"),
+		Entry("should skip the MantleBackup with the deletion timestamp",
+			&testMantleBackup, primaryBackupsWithDeletionTimestamp, testSecondaryMantleBackups,
+			true, "test1"),
+	)
+})
diff --git a/internal/controller/replication.go b/internal/controller/replication.go
index f3be04fe..f7f58193 100644
--- a/internal/controller/replication.go
+++ b/internal/controller/replication.go
@@ -9,6 +9,7 @@ import (
 	"github.com/cybozu-go/mantle/pkg/controller/proto"
 	"google.golang.org/grpc"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/types"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -184,3 +185,22 @@ func (s *SecondaryServer) CreateOrUpdateMantleBackup(
 
 	return &proto.CreateOrUpdateMantleBackupResponse{}, nil
 }
+
+func (s *SecondaryServer) ListMantleBackup(
+	ctx context.Context,
+	req *proto.ListMantleBackupRequest,
+) (*proto.ListMantleBackupResponse, error) {
+	var backupList mantlev1.MantleBackupList
+	err := s.reader.List(ctx, &backupList, &client.ListOptions{
+		LabelSelector: labels.SelectorFromSet(map[string]string{labelRemoteBackupTargetPVCUID: req.PvcUID}),
+		Namespace:     req.Namespace,
+	})
+	if err != nil {
+		return nil, err
+	}
+	data, err := json.Marshal(backupList.Items)
+	if err != nil {
+		return nil, err
+	}
+	return &proto.ListMantleBackupResponse{MantleBackupList: data}, nil
+}
diff --git a/pkg/controller/proto/controller.pb.go b/pkg/controller/proto/controller.pb.go
index bcbcc878..fb9f0d80 100644
--- a/pkg/controller/proto/controller.pb.go
+++ b/pkg/controller/proto/controller.pb.go
@@ -203,6 +203,110 @@ func (*CreateOrUpdateMantleBackupResponse) Descriptor() ([]byte, []int) {
 	return file_pkg_controller_proto_controller_proto_rawDescGZIP(), []int{3}
 }
 
+// ListMantleBackupRequest is a request message for ListMantleBackup RPC.
+type ListMantleBackupRequest struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	PvcUID    string `protobuf:"bytes,1,opt,name=pvcUID,proto3" json:"pvcUID,omitempty"`
+	Namespace string `protobuf:"bytes,2,opt,name=namespace,proto3" json:"namespace,omitempty"`
+}
+
+func (x *ListMantleBackupRequest) Reset() {
+	*x = ListMantleBackupRequest{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_pkg_controller_proto_controller_proto_msgTypes[4]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *ListMantleBackupRequest) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*ListMantleBackupRequest) ProtoMessage() {}
+
+func (x *ListMantleBackupRequest) ProtoReflect() protoreflect.Message {
+	mi := &file_pkg_controller_proto_controller_proto_msgTypes[4]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use ListMantleBackupRequest.ProtoReflect.Descriptor instead.
+func (*ListMantleBackupRequest) Descriptor() ([]byte, []int) {
+	return file_pkg_controller_proto_controller_proto_rawDescGZIP(), []int{4}
+}
+
+func (x *ListMantleBackupRequest) GetPvcUID() string {
+	if x != nil {
+		return x.PvcUID
+	}
+	return ""
+}
+
+func (x *ListMantleBackupRequest) GetNamespace() string {
+	if x != nil {
+		return x.Namespace
+	}
+	return ""
+}
+
+// ListMantleBackupResponse is a response message for ListMantleBackup RPC.
+type ListMantleBackupResponse struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	MantleBackupList []byte `protobuf:"bytes,1,opt,name=mantleBackupList,proto3" json:"mantleBackupList,omitempty"`
+}
+
+func (x *ListMantleBackupResponse) Reset() {
+	*x = ListMantleBackupResponse{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_pkg_controller_proto_controller_proto_msgTypes[5]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *ListMantleBackupResponse) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*ListMantleBackupResponse) ProtoMessage() {}
+
+func (x *ListMantleBackupResponse) ProtoReflect() protoreflect.Message {
+	mi := &file_pkg_controller_proto_controller_proto_msgTypes[5]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use ListMantleBackupResponse.ProtoReflect.Descriptor instead.
+func (*ListMantleBackupResponse) Descriptor() ([]byte, []int) {
+	return file_pkg_controller_proto_controller_proto_rawDescGZIP(), []int{5}
+}
+
+func (x *ListMantleBackupResponse) GetMantleBackupList() []byte {
+	if x != nil {
+		return x.MantleBackupList
+	}
+	return nil
+}
+
 var File_pkg_controller_proto_controller_proto protoreflect.FileDescriptor
 
 var file_pkg_controller_proto_controller_proto_rawDesc = []byte{
@@ -221,25 +325,40 @@ var file_pkg_controller_proto_controller_proto_rawDesc = []byte{
 	0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0c, 0x6d, 0x61, 0x6e, 0x74, 0x6c, 0x65, 0x42, 0x61,
 	0x63, 0x6b, 0x75, 0x70, 0x22, 0x24, 0x0a, 0x22, 0x43, 0x72, 0x65, 0x61, 0x74, 0x65, 0x4f, 0x72,
 	0x55, 0x70, 0x64, 0x61, 0x74, 0x65, 0x4d, 0x61, 0x6e, 0x74, 0x6c, 0x65, 0x42, 0x61, 0x63, 0x6b,
-	0x75, 0x70, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x32, 0xda, 0x01, 0x0a, 0x0d, 0x4d,
-	0x61, 0x6e, 0x74, 0x6c, 0x65, 0x53, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x12, 0x56, 0x0a, 0x11,
-	0x43, 0x72, 0x65, 0x61, 0x74, 0x65, 0x4f, 0x72, 0x55, 0x70, 0x64, 0x61, 0x74, 0x65, 0x50, 0x56,
-	0x43, 0x12, 0x1f, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x43, 0x72, 0x65, 0x61, 0x74, 0x65,
-	0x4f, 0x72, 0x55, 0x70, 0x64, 0x61, 0x74, 0x65, 0x50, 0x56, 0x43, 0x52, 0x65, 0x71, 0x75, 0x65,
-	0x73, 0x74, 0x1a, 0x20, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x43, 0x72, 0x65, 0x61, 0x74,
-	0x65, 0x4f, 0x72, 0x55, 0x70, 0x64, 0x61, 0x74, 0x65, 0x50, 0x56, 0x43, 0x52, 0x65, 0x73, 0x70,
-	0x6f, 0x6e, 0x73, 0x65, 0x12, 0x71, 0x0a, 0x1a, 0x43, 0x72, 0x65, 0x61, 0x74, 0x65, 0x4f, 0x72,
-	0x55, 0x70, 0x64, 0x61, 0x74, 0x65, 0x4d, 0x61, 0x6e, 0x74, 0x6c, 0x65, 0x42, 0x61, 0x63, 0x6b,
-	0x75, 0x70, 0x12, 0x28, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x43, 0x72, 0x65, 0x61, 0x74,
-	0x65, 0x4f, 0x72, 0x55, 0x70, 0x64, 0x61, 0x74, 0x65, 0x4d, 0x61, 0x6e, 0x74, 0x6c, 0x65, 0x42,
-	0x61, 0x63, 0x6b, 0x75, 0x70, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x29, 0x2e, 0x70,
-	0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x43, 0x72, 0x65, 0x61, 0x74, 0x65, 0x4f, 0x72, 0x55, 0x70, 0x64,
-	0x61, 0x74, 0x65, 0x4d, 0x61, 0x6e, 0x74, 0x6c, 0x65, 0x42, 0x61, 0x63, 0x6b, 0x75, 0x70, 0x52,
-	0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x42, 0x32, 0x5a, 0x30, 0x67, 0x69, 0x74, 0x68, 0x75,
-	0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x63, 0x79, 0x62, 0x6f, 0x7a, 0x75, 0x2d, 0x67, 0x6f, 0x2f,
-	0x6d, 0x61, 0x6e, 0x74, 0x6c, 0x65, 0x2f, 0x70, 0x6b, 0x67, 0x2f, 0x63, 0x6f, 0x6e, 0x74, 0x72,
-	0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x06, 0x70, 0x72, 0x6f,
-	0x74, 0x6f, 0x33,
+	0x75, 0x70, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x4f, 0x0a, 0x17, 0x4c, 0x69,
+	0x73, 0x74, 0x4d, 0x61, 0x6e, 0x74, 0x6c, 0x65, 0x42, 0x61, 0x63, 0x6b, 0x75, 0x70, 0x52, 0x65,
+	0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x16, 0x0a, 0x06, 0x70, 0x76, 0x63, 0x55, 0x49, 0x44, 0x18,
+	0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x06, 0x70, 0x76, 0x63, 0x55, 0x49, 0x44, 0x12, 0x1c, 0x0a,
+	0x09, 0x6e, 0x61, 0x6d, 0x65, 0x73, 0x70, 0x61, 0x63, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09,
+	0x52, 0x09, 0x6e, 0x61, 0x6d, 0x65, 0x73, 0x70, 0x61, 0x63, 0x65, 0x22, 0x46, 0x0a, 0x18, 0x4c,
+	0x69, 0x73, 0x74, 0x4d, 0x61, 0x6e, 0x74, 0x6c, 0x65, 0x42, 0x61, 0x63, 0x6b, 0x75, 0x70, 0x52,
+	0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x2a, 0x0a, 0x10, 0x6d, 0x61, 0x6e, 0x74, 0x6c,
+	0x65, 0x42, 0x61, 0x63, 0x6b, 0x75, 0x70, 0x4c, 0x69, 0x73, 0x74, 0x18, 0x01, 0x20, 0x01, 0x28,
+	0x0c, 0x52, 0x10, 0x6d, 0x61, 0x6e, 0x74, 0x6c, 0x65, 0x42, 0x61, 0x63, 0x6b, 0x75, 0x70, 0x4c,
+	0x69, 0x73, 0x74, 0x32, 0xaf, 0x02, 0x0a, 0x0d, 0x4d, 0x61, 0x6e, 0x74, 0x6c, 0x65, 0x53, 0x65,
+	0x72, 0x76, 0x69, 0x63, 0x65, 0x12, 0x56, 0x0a, 0x11, 0x43, 0x72, 0x65, 0x61, 0x74, 0x65, 0x4f,
+	0x72, 0x55, 0x70, 0x64, 0x61, 0x74, 0x65, 0x50, 0x56, 0x43, 0x12, 0x1f, 0x2e, 0x70, 0x72, 0x6f,
+	0x74, 0x6f, 0x2e, 0x43, 0x72, 0x65, 0x61, 0x74, 0x65, 0x4f, 0x72, 0x55, 0x70, 0x64, 0x61, 0x74,
+	0x65, 0x50, 0x56, 0x43, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x20, 0x2e, 0x70, 0x72,
+	0x6f, 0x74, 0x6f, 0x2e, 0x43, 0x72, 0x65, 0x61, 0x74, 0x65, 0x4f, 0x72, 0x55, 0x70, 0x64, 0x61,
+	0x74, 0x65, 0x50, 0x56, 0x43, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x71, 0x0a,
+	0x1a, 0x43, 0x72, 0x65, 0x61, 0x74, 0x65, 0x4f, 0x72, 0x55, 0x70, 0x64, 0x61, 0x74, 0x65, 0x4d,
+	0x61, 0x6e, 0x74, 0x6c, 0x65, 0x42, 0x61, 0x63, 0x6b, 0x75, 0x70, 0x12, 0x28, 0x2e, 0x70, 0x72,
+	0x6f, 0x74, 0x6f, 0x2e, 0x43, 0x72, 0x65, 0x61, 0x74, 0x65, 0x4f, 0x72, 0x55, 0x70, 0x64, 0x61,
+	0x74, 0x65, 0x4d, 0x61, 0x6e, 0x74, 0x6c, 0x65, 0x42, 0x61, 0x63, 0x6b, 0x75, 0x70, 0x52, 0x65,
+	0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x29, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x43, 0x72,
+	0x65, 0x61, 0x74, 0x65, 0x4f, 0x72, 0x55, 0x70, 0x64, 0x61, 0x74, 0x65, 0x4d, 0x61, 0x6e, 0x74,
+	0x6c, 0x65, 0x42, 0x61, 0x63, 0x6b, 0x75, 0x70, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65,
+	0x12, 0x53, 0x0a, 0x10, 0x4c, 0x69, 0x73, 0x74, 0x4d, 0x61, 0x6e, 0x74, 0x6c, 0x65, 0x42, 0x61,
+	0x63, 0x6b, 0x75, 0x70, 0x12, 0x1e, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x4c, 0x69, 0x73,
+	0x74, 0x4d, 0x61, 0x6e, 0x74, 0x6c, 0x65, 0x42, 0x61, 0x63, 0x6b, 0x75, 0x70, 0x52, 0x65, 0x71,
+	0x75, 0x65, 0x73, 0x74, 0x1a, 0x1f, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x4c, 0x69, 0x73,
+	0x74, 0x4d, 0x61, 0x6e, 0x74, 0x6c, 0x65, 0x42, 0x61, 0x63, 0x6b, 0x75, 0x70, 0x52, 0x65, 0x73,
+	0x70, 0x6f, 0x6e, 0x73, 0x65, 0x42, 0x32, 0x5a, 0x30, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e,
+	0x63, 0x6f, 0x6d, 0x2f, 0x63, 0x79, 0x62, 0x6f, 0x7a, 0x75, 0x2d, 0x67, 0x6f, 0x2f, 0x6d, 0x61,
+	0x6e, 0x74, 0x6c, 0x65, 0x2f, 0x70, 0x6b, 0x67, 0x2f, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c,
+	0x6c, 0x65, 0x72, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f,
+	0x33,
 }
 
 var (
@@ -254,20 +373,24 @@ func file_pkg_controller_proto_controller_proto_rawDescGZIP() []byte {
 	return file_pkg_controller_proto_controller_proto_rawDescData
 }
 
-var file_pkg_controller_proto_controller_proto_msgTypes = make([]protoimpl.MessageInfo, 4)
+var file_pkg_controller_proto_controller_proto_msgTypes = make([]protoimpl.MessageInfo, 6)
 var file_pkg_controller_proto_controller_proto_goTypes = []interface{}{
 	(*CreateOrUpdatePVCRequest)(nil),           // 0: proto.CreateOrUpdatePVCRequest
 	(*CreateOrUpdatePVCResponse)(nil),          // 1: proto.CreateOrUpdatePVCResponse
 	(*CreateOrUpdateMantleBackupRequest)(nil),  // 2: proto.CreateOrUpdateMantleBackupRequest
 	(*CreateOrUpdateMantleBackupResponse)(nil), // 3: proto.CreateOrUpdateMantleBackupResponse
+	(*ListMantleBackupRequest)(nil),            // 4: proto.ListMantleBackupRequest
+	(*ListMantleBackupResponse)(nil),           // 5: proto.ListMantleBackupResponse
 }
 var file_pkg_controller_proto_controller_proto_depIdxs = []int32{
 	0, // 0: proto.MantleService.CreateOrUpdatePVC:input_type -> proto.CreateOrUpdatePVCRequest
 	2, // 1: proto.MantleService.CreateOrUpdateMantleBackup:input_type -> proto.CreateOrUpdateMantleBackupRequest
-	1, // 2: proto.MantleService.CreateOrUpdatePVC:output_type -> proto.CreateOrUpdatePVCResponse
-	3, // 3: proto.MantleService.CreateOrUpdateMantleBackup:output_type -> proto.CreateOrUpdateMantleBackupResponse
-	2, // [2:4] is the sub-list for method output_type
-	0, // [0:2] is the sub-list for method input_type
+	4, // 2: proto.MantleService.ListMantleBackup:input_type -> proto.ListMantleBackupRequest
+	1, // 3: proto.MantleService.CreateOrUpdatePVC:output_type -> proto.CreateOrUpdatePVCResponse
+	3, // 4: proto.MantleService.CreateOrUpdateMantleBackup:output_type -> proto.CreateOrUpdateMantleBackupResponse
+	5, // 5: proto.MantleService.ListMantleBackup:output_type -> proto.ListMantleBackupResponse
+	3, // [3:6] is the sub-list for method output_type
+	0, // [0:3] is the sub-list for method input_type
 	0, // [0:0] is the sub-list for extension type_name
 	0, // [0:0] is the sub-list for extension extendee
 	0, // [0:0] is the sub-list for field type_name
@@ -327,6 +450,30 @@ func file_pkg_controller_proto_controller_proto_init() {
 				return nil
 			}
 		}
+		file_pkg_controller_proto_controller_proto_msgTypes[4].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*ListMantleBackupRequest); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_pkg_controller_proto_controller_proto_msgTypes[5].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*ListMantleBackupResponse); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
 	}
 	type x struct{}
 	out := protoimpl.TypeBuilder{
@@ -334,7 +481,7 @@ func file_pkg_controller_proto_controller_proto_init() {
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
 			RawDescriptor: file_pkg_controller_proto_controller_proto_rawDesc,
 			NumEnums:      0,
-			NumMessages:   4,
+			NumMessages:   6,
 			NumExtensions: 0,
 			NumServices:   1,
 		},
diff --git a/pkg/controller/proto/controller.proto b/pkg/controller/proto/controller.proto
index 79adbf1d..d1935094 100644
--- a/pkg/controller/proto/controller.proto
+++ b/pkg/controller/proto/controller.proto
@@ -24,7 +24,19 @@ message CreateOrUpdateMantleBackupResponse {
     // nothing.
 }
 
+// ListMantleBackupRequest is a request message for ListMantleBackup RPC.
+message ListMantleBackupRequest {
+    string pvcUID = 1;
+    string namespace = 2;
+}
+
+// ListMantleBackupResponse is a response message for ListMantleBackup RPC.
+message ListMantleBackupResponse {
+    bytes mantleBackupList = 1;
+}
+
 service MantleService {
     rpc CreateOrUpdatePVC(CreateOrUpdatePVCRequest) returns (CreateOrUpdatePVCResponse);
     rpc CreateOrUpdateMantleBackup(CreateOrUpdateMantleBackupRequest) returns (CreateOrUpdateMantleBackupResponse);
+    rpc ListMantleBackup(ListMantleBackupRequest) returns (ListMantleBackupResponse);
 }
diff --git a/pkg/controller/proto/controller_grpc.pb.go b/pkg/controller/proto/controller_grpc.pb.go
index a000e4ed..5e612e8f 100644
--- a/pkg/controller/proto/controller_grpc.pb.go
+++ b/pkg/controller/proto/controller_grpc.pb.go
@@ -21,6 +21,7 @@ const _ = grpc.SupportPackageIsVersion9
 const (
 	MantleService_CreateOrUpdatePVC_FullMethodName          = "/proto.MantleService/CreateOrUpdatePVC"
 	MantleService_CreateOrUpdateMantleBackup_FullMethodName = "/proto.MantleService/CreateOrUpdateMantleBackup"
+	MantleService_ListMantleBackup_FullMethodName           = "/proto.MantleService/ListMantleBackup"
 )
 
 // MantleServiceClient is the client API for MantleService service.
@@ -29,6 +30,7 @@ const (
 type MantleServiceClient interface {
 	CreateOrUpdatePVC(ctx context.Context, in *CreateOrUpdatePVCRequest, opts ...grpc.CallOption) (*CreateOrUpdatePVCResponse, error)
 	CreateOrUpdateMantleBackup(ctx context.Context, in *CreateOrUpdateMantleBackupRequest, opts ...grpc.CallOption) (*CreateOrUpdateMantleBackupResponse, error)
+	ListMantleBackup(ctx context.Context, in *ListMantleBackupRequest, opts ...grpc.CallOption) (*ListMantleBackupResponse, error)
 }
 
 type mantleServiceClient struct {
@@ -59,12 +61,23 @@ func (c *mantleServiceClient) CreateOrUpdateMantleBackup(ctx context.Context, in
 	return out, nil
 }
 
+func (c *mantleServiceClient) ListMantleBackup(ctx context.Context, in *ListMantleBackupRequest, opts ...grpc.CallOption) (*ListMantleBackupResponse, error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	out := new(ListMantleBackupResponse)
+	err := c.cc.Invoke(ctx, MantleService_ListMantleBackup_FullMethodName, in, out, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
 // MantleServiceServer is the server API for MantleService service.
 // All implementations must embed UnimplementedMantleServiceServer
 // for forward compatibility.
 type MantleServiceServer interface {
 	CreateOrUpdatePVC(context.Context, *CreateOrUpdatePVCRequest) (*CreateOrUpdatePVCResponse, error)
 	CreateOrUpdateMantleBackup(context.Context, *CreateOrUpdateMantleBackupRequest) (*CreateOrUpdateMantleBackupResponse, error)
+	ListMantleBackup(context.Context, *ListMantleBackupRequest) (*ListMantleBackupResponse, error)
 	mustEmbedUnimplementedMantleServiceServer()
 }
 
@@ -81,6 +94,9 @@ func (UnimplementedMantleServiceServer) CreateOrUpdatePVC(context.Context, *Crea
 func (UnimplementedMantleServiceServer) CreateOrUpdateMantleBackup(context.Context, *CreateOrUpdateMantleBackupRequest) (*CreateOrUpdateMantleBackupResponse, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method CreateOrUpdateMantleBackup not implemented")
 }
+func (UnimplementedMantleServiceServer) ListMantleBackup(context.Context, *ListMantleBackupRequest) (*ListMantleBackupResponse, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method ListMantleBackup not implemented")
+}
 func (UnimplementedMantleServiceServer) mustEmbedUnimplementedMantleServiceServer() {}
 func (UnimplementedMantleServiceServer) testEmbeddedByValue()                       {}
 
@@ -138,6 +154,24 @@ func _MantleService_CreateOrUpdateMantleBackup_Handler(srv interface{}, ctx cont
 	return interceptor(ctx, in, info, handler)
 }
 
+func _MantleService_ListMantleBackup_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(ListMantleBackupRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(MantleServiceServer).ListMantleBackup(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: MantleService_ListMantleBackup_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(MantleServiceServer).ListMantleBackup(ctx, req.(*ListMantleBackupRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
 // MantleService_ServiceDesc is the grpc.ServiceDesc for MantleService service.
 // It's only intended for direct use with grpc.RegisterService,
 // and not to be introspected or modified (even as a copy)
@@ -153,6 +187,10 @@ var MantleService_ServiceDesc = grpc.ServiceDesc{
 			MethodName: "CreateOrUpdateMantleBackup",
 			Handler:    _MantleService_CreateOrUpdateMantleBackup_Handler,
 		},
+		{
+			MethodName: "ListMantleBackup",
+			Handler:    _MantleService_ListMantleBackup_Handler,
+		},
 	},
 	Streams:  []grpc.StreamDesc{},
 	Metadata: "pkg/controller/proto/controller.proto",