diff --git a/docs/architecture/s3-tables-support-matrix.md b/docs/architecture/s3-tables-support-matrix.md index df8d8f815..1abc61e74 100644 --- a/docs/architecture/s3-tables-support-matrix.md +++ b/docs/architecture/s3-tables-support-matrix.md @@ -86,14 +86,15 @@ catalog extension. | Snapshot expiration planning | Supported | Produces expiration plans with retained and candidate snapshots. | | Snapshot expiration commit | Preview / controlled | Can manually commit safe snapshot expiration through the catalog. Stale plans fail closed. | | 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 scheduler run endpoint | Preview / controlled | Lets an external scheduler durably queue one maintenance job per table, reuse an active queued job, and recover expired queued leases before requeuing. | +| Maintenance worker run endpoint | Preview / controlled | Supports queued-job claim, run-once execution, current-job backpressure, retry deferral, lease expiry recovery, and heartbeat updates. | +| Maintenance scheduler guardrails | Preview / controlled | Exposes disabled, paused, ready, queued-job handoff, active-job backpressure, retry deferral, quarantine boundary, recommended actions, and recent maintenance job audit timeline state for external schedulers and operators. | | Maintenance audit events | Preview / controlled | Job reports and scheduler job summaries include structured audit events for planning, worker transitions, heartbeats, lease expiry recovery, and mutating quarantine operations. | | 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. | | Delete-file or row-level compaction planning | Preview / controlled | Manifests with position or equality delete files produce machine-readable row-level planning and force the compaction report into manual review before any rewrite can run. | | 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. | +| Built-in periodic scheduler | Not claimed | Operators can trigger scheduler and worker ticks, but continuous in-process scheduling is not claimed. | | Delete-file or row-level compaction execution | Not claimed | RustFS does not rewrite delete files or execute row-level compaction; those cases remain manual-review maintenance items. | ## Recovery And Strong Backing Matrix @@ -162,7 +163,7 @@ RustFS does not currently claim: - full MinIO AIStor Tables private extension parity - full Cloudflare R2 Data Catalog interoperability - full Alibaba OSS Tables interoperability -- built-in periodic maintenance scheduling; external schedulers can inspect scheduler guardrails, but RustFS does not claim a continuous in-process scheduler +- built-in periodic maintenance scheduling; external schedulers can queue maintenance jobs and workers can claim them, but RustFS does not claim a continuous in-process scheduler - active-active multi-region table writes - multi-table transactions - no-long-term-data-credential table bootstrap diff --git a/rustfs/src/admin/handlers/table_catalog.rs b/rustfs/src/admin/handlers/table_catalog.rs index 6109f7e7f..586ad12b0 100644 --- a/rustfs/src/admin/handlers/table_catalog.rs +++ b/rustfs/src/admin/handlers/table_catalog.rs @@ -74,6 +74,7 @@ const TABLE_CATALOG_NAMESPACE_RESOURCE_ROOT: &str = "namespaces"; const TABLE_CATALOG_TABLE_RESOURCE_ROOT: &str = "tables"; const TABLE_CATALOG_VIEW_RESOURCE_ROOT: &str = "views"; const TABLE_CATALOG_ADMIN_OPERATION_SLOW_LOG_THRESHOLD: StdDuration = StdDuration::from_secs(2); +const DEFAULT_TABLE_MAINTENANCE_SCHEDULER_ID: &str = "rustfs-maintenance-scheduler"; const DEFAULT_TABLE_MAINTENANCE_WORKER_ID: &str = "rustfs-maintenance-worker"; const EXTERNAL_CATALOG_BRIDGE_STATUS_UNCONFIGURED: &str = "bridge-unconfigured"; const EXTERNAL_CATALOG_BRIDGE_STATUS_CONFIGURED: &str = "bridge-configured"; @@ -136,6 +137,7 @@ const TABLE_CATALOG_ENDPOINTS: &[&str] = &[ "PUT /{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/config", "GET /{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/jobs/{job}", "GET /{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/scheduler", + "POST /{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/scheduler/run", "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", @@ -182,6 +184,7 @@ static GET_TABLE_MAINTENANCE_CONFIG_HANDLER: GetTableMaintenanceConfigHandler = static PUT_TABLE_MAINTENANCE_CONFIG_HANDLER: PutTableMaintenanceConfigHandler = PutTableMaintenanceConfigHandler {}; static GET_TABLE_MAINTENANCE_JOB_HANDLER: GetTableMaintenanceJobHandler = GetTableMaintenanceJobHandler {}; static GET_TABLE_MAINTENANCE_SCHEDULER_HANDLER: GetTableMaintenanceSchedulerHandler = GetTableMaintenanceSchedulerHandler {}; +static RUN_TABLE_MAINTENANCE_SCHEDULER_HANDLER: RunTableMaintenanceSchedulerHandler = RunTableMaintenanceSchedulerHandler {}; static RUN_TABLE_MAINTENANCE_WORKER_HANDLER: RunTableMaintenanceWorkerHandler = RunTableMaintenanceWorkerHandler {}; static HEARTBEAT_TABLE_MAINTENANCE_JOB_HANDLER: HeartbeatTableMaintenanceJobHandler = HeartbeatTableMaintenanceJobHandler {}; static TABLE_MAINTENANCE_QUARANTINE_HANDLER: TableMaintenanceQuarantineHandler = TableMaintenanceQuarantineHandler {}; @@ -357,6 +360,19 @@ struct TableMetadataMaintenanceRequest { commit_compaction: bool, } +#[derive(Debug, Default, Deserialize)] +#[serde(deny_unknown_fields)] +struct TableMaintenanceSchedulerRunRequest { + #[serde(default, rename = "scheduler-id")] + scheduler_id: Option, +} + +impl TableMaintenanceSchedulerRunRequest { + fn scheduler_id(&self) -> &str { + self.scheduler_id.as_deref().unwrap_or(DEFAULT_TABLE_MAINTENANCE_SCHEDULER_ID) + } +} + #[derive(Debug, Deserialize)] #[serde(deny_unknown_fields)] struct TableMaintenanceWorkerRunRequest { @@ -951,6 +967,11 @@ fn register_table_catalog_prefix_routes(r: &mut S3Router, prefix format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}/tables/{{table}}/maintenance/scheduler").as_str(), AdminOperation(&GET_TABLE_MAINTENANCE_SCHEDULER_HANDLER), )?; + r.insert( + Method::POST, + format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}/tables/{{table}}/maintenance/scheduler/run").as_str(), + AdminOperation(&RUN_TABLE_MAINTENANCE_SCHEDULER_HANDLER), + )?; r.insert( Method::POST, format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}/tables/{{table}}/maintenance/worker/run").as_str(), @@ -5237,6 +5258,32 @@ impl Operation for GetTableMaintenanceSchedulerHandler { pub struct RunTableMaintenanceWorkerHandler {} +#[async_trait::async_trait] +impl Operation for RunTableMaintenanceSchedulerHandler { + 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 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_or_default::(req.input).await?; + let store = table_catalog_store()?; + let response = store + .run_table_maintenance_scheduler_once( + &warehouse, + &namespace.public_name(), + &table, + request.scheduler_id().to_string(), + ) + .await + .map_err(catalog_store_error)?; + build_json_response(StatusCode::OK, &response) + } +} + +pub struct RunTableMaintenanceSchedulerHandler {} + #[async_trait::async_trait] impl Operation for RunTableMaintenanceWorkerHandler { async fn call(&self, req: S3Request, params: Params<'_, '_>) -> S3Result> { @@ -5698,6 +5745,7 @@ mod tests { ("PutTableMaintenanceConfigHandler", "AdminAction::SetTableLifecycleAction"), ("GetTableMaintenanceJobHandler", "AdminAction::GetTableLifecycleAction"), ("GetTableMaintenanceSchedulerHandler", "AdminAction::GetTableLifecycleAction"), + ("RunTableMaintenanceSchedulerHandler", "AdminAction::RunTableMaintenanceAction"), ("TableMaintenanceQuarantineHandler", "AdminAction::RunTableMaintenanceAction"), ("ExportTableCatalogHandler", "AdminAction::GetTableMetadataAction"), ("ImportTableCatalogHandler", "AdminAction::RegisterTableAction"), @@ -5753,6 +5801,7 @@ mod tests { ("PutTableMaintenanceConfigHandler", "AdminAction::SetTableLifecycleAction"), ("GetTableMaintenanceJobHandler", "AdminAction::GetTableLifecycleAction"), ("GetTableMaintenanceSchedulerHandler", "AdminAction::GetTableLifecycleAction"), + ("RunTableMaintenanceSchedulerHandler", "AdminAction::RunTableMaintenanceAction"), ("TableMaintenanceQuarantineHandler", "AdminAction::RunTableMaintenanceAction"), ("ExportTableCatalogHandler", "AdminAction::GetTableMetadataAction"), ("ImportTableCatalogHandler", "AdminAction::RegisterTableAction"), @@ -5810,6 +5859,7 @@ mod tests { "PutTableMaintenanceConfigHandler", "GetTableMaintenanceJobHandler", "GetTableMaintenanceSchedulerHandler", + "RunTableMaintenanceSchedulerHandler", "RunTableMaintenanceWorkerHandler", "HeartbeatTableMaintenanceJobHandler", "TableMaintenanceQuarantineHandler", @@ -5920,6 +5970,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 _: &RunTableMaintenanceSchedulerHandler = &RUN_TABLE_MAINTENANCE_SCHEDULER_HANDLER; let _: &TableMaintenanceQuarantineHandler = &TABLE_MAINTENANCE_QUARANTINE_HANDLER; let _: &ExportTableCatalogHandler = &EXPORT_TABLE_CATALOG_HANDLER; let _: &ImportTableCatalogHandler = &IMPORT_TABLE_CATALOG_HANDLER; @@ -5959,6 +6010,7 @@ mod tests { assert_operation::(); assert_operation::(); assert_operation::(); + assert_operation::(); assert_operation::(); assert_operation::(); assert_operation::(); @@ -6033,6 +6085,33 @@ mod tests { assert_eq!(compaction.max_rewrite_bytes_per_job, 10_737_418_240); } + #[test] + fn table_maintenance_scheduler_run_request_uses_stable_default_scheduler_id() { + let request: TableMaintenanceSchedulerRunRequest = + serde_json::from_value(serde_json::json!({})).expect("scheduler run request should parse"); + + assert_eq!(request.scheduler_id(), "rustfs-maintenance-scheduler"); + } + + #[tokio::test] + async fn table_maintenance_scheduler_run_request_empty_body_uses_default_scheduler_id() { + let request: TableMaintenanceSchedulerRunRequest = read_json_body_or_default(Body::empty()) + .await + .expect("bodyless scheduler run should use the default scheduler id"); + + assert_eq!(request.scheduler_id(), "rustfs-maintenance-scheduler"); + } + + #[test] + fn table_maintenance_scheduler_run_request_accepts_scheduler_id() { + let request: TableMaintenanceSchedulerRunRequest = serde_json::from_value(serde_json::json!({ + "scheduler-id": "scheduler-a" + })) + .expect("scheduler run request should parse scheduler id"); + + assert_eq!(request.scheduler_id(), "scheduler-a"); + } + #[test] fn table_maintenance_worker_run_request_uses_stable_default_worker_id() { let request: TableMaintenanceWorkerRunRequest = diff --git a/rustfs/src/admin/route_policy.rs b/rustfs/src/admin/route_policy.rs index 0bb698bbd..f9f27dc6a 100644 --- a/rustfs/src/admin/route_policy.rs +++ b/rustfs/src/admin/route_policy.rs @@ -835,6 +835,12 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[ GET_TABLE_LIFECYCLE, RouteRiskLevel::Sensitive, ), + admin( + HttpMethod::Post, + "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/scheduler/run", + RUN_TABLE_MAINTENANCE, + RouteRiskLevel::High, + ), admin( HttpMethod::Post, "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/worker/run", @@ -1094,6 +1100,12 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[ GET_TABLE_LIFECYCLE, RouteRiskLevel::Sensitive, ), + admin( + HttpMethod::Post, + "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/scheduler/run", + RUN_TABLE_MAINTENANCE, + RouteRiskLevel::High, + ), admin( HttpMethod::Post, "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/worker/run", @@ -1342,7 +1354,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(), 88); + assert_eq!(table_specs.count(), 90); 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); @@ -1477,6 +1489,16 @@ mod tests { "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/scheduler", GET_TABLE_LIFECYCLE, ); + assert_action( + HttpMethod::Post, + "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/scheduler/run", + RUN_TABLE_MAINTENANCE, + ); + assert_action( + HttpMethod::Post, + "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/scheduler/run", + RUN_TABLE_MAINTENANCE, + ); assert_action( HttpMethod::Post, "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/worker/run", diff --git a/rustfs/src/admin/route_registration_test.rs b/rustfs/src/admin/route_registration_test.rs index 130adb4c8..438a70e9a 100644 --- a/rustfs/src/admin/route_registration_test.rs +++ b/rustfs/src/admin/route_registration_test.rs @@ -424,6 +424,11 @@ fn expected_admin_route_matrix() -> Vec { "/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/scheduler", "/analytics/namespaces/sales/tables/orders/maintenance/scheduler", ), + table_route_sample( + Method::POST, + "/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/scheduler/run", + "/analytics/namespaces/sales/tables/orders/maintenance/scheduler/run", + ), table_route_sample( Method::POST, "/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/worker/run", @@ -608,6 +613,11 @@ fn expected_admin_route_matrix() -> Vec { "/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/scheduler", "/analytics/namespaces/sales/tables/orders/maintenance/scheduler", ), + compat_table_route_sample( + Method::POST, + "/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/scheduler/run", + "/analytics/namespaces/sales/tables/orders/maintenance/scheduler/run", + ), compat_table_route_sample( Method::POST, "/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/worker/run", @@ -864,6 +874,11 @@ fn test_register_routes_cover_representative_admin_paths() { Method::GET, &table_catalog_path("/analytics/namespaces/sales/tables/orders/maintenance/scheduler"), ); + assert_route( + &router, + Method::POST, + &table_catalog_path("/analytics/namespaces/sales/tables/orders/maintenance/scheduler/run"), + ); assert_route( &router, Method::POST, @@ -1021,6 +1036,11 @@ fn test_register_routes_cover_representative_admin_paths() { Method::GET, &compat_table_catalog_path("/analytics/namespaces/sales/tables/orders/maintenance/scheduler"), ); + assert_route( + &router, + Method::POST, + &compat_table_catalog_path("/analytics/namespaces/sales/tables/orders/maintenance/scheduler/run"), + ); assert_route( &router, Method::POST, diff --git a/rustfs/src/table_catalog.rs b/rustfs/src/table_catalog.rs index bdd96b486..203a8e528 100644 --- a/rustfs/src/table_catalog.rs +++ b/rustfs/src/table_catalog.rs @@ -515,6 +515,7 @@ pub(crate) struct TableMaintenanceEffectiveConfig { #[serde(rename_all = "SCREAMING_SNAKE_CASE")] pub(crate) enum TableMaintenanceSchedulerStatus { Ready, + Queued, Disabled, Paused, Backpressured, @@ -543,6 +544,12 @@ pub(crate) struct TableMaintenanceSchedulerReport { pub audit_timeline: Vec, } +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) struct TableMaintenanceSchedulerRunResult { + pub report: TableMetadataMaintenanceReport, + pub scheduler: TableMaintenanceSchedulerReport, +} + #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub(crate) struct TableMaintenanceSchedulerQuarantineBoundary { pub enabled: bool, @@ -565,6 +572,9 @@ pub(crate) enum TableMaintenanceAuditActor { pub(crate) enum TableMaintenanceAuditAction { Planned, WorkerControl, + SchedulerControl, + SchedulerQueued, + SchedulerLeaseExpired, WorkerStarted, WorkerHeartbeat, WorkerLeaseExpired, @@ -623,6 +633,10 @@ pub(crate) struct TableMaintenanceSchedulerJobSummary { pub job_id: String, pub operation: TableMetadataMaintenanceOperation, pub status: TableMetadataMaintenanceJobStatus, + #[serde(default, rename = "scheduler-id")] + pub scheduler_id: Option, + #[serde(default, rename = "scheduled-at")] + pub scheduled_at: Option, pub worker_id: Option, pub attempt: u16, pub started_at: Option, @@ -651,6 +665,12 @@ pub(crate) struct TableMetadataMaintenanceJob { pub recommended_actions: Vec, #[serde(default)] pub config_source: TableMaintenanceConfigSource, + #[serde(default, rename = "scheduler-id")] + pub scheduler_id: Option, + #[serde(default, rename = "scheduler-lease-id")] + pub scheduler_lease_id: String, + #[serde(default, rename = "scheduled-at")] + pub scheduled_at: Option, #[serde(default)] pub worker_id: Option, #[serde(default)] @@ -973,6 +993,7 @@ pub(crate) enum TableMetadataMaintenanceOperation { #[serde(rename_all = "SCREAMING_SNAKE_CASE")] pub(crate) enum TableMetadataMaintenanceJobStatus { NotYetRun, + Queued, Running, #[default] Successful, @@ -985,6 +1006,7 @@ pub(crate) enum TableMetadataMaintenanceJobStatus { #[serde(rename_all = "SCREAMING_SNAKE_CASE")] pub(crate) enum TableMaintenanceRecommendedAction { NoActionRequired, + RunMaintenanceWorker, ReviewAndRunDelete, ReviewQuarantine, EnableDelete, @@ -1079,7 +1101,26 @@ struct TableMaintenanceWorkerControlReport<'a> { now: OffsetDateTime, } +struct TableMaintenanceSchedulerControlReport<'a> { + table_bucket: &'a str, + namespace: &'a str, + table: &'a str, + scheduler_id: String, + effective: &'a TableMaintenanceEffectiveConfig, + status: TableMetadataMaintenanceJobStatus, + reason: &'a str, + now: OffsetDateTime, +} + enum TableMaintenanceWorkerPreflight { + Ready { + effective: TableMaintenanceEffectiveConfig, + queued: Option>, + }, + Complete(Box), +} + +enum TableMaintenanceSchedulerPreflight { Ready(TableMaintenanceEffectiveConfig), Complete(Box), } @@ -4212,6 +4253,12 @@ where { push_unique_maintenance_action(&mut recommended_actions, TableMaintenanceRecommendedAction::WaitForActiveWorker); TableMaintenanceSchedulerStatus::Backpressured + } else if let Some(current) = current.as_ref() + && matches!(current.job.status, TableMetadataMaintenanceJobStatus::Queued) + && table_maintenance_scheduler_lease_is_active(¤t.job, effective.config.worker_lease_timeout_seconds, now) + { + push_unique_maintenance_action(&mut recommended_actions, TableMaintenanceRecommendedAction::RunMaintenanceWorker); + TableMaintenanceSchedulerStatus::Queued } else if let Some(current) = current.as_ref() && table_maintenance_job_retry_is_pending(¤t.job, now) { @@ -4247,6 +4294,130 @@ where }) } + pub(crate) async fn run_table_maintenance_scheduler_once( + &self, + table_bucket: &str, + namespace: &str, + table: &str, + scheduler_id: String, + ) -> TableCatalogStoreResult { + self.run_table_maintenance_scheduler_once_at(table_bucket, namespace, table, scheduler_id, OffsetDateTime::now_utc()) + .await + } + + async fn run_table_maintenance_scheduler_once_at( + &self, + table_bucket: &str, + namespace: &str, + table: &str, + scheduler_id: String, + now: OffsetDateTime, + ) -> TableCatalogStoreResult { + 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 effective = { + let _guard = self.backend.acquire_write_lock(self.catalog_bucket(), &table_path).await?; + match self + .table_metadata_maintenance_scheduler_preflight(table_bucket, &namespace_name, &table_name, &scheduler_id, now) + .await? + { + TableMaintenanceSchedulerPreflight::Ready(effective) => effective, + TableMaintenanceSchedulerPreflight::Complete(report) => { + let scheduler = self + .get_table_maintenance_scheduler_report_at(table_bucket, &namespace_name, &table_name, now) + .await?; + return Ok(TableMaintenanceSchedulerRunResult { + report: *report, + scheduler, + }); + } + } + }; + + let mut report = self + .plan_table_metadata_maintenance( + table_bucket, + &namespace_name, + &table_name, + effective.config.retain_recent_metadata_files, + ) + .await?; + + let report = { + let _guard = self.backend.acquire_write_lock(self.catalog_bucket(), &table_path).await?; + match self + .table_metadata_maintenance_scheduler_preflight(table_bucket, &namespace_name, &table_name, &scheduler_id, now) + .await? + { + TableMaintenanceSchedulerPreflight::Ready(effective) => { + if report.job.retain_recent_metadata_files != effective.config.retain_recent_metadata_files { + return Err(TableCatalogStoreError::Conflict( + "maintenance config changed before scheduler claim".to_string(), + )); + } + let Some((entry, _)) = self.read_table_with_etag_unlocked(table_bucket, &namespace, &table).await? else { + return Err(TableCatalogStoreError::NotFound(format!( + "table {}/{}/{}", + table_bucket, namespace_name, table_name + ))); + }; + if entry.metadata_location != report.current_metadata_location { + return Err(TableCatalogStoreError::Conflict( + "current metadata location changed before maintenance scheduler claim".to_string(), + )); + } + + let before_status = Some(report.job.status.clone()); + let before_quarantined_object_count = Some(report.job.quarantined_object_count); + let scheduled_at = maintenance_timestamp(now); + report.job.operation = if effective.config.delete_enabled { + TableMetadataMaintenanceOperation::Delete + } else { + TableMetadataMaintenanceOperation::DryRun + }; + report.job.status = TableMetadataMaintenanceJobStatus::Queued; + report.job.failure_reason = None; + report.job.config_source = effective.source; + report.job.scheduler_id = Some(scheduler_id); + report.job.scheduler_lease_id = Uuid::new_v4().to_string(); + report.job.scheduled_at = Some(scheduled_at); + report.job.worker_id = None; + report.job.lease_id = String::new(); + report.job.attempt = 0; + report.job.max_retry_attempts = effective.config.max_retry_attempts; + report.job.next_retry_after = None; + report.job.quarantine_enabled = effective.config.quarantine_enabled; + report.job.quarantine_retention_seconds = effective.config.quarantine_retention_seconds; + report.job.heartbeat_at = None; + report.job.started_at = None; + report.job.finished_at = None; + refresh_table_maintenance_report_recommended_actions(&mut report); + push_table_maintenance_audit_event( + &mut report, + now, + TableMaintenanceAuditActor::Scheduler, + TableMaintenanceAuditAction::SchedulerQueued, + None, + before_status, + before_quarantined_object_count, + ); + self.put_table_metadata_maintenance_report(&report).await?; + report + } + TableMaintenanceSchedulerPreflight::Complete(report) => *report, + } + }; + + let scheduler = self + .get_table_maintenance_scheduler_report_at(table_bucket, &namespace_name, &table_name, now) + .await?; + Ok(TableMaintenanceSchedulerRunResult { report, scheduler }) + } + pub(crate) async fn apply_table_maintenance_quarantine_operation( &self, table_bucket: &str, @@ -4379,6 +4550,82 @@ where Ok(reports) } + async fn table_metadata_maintenance_scheduler_preflight( + &self, + table_bucket: &str, + namespace: &str, + table: &str, + scheduler_id: &str, + now: OffsetDateTime, + ) -> TableCatalogStoreResult { + let effective = self + .get_effective_table_maintenance_config(table_bucket, namespace, table) + .await?; + if !effective.config.background_enabled { + let report = self + .put_table_metadata_maintenance_scheduler_control_report(TableMaintenanceSchedulerControlReport { + table_bucket, + namespace, + table, + scheduler_id: scheduler_id.to_string(), + effective: &effective, + status: TableMetadataMaintenanceJobStatus::Disabled, + reason: "background maintenance is disabled", + now, + }) + .await?; + return Ok(TableMaintenanceSchedulerPreflight::Complete(Box::new(report))); + } + if effective.config.worker_paused { + let report = self + .put_table_metadata_maintenance_scheduler_control_report(TableMaintenanceSchedulerControlReport { + table_bucket, + namespace, + table, + scheduler_id: scheduler_id.to_string(), + effective: &effective, + status: TableMetadataMaintenanceJobStatus::Paused, + reason: "background maintenance worker is paused", + now, + }) + .await?; + return Ok(TableMaintenanceSchedulerPreflight::Complete(Box::new(report))); + } + + if let Some(current) = self + .get_table_metadata_maintenance_report(table_bucket, namespace, table, MAINTENANCE_JOB_ALIAS_CURRENT) + .await? + { + if matches!(current.job.status, TableMetadataMaintenanceJobStatus::Running) { + if table_maintenance_job_lease_is_active(¤t.job, effective.config.worker_lease_timeout_seconds, now) { + return Ok(TableMaintenanceSchedulerPreflight::Complete(Box::new(current))); + } + self.expire_table_maintenance_job( + current, + now, + "maintenance worker lease expired", + TableMaintenanceAuditAction::WorkerLeaseExpired, + ) + .await?; + } else if matches!(current.job.status, TableMetadataMaintenanceJobStatus::Queued) { + if table_maintenance_scheduler_lease_is_active(¤t.job, effective.config.worker_lease_timeout_seconds, now) { + return Ok(TableMaintenanceSchedulerPreflight::Complete(Box::new(current))); + } + self.expire_table_maintenance_job( + current, + now, + "maintenance scheduler lease expired", + TableMaintenanceAuditAction::SchedulerLeaseExpired, + ) + .await?; + } else if table_maintenance_job_retry_is_pending(¤t.job, now) { + return Ok(TableMaintenanceSchedulerPreflight::Complete(Box::new(current))); + } + } + + Ok(TableMaintenanceSchedulerPreflight::Ready(effective)) + } + pub(crate) async fn run_table_metadata_maintenance_worker_once( &self, table_bucket: &str, @@ -4404,35 +4651,46 @@ where let namespace_name = namespace.public_name(); let table_name = table.as_str().to_string(); - let effective = { + let (effective, queued) = { let _guard = self.backend.acquire_write_lock(self.catalog_bucket(), &table_path).await?; match self .table_metadata_maintenance_worker_preflight(table_bucket, &namespace_name, &table_name, &worker_id, now) .await? { - TableMaintenanceWorkerPreflight::Ready(effective) => effective, + TableMaintenanceWorkerPreflight::Ready { effective, queued } => (effective, queued), TableMaintenanceWorkerPreflight::Complete(report) => return Ok(*report), } }; - let mut report = self - .plan_table_metadata_maintenance( + let mut report = if let Some(queued) = queued { + *queued + } else { + self.plan_table_metadata_maintenance( table_bucket, &namespace_name, &table_name, effective.config.retain_recent_metadata_files, ) - .await?; + .await? + }; let (report, effective, delete) = { let _guard = self.backend.acquire_write_lock(self.catalog_bucket(), &table_path).await?; - let effective = match self + let (effective, queued) = match self .table_metadata_maintenance_worker_preflight(table_bucket, &namespace_name, &table_name, &worker_id, now) .await? { - TableMaintenanceWorkerPreflight::Ready(effective) => effective, + TableMaintenanceWorkerPreflight::Ready { effective, queued } => (effective, queued), TableMaintenanceWorkerPreflight::Complete(report) => return Ok(*report), }; + if let Some(queued) = queued { + if queued.job.job_id != report.job.job_id { + return Err(TableCatalogStoreError::Conflict( + "queued maintenance job changed before worker claim".to_string(), + )); + } + report = *queued; + } if report.job.retain_recent_metadata_files != effective.config.retain_recent_metadata_files { return Err(TableCatalogStoreError::Conflict( @@ -4451,12 +4709,17 @@ where )); } + let was_queued_claim = matches!(report.job.status, TableMetadataMaintenanceJobStatus::Queued); + let before_status = Some(report.job.status.clone()); + let before_quarantined_object_count = Some(report.job.quarantined_object_count); let started_at = maintenance_timestamp(now); - report.job.operation = if effective.config.delete_enabled { - TableMetadataMaintenanceOperation::Delete - } else { - TableMetadataMaintenanceOperation::DryRun - }; + if !was_queued_claim { + report.job.operation = if effective.config.delete_enabled { + TableMetadataMaintenanceOperation::Delete + } else { + TableMetadataMaintenanceOperation::DryRun + }; + } report.job.status = TableMetadataMaintenanceJobStatus::Running; report.job.failure_reason = None; report.job.config_source = effective.source; @@ -4477,12 +4740,12 @@ where TableMaintenanceAuditActor::Worker, TableMaintenanceAuditAction::WorkerStarted, None, - Some(TableMetadataMaintenanceJobStatus::Successful), - Some(0), + before_status, + before_quarantined_object_count, ); self.put_table_metadata_maintenance_report(&report).await?; - let delete = effective.config.delete_enabled; + let delete = matches!(report.job.operation, TableMetadataMaintenanceOperation::Delete); (report, effective, delete) }; @@ -4540,29 +4803,33 @@ where if table_maintenance_job_lease_is_active(¤t.job, effective.config.worker_lease_timeout_seconds, now) { return Ok(TableMaintenanceWorkerPreflight::Complete(Box::new(current))); } - let mut expired = current; - let before_status = Some(expired.job.status.clone()); - let before_quarantined_object_count = Some(expired.job.quarantined_object_count); - expired.job.status = TableMetadataMaintenanceJobStatus::Failed; - expired.job.failure_reason = Some("maintenance worker lease expired".to_string()); - expired.job.finished_at = Some(maintenance_timestamp(now)); - refresh_table_maintenance_report_recommended_actions(&mut expired); - push_table_maintenance_audit_event( - &mut expired, + self.expire_table_maintenance_job( + current, now, - TableMaintenanceAuditActor::Scheduler, + "maintenance worker lease expired", TableMaintenanceAuditAction::WorkerLeaseExpired, - Some("maintenance worker lease expired".to_string()), - before_status, - before_quarantined_object_count, - ); - self.put_table_metadata_maintenance_report(&expired).await?; + ) + .await?; + } else if matches!(current.job.status, TableMetadataMaintenanceJobStatus::Queued) { + if table_maintenance_scheduler_lease_is_active(¤t.job, effective.config.worker_lease_timeout_seconds, now) { + return Ok(TableMaintenanceWorkerPreflight::Ready { + effective, + queued: Some(Box::new(current)), + }); + } + self.expire_table_maintenance_job( + current, + now, + "maintenance scheduler lease expired", + TableMaintenanceAuditAction::SchedulerLeaseExpired, + ) + .await?; } else if table_maintenance_job_retry_is_pending(¤t.job, now) { return Ok(TableMaintenanceWorkerPreflight::Complete(Box::new(current))); } } - Ok(TableMaintenanceWorkerPreflight::Ready(effective)) + Ok(TableMaintenanceWorkerPreflight::Ready { effective, queued: None }) } pub(crate) async fn heartbeat_table_metadata_maintenance_job( @@ -4643,6 +4910,120 @@ where Ok(report) } + async fn expire_table_maintenance_job( + &self, + mut report: TableMetadataMaintenanceReport, + now: OffsetDateTime, + reason: &str, + action: TableMaintenanceAuditAction, + ) -> TableCatalogStoreResult { + let before_status = Some(report.job.status.clone()); + let before_quarantined_object_count = Some(report.job.quarantined_object_count); + report.job.status = TableMetadataMaintenanceJobStatus::Failed; + report.job.failure_reason = Some(reason.to_string()); + report.job.finished_at = Some(maintenance_timestamp(now)); + refresh_table_maintenance_report_recommended_actions(&mut report); + push_table_maintenance_audit_event( + &mut report, + now, + TableMaintenanceAuditActor::Scheduler, + action, + Some(reason.to_string()), + before_status, + before_quarantined_object_count, + ); + self.put_table_metadata_maintenance_report(&report).await?; + Ok(report) + } + + async fn put_table_metadata_maintenance_scheduler_control_report( + &self, + control: TableMaintenanceSchedulerControlReport<'_>, + ) -> TableCatalogStoreResult { + let namespace = parse_namespace_for_store(control.namespace)?; + let table = parse_table_for_store(control.table)?; + let table_path = self.paths.table_entry_path(control.table_bucket, &namespace, &table); + let Some((entry, _)) = self.read_entry::(self.catalog_bucket(), &table_path).await? else { + return Err(TableCatalogStoreError::NotFound(format!( + "table {}/{}/{}", + control.table_bucket, + namespace.public_name(), + table.as_str() + ))); + }; + let timestamp = maintenance_timestamp(control.now); + let cleanup_watermark_unix_seconds = + (control.now - Duration::seconds(TABLE_METADATA_CLEANUP_SAFETY_WINDOW_SECONDS)).unix_timestamp(); + let current_metadata_location = entry.metadata_location.clone(); + let report = TableMetadataMaintenanceReport { + job: TableMetadataMaintenanceJob { + job_id: Uuid::new_v4().to_string(), + table_bucket: control.table_bucket.to_string(), + namespace: namespace.public_name(), + table: table.as_str().to_string(), + table_id: entry.table_id, + operation: TableMetadataMaintenanceOperation::DryRun, + status: control.status, + failure_reason: Some(control.reason.to_string()), + recommended_actions: Vec::new(), + config_source: control.effective.source, + scheduler_id: Some(control.scheduler_id), + scheduler_lease_id: String::new(), + scheduled_at: Some(timestamp.clone()), + worker_id: None, + lease_id: String::new(), + attempt: 0, + max_retry_attempts: control.effective.config.max_retry_attempts, + next_retry_after: None, + quarantine_enabled: control.effective.config.quarantine_enabled, + quarantine_retention_seconds: control.effective.config.quarantine_retention_seconds, + heartbeat_at: None, + started_at: None, + finished_at: Some(timestamp), + current_metadata_location: current_metadata_location.clone(), + current_generation: entry.generation, + retain_recent_metadata_files: control.effective.config.retain_recent_metadata_files, + safety_window_seconds: TABLE_METADATA_CLEANUP_SAFETY_WINDOW_SECONDS, + cleanup_watermark_unix_seconds, + planned_metadata_file_count: 0, + retained_metadata_file_count: 0, + cleanup_candidate_count: 0, + deletable_metadata_file_count: 0, + deleted_metadata_file_count: 0, + planned_object_file_count: 0, + cleanup_candidate_object_count: 0, + deletable_object_count: 0, + deleted_object_count: 0, + quarantined_object_count: 0, + }, + current_metadata_location, + retained_metadata_locations: Vec::new(), + cleanup_candidate_locations: Vec::new(), + deletable_metadata_locations: Vec::new(), + cleanup_object_candidate_locations: Vec::new(), + deletable_object_locations: Vec::new(), + object_reports: Vec::new(), + object_cleanup_reports: Vec::new(), + referenced_object_reports: Vec::new(), + reachability_graph: TableMaintenanceReachabilityGraphReport::default(), + snapshot_expiration: None, + compaction: None, + audit_events: Vec::new(), + }; + let mut report = table_maintenance_report_with_recommended_actions(report); + push_table_maintenance_audit_event( + &mut report, + control.now, + TableMaintenanceAuditActor::Scheduler, + TableMaintenanceAuditAction::SchedulerControl, + Some(control.reason.to_string()), + None, + None, + ); + self.put_table_metadata_maintenance_report(&report).await?; + Ok(report) + } + async fn put_table_metadata_maintenance_worker_control_report( &self, control: TableMaintenanceWorkerControlReport<'_>, @@ -4674,6 +5055,9 @@ where failure_reason: Some(control.reason.to_string()), recommended_actions: Vec::new(), config_source: control.effective.source, + scheduler_id: None, + scheduler_lease_id: String::new(), + scheduled_at: None, worker_id: Some(control.worker_id), lease_id: String::new(), attempt: 0, @@ -5430,6 +5814,9 @@ where failure_reason: None, recommended_actions: Vec::new(), config_source: TableMaintenanceConfigSource::Default, + scheduler_id: None, + scheduler_lease_id: String::new(), + scheduled_at: None, worker_id: None, lease_id: String::new(), attempt: 0, @@ -6717,6 +7104,23 @@ where } } + pub(crate) async fn run_table_maintenance_scheduler_once( + &self, + table_bucket: &str, + namespace: &str, + table: &str, + scheduler_id: String, + ) -> TableCatalogStoreResult { + match self { + Self::ObjectBacked(store) => { + store + .run_table_maintenance_scheduler_once(table_bucket, namespace, table, scheduler_id) + .await + } + Self::DurableStrong(_) => Err(Self::unsupported_for_durable_strong("table maintenance scheduler")), + } + } + pub(crate) async fn apply_table_maintenance_quarantine_operation( &self, table_bucket: &str, @@ -9471,6 +9875,9 @@ fn table_maintenance_recommended_actions(job: &TableMetadataMaintenanceJob) -> V let mut actions = Vec::new(); match job.status { TableMetadataMaintenanceJobStatus::NotYetRun => {} + TableMetadataMaintenanceJobStatus::Queued => { + actions.push(TableMaintenanceRecommendedAction::RunMaintenanceWorker); + } TableMetadataMaintenanceJobStatus::Running => { actions.push(TableMaintenanceRecommendedAction::WaitForActiveWorker); } @@ -9527,6 +9934,7 @@ fn table_maintenance_report_order_timestamp(report: &TableMetadataMaintenanceRep .clone() .or_else(|| report.job.heartbeat_at.clone()) .or_else(|| report.job.started_at.clone()) + .or_else(|| report.job.scheduled_at.clone()) .unwrap_or_default() } @@ -9535,6 +9943,8 @@ fn table_maintenance_scheduler_job_summary(report: &TableMetadataMaintenanceRepo job_id: report.job.job_id.clone(), operation: report.job.operation.clone(), status: report.job.status.clone(), + scheduler_id: report.job.scheduler_id.clone(), + scheduled_at: report.job.scheduled_at.clone(), worker_id: report.job.worker_id.clone(), attempt: report.job.attempt, started_at: report.job.started_at.clone(), @@ -9573,6 +9983,18 @@ fn table_maintenance_report_with_recommended_actions( report } +fn table_maintenance_scheduler_lease_is_active( + job: &TableMetadataMaintenanceJob, + scheduler_lease_timeout_seconds: u64, + now: OffsetDateTime, +) -> bool { + let Some(scheduled_at) = job.scheduled_at.as_deref().and_then(parse_maintenance_timestamp) else { + return false; + }; + let timeout_seconds = i64::try_from(scheduler_lease_timeout_seconds).unwrap_or(i64::MAX); + scheduled_at.saturating_add(Duration::seconds(timeout_seconds)) > now +} + fn table_maintenance_job_lease_is_active( job: &TableMetadataMaintenanceJob, worker_lease_timeout_seconds: u64, @@ -13171,6 +13593,376 @@ mod tests { assert!(!report.quarantine.active); } + #[tokio::test] + async fn maintenance_scheduler_run_queues_one_durable_job() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + 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"); + let now = OffsetDateTime::UNIX_EPOCH + Duration::seconds(100); + + 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, + retain_recent_metadata_files: 2, + ..Default::default() + }, + ) + .await + .expect("background maintenance config should persist"); + + let result = store + .run_table_maintenance_scheduler_once_at(bucket, "sales", "orders", "scheduler-a".to_string(), now) + .await + .expect("scheduler tick should queue maintenance"); + + assert_eq!(result.report.job.status, TableMetadataMaintenanceJobStatus::Queued); + assert_eq!(result.report.job.scheduler_id.as_deref(), Some("scheduler-a")); + assert!(!result.report.job.scheduler_lease_id.is_empty()); + assert_eq!(result.report.job.retain_recent_metadata_files, 2); + assert_eq!(result.scheduler.status, TableMaintenanceSchedulerStatus::Queued); + assert_eq!( + result.scheduler.current_job.as_ref().map(|job| job.job_id.as_str()), + Some(result.report.job.job_id.as_str()) + ); + assert_eq!( + result.report.audit_events.last().map(|event| event.action.clone()), + Some(TableMaintenanceAuditAction::SchedulerQueued) + ); + } + + #[tokio::test] + async fn maintenance_scheduler_run_persists_disabled_control_report() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + 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"); + let now = OffsetDateTime::UNIX_EPOCH + Duration::seconds(100); + + seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current.clone()).await; + backend + .seed_object(bucket, ¤t, br#"{"metadata-log":[]}"#.to_vec()) + .await; + + let result = store + .run_table_maintenance_scheduler_once_at(bucket, "sales", "orders", "scheduler-a".to_string(), now) + .await + .expect("disabled scheduler tick should persist a control report"); + + assert_eq!(result.report.job.status, TableMetadataMaintenanceJobStatus::Disabled); + assert_eq!(result.report.job.scheduler_id.as_deref(), Some("scheduler-a")); + let stored = store + .get_table_metadata_maintenance_report(bucket, "sales", "orders", &result.report.job.job_id) + .await + .expect("scheduler control report lookup should succeed") + .expect("scheduler control report should be durable"); + assert_eq!(stored.job.job_id, result.report.job.job_id); + assert_eq!(stored.job.status, TableMetadataMaintenanceJobStatus::Disabled); + assert_eq!( + stored.audit_events.last().map(|event| event.action.clone()), + Some(TableMaintenanceAuditAction::SchedulerControl) + ); + let scheduler = store + .get_table_maintenance_scheduler_report_at(bucket, "sales", "orders", now) + .await + .expect("scheduler report should load"); + assert_eq!(scheduler.audit_timeline.len(), 1); + assert_eq!(scheduler.audit_timeline[0].job_id, result.report.job.job_id); + } + + #[tokio::test] + async fn maintenance_scheduler_run_reuses_active_queued_job() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + 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"); + let now = OffsetDateTime::UNIX_EPOCH + Duration::seconds(100); + + 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, + worker_lease_timeout_seconds: 300, + ..Default::default() + }, + ) + .await + .expect("background maintenance config should persist"); + + let first = store + .run_table_maintenance_scheduler_once_at(bucket, "sales", "orders", "scheduler-a".to_string(), now) + .await + .expect("first scheduler tick should queue maintenance"); + let second = store + .run_table_maintenance_scheduler_once_at( + bucket, + "sales", + "orders", + "scheduler-b".to_string(), + now + Duration::seconds(30), + ) + .await + .expect("second scheduler tick should reuse the queued job"); + + assert_eq!(second.report.job.job_id, first.report.job.job_id); + assert_eq!(second.report.job.scheduler_id.as_deref(), Some("scheduler-a")); + assert_eq!(second.report.job.status, TableMetadataMaintenanceJobStatus::Queued); + assert_eq!(second.scheduler.audit_timeline.len(), 1); + } + + #[tokio::test] + async fn maintenance_worker_claims_queued_job_before_running_it() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + 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"); + let now = OffsetDateTime::UNIX_EPOCH + Duration::seconds(100); + + 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, + ..Default::default() + }, + ) + .await + .expect("background maintenance config should persist"); + let queued = store + .run_table_maintenance_scheduler_once_at(bucket, "sales", "orders", "scheduler-a".to_string(), now) + .await + .expect("scheduler tick should queue maintenance"); + + let finished = store + .run_table_metadata_maintenance_worker_once_at( + bucket, + "sales", + "orders", + "worker-a".to_string(), + now + Duration::seconds(10), + ) + .await + .expect("worker tick should claim and finish the queued job"); + + assert_eq!(finished.job.job_id, queued.report.job.job_id); + assert_eq!(finished.job.status, TableMetadataMaintenanceJobStatus::Successful); + assert_eq!(finished.job.worker_id.as_deref(), Some("worker-a")); + assert_eq!(finished.job.attempt, 1); + assert_eq!(finished.job.scheduler_id.as_deref(), Some("scheduler-a")); + let actions = finished + .audit_events + .iter() + .map(|event| event.action.clone()) + .collect::>(); + assert_eq!( + actions, + vec![ + TableMaintenanceAuditAction::Planned, + TableMaintenanceAuditAction::SchedulerQueued, + TableMaintenanceAuditAction::WorkerStarted, + TableMaintenanceAuditAction::WorkerSucceeded, + ] + ); + } + + #[tokio::test] + async fn maintenance_worker_preserves_queued_dry_run_after_delete_is_enabled() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + 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"); + let metadata_dir = default_table_metadata_dir_path(&namespace, &table); + let table_root = format!("{}{}/", default_table_root_prefix(&namespace), table.as_str()); + let manifest_list = format!("{metadata_dir}/snap-10.avro"); + let manifest = format!("{metadata_dir}/manifest-10.avro"); + let data_file = format!("{table_root}data/part-00001.parquet"); + let orphan_data = format!("{table_root}data/orphan.parquet"); + let now = OffsetDateTime::UNIX_EPOCH + Duration::seconds(100); + + seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current.clone()).await; + backend + .seed_object(bucket, &manifest_list, manifest_list_avro_bytes(&[&manifest])) + .await; + backend + .seed_object(bucket, &manifest, manifest_avro_bytes(&[(&data_file, 0)])) + .await; + backend.seed_object(bucket, &data_file, b"data".to_vec()).await; + backend.seed_object(bucket, &orphan_data, b"orphan-data".to_vec()).await; + backend + .seed_object( + bucket, + ¤t, + serde_json::to_vec(&serde_json::json!({ + "metadata-log": [], + "snapshots": [ + { + "snapshot-id": 10, + "manifest-list": manifest_list + } + ], + "refs": { + "main": { + "snapshot-id": 10, + "type": "branch" + } + } + })) + .unwrap(), + ) + .await; + store + .put_table_maintenance_config( + bucket, + "sales", + "orders", + TableMaintenanceConfig { + version: TABLE_MAINTENANCE_CONFIG_VERSION, + background_enabled: true, + delete_enabled: false, + ..Default::default() + }, + ) + .await + .expect("dry-run background maintenance config should persist"); + let queued = store + .run_table_maintenance_scheduler_once_at(bucket, "sales", "orders", "scheduler-a".to_string(), now) + .await + .expect("scheduler tick should queue dry-run maintenance"); + assert_eq!(queued.report.job.operation, TableMetadataMaintenanceOperation::DryRun); + assert_eq!(queued.report.job.deletable_object_count, 1); + + store + .put_table_maintenance_config( + bucket, + "sales", + "orders", + TableMaintenanceConfig { + version: TABLE_MAINTENANCE_CONFIG_VERSION, + background_enabled: true, + delete_enabled: true, + ..Default::default() + }, + ) + .await + .expect("delete-enabled background maintenance config should persist"); + let finished = store + .run_table_metadata_maintenance_worker_once_at( + bucket, + "sales", + "orders", + "worker-a".to_string(), + now + Duration::seconds(10), + ) + .await + .expect("worker tick should preserve the queued dry-run operation"); + + assert_eq!(finished.job.job_id, queued.report.job.job_id); + assert_eq!(finished.job.operation, TableMetadataMaintenanceOperation::DryRun); + assert_eq!(finished.job.status, TableMetadataMaintenanceJobStatus::Successful); + assert_eq!(finished.job.deleted_object_count, 0); + assert!(backend.object_exists(bucket, &orphan_data).await.unwrap()); + } + + #[tokio::test] + async fn maintenance_scheduler_run_recovers_expired_queued_job() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + 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"); + let now = OffsetDateTime::UNIX_EPOCH + Duration::seconds(1000); + + 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, + worker_lease_timeout_seconds: 60, + ..Default::default() + }, + ) + .await + .expect("background maintenance config should persist"); + let first = store + .run_table_maintenance_scheduler_once_at(bucket, "sales", "orders", "scheduler-a".to_string(), now) + .await + .expect("first scheduler tick should queue maintenance"); + + let second = store + .run_table_maintenance_scheduler_once_at( + bucket, + "sales", + "orders", + "scheduler-b".to_string(), + now + Duration::seconds(120), + ) + .await + .expect("scheduler tick should recover expired queued maintenance"); + + assert_ne!(second.report.job.job_id, first.report.job.job_id); + let expired = store + .get_table_metadata_maintenance_report(bucket, "sales", "orders", &first.report.job.job_id) + .await + .expect("expired queued job lookup should succeed") + .expect("expired queued job should remain addressable"); + assert_eq!(expired.job.status, TableMetadataMaintenanceJobStatus::Failed); + assert!( + expired + .job + .failure_reason + .as_deref() + .is_some_and(|reason| reason.contains("scheduler lease expired")) + ); + assert_eq!( + expired.audit_events.last().map(|event| event.action.clone()), + Some(TableMaintenanceAuditAction::SchedulerLeaseExpired) + ); + assert_eq!(second.report.job.status, TableMetadataMaintenanceJobStatus::Queued); + assert_eq!(second.report.job.scheduler_id.as_deref(), Some("scheduler-b")); + } + #[tokio::test] async fn maintenance_scheduler_report_surfaces_active_backpressure_and_audit_timeline() { let backend = TestCatalogObjectBackend::default(); diff --git a/scripts/table-catalog/README.md b/scripts/table-catalog/README.md index 04268dd80..c55430961 100644 --- a/scripts/table-catalog/README.md +++ b/scripts/table-catalog/README.md @@ -230,9 +230,9 @@ The smoke test also probes catalog-backed advanced Iceberg surfaces: `main` cannot be deleted - Iceberg views support basic create, list, load, replace, existence check, and 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; job reports expose +- metadata maintenance supports safe dry-run planning, controlled scheduler + queueing, worker execution checks, and a scheduler status report with disabled, + paused, queued, backpressure, retry, quarantine, and audit-timeline state; job reports expose structured audit events for planning, worker transitions, heartbeat updates, lease expiry, and mutating quarantine operations; quarantined jobs can be inspected, released, retried, or abandoned through an operator endpoint @@ -338,10 +338,10 @@ 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, quarantine operation, and scheduler status endpoints are registered; disabled/paused/backpressure/retry/quarantine/audit-timeline state and per-job audit events are machine-readable; continuous in-process scheduling is not claimed +- background maintenance worker: controlled scheduler run, worker run-once, heartbeat, quarantine operation, and scheduler status endpoints are registered; disabled/paused/queued/backpressure/retry/quarantine/audit-timeline state and per-job audit events are 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 +- automatic maintenance scheduling: external scheduler hook supported through the scheduler run endpoint, worker run endpoint, and scheduler status report; built-in periodic scheduling is not claimed - compaction rewrite: controlled run-once support for partition-local and sort-order-preserving Parquet binpack through metadata maintenance; manifests with position or equality delete files produce machine-readable row-level planning and fail closed before rewrite; built-in periodic scheduling, delete-file rewrite, and row-level compaction execution are not claimed - row-level delete/update/merge commits: standard catalog commit validates append, overwrite, delete, and replace snapshot manifests for table-warehouse scope, referenced object existence, current-live-file deletes, and stale add/delete conflicts; end-to-end SQL DML client coverage remains a compatibility validation item - external catalog bridges: metadata import/register and operator-supplied metadata pointer sync are supported for Polaris/Glue/DLF/Hive identity boundaries; online vendor SDK polling and policy mirroring are not claimed diff --git a/scripts/table-catalog/pyiceberg_smoke.py b/scripts/table-catalog/pyiceberg_smoke.py index 40596631e..779f467ab 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, per-job audit events, lease expiry recovery, heartbeat updates, and operator quarantine inspect/release/retry/abandon actions; built-in periodic scheduling is not claimed", + "expected_behavior": "background-enabled maintenance can be queued by the scheduler run endpoint, claimed by the worker run endpoint, and inspected through the scheduler status endpoint with disabled/paused state, queued-job handoff, current-job backpressure, retry deferral, quarantine boundary, audit timeline, per-job audit events, 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", @@ -1050,6 +1050,7 @@ def run_view_probe(args: argparse.Namespace, deps: RuntimeDeps) -> None: def run_maintenance_probe(args: argparse.Namespace, deps: RuntimeDeps) -> None: config_path = table_endpoint_path(args, "/maintenance/config") scheduler_path = table_endpoint_path(args, "/maintenance/scheduler") + scheduler_run_path = table_endpoint_path(args, "/maintenance/scheduler/run") signed_rest_request(args, deps, "PUT", config_path, default_maintenance_config()) config = signed_rest_request(args, deps, "GET", config_path) if config.get("version") != TABLE_MAINTENANCE_CONFIG_VERSION: @@ -1083,7 +1084,7 @@ def run_maintenance_probe(args: argparse.Namespace, deps: RuntimeDeps) -> None: 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"} + expected_scheduler_statuses = {"READY", "QUEUED", "DISABLED", "PAUSED", "BACKPRESSURED", "RETRY_DEFERRED", "QUARANTINED"} if scheduler.get("status") not in expected_scheduler_statuses: raise RuntimeError("maintenance scheduler endpoint did not return a stable scheduler status") scheduler_timeline = scheduler.get("audit_timeline") @@ -1092,9 +1093,23 @@ def run_maintenance_probe(args: argparse.Namespace, deps: RuntimeDeps) -> None: if scheduler_timeline and not isinstance(scheduler_timeline[0].get("audit-events"), list): raise RuntimeError("maintenance scheduler audit timeline did not include job audit events") + scheduler_config = default_maintenance_config() + scheduler_config["background-enabled"] = True + scheduler_config["worker-paused"] = False + signed_rest_request(args, deps, "PUT", config_path, scheduler_config) + scheduler_run = signed_rest_request(args, deps, "POST", scheduler_run_path, {"scheduler-id": "pyiceberg-smoke-scheduler"}) + scheduler_run_job = scheduler_run.get("report", {}).get("job") + if not isinstance(scheduler_run_job, dict) or scheduler_run_job.get("status") != "QUEUED": + raise RuntimeError("maintenance scheduler run endpoint did not queue a maintenance job") + if scheduler_run_job.get("scheduler-id") != "pyiceberg-smoke-scheduler": + raise RuntimeError("maintenance scheduler run endpoint did not preserve scheduler identity") + scheduler_run_report = scheduler_run.get("scheduler") + if not isinstance(scheduler_run_report, dict) or scheduler_run_report.get("status") != "QUEUED": + raise RuntimeError("maintenance scheduler run endpoint did not return queued scheduler state") + worker = signed_rest_request(args, deps, "POST", table_endpoint_path(args, "/maintenance/worker/run"), {}) worker_job = worker.get("job") - expected_statuses = {"DISABLED", "PAUSED", "FAILED", "SUCCESSFUL", "RUNNING"} + expected_statuses = {"DISABLED", "PAUSED", "FAILED", "SUCCESSFUL", "RUNNING", "QUEUED"} if not isinstance(worker_job, dict) or worker_job.get("status") not in expected_statuses: raise RuntimeError("maintenance worker endpoint did not return a stable job status") if not isinstance(worker.get("audit-events"), list):