Share one blob store across run handles

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
     run_event_seq_prefix(run_id, start_seq)..end
 }
 
-pub(crate) fn blobs_prefix() -> 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<u32> {
     segments.next()?.split_once('-')?.0.parse().ok()
 }
 
-pub(crate) fn parse_blob_id(key: &str) -> Option<RunBlobId> {
-    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
-        );
     }
 }

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<bool> {
         self.repo.exists(id).await
     }
+
+    pub(crate) async fn list(&self) -> Result<Vec<RunBlobId>> {
+        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(

lib/components/fabro-store/src/slate/mod.rs

@@ -146,11 +146,13 @@ impl Database {
     pub async fn create_run(&self, run_id: &RunId) -> Result<RunDatabase> {
         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<RunDatabase> {
         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",

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<BlobStore>,
     // `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<AtomicU32>,
@@ -57,6 +57,7 @@ impl RunDatabase {
     pub(crate) async fn open_writer(
         run_id: RunId,
         db: Db,
+        blob_store: Arc<BlobStore>,
         shared_projection_cache: Arc<RunProjectionCache>,
         run_summary_store: Arc<OnceLock<Arc<RunSummaryStore>>>,
     ) -> Result<Self> {
@@ -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<BlobStore>,
         shared_projection_cache: Arc<RunProjectionCache>,
         run_summary_store: Arc<OnceLock<Arc<RunSummaryStore>>>,
     ) -> Result<Self> {
-        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<BlobStore>,
         shared_projection_cache: Arc<RunProjectionCache>,
         run_summary_store: Arc<OnceLock<Arc<RunSummaryStore>>>,
     ) -> Result<Self> {
@@ -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<Vec<RunBlobId>> {
-        list_blobs(&self.inner.db).await
+        self.inner.blob_store.list().await
     }
 
     pub async fn state(&self) -> Result<RunProjection> {
@@ -904,23 +915,6 @@ where
     Ok(events)
 }
 
-async fn list_blobs<R>(db: &R) -> Result<Vec<RunBlobId>>
-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());