diff --git a/lib/components/fabro-store/src/slate/blob_store.rs b/lib/components/fabro-store/src/slate/blob_store.rs index 925f01789..288755157 100644 --- a/lib/components/fabro-store/src/slate/blob_store.rs +++ b/lib/components/fabro-store/src/slate/blob_store.rs @@ -103,6 +103,16 @@ mod tests { db.blobs().await.unwrap() } + async fn raw_store(name: &str) -> (Arc, BlobStore) { + let raw_db = Arc::new( + slatedb::Db::open(name, Arc::new(InMemory::new())) + .await + .unwrap(), + ); + let store = BlobStore::new(Arc::clone(&raw_db)); + (raw_db, store) + } + #[tokio::test] async fn writes_reads_and_checks_existence() { let store = store().await; @@ -141,12 +151,7 @@ mod tests { #[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 (raw_db, store) = raw_store("blob-store-list-tests").await; let id = store.write(b"valid").await.unwrap(); raw_db @@ -162,12 +167,7 @@ mod tests { #[tokio::test] async fn raw_db_reads_exact_blob_bytes() { - let raw_db = Arc::new( - slatedb::Db::open("blob-store-tests", Arc::new(InMemory::new())) - .await - .unwrap(), - ); - let store = BlobStore::new(Arc::clone(&raw_db)); + let (raw_db, store) = raw_store("blob-store-tests").await; let bytes = b"{\"ok\":true}"; let id = store.write(bytes).await.unwrap(); diff --git a/lib/components/fabro-store/src/slate/mod.rs b/lib/components/fabro-store/src/slate/mod.rs index aed4c490e..8f9c16ac7 100644 --- a/lib/components/fabro-store/src/slate/mod.rs +++ b/lib/components/fabro-store/src/slate/mod.rs @@ -143,20 +143,23 @@ impl Database { .map(RunDatabase::from_inner) } - 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( + /// Builds a run handle wired to the Database-owned shared stores. + async fn open_run_database(&self, run_id: &RunId, read_only: bool) -> Result { + RunDatabase::build( *run_id, - db, - blob_store, + self.open_db().await?, + read_only, + self.blobs().await?, Arc::clone(&self.projection_cache), Arc::clone(&self.run_summary_store), ) - .await?; + .await + } + + pub async fn create_run(&self, run_id: &RunId) -> Result { + self.warm_projection_cache().await?; + self.catalog_index().await?.add(run_id).await?; + let run_store = self.open_run_database(run_id, false).await?; let mut active_runs = self.active_runs.lock().await; Self::cache_active_run(&mut active_runs, &run_store); Ok(run_store) @@ -165,7 +168,6 @@ 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,14 +183,7 @@ impl Database { if !RunDatabase::has_any_events(&db, run_id).await? { return Err(Error::RunNotFound(run_id.to_string())); } - let run_store = RunDatabase::open_writer( - *run_id, - db, - blob_store, - Arc::clone(&self.projection_cache), - Arc::clone(&self.run_summary_store), - ) - .await?; + let run_store = self.open_run_database(run_id, false).await?; Self::cache_active_run(&mut active_runs, &run_store); Ok(run_store) } @@ -206,15 +201,7 @@ 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), - ) - .await + self.open_run_database(run_id, true).await } pub async fn list_runs(&self, query: &ListRunsQuery, now: DateTime) -> Result> { @@ -866,6 +853,10 @@ mod tests { let blob = br#"{"summary":"readable"}"#; let blob_id = run.write_blob(blob).await.unwrap(); + // Evict the cached writer so the reader is built through the real + // `open_run_reader` construction path, not a clone of the writer. + let _ = store.remove_active_run(&test_run_id("run-1")).await; + let reader = store.open_run_reader(&test_run_id("run-1")).await.unwrap(); assert_eq!( reader.read_blob(&blob_id).await.unwrap().as_deref(), diff --git a/lib/components/fabro-store/src/slate/run_store.rs b/lib/components/fabro-store/src/slate/run_store.rs index d3999d964..5365ec30b 100644 --- a/lib/components/fabro-store/src/slate/run_store.rs +++ b/lib/components/fabro-store/src/slate/run_store.rs @@ -54,43 +54,7 @@ pub(crate) struct RunDatabaseInner { } 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 { - Self::build( - run_id, - db, - false, - blob_store, - shared_projection_cache, - run_summary_store, - ) - .await - } - - 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, - blob_store, - shared_projection_cache, - run_summary_store, - ) - .await - } - - async fn build( + pub(crate) async fn build( run_id: RunId, db: Db, read_only: bool,