feat(table-catalog): add metadata maintenance cleanup (#3266)

* feat(table-catalog): add metadata maintenance dry run

* feat(table-catalog): add metadata-only maintenance delete

* fix(table-catalog): protect recent metadata cleanup candidates

---------

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

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(())

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<String>,
+    pub cleanup_candidate_locations: Vec<String>,
+}
+
 #[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<u8>,
     pub etag: Option<String>,
+    pub mod_time: Option<OffsetDateTime>,
 }
 
 #[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<TableMetadataMaintenanceReport> {
+        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::<TableEntry>(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::<serde_json::Value>(&current_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(&current_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<TableMetadataMaintenanceReport> {
+        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<TableMetadataMaintenanceReport> {
+        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::<serde_json::Value>(&current_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(&current_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::<Vec<_>>();
+        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<bool> {
@@ -952,6 +1134,35 @@ fn parse_table_for_store(table: &str) -> TableCatalogStoreResult<IdentifierSegme
     IdentifierSegment::parse(table).map_err(|err| TableCatalogStoreError::Invalid(format!("invalid table name: {err}")))
 }
 
+fn metadata_log_locations(
+    current_metadata: &serde_json::Value,
+    namespace: &Namespace,
+    table: &IdentifierSegment,
+) -> BTreeSet<String> {
+    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<OffsetDateTime>, 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<u8>,
         etag: String,
+        mod_time: Option<OffsetDateTime>,
     }
 
     impl TestCatalogObjectBackend {
         async fn seed_object(&self, bucket: &str, object: &str, data: Vec<u8>) {
+            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<u8>, mod_time: Option<OffsetDateTime>) {
             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<TestCatalogObjectBackend>,
+        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, &current, 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,
+                &current,
+                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(&current));
+        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, &current, 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(&current));
+        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,
+                &current,
+                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, &current).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, &current, 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, &current, 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();