From 10499e707caaf51a8da9dea5726ab6d20da3fb4b Mon Sep 17 00:00:00 2001 From: Scott Werner Date: Mon, 10 Aug 2026 15:32:47 -0400 Subject: [PATCH] Share one blob store across run handles --- lib/components/fabro-store/src/keys.rs | 50 ++---------------- .../fabro-store/src/slate/blob_store.rs | 51 ++++++++++++++++++- lib/components/fabro-store/src/slate/mod.rs | 17 +++++++ .../fabro-store/src/slate/run_store.rs | 51 +++++++++++-------- 4 files changed, 100 insertions(+), 69 deletions(-) diff --git a/lib/components/fabro-store/src/keys.rs b/lib/components/fabro-store/src/keys.rs index 343cdc1d0..13f24a246 100644 --- a/lib/components/fabro-store/src/keys.rs +++ b/lib/components/fabro-store/src/keys.rs @@ -1,7 +1,7 @@ use std::fmt::{self, Write}; use std::ops::Range; -use fabro_types::{RunBlobId, RunId, SessionId}; +use fabro_types::{RunId, SessionId}; pub(crate) const MAX_EVENT_SEQ: u32 = 999_999; @@ -91,10 +91,6 @@ pub(crate) fn run_events_range(run_id: &RunId, start_seq: u32) -> Range SlateKey { - SlateKey::new("blobs").with("sha256").into_prefix() -} - pub(crate) fn sessions_by_id_prefix() -> SlateKey { SlateKey::new("sessions").with("by-id").into_prefix() } @@ -115,21 +111,6 @@ pub(crate) fn parse_event_seq(key: &str) -> Option { segments.next()?.split_once('-')?.0.parse().ok() } -pub(crate) fn parse_blob_id(key: &str) -> Option { - let mut segments = SlateKey::segments(key); - if segments.next()? != "blobs" { - return None; - } - if segments.next()? != "sha256" { - return None; - } - let id = segments.next()?; - if segments.next().is_some() { - return None; - } - id.parse().ok() -} - #[cfg(test)] mod tests { use fabro_types::RunId; @@ -161,14 +142,6 @@ mod tests { ]); } - #[test] - fn blob_key_segments() { - let blob_id = RunBlobId::new(b"summary"); - let key = SlateKey::new("blobs").with("sha256").with(blob_id); - let segments: Vec<&str> = SlateKey::segments(key.as_str()).collect(); - assert_eq!(segments, ["blobs", "sha256", &blob_id.to_string()]); - } - #[test] fn sequence_keys_are_zero_padded() { let run_id: RunId = "01JT56VE4Z5NZ814GZN2JZD65A".parse().unwrap(); @@ -196,20 +169,16 @@ mod tests { } #[test] - fn parse_helpers_roundtrip() { + fn parse_event_seq_roundtrips() { let run_id: RunId = "01JT56VE4Z5NZ814GZN2JZD65A".parse().unwrap(); assert_eq!( parse_event_seq(run_event_key(&run_id, 7, 123).as_str()), Some(7) ); - - let blob_id = RunBlobId::new(b"summary"); - let key = SlateKey::new("blobs").with("sha256").with(blob_id); - assert_eq!(parse_blob_id(key.as_str()), Some(blob_id)); } #[test] - fn parse_helpers_reject_invalid_keys() { + fn parse_event_seq_rejects_invalid_keys() { assert_eq!( parse_event_seq( SlateKey::new("runs") @@ -220,18 +189,5 @@ mod tests { ), None ); - assert_eq!( - parse_blob_id(SlateKey::new("blobs").with("not-a-uuid").as_str()), - None - ); - assert_eq!( - parse_blob_id( - SlateKey::new("blobs") - .with("01JT56VE4Z5NZ814GZN2JZD65A") - .with("not-a-blob") - .as_str() - ), - None - ); } } diff --git a/lib/components/fabro-store/src/slate/blob_store.rs b/lib/components/fabro-store/src/slate/blob_store.rs index 3c058b3c8..925f01789 100644 --- a/lib/components/fabro-store/src/slate/blob_store.rs +++ b/lib/components/fabro-store/src/slate/blob_store.rs @@ -2,9 +2,10 @@ use std::sync::Arc; use bytes::Bytes; use fabro_types::RunBlobId; +use futures::StreamExt; -use crate::Result; use crate::record::{RawBytesCodec, Record, Repository}; +use crate::{Error, Result}; #[derive(Debug, Clone, PartialEq, Eq)] pub struct Blob(pub Bytes); @@ -63,6 +64,20 @@ impl BlobStore { pub async fn exists(&self, id: &RunBlobId) -> Result { self.repo.exists(id).await } + + pub(crate) async fn list(&self) -> Result> { + let mut stream = self.repo.scan_ids_stream(); + let mut ids = Vec::new(); + while let Some(result) = stream.next().await { + match result { + Ok(id) => ids.push(id), + Err(Error::KeyParse(_)) => {} + Err(err) => return Err(err), + } + } + ids.sort(); + Ok(ids) + } } #[cfg(test)] @@ -111,6 +126,40 @@ mod tests { assert_eq!(store.read(&id).await.unwrap(), Some(Bytes::new())); } + #[tokio::test] + async fn list_returns_sorted_ids_and_handles_empty_store() { + let store = store().await; + assert!(store.list().await.unwrap().is_empty()); + + let first_id = store.write(br#"{"z":1}"#).await.unwrap(); + let second_id = store.write(br#"{"a":1}"#).await.unwrap(); + let mut expected = vec![first_id, second_id]; + expected.sort(); + + assert_eq!(store.list().await.unwrap(), expected); + } + + #[tokio::test] + async fn list_skips_malformed_blob_ids() { + let raw_db = Arc::new( + slatedb::Db::open("blob-store-list-tests", Arc::new(InMemory::new())) + .await + .unwrap(), + ); + let store = BlobStore::new(Arc::clone(&raw_db)); + let id = store.write(b"valid").await.unwrap(); + + raw_db + .put( + SlateKey::new("blobs").with("sha256").with("not-a-blob-id"), + b"malformed", + ) + .await + .unwrap(); + + assert_eq!(store.list().await.unwrap(), vec![id]); + } + #[tokio::test] async fn raw_db_reads_exact_blob_bytes() { let raw_db = Arc::new( diff --git a/lib/components/fabro-store/src/slate/mod.rs b/lib/components/fabro-store/src/slate/mod.rs index 5d9ae3bd9..aed4c490e 100644 --- a/lib/components/fabro-store/src/slate/mod.rs +++ b/lib/components/fabro-store/src/slate/mod.rs @@ -146,11 +146,13 @@ impl Database { pub async fn create_run(&self, run_id: &RunId) -> Result { self.warm_projection_cache().await?; let db = self.open_db().await?; + let blob_store = self.blobs().await?; self.catalog_index().await?.add(run_id).await?; let run_store = RunDatabase::open_writer( *run_id, db, + blob_store, Arc::clone(&self.projection_cache), Arc::clone(&self.run_summary_store), ) @@ -163,6 +165,7 @@ impl Database { pub async fn open_run(&self, run_id: &RunId) -> Result { self.warm_projection_cache().await?; let db = self.open_db().await?; + let blob_store = self.blobs().await?; // Keep the active-writer miss and insert atomic. Otherwise concurrent // callers can create independent writers with the same recovered seq. let mut active_runs = self.active_runs.lock().await; @@ -181,6 +184,7 @@ impl Database { let run_store = RunDatabase::open_writer( *run_id, db, + blob_store, Arc::clone(&self.projection_cache), Arc::clone(&self.run_summary_store), ) @@ -202,9 +206,11 @@ impl Database { if !RunDatabase::has_any_events(&db, run_id).await? { return Err(Error::RunNotFound(run_id.to_string())); } + let blob_store = self.blobs().await?; RunDatabase::open_reader( *run_id, db, + blob_store, Arc::clone(&self.projection_cache), Arc::clone(&self.run_summary_store), ) @@ -857,8 +863,19 @@ mod tests { let (_object_store, store) = make_store(); let run = store.create_run(&test_run_id("run-1")).await.unwrap(); append_created(&run, "run-1", dt("2026-03-27T12:00:00Z")).await; + let blob = br#"{"summary":"readable"}"#; + let blob_id = run.write_blob(blob).await.unwrap(); let reader = store.open_run_reader(&test_run_id("run-1")).await.unwrap(); + assert_eq!( + reader.read_blob(&blob_id).await.unwrap().as_deref(), + Some(blob.as_slice()) + ); + assert_eq!(reader.list_blobs().await.unwrap(), vec![blob_id]); + + let err = reader.write_blob(b"blocked").await.unwrap_err(); + assert!(matches!(err, Error::ReadOnly)); + let err = reader .append_event(&event_payload( "run-1", diff --git a/lib/components/fabro-store/src/slate/run_store.rs b/lib/components/fabro-store/src/slate/run_store.rs index ced57d33d..d3999d964 100644 --- a/lib/components/fabro-store/src/slate/run_store.rs +++ b/lib/components/fabro-store/src/slate/run_store.rs @@ -37,7 +37,7 @@ impl std::fmt::Debug for RunDatabase { pub(crate) struct RunDatabaseInner { run_id: RunId, db: Db, - blob_store: BlobStore, + blob_store: Arc, // `None` for reader-built inners: readers never append, so they carry no // next-write sequence and any append through them fails as read-only. event_seq: Option, @@ -57,6 +57,7 @@ impl RunDatabase { pub(crate) async fn open_writer( run_id: RunId, db: Db, + blob_store: Arc, shared_projection_cache: Arc, run_summary_store: Arc>>, ) -> Result { @@ -64,6 +65,7 @@ impl RunDatabase { run_id, db, false, + blob_store, shared_projection_cache, run_summary_store, ) @@ -73,16 +75,26 @@ impl RunDatabase { pub(crate) async fn open_reader( run_id: RunId, db: Db, + blob_store: Arc, shared_projection_cache: Arc, run_summary_store: Arc>>, ) -> Result { - Self::build(run_id, db, true, shared_projection_cache, run_summary_store).await + Self::build( + run_id, + db, + true, + blob_store, + shared_projection_cache, + run_summary_store, + ) + .await } async fn build( run_id: RunId, db: Db, read_only: bool, + blob_store: Arc, shared_projection_cache: Arc, run_summary_store: Arc>>, ) -> Result { @@ -106,7 +118,6 @@ impl RunDatabase { Some(AtomicU32::new(next_seq)) }; let (event_tx, _) = broadcast::channel(DEFAULT_EVENT_TAIL_LIMIT.max(16)); - let blob_store = BlobStore::new(Arc::new(db.clone())); Ok(Self { inner: Arc::new(RunDatabaseInner { run_id, @@ -591,7 +602,7 @@ impl RunDatabase { } pub async fn list_blobs(&self) -> Result> { - list_blobs(&self.inner.db).await + self.inner.blob_store.list().await } pub async fn state(&self) -> Result { @@ -904,23 +915,6 @@ where Ok(events) } -async fn list_blobs(db: &R) -> Result> -where - R: DbRead + Sync, -{ - let mut iter = db.scan_prefix(keys::blobs_prefix()).await?; - let mut blob_ids = Vec::new(); - while let Some(entry) = iter.next().await? { - let key = key_to_str(&entry.key)?; - let Some(blob_id) = keys::parse_blob_id(key) else { - continue; - }; - blob_ids.push(blob_id); - } - blob_ids.sort(); - Ok(blob_ids) -} - fn key_to_str(key: &Bytes) -> Result<&str> { std::str::from_utf8(key) .map_err(|err| Error::Other(format!("stored key is not valid UTF-8: {err}"))) @@ -938,6 +932,21 @@ mod tests { use crate::{Database, Error, EventPayload, keys}; + #[tokio::test] + async fn runs_share_database_blob_store() { + let object_store = Arc::new(InMemory::new()); + let store = Database::new(object_store, "", Duration::from_millis(1), None); + let shared = store.blobs().await.unwrap(); + let first_run_id = "01JT56VE4Z5NZ814GZN2JZD65A".parse().unwrap(); + let second_run_id = "01JT56VE4Z5NZ814GZN2JZD65B".parse().unwrap(); + + let first_run = store.create_run(&first_run_id).await.unwrap(); + let second_run = store.create_run(&second_run_id).await.unwrap(); + + assert!(Arc::ptr_eq(&shared, &first_run.inner.blob_store)); + assert!(Arc::ptr_eq(&shared, &second_run.inner.blob_store)); + } + #[tokio::test] async fn list_blobs_reads_global_cas_namespace() { let object_store = Arc::new(InMemory::new());