Merge pull request #745 from fabro-sh/codex/sqlite-blob-foundation

Add SQLite blob store foundation

lib/components/fabro-store/src/blob_store.rs

@@ -0,0 +1,301 @@
+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<Bytes> 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)
+    }
+}
+
+/// 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<Blob>),
+    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<slatedb::Db>) -> Self {
+        Self {
+            backend: BlobBackend::Slate(Repository::new(db)),
+        }
+    }
+
+    pub async fn write(&self, bytes: &[u8]) -> Result<BlobHash> {
+        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<u8> = 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<Option<Bytes>> {
+        match &self.backend {
+            BlobBackend::Slate(repo) => Ok(repo.get(blob_hash).await?.map(|blob| blob.0)),
+            BlobBackend::Sqlite(pool) => {
+                let stored: Option<Vec<u8>> =
+                    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<bool> {
+        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<T> = std::result::Result<T, Box<dyn std::error::Error>>;
+
+    async fn slate_store() -> Arc<BlobStore> {
+        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<slatedb::Db>, 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?);
+        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())
+            .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(())
+    }
+}

lib/components/fabro-store/src/error.rs

@@ -1,3 +1,5 @@
+use fabro_types::BlobHash;
+
 pub type Result<T> = std::result::Result<T, Error>;
 
 #[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}")]

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;
 

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

@@ -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<Bytes> 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<Blob>,
-}
-
-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<slatedb::Db>) -> Self {
-        Self {
-            repo: Repository::new(db),
-        }
-    }
-
-    pub async fn write(&self, bytes: &[u8]) -> Result<BlobHash> {
-        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<Option<Bytes>> {
-        Ok(self.repo.get(blob_hash).await?.map(|blob| blob.0))
-    }
-
-    pub async fn exists(&self, blob_hash: &BlobHash) -> Result<bool> {
-        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<BlobStore> {
-        let db = Database::new(
-            Arc::new(InMemory::new()),
-            "",
-            Duration::from_millis(1),
-            None,
-        );
-        db.blobs().await.unwrap()
-    }
-
-    async fn raw_store(name: &str) -> (Arc<slatedb::Db>, 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);
-    }
-}

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))

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;

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]*')
+);

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::<String, _>("name"), "hash");
+    assert_eq!(columns[0].get::<String, _>("type"), "TEXT");
+    assert_eq!(columns[0].get::<i64, _>("notnull"), 1);
+    assert_eq!(columns[0].get::<i64, _>("pk"), 1);
+    assert_eq!(columns[0].get::<Option<String>, _>("dflt_value"), None);
+
+    assert_eq!(columns[1].get::<String, _>("name"), "data");
+    assert_eq!(columns[1].get::<String, _>("type"), "BLOB");
+    assert_eq!(columns[1].get::<i64, _>("notnull"), 1);
+    assert_eq!(columns[1].get::<i64, _>("pk"), 0);
+    assert_eq!(columns[1].get::<Option<String>, _>("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<u8> = 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::<u8>::new())
+        .execute(database.pool())
+        .await?;
+    let stored_empty: Vec<u8> = 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::<u8>::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::<u8>::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()?;