From 4865efa4998f3b62ff602a8071179ede9f6ab409 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Tue, 21 Apr 2026 09:00:00 -0400 Subject: [PATCH] refactor(store): simplify Repository and blob wiring - Use hex crate for [u8; 32] RecordId instead of hand-rolled loops - Drop dead prefix_segments cache field; key assembly consumes R::PREFIX.split('/') directly, removing an intermediate Vec<&str> - Cache BlobStore on RunDatabaseInner (built once in open_writer/ open_reader via a new build() helper) instead of per-blob construction - Trim the replay_revocations doc comment to drop a stale plan reference Co-Authored-By: Claude Opus 4.7 (1M context) --- Cargo.lock | 1 + lib/crates/fabro-store/Cargo.toml | 1 + .../fabro-store/src/record/record_id.rs | 25 +----- .../fabro-store/src/record/repository.rs | 76 ++++++++----------- .../fabro-store/src/record/transaction.rs | 13 ++-- .../fabro-store/src/slate/auth_tokens.rs | 5 +- lib/crates/fabro-store/src/slate/run_store.rs | 37 +++------ 7 files changed, 54 insertions(+), 104 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 75efe4397..5a1f79cc0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2102,6 +2102,7 @@ dependencies = [ "dashmap", "fabro-types", "futures", + "hex", "insta", "object_store", "percent-encoding", diff --git a/lib/crates/fabro-store/Cargo.toml b/lib/crates/fabro-store/Cargo.toml index 6b3fe893b..9145a7730 100644 --- a/lib/crates/fabro-store/Cargo.toml +++ b/lib/crates/fabro-store/Cargo.toml @@ -13,6 +13,7 @@ workspace = true [dependencies] fabro-types = { path = "../fabro-types" } +hex.workspace = true slatedb.workspace = true object_store.workspace = true percent-encoding.workspace = true diff --git a/lib/crates/fabro-store/src/record/record_id.rs b/lib/crates/fabro-store/src/record/record_id.rs index 21a483a67..1ce67f069 100644 --- a/lib/crates/fabro-store/src/record/record_id.rs +++ b/lib/crates/fabro-store/src/record/record_id.rs @@ -1,5 +1,3 @@ -use std::fmt::Write; - use fabro_types::{RunBlobId, RunId}; use super::RecordId; @@ -7,11 +5,7 @@ use crate::{Error, Result}; impl RecordId for [u8; 32] { fn key_segments(&self) -> Vec { - let mut encoded = String::with_capacity(self.len() * 2); - for byte in self { - write!(&mut encoded, "{byte:02x}").expect("write to String cannot fail"); - } - vec![encoded] + vec![hex::encode(self)] } fn from_key_segments(segs: &[&str]) -> Result { @@ -21,22 +15,9 @@ impl RecordId for [u8; 32] { segs.len() ))); }; - - if segment.len() != 64 { - return Err(Error::KeyParse(format!( - "expected 64 hex characters for [u8; 32], got {}", - segment.len() - ))); - } - let mut bytes = [0_u8; 32]; - for (index, chunk) in segment.as_bytes().chunks_exact(2).enumerate() { - let chunk = std::str::from_utf8(chunk).map_err(|err| { - Error::KeyParse(format!("hex segment was not valid UTF-8: {err}")) - })?; - bytes[index] = u8::from_str_radix(chunk, 16) - .map_err(|err| Error::KeyParse(format!("invalid hex byte {chunk:?}: {err}")))?; - } + hex::decode_to_slice(segment, &mut bytes) + .map_err(|err| Error::KeyParse(format!("invalid hex segment {segment:?}: {err}")))?; Ok(bytes) } } diff --git a/lib/crates/fabro-store/src/record/repository.rs b/lib/crates/fabro-store/src/record/repository.rs index f6369ef58..bfcc18fdc 100644 --- a/lib/crates/fabro-store/src/record/repository.rs +++ b/lib/crates/fabro-store/src/record/repository.rs @@ -81,16 +81,15 @@ use crate::{Error, Result, keys}; /// named store such as `AuthCodeStore` or `RefreshTokenStore`, which can add /// domain-specific behavior on top of the generic storage primitives here. pub(crate) struct Repository { - db: Arc, - prefix_segments: Vec<&'static str>, - _record: PhantomData, + db: Arc, + _record: PhantomData, } impl Repository { pub(crate) fn new(db: Arc) -> Self { + validate_prefix::(); Self { db, - prefix_segments: prefix_segments::(), _record: PhantomData, } } @@ -141,32 +140,22 @@ impl Repository { extra_segments: &'a [&'a str], ) -> RepositoryStream<'a, (R::Id, R)> { match prefix_key::(extra_segments) { - Ok(prefix) => { - let prefix_segments = self.prefix_segments.clone(); - Box::pin( - scan_entries(Arc::clone(&self.db), &prefix).map(move |result| { - result - .map_err(Into::into) - .and_then(|entry| decode_entry::(&entry, &prefix_segments)) - }), - ) - } + Ok(prefix) => Box::pin(scan_entries(Arc::clone(&self.db), &prefix).map(|result| { + result + .map_err(Into::into) + .and_then(|entry| decode_entry::(&entry)) + })), Err(err) => Box::pin(stream::once(async move { Err(err) })), } } pub(crate) fn scan_ids_stream(&self) -> RepositoryStream<'_, R::Id> { match prefix_key::(&[]) { - Ok(prefix) => { - let prefix_segments = self.prefix_segments.clone(); - Box::pin( - scan_entries(Arc::clone(&self.db), &prefix).map(move |result| { - result - .map_err(Into::into) - .and_then(|entry| parse_entry_id::(&entry, &prefix_segments)) - }), - ) - } + Ok(prefix) => Box::pin(scan_entries(Arc::clone(&self.db), &prefix).map(|result| { + result + .map_err(Into::into) + .and_then(|entry| parse_entry_id::(&entry)) + })), Err(err) => Box::pin(stream::once(async move { Err(err) })), } } @@ -198,52 +187,48 @@ impl Repository { pub(crate) type RepositoryStream<'a, T> = Pin> + Send + 'a>>; pub(super) fn key_for_id(id: &R::Id) -> Result { - let prefix_segments = prefix_segments::(); let id_segments = id.key_segments(); - let id_segments: Vec<&str> = id_segments.iter().map(String::as_str).collect(); key_from_segments( - prefix_segments - .iter() - .copied() - .chain(id_segments.iter().copied()), + R::PREFIX + .split('/') + .chain(id_segments.iter().map(String::as_str)), ) } pub(super) fn prefix_key(extra_segments: &[&str]) -> Result { - let prefix_segments = prefix_segments::(); - prefix_from_segments( - prefix_segments - .iter() - .copied() - .chain(extra_segments.iter().copied()), - ) + prefix_from_segments(R::PREFIX.split('/').chain(extra_segments.iter().copied())) } -fn decode_entry(entry: &KeyValue, prefix_segments: &[&str]) -> Result<(R::Id, R)> { - let id = parse_entry_id::(entry, prefix_segments)?; +fn decode_entry(entry: &KeyValue) -> Result<(R::Id, R)> { + let id = parse_entry_id::(entry)?; let value = R::Codec::decode(&entry.value)?; Ok((id, value)) } -fn parse_entry_id(entry: &KeyValue, prefix_segments: &[&str]) -> Result { +fn parse_entry_id(entry: &KeyValue) -> Result { let raw_key = String::from_utf8(entry.key.to_vec()) .map_err(|err| Error::Other(format!("stored key is not valid UTF-8: {err}")))?; let segments: Vec<&str> = keys::SlateKey::segments(&raw_key).collect(); - if segments.len() < prefix_segments.len() { + let prefix_len = R::PREFIX.split('/').count(); + if segments.len() < prefix_len { return Err(Error::KeyParse(format!( "key {raw_key:?} had {} segments, expected at least {} for prefix {}", segments.len(), - prefix_segments.len(), + prefix_len, R::PREFIX ))); } - if segments[..prefix_segments.len()] != prefix_segments[..] { + if !segments[..prefix_len] + .iter() + .copied() + .eq(R::PREFIX.split('/')) + { return Err(Error::KeyParse(format!( "key {raw_key:?} did not match expected prefix {}", R::PREFIX ))); } - R::Id::from_key_segments(&segments[prefix_segments.len()..]) + R::Id::from_key_segments(&segments[prefix_len..]) } fn scan_entries( @@ -274,7 +259,7 @@ fn scan_entries( ) } -fn prefix_segments() -> Vec<&'static str> { +fn validate_prefix() { debug_assert!( !R::PREFIX.is_empty() && !R::PREFIX.starts_with('/') @@ -283,7 +268,6 @@ fn prefix_segments() -> Vec<&'static str> { "Record::PREFIX must be a non-empty '/'-separated path with no empty segments: {}", R::PREFIX ); - R::PREFIX.split('/').collect() } fn key_from_segments<'a>(segments: impl IntoIterator) -> Result { diff --git a/lib/crates/fabro-store/src/record/transaction.rs b/lib/crates/fabro-store/src/record/transaction.rs index a66e7e80f..8c7c710ac 100644 --- a/lib/crates/fabro-store/src/record/transaction.rs +++ b/lib/crates/fabro-store/src/record/transaction.rs @@ -10,20 +10,21 @@ where { let mut tx = Tx::new(); let value = f(&mut tx)?; - if !tx.has_ops { - return Ok(value); + if tx.has_ops { + db.write(tx.batch).await?; } - db.write(tx.into_batch()).await?; Ok(value) } pub(crate) struct Tx { batch: WriteBatch, + /// SlateDB rejects empty `WriteBatch` commits; skip the write entirely + /// when the closure produced no operations. has_ops: bool, } impl Tx { - pub(crate) fn new() -> Self { + fn new() -> Self { Self { batch: WriteBatch::new(), has_ops: false, @@ -51,10 +52,6 @@ impl Tx { self.has_ops = true; Ok(self) } - - fn into_batch(self) -> WriteBatch { - self.batch - } } #[cfg(test)] diff --git a/lib/crates/fabro-store/src/slate/auth_tokens.rs b/lib/crates/fabro-store/src/slate/auth_tokens.rs index 9357dd5b8..97cb1a8aa 100644 --- a/lib/crates/fabro-store/src/slate/auth_tokens.rs +++ b/lib/crates/fabro-store/src/slate/auth_tokens.rs @@ -49,9 +49,8 @@ pub struct RefreshTokenStore { db: Arc, repo: Repository, consume_locks: KeyedMutex<[u8; 32]>, - /// In-memory only by design (origin R6): persisting attacker-supplied - /// hashes adds an unbounded-growth surface under token-stuffing attack with - /// no security benefit. Do not migrate this into Repository. + /// In-memory only: persisting attacker-supplied hashes would be an + /// unbounded-growth surface under a token-stuffing attack. replay_revocations: DashMap<[u8; 32], DateTime>, } diff --git a/lib/crates/fabro-store/src/slate/run_store.rs b/lib/crates/fabro-store/src/slate/run_store.rs index be565bb63..a334281b9 100644 --- a/lib/crates/fabro-store/src/slate/run_store.rs +++ b/lib/crates/fabro-store/src/slate/run_store.rs @@ -33,6 +33,7 @@ impl std::fmt::Debug for RunDatabase { pub(crate) struct RunDatabaseInner { run_id: RunId, db: Db, + blob_store: BlobStore, event_seq: AtomicU32, close_lock: Mutex<()>, state_lock: Mutex<()>, @@ -44,33 +45,23 @@ pub(crate) struct RunDatabaseInner { impl RunDatabase { pub(crate) async fn open_writer(run_id: RunId, db: Db) -> Result { - let event_seq = - recover_next_seq(&db, keys::run_events_prefix(&run_id), keys::parse_event_seq).await?; - let (event_tx, _) = broadcast::channel(DEFAULT_EVENT_TAIL_LIMIT.max(16)); - Ok(Self { - inner: Arc::new(RunDatabaseInner { - run_id, - db, - event_seq: AtomicU32::new(event_seq), - close_lock: Mutex::new(()), - state_lock: Mutex::new(()), - projection_cache: Mutex::new(EventProjectionCache::default()), - recent_events: Mutex::new(VecDeque::with_capacity(DEFAULT_EVENT_TAIL_LIMIT)), - recent_event_limit: DEFAULT_EVENT_TAIL_LIMIT, - event_tx, - }), - read_only: false, - }) + Self::build(run_id, db, false).await } pub(crate) async fn open_reader(run_id: RunId, db: Db) -> Result { + Self::build(run_id, db, true).await + } + + async fn build(run_id: RunId, db: Db, read_only: bool) -> Result { let event_seq = recover_next_seq(&db, keys::run_events_prefix(&run_id), keys::parse_event_seq).await?; let (event_tx, _) = broadcast::channel(DEFAULT_EVENT_TAIL_LIMIT.max(16)); + let blob_store = BlobStore::new(Arc::new(db.clone())); Ok(Self { - inner: Arc::new(RunDatabaseInner { + inner: Arc::new(RunDatabaseInner { run_id, db, + blob_store, event_seq: AtomicU32::new(event_seq), close_lock: Mutex::new(()), state_lock: Mutex::new(()), @@ -79,7 +70,7 @@ impl RunDatabase { recent_event_limit: DEFAULT_EVENT_TAIL_LIMIT, event_tx, }), - read_only: true, + read_only, }) } @@ -285,15 +276,11 @@ impl RunDatabase { if self.read_only { return Err(Error::ReadOnly); } - BlobStore::new(Arc::new(self.inner.db.clone())) - .write(data) - .await + self.inner.blob_store.write(data).await } pub async fn read_blob(&self, id: &RunBlobId) -> Result> { - BlobStore::new(Arc::new(self.inner.db.clone())) - .read(id) - .await + self.inner.blob_store.read(id).await } pub async fn list_blobs(&self) -> Result> {