diff --git a/rustfs/src/admin/handlers/table_catalog.rs b/rustfs/src/admin/handlers/table_catalog.rs index 9a14db665..1854910ad 100644 --- a/rustfs/src/admin/handlers/table_catalog.rs +++ b/rustfs/src/admin/handlers/table_catalog.rs @@ -1076,6 +1076,7 @@ mod tests { crate::table_catalog::TableCatalogObject { data, etag: Some("etag".to_string()), + mod_time: None, }, ); } @@ -1116,6 +1117,7 @@ mod tests { crate::table_catalog::TableCatalogObject { data, etag: Some("etag".to_string()), + mod_time: None, }, ); Ok(()) diff --git a/rustfs/src/table_catalog.rs b/rustfs/src/table_catalog.rs index 873ee86ce..1d4eea293 100644 --- a/rustfs/src/table_catalog.rs +++ b/rustfs/src/table_catalog.rs @@ -38,6 +38,7 @@ use rustfs_ecstore::{ }; use serde::{Deserialize, Serialize, de::DeserializeOwned}; use sha2::{Digest, Sha256}; +use time::{Duration, OffsetDateTime}; use tokio::io::AsyncReadExt; use uuid::Uuid; @@ -67,6 +68,7 @@ const TABLE_ENTRY_FILE: &str = "table-entry.json"; const COMMIT_LOG_ROOT: &str = "commits"; const COMMIT_IDEMPOTENCY_ROOT: &str = "commit-idempotency"; const TABLE_CATALOG_LIST_MAX_KEYS: i32 = 1000; +const TABLE_METADATA_CLEANUP_SAFETY_WINDOW_SECONDS: i64 = 15 * 60; #[derive(Debug, Clone, PartialEq, Eq)] pub enum CatalogIdentifierError { @@ -239,6 +241,13 @@ pub(crate) struct TableCommitResult { pub commit_log: CommitLogEntry, } +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub(crate) struct TableMetadataMaintenanceReport { + pub current_metadata_location: String, + pub retained_metadata_locations: Vec, + pub cleanup_candidate_locations: Vec, +} + #[derive(Debug, Clone, PartialEq, Eq)] pub(crate) enum TableCatalogStoreError { NotFound(String), @@ -307,6 +316,7 @@ pub(crate) trait TableCatalogStore: Send + Sync { pub(crate) struct TableCatalogObject { pub data: Vec, pub etag: Option, + pub mod_time: Option, } #[derive(Debug, Clone, PartialEq, Eq)] @@ -504,6 +514,174 @@ where } Ok(()) } + + pub(crate) async fn plan_table_metadata_maintenance( + &self, + table_bucket: &str, + namespace: &str, + table: &str, + retain_recent_metadata_files: usize, + ) -> TableCatalogStoreResult { + let namespace = parse_namespace_for_store(namespace)?; + let table = parse_table_for_store(table)?; + let table_path = self.paths.table_entry_path(&namespace, &table); + let Some((entry, _)) = self.read_entry::(table_bucket, &table_path).await? else { + return Err(TableCatalogStoreError::NotFound(format!( + "table {}/{}/{}", + table_bucket, + namespace.public_name(), + table.as_str() + ))); + }; + if !is_valid_table_metadata_location(&namespace, &table, &entry.metadata_location) { + return Err(TableCatalogStoreError::Invalid( + "current metadata location must be inside the table metadata directory".to_string(), + )); + } + + let Some(current_metadata_object) = self.backend.read_object(table_bucket, &entry.metadata_location).await? else { + return Err(TableCatalogStoreError::NotFound(format!( + "current metadata object {}", + entry.metadata_location + ))); + }; + let current_metadata = serde_json::from_slice::(¤t_metadata_object.data).map_err(|err| { + TableCatalogStoreError::Invalid(format!("failed to parse current metadata {}: {err}", entry.metadata_location)) + })?; + if !current_metadata.is_object() { + return Err(TableCatalogStoreError::Invalid(format!( + "current metadata {} must be a JSON object", + entry.metadata_location + ))); + } + + let mut retained = metadata_log_locations(¤t_metadata, &namespace, &table); + retained.insert(entry.metadata_location.clone()); + + let mut metadata_locations = Vec::new(); + let metadata_prefix = format!("{}/", default_table_metadata_dir_path(&namespace, &table)); + for object in self.backend.list_objects(table_bucket, &metadata_prefix).await? { + if let Some(metadata_location) = metadata_location_from_metadata_file_path(&namespace, &table, &object) { + metadata_locations.push(metadata_location); + } + } + metadata_locations.sort(); + metadata_locations.dedup(); + + for metadata_location in metadata_locations.iter().rev().take(retain_recent_metadata_files) { + retained.insert(metadata_location.clone()); + } + + let cleanup_candidate_locations = metadata_locations + .into_iter() + .filter(|metadata_location| !retained.contains(metadata_location)) + .collect(); + + Ok(TableMetadataMaintenanceReport { + current_metadata_location: entry.metadata_location, + retained_metadata_locations: retained.into_iter().collect(), + cleanup_candidate_locations, + }) + } + + pub(crate) async fn delete_table_metadata_maintenance_candidates( + &self, + table_bucket: &str, + namespace: &str, + table: &str, + retain_recent_metadata_files: usize, + ) -> TableCatalogStoreResult { + let report = self + .plan_table_metadata_maintenance(table_bucket, namespace, table, retain_recent_metadata_files) + .await?; + self.delete_table_metadata_maintenance_report(table_bucket, namespace, table, report) + .await + } + + async fn delete_table_metadata_maintenance_report( + &self, + table_bucket: &str, + namespace: &str, + table: &str, + report: TableMetadataMaintenanceReport, + ) -> TableCatalogStoreResult { + let namespace = parse_namespace_for_store(namespace)?; + let table = parse_table_for_store(table)?; + if !is_valid_table_metadata_location(&namespace, &table, &report.current_metadata_location) { + return Err(TableCatalogStoreError::Invalid( + "maintenance report current metadata location must be inside the table metadata directory".to_string(), + )); + } + + let table_path = self.paths.table_entry_path(&namespace, &table); + let _guard = self.backend.acquire_write_lock(table_bucket, &table_path).await?; + let Some((entry, _)) = self.read_table_with_etag(table_bucket, &namespace, &table).await? else { + return Err(TableCatalogStoreError::NotFound(format!( + "table {}/{}/{}", + table_bucket, + namespace.public_name(), + table.as_str() + ))); + }; + if entry.metadata_location != report.current_metadata_location { + return Err(TableCatalogStoreError::Conflict( + "current metadata location changed before maintenance delete".to_string(), + )); + } + + let Some(current_metadata_object) = self.backend.read_object(table_bucket, &entry.metadata_location).await? else { + return Err(TableCatalogStoreError::NotFound(format!( + "current metadata object {}", + entry.metadata_location + ))); + }; + let current_metadata = serde_json::from_slice::(¤t_metadata_object.data).map_err(|err| { + TableCatalogStoreError::Invalid(format!("failed to parse current metadata {}: {err}", entry.metadata_location)) + })?; + if !current_metadata.is_object() { + return Err(TableCatalogStoreError::Invalid(format!( + "current metadata {} must be a JSON object", + entry.metadata_location + ))); + } + + let mut protected = metadata_log_locations(¤t_metadata, &namespace, &table); + protected.insert(entry.metadata_location.clone()); + protected.extend(report.retained_metadata_locations.iter().cloned()); + + let mut cleanup_candidate_locations = BTreeSet::new(); + let now = OffsetDateTime::now_utc(); + for metadata_location in report.cleanup_candidate_locations { + if !is_valid_table_metadata_location(&namespace, &table, &metadata_location) { + return Err(TableCatalogStoreError::Invalid(format!( + "cleanup candidate {metadata_location} must be inside the table metadata directory" + ))); + } + if protected.contains(&metadata_location) { + return Err(TableCatalogStoreError::Conflict(format!( + "cleanup candidate {metadata_location} is retained by current metadata" + ))); + } + let Some(candidate_object) = self.backend.read_object(table_bucket, &metadata_location).await? else { + continue; + }; + if !metadata_candidate_is_past_safety_window(candidate_object.mod_time, now) { + continue; + } + cleanup_candidate_locations.insert(metadata_location); + } + + let cleanup_candidate_locations = cleanup_candidate_locations.into_iter().collect::>(); + for metadata_location in &cleanup_candidate_locations { + self.backend.delete_object(table_bucket, metadata_location).await?; + } + + Ok(TableMetadataMaintenanceReport { + current_metadata_location: entry.metadata_location, + retained_metadata_locations: protected.into_iter().collect(), + cleanup_candidate_locations, + }) + } } #[async_trait::async_trait] @@ -863,7 +1041,11 @@ where .read_to_end(&mut data) .await .map_err(|err| TableCatalogStoreError::Internal(format!("failed to read catalog object {bucket}/{object}: {err}")))?; - Ok(Some(TableCatalogObject { data, etag: info.etag })) + Ok(Some(TableCatalogObject { + data, + etag: info.etag, + mod_time: info.mod_time, + })) } async fn object_exists(&self, bucket: &str, object: &str) -> TableCatalogStoreResult { @@ -952,6 +1134,35 @@ fn parse_table_for_store(table: &str) -> TableCatalogStoreResult BTreeSet { + let mut locations = BTreeSet::new(); + let Some(metadata_log) = current_metadata.get("metadata-log").and_then(serde_json::Value::as_array) else { + return locations; + }; + + for entry in metadata_log { + let Some(metadata_location) = entry.get("metadata-file").and_then(serde_json::Value::as_str) else { + continue; + }; + if is_valid_table_metadata_location(namespace, table, metadata_location) { + locations.insert(metadata_location.to_string()); + } + } + + locations +} + +fn metadata_candidate_is_past_safety_window(mod_time: Option, now: OffsetDateTime) -> bool { + let Some(mod_time) = mod_time else { + return false; + }; + mod_time <= now - Duration::seconds(TABLE_METADATA_CLEANUP_SAFETY_WINDOW_SECONDS) +} + fn catalog_path_hash(value: &str) -> String { let digest = Sha256::digest(value.as_bytes()); let mut output = String::with_capacity(digest.len() * 2); @@ -1743,15 +1954,21 @@ mod tests { struct TestCatalogObjectRecord { data: Vec, etag: String, + mod_time: Option, } impl TestCatalogObjectBackend { async fn seed_object(&self, bucket: &str, object: &str, data: Vec) { + self.seed_object_with_mod_time(bucket, object, data, Some(OffsetDateTime::UNIX_EPOCH)) + .await; + } + + async fn seed_object_with_mod_time(&self, bucket: &str, object: &str, data: Vec, mod_time: Option) { let mut state = self.state.lock().await; let etag = state.next_etag(); state .objects - .insert((bucket.to_string(), object.to_string()), TestCatalogObjectRecord { data, etag }); + .insert((bucket.to_string(), object.to_string()), TestCatalogObjectRecord { data, etag, mod_time }); } async fn fail_put_attempt(&self, bucket: &str, object: &str, attempt: usize) { @@ -1781,6 +1998,7 @@ mod tests { .map(|record| TableCatalogObject { data: record.data.clone(), etag: Some(record.etag.clone()), + mod_time: record.mod_time, })) } @@ -1828,7 +2046,14 @@ mod tests { } let etag = state.next_etag(); - state.objects.insert(key, TestCatalogObjectRecord { data, etag }); + state.objects.insert( + key, + TestCatalogObjectRecord { + data, + etag, + mod_time: Some(OffsetDateTime::now_utc()), + }, + ); Ok(()) } @@ -1900,6 +2125,249 @@ mod tests { } } + async fn seed_table_for_metadata_maintenance( + store: &ObjectTableCatalogStore, + bucket: &str, + namespace: &Namespace, + table: &IdentifierSegment, + current_metadata: String, + ) { + store.put_table_bucket(test_bucket_entry(bucket)).await.unwrap(); + store.create_namespace(test_namespace_entry(bucket, namespace)).await.unwrap(); + store + .create_table(test_table_entry(bucket, namespace, table, current_metadata)) + .await + .unwrap(); + } + + #[tokio::test] + async fn maintenance_dry_run_keeps_current_metadata() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").unwrap(); + let table = IdentifierSegment::parse("orders").unwrap(); + let v1 = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json"); + let v2 = default_table_metadata_file_path(&namespace, &table, "00002.metadata.json"); + let current = default_table_metadata_file_path(&namespace, &table, "00003.metadata.json"); + + seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current.clone()).await; + backend.seed_object(bucket, &v1, b"{}".to_vec()).await; + backend.seed_object(bucket, &v2, b"{}".to_vec()).await; + backend + .seed_object(bucket, ¤t, br#"{"metadata-log":[]}"#.to_vec()) + .await; + + let report = store + .plan_table_metadata_maintenance(bucket, "sales", "orders", 0) + .await + .unwrap(); + + assert_eq!(report.current_metadata_location, current); + assert!(report.retained_metadata_locations.contains(&report.current_metadata_location)); + assert!(!report.cleanup_candidate_locations.contains(&report.current_metadata_location)); + assert_eq!(report.cleanup_candidate_locations, vec![v1, v2]); + } + + #[tokio::test] + async fn maintenance_dry_run_keeps_metadata_log_references() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").unwrap(); + let table = IdentifierSegment::parse("orders").unwrap(); + let v1 = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json"); + let logged = default_table_metadata_file_path(&namespace, &table, "00002.metadata.json"); + let v3 = default_table_metadata_file_path(&namespace, &table, "00003.metadata.json"); + let current = default_table_metadata_file_path(&namespace, &table, "00004.metadata.json"); + + seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current.clone()).await; + for metadata in [&v1, &logged, &v3] { + backend.seed_object(bucket, metadata, b"{}".to_vec()).await; + } + backend + .seed_object( + bucket, + ¤t, + serde_json::to_vec(&serde_json::json!({ + "metadata-log": [ + { + "timestamp-ms": 1, + "metadata-file": logged + } + ] + })) + .unwrap(), + ) + .await; + + let report = store + .plan_table_metadata_maintenance(bucket, "sales", "orders", 0) + .await + .unwrap(); + + assert!(report.retained_metadata_locations.contains(¤t)); + assert!(report.retained_metadata_locations.contains(&logged)); + assert_eq!(report.cleanup_candidate_locations, vec![v1, v3]); + } + + #[tokio::test] + async fn maintenance_dry_run_keeps_recent_metadata_files_and_ignores_non_metadata_objects() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").unwrap(); + let table = IdentifierSegment::parse("orders").unwrap(); + let v1 = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json"); + let v2 = default_table_metadata_file_path(&namespace, &table, "00002.metadata.json"); + let recent = default_table_metadata_file_path(&namespace, &table, "00003.metadata.json"); + let current = default_table_metadata_file_path(&namespace, &table, "00004.metadata.json"); + let manifest = format!("{}/snap-1.avro", default_table_metadata_dir_path(&namespace, &table)); + + seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current.clone()).await; + for metadata in [&v1, &v2, &recent] { + backend.seed_object(bucket, metadata, b"{}".to_vec()).await; + } + backend + .seed_object(bucket, ¤t, br#"{"metadata-log":[]}"#.to_vec()) + .await; + backend.seed_object(bucket, &manifest, b"manifest".to_vec()).await; + + let report = store + .plan_table_metadata_maintenance(bucket, "sales", "orders", 2) + .await + .unwrap(); + + assert!(report.retained_metadata_locations.contains(&recent)); + assert!(report.retained_metadata_locations.contains(¤t)); + assert_eq!(report.cleanup_candidate_locations, vec![v1, v2]); + assert!(!report.cleanup_candidate_locations.contains(&manifest)); + } + + #[tokio::test] + async fn maintenance_delete_removes_only_dry_run_metadata_candidates() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").unwrap(); + let table = IdentifierSegment::parse("orders").unwrap(); + let old = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json"); + let retained = default_table_metadata_file_path(&namespace, &table, "00002.metadata.json"); + let current = default_table_metadata_file_path(&namespace, &table, "00003.metadata.json"); + let manifest = format!("{}/snap-1.avro", default_table_metadata_dir_path(&namespace, &table)); + + seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current.clone()).await; + backend.seed_object(bucket, &old, b"{}".to_vec()).await; + backend.seed_object(bucket, &retained, b"{}".to_vec()).await; + backend + .seed_object( + bucket, + ¤t, + serde_json::to_vec(&serde_json::json!({ + "metadata-log": [ + { + "timestamp-ms": 1, + "metadata-file": retained + } + ] + })) + .unwrap(), + ) + .await; + backend.seed_object(bucket, &manifest, b"manifest".to_vec()).await; + + let report = store + .delete_table_metadata_maintenance_candidates(bucket, "sales", "orders", 0) + .await + .unwrap(); + + assert_eq!(report.cleanup_candidate_locations, vec![old.clone()]); + assert!(!backend.object_exists(bucket, &old).await.unwrap()); + assert!(backend.object_exists(bucket, &retained).await.unwrap()); + assert!(backend.object_exists(bucket, ¤t).await.unwrap()); + assert!(backend.object_exists(bucket, &manifest).await.unwrap()); + } + + #[tokio::test] + async fn maintenance_delete_skips_recent_uncommitted_metadata_candidates() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").unwrap(); + let table = IdentifierSegment::parse("orders").unwrap(); + let old = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json"); + let current = default_table_metadata_file_path(&namespace, &table, "00002.metadata.json"); + let fresh = default_table_metadata_file_path(&namespace, &table, "00003.metadata.json"); + + seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current.clone()).await; + backend.seed_object(bucket, &old, b"{}".to_vec()).await; + backend + .seed_object(bucket, ¤t, br#"{"metadata-log":[]}"#.to_vec()) + .await; + backend + .seed_object_with_mod_time(bucket, &fresh, br#"{"metadata-log":[]}"#.to_vec(), Some(OffsetDateTime::now_utc())) + .await; + + let report = store + .delete_table_metadata_maintenance_candidates(bucket, "sales", "orders", 0) + .await + .unwrap(); + + assert_eq!(report.cleanup_candidate_locations, vec![old.clone()]); + assert!(!backend.object_exists(bucket, &old).await.unwrap()); + assert!(backend.object_exists(bucket, &fresh).await.unwrap()); + } + + #[tokio::test] + async fn maintenance_delete_conflicts_when_current_pointer_changes_before_delete() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").unwrap(); + let table = IdentifierSegment::parse("orders").unwrap(); + let old = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json"); + let current = default_table_metadata_file_path(&namespace, &table, "00002.metadata.json"); + let next = default_table_metadata_file_path(&namespace, &table, "00003.metadata.json"); + + seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current.clone()).await; + backend.seed_object(bucket, &old, b"{}".to_vec()).await; + backend + .seed_object(bucket, ¤t, br#"{"metadata-log":[]}"#.to_vec()) + .await; + + let report = store + .plan_table_metadata_maintenance(bucket, "sales", "orders", 0) + .await + .unwrap(); + assert_eq!(report.cleanup_candidate_locations, vec![old.clone()]); + + backend.seed_object(bucket, &next, br#"{"metadata-log":[]}"#.to_vec()).await; + store + .commit_table(TableCommitRequest { + table_bucket: bucket.to_string(), + namespace: namespace.public_name(), + table: table.as_str().to_string(), + commit_id: "commit-id".to_string(), + idempotency_key: None, + operation: "append".to_string(), + expected_version_token: "token-v1".to_string(), + expected_metadata_location: current, + new_metadata_location: next, + requirements: Vec::new(), + writer: Some("test".to_string()), + }) + .await + .unwrap(); + + let err = store + .delete_table_metadata_maintenance_report(bucket, "sales", "orders", report) + .await + .unwrap_err(); + + assert!(matches!(err, TableCatalogStoreError::Conflict(_))); + assert!(backend.object_exists(bucket, &old).await.unwrap()); + } + #[tokio::test] async fn object_table_catalog_store_commits_with_token_match_and_writes_log() { let backend = TestCatalogObjectBackend::default();