From cc163625280258038db0175ed7c98016fea7da6b Mon Sep 17 00:00:00 2001 From: Scott Werner Date: Thu, 13 Aug 2026 15:09:26 -0400 Subject: [PATCH 1/4] Add SQLite blob store foundation --- lib/components/fabro-store/src/blob_store.rs | 292 ++++++++++++++++++ lib/components/fabro-store/src/error.rs | 6 + lib/components/fabro-store/src/lib.rs | 6 +- .../fabro-store/src/slate/blob_store.rs | 137 -------- lib/components/fabro-store/src/slate/mod.rs | 6 +- .../fabro-store/src/slate/run_store.rs | 4 +- .../fabro-db/migrations/2026081301_blobs.sql | 7 + lib/foundation/fabro-db/tests/sqlite.rs | 94 ++++++ 8 files changed, 407 insertions(+), 145 deletions(-) create mode 100644 lib/components/fabro-store/src/blob_store.rs delete mode 100644 lib/components/fabro-store/src/slate/blob_store.rs create mode 100644 lib/foundation/fabro-db/migrations/2026081301_blobs.sql diff --git a/lib/components/fabro-store/src/blob_store.rs b/lib/components/fabro-store/src/blob_store.rs new file mode 100644 index 000000000..b43c94cdc --- /dev/null +++ b/lib/components/fabro-store/src/blob_store.rs @@ -0,0 +1,292 @@ +use std::sync::Arc; + +use bytes::Bytes; +use fabro_types::BlobHash; +use sqlx::SqlitePool; + +use crate::record::{RawBytesCodec, Record, Repository}; +use crate::{Error, Result}; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Blob(pub Bytes); + +impl AsRef<[u8]> for Blob { + fn as_ref(&self) -> &[u8] { + self.0.as_ref() + } +} + +impl From for Blob { + fn from(value: Bytes) -> Self { + Self(value) + } +} + +impl Record for Blob { + type Id = BlobHash; + type Codec = RawBytesCodec; + + const PREFIX: &'static str = "blobs/sha256"; + + fn id(&self) -> Self::Id { + BlobHash::new(&self.0) + } +} + +enum BlobBackend { + Slate(Repository), + Sqlite(SqlitePool), +} + +pub struct BlobStore { + backend: BlobBackend, +} + +impl std::fmt::Debug for BlobStore { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + let backend = match &self.backend { + BlobBackend::Slate(_) => "slate", + BlobBackend::Sqlite(_) => "sqlite", + }; + f.debug_struct("BlobStore") + .field("backend", &backend) + .finish_non_exhaustive() + } +} + +impl BlobStore { + /// Creates a blob store backed by a SQLite pool whose migrations have run. + #[must_use] + pub fn new(pool: SqlitePool) -> Self { + Self { + backend: BlobBackend::Sqlite(pool), + } + } + + pub(crate) fn from_slate(db: Arc) -> Self { + Self { + backend: BlobBackend::Slate(Repository::new(db)), + } + } + + pub async fn write(&self, bytes: &[u8]) -> Result { + match &self.backend { + BlobBackend::Slate(repo) => { + let blob = Blob(Bytes::copy_from_slice(bytes)); + let id = blob.id(); + repo.put(&blob).await?; + Ok(id) + } + BlobBackend::Sqlite(pool) => { + let blob_hash = BlobHash::new(bytes); + let result = sqlx::query( + "INSERT INTO blobs (hash, data) VALUES (?, ?) \ + ON CONFLICT(hash) DO NOTHING", + ) + .bind(blob_hash.to_string()) + .bind(bytes) + .execute(pool) + .await?; + + if result.rows_affected() == 1 { + return Ok(blob_hash); + } + + let stored: Vec = sqlx::query_scalar("SELECT data FROM blobs WHERE hash = ?") + .bind(blob_hash.to_string()) + .fetch_one(pool) + .await?; + if stored == bytes { + Ok(blob_hash) + } else { + Err(Error::BlobHashConflict { blob_hash }) + } + } + } + } + + pub async fn read(&self, blob_hash: &BlobHash) -> Result> { + match &self.backend { + BlobBackend::Slate(repo) => Ok(repo.get(blob_hash).await?.map(|blob| blob.0)), + BlobBackend::Sqlite(pool) => { + let stored: Option> = + sqlx::query_scalar("SELECT data FROM blobs WHERE hash = ?") + .bind(blob_hash.to_string()) + .fetch_optional(pool) + .await?; + let Some(stored) = stored else { + return Ok(None); + }; + if BlobHash::new(&stored) != *blob_hash { + return Err(Error::BlobIntegrity { + blob_hash: *blob_hash, + }); + } + Ok(Some(Bytes::from(stored))) + } + } + } + + pub async fn exists(&self, blob_hash: &BlobHash) -> Result { + match &self.backend { + BlobBackend::Slate(repo) => repo.exists(blob_hash).await, + BlobBackend::Sqlite(pool) => { + let exists: bool = + sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM blobs WHERE hash = ?)") + .bind(blob_hash.to_string()) + .fetch_one(pool) + .await?; + Ok(exists) + } + } + } + +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + use std::time::Duration; + + use bytes::Bytes; + use fabro_types::BlobHash; + use object_store::memory::InMemory; + + use super::BlobStore; + use crate::keys::SlateKey; + use crate::{Database, Error}; + + type TestResult = std::result::Result>; + + async fn slate_store() -> Arc { + let db = Database::new( + Arc::new(InMemory::new()), + "", + Duration::from_millis(1), + None, + ); + db.blobs().await.unwrap() + } + + async fn raw_slate_store(name: &str) -> (Arc, BlobStore) { + let raw_db = Arc::new( + slatedb::Db::open(name, Arc::new(InMemory::new())) + .await + .unwrap(), + ); + let store = BlobStore::from_slate(raw_db.clone()); + (raw_db, store) + } + + async fn sqlite_store() -> TestResult<(tempfile::TempDir, fabro_db::Database, BlobStore)> { + let dir = tempfile::tempdir()?; + let database = fabro_db::Database::connect(dir.path().join("fabro.sqlite3")).await?; + database.migrate().await?; + let store = BlobStore::new(database.clone_pool()); + Ok((dir, database, store)) + } + + #[tokio::test] + async fn slate_writes_reads_and_checks_existence() { + let store = slate_store().await; + let bytes = b"hello world"; + let id = store.write(bytes).await.unwrap(); + + assert_eq!( + store.read(&id).await.unwrap(), + Some(Bytes::from_static(bytes)) + ); + assert_eq!(store.write(bytes).await.unwrap(), id); + assert!(store.exists(&id).await.unwrap()); + assert!(!store.exists(&BlobHash::new(b"missing")).await.unwrap()); + } + + #[tokio::test] + async fn slate_empty_blobs_round_trip() { + let store = slate_store().await; + let id = store.write(b"").await.unwrap(); + + assert_eq!(store.read(&id).await.unwrap(), Some(Bytes::new())); + } + + #[tokio::test] + async fn raw_slate_db_reads_exact_blob_bytes() { + let (raw_db, store) = raw_slate_store("blob-store-tests").await; + let bytes = b"{\"ok\":true}"; + let id = store.write(bytes).await.unwrap(); + + let saved = raw_db + .get(SlateKey::new("blobs").with("sha256").with(id)) + .await + .unwrap() + .unwrap(); + assert_eq!(saved.as_ref(), bytes); + } + + #[tokio::test] + async fn sqlite_writes_reads_and_checks_existence() -> TestResult<()> { + let (_dir, database, store) = sqlite_store().await?; + let store = Arc::new(store); + + let binary = [0_u8, 0xff, 0x80, b'a']; + let (first_write, concurrent_write) = + tokio::join!(store.write(&binary), store.write(&binary)); + let binary_hash = first_write?; + assert_eq!(concurrent_write?, binary_hash); + let empty_hash = store.write(b"").await?; + + assert_eq!(store.write(&binary).await?, binary_hash); + assert_eq!( + store.read(&binary_hash).await?, + Some(Bytes::copy_from_slice(&binary)) + ); + assert_eq!(store.read(&empty_hash).await?, Some(Bytes::new())); + assert!(store.exists(&binary_hash).await?); + assert!(!store.exists(&BlobHash::new(b"missing")).await?); + + let row_count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM blobs") + .fetch_one(database.pool()) + .await?; + assert_eq!(row_count, 2); + Ok(()) + } + + #[tokio::test] + async fn sqlite_write_rejects_conflicting_stored_bytes() -> TestResult<()> { + let (_dir, database, store) = sqlite_store().await?; + let expected = b"expected"; + let blob_hash = BlobHash::new(expected); + sqlx::query("INSERT INTO blobs (hash, data) VALUES (?, ?)") + .bind(blob_hash.to_string()) + .bind(b"different".as_slice()) + .execute(database.pool()) + .await?; + + let error = store + .write(expected) + .await + .expect_err("conflicting bytes should fail"); + assert!( + matches!(error, Error::BlobHashConflict { blob_hash: value } if value == blob_hash) + ); + Ok(()) + } + + #[tokio::test] + async fn sqlite_read_rejects_bytes_that_do_not_match_hash() -> TestResult<()> { + let (_dir, database, store) = sqlite_store().await?; + let blob_hash = BlobHash::new(b"expected"); + sqlx::query("INSERT INTO blobs (hash, data) VALUES (?, ?)") + .bind(blob_hash.to_string()) + .bind(b"different".as_slice()) + .execute(database.pool()) + .await?; + + let error = store + .read(&blob_hash) + .await + .expect_err("mismatched stored bytes should fail"); + assert!(matches!(error, Error::BlobIntegrity { blob_hash: value } if value == blob_hash)); + Ok(()) + } +} diff --git a/lib/components/fabro-store/src/error.rs b/lib/components/fabro-store/src/error.rs index 3b6c93f5d..7cc31d9f5 100644 --- a/lib/components/fabro-store/src/error.rs +++ b/lib/components/fabro-store/src/error.rs @@ -1,3 +1,5 @@ +use fabro_types::BlobHash; + pub type Result = std::result::Result; #[derive(Debug, thiserror::Error)] @@ -10,6 +12,10 @@ pub enum Error { Serde(#[from] serde_json::Error), #[error("SQLite error: {0}")] Sqlite(#[from] sqlx::Error), + #[error("stored blob {blob_hash} has bytes that conflict with its hash")] + BlobHashConflict { blob_hash: BlobHash }, + #[error("stored blob data does not match requested hash {blob_hash}")] + BlobIntegrity { blob_hash: BlobHash }, #[error("I/O error: {0}")] Io(#[from] std::io::Error), #[error("Invalid event payload: {0}")] diff --git a/lib/components/fabro-store/src/lib.rs b/lib/components/fabro-store/src/lib.rs index 1b514a8e0..00a95d4e0 100644 --- a/lib/components/fabro-store/src/lib.rs +++ b/lib/components/fabro-store/src/lib.rs @@ -1,6 +1,7 @@ use chrono::{DateTime, Utc}; mod artifact_store; +mod blob_store; mod error; mod keyed_mutex; mod keys; @@ -18,6 +19,7 @@ pub use artifact_store::{ ArtifactKey, ArtifactStore, NodeArtifact, StageArtifactEntry, retry_storage_segment, stage_storage_segment, }; +pub use blob_store::{Blob, BlobStore}; pub use error::{Error, Result}; pub use fabro_types::{ BlobHash, EventEnvelope, PendingInterviewRecord, Run, RunProjection, StageId, StageProjection, @@ -34,8 +36,8 @@ pub use run_summary_store::{ }; pub use serializable_projection::SerializableProjection; pub use slate::{ - AuthCode, AuthCodeStore, Blob, BlobStore, CachedRunProjection, ConsumeOutcome, Database, - RefreshToken, RefreshTokenStore, RunCatalogIndex, RunDatabase, Runs, UnreadableRun, + AuthCode, AuthCodeStore, CachedRunProjection, ConsumeOutcome, Database, RefreshToken, + RefreshTokenStore, RunCatalogIndex, RunDatabase, Runs, UnreadableRun, }; pub use types::EventPayload; diff --git a/lib/components/fabro-store/src/slate/blob_store.rs b/lib/components/fabro-store/src/slate/blob_store.rs deleted file mode 100644 index 8cec4c296..000000000 --- a/lib/components/fabro-store/src/slate/blob_store.rs +++ /dev/null @@ -1,137 +0,0 @@ -use std::sync::Arc; - -use bytes::Bytes; -use fabro_types::BlobHash; - -use crate::Result; -use crate::record::{RawBytesCodec, Record, Repository}; - -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct Blob(pub Bytes); - -impl AsRef<[u8]> for Blob { - fn as_ref(&self) -> &[u8] { - self.0.as_ref() - } -} - -impl From for Blob { - fn from(value: Bytes) -> Self { - Self(value) - } -} - -impl Record for Blob { - type Id = BlobHash; - type Codec = RawBytesCodec; - - const PREFIX: &'static str = "blobs/sha256"; - - fn id(&self) -> Self::Id { - BlobHash::new(&self.0) - } -} - -pub struct BlobStore { - repo: Repository, -} - -impl std::fmt::Debug for BlobStore { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.debug_struct("BlobStore").finish_non_exhaustive() - } -} - -impl BlobStore { - pub(crate) fn new(db: Arc) -> Self { - Self { - repo: Repository::new(db), - } - } - - pub async fn write(&self, bytes: &[u8]) -> Result { - let blob = Blob(Bytes::copy_from_slice(bytes)); - let id = blob.id(); - self.repo.put(&blob).await?; - Ok(id) - } - - pub async fn read(&self, blob_hash: &BlobHash) -> Result> { - Ok(self.repo.get(blob_hash).await?.map(|blob| blob.0)) - } - - pub async fn exists(&self, blob_hash: &BlobHash) -> Result { - self.repo.exists(blob_hash).await - } -} - -#[cfg(test)] -mod tests { - use std::sync::Arc; - use std::time::Duration; - - use bytes::Bytes; - use fabro_types::BlobHash; - use object_store::memory::InMemory; - - use super::BlobStore; - use crate::Database; - use crate::keys::SlateKey; - - async fn store() -> Arc { - let db = Database::new( - Arc::new(InMemory::new()), - "", - Duration::from_millis(1), - None, - ); - 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; - let bytes = b"hello world"; - let id = store.write(bytes).await.unwrap(); - - assert_eq!( - store.read(&id).await.unwrap(), - Some(Bytes::from_static(bytes)) - ); - assert_eq!(store.write(bytes).await.unwrap(), id); - assert!(store.exists(&id).await.unwrap()); - assert!(!store.exists(&BlobHash::new(b"missing")).await.unwrap()); - } - - #[tokio::test] - async fn empty_blobs_round_trip() { - let store = store().await; - let id = store.write(b"").await.unwrap(); - - assert_eq!(store.read(&id).await.unwrap(), Some(Bytes::new())); - } - - #[tokio::test] - async fn raw_db_reads_exact_blob_bytes() { - let (raw_db, store) = raw_store("blob-store-tests").await; - let bytes = b"{\"ok\":true}"; - let id = store.write(bytes).await.unwrap(); - - let saved = raw_db - .get(SlateKey::new("blobs").with("sha256").with(id)) - .await - .unwrap() - .unwrap(); - assert_eq!(saved.as_ref(), bytes); - } -} diff --git a/lib/components/fabro-store/src/slate/mod.rs b/lib/components/fabro-store/src/slate/mod.rs index e0b66d6e8..efed28796 100644 --- a/lib/components/fabro-store/src/slate/mod.rs +++ b/lib/components/fabro-store/src/slate/mod.rs @@ -1,6 +1,5 @@ mod auth_codes; mod auth_tokens; -mod blob_store; mod projection_cache; mod run_catalog_index; mod run_store; @@ -12,7 +11,6 @@ use std::time::Duration; pub use auth_codes::{AuthCode, AuthCodeStore}; pub use auth_tokens::{ConsumeOutcome, RefreshToken, RefreshTokenStore}; -pub use blob_store::{Blob, BlobStore}; use chrono::{DateTime, Utc}; use fabro_types::{Run, RunId, SessionId}; use object_store::ObjectStore; @@ -25,7 +23,7 @@ use slatedb::config::{CompressionCodec, Settings}; use tokio::sync::{Mutex, OnceCell}; use tracing::warn; -use crate::{Error, ListRunsQuery, Result, RunProjection, RunSummaryStore, keys}; +use crate::{BlobStore, Error, ListRunsQuery, Result, RunProjection, RunSummaryStore, keys}; #[derive(Debug, Clone, PartialEq, Eq)] pub struct UnreadableRun { @@ -449,7 +447,7 @@ impl Database { .blobs .get_or_try_init(|| async { let db = Arc::new(self.open_db().await?); - Ok::<_, Error>(Arc::new(BlobStore::new(db))) + Ok::<_, Error>(Arc::new(BlobStore::from_slate(db))) }) .await?; Ok(Arc::clone(store)) diff --git a/lib/components/fabro-store/src/slate/run_store.rs b/lib/components/fabro-store/src/slate/run_store.rs index c4c590a0d..81450b44a 100644 --- a/lib/components/fabro-store/src/slate/run_store.rs +++ b/lib/components/fabro-store/src/slate/run_store.rs @@ -11,11 +11,11 @@ use tokio::sync::{Mutex, broadcast, mpsc}; use tokio_stream::wrappers::UnboundedReceiverStream; use tracing::warn; -use super::blob_store::BlobStore; use super::projection_cache::{CachedRunProjection, RunProjectionCache}; use crate::run_state::{EventProjectionCache, RunProjectionReducer}; use crate::{ - Error, EventEnvelope, EventPayload, Result, RunProjection, RunSummaryStore, StageId, keys, + BlobStore, Error, EventEnvelope, EventPayload, Result, RunProjection, RunSummaryStore, StageId, + keys, }; const DEFAULT_EVENT_TAIL_LIMIT: usize = 1024; diff --git a/lib/foundation/fabro-db/migrations/2026081301_blobs.sql b/lib/foundation/fabro-db/migrations/2026081301_blobs.sql new file mode 100644 index 000000000..795e705a4 --- /dev/null +++ b/lib/foundation/fabro-db/migrations/2026081301_blobs.sql @@ -0,0 +1,7 @@ +CREATE TABLE blobs ( + hash TEXT PRIMARY KEY NOT NULL, + data BLOB NOT NULL, + CHECK (length(hash) = 64), + CHECK (hash = lower(hash)), + CHECK (hash NOT GLOB '*[^0-9a-f]*') +); diff --git a/lib/foundation/fabro-db/tests/sqlite.rs b/lib/foundation/fabro-db/tests/sqlite.rs index bc4981b06..bdb1179e4 100644 --- a/lib/foundation/fabro-db/tests/sqlite.rs +++ b/lib/foundation/fabro-db/tests/sqlite.rs @@ -73,6 +73,13 @@ async fn connect_creates_parent_directory_and_migrate_is_idempotent() -> anyhow: .await?; assert_eq!(runs_table_count, 1); + let blobs_table_count: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = 'blobs'", + ) + .fetch_one(database.pool()) + .await?; + assert_eq!(blobs_table_count, 1); + let legacy_import_table_count: i64 = sqlx::query_scalar( "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = 'legacy_imports'", ) @@ -89,6 +96,93 @@ async fn connect_creates_parent_directory_and_migrate_is_idempotent() -> anyhow: Ok(()) } +#[tokio::test] +async fn blobs_schema_enforces_canonical_hashes_and_required_data() -> anyhow::Result<()> { + let dir = tempfile::tempdir()?; + let database = fabro_db::Database::connect(dir.path().join("fabro.sqlite3")).await?; + database.migrate().await?; + + let columns = sqlx::query("PRAGMA table_info(blobs)") + .fetch_all(database.pool()) + .await?; + assert_eq!(columns.len(), 2); + + assert_eq!(columns[0].get::("name"), "hash"); + assert_eq!(columns[0].get::("type"), "TEXT"); + assert_eq!(columns[0].get::("notnull"), 1); + assert_eq!(columns[0].get::("pk"), 1); + assert_eq!(columns[0].get::, _>("dflt_value"), None); + + assert_eq!(columns[1].get::("name"), "data"); + assert_eq!(columns[1].get::("type"), "BLOB"); + assert_eq!(columns[1].get::("notnull"), 1); + assert_eq!(columns[1].get::("pk"), 0); + assert_eq!(columns[1].get::, _>("dflt_value"), None); + + let binary_hash = "0".repeat(64); + let binary_data = vec![0, 0xff, 0x80, b'a']; + sqlx::query("INSERT INTO blobs (hash, data) VALUES (?, ?)") + .bind(&binary_hash) + .bind(&binary_data) + .execute(database.pool()) + .await?; + let stored_binary: Vec = sqlx::query_scalar("SELECT data FROM blobs WHERE hash = ?") + .bind(&binary_hash) + .fetch_one(database.pool()) + .await?; + assert_eq!(stored_binary, binary_data); + + let empty_hash = "1".repeat(64); + sqlx::query("INSERT INTO blobs (hash, data) VALUES (?, ?)") + .bind(&empty_hash) + .bind(Vec::::new()) + .execute(database.pool()) + .await?; + let stored_empty: Vec = sqlx::query_scalar("SELECT data FROM blobs WHERE hash = ?") + .bind(&empty_hash) + .fetch_one(database.pool()) + .await?; + assert!(stored_empty.is_empty()); + + for invalid_hash in [ + "a".repeat(63), + "a".repeat(65), + "A".repeat(64), + "g".repeat(64), + ] { + let result = sqlx::query("INSERT INTO blobs (hash, data) VALUES (?, ?)") + .bind(&invalid_hash) + .bind(Vec::::new()) + .execute(database.pool()) + .await; + assert!( + result.is_err(), + "invalid blob hash should be rejected: {invalid_hash:?}" + ); + } + + let null_hash = sqlx::query("INSERT INTO blobs (hash, data) VALUES (NULL, ?)") + .bind(Vec::::new()) + .execute(database.pool()) + .await; + assert!(null_hash.is_err()); + + let null_data = sqlx::query("INSERT INTO blobs (hash, data) VALUES (?, NULL)") + .bind("2".repeat(64)) + .execute(database.pool()) + .await; + assert!(null_data.is_err()); + + let duplicate_hash = sqlx::query("INSERT INTO blobs (hash, data) VALUES (?, ?)") + .bind(&binary_hash) + .bind(vec![1_u8]) + .execute(database.pool()) + .await; + assert!(duplicate_hash.is_err()); + + Ok(()) +} + #[tokio::test] async fn mcp_servers_schema_rejects_invalid_transport_rows() -> anyhow::Result<()> { let dir = tempfile::tempdir()?; From 01efe7c883dc813a837d45b1dcd436b531fa3964 Mon Sep 17 00:00:00 2001 From: Scott Werner Date: Sun, 16 Aug 2026 10:21:45 -0400 Subject: [PATCH 2/4] Document BlobBackend as a transitional enum Mark the Slate arm as temporary and record that the SQLite arm's verified-read and hash-conflict semantics are the intended end state, so the dual-backend enum reads as a rollout vehicle rather than a permanent abstraction. Co-Authored-By: Claude Fable 5 --- lib/components/fabro-store/src/blob_store.rs | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/lib/components/fabro-store/src/blob_store.rs b/lib/components/fabro-store/src/blob_store.rs index b43c94cdc..53d7bd75b 100644 --- a/lib/components/fabro-store/src/blob_store.rs +++ b/lib/components/fabro-store/src/blob_store.rs @@ -33,6 +33,14 @@ impl Record for Blob { } } +/// Which storage engine holds the blobs. +/// +/// This enum is a transition vehicle, not a permanent abstraction: `Slate` +/// preserves current production behavior while the SQLite backend rolls out. +/// Once runtime blob storage switches to SQLite and legacy blobs are +/// imported, delete the `Slate` arm (and this enum) and inline the SQLite +/// implementation into [`BlobStore`]. The SQLite arm's semantics — verified +/// reads and loud failure on hash conflicts — are the intended end state. enum BlobBackend { Slate(Repository), Sqlite(SqlitePool), @@ -140,7 +148,6 @@ impl BlobStore { } } } - } #[cfg(test)] From 9a03b813b20a8acfa64aea1ed685b12753596e9e Mon Sep 17 00:00:00 2001 From: Scott Werner Date: Mon, 17 Aug 2026 16:16:18 -0400 Subject: [PATCH 3/4] Trigger CI From 7b47ef2d0536f21c7418a7e4edd7736f6af4f9d2 Mon Sep 17 00:00:00 2001 From: Scott Werner Date: Tue, 18 Aug 2026 17:43:22 -0400 Subject: [PATCH 4/4] Cover missing SQLite blob reads --- lib/components/fabro-store/src/blob_store.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/lib/components/fabro-store/src/blob_store.rs b/lib/components/fabro-store/src/blob_store.rs index 53d7bd75b..6d19ed978 100644 --- a/lib/components/fabro-store/src/blob_store.rs +++ b/lib/components/fabro-store/src/blob_store.rs @@ -249,7 +249,9 @@ mod tests { ); assert_eq!(store.read(&empty_hash).await?, Some(Bytes::new())); assert!(store.exists(&binary_hash).await?); - assert!(!store.exists(&BlobHash::new(b"missing")).await?); + let missing_hash = BlobHash::new(b"missing"); + assert_eq!(store.read(&missing_hash).await?, None); + assert!(!store.exists(&missing_hash).await?); let row_count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM blobs") .fetch_one(database.pool())