feat(table-catalog): add durable backing migration guardrails (#4038)

Co-authored-by: Henry Guo <marshawcoco@users.noreply.github.com>
Co-authored-by: houseme <housemecn@gmail.com>

docs/architecture/s3-tables-support-matrix.md

@@ -100,10 +100,31 @@ catalog extension.
 | Post-CAS finalization recovery | Supported | Diagnostics and recovery can repair stale or missing idempotency indexes without changing the current table pointer. |
 | Catalog export | Supported | Exposes table state, commit recovery state, and backing migration information for operator inspection. |
 | Strong backing migration contract | Supported as a contract | Diagnostics publish object-backed manifest state, recoverable commit-log WAL state, target backing type, replay requirements, and blockers. |
-| Strong KV/WAL backing cutover | Not claimed | The contract and diagnostics exist, but this matrix does not claim a completed backing-store migration. |
+| Durable backing migration dry-run | Supported | `GET /iceberg/v1/{warehouse}/catalog/migration` and the `/_iceberg/v1` alias inspect object-backed catalog inventory, commit recovery blockers, idempotency indexes, warehouse prefix index readiness, recommended actions, and rollback configuration without mutating catalog state. |
+| Strong KV/WAL backing cutover | Preview / controlled | Operators can explicitly select durable strong backing with `RUSTFS_TABLE_CATALOG_BACKING=durable-strong`. Run the migration dry-run first and keep the object-backed catalog available for rollback. Object-only advanced operations fail closed in durable strong mode. |
 | Single active writer region | Supported policy | Diagnostics publish single-active-writer semantics and read-only replica limits. |
 | Active-active multi-region writes | Not claimed | A table must not accept independent concurrent writers in multiple active regions. |
 
+## Durable Backing Cutover Runbook
+
+Use the migration dry-run before changing the table catalog backing for a
+warehouse:
+
+1. Run `GET /iceberg/v1/{warehouse}/catalog/migration` with a principal that has
+   `GetTableCatalogAction` on the table bucket.
+2. Treat any `blockers` entry as fail-closed. Run catalog recovery first when
+   commit recovery or idempotency repair is required, and backfill the warehouse
+   prefix index before switching backing modes.
+3. Take an object-backed catalog snapshot or backup before setting
+   `RUSTFS_TABLE_CATALOG_BACKING=durable-strong`.
+4. Restart with durable strong backing enabled, then verify catalog config,
+   table load, commit recovery diagnostics, and table data-plane policy
+   resolution for representative tables.
+5. To roll back, restore `RUSTFS_TABLE_CATALOG_BACKING=object-backed` and restart
+   while preserving the object-backed catalog objects. Do not delete object-backed
+   catalog state until durable strong backing has passed the operator's retention
+   window.
+
 ## Production Failure Coverage
 
 Positive client smoke proves a client can use a table. Production failure probes

rustfs/src/admin/handlers/table_catalog.rs

@@ -106,6 +106,7 @@ const TABLE_CATALOG_ENDPOINTS: &[&str] = &[
     "DELETE /v1/{prefix}/namespaces/{namespace}/tables/{table}",
     "PUT /buckets/{warehouse}",
     "GET /buckets/{warehouse}",
+    "GET /{warehouse}/catalog/migration",
     "GET /{warehouse}/namespaces",
     "POST /{warehouse}/namespaces",
     "GET /{warehouse}/namespaces/{namespace}",
@@ -149,6 +150,7 @@ const TABLE_CATALOG_ENDPOINTS: &[&str] = &[
 static GET_CONFIG_HANDLER: GetCatalogConfigHandler = GetCatalogConfigHandler {};
 static ENABLE_TABLE_BUCKET_HANDLER: EnableTableBucketHandler = EnableTableBucketHandler {};
 static GET_TABLE_BUCKET_HANDLER: GetTableBucketHandler = GetTableBucketHandler {};
+static GET_TABLE_CATALOG_MIGRATION_HANDLER: GetTableCatalogMigrationHandler = GetTableCatalogMigrationHandler {};
 static LIST_NAMESPACES_HANDLER: RestListNamespacesHandler = RestListNamespacesHandler {};
 static CREATE_NAMESPACE_HANDLER: RestCreateNamespaceHandler = RestCreateNamespaceHandler {};
 static GET_NAMESPACE_HANDLER: RestGetNamespaceHandler = RestGetNamespaceHandler {};
@@ -795,6 +797,11 @@ fn register_table_catalog_prefix_routes(r: &mut S3Router<AdminOperation>, prefix
         format!("{prefix}/buckets/{{warehouse}}").as_str(),
         AdminOperation(&GET_TABLE_BUCKET_HANDLER),
     )?;
+    r.insert(
+        Method::GET,
+        format!("{prefix}/{{warehouse}}/catalog/migration").as_str(),
+        AdminOperation(&GET_TABLE_CATALOG_MIGRATION_HANDLER),
+    )?;
     r.insert(
         Method::GET,
         format!("{prefix}/{{warehouse}}/namespaces").as_str(),
@@ -4665,6 +4672,27 @@ impl Operation for GetTableBucketHandler {
     }
 }
 
+pub struct GetTableCatalogMigrationHandler {}
+
+#[async_trait::async_trait]
+impl Operation for GetTableCatalogMigrationHandler {
+    async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        let warehouse = warehouse_from_params(&params)?;
+        let resource = TableCatalogResource::warehouse(&warehouse);
+        authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableCatalogAction).await?;
+        ensure_table_bucket_enabled(&warehouse).await?;
+        let store = table_catalog_object_store()?;
+        let started = Instant::now();
+        let result = store
+            .plan_durable_strong_backing_migration(&warehouse)
+            .await
+            .map_err(catalog_store_error);
+        record_table_catalog_admin_operation_result("migration", &warehouse, "", "", started, &result);
+        let response = result?;
+        build_json_response(StatusCode::OK, &response)
+    }
+}
+
 pub struct RestListNamespacesHandler {}
 
 #[async_trait::async_trait]
@@ -5447,6 +5475,7 @@ mod tests {
         assert_eq!(response.admin_discovery.cluster_snapshot, "/rustfs/admin/v4/cluster/snapshot");
         assert_eq!(response.admin_discovery.extensions_catalog, "/rustfs/admin/v4/extensions/catalog");
         assert!(response.endpoints.contains(&"GET /v1/{prefix}/namespaces"));
+        assert!(response.endpoints.contains(&"GET /{warehouse}/catalog/migration"));
         assert!(response.endpoints.contains(&"HEAD /v1/{prefix}/namespaces/{namespace}"));
         assert!(
             response
@@ -5586,6 +5615,7 @@ mod tests {
         for (handler, action) in [
             ("EnableTableBucketHandler", "AdminAction::SetTableBucketAction"),
             ("GetTableBucketHandler", "AdminAction::GetTableBucketAction"),
+            ("GetTableCatalogMigrationHandler", "AdminAction::GetTableCatalogAction"),
             ("RestListNamespacesHandler", "AdminAction::GetTableNamespaceAction"),
             ("RestCreateNamespaceHandler", "AdminAction::SetTableNamespaceAction"),
             ("RestGetNamespaceHandler", "AdminAction::GetTableNamespaceAction"),
@@ -5645,6 +5675,12 @@ mod tests {
             "external catalog sync should branch authorization on current table existence"
         );
 
+        let migration_block = operation_block(src, "GetTableCatalogMigrationHandler");
+        assert!(
+            migration_block.contains("TableCatalogResource::warehouse(&warehouse)"),
+            "catalog migration dry-run should authorize against the warehouse resource"
+        );
+
         for (handler, action) in [
             ("RestLoadTableHandler", "AdminAction::GetTableMetadataAction"),
             ("RestTableExistsHandler", "AdminAction::GetTableAction"),
@@ -5684,6 +5720,7 @@ mod tests {
         let src = include_str!("table_catalog.rs");
 
         for handler in [
+            "GetTableCatalogMigrationHandler",
             "RestListNamespacesHandler",
             "RestCreateNamespaceHandler",
             "RestGetNamespaceHandler",
@@ -5794,6 +5831,7 @@ mod tests {
 
         let _: &EnableTableBucketHandler = &ENABLE_TABLE_BUCKET_HANDLER;
         let _: &GetTableBucketHandler = &GET_TABLE_BUCKET_HANDLER;
+        let _: &GetTableCatalogMigrationHandler = &GET_TABLE_CATALOG_MIGRATION_HANDLER;
         let _: &RestListNamespacesHandler = &LIST_NAMESPACES_HANDLER;
         let _: &RestCreateNamespaceHandler = &CREATE_NAMESPACE_HANDLER;
         let _: &RestGetNamespaceHandler = &GET_NAMESPACE_HANDLER;
@@ -5830,6 +5868,7 @@ mod tests {
 
         assert_operation::<EnableTableBucketHandler>();
         assert_operation::<GetTableBucketHandler>();
+        assert_operation::<GetTableCatalogMigrationHandler>();
         assert_operation::<RestListNamespacesHandler>();
         assert_operation::<RestCreateNamespaceHandler>();
         assert_operation::<RestGetNamespaceHandler>();

rustfs/src/admin/route_policy.rs

@@ -655,6 +655,12 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[
         GET_TABLE_BUCKET,
         RouteRiskLevel::Sensitive,
     ),
+    admin(
+        HttpMethod::Get,
+        "/iceberg/v1/{warehouse}/catalog/migration",
+        GET_TABLE_CATALOG,
+        RouteRiskLevel::Sensitive,
+    ),
     admin(
         HttpMethod::Get,
         "/iceberg/v1/{warehouse}/namespaces",
@@ -896,6 +902,12 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[
         GET_TABLE_BUCKET,
         RouteRiskLevel::Sensitive,
     ),
+    admin(
+        HttpMethod::Get,
+        "/_iceberg/v1/{warehouse}/catalog/migration",
+        GET_TABLE_CATALOG,
+        RouteRiskLevel::Sensitive,
+    ),
     admin(
         HttpMethod::Get,
         "/_iceberg/v1/{warehouse}/namespaces",
@@ -1306,7 +1318,7 @@ mod tests {
         let table_specs = ADMIN_ROUTE_POLICY_SPECS
             .iter()
             .filter(|spec| spec.path().starts_with("/iceberg/v1") || spec.path().starts_with("/_iceberg/v1"));
-        assert_eq!(table_specs.count(), 82);
+        assert_eq!(table_specs.count(), 84);
         assert_action(HttpMethod::Put, "/iceberg/v1/buckets/{warehouse}", SET_TABLE_BUCKET);
         assert_action(HttpMethod::Get, "/_iceberg/v1/buckets/{warehouse}", GET_TABLE_BUCKET);
         assert_action(HttpMethod::Get, "/iceberg/v1/{warehouse}/namespaces", GET_TABLE_NAMESPACE);
@@ -1441,6 +1453,8 @@ mod tests {
             "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/jobs/{job}/heartbeat",
             RUN_TABLE_MAINTENANCE,
         );
+        assert_action(HttpMethod::Get, "/iceberg/v1/{warehouse}/catalog/migration", GET_TABLE_CATALOG);
+        assert_action(HttpMethod::Get, "/_iceberg/v1/{warehouse}/catalog/migration", GET_TABLE_CATALOG);
         assert_action(
             HttpMethod::Post,
             "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/import",

rustfs/src/admin/route_registration_test.rs

@@ -298,6 +298,7 @@ fn expected_admin_route_matrix() -> Vec<RouteMatrixEntry> {
         table_route(Method::GET, "/config"),
         table_route_sample(Method::PUT, "/buckets/{warehouse}", "/buckets/analytics"),
         table_route_sample(Method::GET, "/buckets/{warehouse}", "/buckets/analytics"),
+        table_route_sample(Method::GET, "/{warehouse}/catalog/migration", "/analytics/catalog/migration"),
         table_route_sample(Method::GET, "/{warehouse}/namespaces", "/analytics/namespaces"),
         table_route_sample(Method::POST, "/{warehouse}/namespaces", "/analytics/namespaces"),
         table_route_sample(Method::GET, "/{warehouse}/namespaces/{namespace}", "/analytics/namespaces/sales"),
@@ -471,6 +472,7 @@ fn expected_admin_route_matrix() -> Vec<RouteMatrixEntry> {
         compat_table_route(Method::GET, "/config"),
         compat_table_route_sample(Method::PUT, "/buckets/{warehouse}", "/buckets/analytics"),
         compat_table_route_sample(Method::GET, "/buckets/{warehouse}", "/buckets/analytics"),
+        compat_table_route_sample(Method::GET, "/{warehouse}/catalog/migration", "/analytics/catalog/migration"),
         compat_table_route_sample(Method::GET, "/{warehouse}/namespaces", "/analytics/namespaces"),
         compat_table_route_sample(Method::POST, "/{warehouse}/namespaces", "/analytics/namespaces"),
         compat_table_route_sample(Method::GET, "/{warehouse}/namespaces/{namespace}", "/analytics/namespaces/sales"),

rustfs/src/table_catalog.rs

@@ -981,6 +981,24 @@ pub(crate) struct TableCatalogBackingMigrationPlan {
     pub blockers: Vec<TableCatalogBackingMigrationBlocker>,
 }
 
+#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
+pub(crate) struct TableCatalogBackingMigrationDryRunReport {
+    pub table_bucket: String,
+    pub source_kind: TableCatalogBackingKind,
+    pub target_kind: TableCatalogBackingKind,
+    pub status: TableCatalogBackingMigrationStatus,
+    pub namespace_count: usize,
+    pub table_count: usize,
+    pub view_count: usize,
+    pub commit_log_count: usize,
+    pub idempotency_index_count: usize,
+    pub warehouse_prefix_count: usize,
+    pub warehouse_index_ready: bool,
+    pub blockers: Vec<TableCatalogBackingMigrationBlocker>,
+    pub recommended_actions: Vec<TableCatalogBackingMigrationAction>,
+    pub rollback: TableCatalogBackingRollbackPlan,
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, Serialize)]
 #[serde(rename_all = "SCREAMING_SNAKE_CASE")]
 pub(crate) enum TableCatalogBackingMigrationStatus {
@@ -1004,6 +1022,29 @@ pub(crate) enum TableCatalogBackingMigrationStep {
 pub(crate) enum TableCatalogBackingMigrationBlocker {
     CommitRecoveryRequired,
     CommitManualReviewRequired,
+    WarehouseIndexBackfillRequired,
+    DuplicateWarehousePrefix,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
+pub(crate) enum TableCatalogBackingMigrationAction {
+    RunCatalogRecovery,
+    BackfillWarehouseIndex,
+    ReviewDuplicateWarehousePrefixes,
+    SnapshotObjectBackedCatalog,
+    EnableDurableStrongBacking,
+    VerifyDurableStrongSnapshot,
+    KeepObjectBackedRollbackConfig,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
+pub(crate) struct TableCatalogBackingRollbackPlan {
+    pub backing_config_key: &'static str,
+    pub current_backing_value: &'static str,
+    pub rollback_backing_value: &'static str,
+    pub preserves_object_backed_catalog: bool,
+    pub requires_operator_restart: bool,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize)]
@@ -4334,6 +4375,120 @@ where
         })
     }
 
+    pub(crate) async fn plan_durable_strong_backing_migration(
+        &self,
+        table_bucket: &str,
+    ) -> TableCatalogStoreResult<TableCatalogBackingMigrationDryRunReport> {
+        if self.get_table_bucket(table_bucket).await?.is_none() {
+            return Err(TableCatalogStoreError::NotFound(format!("table bucket {table_bucket}")));
+        }
+
+        let namespaces = self.list_namespaces(table_bucket).await?;
+        let mut table_count: usize = 0;
+        let mut view_count: usize = 0;
+        let mut commit_log_count: usize = 0;
+        let mut idempotency_index_count: usize = 0;
+        let mut recovery_required_count: usize = 0;
+        let mut manual_review_count: usize = 0;
+        let mut warehouse_prefix_owners = BTreeMap::<String, usize>::new();
+
+        for namespace in &namespaces {
+            let tables = self.list_tables(table_bucket, &namespace.namespace).await?;
+            for table in tables {
+                table_count += 1;
+                if table.state == TableCatalogEntryState::Active {
+                    let warehouse_prefix = table_warehouse_object_prefix(&table)?;
+                    warehouse_prefix_owners
+                        .entry(warehouse_prefix)
+                        .and_modify(|count| *count = count.saturating_add(1))
+                        .or_insert(1);
+                }
+
+                let recovery = self.table_commit_recovery_report_for_entry(&table, 0).await?;
+                commit_log_count = commit_log_count.saturating_add(recovery.commits.len());
+                idempotency_index_count = idempotency_index_count.saturating_add(
+                    self.backend
+                        .list_objects(
+                            self.catalog_bucket(),
+                            &self.paths.commit_idempotency_entries_prefix(table_bucket, &table.table_id),
+                        )
+                        .await?
+                        .into_iter()
+                        .filter(|object| object.ends_with(".json"))
+                        .count(),
+                );
+                recovery_required_count = recovery_required_count
+                    .saturating_add(recovery.staged_before_table_update_count)
+                    .saturating_add(recovery.finalization_required_count)
+                    .saturating_add(recovery.idempotency_repair_required_count);
+                manual_review_count = manual_review_count.saturating_add(recovery.manual_review_count);
+            }
+            view_count = view_count.saturating_add(self.list_views(table_bucket, &namespace.namespace).await?.len());
+        }
+
+        let warehouse_index_ready = self.warehouse_index_ready(table_bucket).await?;
+        let duplicate_warehouse_prefix_count = warehouse_prefix_owners.values().filter(|count| **count > 1).count();
+        let mut blockers = Vec::new();
+        let mut recommended_actions = Vec::new();
+        if recovery_required_count > 0 {
+            blockers.push(TableCatalogBackingMigrationBlocker::CommitRecoveryRequired);
+        }
+        if manual_review_count > 0 {
+            blockers.push(TableCatalogBackingMigrationBlocker::CommitManualReviewRequired);
+        }
+        if recovery_required_count > 0 || manual_review_count > 0 {
+            recommended_actions.push(TableCatalogBackingMigrationAction::RunCatalogRecovery);
+        }
+        if !warehouse_index_ready {
+            blockers.push(TableCatalogBackingMigrationBlocker::WarehouseIndexBackfillRequired);
+            recommended_actions.push(TableCatalogBackingMigrationAction::BackfillWarehouseIndex);
+        }
+        if duplicate_warehouse_prefix_count > 0 {
+            blockers.push(TableCatalogBackingMigrationBlocker::DuplicateWarehousePrefix);
+            recommended_actions.push(TableCatalogBackingMigrationAction::ReviewDuplicateWarehousePrefixes);
+        }
+
+        let status = if manual_review_count > 0 || duplicate_warehouse_prefix_count > 0 {
+            TableCatalogBackingMigrationStatus::ManualReviewRequired
+        } else if recovery_required_count > 0 || !warehouse_index_ready {
+            TableCatalogBackingMigrationStatus::RecoveryRequired
+        } else {
+            TableCatalogBackingMigrationStatus::ReadyToSnapshot
+        };
+
+        if status == TableCatalogBackingMigrationStatus::ReadyToSnapshot {
+            recommended_actions.extend([
+                TableCatalogBackingMigrationAction::SnapshotObjectBackedCatalog,
+                TableCatalogBackingMigrationAction::EnableDurableStrongBacking,
+                TableCatalogBackingMigrationAction::VerifyDurableStrongSnapshot,
+                TableCatalogBackingMigrationAction::KeepObjectBackedRollbackConfig,
+            ]);
+        }
+
+        Ok(TableCatalogBackingMigrationDryRunReport {
+            table_bucket: table_bucket.to_string(),
+            source_kind: TableCatalogBackingKind::ObjectBacked,
+            target_kind: TableCatalogBackingKind::StrongKvWal,
+            status,
+            namespace_count: namespaces.len(),
+            table_count,
+            view_count,
+            commit_log_count,
+            idempotency_index_count,
+            warehouse_prefix_count: warehouse_prefix_owners.len(),
+            warehouse_index_ready,
+            blockers,
+            recommended_actions,
+            rollback: TableCatalogBackingRollbackPlan {
+                backing_config_key: ENV_TABLE_CATALOG_BACKING,
+                current_backing_value: TABLE_CATALOG_BACKING_DURABLE_STRONG,
+                rollback_backing_value: TABLE_CATALOG_BACKING_OBJECT,
+                preserves_object_backed_catalog: true,
+                requires_operator_restart: true,
+            },
+        })
+    }
+
     pub(crate) async fn diagnose_table_catalog(
         &self,
         table_bucket: &str,
@@ -13657,6 +13812,177 @@ mod tests {
         assert!(!export.backing_manifest.ha.active_active_supported);
     }
 
+    #[tokio::test]
+    async fn durable_strong_migration_dry_run_reports_ready_catalog_inventory() {
+        let backend = TestCatalogObjectBackend::default();
+        let store = ObjectTableCatalogStore::new(backend.clone());
+        let bucket = "analytics";
+        let namespace = Namespace::parse("sales").unwrap();
+        let table = IdentifierSegment::parse("orders").unwrap();
+        let view = IdentifierSegment::parse("recent_orders").unwrap();
+        let current_metadata = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json");
+        let next_metadata = default_table_metadata_file_path(&namespace, &table, "00002.metadata.json");
+        let view_metadata = default_view_metadata_file_path(&namespace, &view, "00001.metadata.json");
+
+        seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current_metadata.clone()).await;
+        store
+            .create_view(test_view_entry(bucket, &namespace, &view, view_metadata))
+            .await
+            .unwrap();
+        backend.seed_object(bucket, &next_metadata, b"{}".to_vec()).await;
+        store
+            .commit_table(TableCommitRequest {
+                table_bucket: bucket.to_string(),
+                namespace: namespace.public_name(),
+                table: table.as_str().to_string(),
+                commit_id: "commit-1".to_string(),
+                idempotency_key: Some("client-request".to_string()),
+                operation: "append".to_string(),
+                expected_version_token: "token-v1".to_string(),
+                expected_metadata_location: current_metadata,
+                new_metadata_location: next_metadata,
+                requirements: Vec::new(),
+                writer: Some("pyiceberg/test".to_string()),
+            })
+            .await
+            .unwrap();
+
+        let report = store.plan_durable_strong_backing_migration(bucket).await.unwrap();
+
+        assert_eq!(report.table_bucket, bucket);
+        assert_eq!(report.source_kind, TableCatalogBackingKind::ObjectBacked);
+        assert_eq!(report.target_kind, TableCatalogBackingKind::StrongKvWal);
+        assert_eq!(report.status, TableCatalogBackingMigrationStatus::ReadyToSnapshot);
+        assert_eq!(report.namespace_count, 1);
+        assert_eq!(report.table_count, 1);
+        assert_eq!(report.view_count, 1);
+        assert_eq!(report.commit_log_count, 1);
+        assert_eq!(report.idempotency_index_count, 1);
+        assert_eq!(report.warehouse_prefix_count, 1);
+        assert!(report.blockers.is_empty());
+        assert!(
+            report
+                .recommended_actions
+                .contains(&TableCatalogBackingMigrationAction::SnapshotObjectBackedCatalog)
+        );
+        assert!(
+            report
+                .recommended_actions
+                .contains(&TableCatalogBackingMigrationAction::EnableDurableStrongBacking)
+        );
+        assert_eq!(report.rollback.backing_config_key, ENV_TABLE_CATALOG_BACKING);
+        assert_eq!(report.rollback.rollback_backing_value, TABLE_CATALOG_BACKING_OBJECT);
+    }
+
+    #[tokio::test]
+    async fn durable_strong_migration_dry_run_reports_recovery_blockers() {
+        let backend = TestCatalogObjectBackend::default();
+        let store = ObjectTableCatalogStore::new(backend.clone());
+        let bucket = "analytics";
+        let namespace = Namespace::parse("sales").unwrap();
+        let table = IdentifierSegment::parse("orders").unwrap();
+        let current_metadata = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json");
+        let next_metadata = default_table_metadata_file_path(&namespace, &table, "00002.metadata.json");
+
+        seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current_metadata.clone()).await;
+        backend.seed_object(bucket, &next_metadata, b"{}".to_vec()).await;
+        store
+            .commit_table(TableCommitRequest {
+                table_bucket: bucket.to_string(),
+                namespace: namespace.public_name(),
+                table: table.as_str().to_string(),
+                commit_id: "commit-1".to_string(),
+                idempotency_key: Some("client-request".to_string()),
+                operation: "append".to_string(),
+                expected_version_token: "token-v1".to_string(),
+                expected_metadata_location: current_metadata,
+                new_metadata_location: next_metadata,
+                requirements: Vec::new(),
+                writer: Some("pyiceberg/test".to_string()),
+            })
+            .await
+            .unwrap();
+        let idempotency_path = store
+            .paths
+            .commit_idempotency_entry_path(bucket, "table-id", "client-request");
+        backend.delete_object(RUSTFS_META_BUCKET, &idempotency_path).await.unwrap();
+
+        let report = store.plan_durable_strong_backing_migration(bucket).await.unwrap();
+
+        assert_eq!(report.status, TableCatalogBackingMigrationStatus::RecoveryRequired);
+        assert_eq!(report.commit_log_count, 1);
+        assert_eq!(report.idempotency_index_count, 0);
+        assert!(
+            report
+                .blockers
+                .contains(&TableCatalogBackingMigrationBlocker::CommitRecoveryRequired)
+        );
+        assert!(
+            report
+                .recommended_actions
+                .contains(&TableCatalogBackingMigrationAction::RunCatalogRecovery)
+        );
+        assert!(
+            !report
+                .recommended_actions
+                .contains(&TableCatalogBackingMigrationAction::SnapshotObjectBackedCatalog)
+        );
+        assert!(
+            !report
+                .recommended_actions
+                .contains(&TableCatalogBackingMigrationAction::EnableDurableStrongBacking)
+        );
+        assert!(
+            !report
+                .recommended_actions
+                .contains(&TableCatalogBackingMigrationAction::VerifyDurableStrongSnapshot)
+        );
+    }
+
+    #[tokio::test]
+    async fn durable_strong_migration_dry_run_requires_ready_warehouse_index() {
+        let backend = TestCatalogObjectBackend::default();
+        let store = ObjectTableCatalogStore::new(backend.clone());
+        let bucket = "analytics";
+        let namespace = Namespace::parse("sales").unwrap();
+        let table = IdentifierSegment::parse("orders").unwrap();
+        let current_metadata = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json");
+
+        seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current_metadata).await;
+        let state_path = store.paths.warehouse_index_state_path(bucket);
+        backend.delete_object(RUSTFS_META_BUCKET, &state_path).await.unwrap();
+
+        let report = store.plan_durable_strong_backing_migration(bucket).await.unwrap();
+
+        assert_eq!(report.status, TableCatalogBackingMigrationStatus::RecoveryRequired);
+        assert!(!report.warehouse_index_ready);
+        assert!(
+            report
+                .blockers
+                .contains(&TableCatalogBackingMigrationBlocker::WarehouseIndexBackfillRequired)
+        );
+        assert!(
+            report
+                .recommended_actions
+                .contains(&TableCatalogBackingMigrationAction::BackfillWarehouseIndex)
+        );
+        assert!(
+            !report
+                .recommended_actions
+                .contains(&TableCatalogBackingMigrationAction::SnapshotObjectBackedCatalog)
+        );
+        assert!(
+            !report
+                .recommended_actions
+                .contains(&TableCatalogBackingMigrationAction::EnableDurableStrongBacking)
+        );
+        assert!(
+            !report
+                .recommended_actions
+                .contains(&TableCatalogBackingMigrationAction::VerifyDurableStrongSnapshot)
+        );
+    }
+
     #[tokio::test]
     async fn strong_catalog_backing_commit_is_atomic_with_wal_and_idempotency() {
         let backend = TestCatalogObjectBackend::default();