test(table-catalog): add disaster recovery rehearsal probes (#4119)

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

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

@@ -102,6 +102,7 @@ catalog extension.
 | 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. |
 | 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. |
+| Disaster recovery rehearsal | Manual/live harness | `failure_coverage.py --print-disaster-recovery-rehearsal` generates an operator runbook covering catalog export, diagnostics, safe recovery repair, rollback/import, durable backing migration dry-run, post-recovery loadTable, and table data-plane policy probes. |
 | 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. |
@@ -194,6 +195,12 @@ python3 scripts/table-catalog/failure_coverage.py \
   --namespace smoke \
   --table events \
   --print-failure-probes
+python3 scripts/table-catalog/failure_coverage.py \
+  --warehouse rustfs-s3table-smoke \
+  --namespace smoke \
+  --table events \
+  --table-warehouse-location s3://rustfs-s3table-smoke/tables/table-id \
+  --print-disaster-recovery-rehearsal
 ```
 
 ## Release Claim Guidance
@@ -205,7 +212,8 @@ Acceptable wording:
 > RustFS includes a core Iceberg REST Catalog-based S3 Tables implementation
 > with PyIceberg smoke coverage, table-aware S3 data-plane policy checks,
 > controlled maintenance, catalog recovery diagnostics, Spark manual/live
-> conformance input, and production-failure probe harnesses.
+> conformance input, production-failure probe harnesses, and disaster-recovery
+> rehearsal probes.
 
 Do not claim:
 

scripts/table-catalog/README.md

@@ -204,6 +204,12 @@ python3 scripts/table-catalog/failure_coverage.py \
   --table events \
   --rest-path /iceberg \
   --print-failure-probes
+python3 scripts/table-catalog/failure_coverage.py \
+  --warehouse rustfs-s3table-smoke \
+  --namespace smoke \
+  --table events \
+  --table-warehouse-location s3://rustfs-s3table-smoke/tables/table-id \
+  --print-disaster-recovery-rehearsal
 ```
 
 The generated probe plan covers stale-token commit conflicts, missing metadata
@@ -271,6 +277,44 @@ Do not promote a failure case from `probe-required` or `load-test-required` to
 an automated claim until the live probe or stress harness is repeatable and its
 RustFS build, client version, and expected response shape are recorded.
 
+## Disaster Recovery Rehearsal
+
+The disaster recovery rehearsal plan is a machine-readable operator runbook. It
+does not mutate state by itself and does not claim automatic repair. It records
+the REST and S3 probes an operator or CI opt-in job should run against a
+prepared table when validating recovery behavior.
+
+Generate the rehearsal plan:
+
+```bash
+python3 scripts/table-catalog/failure_coverage.py \
+  --warehouse rustfs-s3table-smoke \
+  --namespace smoke \
+  --table events \
+  --table-warehouse-location s3://rustfs-s3table-smoke/tables/table-id \
+  --print-disaster-recovery-rehearsal
+```
+
+The plan is gated for CI by:
+
+```text
+RUSTFS_TABLE_CATALOG_DR_REHEARSAL=1
+```
+
+The generated phases cover:
+
+- baseline capture through catalog export and `loadTable`
+- recovery diagnostics and safe idempotency/commit repair
+- operator-selected rollback or import of a known metadata location
+- durable backing migration dry-run blocker checks
+- post-recovery `loadTable` and table warehouse data-plane policy probes
+
+Record the RustFS build, catalog backing mode, table identifier, metadata
+location, and expected response status for each run. Treat migration blockers,
+manual-review diagnostics, stale rollback/import conflicts, and data-plane
+policy failures as fail-closed results that require operator investigation
+before cutover or release claims.
+
 ## Vendor Profile References
 
 | Profile | Catalog shape | Warehouse shape | Signing name | RustFS claim |

scripts/table-catalog/failure_coverage.py

@@ -93,6 +93,11 @@ def table_path(warehouse: str, namespace: str, table: str, suffix: str = "", res
     return f"{base}{suffix}"
 
 
+def warehouse_path(warehouse: str, suffix: str = "", rest_path: str = "/iceberg") -> str:
+    base = f"{catalog_prefix(rest_path)}/{warehouse}"
+    return f"{base}{suffix}"
+
+
 def probe_step(
     name: str,
     method: str,
@@ -196,14 +201,175 @@ def failure_probe_plan(warehouse: str, namespace: str, table: str, rest_path: st
     ]
 
 
+def rehearsal_phase(name: str, objective: str, steps: list[dict[str, Any]]) -> dict[str, Any]:
+    return {
+        "name": name,
+        "objective": objective,
+        "steps": steps,
+    }
+
+
+def disaster_recovery_rehearsal_plan(
+    *,
+    warehouse: str,
+    namespace: str,
+    table: str,
+    rest_path: str = "/iceberg",
+    table_warehouse_location: str | None = None,
+) -> dict[str, Any]:
+    table_endpoint = table_path(warehouse, namespace, table, rest_path=rest_path)
+    table_warehouse_location = table_warehouse_location or f"s3://{warehouse}/tables/table-id"
+    return {
+        "mode": "manual-or-ci-optional",
+        "ci_gate": "RUSTFS_TABLE_CATALOG_DR_REHEARSAL=1",
+        "preconditions": [
+            "record the RustFS build and catalog backing mode before starting",
+            "run against a disposable table or a backed-up table warehouse",
+            "capture the current metadata location and version token from loadTable",
+            "keep object-backed catalog state available until durable backing cutover is accepted",
+        ],
+        "expected_invariants": [
+            "current metadata pointer remains recoverable or deliberately rolled back",
+            "recovery repair does not move the table pointer",
+            "rollback/import actions require explicit operator-selected metadata locations",
+            "durable backing cutover remains blocked while migration blockers are present",
+            "post-recovery loadTable and table data-plane policy checks still succeed",
+        ],
+        "phases": [
+            rehearsal_phase(
+                "capture-baseline",
+                "Capture table and catalog state before injecting or repairing a failure.",
+                [
+                    probe_step(
+                        "export-catalog-state",
+                        "GET",
+                        table_path(warehouse, namespace, table, "/catalog/export", rest_path),
+                        "200",
+                        "export includes table entry, current metadata location, commit recovery state, and backing manifest",
+                    ),
+                    probe_step(
+                        "load-table-before-recovery",
+                        "GET",
+                        table_endpoint,
+                        "200",
+                        "baseline loadTable returns the current metadata location and version token",
+                    ),
+                ],
+            ),
+            rehearsal_phase(
+                "diagnose-and-repair",
+                "Inspect recovery state and run only safe repair actions.",
+                [
+                    probe_step(
+                        "read-recovery-diagnostics",
+                        "GET",
+                        table_path(warehouse, namespace, table, "/catalog/diagnostics", rest_path),
+                        "200",
+                        "diagnostics expose commit recovery state, idempotency index state, recommended actions, and manual-review blockers",
+                    ),
+                    probe_step(
+                        "safe-recovery-repair",
+                        "POST",
+                        table_path(warehouse, namespace, table, "/catalog/recovery", rest_path),
+                        "200",
+                        "safe repair can finalize recoverable commit records or repair idempotency indexes without pointer movement",
+                        {
+                            "mode": "safe-repair",
+                        },
+                    ),
+                    probe_step(
+                        "diagnostics-after-repair",
+                        "GET",
+                        table_path(warehouse, namespace, table, "/catalog/diagnostics", rest_path),
+                        "200",
+                        "recovery state is clean or still reports manual-review blockers without advancing table state",
+                    ),
+                ],
+            ),
+            rehearsal_phase(
+                "rollback-or-import",
+                "Exercise explicit operator-selected rollback/import paths.",
+                [
+                    probe_step(
+                        "rollback-to-known-metadata",
+                        "POST",
+                        table_path(warehouse, namespace, table, "/catalog/rollback", rest_path),
+                        "200-or-409",
+                        "rollback commits only a validated operator-selected metadata location, or conflicts without pointer movement",
+                        {
+                            "metadata-location": "metadata-location-from-catalog-export-or-backup",
+                            "version-token": "current-version-token-from-load-table",
+                        },
+                    ),
+                    probe_step(
+                        "import-known-metadata",
+                        "POST",
+                        table_path(warehouse, namespace, table, "/catalog/import", rest_path),
+                        "200-or-409",
+                        "import/register validates metadata identity and conflicts without pointer/token/generation advancement when stale",
+                        {
+                            "metadata-location": "metadata-location-from-catalog-export-or-backup",
+                            "properties": {
+                                "recovery-source": "catalog-export-or-backup",
+                            },
+                        },
+                    ),
+                ],
+            ),
+            rehearsal_phase(
+                "migration-preflight",
+                "Verify durable backing cutover remains explainable and fail-closed.",
+                [
+                    probe_step(
+                        "durable-backing-migration-dry-run",
+                        "GET",
+                        warehouse_path(warehouse, "/catalog/migration", rest_path),
+                        "200",
+                        "migration blockers must be empty before cutover",
+                    ),
+                ],
+            ),
+            rehearsal_phase(
+                "post-recovery-validation",
+                "Check the recovered table can still be loaded and data-plane policy still resolves to the table.",
+                [
+                    probe_step(
+                        "load-table-after-recovery",
+                        "GET",
+                        table_endpoint,
+                        "200",
+                        "loadTable returns the intended current metadata location after repair, rollback, or import",
+                    ),
+                    probe_step(
+                        "table-data-plane-policy-probe",
+                        "S3-PROBE",
+                        table_warehouse_location,
+                        "inside-allowed-outside-denied",
+                        "ordinary S3 object access still maps table warehouse objects to the table policy boundary",
+                    ),
+                    probe_step(
+                        "diagnostics-after-recovery",
+                        "GET",
+                        table_path(warehouse, namespace, table, "/catalog/diagnostics", rest_path),
+                        "200",
+                        "diagnostics no longer report unexpected recovery blockers after the rehearsal",
+                    ),
+                ],
+            ),
+        ],
+    }
+
+
 def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
     parser = argparse.ArgumentParser(description="Print RustFS S3 Tables production failure coverage helpers.")
     parser.add_argument("--warehouse", default="rustfs-s3table-smoke")
     parser.add_argument("--namespace", default="smoke")
     parser.add_argument("--table", default="events")
     parser.add_argument("--rest-path", default="/iceberg")
+    parser.add_argument("--table-warehouse-location")
     parser.add_argument("--print-failure-matrix", action="store_true")
     parser.add_argument("--print-failure-probes", action="store_true")
+    parser.add_argument("--print-disaster-recovery-rehearsal", action="store_true")
     return parser.parse_args(argv)
 
 
@@ -239,6 +405,20 @@ def run(args: argparse.Namespace, output: StringIO | None = None) -> None:
             output,
         )
         printed = True
+    if args.print_disaster_recovery_rehearsal:
+        print_json(
+            {
+                "disaster_recovery_rehearsal": disaster_recovery_rehearsal_plan(
+                    warehouse=args.warehouse,
+                    namespace=args.namespace,
+                    table=args.table,
+                    rest_path=args.rest_path,
+                    table_warehouse_location=args.table_warehouse_location,
+                )
+            },
+            output,
+        )
+        printed = True
     if not printed:
         print_json({"production_failure_coverage": production_failure_matrix()}, output)
 

scripts/table-catalog/test_failure_coverage.py

@@ -85,6 +85,80 @@ class FailureCoverageTest(unittest.TestCase):
             "/_iceberg/v1/lake/namespaces/sales/tables/orders",
         )
 
+    def test_disaster_recovery_rehearsal_plan_covers_operator_recovery_path(self) -> None:
+        rehearsal = failure_coverage.disaster_recovery_rehearsal_plan(
+            warehouse="lake",
+            namespace="sales",
+            table="orders",
+            rest_path="/iceberg",
+            table_warehouse_location="s3://lake/tables/orders",
+        )
+
+        self.assertEqual(rehearsal["mode"], "manual-or-ci-optional")
+        self.assertEqual(rehearsal["ci_gate"], "RUSTFS_TABLE_CATALOG_DR_REHEARSAL=1")
+        self.assertIn("record the RustFS build and catalog backing mode before starting", rehearsal["preconditions"])
+        self.assertIn("current metadata pointer remains recoverable or deliberately rolled back", rehearsal["expected_invariants"])
+
+        phases = {phase["name"]: phase for phase in rehearsal["phases"]}
+        self.assertIn("capture-baseline", phases)
+        self.assertIn("diagnose-and-repair", phases)
+        self.assertIn("rollback-or-import", phases)
+        self.assertIn("migration-preflight", phases)
+        self.assertIn("post-recovery-validation", phases)
+
+        baseline_steps = {step["name"]: step for step in phases["capture-baseline"]["steps"]}
+        self.assertEqual(
+            baseline_steps["export-catalog-state"]["path"],
+            "/iceberg/v1/lake/namespaces/sales/tables/orders/catalog/export",
+        )
+        self.assertEqual(baseline_steps["load-table-before-recovery"]["method"], "GET")
+
+        repair_steps = {step["name"]: step for step in phases["diagnose-and-repair"]["steps"]}
+        self.assertEqual(repair_steps["read-recovery-diagnostics"]["path"], "/iceberg/v1/lake/namespaces/sales/tables/orders/catalog/diagnostics")
+        self.assertEqual(repair_steps["safe-recovery-repair"]["method"], "POST")
+        self.assertEqual(repair_steps["safe-recovery-repair"]["body"], {"mode": "safe-repair"})
+
+        rollback_steps = {step["name"]: step for step in phases["rollback-or-import"]["steps"]}
+        self.assertEqual(rollback_steps["rollback-to-known-metadata"]["path"], "/iceberg/v1/lake/namespaces/sales/tables/orders/catalog/rollback")
+        self.assertIn("metadata-location", rollback_steps["rollback-to-known-metadata"]["body"])
+        self.assertIn("version-token", rollback_steps["rollback-to-known-metadata"]["body"])
+        self.assertNotIn("expected-version-token", rollback_steps["rollback-to-known-metadata"]["body"])
+        self.assertEqual(rollback_steps["import-known-metadata"]["path"], "/iceberg/v1/lake/namespaces/sales/tables/orders/catalog/import")
+        self.assertIn("metadata-location", rollback_steps["import-known-metadata"]["body"])
+        self.assertIn("properties", rollback_steps["import-known-metadata"]["body"])
+        self.assertNotIn("external-version-token", rollback_steps["import-known-metadata"]["body"])
+
+        migration_steps = {step["name"]: step for step in phases["migration-preflight"]["steps"]}
+        self.assertEqual(migration_steps["durable-backing-migration-dry-run"]["path"], "/iceberg/v1/lake/catalog/migration")
+        self.assertEqual(migration_steps["durable-backing-migration-dry-run"]["expected_behavior"], "migration blockers must be empty before cutover")
+
+        validation_steps = {step["name"]: step for step in phases["post-recovery-validation"]["steps"]}
+        self.assertEqual(validation_steps["load-table-after-recovery"]["path"], "/iceberg/v1/lake/namespaces/sales/tables/orders")
+        self.assertEqual(validation_steps["table-data-plane-policy-probe"]["path"], "s3://lake/tables/orders")
+        self.assertEqual(validation_steps["table-data-plane-policy-probe"]["method"], "S3-PROBE")
+
+    def test_cli_prints_disaster_recovery_rehearsal_plan(self) -> None:
+        payload = failure_coverage.cli_json(
+            [
+                "--warehouse",
+                "lake",
+                "--namespace",
+                "sales",
+                "--table",
+                "orders",
+                "--rest-path",
+                "/_iceberg",
+                "--table-warehouse-location",
+                "s3://lake/tables/orders",
+                "--print-disaster-recovery-rehearsal",
+            ]
+        )
+        document = json.loads(payload)
+
+        self.assertEqual(document["disaster_recovery_rehearsal"]["phases"][0]["name"], "capture-baseline")
+        first_step = document["disaster_recovery_rehearsal"]["phases"][0]["steps"][0]
+        self.assertEqual(first_step["path"], "/_iceberg/v1/lake/namespaces/sales/tables/orders/catalog/export")
+
 
 if __name__ == "__main__":
     unittest.main()