diff --git a/docs/architecture/s3-tables-support-matrix.md b/docs/architecture/s3-tables-support-matrix.md index 6c3768d51..5b818650a 100644 --- a/docs/architecture/s3-tables-support-matrix.md +++ b/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. | diff --git a/rustfs/src/admin/handlers/table_catalog.rs b/rustfs/src/admin/handlers/table_catalog.rs index ad030c7ee..6109f7e7f 100644 --- a/rustfs/src/admin/handlers/table_catalog.rs +++ b/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, 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, params: Params<'_, '_>) -> S3Result> { + let warehouse = warehouse_from_params(¶ms)?; + let namespace = namespace_from_params(¶ms)?; + let table = table_name_from_params(¶ms)?; + let job = job_id_from_params(¶ms)?; + 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::(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::(); assert_operation::(); assert_operation::(); + assert_operation::(); assert_operation::(); assert_operation::(); assert_operation::(); diff --git a/rustfs/src/admin/route_policy.rs b/rustfs/src/admin/route_policy.rs index 4d5538e97..0bb698bbd 100644 --- a/rustfs/src/admin/route_policy.rs +++ b/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( diff --git a/rustfs/src/admin/route_registration_test.rs b/rustfs/src/admin/route_registration_test.rs index 397550e49..130adb4c8 100644 --- a/rustfs/src/admin/route_registration_test.rs +++ b/rustfs/src/admin/route_registration_test.rs @@ -434,6 +434,11 @@ fn expected_admin_route_matrix() -> Vec { "/{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 { "/{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, diff --git a/rustfs/src/table_catalog.rs b/rustfs/src/table_catalog.rs index 2d31d1200..525ae7be6 100644 --- a/rustfs/src/table_catalog.rs +++ b/rustfs/src/table_catalog.rs @@ -552,6 +552,30 @@ pub(crate) struct TableMaintenanceSchedulerQuarantineBoundary { pub source_job_id: Option, } +#[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, +} + +#[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 { + 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 { + 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::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 { 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, + backend: &TestCatalogObjectBackend, + bucket: &str, + now: OffsetDateTime, + next_retry_after: Option, + ) -> (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, ¤t, 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(¤t_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(); diff --git a/scripts/table-catalog/README.md b/scripts/table-catalog/README.md index 12b44d451..187e101e5 100644 --- a/scripts/table-catalog/README.md +++ b/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 diff --git a/scripts/table-catalog/pyiceberg_smoke.py b/scripts/table-catalog/pyiceberg_smoke.py index acbd6fd27..064a077a5 100755 --- a/scripts/table-catalog/pyiceberg_smoke.py +++ b/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: diff --git a/scripts/table-catalog/test_pyiceberg_smoke.py b/scripts/table-catalog/test_pyiceberg_smoke.py index 496ed8fa4..ca57306fe 100644 --- a/scripts/table-catalog/test_pyiceberg_smoke.py +++ b/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: