feat(table-catalog): add maintenance quarantine operations (#4201)

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

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

@@ -88,6 +88,7 @@ catalog extension.
 | Manifest/data/delete reachability cleanup | Supported | Reads manifest-list and manifest Avro references, reports reachable objects, and deletes only unreferenced table objects that pass the safety window. |
 | Maintenance worker run endpoint | Preview / controlled | Supports run-once execution, current-job backpressure, retry deferral, lease expiry recovery, and heartbeat updates. |
 | Maintenance scheduler guardrails | Preview / controlled | Exposes disabled, paused, ready, active-job backpressure, retry deferral, quarantine boundary, recommended actions, and recent maintenance job audit timeline state for external schedulers and operators. |
+| Maintenance quarantine operations | Preview / controlled | Lets operators inspect, release, retry, or abandon the current quarantined maintenance job without moving the table pointer. |
 | Compaction planning | Preview / controlled | Plans partition-local and sort-order-local binpack candidates for Parquet files and does not mix data files from different partition directories or sort orders in one rewrite group. |
 | Compaction commit | Preview / controlled | Can commit a safe partition-local Parquet rewrite through the catalog while preserving Iceberg data file sort order IDs in the rewritten manifest. |
 | Built-in periodic scheduler | Not claimed | Operators can trigger worker runs, but continuous in-process scheduling is not claimed. |

rustfs/src/admin/handlers/table_catalog.rs

@@ -138,6 +138,7 @@ const TABLE_CATALOG_ENDPOINTS: &[&str] = &[
     "GET /{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/scheduler",
     "POST /{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/worker/run",
     "POST /{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/jobs/{job}/heartbeat",
+    "POST /{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/jobs/{job}/quarantine",
     "GET /{warehouse}/namespaces/{namespace}/tables/{table}/catalog/export",
     "POST /{warehouse}/namespaces/{namespace}/tables/{table}/catalog/import",
     "GET /{warehouse}/namespaces/{namespace}/tables/{table}/catalog/external",
@@ -183,6 +184,7 @@ static GET_TABLE_MAINTENANCE_JOB_HANDLER: GetTableMaintenanceJobHandler = GetTab
 static GET_TABLE_MAINTENANCE_SCHEDULER_HANDLER: GetTableMaintenanceSchedulerHandler = GetTableMaintenanceSchedulerHandler {};
 static RUN_TABLE_MAINTENANCE_WORKER_HANDLER: RunTableMaintenanceWorkerHandler = RunTableMaintenanceWorkerHandler {};
 static HEARTBEAT_TABLE_MAINTENANCE_JOB_HANDLER: HeartbeatTableMaintenanceJobHandler = HeartbeatTableMaintenanceJobHandler {};
+static TABLE_MAINTENANCE_QUARANTINE_HANDLER: TableMaintenanceQuarantineHandler = TableMaintenanceQuarantineHandler {};
 static EXPORT_TABLE_CATALOG_HANDLER: ExportTableCatalogHandler = ExportTableCatalogHandler {};
 static IMPORT_TABLE_CATALOG_HANDLER: ImportTableCatalogHandler = ImportTableCatalogHandler {};
 static EXTERNAL_CATALOG_BRIDGE_HANDLER: ExternalCatalogBridgeHandler = ExternalCatalogBridgeHandler {};
@@ -959,6 +961,11 @@ fn register_table_catalog_prefix_routes(r: &mut S3Router<AdminOperation>, prefix
         format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}/tables/{{table}}/maintenance/jobs/{{job}}/heartbeat").as_str(),
         AdminOperation(&HEARTBEAT_TABLE_MAINTENANCE_JOB_HANDLER),
     )?;
+    r.insert(
+        Method::POST,
+        format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}/tables/{{table}}/maintenance/jobs/{{job}}/quarantine").as_str(),
+        AdminOperation(&TABLE_MAINTENANCE_QUARANTINE_HANDLER),
+    )?;
     r.insert(
         Method::GET,
         format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}/tables/{{table}}/catalog/export").as_str(),
@@ -5283,6 +5290,28 @@ impl Operation for HeartbeatTableMaintenanceJobHandler {
     }
 }
 
+pub struct TableMaintenanceQuarantineHandler {}
+
+#[async_trait::async_trait]
+impl Operation for TableMaintenanceQuarantineHandler {
+    async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        let warehouse = warehouse_from_params(&params)?;
+        let namespace = namespace_from_params(&params)?;
+        let table = table_name_from_params(&params)?;
+        let job = job_id_from_params(&params)?;
+        let resource = TableCatalogResource::table(&warehouse, &namespace, &table);
+        authorize_table_catalog_resource_request(&req, &resource, AdminAction::RunTableMaintenanceAction).await?;
+        ensure_table_bucket_enabled(&warehouse).await?;
+        let request = read_json_body::<crate::table_catalog::TableMaintenanceQuarantineOperationRequest>(req.input).await?;
+        let store = table_catalog_store()?;
+        let response = store
+            .apply_table_maintenance_quarantine_operation(&warehouse, &namespace.public_name(), &table, &job, request)
+            .await
+            .map_err(catalog_store_error)?;
+        build_json_response(StatusCode::OK, &response)
+    }
+}
+
 pub struct ExportTableCatalogHandler {}
 
 #[async_trait::async_trait]
@@ -5669,6 +5698,7 @@ mod tests {
             ("PutTableMaintenanceConfigHandler", "AdminAction::SetTableLifecycleAction"),
             ("GetTableMaintenanceJobHandler", "AdminAction::GetTableLifecycleAction"),
             ("GetTableMaintenanceSchedulerHandler", "AdminAction::GetTableLifecycleAction"),
+            ("TableMaintenanceQuarantineHandler", "AdminAction::RunTableMaintenanceAction"),
             ("ExportTableCatalogHandler", "AdminAction::GetTableMetadataAction"),
             ("ImportTableCatalogHandler", "AdminAction::RegisterTableAction"),
             ("ExternalCatalogBridgeHandler", "AdminAction::GetTableMetadataAction"),
@@ -5723,6 +5753,7 @@ mod tests {
             ("PutTableMaintenanceConfigHandler", "AdminAction::SetTableLifecycleAction"),
             ("GetTableMaintenanceJobHandler", "AdminAction::GetTableLifecycleAction"),
             ("GetTableMaintenanceSchedulerHandler", "AdminAction::GetTableLifecycleAction"),
+            ("TableMaintenanceQuarantineHandler", "AdminAction::RunTableMaintenanceAction"),
             ("ExportTableCatalogHandler", "AdminAction::GetTableMetadataAction"),
             ("ImportTableCatalogHandler", "AdminAction::RegisterTableAction"),
             ("ExternalCatalogBridgeHandler", "AdminAction::GetTableMetadataAction"),
@@ -5781,6 +5812,7 @@ mod tests {
             "GetTableMaintenanceSchedulerHandler",
             "RunTableMaintenanceWorkerHandler",
             "HeartbeatTableMaintenanceJobHandler",
+            "TableMaintenanceQuarantineHandler",
             "ExportTableCatalogHandler",
             "ImportTableCatalogHandler",
             "ExternalCatalogBridgeHandler",
@@ -5888,6 +5920,7 @@ mod tests {
         let _: &PutTableMaintenanceConfigHandler = &PUT_TABLE_MAINTENANCE_CONFIG_HANDLER;
         let _: &GetTableMaintenanceJobHandler = &GET_TABLE_MAINTENANCE_JOB_HANDLER;
         let _: &GetTableMaintenanceSchedulerHandler = &GET_TABLE_MAINTENANCE_SCHEDULER_HANDLER;
+        let _: &TableMaintenanceQuarantineHandler = &TABLE_MAINTENANCE_QUARANTINE_HANDLER;
         let _: &ExportTableCatalogHandler = &EXPORT_TABLE_CATALOG_HANDLER;
         let _: &ImportTableCatalogHandler = &IMPORT_TABLE_CATALOG_HANDLER;
         let _: &ExternalCatalogBridgeHandler = &EXTERNAL_CATALOG_BRIDGE_HANDLER;
@@ -5928,6 +5961,7 @@ mod tests {
         assert_operation::<GetTableMaintenanceSchedulerHandler>();
         assert_operation::<RunTableMaintenanceWorkerHandler>();
         assert_operation::<HeartbeatTableMaintenanceJobHandler>();
+        assert_operation::<TableMaintenanceQuarantineHandler>();
         assert_operation::<ExportTableCatalogHandler>();
         assert_operation::<ImportTableCatalogHandler>();
         assert_operation::<ExternalCatalogBridgeHandler>();

rustfs/src/admin/route_policy.rs

@@ -847,6 +847,12 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[
         RUN_TABLE_MAINTENANCE,
         RouteRiskLevel::High,
     ),
+    admin(
+        HttpMethod::Post,
+        "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/jobs/{job}/quarantine",
+        RUN_TABLE_MAINTENANCE,
+        RouteRiskLevel::High,
+    ),
     admin(
         HttpMethod::Get,
         "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/export",
@@ -1100,6 +1106,12 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[
         RUN_TABLE_MAINTENANCE,
         RouteRiskLevel::High,
     ),
+    admin(
+        HttpMethod::Post,
+        "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/jobs/{job}/quarantine",
+        RUN_TABLE_MAINTENANCE,
+        RouteRiskLevel::High,
+    ),
     admin(
         HttpMethod::Get,
         "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/export",
@@ -1330,7 +1342,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(), 86);
+        assert_eq!(table_specs.count(), 88);
         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);
@@ -1475,6 +1487,16 @@ mod tests {
             "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/jobs/{job}/heartbeat",
             RUN_TABLE_MAINTENANCE,
         );
+        assert_action(
+            HttpMethod::Post,
+            "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/jobs/{job}/quarantine",
+            RUN_TABLE_MAINTENANCE,
+        );
+        assert_action(
+            HttpMethod::Post,
+            "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/jobs/{job}/quarantine",
+            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(

rustfs/src/admin/route_registration_test.rs

@@ -434,6 +434,11 @@ fn expected_admin_route_matrix() -> Vec<RouteMatrixEntry> {
             "/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/jobs/{job}/heartbeat",
             "/analytics/namespaces/sales/tables/orders/maintenance/jobs/job-1/heartbeat",
         ),
+        table_route_sample(
+            Method::POST,
+            "/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/jobs/{job}/quarantine",
+            "/analytics/namespaces/sales/tables/orders/maintenance/jobs/job-1/quarantine",
+        ),
         table_route_sample(
             Method::GET,
             "/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/export",
@@ -613,6 +618,11 @@ fn expected_admin_route_matrix() -> Vec<RouteMatrixEntry> {
             "/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/jobs/{job}/heartbeat",
             "/analytics/namespaces/sales/tables/orders/maintenance/jobs/job-1/heartbeat",
         ),
+        compat_table_route_sample(
+            Method::POST,
+            "/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/jobs/{job}/quarantine",
+            "/analytics/namespaces/sales/tables/orders/maintenance/jobs/job-1/quarantine",
+        ),
         compat_table_route_sample(
             Method::GET,
             "/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/export",
@@ -864,6 +874,11 @@ fn test_register_routes_cover_representative_admin_paths() {
         Method::POST,
         &table_catalog_path("/analytics/namespaces/sales/tables/orders/maintenance/jobs/job-1/heartbeat"),
     );
+    assert_route(
+        &router,
+        Method::POST,
+        &table_catalog_path("/analytics/namespaces/sales/tables/orders/maintenance/jobs/job-1/quarantine"),
+    );
     assert_route(
         &router,
         Method::GET,
@@ -1016,6 +1031,11 @@ fn test_register_routes_cover_representative_admin_paths() {
         Method::POST,
         &compat_table_catalog_path("/analytics/namespaces/sales/tables/orders/maintenance/jobs/job-1/heartbeat"),
     );
+    assert_route(
+        &router,
+        Method::POST,
+        &compat_table_catalog_path("/analytics/namespaces/sales/tables/orders/maintenance/jobs/job-1/quarantine"),
+    );
     assert_route(
         &router,
         Method::GET,

rustfs/src/table_catalog.rs

@@ -552,6 +552,30 @@ pub(crate) struct TableMaintenanceSchedulerQuarantineBoundary {
     pub source_job_id: Option<String>,
 }
 
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
+pub(crate) enum TableMaintenanceQuarantineAction {
+    Inspect,
+    Release,
+    Retry,
+    Abandon,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub(crate) struct TableMaintenanceQuarantineOperationRequest {
+    pub action: TableMaintenanceQuarantineAction,
+    #[serde(default)]
+    pub reason: Option<String>,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub(crate) struct TableMaintenanceQuarantineOperationResult {
+    pub action: TableMaintenanceQuarantineAction,
+    pub report: TableMetadataMaintenanceReport,
+    pub scheduler: TableMaintenanceSchedulerReport,
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub(crate) struct TableMaintenanceSchedulerJobSummary {
     pub job_id: String,
@@ -3997,6 +4021,90 @@ where
         })
     }
 
+    pub(crate) async fn apply_table_maintenance_quarantine_operation(
+        &self,
+        table_bucket: &str,
+        namespace: &str,
+        table: &str,
+        job_id: &str,
+        request: TableMaintenanceQuarantineOperationRequest,
+    ) -> TableCatalogStoreResult<TableMaintenanceQuarantineOperationResult> {
+        let action = request.action.clone();
+        let namespace = parse_namespace_for_store(namespace)?;
+        let table = parse_table_for_store(table)?;
+        let table_path = self.paths.table_entry_path(table_bucket, &namespace, &table);
+        let namespace_name = namespace.public_name();
+        let table_name = table.as_str().to_string();
+
+        let report = if matches!(action, TableMaintenanceQuarantineAction::Inspect) {
+            self.get_table_metadata_maintenance_report(table_bucket, &namespace_name, &table_name, job_id)
+                .await?
+                .ok_or_else(|| {
+                    TableCatalogStoreError::NotFound(format!(
+                        "maintenance job {}/{}/{}/{}",
+                        table_bucket, namespace_name, table_name, job_id
+                    ))
+                })?
+        } else {
+            let _guard = self.backend.acquire_write_lock(self.catalog_bucket(), &table_path).await?;
+            let Some(mut report) = self
+                .get_table_metadata_maintenance_report(table_bucket, &namespace_name, &table_name, MAINTENANCE_JOB_ALIAS_CURRENT)
+                .await?
+            else {
+                return Err(TableCatalogStoreError::NotFound(format!(
+                    "maintenance job {}/{}/{}/{}",
+                    table_bucket, namespace_name, table_name, job_id
+                )));
+            };
+            if report.job.job_id != job_id {
+                return Err(TableCatalogStoreError::Conflict("maintenance job is not current".to_string()));
+            }
+            if !matches!(report.job.status, TableMetadataMaintenanceJobStatus::Failed) {
+                return Err(TableCatalogStoreError::Conflict(
+                    "maintenance quarantine operation requires a failed job".to_string(),
+                ));
+            }
+            if !report.job.quarantine_enabled || report.job.quarantined_object_count == 0 {
+                return Err(TableCatalogStoreError::Conflict(
+                    "maintenance job has no active quarantine boundary".to_string(),
+                ));
+            }
+
+            report.job.quarantined_object_count = 0;
+            match &action {
+                TableMaintenanceQuarantineAction::Inspect => unreachable!("inspect branch handled before mutation"),
+                TableMaintenanceQuarantineAction::Release => {
+                    report.job.failure_reason =
+                        Some(table_maintenance_quarantine_operator_reason("released", request.reason.as_deref()));
+                }
+                TableMaintenanceQuarantineAction::Retry => {
+                    report.job.next_retry_after = None;
+                    report.job.failure_reason = Some(table_maintenance_quarantine_operator_reason(
+                        "released for retry",
+                        request.reason.as_deref(),
+                    ));
+                }
+                TableMaintenanceQuarantineAction::Abandon => {
+                    report.job.next_retry_after = None;
+                    report.job.failure_reason =
+                        Some(table_maintenance_quarantine_operator_reason("abandoned", request.reason.as_deref()));
+                }
+            }
+            refresh_table_maintenance_report_recommended_actions(&mut report);
+            self.put_table_metadata_maintenance_report(&report).await?;
+            report
+        };
+
+        let scheduler = self
+            .get_table_maintenance_scheduler_report(table_bucket, &namespace_name, &table_name)
+            .await?;
+        Ok(TableMaintenanceQuarantineOperationResult {
+            action,
+            report,
+            scheduler,
+        })
+    }
+
     async fn list_table_metadata_maintenance_audit_reports(
         &self,
         table_bucket: &str,
@@ -6254,6 +6362,24 @@ where
         }
     }
 
+    pub(crate) async fn apply_table_maintenance_quarantine_operation(
+        &self,
+        table_bucket: &str,
+        namespace: &str,
+        table: &str,
+        job_id: &str,
+        request: TableMaintenanceQuarantineOperationRequest,
+    ) -> TableCatalogStoreResult<TableMaintenanceQuarantineOperationResult> {
+        match self {
+            Self::ObjectBacked(store) => {
+                store
+                    .apply_table_maintenance_quarantine_operation(table_bucket, namespace, table, job_id, request)
+                    .await
+            }
+            Self::DurableStrong(_) => Err(Self::unsupported_for_durable_strong("table maintenance quarantine")),
+        }
+    }
+
     pub(crate) async fn run_table_metadata_maintenance_worker_once(
         &self,
         table_bucket: &str,
@@ -8815,6 +8941,14 @@ fn parse_maintenance_timestamp(timestamp: &str) -> Option<OffsetDateTime> {
     OffsetDateTime::parse(timestamp, &time::format_description::well_known::Rfc3339).ok()
 }
 
+fn table_maintenance_quarantine_operator_reason(action: &str, reason: Option<&str>) -> String {
+    let reason = reason.map(str::trim).filter(|reason| !reason.is_empty());
+    match reason {
+        Some(reason) => format!("maintenance quarantine {action} by operator: {reason}"),
+        None => format!("maintenance quarantine {action} by operator"),
+    }
+}
+
 fn table_maintenance_recommended_actions(job: &TableMetadataMaintenanceJob) -> Vec<TableMaintenanceRecommendedAction> {
     let mut actions = Vec::new();
     match job.status {
@@ -10842,6 +10976,58 @@ mod tests {
         store.backend.reset_call_counts().await;
     }
 
+    async fn seed_quarantined_table_maintenance(
+        store: &ObjectTableCatalogStore<TestCatalogObjectBackend>,
+        backend: &TestCatalogObjectBackend,
+        bucket: &str,
+        now: OffsetDateTime,
+        next_retry_after: Option<OffsetDateTime>,
+    ) -> (Namespace, IdentifierSegment, TableMetadataMaintenanceReport) {
+        let namespace = Namespace::parse("sales").expect("namespace should parse");
+        let table = IdentifierSegment::parse("orders").expect("table should parse");
+        let current = default_table_metadata_file_path(&namespace, &table, "00002.metadata.json");
+
+        seed_table_for_metadata_maintenance(store, bucket, &namespace, &table, current.clone()).await;
+        backend
+            .seed_object(bucket, &current, br#"{"metadata-log":[]}"#.to_vec())
+            .await;
+        store
+            .put_table_maintenance_config(
+                bucket,
+                "sales",
+                "orders",
+                TableMaintenanceConfig {
+                    version: TABLE_MAINTENANCE_CONFIG_VERSION,
+                    background_enabled: true,
+                    max_retry_attempts: 2,
+                    retry_initial_backoff_seconds: 60,
+                    retry_max_backoff_seconds: 300,
+                    quarantine_enabled: true,
+                    quarantine_retention_seconds: 86_400,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("background maintenance config should persist");
+        let mut failed = store
+            .plan_table_metadata_maintenance(bucket, "sales", "orders", 0)
+            .await
+            .expect("maintenance report should be planned");
+        failed.job.status = TableMetadataMaintenanceJobStatus::Failed;
+        failed.job.failure_reason = Some("quarantine retained failed cleanup candidates".to_string());
+        failed.job.max_retry_attempts = 2;
+        failed.job.next_retry_after = next_retry_after.map(maintenance_timestamp);
+        failed.job.quarantine_enabled = true;
+        failed.job.quarantine_retention_seconds = 86_400;
+        failed.job.quarantined_object_count = 2;
+        failed.job.finished_at = Some(maintenance_timestamp(now - Duration::seconds(10)));
+        store
+            .put_table_metadata_maintenance_report(&failed)
+            .await
+            .expect("failed maintenance report should be seeded");
+        (namespace, table, failed)
+    }
+
     #[tokio::test]
     async fn object_table_catalog_store_writes_catalog_entries_to_internal_meta_bucket() {
         let backend = TestCatalogObjectBackend::default();
@@ -12525,6 +12711,194 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn maintenance_quarantine_retry_clears_boundary_and_unblocks_scheduler() {
+        let backend = TestCatalogObjectBackend::default();
+        let store = ObjectTableCatalogStore::new(backend.clone());
+        let bucket = "analytics";
+        let now = OffsetDateTime::now_utc();
+        let (_namespace, _table, failed) =
+            seed_quarantined_table_maintenance(&store, &backend, bucket, now, Some(now + Duration::seconds(300))).await;
+
+        let result = store
+            .apply_table_maintenance_quarantine_operation(
+                bucket,
+                "sales",
+                "orders",
+                &failed.job.job_id,
+                TableMaintenanceQuarantineOperationRequest {
+                    action: TableMaintenanceQuarantineAction::Retry,
+                    reason: Some("operator reviewed retained candidates".to_string()),
+                },
+            )
+            .await
+            .expect("quarantine retry should update the current maintenance job");
+
+        assert_eq!(result.action, TableMaintenanceQuarantineAction::Retry);
+        assert_eq!(result.report.job.job_id, failed.job.job_id);
+        assert_eq!(result.report.job.quarantined_object_count, 0);
+        assert!(result.report.job.next_retry_after.is_none());
+        assert!(
+            !result
+                .report
+                .job
+                .recommended_actions
+                .contains(&TableMaintenanceRecommendedAction::ReviewQuarantine)
+        );
+        assert_eq!(result.scheduler.status, TableMaintenanceSchedulerStatus::Ready);
+        assert!(!result.scheduler.quarantine.active);
+
+        let current_report = store
+            .get_table_metadata_maintenance_report(bucket, "sales", "orders", MAINTENANCE_JOB_ALIAS_CURRENT)
+            .await
+            .expect("current maintenance report should load")
+            .expect("current maintenance report should exist");
+        assert_eq!(current_report.job.job_id, failed.job.job_id);
+        assert_eq!(current_report.job.quarantined_object_count, 0);
+    }
+
+    #[tokio::test]
+    async fn maintenance_quarantine_inspect_reports_without_mutating_current_job() {
+        let backend = TestCatalogObjectBackend::default();
+        let store = ObjectTableCatalogStore::new(backend.clone());
+        let bucket = "analytics";
+        let now = OffsetDateTime::now_utc();
+        let (_namespace, _table, failed) =
+            seed_quarantined_table_maintenance(&store, &backend, bucket, now, Some(now + Duration::seconds(300))).await;
+
+        let result = store
+            .apply_table_maintenance_quarantine_operation(
+                bucket,
+                "sales",
+                "orders",
+                &failed.job.job_id,
+                TableMaintenanceQuarantineOperationRequest {
+                    action: TableMaintenanceQuarantineAction::Inspect,
+                    reason: Some("ignored for inspect".to_string()),
+                },
+            )
+            .await
+            .expect("quarantine inspect should load the maintenance job");
+
+        assert_eq!(result.action, TableMaintenanceQuarantineAction::Inspect);
+        assert_eq!(result.report.job.job_id, failed.job.job_id);
+        assert_eq!(result.report.job.quarantined_object_count, 2);
+        let expected_retry_after = maintenance_timestamp(now + Duration::seconds(300));
+        assert_eq!(result.report.job.next_retry_after.as_deref(), Some(expected_retry_after.as_str()));
+        assert_eq!(result.scheduler.status, TableMaintenanceSchedulerStatus::RetryDeferred);
+
+        let current_report = store
+            .get_table_metadata_maintenance_report(bucket, "sales", "orders", MAINTENANCE_JOB_ALIAS_CURRENT)
+            .await
+            .expect("current maintenance report should load")
+            .expect("current maintenance report should exist");
+        assert_eq!(current_report.job.quarantined_object_count, 2);
+    }
+
+    #[tokio::test]
+    async fn maintenance_quarantine_release_preserves_retry_deferral() {
+        let backend = TestCatalogObjectBackend::default();
+        let store = ObjectTableCatalogStore::new(backend.clone());
+        let bucket = "analytics";
+        let now = OffsetDateTime::now_utc();
+        let (_namespace, _table, failed) =
+            seed_quarantined_table_maintenance(&store, &backend, bucket, now, Some(now + Duration::seconds(300))).await;
+
+        let result = store
+            .apply_table_maintenance_quarantine_operation(
+                bucket,
+                "sales",
+                "orders",
+                &failed.job.job_id,
+                TableMaintenanceQuarantineOperationRequest {
+                    action: TableMaintenanceQuarantineAction::Release,
+                    reason: Some("objects retained for later retry".to_string()),
+                },
+            )
+            .await
+            .expect("quarantine release should update the current maintenance job");
+
+        assert_eq!(result.action, TableMaintenanceQuarantineAction::Release);
+        assert_eq!(result.report.job.quarantined_object_count, 0);
+        let expected_retry_after = maintenance_timestamp(now + Duration::seconds(300));
+        assert_eq!(result.report.job.next_retry_after.as_deref(), Some(expected_retry_after.as_str()));
+        assert_eq!(result.scheduler.status, TableMaintenanceSchedulerStatus::RetryDeferred);
+        assert!(result.report.job.failure_reason.as_deref().is_some_and(|reason| {
+            reason.contains("maintenance quarantine released by operator") && reason.contains("objects retained for later retry")
+        }));
+    }
+
+    #[tokio::test]
+    async fn maintenance_quarantine_abandon_clears_boundary_and_retry() {
+        let backend = TestCatalogObjectBackend::default();
+        let store = ObjectTableCatalogStore::new(backend.clone());
+        let bucket = "analytics";
+        let now = OffsetDateTime::now_utc();
+        let (_namespace, _table, failed) =
+            seed_quarantined_table_maintenance(&store, &backend, bucket, now, Some(now + Duration::seconds(300))).await;
+
+        let result = store
+            .apply_table_maintenance_quarantine_operation(
+                bucket,
+                "sales",
+                "orders",
+                &failed.job.job_id,
+                TableMaintenanceQuarantineOperationRequest {
+                    action: TableMaintenanceQuarantineAction::Abandon,
+                    reason: Some("operator accepted retained objects".to_string()),
+                },
+            )
+            .await
+            .expect("quarantine abandon should update the current maintenance job");
+
+        assert_eq!(result.action, TableMaintenanceQuarantineAction::Abandon);
+        assert_eq!(result.report.job.quarantined_object_count, 0);
+        assert!(result.report.job.next_retry_after.is_none());
+        assert_eq!(result.scheduler.status, TableMaintenanceSchedulerStatus::Ready);
+        assert!(result.report.job.failure_reason.as_deref().is_some_and(|reason| {
+            reason.contains("maintenance quarantine abandoned by operator")
+                && reason.contains("operator accepted retained objects")
+        }));
+    }
+
+    #[tokio::test]
+    async fn maintenance_quarantine_rejects_mutating_non_current_job() {
+        let backend = TestCatalogObjectBackend::default();
+        let store = ObjectTableCatalogStore::new(backend.clone());
+        let bucket = "analytics";
+        let now = OffsetDateTime::now_utc();
+        let (_namespace, _table, old_failed) =
+            seed_quarantined_table_maintenance(&store, &backend, bucket, now, Some(now + Duration::seconds(300))).await;
+        let mut current_failed = store
+            .plan_table_metadata_maintenance(bucket, "sales", "orders", 0)
+            .await
+            .expect("maintenance report should be planned");
+        current_failed.job.status = TableMetadataMaintenanceJobStatus::Failed;
+        current_failed.job.quarantine_enabled = true;
+        current_failed.job.quarantine_retention_seconds = 86_400;
+        current_failed.job.quarantined_object_count = 1;
+        store
+            .put_table_metadata_maintenance_report(&current_failed)
+            .await
+            .expect("new current maintenance report should be seeded");
+
+        let error = store
+            .apply_table_maintenance_quarantine_operation(
+                bucket,
+                "sales",
+                "orders",
+                &old_failed.job.job_id,
+                TableMaintenanceQuarantineOperationRequest {
+                    action: TableMaintenanceQuarantineAction::Release,
+                    reason: None,
+                },
+            )
+            .await
+            .expect_err("mutating a non-current quarantine job should fail");
+
+        assert_eq!(error, TableCatalogStoreError::Conflict("maintenance job is not current".to_string()));
+    }
+
     #[tokio::test]
     async fn maintenance_worker_run_defers_until_retry_after() {
         let backend = TestCatalogObjectBackend::default();

scripts/table-catalog/README.md

@@ -232,7 +232,8 @@ The smoke test also probes catalog-backed advanced Iceberg surfaces:
   drop routes with persisted view metadata and view-scoped authorization
 - metadata maintenance supports safe dry-run planning, controlled worker
   execution checks, and a scheduler status report with disabled, paused,
-  backpressure, retry, quarantine, and audit-timeline state
+  backpressure, retry, quarantine, and audit-timeline state; quarantined jobs
+  can be inspected, released, retried, or abandoned through an operator endpoint
 - catalog diagnostics exposes the table recovery and consistency state used by
   operators
 - catalog export and diagnostics expose the current catalog backing manifest,
@@ -335,7 +336,7 @@ Unsupported behavior is documented instead of hidden behind internal errors. The
 current unsupported inventory is:
 
 - credential vending: automated after table bootstrap with exact-prefix validation and a data-plane scope probe; full no-long-term-data-credential bootstrap is not claimed
-- background maintenance worker: controlled run-once, heartbeat, and scheduler status endpoints are registered; disabled/paused/backpressure/retry/quarantine/audit-timeline state is machine-readable; continuous in-process scheduling is not claimed
+- background maintenance worker: controlled run-once, heartbeat, quarantine operation, and scheduler status endpoints are registered; disabled/paused/backpressure/retry/quarantine/audit-timeline state is machine-readable; continuous in-process scheduling is not claimed
 - manifest/data reachability cleanup: metadata maintenance reads manifest-list and manifest Avro references, reports manifest/data/delete reachability, and deletes only unreferenced table objects that pass the safety window
 - snapshot expiration dry-run planning and manual catalog commit: supported through metadata maintenance reports
 - automatic maintenance scheduling: external scheduler hook supported through the worker run endpoint and scheduler status report; built-in periodic scheduling is not claimed

scripts/table-catalog/pyiceberg_smoke.py

@@ -183,7 +183,7 @@ UNSUPPORTED_INVENTORY: list[dict[str, str]] = [
         "status": "controlled-run-once-supported",
         "roadmap_area": "maintenance-worker",
         "catalog_endpoint": "GET /v1/{prefix}/namespaces/{namespace}/tables/{table}/maintenance/scheduler",
-        "expected_behavior": "background-enabled maintenance can be driven by the worker run endpoint and inspected through the scheduler status endpoint with disabled/paused state, current-job backpressure, retry deferral, quarantine boundary, audit timeline, lease expiry recovery, and heartbeat updates; built-in periodic scheduling is not claimed",
+        "expected_behavior": "background-enabled maintenance can be driven by the worker run endpoint and inspected through the scheduler status endpoint with disabled/paused state, current-job backpressure, retry deferral, quarantine boundary, audit timeline, lease expiry recovery, heartbeat updates, and operator quarantine inspect/release/retry/abandon actions; built-in periodic scheduling is not claimed",
     },
     {
         "capability": "manifest-data-reachability-cleanup",
@@ -1066,6 +1066,15 @@ def run_maintenance_probe(args: argparse.Namespace, deps: RuntimeDeps) -> None:
     if not isinstance(job, dict) or not job.get("job-id"):
         raise RuntimeError("maintenance metadata endpoint did not return a job id")
     signed_rest_request(args, deps, "GET", table_endpoint_path(args, f"/maintenance/jobs/{job['job-id']}"))
+    quarantine = signed_rest_request(
+        args,
+        deps,
+        "POST",
+        table_endpoint_path(args, f"/maintenance/jobs/{job['job-id']}/quarantine"),
+        {"action": "INSPECT"},
+    )
+    if quarantine.get("action") != "INSPECT" or not isinstance(quarantine.get("report"), dict):
+        raise RuntimeError("maintenance quarantine endpoint did not return an inspection report")
     scheduler = signed_rest_request(args, deps, "GET", scheduler_path)
     expected_scheduler_statuses = {"READY", "DISABLED", "PAUSED", "BACKPRESSURED", "RETRY_DEFERRED", "QUARANTINED"}
     if scheduler.get("status") not in expected_scheduler_statuses:

scripts/table-catalog/test_pyiceberg_smoke.py

@@ -294,6 +294,7 @@ class PyIcebergSmokeConfigTest(unittest.TestCase):
         config_path = pyiceberg_smoke.table_endpoint_path(args, "/maintenance/config")
         maintenance_path = pyiceberg_smoke.table_endpoint_path(args, "/maintenance/metadata")
         job_path = pyiceberg_smoke.table_endpoint_path(args, "/maintenance/jobs/job-1")
+        quarantine_path = pyiceberg_smoke.table_endpoint_path(args, "/maintenance/jobs/job-1/quarantine")
         scheduler_path = pyiceberg_smoke.table_endpoint_path(args, "/maintenance/scheduler")
         worker_path = pyiceberg_smoke.table_endpoint_path(args, "/maintenance/worker/run")
 
@@ -312,6 +313,8 @@ class PyIcebergSmokeConfigTest(unittest.TestCase):
                 return {"job": {"job-id": "job-1"}}
             if (method, path) == ("GET", job_path):
                 return {"job": {"job-id": "job-1", "status": "SUCCESSFUL"}}
+            if (method, path) == ("POST", quarantine_path):
+                return {"action": "INSPECT", "report": {"job": {"job-id": "job-1"}}}
             if (method, path) == ("GET", scheduler_path):
                 return {"status": "DISABLED", "audit_timeline": [{"job_id": "job-1"}]}
             if (method, path) == ("POST", worker_path):
@@ -333,6 +336,7 @@ class PyIcebergSmokeConfigTest(unittest.TestCase):
         view_path = pyiceberg_smoke.view_endpoint_path(args)
         config_path = pyiceberg_smoke.table_endpoint_path(args, "/maintenance/config")
         maintenance_path = pyiceberg_smoke.table_endpoint_path(args, "/maintenance/metadata")
+        quarantine_path = pyiceberg_smoke.table_endpoint_path(args, "/maintenance/jobs/job-1/quarantine")
         scheduler_path = pyiceberg_smoke.table_endpoint_path(args, "/maintenance/scheduler")
         worker_path = pyiceberg_smoke.table_endpoint_path(args, "/maintenance/worker/run")
         diagnostics_path = pyiceberg_smoke.table_endpoint_path(args, "/catalog/diagnostics")
@@ -373,6 +377,8 @@ class PyIcebergSmokeConfigTest(unittest.TestCase):
                 return {"job": {"job-id": "job-1"}}
             if (method, path) == ("GET", pyiceberg_smoke.table_endpoint_path(args, "/maintenance/jobs/job-1")):
                 return {"job": {"job-id": "job-1", "status": "SUCCESSFUL"}}
+            if (method, path) == ("POST", quarantine_path):
+                return {"action": "INSPECT", "report": {"job": {"job-id": "job-1"}}}
             if (method, path) == ("GET", scheduler_path):
                 return {"status": "DISABLED", "audit_timeline": [{"job_id": "job-1"}]}
             if (method, path) == ("POST", worker_path):
@@ -405,6 +411,7 @@ class PyIcebergSmokeConfigTest(unittest.TestCase):
         self.assertIn(("GET", metadata_location_path, None), calls)
         self.assertIn(("GET", diagnostics_path, None), calls)
         self.assertIn(("GET", scheduler_path, None), calls)
+        self.assertIn(("POST", quarantine_path, {"action": "INSPECT"}), calls)
         self.assertIn(("POST", worker_path, {}), calls)
 
     def test_table_ref_probe_force_deletes_smoke_ref_after_validation_failure(self) -> None: