From 787dd8eee32e700771ba42ad43202bfd5a03fa9c Mon Sep 17 00:00:00 2001 From: James Ross Date: Sat, 27 Jun 2026 20:05:45 -0700 Subject: [PATCH] feat(echo-cas): add durable disk tier --- CHANGELOG.md | 3 + crates/echo-cas/README.md | 6 + crates/echo-cas/src/disk.rs | 281 +++++++++++++++++++++++++++++ crates/echo-cas/src/lib.rs | 7 +- crates/echo-cas/src/memory.rs | 6 +- crates/echo-cas/tests/disk_tier.rs | 130 +++++++++++++ 6 files changed, 428 insertions(+), 5 deletions(-) create mode 100644 crates/echo-cas/src/disk.rs create mode 100644 crates/echo-cas/tests/disk_tier.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index c30ea40a..f913743c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -55,6 +55,9 @@ bundle envelope metadata, verify self-contained bundles without the original WAL root, and report unavailable ref-only segment bytes as typed material obstructions in JSON output. +- `echo-cas` now exposes a fallible filesystem-backed `DiskTier` for durable + retained blobs, preserving content-only BLAKE3 hash semantics across process + reconstruction while keeping missing blobs as explicit absence. - `cargo xtask test-slice durable-runtime-wal` now runs the release-grade filesystem runtime WAL durability gate, joining filesystem ACK recovery, filesystem failure atomicity, CLI submission posture JSON, stale-claim, and diff --git a/crates/echo-cas/README.md b/crates/echo-cas/README.md index dfc8adb2..5db4106d 100644 --- a/crates/echo-cas/README.md +++ b/crates/echo-cas/README.md @@ -4,3 +4,9 @@ # echo-cas Content-addressed blob store for Echo. + +`MemoryTier` provides infallible in-process content-addressed storage. +`DiskTier` provides a fallible filesystem-backed retained blob tier for material +that must survive process reconstruction while preserving content-only BLAKE3 +hash semantics. Because filesystem writes can fail, `DiskTier` exposes fallible +methods directly instead of hiding I/O behind the infallible `BlobStore` trait. diff --git a/crates/echo-cas/src/disk.rs b/crates/echo-cas/src/disk.rs new file mode 100644 index 00000000..1631c2ce --- /dev/null +++ b/crates/echo-cas/src/disk.rs @@ -0,0 +1,281 @@ +// SPDX-License-Identifier: Apache-2.0 +// © James Ross Ω FLYING•ROBOTS +//! Filesystem-backed content-addressed blob tier. + +use std::collections::HashSet; +use std::fs; +use std::io::{self, ErrorKind}; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::Arc; + +use thiserror::Error; + +use crate::{blob_hash, BlobHash, CasError}; + +static TEMP_COUNTER: AtomicU64 = AtomicU64::new(0); + +/// Filesystem-backed content-addressed blob tier. +/// +/// `DiskTier` persists blob bytes under their content-only BLAKE3 hash. It is +/// intentionally fallible instead of implementing [`crate::BlobStore`], whose +/// current `put`/`get` surface has no way to report filesystem failures. +pub struct DiskTier { + root: PathBuf, + blobs_dir: PathBuf, + pins: HashSet, +} + +impl DiskTier { + /// Opens or creates a disk tier rooted at `root`. + /// + /// # Errors + /// + /// Returns [`DiskTierError::Io`] when the tier directories cannot be + /// created. + pub fn open(root: impl AsRef) -> Result { + let root = root.as_ref().to_path_buf(); + let blobs_dir = root.join("blobs"); + fs::create_dir_all(&blobs_dir) + .map_err(|source| DiskTierError::io("create_dir_all", &blobs_dir, source))?; + Ok(Self { + root, + blobs_dir, + pins: HashSet::new(), + }) + } + + /// Returns the tier root directory. + #[must_use] + pub fn root(&self) -> &Path { + &self.root + } + + /// Computes the content hash and stores `bytes`. + /// + /// # Errors + /// + /// Returns [`DiskTierError::Io`] when the blob cannot be written. + pub fn put(&self, bytes: &[u8]) -> Result { + let hash = blob_hash(bytes); + self.put_verified(hash, bytes)?; + Ok(hash) + } + + /// Stores `bytes` only if they match `expected`. + /// + /// # Errors + /// + /// Returns [`DiskTierError::Cas`] when `bytes` hash to a different + /// [`BlobHash`], without mutating the tier. Returns [`DiskTierError::Io`] + /// when verified bytes cannot be written. + pub fn put_verified(&self, expected: BlobHash, bytes: &[u8]) -> Result<(), DiskTierError> { + let computed = blob_hash(bytes); + if computed != expected { + return Err(CasError::HashMismatch { expected, computed }.into()); + } + let path = self.blob_path(&expected); + let parent = path + .parent() + .ok_or_else(|| DiskTierError::invalid_blob_path(path.clone()))?; + fs::create_dir_all(parent) + .map_err(|source| DiskTierError::io("create_dir_all", parent, source))?; + let temp_path = Self::temp_path(parent, &expected); + fs::write(&temp_path, bytes) + .map_err(|source| DiskTierError::io("write", &temp_path, source))?; + fs::rename(&temp_path, &path).map_err(|source| { + let _ = fs::remove_file(&temp_path); + DiskTierError::io("rename", &path, source) + })?; + Ok(()) + } + + /// Retrieves blob bytes by content hash. + /// + /// Missing blobs return `Ok(None)`. + /// + /// # Errors + /// + /// Returns [`DiskTierError::Io`] for filesystem errors other than absence + /// and [`DiskTierError::Cas`] when stored bytes no longer match their path + /// hash. + pub fn get(&self, hash: &BlobHash) -> Result>, DiskTierError> { + let path = self.blob_path(hash); + let bytes = match fs::read(&path) { + Ok(bytes) => bytes, + Err(error) if error.kind() == ErrorKind::NotFound => return Ok(None), + Err(source) => return Err(DiskTierError::io("read", &path, source)), + }; + let computed = blob_hash(&bytes); + if computed != *hash { + return Err(CasError::HashMismatch { + expected: *hash, + computed, + } + .into()); + } + Ok(Some(Arc::from(bytes))) + } + + /// Returns whether a blob exists without reading its bytes. + /// + /// # Errors + /// + /// Returns [`DiskTierError::Io`] when metadata lookup fails for reasons + /// other than absence. + pub fn has(&self, hash: &BlobHash) -> Result { + let path = self.blob_path(hash); + match fs::metadata(&path) { + Ok(metadata) => Ok(metadata.is_file()), + Err(error) if error.kind() == ErrorKind::NotFound => Ok(false), + Err(source) => Err(DiskTierError::io("metadata", &path, source)), + } + } + + /// Returns all stored blob hashes in sorted order. + /// + /// # Errors + /// + /// Returns [`DiskTierError::Io`] when the blob directory cannot be read. + pub fn list(&self) -> Result, DiskTierError> { + let mut hashes = Vec::new(); + let entries = match fs::read_dir(&self.blobs_dir) { + Ok(entries) => entries, + Err(error) if error.kind() == ErrorKind::NotFound => return Ok(hashes), + Err(source) => return Err(DiskTierError::io("read_dir", &self.blobs_dir, source)), + }; + for shard in entries { + let shard = shard + .map_err(|source| DiskTierError::io("read_dir_entry", &self.blobs_dir, source))?; + let shard_path = shard.path(); + if !shard_path.is_dir() { + continue; + } + let blob_entries = fs::read_dir(&shard_path) + .map_err(|source| DiskTierError::io("read_dir", &shard_path, source))?; + for blob in blob_entries { + let blob = blob + .map_err(|source| DiskTierError::io("read_dir_entry", &shard_path, source))?; + let path = blob.path(); + if !path.is_file() { + continue; + } + let Some(file_name) = path.file_name().and_then(|name| name.to_str()) else { + return Err(DiskTierError::invalid_blob_path(path)); + }; + if file_name.starts_with('.') { + continue; + } + let Some(hash) = blob_hash_from_hex(file_name) else { + return Err(DiskTierError::invalid_blob_path(path)); + }; + hashes.push(hash); + } + } + hashes.sort_unstable(); + Ok(hashes) + } + + /// Marks a hash as pinned in this process. + pub fn pin(&mut self, hash: &BlobHash) { + self.pins.insert(*hash); + } + + /// Removes a process-local pin. + pub fn unpin(&mut self, hash: &BlobHash) { + self.pins.remove(hash); + } + + /// Returns whether the hash is pinned in this process. + #[must_use] + pub fn is_pinned(&self, hash: &BlobHash) -> bool { + self.pins.contains(hash) + } + + /// Number of process-local pins. + #[must_use] + pub fn pinned_count(&self) -> usize { + self.pins.len() + } + + fn blob_path(&self, hash: &BlobHash) -> PathBuf { + let hex = blob_hash_hex(hash); + self.blobs_dir.join(&hex[..2]).join(hex) + } + + fn temp_path(parent: &Path, hash: &BlobHash) -> PathBuf { + let counter = TEMP_COUNTER.fetch_add(1, Ordering::Relaxed); + parent.join(format!(".{}.{}.tmp", blob_hash_hex(hash), counter)) + } +} + +/// Errors returned by [`DiskTier`]. +#[derive(Debug, Error)] +pub enum DiskTierError { + /// Content hash verification failed. + #[error(transparent)] + Cas(#[from] CasError), + /// Filesystem operation failed. + #[error("[CAS_IO] {operation} {path}: {source}")] + Io { + /// Filesystem operation being performed. + operation: &'static str, + /// Path associated with the failure. + path: PathBuf, + /// Source I/O error. + #[source] + source: io::Error, + }, + /// Blob path did not match the tier layout. + #[error("[CAS_INVALID_BLOB_PATH] {path}")] + InvalidBlobPath { + /// Invalid path. + path: PathBuf, + }, +} + +impl DiskTierError { + fn io(operation: &'static str, path: &Path, source: io::Error) -> Self { + Self::Io { + operation, + path: path.to_path_buf(), + source, + } + } + + fn invalid_blob_path(path: PathBuf) -> Self { + Self::InvalidBlobPath { path } + } +} + +fn blob_hash_hex(hash: &BlobHash) -> String { + const HEX: &[u8; 16] = b"0123456789abcdef"; + let mut out = String::with_capacity(64); + for byte in hash.as_bytes() { + out.push(char::from(HEX[usize::from(byte >> 4)])); + out.push(char::from(HEX[usize::from(byte & 0x0f)])); + } + out +} + +fn blob_hash_from_hex(input: &str) -> Option { + if input.len() != 64 { + return None; + } + let mut bytes = [0_u8; 32]; + for (index, chunk) in input.as_bytes().chunks_exact(2).enumerate() { + let high = hex_nibble(chunk[0])?; + let low = hex_nibble(chunk[1])?; + bytes[index] = (high << 4) | low; + } + Some(BlobHash::from_bytes(bytes)) +} + +fn hex_nibble(byte: u8) -> Option { + match byte { + b'0'..=b'9' => Some(byte - b'0'), + b'a'..=b'f' => Some(byte - b'a' + 10), + b'A'..=b'F' => Some(byte - b'A' + 10), + _ => None, + } +} diff --git a/crates/echo-cas/src/lib.rs b/crates/echo-cas/src/lib.rs index 425a018b..8c67e448 100644 --- a/crates/echo-cas/src/lib.rs +++ b/crates/echo-cas/src/lib.rs @@ -3,8 +3,9 @@ //! Content-addressed blob store for Echo. //! //! `echo-cas` provides a [`BlobStore`] trait for content-addressed storage keyed by -//! BLAKE3 hash. Phase 1 ships [`MemoryTier`] — sufficient for the in-browser website -//! demo. Disk/cold tiers, wire protocol, and GC come in Phase 3. +//! BLAKE3 hash. [`MemoryTier`] is the infallible in-process implementation. +//! [`DiskTier`] is the fallible filesystem-backed retained blob tier. Cold tiers, +//! wire protocol, and GC come in later phases. //! //! # Hash Domain Policy //! @@ -21,8 +22,10 @@ //! return results sorted by [`BlobHash`]. #![forbid(unsafe_code)] +mod disk; mod memory; mod retention; +pub use disk::{DiskTier, DiskTierError}; pub use memory::MemoryTier; pub use retention::{ RetainedBlob, RetainedBlobDescriptor, RetainedBlobIndex, RetainedBlobRange, RetainedBlobRole, diff --git a/crates/echo-cas/src/memory.rs b/crates/echo-cas/src/memory.rs index 63dd7b86..de003d04 100644 --- a/crates/echo-cas/src/memory.rs +++ b/crates/echo-cas/src/memory.rs @@ -2,9 +2,9 @@ // © James Ross Ω FLYING•ROBOTS //! In-memory content-addressed blob store. //! -//! [`MemoryTier`] is the Phase 1 `BlobStore` implementation — sufficient for the -//! in-browser website demo (single tab, no persistence). Disk and cold tiers are -//! deferred to Phase 3. +//! [`MemoryTier`] is the infallible `BlobStore` implementation for in-process +//! use. Use [`crate::DiskTier`] when retained bytes must survive process +//! reconstruction. use std::collections::{HashMap, HashSet}; use std::sync::Arc; diff --git a/crates/echo-cas/tests/disk_tier.rs b/crates/echo-cas/tests/disk_tier.rs new file mode 100644 index 00000000..387bfaf9 --- /dev/null +++ b/crates/echo-cas/tests/disk_tier.rs @@ -0,0 +1,130 @@ +// SPDX-License-Identifier: Apache-2.0 +// © James Ross Ω FLYING•ROBOTS +//! Filesystem-backed retained blob tier tests. + +use std::fs; +use std::io::{self, ErrorKind}; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicU64, Ordering}; + +use echo_cas::{BlobHash, CasError, DiskTier, DiskTierError}; + +type TestResult = Result<(), Box>; + +static TEMP_COUNTER: AtomicU64 = AtomicU64::new(0); + +struct TestDir { + path: PathBuf, +} + +impl TestDir { + fn new() -> io::Result { + let counter = TEMP_COUNTER.fetch_add(1, Ordering::Relaxed); + let path = std::env::temp_dir().join(format!( + "echo-cas-disk-tier-{}-{counter}", + std::process::id() + )); + match fs::remove_dir_all(&path) { + Ok(()) => {} + Err(error) if error.kind() == ErrorKind::NotFound => {} + Err(error) => return Err(error), + } + fs::create_dir_all(&path)?; + Ok(Self { path }) + } + + fn path(&self) -> &Path { + &self.path + } +} + +impl Drop for TestDir { + fn drop(&mut self) { + let _ = fs::remove_dir_all(&self.path); + } +} + +fn missing_blob(message: &'static str) -> io::Error { + io::Error::other(message) +} + +#[test] +fn disk_tier_persists_bytes_across_reopen() -> TestResult { + let root = TestDir::new()?; + let bytes = b"durable retained bytes"; + let hash = { + let tier = DiskTier::open(root.path())?; + tier.put(bytes)? + }; + + let reopened = DiskTier::open(root.path())?; + let stored = reopened + .get(&hash)? + .ok_or_else(|| missing_blob("reopened disk tier did not contain retained bytes"))?; + + assert_eq!(stored.as_ref(), bytes); + assert!(reopened.has(&hash)?); + Ok(()) +} + +#[test] +fn disk_tier_put_verified_rejects_mismatch_without_mutating_store() -> TestResult { + let root = TestDir::new()?; + let tier = DiskTier::open(root.path())?; + let original = b"original retained bytes"; + let original_hash = tier.put(original)?; + let bad_hash = BlobHash::from_bytes([0xFF; 32]); + + let result = tier.put_verified(bad_hash, b"mismatched retained bytes"); + + assert!(matches!( + result, + Err(DiskTierError::Cas(CasError::HashMismatch { expected, .. })) if expected == bad_hash + )); + assert!(!tier.has(&bad_hash)?); + assert!(tier.get(&bad_hash)?.is_none()); + let stored = tier + .get(&original_hash)? + .ok_or_else(|| missing_blob("original blob disappeared after rejected write"))?; + assert_eq!(stored.as_ref(), original); + assert_eq!(tier.list()?, vec![original_hash]); + Ok(()) +} + +#[test] +fn disk_tier_missing_blob_remains_absence() -> TestResult { + let root = TestDir::new()?; + let tier = DiskTier::open(root.path())?; + let missing = BlobHash::from_bytes([0x55; 32]); + + assert!(!tier.has(&missing)?); + assert!(tier.get(&missing)?.is_none()); + Ok(()) +} + +#[test] +fn disk_tier_list_returns_sorted_hashes() -> TestResult { + let root = TestDir::new()?; + let tier = DiskTier::open(root.path())?; + let payloads: [&[u8]; 4] = [ + b"z-last".as_slice(), + b"a-first".as_slice(), + b"middle".as_slice(), + b"another".as_slice(), + ]; + let mut expected = Vec::new(); + for payload in payloads { + expected.push(tier.put(payload)?); + } + expected.sort_unstable(); + expected.dedup(); + let first_hex = expected[0].to_string(); + let temp_shard = root.path().join("blobs").join(&first_hex[..2]); + fs::write(temp_shard.join(".stale-write.tmp"), b"partial")?; + + assert_eq!(tier.list()?, expected); + + let reopened = DiskTier::open(root.path())?; + assert_eq!(reopened.list()?, expected); + Ok(()) +}