From 374b224ac4f2165df139a94dfb5b62ea64cd5598 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Fri, 3 Apr 2026 15:40:16 -0700 Subject: [PATCH] refactor(run): remove dead legacy cleanup paths --- lib/crates/fabro-store/src/slate/catalog.rs | 148 +++++++++++--------- lib/crates/fabro-store/src/slate/mod.rs | 12 +- 2 files changed, 84 insertions(+), 76 deletions(-) diff --git a/lib/crates/fabro-store/src/slate/catalog.rs b/lib/crates/fabro-store/src/slate/catalog.rs index 436defc53..23df4f4de 100644 --- a/lib/crates/fabro-store/src/slate/catalog.rs +++ b/lib/crates/fabro-store/src/slate/catalog.rs @@ -71,65 +71,6 @@ pub(crate) async fn list_catalogs( Ok(records) } -pub(super) async fn repair_catalog(store: Arc, base_prefix: &str) -> Result<()> { - let by_id_prefix = Path::from(format!("{base_prefix}by-id")); - let by_start_prefix = Path::from(format!("{base_prefix}by-start")); - - let by_id_metas = store - .list(Some(&by_id_prefix)) - .try_collect::>() - .await?; - let mut canonical = HashMap::new(); - for meta in by_id_metas { - if let Some(record) = read_catalog_path(store.clone(), meta.location).await? { - canonical.insert(record.run_id, record); - } - } - - for record in canonical.values() { - let path = by_start_path(base_prefix, record.created_at, &record.run_id); - if !object_exists(store.clone(), &path).await? { - store.put(&path, serde_json::to_vec(record)?.into()).await?; - } - } - - let by_start_metas = store - .list(Some(&by_start_prefix)) - .try_collect::>() - .await?; - let mut seen = HashSet::new(); - for meta in by_start_metas { - let location = meta.location.clone(); - let Some(record) = read_catalog_path(store.clone(), location.clone()).await? else { - delete_if_exists(store.clone(), &location).await?; - continue; - }; - let expected = canonical.get(&record.run_id).map(|canonical_record| { - by_start_path(base_prefix, canonical_record.created_at, &record.run_id) - }); - match expected { - Some(expected) if expected == location => { - seen.insert(record.run_id); - } - _ => { - delete_if_exists(store.clone(), &location).await?; - } - } - } - - for record in canonical.values() { - if !seen.contains(&record.run_id) { - store - .put( - &by_start_path(base_prefix, record.created_at, &record.run_id), - serde_json::to_vec(record)?.into(), - ) - .await?; - } - } - Ok(()) -} - pub(crate) fn db_prefix(base_prefix: &str, created_at: DateTime, run_id: &RunId) -> String { format!( "{base_prefix}db/{}/{run_id}/", @@ -159,17 +100,84 @@ pub(crate) async fn read_catalog_path( } } -async fn object_exists(store: Arc, path: &Path) -> Result { - match store.head(path).await { - Ok(_) => Ok(true), - Err(object_store::Error::NotFound { .. }) => Ok(false), - Err(err) => Err(err.into()), - } -} +#[cfg(test)] +pub(super) mod test_support { + use super::*; -async fn delete_if_exists(store: Arc, path: &Path) -> Result<()> { - match store.delete(path).await { - Ok(()) | Err(object_store::Error::NotFound { .. }) => Ok(()), - Err(err) => Err(err.into()), + pub(crate) async fn repair_catalog( + store: Arc, + base_prefix: &str, + ) -> Result<()> { + let by_id_prefix = Path::from(format!("{base_prefix}by-id")); + let by_start_prefix = Path::from(format!("{base_prefix}by-start")); + + let by_id_metas = store + .list(Some(&by_id_prefix)) + .try_collect::>() + .await?; + let mut canonical = HashMap::new(); + for meta in by_id_metas { + if let Some(record) = read_catalog_path(store.clone(), meta.location).await? { + canonical.insert(record.run_id, record); + } + } + + for record in canonical.values() { + let path = by_start_path(base_prefix, record.created_at, &record.run_id); + if !object_exists(store.clone(), &path).await? { + store.put(&path, serde_json::to_vec(record)?.into()).await?; + } + } + + let by_start_metas = store + .list(Some(&by_start_prefix)) + .try_collect::>() + .await?; + let mut seen = HashSet::new(); + for meta in by_start_metas { + let location = meta.location.clone(); + let Some(record) = read_catalog_path(store.clone(), location.clone()).await? else { + delete_if_exists(store.clone(), &location).await?; + continue; + }; + let expected = canonical.get(&record.run_id).map(|canonical_record| { + by_start_path(base_prefix, canonical_record.created_at, &record.run_id) + }); + match expected { + Some(expected) if expected == location => { + seen.insert(record.run_id); + } + _ => { + delete_if_exists(store.clone(), &location).await?; + } + } + } + + for record in canonical.values() { + if !seen.contains(&record.run_id) { + store + .put( + &by_start_path(base_prefix, record.created_at, &record.run_id), + serde_json::to_vec(record)?.into(), + ) + .await?; + } + } + Ok(()) + } + + async fn object_exists(store: Arc, path: &Path) -> Result { + match store.head(path).await { + Ok(_) => Ok(true), + Err(object_store::Error::NotFound { .. }) => Ok(false), + Err(err) => Err(err.into()), + } + } + + async fn delete_if_exists(store: Arc, path: &Path) -> Result<()> { + match store.delete(path).await { + Ok(()) | Err(object_store::Error::NotFound { .. }) => Ok(()), + Err(err) => Err(err.into()), + } } } diff --git a/lib/crates/fabro-store/src/slate/mod.rs b/lib/crates/fabro-store/src/slate/mod.rs index 4eab04012..4c6d9f1e3 100644 --- a/lib/crates/fabro-store/src/slate/mod.rs +++ b/lib/crates/fabro-store/src/slate/mod.rs @@ -50,10 +50,6 @@ impl SlateStore { } } - pub async fn repair_catalog(&self) -> Result<()> { - catalog::repair_catalog(self.object_store.clone(), &self.base_prefix).await - } - async fn open_db(&self, db_prefix: &str) -> Result { Ok( slatedb::Db::builder(db_prefix.to_string(), self.object_store.clone()) @@ -408,6 +404,10 @@ mod tests { (object_store, store) } + async fn repair_catalog_for_tests(store: &SlateStore) -> Result<()> { + catalog::test_support::repair_catalog(store.object_store.clone(), &store.base_prefix).await + } + fn test_run_id(label: &str) -> RunId { match label { "run-1" => fixtures::RUN_1, @@ -725,7 +725,7 @@ mod tests { .is_empty() ); - store.repair_catalog().await.unwrap(); + repair_catalog_for_tests(&store).await.unwrap(); let listed = store.list_runs(&ListRunsQuery::default()).await.unwrap(); assert_eq!(listed.len(), 1); assert!( @@ -1041,7 +1041,7 @@ mod tests { .await .unwrap(); - store.repair_catalog().await.unwrap(); + repair_catalog_for_tests(&store).await.unwrap(); let paths = list_paths(object_store, "runs/by-start").await; assert_eq!(paths.len(), 1); assert!(paths[0].contains(&format!("2026-03-27-12-00/{}.json", test_run_id("run-1"))));