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) <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-04-21 09:00:00 -04:00
parent b862762b29
commit 4865efa499
No known key found for this signature in database
7 changed files with 54 additions and 104 deletions

1
Cargo.lock generated
View file

@ -2102,6 +2102,7 @@ dependencies = [
"dashmap",
"fabro-types",
"futures",
"hex",
"insta",
"object_store",
"percent-encoding",

View file

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

View file

@ -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<String> {
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<Self> {
@ -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)
}
}

View file

@ -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<R: Record> {
db: Arc<Db>,
prefix_segments: Vec<&'static str>,
_record: PhantomData<R>,
db: Arc<Db>,
_record: PhantomData<R>,
}
impl<R: Record> Repository<R> {
pub(crate) fn new(db: Arc<Db>) -> Self {
validate_prefix::<R>();
Self {
db,
prefix_segments: prefix_segments::<R>(),
_record: PhantomData,
}
}
@ -141,32 +140,22 @@ impl<R: Record> Repository<R> {
extra_segments: &'a [&'a str],
) -> RepositoryStream<'a, (R::Id, R)> {
match prefix_key::<R>(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::<R>(&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::<R>(&entry))
})),
Err(err) => Box::pin(stream::once(async move { Err(err) })),
}
}
pub(crate) fn scan_ids_stream(&self) -> RepositoryStream<'_, R::Id> {
match prefix_key::<R>(&[]) {
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::<R>(&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::<R>(&entry))
})),
Err(err) => Box::pin(stream::once(async move { Err(err) })),
}
}
@ -198,52 +187,48 @@ impl<R: Record> Repository<R> {
pub(crate) type RepositoryStream<'a, T> = Pin<Box<dyn Stream<Item = Result<T>> + Send + 'a>>;
pub(super) fn key_for_id<R: Record>(id: &R::Id) -> Result<keys::SlateKey> {
let prefix_segments = prefix_segments::<R>();
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<R: Record>(extra_segments: &[&str]) -> Result<keys::SlateKey> {
let prefix_segments = prefix_segments::<R>();
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<R: Record>(entry: &KeyValue, prefix_segments: &[&str]) -> Result<(R::Id, R)> {
let id = parse_entry_id::<R>(entry, prefix_segments)?;
fn decode_entry<R: Record>(entry: &KeyValue) -> Result<(R::Id, R)> {
let id = parse_entry_id::<R>(entry)?;
let value = R::Codec::decode(&entry.value)?;
Ok((id, value))
}
fn parse_entry_id<R: Record>(entry: &KeyValue, prefix_segments: &[&str]) -> Result<R::Id> {
fn parse_entry_id<R: Record>(entry: &KeyValue) -> Result<R::Id> {
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<R: Record>() -> Vec<&'static str> {
fn validate_prefix<R: Record>() {
debug_assert!(
!R::PREFIX.is_empty()
&& !R::PREFIX.starts_with('/')
@ -283,7 +268,6 @@ fn prefix_segments<R: Record>() -> 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<Item = &'a str>) -> Result<keys::SlateKey> {

View file

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

View file

@ -49,9 +49,8 @@ pub struct RefreshTokenStore {
db: Arc<slatedb::Db>,
repo: Repository<RefreshToken>,
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<R>.
/// In-memory only: persisting attacker-supplied hashes would be an
/// unbounded-growth surface under a token-stuffing attack.
replay_revocations: DashMap<[u8; 32], DateTime<Utc>>,
}

View file

@ -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<Self> {
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> {
Self::build(run_id, db, true).await
}
async fn build(run_id: RunId, db: Db, read_only: bool) -> Result<Self> {
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<Option<Bytes>> {
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<Vec<RunBlobId>> {