scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-15 23:32:46 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

refactor(run): remove dead legacy cleanup paths

Browse source
This commit is contained in:
Bryan Helmkamp 2026-04-03 15:40:16 -07:00
parent 6c2c4ce657
commit 374b224ac4
No known key found for this signature in database
2 changed files with 84 additions and 76 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

148
lib/crates/fabro-store/src/slate/catalog.rs
View file

@ -71,65 +71,6 @@ pub(crate) async fn list_catalogs(
Ok(records)
}
pub(super) async fn repair_catalog(store: Arc<dyn ObjectStore>, 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::<Vec<_>>()
.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::<Vec<_>>()
.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<Utc>, 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<dyn ObjectStore>, path: &Path) -> Result<bool> {
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<dyn ObjectStore>, 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<dyn ObjectStore>,
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::<Vec<_>>()
.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::<Vec<_>>()
.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<dyn ObjectStore>, path: &Path) -> Result<bool> {
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<dyn ObjectStore>, path: &Path) -> Result<()> {
match store.delete(path).await {
Ok(()) | Err(object_store::Error::NotFound { .. }) => Ok(()),
Err(err) => Err(err.into()),
}
}
}

12
lib/crates/fabro-store/src/slate/mod.rs
View file

@ -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<slatedb::Db> {
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"))));