feat: validate restore schema-version compatibility

docs/operator-public-documentation/preview/operations/backup-and-restore.md

@@ -166,6 +166,32 @@ List backups for your DocumentDB cluster and choose one in `completed` status:
 kubectl get backups -n <namespace>
 ```
 
+The `SchemaVersion` column shows the schema version the backup was taken at. You
+need it to choose a compatible version for the restore:
+
+```bash
+kubectl get backup my-backup -n <namespace> -o jsonpath='{.status.schemaVersion}'
+```
+
+### Version compatibility
+
+A restore brings your data back at the schema version from **when the backup was
+taken**. Set the new cluster's `documentDBVersion` to **that version or newer** —
+ideally the same version the backup was taken with. Never restore onto an **older**
+version: it runs against a newer, irreversible schema and may cause **data
+corruption**.
+
+| Restore version vs. backup schema | Behavior |
+| --- | --- |
+| equal or newer | Allowed |
+| older | **Rejected** at admission (when the schema version is known) |
+| schema version unknown | Allowed with a warning — verify the version yourself |
+
+The schema version is known from the `Backup`'s `status.schemaVersion` or a
+retained PV's `documentdb.io/schema-version` annotation. A PV imported from
+outside the operator (e.g. a raw disk or external snapshot) carries no annotation,
+so its schema version is unknown.
+
 ### Step 2: Create a New DocumentDB Cluster
 
 ```yaml title="restore.yaml"
@@ -177,6 +203,8 @@ metadata:
 spec:
   nodeCount: 1
   instancesPerNode: 1
+  # Set to the backup's schema version or newer (see Version compatibility).
+  documentDBVersion: "0.110.0"
   resource:
     storage:
       pvcSize: 10Gi
@@ -207,6 +235,7 @@ Once the status shows `Cluster in healthy state`, connect and verify your data.
 - The backup must be in `completed` status.
 - The VolumeSnapshot referenced by the backup must still exist — if it was manually deleted, the backup cannot be used for recovery.
 - You cannot specify both `backup` and `persistentVolume` in the same recovery spec.
+- The restore version must be **>= the backup's schema version** when that version is known (from the `Backup` or a PV's `documentdb.io/schema-version` annotation); otherwise the restore is allowed with a warning. See [Version compatibility](#version-compatibility).
 
 For additional recovery options (including PV-based recovery), see [Restore a Deleted DocumentDB Cluster](restore-deleted-cluster.md).
 
@@ -236,4 +265,3 @@ The operator resolves retention in priority order: per-backup > per-schedule > p
 - Failed backups still expire (timer starts at creation).
 - Deleting the DocumentDB cluster does **not** immediately delete its `Backup` objects — they wait for expiration.
 - There is no "keep forever" option. Export backups externally for permanent archival.
-

docs/operator-public-documentation/preview/operations/restore-deleted-cluster.md

@@ -48,6 +48,15 @@ The PV should be in `Released` or `Available` status.
 
 ### Step 2: Create a New DocumentDB Cluster with PV Recovery
 
+!!! warning "Schema-version compatibility"
+    Set `documentDBVersion` to the schema version of the data on the PV **or newer** —
+    never older, which risks **data corruption**. Retained PVs carry a
+    `documentdb.io/schema-version` annotation; when present, admission rejects a
+    restore onto an older version. If the annotation is missing (e.g. a PV imported
+    from outside the operator), the version can't be verified and the restore is
+    allowed with a warning — set the right version yourself. See
+    [Version compatibility](backup-and-restore.md#version-compatibility).
+
 ```yaml title="restore-from-pv.yaml"
 apiVersion: documentdb.io/preview
 kind: DocumentDB
@@ -57,6 +66,8 @@ metadata:
 spec:
   nodeCount: 1
   instancesPerNode: 1
+  # Set to the PV data's schema version or newer (see Version compatibility).
+  documentDBVersion: "0.110.0"
   documentDbCredentialSecret: documentdb-credentials
   resource:
     storage:
@@ -90,4 +101,3 @@ After confirming the recovery is successful, delete the source PV:
 ```bash
 kubectl delete pv pvc-abc123-def456-789
 ```
-

operator/documentdb-helm-chart/crds/documentdb.io_backups.yaml

@@ -41,6 +41,10 @@ spec:
       jsonPath: .status.message
       name: Message
       type: string
+    - description: DocumentDB schema version at backup time
+      jsonPath: .status.schemaVersion
+      name: SchemaVersion
+      type: string
     name: preview
     schema:
       openAPIV3Schema:
@@ -104,6 +108,14 @@ spec:
               phase:
                 description: Phase represents the current phase of the backup operation.
                 type: string
+              schemaVersion:
+                description: |-
+                  SchemaVersion is the DocumentDB extension schema version of the source
+                  cluster at backup time, captured from the source DocumentDB's
+                  status.schemaVersion. It is used to validate restore compatibility:
+                  a restore must target a binary version >= this schema version, otherwise
+                  an older binary would run against a newer, irreversible schema.
+                type: string
               startedAt:
                 description: StartedAt is the time when the backup operation started.
                 format: date-time

operator/src/api/preview/backup_funcs.go

@@ -34,8 +34,9 @@ func (backup *Backup) CreateCNPGBackup(scheme *runtime.Scheme, clusterName strin
 	return cnpgBackup, nil
 }
 
-// UpdateStatus updates the Backup status based on the CNPG Backup status and backup configuration.
-func (backup *Backup) UpdateStatus(cnpgBackup *cnpgv1.Backup, backupConfiguration *BackupConfiguration) bool {
+// UpdateStatus reconciles the Backup status from the observed inputs and returns
+// whether any status field changed.
+func (backup *Backup) UpdateStatus(cnpgBackup *cnpgv1.Backup, backupConfiguration *BackupConfiguration, sourceSchemaVersion string) bool {
 	needsUpdate := false
 	if backup.Status.Phase != cnpgBackup.Status.Phase {
 		backup.Status.Phase = cnpgBackup.Status.Phase
@@ -63,6 +64,14 @@ func (backup *Backup) UpdateStatus(cnpgBackup *cnpgv1.Backup, backupConfiguratio
 		needsUpdate = true
 	}
 
+	// Record the source cluster's schema version at backup time: captured once
+	// when first available and left immutable afterward, so a later restore can
+	// validate binary-vs-schema compatibility.
+	if backup.Status.SchemaVersion == "" && sourceSchemaVersion != "" {
+		backup.Status.SchemaVersion = sourceSchemaVersion
+		needsUpdate = true
+	}
+
 	return needsUpdate
 }
 

operator/src/api/preview/backup_funcs_test.go

@@ -83,7 +83,7 @@ var _ = Describe("Backup", func() {
 				},
 			}
 
-			needsUpdate := backup.UpdateStatus(cnpg, nil)
+			needsUpdate := backup.UpdateStatus(cnpg, nil, "")
 			Expect(needsUpdate).To(BeTrue())
 			Expect(string(backup.Status.Phase)).To(Equal(cnpgv1.BackupPhaseCompleted))
 			Expect(backup.Status.StartedAt).To(Equal(&startedAt))
@@ -119,9 +119,30 @@ var _ = Describe("Backup", func() {
 				},
 			}
 
-			needsUpdate := backup.UpdateStatus(cnpg, nil)
+			needsUpdate := backup.UpdateStatus(cnpg, nil, "")
 			Expect(needsUpdate).To(BeFalse())
 		})
+
+		It("captures the source schema version once and leaves it immutable", func() {
+			cnpg := &cnpgv1.Backup{
+				Status: cnpgv1.BackupStatus{Phase: cnpgv1.BackupPhaseCompleted},
+			}
+			backup := &Backup{Spec: BackupSpec{}}
+
+			// First observation records the source schema version.
+			Expect(backup.UpdateStatus(cnpg, nil, "0.110.0")).To(BeTrue())
+			Expect(backup.Status.SchemaVersion).To(Equal("0.110.0"))
+
+			// A later, differing source schema version must NOT overwrite it
+			// (it reflects the schema at backup time), and must not by itself
+			// mark the status dirty.
+			Expect(backup.UpdateStatus(cnpg, nil, "0.112.0")).To(BeFalse())
+			Expect(backup.Status.SchemaVersion).To(Equal("0.110.0"))
+
+			// An empty source schema version is a no-op and never clears it.
+			Expect(backup.UpdateStatus(cnpg, nil, "")).To(BeFalse())
+			Expect(backup.Status.SchemaVersion).To(Equal("0.110.0"))
+		})
 	})
 
 	Describe("CalculateExpirationTime", func() {

operator/src/api/preview/backup_types.go

@@ -48,6 +48,14 @@ type BackupStatus struct {
 	// For skipped backups, this explains why the backup was skipped.
 	// +optional
 	Message string `json:"message,omitempty"`
+
+	// SchemaVersion is the DocumentDB extension schema version of the source
+	// cluster at backup time, captured from the source DocumentDB's
+	// status.schemaVersion. It is used to validate restore compatibility:
+	// a restore must target a binary version >= this schema version, otherwise
+	// an older binary would run against a newer, irreversible schema.
+	// +optional
+	SchemaVersion string `json:"schemaVersion,omitempty"`
 }
 
 // +kubebuilder:object:root=true
@@ -59,6 +67,7 @@ type BackupStatus struct {
 // +kubebuilder:printcolumn:name="StoppedAt",type=string,JSONPath=".status.stoppedAt",description="Backup completion time"
 // +kubebuilder:printcolumn:name="ExpiredAt",type=string,JSONPath=".status.expiredAt",description="Backup expiration time"
 // +kubebuilder:printcolumn:name="Message",type=string,JSONPath=".status.message",description="Backup status message"
+// +kubebuilder:printcolumn:name="SchemaVersion",type=string,JSONPath=".status.schemaVersion",description="DocumentDB schema version at backup time"
 // +kubebuilder:metadata:labels=app=documentdb-operator
 type Backup struct {
 	metav1.TypeMeta   `json:",inline"`

operator/src/config/crd/bases/documentdb.io_backups.yaml

@@ -41,6 +41,10 @@ spec:
       jsonPath: .status.message
       name: Message
       type: string
+    - description: DocumentDB schema version at backup time
+      jsonPath: .status.schemaVersion
+      name: SchemaVersion
+      type: string
     name: preview
     schema:
       openAPIV3Schema:
@@ -104,6 +108,14 @@ spec:
               phase:
                 description: Phase represents the current phase of the backup operation.
                 type: string
+              schemaVersion:
+                description: |-
+                  SchemaVersion is the DocumentDB extension schema version of the source
+                  cluster at backup time, captured from the source DocumentDB's
+                  status.schemaVersion. It is used to validate restore compatibility:
+                  a restore must target a binary version >= this schema version, otherwise
+                  an older binary would run against a newer, irreversible schema.
+                type: string
               startedAt:
                 description: StartedAt is the time when the backup operation started.
                 format: date-time

operator/src/internal/controller/backup_controller.go

@@ -108,8 +108,7 @@ func (r *BackupReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctr
 		return ctrl.Result{}, err
 	}
 
-	// Update status based on CNPG Backup status
-	return r.updateBackupStatus(ctx, backup, cnpgBackup, cluster.Spec.Backup)
+	return r.updateBackupStatus(ctx, backup, cnpgBackup, cluster.Spec.Backup, cluster.Status.SchemaVersion)
 }
 
 // ensureVolumeSnapshotClass creates a VolumeSnapshotClass based on the cloud environment
@@ -193,10 +192,10 @@ func (r *BackupReconciler) createCNPGBackup(ctx context.Context, backup *dbprevi
 	return ctrl.Result{RequeueAfter: 5 * time.Second}, nil
 }
 
-// updateBackupStatus updates the Backup status based on CNPG Backup status
-func (r *BackupReconciler) updateBackupStatus(ctx context.Context, backup *dbpreview.Backup, cnpgBackup *cnpgv1.Backup, backupConfiguration *dbpreview.BackupConfiguration) (ctrl.Result, error) {
+// updateBackupStatus reconciles and persists the Backup's status, returning the reconcile result.
+func (r *BackupReconciler) updateBackupStatus(ctx context.Context, backup *dbpreview.Backup, cnpgBackup *cnpgv1.Backup, backupConfiguration *dbpreview.BackupConfiguration, sourceSchemaVersion string) (ctrl.Result, error) {
 	original := backup.DeepCopy()
-	needsUpdate := backup.UpdateStatus(cnpgBackup, backupConfiguration)
+	needsUpdate := backup.UpdateStatus(cnpgBackup, backupConfiguration, sourceSchemaVersion)
 
 	if needsUpdate {
 		if err := r.Status().Patch(ctx, backup, client.MergeFrom(original)); err != nil {

operator/src/internal/controller/backup_controller_test.go

@@ -193,18 +193,19 @@ var _ = Describe("Backup Controller", func() {
 				},
 			}
 
-			res, err := reconciler.updateBackupStatus(ctx, backup, cnpgBackup, nil)
+			res, err := reconciler.updateBackupStatus(ctx, backup, cnpgBackup, nil, "0.112.0")
 			Expect(err).ToNot(HaveOccurred())
 			Expect(res.RequeueAfter).NotTo(Equal(0))
 
-			// Verify status was updated with times
+			// Verify status was updated with times and the source schema version.
 			updated := &dbpreview.Backup{}
 			Expect(fakeClient.Get(ctx, client.ObjectKey{Name: backupName, Namespace: backupNamespace}, updated)).To(Succeed())
 			Expect(string(updated.Status.Phase)).To(Equal(string(cnpgv1.BackupPhaseCompleted)))
 			Expect(updated.Status.StartedAt).ToNot(BeNil())
 			Expect(updated.Status.StoppedAt).ToNot(BeNil())
 			Expect(updated.Status.StartedAt.Time.Unix()).To(Equal(cnpgBackup.Status.StartedAt.Time.Unix()))
 			Expect(updated.Status.StoppedAt.Time.Unix()).To(Equal(cnpgBackup.Status.StoppedAt.Time.Unix()))
+			Expect(updated.Status.SchemaVersion).To(Equal("0.112.0"))
 		})
 
 		It("stops reconciling (returns zero result) when CNPG Backup phase is Failed", func() {
@@ -249,7 +250,7 @@ var _ = Describe("Backup Controller", func() {
 				},
 			}
 
-			res, err := reconciler.updateBackupStatus(ctx, backup, cnpgBackup, nil)
+			res, err := reconciler.updateBackupStatus(ctx, backup, cnpgBackup, nil, "")
 			Expect(err).ToNot(HaveOccurred())
 			Expect(res.RequeueAfter).NotTo(Equal(0))
 
@@ -300,7 +301,7 @@ var _ = Describe("Backup Controller", func() {
 				},
 			}
 
-			res, err := reconciler.updateBackupStatus(ctx, backup, cnpgBackup, nil)
+			res, err := reconciler.updateBackupStatus(ctx, backup, cnpgBackup, nil, "")
 			Expect(err).ToNot(HaveOccurred())
 			// Still in progress, requeue
 			Expect(res.RequeueAfter).To(Equal(10 * time.Second))

operator/src/internal/controller/pv_controller.go

@@ -151,6 +151,25 @@ func (r *PersistentVolumeReconciler) applyDesiredPVConfiguration(ctx context.Con
 		needsUpdate = true
 	}
 
+	// Stamp the installed schema version so a future PV restore can validate
+	// binary/schema compatibility at admission time. Only set it once the cluster
+	// has reported a schema version; never clear a previously stamped value (a
+	// transient empty status must not erase the last-known-good version, which is
+	// what a retained PV would be restored from).
+	if schemaVersion := documentdb.Status.SchemaVersion; schemaVersion != "" &&
+		pv.Annotations[util.AnnotationSchemaVersion] != schemaVersion {
+		if pv.Annotations == nil {
+			pv.Annotations = make(map[string]string)
+		}
+		logger.Info("PV schema-version annotation needs update",
+			"pv", pv.Name,
+			"currentValue", pv.Annotations[util.AnnotationSchemaVersion],
+			"desiredValue", schemaVersion,
+			"documentdb", documentdb.Name)
+		pv.Annotations[util.AnnotationSchemaVersion] = schemaVersion
+		needsUpdate = true
+	}
+
 	// Check if reclaim policy needs update
 	desiredPolicy := r.getDesiredReclaimPolicy(documentdb)
 	if pv.Spec.PersistentVolumeReclaimPolicy != desiredPolicy {
@@ -402,17 +421,21 @@ func (r *PersistentVolumeReconciler) SetupWithManager(mgr ctrl.Manager) error {
 		// Apply pvPredicate only to PersistentVolume events, not globally
 		For(&corev1.PersistentVolume{}, builder.WithPredicates(pvPredicate())).
 		// Watch DocumentDB changes and trigger reconciliation of associated PVs
+		// when the reclaim policy or the installed schema version changes.
 		Watches(
 			&dbpreview.DocumentDB{},
 			handler.EnqueueRequestsFromMapFunc(r.findPVsForDocumentDB),
-			builder.WithPredicates(documentDBReclaimPolicyPredicate()),
+			builder.WithPredicates(documentDBPVRelevantPredicate()),
 		).
 		Named("pv-controller").
 		Complete(r)
 }
 
-// documentDBReclaimPolicyPredicate only triggers when the reclaim policy field changes
-func documentDBReclaimPolicyPredicate() predicate.Predicate {
+// documentDBPVRelevantPredicate triggers PV reconciliation when a DocumentDB
+// change affects PV-level state: the reclaim policy (mutates the PV spec) or the
+// installed schema version (stamped onto the PV as an annotation for restore
+// compatibility validation).
+func documentDBPVRelevantPredicate() predicate.Predicate {
 	return predicate.Funcs{
 		UpdateFunc: func(e event.UpdateEvent) bool {
 			oldDB, ok := e.ObjectOld.(*dbpreview.DocumentDB)
@@ -423,7 +446,8 @@ func documentDBReclaimPolicyPredicate() predicate.Predicate {
 			if !ok {
 				return false
 			}
-			return oldDB.Spec.Resource.Storage.PersistentVolumeReclaimPolicy != newDB.Spec.Resource.Storage.PersistentVolumeReclaimPolicy
+			return oldDB.Spec.Resource.Storage.PersistentVolumeReclaimPolicy != newDB.Spec.Resource.Storage.PersistentVolumeReclaimPolicy ||
+				oldDB.Status.SchemaVersion != newDB.Status.SchemaVersion
 		},
 		CreateFunc:  func(e event.CreateEvent) bool { return false },
 		DeleteFunc:  func(e event.DeleteEvent) bool { return false },