wip: T0.7 BlobStore redb backend (in progress, salvaged before restart)

2026-08-20 14:22:46 +00:00
parent 3674ffc662
commit 20a54e42cf
7 changed files with 125 additions and 23 deletions
@@ -220,7 +220,7 @@ version = "2.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "900d271a03799a1ee8d1ca9b19893b48ca674a9284fefcfb85f05e74ed314217"
 dependencies = [
- "log 0.4.33",
+ "log",
 "regex",
 ]

@@ -234,7 +234,7 @@ dependencies = [
 "anstyle",
 "env_filter",
 "jiff",
- "log 0.4.33",
+ "log",
 ]

 [[package]]
@@ -382,7 +382,7 @@ dependencies = [
 "defmt",
 "jiff-core",
 "jiff-static",
- "log 0.4.33",
+ "log",
 "portable-atomic",
 "portable-atomic-util",
 "serde_core",
@@ -439,21 +439,6 @@ version = "0.12.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "32a66949e030da00e8c7d4434b251670a91556f4144941d37452769c25d58a53"

-[[package]]
-name = "log"
-version = "0.1.0"
-dependencies = [
- "async-trait",
- "env_logger",
- "ids",
- "log 0.4.33",
- "once_cell",
- "redb",
- "serde",
- "serde_cbor",
- "tempfile",
-]
-
 [[package]]
 name = "log"
 version = "0.4.33"
@@ -493,6 +478,21 @@ version = "0.2.17"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd"

+[[package]]
+name = "poimen-log"
+version = "0.1.0"
+dependencies = [
+ "async-trait",
+ "env_logger",
+ "ids",
+ "log",
+ "once_cell",
+ "redb",
+ "serde",
+ "serde_cbor",
+ "tempfile",
+]
+
 [[package]]
 name = "portable-atomic"
 version = "1.15.0"
@@ -762,6 +762,20 @@ dependencies = [
 "serde",
 ]

+[[package]]
+name = "storage"
+version = "0.1.0"
+dependencies = [
+ "async-trait",
+ "ids",
+ "redb",
+ "serde",
+ "serde_cbor",
+ "tempfile",
+ "thiserror",
+ "tokio",
+]
+
 [[package]]
 name = "syn"
 version = "2.0.119"
@@ -832,6 +846,15 @@ dependencies = [
 "syn 3.0.3",
 ]

+[[package]]
+name = "tokio"
+version = "1.53.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "202caea871b69668250d242070849eb495be178ed697a3e98aebce5bc81a0bed"
+dependencies = [
+ "pin-project-lite",
+]
+
 [[package]]
 name = "toml"
 version = "1.1.4+spec-1.1.0"
@@ -4,6 +4,7 @@ members = [
    "crates/ids",
    "crates/kernel",
    "crates/log",
+    "crates/storage",
 ]

 [workspace.package]
@@ -14,6 +15,8 @@ license = "MIT OR Apache-2.0"

 [workspace.dependencies]
 ids = { path = "crates/ids" }
+storage = { path = "crates/storage" }
+
 uuid = { version = "1.7", features = ["v4", "serde"] }
 ulid = { version = "1.1", features = ["serde"] }
 smol_str = { version = "0.2", features = ["serde"] }
@@ -1,5 +1,5 @@
 [package]
-name = "log"
+name = "poimen-log"
 version.workspace = true
 edition.workspace = true
 authors.workspace = true
@@ -1,5 +1,5 @@
 use ids::{BranchId, Lsn, RunId, TenantId};
-use log::{BlobRef, BranchKey, LogRecord, SchemaVersion, Timestamp, WorkEvent};
+use poimen_log::{BlobRef, BranchKey, LogRecord, SchemaVersion, Timestamp, WorkEvent};
 use std::fs;
 use std::path::Path;

@@ -136,7 +136,7 @@ fn regenerate_fixtures() {
    ];

    for (name, record) in variants {
-        let encoded = log::encode(&record).expect("encode failed");
+        let encoded = poimen_log::encode(&record).expect("encode failed");
        let path = base_path.join(format!("{}.cbor", name));
        fs::write(&path, &encoded).expect(&format!("Failed to write {}", path.display()));
        println!("Generated fixture: {}", path.display());
@@ -2,7 +2,7 @@
 //! decodes them via the wire codec, asserts non-zero schema presence, and checks
 //! every WorkEvent variant appears in at least one fixture.

-use log::{decode, encode, SchemaVersion, WorkEvent};
+use poimen_log::{decode, encode, SchemaVersion, WorkEvent};
 use std::fs;
 use std::path::{Path, PathBuf};

@@ -37,7 +37,7 @@ fn collect_files(dir: &Path) -> Vec<PathBuf> {
 }

 /// Read CBOR bytes, decode into a LogRecord. Panics with file path on failure — that is the assertion.
-fn load_fixture(path: &Path) -> log::LogRecord {
+fn load_fixture(path: &Path) -> poimen_log::LogRecord {
    let bytes = fs::read(path).unwrap_or_else(|e| panic!("read {}: {}", path.display(), e));
    decode(&bytes).unwrap_or_else(|e| panic!("decode {}: {:?}", path.display(), e))
 }
@@ -0,0 +1,18 @@
+[package]
+name = "storage"
+version.workspace = true
+edition.workspace = true
+authors.workspace = true
+license.workspace = true
+
+[dependencies]
+ids = { path = "../ids" }
+async-trait.workspace = true
+serde = { workspace = true, features = ["derive"] }
+serde_cbor = { workspace = true }
+redb.workspace = true
+thiserror = "2.0"
+
+[dev-dependencies]
+tempfile = "3"
+tokio = "1"
@@ -0,0 +1,58 @@
+// BlobStore port + redb implementation (T0.7). Content-addressed, tenant-namespaced blob storage.
+
+use async_trait::async_trait;
+use serde::{Deserialize, Serialize};
+use std::path::PathBuf;
+
+const BLOBS_TABLE: &str = "BLOBS";
+
+// ─── BlobRef (Blake3 hash newtype, spec step 1) -----------------------------
+
+/// Content-addressed blob reference — the Blake3 digest of what we stored.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
+pub struct BlobRef(pub [u8; 32]);
+
+impl BlobRef {
+    /// Construct from a blob's bytes. Hash is deterministic: `blake3::hash(content)`.
+    pub fn new(content: &[u8]) -> Self {
+        let digest = blake3::hash(content);
+        Self(*digest.as_bytes())
+    }
+
+    /// Raw 32-byte hasher output as a slice (zero allocation).
+    pub fn as_slice(&self) -> &[u8] { &self.0 }
+}
+
+// ─── Errors ------------------------------------------------------------------
+
+#[derive(Debug, thiserror::Error)]
+pub enum BlobStoreError {
+    #[error("blob not found: ref={0:?}")] NotFound(BlobRef),
+    #[error("I/O error in blob store: {0}")] Io(std::io::Error),
+    #[error("redb storage error: {0:#}")] Redb(#[from] redb::StorageError),
+    #[error("blobStore serialization failure: {0}")] Serialization(String),
+}
+type Result<T> = std::result::Result<T, BlobStoreError>;
+
+// ─── The port (public surface) -----------------------------------------------
+
+#[async_trait]
+pub trait BlobStore: Send + Sync {
+    async fn put(&self, tenant_id: ids::TenantId, content: Vec<u8>) -> Result<BlobRef>;
+    /// Returns `None` on miss. Normal outcome post reduction (T8.4).
+    async fn get(&self, tenant_id: ids::TenantId, ref_: &BlobRef) -> Result<Option<Vec<u8>>>;
+    async fn delete(&self, tenant_id: ids::TenantId, ref_: &BlobRef) -> Result<()>;
+}
+
+// ─── BlobStore implementation — redb backend ---------------------------------
+
+/// `BlobStore` over a single-process redb database.
+/// Storage key: serialize `(tenant_id, blob_ref)` with serde_cbor to raw bytes. Table value is the full body as `Vec<u8>`.
+pub struct RedBBackend { db: redb::Database }
+
+impl RedBBackend {
+    /// Open (or create) a new BlobStore file at `path`. Each test that needs isolation must pass its own tempfile path.
+    pub fn new(path: impl AsRef<std::path::Path>) -> std::io::Result<Self> {
+        let db = redb::Database::create(path.as_ref())?; // need to convert StorageError — actually it's a storage error but for our purposes just use unwrap or map...
+        Ok(Self { db })
+    }