fix(replication-restore): adjust retain latest mvcc fallback

cmd/backup-manager/app/cmd/restore.go

@@ -95,6 +95,8 @@ func NewRestoreCommand() *cobra.Command {
 	cmd.Flags().BoolVar(&ro.Abort, "abort", false, "Whether to abort/cleanup a failed restore operation")
 	cmd.Flags().IntVar(&ro.ReplicationPhase, "replicationPhase", 0,
 		"Replication restore phase: 1 = snapshot, 2 = log. Omit (or 0) for standard PiTR / snapshot.")
+	cmd.Flags().BoolVar(&ro.ReplicationRetainLatestMVCCVersion, "replicationRetainLatestMVCCVersion", true,
+		"Whether replication restore passes --retain-latest-mvcc-version to BR.")
 	return cmd
 }
 

cmd/backup-manager/app/restore/restore.go

@@ -51,16 +51,19 @@ const (
 // (= not a replication restore), in which case the caller's
 // `append(args, replicationBRFlags(0)...)` is a no-op and standard
 // PiTR semantics apply unchanged. Spec §6.
-func replicationBRFlags(phase int) []string {
+func replicationBRFlags(phase int, retainLatestMVCCVersion bool) []string {
 	if phase == 0 {
 		return nil
 	}
-	return []string{
+	args := []string{
 		fmt.Sprintf("--restore-phase=%d", phase),
 		fmt.Sprintf("--pitr-concurrency=%d", replicationPiTRConcurrency),
 		fmt.Sprintf("--metadata-download-batch-size=%d", replicationMetadataDownloadBatchSize),
-		"--retain-latest-mvcc-version",
 	}
+	if retainLatestMVCCVersion {
+		args = append(args, "--retain-latest-mvcc-version")
+	}
+	return args
 }
 
 type Options struct {
@@ -83,6 +86,9 @@ type Options struct {
 	// CLI flag was not passed and standard PiTR / snapshot semantics
 	// apply. See spec §3 (Option B) and §6 (BR CLI surface).
 	ReplicationPhase int
+	// ReplicationRetainLatestMVCCVersion controls whether replication PiTR
+	// restore passes --retain-latest-mvcc-version to BR.
+	ReplicationRetainLatestMVCCVersion bool
 }
 
 func (ro *Options) restoreData(
@@ -139,7 +145,7 @@ func (ro *Options) restoreData(
 		} else {
 			args = append(args, fullBackupArgs...)
 		}
-		args = append(args, replicationBRFlags(ro.ReplicationPhase)...)
+		args = append(args, replicationBRFlags(ro.ReplicationPhase, ro.ReplicationRetainLatestMVCCVersion)...)
 		restoreType = "point"
 	case string(v1alpha1.RestoreModeVolumeSnapshot):
 		// Currently, we only support aws ebs volume snapshot.

cmd/backup-manager/app/restore/restore_test.go

@@ -36,27 +36,36 @@ func TestOptions_ReplicationPhase_DefaultsToZero(t *testing.T) {
 // PiTR); phases 1 and 2 produce the flags spec §6 mandates.
 func TestReplicationBRFlags(t *testing.T) {
 	cases := []struct {
-		phase int
-		want  []string
+		name                    string
+		phase                   int
+		retainLatestMVCCVersion bool
+		want                    []string
 	}{
-		{phase: 0, want: nil},
-		{phase: 1, want: []string{
+		{name: "standard restore", phase: 0, retainLatestMVCCVersion: true, want: nil},
+		{name: "phase 1 retains latest mvcc", phase: 1, retainLatestMVCCVersion: true, want: []string{
 			"--restore-phase=1",
 			"--pitr-concurrency=1024",
 			"--metadata-download-batch-size=512",
 			"--retain-latest-mvcc-version",
 		}},
-		{phase: 2, want: []string{
+		{name: "phase 2 retains latest mvcc after all compact shards complete", phase: 2, retainLatestMVCCVersion: true, want: []string{
 			"--restore-phase=2",
 			"--pitr-concurrency=1024",
 			"--metadata-download-batch-size=512",
 			"--retain-latest-mvcc-version",
 		}},
+		{name: "phase 2 omits retain latest mvcc when compact has failed shards", phase: 2, retainLatestMVCCVersion: false, want: []string{
+			"--restore-phase=2",
+			"--pitr-concurrency=1024",
+			"--metadata-download-batch-size=512",
+		}},
 	}
 	for _, c := range cases {
-		got := replicationBRFlags(c.phase)
-		if !reflect.DeepEqual(got, c.want) {
-			t.Fatalf("phase=%d: got %v, want %v", c.phase, got, c.want)
-		}
+		t.Run(c.name, func(t *testing.T) {
+			got := replicationBRFlags(c.phase, c.retainLatestMVCCVersion)
+			if !reflect.DeepEqual(got, c.want) {
+				t.Fatalf("phase=%d: got %v, want %v", c.phase, got, c.want)
+			}
+		})
 	}
 }

pkg/backup/restore/replication_handler.go

@@ -56,7 +56,7 @@ import (
 //
 // CompactBackup observation is independent of the step dispatch. Until
 // CompactSettled is written, each Sync tries to settle it with one of the
-// marker reasons consumed by phase-2 BR: AllShardsComplete,
+// marker reasons used when creating the phase-2 Job: AllShardsComplete,
 // ShardsPartialFailed, CompactBackupMismatch, or CompactBackupWaitTimeout.
 // jobBuilderFunc produces the standard PiTR Restore Job that
 // applyReplicationPhase post-processes for one phase of a replication
@@ -388,10 +388,21 @@ func applyReplicationPhase(job *batchv1.Job, restore *v1alpha1.Restore, step str
 			job.Spec.Template.Spec.Containers[0].Args,
 			fmt.Sprintf("--replicationPhase=%d", phase),
 		)
+		if step == label.ReplicationStepLogRestoreVal && shouldDisableRetainLatestMVCCVersion(restore) {
+			job.Spec.Template.Spec.Containers[0].Args = append(
+				job.Spec.Template.Spec.Containers[0].Args,
+				"--replicationRetainLatestMVCCVersion=false",
+			)
+		}
 	}
 	return job
 }
 
+func shouldDisableRetainLatestMVCCVersion(restore *v1alpha1.Restore) bool {
+	_, c := v1alpha1.GetRestoreCondition(&restore.Status, v1alpha1.RestoreCompactSettled)
+	return c != nil && c.Status == corev1.ConditionTrue && c.Reason != "AllShardsComplete"
+}
+
 // compactIsTerminal returns true if CompactBackup has reached either
 // BackupComplete or BackupFailed. The Failed sub-case is intentional — per
 // the spec, the gate proceeds to phase-2 even when compact failed, because

pkg/backup/restore/replication_handler_test.go

@@ -778,9 +778,9 @@ func TestSync_NoCBYet_SnapshotPhaseDrivesIndependently(t *testing.T) {
 }
 
 // CompactSettled with a non-success Reason (e.g. CompactBackupWaitTimeout from
-// the top block) must NOT block the gate at B9. BR phase-2 reads the reason
-// and decides fallback; the controller's responsibility ends at "marker is
-// True", regardless of why.
+// the top block) must NOT block the gate at B9. The phase-2 Job args are
+// derived from the reason; the gate itself only requires "marker is True",
+// regardless of why.
 func TestSync_CompactSettledFailure_DoesNotBlockGate(t *testing.T) {
 	g := NewGomegaWithT(t)
 	handler, h := newHandlerForTest(t)
@@ -980,6 +980,29 @@ func TestApplyReplicationPhase_AppendsReplicationPhaseArg(t *testing.T) {
 		To(Equal([]string{"restore", "--mode=pitr", "--replicationPhase=2"}))
 }
 
+func TestApplyReplicationPhase_LogRestoreDisablesRetainLatestMVCCVersionUnlessAllShardsComplete(t *testing.T) {
+	g := NewGomegaWithT(t)
+	r := newReplicationRestoreFixture("r1", "ns1", "cb1", nil)
+	appendRestoreMarker(r, v1alpha1.RestoreCompactSettled, "ShardsPartialFailed", "")
+
+	out := applyReplicationPhase(fakeBuiltJob(r), r, label.ReplicationStepLogRestoreVal)
+	g.Expect(out.Spec.Template.Spec.Containers[0].Args).To(Equal([]string{
+		"restore",
+		"--mode=pitr",
+		"--replicationPhase=2",
+		"--replicationRetainLatestMVCCVersion=false",
+	}))
+
+	r = newReplicationRestoreFixture("r1", "ns1", "cb1", nil)
+	appendRestoreMarker(r, v1alpha1.RestoreCompactSettled, "AllShardsComplete", "")
+	out = applyReplicationPhase(fakeBuiltJob(r), r, label.ReplicationStepLogRestoreVal)
+	g.Expect(out.Spec.Template.Spec.Containers[0].Args).To(Equal([]string{
+		"restore",
+		"--mode=pitr",
+		"--replicationPhase=2",
+	}))
+}
+
 // TestMakeReplicationBRJob_DelegatesToBuilder verifies that the handler's
 // makeReplicationBRJob calls the injected builder (with isPruneJob=false)
 // and post-processes the returned Job via applyReplicationPhase. The