diff --git a/Cargo.lock b/Cargo.lock index c779b34bf..e6d735c00 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2074,6 +2074,7 @@ dependencies = [ "dashmap", "fabro-types", "futures", + "insta", "object_store", "percent-encoding", "serde", diff --git a/docs/brainstorms/2026-04-20-fabro-store-record-abstractions-requirements.md b/docs/brainstorms/2026-04-20-fabro-store-record-abstractions-requirements.md new file mode 100644 index 000000000..90848007f --- /dev/null +++ b/docs/brainstorms/2026-04-20-fabro-store-record-abstractions-requirements.md @@ -0,0 +1,146 @@ +--- +date: 2026-04-20 +topic: fabro-store-record-abstractions +--- + +# Fabro-store Record Abstractions + +## Problem Frame + +Adding a new persisted record type to `fabro-store` today means writing ~200–400 LOC of repetitive plumbing in a new `Slate*Store`: key construction, JSON serialization, `get`/`put`/`delete`/`scan_prefix` wrappers, optional per-key consume mutex, optional GC-by-prefix-scan, optional secondary index. The existing four record families — `RefreshToken`, `AuthCode`, `Blob`, and the Run catalog index — share most of that plumbing mechanically, but each duplicates it inside `lib/crates/fabro-store/src/slate/`. As more record types land (new auth flows, sessions, vault entries, agent state), the duplication compounds and each copy is one more place where serialization, locking, or key encoding can drift. + +Greenfield, no production deployments — backwards compat can be broken freely. + +## Architecture + +``` +lib/crates/fabro-store/ + Database + .refresh_tokens() -> Arc + .auth_codes() -> Arc + .blobs() -> Arc + .catalog_index() -> Arc + .runs() -> Runs (unchanged) + + *Store wrappers (one per Record) + own: Repository + domain helpers (KeyedMutex, ReplayCache, …) + expose: domain-named methods (consume_and_rotate, delete_chain, …) + + Repository ← reusable typed K/V layer + get / put / delete / scan_stream / scan_prefix_stream / gc + serializes via R::Codec + keys built from R::PREFIX + R::Id::key_segments + pub(crate) — wrapper Stores own it; never exposed on Database + + trait Record / trait RecordId / trait Codec + Record::PREFIX (const) + Record::Codec (associated type; each impl writes `type Codec = JsonCodec;`) + Record::id(&self) (-> Self::Id) + RecordId::key_segments(&self) -> Vec (Repository assembles SlateKey) + + transaction(&db, |tx| { tx.put(&r1)?; tx.put(&r2)?; }) + ← cross-record atomic batch +``` + +Run's event log, projection cache, broadcast channel, and per-run mutex stay in `RunDatabase` and are deliberately out of scope. The Run catalog *index* (today's `slate/catalog.rs`) is in scope and becomes `RunCatalogIndex` on top of `Repository`. + +## Records in Scope + +| Record | PREFIX | Id type | Codec | Special semantics | +|------------------|-------------------------|---------------------|-----------|---------------------------------------------------------| +| RefreshToken | `auth/refresh` | `[u8; 32]` (hex) | JsonCodec | KeyedMutex consume lock, in-memory replay revocation | +| AuthCode | `auth/code` | `String` (opaque) | JsonCodec | KeyedMutex consume lock, single-use | +| Blob | `blobs/sha256` | `RunBlobId` | RawBytesCodec | Immutable, content-addressed, never deleted | +| RunCatalogEntry | `runs/_index/by-start` | `RunId` | EmptyCodec | Empty value; date segment derived from `RunId.created_at()` | + +## Requirements + +**Core abstraction** +- R1. Define `trait Record: Sized + Send + Sync + 'static` with associated `type Id: RecordId`, `type Codec: Codec`, `const PREFIX: &'static str`, and `fn id(&self) -> Self::Id`. Each `impl Record` writes `type Codec = JsonCodec;` (or its override) explicitly — no associated-type default, since `associated_type_defaults` is unstable on stable Rust as of 2026. One extra line per impl is the lowest-magic alternative; the workspace doesn't have a proc-macro sub-crate today and adding one is more weight than the savings justifies. `PREFIX` is a `/`-separated path of segments (e.g. `"auth/refresh"`, `"runs/_index/by-start"`) that `Repository` splits before assembling the `\0`-separated `SlateKey`; `/` is therefore reserved inside any single segment. +- R2. Define `trait RecordId { fn key_segments(&self) -> Vec; }`. Implementations return segment data; `Repository` assembles the SlateKey internally so `SlateKey` (and its `\0`-segment-boundary invariant) stay `pub(crate)`. Built-in impls: `[u8; 32]` writes one hex segment; `String` writes itself; `RunBlobId` writes its sha256 hex segment; `RunId` writes two segments — `` derived from `RunId.created_at()` followed by the RunId string. (`RunId`-as-Record-Id is used only by `RunCatalogEntry` today; if a future record needs a single-segment RunId encoding it uses a newtype wrapper.) +- R3. Define `trait Codec { fn encode(r: &R) -> Result>; fn decode(bytes: &[u8]) -> Result; }` with built-in implementations: `JsonCodec` (used by most records), `RawBytesCodec` (for `Blob`), `EmptyCodec` (for index entries with no value). `JsonCodec::encode` is implemented as a one-line literal forwarding to `serde_json::to_vec(&value)` (and `decode` to `serde_json::from_slice`); the implementation IS the proof of byte-identity. A snapshot test on a representative `RefreshToken` and `AuthCode` is kept as a regression net so any accidental change to the forwarding impl (e.g. someone adds an envelope) fails CI loudly. The snapshot files commit the on-disk wire format to the repo and are reviewed as wire-format changes when they change. +- R4. Provide `Repository` exposing the typed K/V primitives: `get(&R::Id)`, `put(&R)`, `delete(&R::Id)`, `scan_stream() -> impl Stream>` (full prefix), `scan_prefix_stream(extra_segments)`, and `gc(predicate: impl Fn(&R) -> bool) -> Result` (scans, filters, deletes — replaces the three hand-rolled GC-by-prefix-scan loops in today's stores). `Repository` is `pub(crate)`. The `gc` predicate is sync `Fn` (so it cannot mutate caller state across the scan), MUST NOT perform I/O or block (it runs once per scanned record), and MUST be free of side effects (gc may be called more than once on the same record set during retries or future caller logic). `gc` issues all deletes in a single `slatedb::WriteBatch` — atomic vs today's per-key delete loop, deliberate behaviour change. No capability traits, no marker subtraits — domain logic lives in the wrapper Store. + +**Wrapper stores** +- R5. Each record family has a named domain `*Store` type that owns a `Repository` and any domain-specific helpers (per-key mutex, in-memory caches, replay revocation set). Trivial stores (`BlobStore`, `RunCatalogIndex`) are still concrete named types even when they are thin pass-throughs. **Security boundary**: `Repository` is `pub(crate)`; construction is gated to the wrapper Store and to test fixtures. The `Repository` field on each wrapper Store is private (`pub(super)` at most); wrapper Stores never expose `.scan_stream()` / `.gc()` / `.repository()` accessors. `Database` exposes only the wrapper Stores, never `Repository` directly. Domain methods that need a scan (e.g. `gc_expired`, `delete_chain`) perform it internally and return aggregate results, never raw records. The boundary is enforced for crate-external callers by the `pub(crate)` visibility; for crate-internal contributors it is convention + code review (a future fabro-store author who constructs `Repository` directly is bypassing the design and the PR review should catch it). +- R6. `RefreshTokenStore` keeps its existing API (`insert_refresh_token`, `find_refresh_token`, `consume_and_rotate`, `delete_chain`, `gc_expired`, `mark_refresh_token_replay`, `was_recently_replay_revoked`). `consume_and_rotate` uses the new `transaction` helper while still holding its `KeyedMutex` guard around the call. The replay revocation set (`mark_refresh_token_replay` / `was_recently_replay_revoked`) is in-memory only with a 60s TTL — it MUST NOT be moved into `Repository` or otherwise persisted. Rationale: it's a transient signal so the current process can recognize replays within the rotation window; persisting attacker-supplied token hashes adds disk I/O and a new GC surface with no security benefit and an unbounded-growth risk under token-stuffing attack. +- R7. `AuthCodeStore` keeps its existing API (`insert`, `consume`, `gc_expired`). The `AuthCode` struct gains a `code: String` first field carrying its own key (today's struct receives the code as a separate parameter to `insert`); this lets `Record::id(&self)` return the key from the value, which `Repository::scan_stream` requires. The on-disk JSON shape gains one field — acceptable under greenfield. +- R8. `BlobStore` exposes `read(&RunBlobId)`, `write(bytes) -> RunBlobId`, `exists(&RunBlobId)`. `delete` is intentionally not exposed — current behaviour is that blobs survive run deletion (enforced by `delete_run_keeps_global_cas_blobs` test). `RunDatabase::write_blob` / `read_blob` / `list_blobs` keep their signatures and delegate to `BlobStore` internally so existing callers (and tests) require no changes. +- R9. `RunCatalogIndex` exposes `add(&RunId)`, `remove(&RunId)`, `list(query: &ListRunsQuery) -> Vec`, replacing the free functions in `lib/crates/fabro-store/src/slate/catalog.rs`. `list` scans the prefix, applies `query.start`/`query.end` filtering against `run_id.created_at()`, and sorts by the same key as today's `catalog.rs:39-49` — `(year, month, day, hour, minute, run_id)` ascending — derived inside `RunCatalogIndex::list` from each `RunId`'s embedded ULID timestamp. The post-list summary-building loop in `Database::list_runs` (`slate/mod.rs:170-183`) is **not** absorbed into `RunCatalogIndex`; only the catalog scan + filter + sort move. + +**Cross-record atomic writes** +- R10. Provide a `transaction(&db, |tx| { … })` helper producing a single `slatedb::WriteBatch`. `Tx` exposes `put(&R)` and `delete(&R::Id)` for any `R: Record`, so a single transaction can span record types. Replaces the hand-built `slatedb::WriteBatch` in `RefreshTokenStore::consume_and_rotate`. **Atomicity invariants** (security-critical for token rotation): the closure is `FnOnce(&mut Tx) -> Result` (synchronous); on closure `Err`, no slatedb write occurs (codec errors short-circuit via `?`); on closure `Ok`, exactly one `slatedb::WriteBatch::write` commit is issued; no retry, no partial flush, no commit-on-drop. A fault-injection test asserts that an encode failure on the Nth `put` leaves the database unchanged. To restrict the cross-type capability to authorized callers, `transaction` is `pub(crate)` — wrapper Stores expose domain-named atomic methods (e.g. `RefreshTokenStore::consume_and_rotate`) that compose `transaction` internally. + +**Database surface** +- R11. `Database` exposes `refresh_tokens()`, `auth_codes()`, `blobs()`, `catalog_index()`, and `runs()`. Each lazily initializes its store via `OnceCell` and returns `Arc<*Store>` (mirrors today's pattern in `lib/crates/fabro-store/src/slate/mod.rs:208-228`). +- R12. `RunDatabase`'s event-sourcing machinery (event log, projection cache, broadcast channel, atomic seq counter, per-run state mutex) and public method signatures are unchanged. Two internal touch-points exist: (a) the catalog call sites all live in `Database` itself — `Database::create_run` (`slate/mod.rs:119,127`), `Database::list_runs` (`slate/mod.rs:169`), and `Database::delete_run` (`slate/mod.rs:204`) — and migrate from `catalog::write_index` / `catalog::delete_index` / `catalog::list_run_ids` to the new `RunCatalogIndex` (lazily initialized on `Database` via `OnceCell`, like the other stores); (b) `RunDatabase::write_blob` / `read_blob` / `list_blobs` (`run_store.rs:283-302`) keep their public signatures and behaviour but delegate internally to `BlobStore` per R8. No callers change. + +**Reusable helpers** +- R13. Extract `KeyedMutex` (per-key async mutex on top of `DashMap>>` with auto-cleanup when strong count drops to 2) as a shared helper inside `fabro-store`, used by `RefreshTokenStore` and `AuthCodeStore`. **Security purpose**: `KeyedMutex` serializes concurrent `consume` operations on the same key. Required by single-use semantics for `AuthCode` and replay detection for `RefreshToken` — without it, a TOCTOU race between the find/get and the delete/mark-used write permits double-consumption. Auto-cleanup at `strong_count == 2` bounds the map size so a high volume of distinct codes/hashes cannot grow it without limit. The cleanup check and the entry insertion happen under the same `DashMap` shard lock so a concurrent `lock(&key)` cannot observe a soon-to-be-dropped Arc and acquire a different `Mutex` instance for the same key. `KeyedMutex` exposes only `lock(&self, key: K) -> Guard<'_>` — never returns or clones the inner `Arc`; `Guard` holds the Arc internally so the strong-count==2 check happens after `Guard` drops. The lock MUST NOT be made optional, replaced with a TTL/janitor, or relocated to a separate process without explicit security review. + +## Success Criteria + +- Adding a 5th simple K/V record type requires writing only: a struct + `impl Record` + an `impl RecordId` (if its ID type isn't already covered) + a thin `*Store` wrapper. No new key-construction code, no `serde_json::to_vec`/`from_slice` per store, no copy of get/put/delete/scan plumbing. +- All existing public store APIs (`RefreshTokenStore::consume_and_rotate`, `AuthCodeStore::consume`, etc.) keep their signatures and observable behaviour. Existing unit and integration tests pass without modification. +- `lib/crates/fabro-store/src/slate/auth_tokens.rs` and `auth_codes.rs` shrink to roughly their domain logic (consume lock, GC traversal, replay cache) — kv plumbing moves into `Repository`. +- `lib/crates/fabro-store/src/slate/catalog.rs` is deleted; `RunCatalogIndex` exposes the same operations through `Repository`. +- A single `transaction(…)` call replaces the hand-built `slatedb::WriteBatch` in `RefreshTokenStore::consume_and_rotate`. + +## Scope Boundaries + +- Run aggregate's event-sourcing machinery (`RunDatabase` event log, projection cache, broadcast channel, `recover_next_seq`, `EventProjectionCache`, atomic seq counter) is not refactored. `RunDatabase`'s blob methods are touched only to delegate to `BlobStore` per R8 — no signature or behaviour change. +- No marker capability traits (`ExpiringRecord`, `ConsumableRecord`, `IndexedRecord`, `ContentAddressed`). Capabilities are hand-written per `*Store`. +- No record schema versioning or migration framework. Greenfield, can break compat. +- No swap-out of the storage backend (still SlateDB on `Arc`). Repository is parameterized by `R`, not by storage. +- No changes to `ArtifactStore` or other non-SlateDB storage. +- No changes to the public `Database::create_run` / `open_run` / `delete_run` / `list_runs` API signatures or observable behaviour. Internal implementations of `create_run`, `list_runs`, and `delete_run` consume the new `RunCatalogIndex` per R12. + +## Key Decisions + +- **Two worlds** (Run separate from simple K/V records): unifying an event-sourced aggregate (atomic seq counter, projection cache, live broadcast, per-run state mutex) with single-key records would either flatten to a lowest-common-denominator API or force Run's complexity onto records that don't need it. +- **Hand-written `*Store` wrappers on a thin `Repository`** (not marker capability traits): chose lowest magic. Capability variations are small in number, infrequently added, and easy to read inline. A reader of `RefreshTokenStore` should see exactly what `consume_and_rotate` does without crossing trait-bound boundaries. +- **Multi-segment IDs via data-only `key_segments() -> Vec`**: `SlateKey` stays `pub(crate)`. `Repository` assembles the key internally — `RecordId` impls cannot violate the `\0`-segment-boundary invariant by accident. Small allocation per key derivation is acceptable for the encapsulation gain. +- **Codec specified explicitly per impl** (no associated-type default): `associated_type_defaults` is unstable on stable Rust as of 2026. Rather than wait, pull in a derive-macro sub-crate, or split into sub-traits, every `impl Record` writes `type Codec = JsonCodec;` (or its override). One extra line per impl, no nightly, no proc-macro infra. +- **`transaction(...)` as a single-batch atomic primitive, `pub(crate)`**: `consume_and_rotate` already needs an atomic two-key write; one helper covers it and any future cross-record flows. Closure is `FnOnce`, all-or-nothing, no commit-on-drop. Crate-private visibility prevents callers from bypassing per-store invariants — wrapper Stores expose domain-named atomic methods that compose `transaction` internally. +- **Repository for security-sensitive records is store-private**: `Database` exposes only wrapper Stores. `RefreshTokenStore` and `AuthCodeStore` never hand out a `Repository` accessor or generic `scan`. Prevents a new caller in `fabro-server` (or anywhere else) from walking the entire token table. +- **Named `*Store` for every record**: discoverability matters more than line count. `db.blobs()` and `db.catalog_index()` read more clearly than `db.repository::()`. The trivial wrappers (`BlobStore`, `RunCatalogIndex`) are accepted overhead — see Alternatives Considered for the leaner shape we rejected. + +## Alternatives Considered + +Three lighter shapes were considered. The full scaffold (R1–R13) was chosen anyway. Comparison: + +| Option | New types | Approx. scaffold | Per-record cost (5th) | Trade | +|---|---|---|---|---| +| (1) Copy-paste-and-rename | 0 | 0 LOC | ~200 LOC duplicated | Zero abstraction tax now; drift risk grows linearly. | +| (2) `JsonStore` helper only | 1 (~30 LOC) | ~30 LOC | ~80 LOC | Removes JSON serde duplication only. No typed key, no codec abstraction, no transaction helper, no security boundary. Each store keeps bespoke key construction and consume locks. | +| (3) Minimum-viable subset (R1+R2+R3 JsonCodec only+R4+R6+R7+R13) | ~5 traits + Repository + KeyedMutex | ~120 LOC | ~30 LOC | Touches only `auth_tokens.rs` and `auth_codes.rs`. Drops `BlobStore`, `RunCatalogIndex`, cross-record `transaction`, `RawBytesCodec`, `EmptyCodec`. Loses the boundary requirement (no Blob/catalog pass-through to demonstrate the pattern). | +| **(chosen) Full scaffold (R1–R13)** | ~6 traits + Repository + Tx + 4 wrapper stores + KeyedMutex | ~250 LOC | ~30 LOC | All of (3) plus Blob CAS and catalog refactor onto the same shape. R8 (`BlobStore` as a public Database surface) and R9 (catalog migration) are forward-looking in their *value* — they don't have a new today-consumer beyond what already works through `RunDatabase::write_blob` and `catalog::*` — but they are concrete refactors of existing code, not speculative new code. R10's cross-type Tx is similarly forward-looking. | + +**Why (chosen) over (3)**: standardizes the shape across every K/V record the crate has, so new entrants (sessions, vault, agent state) hit one well-understood pattern instead of choosing between "use `Repository`" and "use the older bespoke style." Designs the security boundary (R5) once for the whole crate so a future security-sensitive record inherits the discipline by default rather than reinventing it. (Note: only `RefreshToken` and `AuthCode` actually need the boundary today; for `Blob` and `RunCatalogEntry` the boundary is "free" — there's nothing sensitive to protect — so the security argument is forward-leaning, not load-bearing right now.) + +**Why not (2) `JsonStore`**: removes the JSON duplication but nothing else. Each store still hand-writes its own key construction (so the scan/key-parsing invariants stay scattered), its own consume mutex (so KeyedMutex can't naturally land), and its own atomic-write batch (so the security invariants in R10 stay implicit). Reduction in scaffolding is real, reduction in surface-to-think-about is much smaller than it looks. + +**Why not (1) copy-paste**: works fine until two records drift. The first such drift (e.g., one store fixes a key-encoding bug another doesn't) is a class of bug the abstraction prevents structurally. + +**Acknowledged speculative scope**: R8 (`BlobStore` public surface), R9 (catalog sweep), and R10's cross-type Tx are all forward-looking. The bet is that paying the setup cost once now is cheaper than retrofitting later. If that bet is wrong, the cost is the LOC delta between (3) and (chosen) plus the ongoing readability cost: a new contributor must learn `Record`/`RecordId`/`Codec`/`Repository`/`Tx`/`KeyedMutex` to add a record, vs. learning one `JsonStore` API; trait-resolution makes "where is this key constructed?" harder to grep; generic error messages reference `::Codec` instead of `JsonCodec`. The per-record LOC numbers in the table above are rough estimates — the actual delta between (2)/(3) and (chosen) for a 5th record is small (single-digit LOC). The primary argument for (chosen) over (3) is consistency-across-records and one-time security-boundary design, not per-record LOC. + +## Dependencies / Assumptions + +- SlateDB's `WriteBatch` is suitable as the underlying primitive for the `transaction` helper. Already used by `RefreshTokenStore::consume_and_rotate`. +- The current public consumers of `Database::auth_codes()` / `auth_tokens()` are limited to `fabro-server` and crate-internal tests. The `Slate*Store` type names are referenced in `fabro-server` (`serve.rs:799-829` types `Arc` etc.); renaming to `*Store` is a mechanical import update on the order of one PR for `fabro-server`, not a deep API change. + +## Outstanding Questions + +### Resolve Before Planning + +(none) + +### Deferred to Planning + +- [Affects R13] [Technical] Whether `KeyedMutex` lives at `lib/crates/fabro-store/src/keyed_mutex.rs` or in a small shared util module. Either is fine. +- [Affects R5–R9] [Technical] Migration order — which store moves to `Repository` first and in what PR sequence. Suggested order: R13 (`KeyedMutex` extraction, can land first as a precursor PR) → `BlobStore` (greenfield, simplest) → `RunCatalogIndex` (replaces `catalog.rs`) → `AuthCodeStore` → `RefreshTokenStore` (most complex due to consume-and-rotate + replay cache). Planning concern. +- [Affects R10] [Technical] Whether to add a separate `per_repository_batch` helper for the common single-type case so callers don't pay the cross-record machinery cost when they don't need it. Performance/ergonomics tweak, not a correctness question. + +## Next Steps + +→ `/ce:plan` for structured implementation planning diff --git a/docs/plans/2026-04-20-003-refactor-fabro-store-record-abstractions-plan.md b/docs/plans/2026-04-20-003-refactor-fabro-store-record-abstractions-plan.md new file mode 100644 index 000000000..335912d1b --- /dev/null +++ b/docs/plans/2026-04-20-003-refactor-fabro-store-record-abstractions-plan.md @@ -0,0 +1,725 @@ +--- +title: "refactor: Extract Record/Repository abstractions in fabro-store" +type: refactor +status: completed +date: 2026-04-20 +origin: docs/brainstorms/2026-04-20-fabro-store-record-abstractions-requirements.md +--- + +# refactor: Extract Record/Repository abstractions in fabro-store + +## Overview + +`fabro-store` currently has four hand-written `Slate*Store` types (`SlateAuthCodeStore`, `SlateAuthTokenStore`, blob plumbing on `RunDatabase`, free functions in `catalog.rs`) that each duplicate the same K/V plumbing on top of SlateDB: key construction, JSON serialization, get/put/delete/scan loops, optional per-key consume mutex, optional GC-by-prefix-scan. This plan replaces that duplication with a thin reusable typed K/V layer (`Repository`) plus per-record wrapper Stores that hold any domain-specific helpers (consume locks, replay caches, etc.). + +The Run aggregate (event log, projection cache, broadcast channel) stays as it is. Only the simple K/V records (RefreshToken, AuthCode, Blob, RunCatalogEntry) move to the new shape. + +This is a greenfield refactor — no production deployments, backwards-compat can be broken freely (see origin: `docs/brainstorms/2026-04-20-fabro-store-record-abstractions-requirements.md`). + +## Problem Frame + +Adding a new persisted record type today requires copying ~200 LOC of plumbing into a new `Slate*Store` and editing names. Each copy is one more place where serialization, locking, or key encoding can drift. As more record families land (sessions, vault entries, agent state), the duplication compounds. + +The brainstorm's chosen response: a `Record` trait + `Repository` typed K/V layer + named domain `*Store` wrappers + a security boundary that's designed once for the whole crate (see origin Key Decisions). + +## Requirements Trace + +All requirements traced from the origin document. + +- **R1.** `trait Record` with associated `type Id: RecordId`, `type Codec: Codec` (no default — explicit per impl), `const PREFIX: &'static str`, and `fn id(&self) -> Self::Id`. `PREFIX` MUST be a non-empty `/`-separated path of non-empty segments — no leading or trailing `/`, no empty segments, `/` is reserved within a segment. `Repository::new` `debug_assert!`s the constraint. (origin R1) +- **R2.** `trait RecordId: Sized { fn key_segments(&self) -> Vec; fn from_key_segments(segs: &[&str]) -> Result; }` with built-in impls for `[u8; 32]` (hex), `String`, `RunBlobId`, and `RunId` (date + ulid). `from_key_segments` is the parse counterpart used by `Repository::scan_stream` and `scan_ids_stream` — the only path to reconstruct `R::Id` from the SlateDB key for both value-carrying records and ZST marker records (e.g. `RunCatalogEntry`). Implementations of `key_segments` MUST NOT include the `\0` byte in any returned segment. `Repository` validates this **at runtime** before assembling any SlateKey (in `put`/`put_at`/`get`/`delete`/`exists`/scan-prefix construction): a segment containing `\0` returns `Err(Error::InvalidKeySegment)` rather than silently corrupting the keyspace. Validation is one byte scan per segment per call — negligible vs the SlateDB op. (origin R2 — extended at plan time) +- **R3.** `trait Codec` with three built-in implementations: `JsonCodec` (literal `serde_json` forwarding for value-carrying records), `RawBytesCodec` (byte-identity for `Blob`), `MarkerCodec` (for ZST marker records — `encode(_) -> Vec::new()`, `decode(&[]) -> R::default()`; bound `R: Default + Sized`). Snapshot test on a stable synthetic struct locks the `JsonCodec` wire format. **Wire-format change tracking:** `RefreshToken` JSON wire format is byte-identical to today (literal serde_json forward). `AuthCode` JSON wire format gains a new `code` field per R7 — acceptable under greenfield. (origin R3) +- **R4.** `Repository` (`pub(crate)`) exposes: + - For value-carrying records: `get(&id)`, `put(&r)`, `delete(&id)`, `scan_stream() -> Stream<(R::Id, R)>`, `scan_prefix_stream(extra_segments)`, `gc(predicate)`. + - For all records (including markers): `put_at(&id, &r)` (writes a key derived from the explicit id rather than from `r.id()`; `put` becomes sugar for `put_at(&r.id(), r)`), `exists(&id) -> bool`, `scan_ids_stream() -> Stream` (keys-only enumeration; doesn't touch values). Marker records (`R::Codec = MarkerCodec`) interact with `Repository` exclusively through the id-only API: `put_at`, `delete`, `exists`, `scan_ids_stream`. (origin R4 — extended at plan time) +- **R5.** Named domain `*Store` wrappers; `Repository` field is private; security boundary against external callers walking sensitive tables. (origin R5) +- **R6.** `RefreshTokenStore` keeps existing API; `consume_and_rotate` uses `transaction(...)` under `KeyedMutex` guard; replay cache stays in-memory. (origin R6) +- **R7.** `AuthCodeStore` keeps observable behaviour but `insert` changes signature from `insert(code: &str, entry: AuthCode)` to `insert(entry: AuthCode)` since `AuthCode` now carries its own `code` field (per the brainstorm decision). All existing callers update their construction pattern: compute the code, then construct `AuthCode { code, identity, ... }` in one literal. This is a public API change accepted under greenfield. (origin R7 — clarified at plan time) +- **R8.** `BlobStore` exposes `read`/`write`/`exists` (no `delete`, no `list`/`scan` — see R5 boundary). `RunDatabase::write_blob` and `RunDatabase::read_blob` keep signatures and delegate to `BlobStore`. `RunDatabase::list_blobs` keeps signature and continues using its existing free-function key scan path until/unless a future record needs blob enumeration through `Repository`. (origin R8 — clarified at plan time) +- **R9.** `RunCatalogIndex` exposes `add(&run_id)` (writes a `RunCatalogEntry` marker key), `remove(&run_id)`, `list(query)` (uses `Repository::scan_ids_stream` to enumerate marker keys, applies query filter, sorts by `(year, month, day, hour, minute, run_id)` per `catalog.rs:39-49`). Replaces `catalog.rs` free functions. `RunCatalogEntry` is a ZST marker (no fields); the run id lives only in the key. (origin R9 — extended at plan time) +- **R10.** `transaction(&db, |tx| ...)` helper, `pub(crate)`, `FnOnce`, single-batch atomic write, encode errors short-circuit; fault-injection test required. (origin R10) +- **R11.** `Database` exposes five accessors (lazy `OnceCell` init): `blobs()` lands in U2b, `catalog_index()` in U4, `auth_codes()` is preserved in U5, `refresh_tokens()` (renamed from `auth_tokens()`) lands in U6, `runs()` is unchanged. (origin R11) +- **R12.** `RunDatabase`'s event-sourcing machinery + public method signatures unchanged. `Database` consumes `RunCatalogIndex`. `RunDatabase::write_blob`/`read_blob` delegate to `BlobStore` (per R8); `RunDatabase::list_blobs` keeps its current free-function path. (origin R12 — clarified at plan time) +- **R13.** `KeyedMutex` extracted as shared helper; cleanup-under-shard-lock invariant; never exposes the inner Arc. (origin R13) + +## Scope Boundaries + +- Run aggregate event-sourcing machinery (`RunDatabase` event log, projection cache, broadcast channel, `recover_next_seq`, `EventProjectionCache`, atomic seq counter) is not refactored. +- No marker capability traits (`ExpiringRecord`, `ConsumableRecord`, `IndexedRecord`, `ContentAddressed`). +- No record schema versioning or migration framework. +- No swap-out of the storage backend (still SlateDB on `Arc`). +- No changes to `ArtifactStore` or other non-SlateDB storage. +- No changes to public `Database::create_run` / `open_run` / `delete_run` / `list_runs` API signatures or observable behaviour. Internals consume new `RunCatalogIndex` per R12. + +## Context & Research + +### Relevant Code and Patterns + +- `lib/crates/fabro-store/src/slate/auth_tokens.rs` — `SlateAuthTokenStore`, the most complex existing store (consume_and_rotate with WriteBatch, KeyedMutex, replay cache). +- `lib/crates/fabro-store/src/slate/auth_codes.rs` — `SlateAuthCodeStore`, mirror of the same pattern minus replay cache. +- `lib/crates/fabro-store/src/slate/run_store.rs:283-302` — `RunDatabase::write_blob`/`read_blob`/`list_blobs`, today's blob plumbing. +- `lib/crates/fabro-store/src/slate/catalog.rs` — three free functions (`write_index`, `delete_index`, `list_run_ids`) called from `Database::create_run`/`list_runs`/`delete_run`. +- `lib/crates/fabro-store/src/slate/mod.rs` — `Database` (lazy `OnceCell` per store), `Runs` accessor, run lifecycle methods that thread catalog calls. +- `lib/crates/fabro-store/src/keys.rs` — `SlateKey` builder (`pub(crate)`, `\0`-separated segments), per-record key constructors, parse helpers. **Stays `pub(crate)`** under R2's data-only `key_segments()` design. +- `lib/crates/fabro-store/src/lib.rs` — current `pub use` re-exports of `Slate*Store` types. + +### Pattern: lazy store init via `OnceCell` + +Today (`slate/mod.rs:208-228`): + +```text +auth_codes: Arc>> +auth_tokens: Arc>> +``` + +`Database::auth_codes()` / `auth_tokens()` use `get_or_try_init` to construct the store with `Arc::new(self.open_db().await?)`. New `BlobStore`, `RunCatalogIndex` follow the same shape. No new pattern needed. + +### Pattern: KeyedMutex (today inlined per store) + +Both `auth_tokens.rs:85-118` and `auth_codes.rs:48-77` carry the same `DashMap>>` + clone Arc + lock + decide-on-drop-via-strong-count==2 pattern. R13 extracts it. + +### Pattern: today's WriteBatch usage + +`auth_tokens.rs:101-110` builds one `slatedb::WriteBatch`, puts both keys, calls `db.write(batch).await?`. The `transaction(...)` helper preserves this exact shape — single batch, single commit. + +### External References + +None — this is an internal refactor. SlateDB API is already understood from existing usage. No new dependencies. + +### Institutional Learnings + +No prior `docs/solutions/` entries on this topic — first refactor of this kind in `fabro-store`. + +## Key Technical Decisions + +Most decisions carry forward from the origin Key Decisions section. + +- **Two worlds**: Run aggregate stays as-is; only K/V records move to `Repository`. (origin) +- **Hand-written `*Store` wrappers on a thin `Repository`**, no marker capability traits. (origin) +- **Multi-segment IDs via data-only `key_segments() -> Vec`**: keeps `SlateKey` `pub(crate)`. (origin) +- **Codec specified explicitly per impl** (no associated-type default): avoids unstable `associated_type_defaults`. (origin) +- **`transaction(...)` is `pub(crate)`**, single-batch atomic, `FnOnce`. Wrapper Stores compose it via domain-named methods. (origin) +- **Repository for security-sensitive records is store-private** (`pub(crate)` Repository, `pub(super)` field on the wrapper Store). External boundary enforced by visibility; internal discipline by code review. (origin + pass-2 review) + +Plan-time additions: + +- **KeyedMutex location**: `lib/crates/fabro-store/src/keyed_mutex.rs` (top-level module, not under `slate/`, since it's not slatedb-specific). Resolves origin deferred Q1. +- **Migration order** (resolves origin deferred Q2): U1 (KeyedMutex precursor) and U2a (trait scaffolding + Repository + transaction) are independent and can land in either order. U2b (Blob + BlobStore + Database::blobs) follows U2a. Then U3, U4, U5, U6 can be parallelized. U7 closes with re-exports + fabro-server cleanup. Each unit lands as one commit; each compiles and tests pass independently. See dependency graph in Implementation Units. +- **Per-Repository batch helper deferred** (resolves origin deferred Q3): not added in this refactor. Only `transaction(...)` ships. If profiling shows the cross-record machinery cost is meaningful for the single-type case, add later. +- **Module organization**: traits + Repository + Tx live in a new top-level `record/` module (`lib/crates/fabro-store/src/record/{mod,codec,repository,transaction}.rs`); per-record wrapper Stores stay under `slate/` (since they're slatedb-bound). `KeyedMutex` is its own top-level module. +- **`AuthCode` schema change**: `code` becomes the first field. JSON shape changes by one field. Acceptable under greenfield; test fixtures update mechanically. +- **Test approach**: continue the existing pattern — real SlateDB on `object_store::memory::InMemory`, no mocks (per repo `CLAUDE.md` testing posture and existing `auth_tokens.rs:194-201` style). + +## Open Questions + +### Resolved During Planning + +- **Where does `KeyedMutex` live?** → `lib/crates/fabro-store/src/keyed_mutex.rs` (top-level module). +- **Migration order?** → U1 + U2a parallelizable → U2b → U3, U4, U5, U6 parallelizable → U7. See Implementation Units dependency graph. +- **Per-Repository batch helper?** → Deferred, not in this refactor. +- **Module organization for traits?** → New `record/` module at `lib/crates/fabro-store/src/record/`. + +### Deferred to Implementation + +- **Exact internal types for `Tx`** — whether it owns or borrows the `WriteBatch`, whether `put` returns `&mut Self` for chaining. Mechanical; affects no caller. The contract (`FnOnce`, single commit, encode short-circuit) is fixed in R10. +- **Stream adapter for `scan_stream`** — `slatedb::DbIterator` exposes `async fn next(&mut self)` but does not impl `Stream`. Implementer chooses between `futures::stream::unfold`, a hand-rolled `poll_next`, or returning the iterator directly behind a thin newtype. None affect callers; pick the simplest that types cleanly. +- **`gc(predicate)` exact signature for the closure capture lifetime** — `impl Fn(&R) -> bool` may need a `+ Send` bound depending on how the stream-and-batch implementation interleaves with `Send` futures. Implementer adjusts at the type-error site. +- **Whether `BlobStore` lives at `lib/crates/fabro-store/src/slate/blob_store.rs` or a deeper path** — file location is mechanical. + +## High-Level Technical Design + +> *This illustrates the intended approach and is directional guidance for review, not implementation specification. The implementing agent should treat it as context, not code to reproduce.* + +### Trait scaffolding shape + +```text +trait Record: Sized + Send + Sync + 'static { + type Id: RecordId; + type Codec: Codec; + const PREFIX: &'static str; // "/"-separated, split by Repository + fn id(&self) -> Self::Id; +} + +trait RecordId: Sized { + fn key_segments(&self) -> Vec; + fn from_key_segments(segs: &[&str]) -> Result; +} + +trait Codec { + fn encode(r: &R) -> Result>; + fn decode(bytes: &[u8]) -> Result; +} + +// Built-in codecs +struct JsonCodec; // literal forward to serde_json::to_vec / from_slice +struct RawBytesCodec; // identity on bytes — encode(&Blob(bytes)) returns bytes; decode(bytes) returns Blob(bytes) +struct MarkerCodec; // for ZST marker records (R: Default + Sized) — encode(_) -> Vec::new(); decode(&[]) -> R::default() + +// Built-in RecordId impls (each implements both directions) +impl RecordId for [u8; 32] { /* key_segments: vec![hex(self)]; from_key_segments: parse hex */ } +impl RecordId for String { /* key_segments: vec![self.clone()]; from_key_segments: clone segs[0] */ } +impl RecordId for RunBlobId { /* one hex segment, both directions */ } +impl RecordId for RunId { /* key_segments: [, ]; from_key_segments: parse segs[1] */ } +``` + +### Repository shape + +```text +pub(crate) struct Repository { + db: Arc, + _r: PhantomData, +} + +impl Repository { + // Value-carrying API + pub(crate) async fn get(&self, id: &R::Id) -> Result>; + pub(crate) async fn put(&self, r: &R) -> Result<()>; // sugar for put_at(&r.id(), r) + pub(crate) async fn delete(&self, id: &R::Id) -> Result<()>; + pub(crate) fn scan_stream(&self) -> impl Stream>; + pub(crate) fn scan_prefix_stream(&self, extra: &[&str]) -> impl Stream>; + pub(crate) async fn gc(&self, predicate: impl Fn(&R) -> bool) -> Result; + + // Id-only API — works for any record, required for ZST marker records + pub(crate) async fn put_at(&self, id: &R::Id, r: &R) -> Result<()>; + pub(crate) async fn exists(&self, id: &R::Id) -> Result; + pub(crate) fn scan_ids_stream(&self) -> impl Stream>; + pub(crate) fn scan_prefix_ids_stream(&self, extra: &[&str]) -> impl Stream>; +} + +// Marker (index) record example +#[derive(Default)] +pub(crate) struct RunCatalogEntry; // ZST — id lives only in the key + +impl Record for RunCatalogEntry { + type Id = RunId; + type Codec = MarkerCodec; + const PREFIX: &'static str = "runs/_index/by-start"; + fn id(&self) -> RunId { unreachable!("marker — use put_at(&id, &RunCatalogEntry)") } +} + +// RunCatalogIndex usage: +// add(&run_id) -> repo.put_at(&run_id, &RunCatalogEntry).await +// remove(&run_id) -> repo.delete(&run_id).await +// list(query) -> repo.scan_ids_stream().filter(query).sort() +``` + +### transaction shape + +```text +pub(crate) async fn transaction(db: &slatedb::Db, f: F) -> Result +where F: FnOnce(&mut Tx) -> Result +{ + let mut tx = Tx::new(); + let value = f(&mut tx)?; // closure Err → return early, no DB write + db.write(tx.into_batch()).await?; // exactly one commit + Ok(value) +} + +pub(crate) struct Tx { batch: WriteBatch } +impl Tx { + pub(crate) fn put(&mut self, r: &R) -> Result<&mut Self> { ... } + pub(crate) fn delete(&mut self, id: &R::Id) -> Result<&mut Self> { ... } +} +``` + +### Wrapper Store shape (RefreshTokenStore example) + +```text +pub struct RefreshTokenStore { + db: Arc, // for transaction(...) + repo: Repository, // pub(super) at most + consume_locks: KeyedMutex<[u8; 32]>, + replay_revocations: DashMap<[u8; 32], DateTime>, // in-memory only +} + +impl RefreshTokenStore { + pub async fn consume_and_rotate(...) -> Result { + let _guard = self.consume_locks.lock(presented_hash).await; + let existing = self.repo.get(&presented_hash).await?; + match existing { + None => Ok(ConsumeOutcome::NotFound), + Some(token) if now >= token.expires_at => Ok(ConsumeOutcome::Expired), + Some(token) if token.used => Ok(ConsumeOutcome::Reused(token)), + Some(token) => { + let mut old = token.clone(); + old.used = true; + old.last_used_at = now; + transaction(&self.db, |tx| { + tx.put(&old)?; + tx.put(&new_token)?; + Ok(()) + }).await?; + Ok(ConsumeOutcome::Rotated(old, Box::new(new_token))) + } + } + } +} +``` + +## Implementation Units + +```mermaid +graph TB + U1[U1: KeyedMutex extraction] + U2a[U2a: Traits + Repository + transaction] + U2b[U2b: Blob + BlobStore + Database::blobs] + U3[U3: RunDatabase blob delegation] + U4[U4: RunCatalogIndex; delete catalog.rs] + U5[U5: AuthCodeStore migration] + U6[U6: RefreshTokenStore migration; transaction wired in] + U7[U7: Re-exports + fabro-server import cleanup] + + U2a --> U2b + U2b --> U3 + U2a --> U4 + U2a --> U5 + U2a --> U6 + U1 --> U5 + U1 --> U6 + U2b --> U7 + U3 --> U7 + U4 --> U7 + U5 --> U7 + U6 --> U7 +``` + +U1 and U2a are independent of each other and can land in either order. U2b depends on U2a. U3 depends on U2b (it needs `BlobStore`). U4, U5, U6 each depend on U2a (they need `Repository` and `transaction`); U5 and U6 also depend on U1 (`KeyedMutex`). U3, U4, U5, U6 can be parallelized after their dependencies land. U7 is the final cleanup pass. + +--- + +- [x] **Unit 1: Extract `KeyedMutex` shared helper** + +**Goal:** Move the per-key `DashMap>>` + auto-cleanup-at-strong-count-2 pattern out of `auth_tokens.rs` and `auth_codes.rs` into a single shared helper. No behaviour change. + +**Requirements:** R13. + +**Dependencies:** None. Independent precursor PR. + +**Files:** +- Create: `lib/crates/fabro-store/src/keyed_mutex.rs` +- Modify: `lib/crates/fabro-store/src/lib.rs` (add `mod keyed_mutex;` + `pub(crate) use`) +- Modify: `lib/crates/fabro-store/src/slate/auth_tokens.rs` (replace `refresh_locks: DashMap<...>` field + inline lock pattern with `consume_locks: KeyedMutex<[u8; 32]>`) +- Modify: `lib/crates/fabro-store/src/slate/auth_codes.rs` (replace `code_locks: DashMap<...>` similarly) +- Test: `lib/crates/fabro-store/src/keyed_mutex.rs` (unit tests inline) + +**Approach:** +- `KeyedMutex` exposes only `pub async fn lock(&self, key: K) -> Guard<'_>`. +- `Guard<'_>` holds the `Arc>` internally, releases the inner `Mutex` on drop, then performs the cleanup check (strong_count == 2 → remove from `DashMap`). +- The cleanup check and entry insertion happen under the same `DashMap` shard lock — use `DashMap::entry().and_modify().or_insert_with()` or equivalent so a concurrent `lock(&key)` cannot observe a stale Arc. +- No public access to the inner `Arc`. + +**Patterns to follow:** +- `lib/crates/fabro-store/src/slate/auth_tokens.rs:85-118` (today's inline pattern — preserve semantics). +- `lib/crates/fabro-store/src/slate/auth_codes.rs:48-77` (mirror pattern). + +**Test scenarios:** +- *Happy path:* `lock(K1)` then `lock(K2)` proceed independently (no contention between distinct keys). +- *Edge case:* `lock(K1)` from two tasks serializes — second waits until first drops Guard. +- *Edge case:* Auto-cleanup — after the only outstanding Guard drops, the `DashMap` no longer contains the entry for that key (verified via internal accessor or by counting `len()`). +- *Integration:* Stress test — 16 concurrent tasks `lock(K)` on the same key; verify exactly one progresses at a time and the map is empty after all complete. +- *Integration race regression:* Stress test that hammers the same key from N tasks across many drop cycles — verify two threads cannot acquire `lock(&key)` and end up with different `Mutex` instances (cleanup-under-shard-lock invariant). + +**Verification:** +- `cargo nextest run -p fabro-store` passes. +- `auth_tokens.rs` and `auth_codes.rs` no longer carry their inline `refresh_locks` / `code_locks` fields and inline lock plumbing. +- All existing concurrent_consume_has_one_winner / concurrent_rotation_has_one_winner tests pass unmodified. + +--- + +- [x] **Unit 2a: Trait scaffolding (`Record`, `RecordId`, `Codec`, `Repository`, `transaction`)** + +**Goal:** Land the new abstraction with a synthetic test-only Record exercising the full `Repository` API and the `transaction(...)` atomicity contract. No production consumer yet. + +**Requirements:** R1, R2, R3, R4, R5 (boundary spec), R10 (transaction). + +**Dependencies:** None. Independent of U1; either order fine. + +**Files:** +- Create: `lib/crates/fabro-store/src/record/mod.rs` (re-exports `Record`, `RecordId`) +- Create: `lib/crates/fabro-store/src/record/codec.rs` (`trait Codec`, `JsonCodec`, `RawBytesCodec`, `MarkerCodec`) +- Create: `lib/crates/fabro-store/src/record/repository.rs` (`Repository` and methods) +- Create: `lib/crates/fabro-store/src/record/transaction.rs` (`Tx`, `transaction()`) +- Create: `lib/crates/fabro-store/src/record/record_id.rs` (built-in `impl RecordId` for `[u8; 32]`, `String`, `RunId`, `RunBlobId` — both `key_segments` and `from_key_segments` directions) +- Modify: `lib/crates/fabro-store/src/error.rs` — add a new public variant `Error::InvalidKeySegment { segment: String }` returned by `Repository` key-assembly when a `RecordId::key_segments` impl produces a segment containing the `\0` separator byte. Also add `Error::KeyParse(String)` returned by `Repository::scan_stream` / `scan_ids_stream` when `RecordId::from_key_segments` rejects the parsed segments. +- Modify: `lib/crates/fabro-store/src/lib.rs` (add `mod record;`, `pub(crate) use record::...;`) +- Test: `lib/crates/fabro-store/src/record/repository.rs` (unit tests with a synthetic test-only Record to exercise get/put/delete/scan_stream/scan_prefix_stream/gc) +- Test: `lib/crates/fabro-store/src/record/transaction.rs` (unit tests for `transaction(...)` including encode-failure fault injection) +- Test: `lib/crates/fabro-store/src/record/codec.rs` (snapshot test for `JsonCodec` byte-identity using a stable synthetic struct via `insta::assert_snapshot!`) + +**Approach:** +- `Record::PREFIX` is a `&'static str` containing `/`-separated segments (`"auth/refresh"`, `"blobs/sha256"`, etc.). `Repository` splits on `/` once and writes `\0`-separated `SlateKey` segments. `/` is reserved inside any single segment. **Key-assembly invariant (runtime-enforced):** `Repository` checks every segment (PREFIX-derived OR id-derived) for `\0` before assembling the SlateKey, in every operation that constructs a key (`put`/`put_at`/`get`/`delete`/`exists`/scan-prefix). A segment containing `\0` returns `Err(Error::InvalidKeySegment { segment: String })`. This is a runtime error, not a `debug_assert!` — release builds get the same protection. Adds one byte-scan per segment per call (negligible). +- `Repository::scan_stream` adapts `slatedb::DbIterator::next` into `impl Stream>` via `futures::stream::unfold` or hand-rolled `poll_next` (implementer's choice). For each scanned entry: parse the SlateDB key into segments (split on `\0`), strip the `R::PREFIX` segments, pass the remaining segments to `R::Id::from_key_segments` to reconstruct the typed Id; decode the value via `R::Codec::decode`; yield `(id, value)`. For ZST marker records using `MarkerCodec`, decode is trivially `R::default()` — yielding `(id, marker)` where the marker carries no data and the id from the key is the actual information. +- `Repository::scan_ids_stream` is a keys-only variant: same key-parsing as `scan_stream`, but skips value bytes entirely (yields `Stream>`). Required for ZST marker records (where `decode` would be wasted), useful as a perf optimization for any record where the caller only needs ids. +- `Repository::put_at(id, &r)` is the explicit-id put: writes the key derived from `id` (not from `r.id()`). `Repository::put(&r)` is sugar for `put_at(&r.id(), r)`. ZST markers use `put_at` exclusively because they don't carry an id (their `Record::id` is `unreachable!`). +- `Repository::gc(predicate)` scans the prefix, decodes each value, evaluates the (sync, no-I/O) predicate, collects matching keys into a `Vec`, then issues a single `slatedb::WriteBatch` containing all deletes. Returns the count. +- `JsonCodec::encode` is literally `serde_json::to_vec(value)`; `decode` is `serde_json::from_slice(bytes)`. The implementation IS the byte-identity proof. Snapshot test guards against accidental future changes. +- `MarkerCodec::encode(_: &R) -> Ok(Vec::new())`; `decode(bytes) -> Ok(R::default())` (asserts `bytes.is_empty()`). +- `transaction(&db, f)` runs `f(&mut tx)`. On `Err`, returns the error without writing anything. On `Ok`, calls `db.write(tx.into_batch()).await`. `Tx::put` calls `R::Codec::encode(value)` and pushes onto the batch — encode errors propagate via `?` from inside the closure. `Tx::put_at(&id, &r)` is the explicit-id variant for ZST markers. + +**Patterns to follow:** +- `lib/crates/fabro-store/src/slate/mod.rs:208-228` (`OnceCell`-based store accessor on `Database`). +- `lib/crates/fabro-store/src/slate/auth_tokens.rs:79-121` (single `WriteBatch` + single `db.write` — `transaction(...)` mirrors this). +- `lib/crates/fabro-store/src/keys.rs:11-23` (`SlateKey` builder usage — `Repository` calls these `pub(crate)` methods). +- `lib/crates/fabro-store/src/slate/auth_tokens.rs:194-201` (test fixture pattern: `Database::new(Arc::new(InMemory::new()), "", Duration::from_millis(1), None)` then `db.().await.unwrap()`). + +**Test scenarios:** +- *Happy path (Repository):* `put` then `get` returns the same value; `delete` then `get` returns `None`. +- *Happy path (scan_stream):* `put` 5 records, `scan_stream().collect()` yields all 5 with correct `(id, value)` pairs. +- *Happy path (scan_prefix_stream):* records under different sub-prefixes filter correctly. +- *Happy path (gc):* `put` 10 records, `gc(|r| predicate true for half)` deletes 5, returns count 5; `scan_stream` confirms 5 remain. +- *Edge case (Repository):* `get` on missing id returns `None`; `delete` on missing id is a no-op. +- *Edge case (Repository):* empty prefix scan returns empty stream. +- *Error path (Codec):* malformed bytes from `decode` propagate as `Error::Other` (or whatever the existing error type is). +- *Error path (transaction):* closure returns `Err` → `db.write` is NOT called (verify via store state unchanged after). +- *Error path (transaction fault injection):* in the test module, define `TestRecord` with `type Codec = TestFailCodec;` where `TestFailCodec::encode` returns `Err` on a poisoned input. Run `transaction(&db, |tx| { tx.put(&good)?; tx.put(&poisoned)?; Ok(()) })` and assert: (1) the helper returns `Err`, (2) `db.get(key_for(&good))` returns `None`, (3) `db.get(key_for(&poisoned))` returns `None`. This proves the all-or-nothing invariant against the encode-error-mid-batch path (not just the closure-Err short-circuit). **Required by R10.** +- *Edge case (transaction):* empty closure (no puts/deletes) → no batch commit overhead, returns `Ok`. +- *Happy path (MarkerCodec / ZST marker):* define a synthetic `TestMarker` ZST + `impl Record for TestMarker { type Codec = MarkerCodec; ... }`. `repo.put_at(&id, &TestMarker)` then `repo.exists(&id) == true`; `repo.scan_ids_stream().collect()` yields the id; `repo.delete(&id)` then `exists == false`. +- *Edge case (key-assembly invariant):* in release builds, `RecordId::key_segments` returning a string containing `\0` causes the next `Repository` op to return `Err(Error::InvalidKeySegment)` rather than silently corrupting the keyspace. Test asserts the error is returned and the DB is unchanged. +- *Snapshot (JsonCodec byte-identity):* `insta::assert_snapshot!(std::str::from_utf8(&JsonCodec::encode(&sample)?).unwrap())` against a fixed synthetic struct with deterministic timestamps and IDs. Locks the JSON wire format string itself; any future change to `JsonCodec` (envelope, version tag, field reordering) fails the snapshot. + +**Verification:** +- `cargo nextest run -p fabro-store` passes including the new synthetic-record tests. +- `cargo +nightly-2026-04-14 clippy --workspace --all-targets -- -D warnings` passes. +- `Repository` and `transaction` are `pub(crate)`; the `record/` module is internal to `fabro-store`. + +--- + +- [x] **Unit 2b: `Blob` record + `BlobStore` wrapper + `Database::blobs()` accessor** + +**Goal:** First production consumer of the new abstraction. Land `BlobStore` and the `Database::blobs()` accessor. `RunDatabase::write_blob` continues to use the raw DB this unit (delegation lands in U3). + +**Requirements:** R5 (boundary applied to first wrapper), R8 (`BlobStore` API), R11 (`Database::blobs()`). + +**Dependencies:** U2a. + +**Files:** +- Create: `lib/crates/fabro-store/src/slate/blob_store.rs` (`Blob` record + `BlobStore` wrapper) +- Modify: `lib/crates/fabro-store/src/lib.rs` (public re-export of `BlobStore` and `Blob`) +- Modify: `lib/crates/fabro-store/src/slate/mod.rs` (`Database` gets `blobs: Arc>>` field + `Database::blobs() -> Result>` accessor) +- Test: `lib/crates/fabro-store/src/slate/blob_store.rs` (unit tests inline) + +**Approach:** +- `Blob(pub Bytes)` newtype. `impl Record for Blob { type Id = RunBlobId; type Codec = RawBytesCodec; const PREFIX = "blobs/sha256"; fn id(&self) -> RunBlobId { RunBlobId::new(&self.0) } }`. (`RunBlobId::new(content)` is the actual constructor in `lib/crates/fabro-types/src/run_blob_id.rs`; it computes the sha256 internally.) +- `BlobStore` wraps `Repository`. `Repository` field is private. No `.scan_stream()` or `.gc()` exposed. +- `Database::blobs()` follows the existing `auth_codes()`/`auth_tokens()` pattern at `slate/mod.rs:208-228`. + +**Patterns to follow:** +- `lib/crates/fabro-store/src/slate/mod.rs:208-228` (`OnceCell`-based store accessor on `Database`). +- `lib/crates/fabro-store/src/slate/auth_tokens.rs:194-201` (test fixture pattern: `Database::new(Arc::new(InMemory::new()), "", Duration::from_millis(1), None)`). + +**Test scenarios:** +- *Happy path:* `write(bytes_a)` returns `id_a`; `read(&id_a)` returns `Some(bytes_a)`; `write(bytes_a)` again returns the same `id_a` (content-addressed); `exists(&id_a) == true`, `exists(&unknown) == false`. +- *Edge case:* `write(empty bytes)` succeeds; reading back returns `Some(empty)`. +- *Integration (no delete):* `BlobStore` exposes no `delete` — verified at compile time by the absence of the method. +- *Integration (cross-path equivalence with raw DB):* a blob written via `BlobStore::write(bytes)` is byte-identical when read directly from SlateDB via `db.get(blob_key(&id))`. Locks the invariant that `RawBytesCodec::encode` is byte-identity (no envelope, no length prefix). The same equivalence is checked end-to-end in U3 once `RunDatabase::write_blob` delegates. + +**Verification:** +- `cargo nextest run -p fabro-store` passes including new tests. +- `Database::blobs()` returns an `Arc` and supports the round-trip tests. +- Existing tests (including `delete_run_keeps_global_cas_blobs` at `slate/mod.rs:451-466`) still pass — `RunDatabase::write_blob` continues to use the raw DB this unit, so no existing observable behaviour changes. + +--- + +- [x] **Unit 3: `RunDatabase` blob methods delegate to `BlobStore`** + +**Goal:** Internal refactor only — `RunDatabase::write_blob` and `RunDatabase::read_blob` keep their public signatures and delegate to `BlobStore`. `RunDatabase::list_blobs` is **not touched**; it keeps its existing free-function key-scan path at `run_store.rs:365-380` since `BlobStore` intentionally has no `list` method per R8's boundary. No callers change. No tests change. (If a future record needs blob enumeration through `Repository`, `BlobStore::list_ids` could be added later via `Repository::scan_ids_stream`; out of scope for this refactor.) + +**Requirements:** R8 (delegation), R12 (`RunDatabase` public surface unchanged). + +**Dependencies:** U2b. + +**Files:** +- Modify: `lib/crates/fabro-store/src/slate/run_store.rs` — `write_blob` and `read_blob` only. Each constructs a stack-local `BlobStore::new(self.inner.db.clone())` and delegates. `list_blobs` is left exactly as-is. No `RunDatabaseInner` field changes. No constructor signature changes. No `Database::create_run`/`open_run`/`open_run_reader` changes. + +**Approach:** +- `RunDatabase` keeps the raw `Db` field as today. `write_blob` and `read_blob` construct a stack-local `BlobStore` from the existing `Arc` and delegate (~24-byte stack alloc per call; `BlobStore` is stateless). `list_blobs` is unchanged — it continues calling the existing free function at `run_store.rs:365-380`. +- Existing `delete_run_keeps_global_cas_blobs` test (`slate/mod.rs:451-466`) MUST continue to pass — it accesses blobs through `RunDatabase::write_blob`/`read_blob` and the test is the canonical contract that blobs survive run deletion. + +**Patterns to follow:** +- `lib/crates/fabro-store/src/slate/run_store.rs:283-302` (today's `write_blob`/`read_blob` bodies — preserve return types and error behavior; only the implementation changes). +- `lib/crates/fabro-store/src/slate/run_store.rs:365-380` (today's `list_blobs` free function — left untouched). + +**Test scenarios:** +- *Integration:* `RunDatabase::write_blob(bytes)` followed by `RunDatabase::read_blob(&id)` returns `Some(bytes)` (existing behaviour). +- *Integration:* `Database::create_run` → `run.write_blob` → `Database::delete_run` → blob still readable via a different run's `read_blob` (the `delete_run_keeps_global_cas_blobs` invariant). +- *Integration:* `RunDatabase::list_blobs` returns the same set as before for a run that wrote N blobs. + +**Verification:** +- All existing tests in `lib/crates/fabro-store/src/slate/mod.rs` pass unmodified. +- `RunDatabase::write_blob`/`read_blob`/`list_blobs` signatures and behaviour are observably unchanged (no caller in `fabro-workflow`, `fabro-server`, etc. needs updating). +- `cargo nextest run --workspace` passes. + +--- + +- [x] **Unit 4: `RunCatalogIndex` replaces `catalog.rs`** + +**Goal:** Migrate the three free functions in `catalog.rs` to a `RunCatalogIndex` wrapper Store on top of `Repository`. Wire `Database::create_run`/`list_runs`/`delete_run` to use it. Delete `catalog.rs`. + +**Requirements:** R9 (`RunCatalogIndex` API + sort matches today), R12 (catalog calls live on `Database`), R11 (`Database::catalog_index()`). + +**Dependencies:** U2a. + +**Files:** +- Create: `lib/crates/fabro-store/src/slate/run_catalog_index.rs` (`RunCatalogEntry` record + `RunCatalogIndex` wrapper) +- Modify: `lib/crates/fabro-store/src/slate/mod.rs`: + - Add `catalog_index: Arc>>` field on `Database` + - Add `Database::catalog_index()` accessor + - `Database::create_run` (lines 119, 127) replaces `catalog::write_index(&db, run_id).await?` with `self.catalog_index().await?.add(run_id).await?` + - `Database::list_runs` (line 169) replaces `catalog::list_run_ids(&db, query).await?` with `self.catalog_index().await?.list(query).await?` + - `Database::delete_run` (line 204) replaces `catalog::delete_index(&db, run_id).await?` with `self.catalog_index().await?.remove(run_id).await?` +- Delete: `lib/crates/fabro-store/src/slate/catalog.rs` +- Modify: `lib/crates/fabro-store/src/slate/mod.rs` (remove `mod catalog;` import; remove the test-side `use catalog::*` if any) +- Test: `lib/crates/fabro-store/src/slate/run_catalog_index.rs` (unit tests inline) + +**Approach:** +- `RunCatalogEntry` is a ZST marker: `#[derive(Default)] pub(crate) struct RunCatalogEntry;`. `Record::Id = RunId`, `Codec = MarkerCodec`, `PREFIX = "runs/_index/by-start"`. `id(&self)` is `unreachable!("marker — use put_at(&id, &RunCatalogEntry)")` since the marker carries no id. +- `impl RecordId for RunId` writes two segments: `` (from `self.created_at().format("%Y-%m-%d")`) followed by `self.to_string()`. `from_key_segments` parses the ULID from `segs[1]` (the date segment is redundant — derivable from the ULID timestamp). +- `RunCatalogIndex::add(&RunId)` calls `repo.put_at(&run_id, &RunCatalogEntry).await`. Value bytes are empty per `MarkerCodec`. +- `RunCatalogIndex::remove(&RunId)` calls `repo.delete(&run_id).await`. +- `RunCatalogIndex::list(query)` uses `repo.scan_ids_stream()` to enumerate marker keys (no value decoding — cheap), collects into a `Vec`, filters by `query.start`/`query.end` against `run_id.created_at()` (today's logic at `catalog.rs:27-37`), and sorts by `(year, month, day, hour, minute, run_id)` — the same key as `catalog.rs:39-49`. +- `Database::list_runs`'s post-list summary-building loop (lines 170-183) is **NOT** absorbed into `RunCatalogIndex` — only the catalog scan + filter + sort move. The summary loop stays in `Database` because it needs `RunDatabase::has_any_events` and `RunDatabase::build_summary`. + +**Patterns to follow:** +- `lib/crates/fabro-store/src/slate/catalog.rs:7-50` (today's three functions — preserve filter and sort semantics exactly). +- `lib/crates/fabro-store/src/slate/auth_codes.rs::SlateAuthCodeStore::new` (wrapper Store construction signature). +- `lib/crates/fabro-store/src/slate/mod.rs:208-228` (`OnceCell` accessor pattern). + +**Test scenarios:** +- *Happy path:* `add(run_id_1); add(run_id_2); list(default query)` returns both, sorted by `(year, month, day, hour, minute, run_id)` ascending exactly per `catalog.rs:39-49`. Existing `create_open_list_and_delete_full_lifecycle_in_shared_db` test (`slate/mod.rs:421-447`) is the canonical regression — it MUST pass with identical output before and after migration. +- *Edge case:* `list` with `query.start` set later than all `run_id.created_at()` returns empty. +- *Edge case:* `list` with `query.end` set earlier than all `run_id.created_at()` returns empty. +- *Edge case:* `add(run_id); remove(run_id); list()` returns empty. +- *Integration:* `Database::create_run(rid_1); Database::list_runs(default)` returns the run summary (full integration through `Database` — verifies the migration didn't break the surface). +- *Integration:* `Database::delete_run(rid_1); Database::list_runs(default)` no longer returns it. +- *Integration:* All existing tests in `slate/mod.rs::tests` (notably `create_open_list_and_delete_full_lifecycle_in_shared_db`, `delete_run_keeps_global_cas_blobs`, `reopening_store_rebuilds_from_shared_db`) pass unmodified. + +**Verification:** +- `lib/crates/fabro-store/src/slate/catalog.rs` is deleted. +- `cargo nextest run -p fabro-store` passes. +- `Database::list_runs` produces identical results to the pre-refactor state for the same input. + +--- + +- [x] **Unit 5: Migrate `AuthCodeStore` to `Repository`; add `code` field to `AuthCode`** + +**Goal:** Refactor `SlateAuthCodeStore` → `AuthCodeStore` on top of `Repository` and `KeyedMutex`. Add `code: String` as the first field of `AuthCode` so `Record::id` can return it. Update insert/consume signatures and call sites. + +**Requirements:** R7 (API + struct change), R5 (security boundary), R13 (uses KeyedMutex from U1). + +**Dependencies:** U1, U2a. + +**Files:** +- Modify: `lib/crates/fabro-store/src/slate/auth_codes.rs`: + - Rename `SlateAuthCodeStore` → `AuthCodeStore` + - Add `code: String` as first field of `AuthCode` + - `impl Record for AuthCode { type Id = String; type Codec = JsonCodec; const PREFIX = "auth/code"; fn id(&self) -> String { self.code.clone() } }` + - Refactor `insert(code: &str, entry: AuthCode)` → `insert(&self, entry: AuthCode)` per R7 (the `entry` now carries its own `code`; one-arg form is the chosen signature) + - Refactor `consume(&self, code: &str)` to use `self.repo.get(&code.to_string())` then `self.repo.delete(...)` under `KeyedMutex` guard + - Refactor `gc_expired(&self, cutoff)` to use `self.repo.gc(|entry| entry.expires_at <= cutoff)` + - Field changes: drop `db: Arc` (now via `repo`); drop `code_locks: DashMap<...>` (now via `consume_locks: KeyedMutex`) +- Modify: `lib/crates/fabro-store/src/slate/mod.rs`: + - Rename the field `auth_codes: Arc>>` → `Arc>>` + - Update `Database::auth_codes()` return type +- Modify: `lib/crates/fabro-store/src/lib.rs`: drop `SlateAuthCodeStore` from `pub use`; add `AuthCodeStore` and `AuthCode`; **add temporary alias** `pub type SlateAuthCodeStore = AuthCodeStore;` so `fabro-server` keeps compiling until U7 lands. (Without this alias the workspace breaks between U5 and U7.) +- Modify: `lib/crates/fabro-server/src/auth/cli_flow.rs:1226` (OAuth code-mint AuthCode literal — add `code: code.clone()` as first field; reorder so `code` is computed before the literal). After this change, the call switches from `store.insert(&code, entry)` (today's `auth_codes.rs:41`) to `store.insert(entry)` per R7. +- Modify: `lib/crates/fabro-server/src/auth/cli_flow.rs:1390` (test fixture AuthCode literal) +- Modify: `lib/crates/fabro-server/tests/it/api/cli_auth_token.rs:82` (test fixture AuthCode literal) +- Test: `lib/crates/fabro-store/src/slate/auth_codes.rs` (existing test module updates to construct `AuthCode { code: "...", ... }` — `auth_codes.rs:126`) + +**Approach:** +- The `code` field becomes the first field of `AuthCode` so the JSON shape is `{"code":"...","identity":...,...}` — adds one field to the wire format. Greenfield → acceptable. +- `AuthCodeStore::insert(entry)` calls `self.repo.put(&entry)` (no extra arg needed since `entry.code` carries the key). +- `AuthCodeStore::consume(code)` takes `&str`, holds `consume_locks.lock(code.to_string())` guard, then `repo.get(&code.to_string())`, branches on expiry/found, deletes via `repo.delete(&code.to_string())`. Single-use semantics preserved. +- `gc_expired` is one line via `repo.gc(|c| c.expires_at <= cutoff)`. +- Repository field is `pub(super)` per R5 boundary. + +**Patterns to follow:** +- `lib/crates/fabro-store/src/slate/auth_codes.rs:33-100` (preserve API behavior). +- `lib/crates/fabro-store/src/slate/auth_codes.rs:103-200` (existing test patterns — update for new struct shape, otherwise preserve). + +**Test scenarios:** +- *Happy path:* `insert(AuthCode { code, ... })` then `consume(&code)` returns the entry and a second `consume` returns `None` (single-use). +- *Edge case:* `consume` on a non-existent code returns `None`. +- *Edge case:* `consume` on an expired code deletes the row and returns `None` (today's behaviour at `auth_codes.rs:67-71`). +- *Error path:* none surfaced by today's code beyond serde errors — preserve. +- *Integration:* All existing `auth_codes.rs` tests pass with updated struct construction (e.g., `concurrent_consume_has_one_winner` at `auth_codes.rs:152` MUST still pass — KeyedMutex preserves single-winner behaviour). +- *Integration (boundary):* `AuthCodeStore`'s `repo` field is not accessible from outside the wrapper (compile-time check via test that the field is private; or simply: no public method on `AuthCodeStore` returns `&Repository`). + +**Verification:** +- `cargo nextest run -p fabro-store` passes. +- `cargo nextest run -p fabro-server` passes (if server uses `AuthCode` struct literally; update imports per U7 if needed). +- `lib/crates/fabro-store/src/slate/auth_codes.rs` no longer imports `slatedb::Db` directly. + +--- + +- [x] **Unit 6: Migrate `RefreshTokenStore` to `Repository` + `transaction(...)` for `consume_and_rotate`** + +**Goal:** The most complex migration. Refactor `SlateAuthTokenStore` → `RefreshTokenStore` on top of `Repository` and `KeyedMutex<[u8; 32]>`. Replace the hand-built `slatedb::WriteBatch` in `consume_and_rotate` with `transaction(&db, ...)`. Preserve replay revocation cache as in-memory only. R10 atomicity is verified by U2a's transaction-layer fault-injection test (no separate U6 fault-injection test — see Execution note). + +**Requirements:** R6 (API + transaction usage + replay cache invariant), R5 (security boundary), R10 (transaction atomicity), R13 (KeyedMutex). + +**Dependencies:** U1, U2a. + +**Execution note:** `consume_and_rotate` is security-sensitive. The R10 atomicity contract is exercised by U2a's transaction-layer fault-injection test using a synthetic test-only `Record` + `Codec` whose `encode` fails on the second `put` (`RefreshToken` cannot itself trigger a JSON encode failure). U6 keeps the existing `concurrent_rotation_has_one_winner` (`auth_tokens.rs:311`) test as the consume-layer atomicity regression net. + +**Files:** +- Modify: `lib/crates/fabro-store/src/slate/auth_tokens.rs`: + - Rename `SlateAuthTokenStore` → `RefreshTokenStore` + - `impl Record for RefreshToken { type Id = [u8; 32]; type Codec = JsonCodec; const PREFIX = "auth/refresh"; fn id(&self) -> [u8; 32] { self.token_hash } }` + - Refactor `insert_refresh_token(token)` → uses `self.repo.put(&token)` + - Refactor `find_refresh_token(&hash)` → uses `self.repo.get(&hash)` + - Refactor `consume_and_rotate(presented_hash, new_token, now)` → uses `transaction(&self.db, |tx| { tx.put(&old_token)?; tx.put(&new_token)?; Ok(()) })` while holding `consume_locks.lock(presented_hash)` guard. Outcomes (`Rotated`, `Reused`, `Expired`, `NotFound`) preserved exactly. + - Refactor `delete_chain(chain_id)` → uses `self.repo.gc(|t| t.chain_id == chain_id)` + - Refactor `gc_expired(cutoff)` → uses `self.repo.gc(|t| t.expires_at <= cutoff)` + - Field changes: drop `refresh_locks: DashMap<...>` (now `consume_locks: KeyedMutex<[u8; 32]>`); keep `replay_revocations: DashMap<[u8; 32], DateTime>` UNCHANGED (in-memory only per R6). Add a doc-comment on the `replay_revocations` field: `/// 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 to Repository.` so future contributors don't "complete the abstraction" by routing this through a persisted Repository. + - `mark_refresh_token_replay` and `was_recently_replay_revoked` keep their existing impls — they touch only the in-memory `DashMap`. +- Modify: `lib/crates/fabro-store/src/slate/mod.rs` (rename field type; update `Database::auth_tokens()` → `Database::refresh_tokens()` per R11; **add temporary inherent shim** `pub async fn auth_tokens(&self) -> Result> { self.refresh_tokens().await }` so existing `fabro-server` callers compile until U7 removes them.) +- Modify: `lib/crates/fabro-store/src/lib.rs`: drop `SlateAuthTokenStore` from `pub use`; add `RefreshTokenStore` and `RefreshToken`; **add temporary alias** `pub type SlateAuthTokenStore = RefreshTokenStore;` so `fabro-server` keeps compiling until U7 lands. +- Modify: `lib/crates/fabro-store/src/slate/auth_tokens.rs:201` (the in-crate test fixture call `db.auth_tokens().await.unwrap()` → `db.refresh_tokens().await.unwrap()`) +- Test: `lib/crates/fabro-store/src/slate/auth_tokens.rs` (preserve all existing tests; the U2a transaction-layer fault-injection test covers R10) + +**Approach:** +- `RefreshTokenStore` holds `db: Arc` (for `transaction`), `repo: Repository` (`pub(super)`), `consume_locks: KeyedMutex<[u8; 32]>`, `replay_revocations: DashMap<[u8; 32], DateTime>`. +- `consume_and_rotate` flow: + 1. `let _guard = self.consume_locks.lock(presented_hash).await;` + 2. `let existing = self.repo.get(&presented_hash).await?;` + 3. Branch on `None` / expired / used / proceed exactly as today. + 4. For the "proceed" branch: clone `existing` as `old_token`, set `used = true`, `last_used_at = now`. Then `transaction(&self.db, |tx| { tx.put(&old_token)?; tx.put(&new_token)?; Ok(()) }).await?`. + 5. Return `ConsumeOutcome::Rotated(old_token, Box::new(new_token))`. + 6. Mutex strong-count cleanup happens via `KeyedMutex` per U1 — no inline check here. +- Replay cache: `mark_refresh_token_replay` and `was_recently_replay_revoked` stay as-is. R6 explicitly forbids moving them into `Repository`. + +**Patterns to follow:** +- `lib/crates/fabro-store/src/slate/auth_tokens.rs:79-121` (today's `consume_and_rotate` — preserve outcomes and ordering exactly; only the WriteBatch construction moves into `transaction(...)`). +- `lib/crates/fabro-store/src/slate/auth_tokens.rs:123-161` (today's `delete_chain` and `gc_expired` — both collapse to `repo.gc(predicate)` calls). +- `lib/crates/fabro-store/src/slate/auth_tokens.rs:163-178` (replay revocation methods — preserve unchanged). + +**Test scenarios:** +- *Happy path:* `insert_refresh_token(t1); find_refresh_token(&hash) == Some(t1)`. +- *Happy path (rotation):* `insert(old); consume_and_rotate(old_hash, new, now)` returns `Rotated(old_used, new)`; both `find(&old_hash)` and `find(&new_hash)` succeed; `old.used == true`. +- *Edge case (NotFound):* `consume_and_rotate(unknown_hash, ...)` returns `NotFound`. +- *Edge case (Expired):* `consume_and_rotate(expired_hash, ...)` returns `Expired`; old token is NOT marked `used`. +- *Edge case (Reused):* `consume_and_rotate(used_hash, ...)` returns `Reused(existing)`. +- *Integration (concurrency):* 16 concurrent `consume_and_rotate` on the same hash — exactly one returns `Rotated`, 15 return `Reused`. (Existing `concurrent_rotation_has_one_winner` test, MUST pass unmodified.) +- *Integration (delete_chain):* `insert` two tokens with the same `chain_id`, `delete_chain(chain_id)` returns 2, both vanish. +- *Integration (gc_expired):* `insert` an expired token + a live token, `gc_expired(now)` returns 1, expired vanishes, live remains. +- *Integration (replay revocation):* `mark_refresh_token_replay(h)` then `was_recently_replay_revoked(&h)` is true; after 60s TTL, false. +- *Integration (atomicity):* the R10 transaction-layer atomicity test lives in U2a (synthetic Record + Codec). U6 does not duplicate it — `RefreshToken` cannot itself trigger a JSON encode failure without unrealistic fixtures, so the consume-layer test would either reduce to a closure-Err short-circuit (strictly weaker) or require a fragile mock. The U2a test plus `concurrent_rotation_has_one_winner` are the regression net. + +**Verification:** +- All existing tests in `auth_tokens.rs:181-403` pass without behavioural change (the test fixtures may need import-rename updates only). +- `cargo nextest run -p fabro-store` passes. +- `cargo nextest run --workspace` passes. +- `lib/crates/fabro-store/src/slate/auth_tokens.rs` no longer imports `slatedb::WriteBatch` directly. +- `RefreshTokenStore`'s `repo` field is private (compile-time check). + +--- + +- [x] **Unit 7: Update public re-exports + `fabro-server` imports** + +**Goal:** Update `fabro-store::lib.rs` re-exports to the new names; update `fabro-server` to use the new type names. Mechanical cleanup. + +**Requirements:** R11 (Database surface). + +**Dependencies:** U2b, U3, U4, U5, U6. + +**Files:** +- Modify: `lib/crates/fabro-store/src/lib.rs`: + - Drop temporary `pub type SlateAuthCodeStore = AuthCodeStore;` and `pub type SlateAuthTokenStore = RefreshTokenStore;` aliases that U5/U6 introduced + - Add `BlobStore`, `RunCatalogIndex` to `pub use slate::{...}` (the new accessor types from U2b/U4) + - Verify other re-exports (`AuthCode`, `RefreshToken`, `ConsumeOutcome`, `Database`, `RunDatabase`, `Runs`, `AuthCodeStore`, `RefreshTokenStore`) are still correct +- Modify: `lib/crates/fabro-store/src/slate/mod.rs`: drop the temporary `Database::auth_tokens()` shim that U6 introduced (callers now use `refresh_tokens()` directly) +- Modify: `lib/crates/fabro-server/src/serve.rs:799-829` (4 type-name sites: `Arc` → `Arc`, `Arc` → `Arc`) +- Modify: `lib/crates/fabro-server/src/serve.rs:507,508` (2 accessor sites: `db.auth_tokens()` → `db.refresh_tokens()`) +- Modify: `lib/crates/fabro-server/src/auth/cli_flow.rs:380, 462, 541, 660, 1236, 1388, 1660, 1898, 1987, 2042, 2126` (12 accessor sites: `db.auth_tokens()` → `db.refresh_tokens()`; verify each by grep before edit) +- Modify: `lib/crates/fabro-server/src/auth/cli_flow.rs:449, 1039, 1410, 2129` (4 `RefreshToken { ... }` literal sites — should be unchanged in shape since `RefreshToken` struct itself doesn't change, but verify imports) +- Modify: `lib/crates/fabro-server/tests/it/api/cli_auth_token.rs:80, 82, 146, 149` (4 sites: accessor calls + `AuthCode { ... }` and `RefreshToken { ... }` literals — `AuthCode` literal needs `code` field per U5) +- Note: `server.rs`, `jwt_auth.rs`, `auth/mod.rs`, `web_auth.rs` were spot-checked and contain no `SlateAuth*Store` or accessor references — no edits needed there. Final verification grep below catches drift. + +**Approach:** +- Rename via grep-replace, since the types are referenced by qualified name only. +- `Database::auth_tokens()` → `Database::refresh_tokens()` per R11. Update all callers accordingly. +- If any `fabro-server` code constructs `AuthCode` literals (e.g., during the OAuth code flow), add the `code` field to those literals. + +**Patterns to follow:** +- N/A — pure rename + import update. + +**Test scenarios:** +- *Test expectation: none — pure rename / re-export update.* No new behavioural test required. +- *Verification:* `cargo build --workspace` passes; `cargo nextest run --workspace` passes. + +**Verification:** +- `cargo build --workspace` succeeds. +- `cargo nextest run --workspace` passes (including `fabro-server` and `fabro-cli` integration tests). +- `cargo +nightly-2026-04-14 clippy --workspace --all-targets -- -D warnings` passes. +- `cargo +nightly-2026-04-14 fmt --check --all` passes. +- `grep -r "Slate\(Auth\|RefreshToken\)" lib/ apps/` returns no production references (test-fixture references are also updated). +- `grep -rn '\.auth_tokens()' lib/ apps/` returns zero matches (the `auth_tokens()` accessor was renamed to `refresh_tokens()`; this catches any caller the import-update missed, since a stale `.auth_tokens()` would only fail at compile time if no other type in scope happens to expose a method by that name). + +## System-Wide Impact + +- **Interaction graph:** `fabro-server` (`serve.rs`, `auth/mod.rs`, `auth/cli_flow.rs`, `jwt_auth.rs`, `web_auth.rs`) imports `Slate*Store` types from `fabro-store` and calls `Database::auth_codes()` / `auth_tokens()`. All of these need import + accessor-name updates in U7. No `fabro-server` logic changes — purely mechanical. +- **Error propagation:** `fabro_store::Error` gains two new public variants in U2a: `InvalidKeySegment { segment: String }` (returned by `Repository` key-assembly when a `RecordId::key_segments` impl produces a segment containing the `\0` separator) and `KeyParse(String)` (returned by `scan_stream`/`scan_ids_stream` when `RecordId::from_key_segments` rejects the parsed segments). Existing variants are unchanged. New error paths from `JsonCodec::encode` (`serde_json::Error`) and `MarkerCodec` propagate through the same `Result` shape via the existing variants. +- **State lifecycle risks:** + - U2a's `transaction(...)` MUST commit all-or-nothing — covered by R10's fault-injection test in U2a (synthetic Record + TestFailCodec). + - U5's `AuthCode` schema change (adding `code` field) means any in-flight or persisted code from before the migration would not deserialize. Greenfield → no concern in practice; flag in PR description anyway. + - U1's `KeyedMutex` cleanup race (cleanup-under-shard-lock invariant) is the load-bearing concurrency guarantee — covered by U1 stress test. +- **API surface parity:** `Database::auth_tokens()` rename → `refresh_tokens()` is observable. Callers in `fabro-server` are the only consumers; updating them in U7. No external repos consume `fabro-store`. +- **Integration coverage:** Cross-layer scenarios (Database → wrapper Store → Repository → SlateDB) are covered by: + - U2a's transaction fault-injection test (proves R10 atomicity at the helper layer) + - U2b's `BlobStore` round-trip + cross-path equivalence test (proves the full stack works for the simplest record and that `RawBytesCodec` is byte-identity) + - U4's `Database::list_runs` integration test (proves the catalog migration preserves observable behaviour) + - U6's `concurrent_rotation_has_one_winner` (proves KeyedMutex + transaction integration at the consume layer) + - The `delete_run_keeps_global_cas_blobs` test in `slate/mod.rs:451` continues to enforce the cross-store blob invariant. +- **Unchanged invariants:** + - `RunDatabase` event log, projection cache, broadcast channel, `recover_next_seq`, `EventProjectionCache` — all untouched. + - `Database::create_run` / `open_run` / `delete_run` / `list_runs` public signatures and observable behaviour — preserved. + - `RunDatabase::write_blob` / `read_blob` / `list_blobs` public signatures — preserved (delegation is internal). + - All existing `*_db.rs` tests pass without modification (tests touching renamed types update imports; tests touching `AuthCode` literals add the `code` field). + - Wire format for `RefreshToken` JSON is byte-identical to today (`JsonCodec` is literal `serde_json::to_vec` forwarding; U2a's snapshot test on a synthetic struct catches accidental future envelope/version/reordering changes). + - Wire format for `AuthCode` JSON gains a new `code` field per R7/U5 — the value goes from `{"identity":...,"login":...,...}` to `{"code":"...","identity":...,...}`. Acceptable under greenfield (no production deployments). All existing AuthCode rows in any non-empty test/dev SlateDB become un-decodable after upgrade and must be discarded. + +## Risks & Dependencies + +| Risk | Mitigation | +|------|------------| +| `transaction(...)` is implemented with non-atomic semantics (e.g., commits per-put on closure error), silently weakening refresh-token rotation atomicity. | R10 fault-injection test in U2a explicitly asserts an encode error on the Nth `put` leaves the DB unchanged. Implementer cannot ship a non-atomic helper without breaking the test. `concurrent_rotation_has_one_winner` in U6 is the consume-layer regression net. | +| `AuthCode` struct change in U5 breaks `fabro-server` OAuth code flow. | U5 includes a grep for `AuthCode { ` literals; all are updated in the same commit. `cargo build --workspace` failure is the canary. | +| `KeyedMutex` cleanup race: a concurrent `lock(K)` observes a stale Arc after `strong_count == 2` cleanup, leading to two threads holding "the lock for K" against different Mutex instances. | U1 implementation MUST hold the `DashMap` shard lock for both the cleanup check and the entry insertion. U1 includes a stress test that hammers the same key across many drop cycles. | +| `Repository::scan_stream` cannot reconstruct `R::Id` from values for records whose Id isn't a field (e.g. ZST markers, content-addressed records). | Resolved at plan time: `RecordId::from_key_segments` parses Id from the key, and `Repository::scan_ids_stream` enumerates ids without touching values. ZST markers (`MarkerCodec`) use the id-only API exclusively. | +| `RunDatabase` blob delegation in U3 changes construction of `RunDatabase` inside `Database::create_run`/`open_run`/`open_run_reader`, breaking subtle ordering assumptions. | Resolved at plan time: U3 uses stack-local `BlobStore` construction inside each blob method — no `RunDatabaseInner` field changes, no constructor signature changes. Existing `delete_run_keeps_global_cas_blobs` and full-lifecycle tests are the regression net. | +| `gc(predicate)` predicate is `impl Fn(&R) -> bool` — a future caller passes an `async` closure or one that does I/O, blocking the scan. | R4 explicitly states the predicate is sync `Fn` and MUST NOT perform I/O or block. Documented; relies on convention + code review (no compile-time enforcement). | +| `JsonCodec::encode` is "literally `serde_json::to_vec`" but a future maintainer adds an envelope/version tag for "v2" without realizing it breaks every existing on-disk record. | U2a's snapshot test on a stable synthetic struct fails loudly on any byte change to JsonCodec output (envelope, version tag, field reordering). PR review should catch the snapshot acceptance. | +| Migration order assumption: U2b lands and exposes `BlobStore` publicly before U3 wires `RunDatabase` to it; an external caller could start using `Database::blobs()` directly between U2b and U3. | Acceptable — both paths produce identical results; cross-path equivalence is locked by U2b's test (RawBytesCodec is byte-identity). After U3, `RunDatabase::write_blob` and `Database::blobs().write` are equivalent surfaces. | +| Pinning `Repository` to `pub(crate)` is by convention only against future in-crate contributors — nothing prevents a new module under `fabro-store/src/` from constructing a second `Repository` and bypassing `RefreshTokenStore`. | Documented in R5: "for crate-internal contributors it is convention + code review." Considered alternative (sealed-per-record module with `pub(super)`) was rejected in pass-2 review for being heavier than the threat warrants. | + +## Documentation / Operational Notes + +- No external docs to update — `fabro-store` has no public-facing documentation in `docs/`. +- No CLI changes; no rollout/monitoring impact. +- No migration script needed — greenfield, on-disk format for `RefreshToken` is byte-identical (locked by snapshot); `AuthCode` schema change in U5 is an additive field, but greenfield means existing dev databases are throwaway. +- After U7 lands, a follow-up issue may be filed for: (a) per-Repository batch helper (deferred from origin Q3); (b) `BlobStore` lifecycle question if the simpler stack-local construction in U3 ages poorly. + +## Verification Notes (2026-04-21) + +Post-implementation audit against the plan. All 7 units are implemented; build, nextest (`fabro-store` 68 / `fabro-server` 359), clippy (nightly-2026-04-14 `-D warnings`), and fmt pass. U7 grep checks (`Slate(Auth|RefreshToken)`, `.auth_tokens()`) return zero hits. `slate/catalog.rs` is deleted. + +Minor gaps that do not affect correctness or production surface: + +- **U2a — `scan_prefix_ids_stream` not implemented.** Appears in the Repository *design* sketch (line 186) but is not listed in R4. No current caller. Add lazily if a future consumer needs prefix-scoped id enumeration. +- **U4 — two edge-case tests from plan not individually present.** `run_catalog_index.rs` tests cover add/list/remove round-trip and start/end filter, but not (a) `add(rid); remove(rid); list()` returns empty, nor (b) start-later-than-all / end-earlier-than-all returning empty. The integration paths in `slate::tests` and `list_applies_start_and_end_filters` exercise the same code paths, so no functional regression risk. +- **U3 — `BlobStore::new(Arc::new(self.inner.db.clone()))` shape.** Plan suggested stack-local `BlobStore::new(self.inner.db.clone())`, but `BlobStore::new` requires `Arc` so the call wraps in `Arc::new`. Functionally equivalent; cosmetic. + +## Sources & References + +- **Origin document:** `docs/brainstorms/2026-04-20-fabro-store-record-abstractions-requirements.md` +- Today's K/V stores: + - `lib/crates/fabro-store/src/slate/auth_tokens.rs` + - `lib/crates/fabro-store/src/slate/auth_codes.rs` + - `lib/crates/fabro-store/src/slate/catalog.rs` + - `lib/crates/fabro-store/src/slate/run_store.rs` +- Database accessor pattern: `lib/crates/fabro-store/src/slate/mod.rs:208-228` +- Key construction: `lib/crates/fabro-store/src/keys.rs` +- Public re-exports today: `lib/crates/fabro-store/src/lib.rs` +- Server callers: `lib/crates/fabro-server/src/serve.rs:799-829`, `lib/crates/fabro-server/src/server.rs`, `lib/crates/fabro-server/src/auth/mod.rs` +- Workspace build commands and test posture: `CLAUDE.md` (`cargo nextest`, `cargo +nightly-2026-04-14 clippy`, `cargo +nightly-2026-04-14 fmt`) diff --git a/lib/crates/fabro-server/src/auth/cli_flow.rs b/lib/crates/fabro-server/src/auth/cli_flow.rs index 6d6df73ac..0fa1d163a 100644 --- a/lib/crates/fabro-server/src/auth/cli_flow.rs +++ b/lib/crates/fabro-server/src/auth/cli_flow.rs @@ -459,7 +459,7 @@ async fn token( used: false, user_agent: sanitize_user_agent(request_user_agent(&headers)), }; - let auth_tokens = match state.store_ref().auth_tokens().await { + let auth_tokens = match state.store_ref().refresh_tokens().await { Ok(store) => store, Err(err) => { warn!(error = %err, "Failed to open refresh token store"); @@ -538,7 +538,7 @@ async fn refresh( "Could not refresh authentication", ); }; - let auth_tokens = match state.store_ref().auth_tokens().await { + let auth_tokens = match state.store_ref().refresh_tokens().await { Ok(store) => store, Err(err) => { warn!(error = %err, "Failed to open refresh token store"); @@ -657,7 +657,7 @@ async fn logout( let Some(secret) = refresh_secret_from_headers(&headers) else { return StatusCode::NO_CONTENT.into_response(); }; - let auth_tokens = match state.store_ref().auth_tokens().await { + let auth_tokens = match state.store_ref().refresh_tokens().await { Ok(store) => store, Err(err) => { warn!(error = %err, "Failed to open refresh token store"); @@ -1224,6 +1224,7 @@ async fn issue_auth_code_response( return static_error_page(INVALID_REDIRECT_URI); }; let entry = AuthCode { + code: code.clone(), identity, login: session.login.clone(), name: session.name.clone(), @@ -1246,7 +1247,7 @@ async fn issue_auth_code_response( } }; - if let Err(err) = store.insert(&code, entry).await { + if let Err(err) = store.insert(entry).await { warn!(error = %err, "Failed to persist auth code"); return redirect_with_error( &redirect_uri, @@ -1387,7 +1388,8 @@ mod tests { async fn insert_auth_code(state: &crate::server::AppState, code: &str, verifier: &str) { let auth_codes = state.store_ref().auth_codes().await.unwrap(); auth_codes - .insert(code, AuthCode { + .insert(AuthCode { + code: code.to_string(), identity: fabro_types::IdpIdentity::new("https://github.com", "12345") .expect("identity should be valid"), login: "octocat".to_string(), @@ -1895,7 +1897,7 @@ mod tests { .unwrap() .strip_prefix("fabro_refresh_") .unwrap(); - let auth_tokens = state.store_ref().auth_tokens().await.unwrap(); + let auth_tokens = state.store_ref().refresh_tokens().await.unwrap(); let refresh = auth_tokens .find_refresh_token(&hash_refresh_secret(refresh_secret)) .await @@ -1984,7 +1986,7 @@ mod tests { async fn refresh_rotates_tokens_and_replay_revokes_chain() { let (app, state) = test_router(github_settings("https://fabro.example")); let initial_secret = "refresh-secret-1"; - let auth_tokens = state.store_ref().auth_tokens().await.unwrap(); + let auth_tokens = state.store_ref().refresh_tokens().await.unwrap(); auth_tokens .insert_refresh_token(refresh_row(initial_secret)) .await @@ -2039,7 +2041,7 @@ mod tests { async fn concurrent_refresh_has_one_winner_and_revokes_chain() { let (app, state) = test_router(github_settings("https://fabro.example")); let initial_secret = "refresh-secret-concurrent"; - let auth_tokens = state.store_ref().auth_tokens().await.unwrap(); + let auth_tokens = state.store_ref().refresh_tokens().await.unwrap(); auth_tokens .insert_refresh_token(refresh_row(initial_secret)) .await @@ -2123,7 +2125,7 @@ mod tests { let secret = "refresh-secret-logout"; let token = refresh_row(secret); let chain_id = token.chain_id; - let auth_tokens = state.store_ref().auth_tokens().await.unwrap(); + let auth_tokens = state.store_ref().refresh_tokens().await.unwrap(); auth_tokens.insert_refresh_token(token).await.unwrap(); let sibling = RefreshToken { diff --git a/lib/crates/fabro-server/src/serve.rs b/lib/crates/fabro-server/src/serve.rs index cbb35d066..7b4b2ba96 100644 --- a/lib/crates/fabro-server/src/serve.rs +++ b/lib/crates/fabro-server/src/serve.rs @@ -505,7 +505,7 @@ where cache_path, )); let auth_code_store = store.auth_codes().await?; - let auth_token_store = store.auth_tokens().await?; + let auth_token_store = store.refresh_tokens().await?; let (artifact_object_store, artifact_prefix) = build_artifact_object_store(&resolved_server_settings)?; let artifact_store = fabro_store::ArtifactStore::new(artifact_object_store, artifact_prefix); @@ -796,8 +796,8 @@ async fn wait_for_shutdown(mut shutdown_rx: watch::Receiver) { } fn spawn_auth_store_reapers( - auth_codes: Arc, - auth_tokens: Arc, + auth_codes: Arc, + auth_tokens: Arc, shutdown_rx: watch::Receiver, ) { spawn_auth_code_reaper(auth_codes, shutdown_rx.clone()); @@ -805,7 +805,7 @@ fn spawn_auth_store_reapers( } fn spawn_auth_code_reaper( - auth_codes: Arc, + auth_codes: Arc, mut shutdown_rx: watch::Receiver, ) { tokio::spawn(async move { @@ -826,7 +826,7 @@ fn spawn_auth_code_reaper( } fn spawn_refresh_token_reaper( - auth_tokens: Arc, + auth_tokens: Arc, mut shutdown_rx: watch::Receiver, ) { tokio::spawn(async move { diff --git a/lib/crates/fabro-server/tests/it/api/cli_auth_token.rs b/lib/crates/fabro-server/tests/it/api/cli_auth_token.rs index 818cbd928..1fb8ab3ea 100644 --- a/lib/crates/fabro-server/tests/it/api/cli_auth_token.rs +++ b/lib/crates/fabro-server/tests/it/api/cli_auth_token.rs @@ -79,7 +79,8 @@ client_id = "Iv1.test" )); let auth_codes = store.auth_codes().await.unwrap(); auth_codes - .insert("integration-code", AuthCode { + .insert(AuthCode { + code: "integration-code".to_string(), identity: fabro_types::IdpIdentity::new("https://github.com", "12345").unwrap(), login: "octocat".to_string(), name: "The Octocat".to_string(), @@ -143,7 +144,7 @@ url = "https://fabro.example" client_id = "Iv1.test" "#, )); - let auth_tokens = store.auth_tokens().await.unwrap(); + let auth_tokens = store.refresh_tokens().await.unwrap(); let now = chrono::Utc::now(); auth_tokens .insert_refresh_token(RefreshToken { diff --git a/lib/crates/fabro-store/Cargo.toml b/lib/crates/fabro-store/Cargo.toml index ce458b884..6b3fe893b 100644 --- a/lib/crates/fabro-store/Cargo.toml +++ b/lib/crates/fabro-store/Cargo.toml @@ -33,3 +33,4 @@ uuid.workspace = true tokio = { workspace = true, features = ["test-util", "macros"] } tempfile = "3" ulid.workspace = true +insta = { workspace = true } diff --git a/lib/crates/fabro-store/src/error.rs b/lib/crates/fabro-store/src/error.rs index f73aa472a..e66da4134 100644 --- a/lib/crates/fabro-store/src/error.rs +++ b/lib/crates/fabro-store/src/error.rs @@ -16,6 +16,10 @@ pub enum Error { RunAlreadyExists(String), #[error("run store is read-only")] ReadOnly, + #[error("invalid key segment: {segment:?}")] + InvalidKeySegment { segment: String }, + #[error("failed to parse key: {0}")] + KeyParse(String), #[error("{0}")] Other(String), } diff --git a/lib/crates/fabro-store/src/keyed_mutex.rs b/lib/crates/fabro-store/src/keyed_mutex.rs new file mode 100644 index 000000000..576ce3352 --- /dev/null +++ b/lib/crates/fabro-store/src/keyed_mutex.rs @@ -0,0 +1,180 @@ +use std::hash::Hash; + +use dashmap::DashMap; +use dashmap::mapref::entry::Entry; +use tokio::sync::{Mutex, OwnedMutexGuard}; + +#[derive(Debug)] +pub(crate) struct KeyedMutex +where + K: Eq + Hash + Clone, +{ + mutexes: DashMap>>, +} + +impl Default for KeyedMutex +where + K: Eq + Hash + Clone, +{ + fn default() -> Self { + Self::new() + } +} + +impl KeyedMutex +where + K: Eq + Hash + Clone, +{ + pub(crate) fn new() -> Self { + Self { + mutexes: DashMap::new(), + } + } + + #[cfg(test)] + pub(crate) fn len(&self) -> usize { + self.mutexes.len() + } +} + +impl KeyedMutex +where + K: Eq + Hash + Clone, +{ + pub(crate) async fn lock(&self, key: K) -> Guard<'_, K> { + let mutex = self + .mutexes + .entry(key.clone()) + .or_insert_with(|| std::sync::Arc::new(Mutex::new(()))) + .clone(); + + Guard { + keyed_mutex: self, + key, + held_lock: Some(mutex.lock_owned().await), + } + } +} + +pub(crate) struct Guard<'a, K> +where + K: Eq + Hash + Clone, +{ + keyed_mutex: &'a KeyedMutex, + key: K, + held_lock: Option>, +} + +impl Drop for Guard<'_, K> +where + K: Eq + Hash + Clone, +{ + fn drop(&mut self) { + self.held_lock.take(); + + if let Entry::Occupied(entry) = self.keyed_mutex.mutexes.entry(self.key.clone()) { + if std::sync::Arc::strong_count(entry.get()) == 1 { + entry.remove(); + } + } + } +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; + use std::time::Duration; + + use tokio::task::{JoinSet, yield_now}; + use tokio::time::{sleep, timeout}; + + use super::KeyedMutex; + + #[tokio::test] + async fn distinct_keys_do_not_contend() { + let keyed_mutex = KeyedMutex::new(); + let _first = keyed_mutex.lock("alpha".to_string()).await; + + timeout( + Duration::from_millis(50), + keyed_mutex.lock("beta".to_string()), + ) + .await + .expect("distinct keys should not block"); + } + + #[tokio::test] + async fn same_key_serializes_access() { + let keyed_mutex = Arc::new(KeyedMutex::new()); + let first = keyed_mutex.lock("alpha".to_string()).await; + let acquired = Arc::new(AtomicBool::new(false)); + + let worker_mutex = Arc::clone(&keyed_mutex); + let worker_acquired = Arc::clone(&acquired); + let waiter = tokio::spawn(async move { + let _second = worker_mutex.lock("alpha".to_string()).await; + worker_acquired.store(true, Ordering::SeqCst); + }); + + sleep(Duration::from_millis(10)).await; + assert!( + !acquired.load(Ordering::SeqCst), + "second waiter should block while first guard is held" + ); + + drop(first); + timeout(Duration::from_millis(100), waiter) + .await + .expect("waiter should proceed after first guard drops") + .unwrap(); + assert!(acquired.load(Ordering::SeqCst)); + } + + #[tokio::test] + async fn drops_unused_entries_after_last_guard_releases() { + let keyed_mutex = KeyedMutex::new(); + let guard = keyed_mutex.lock("alpha".to_string()).await; + + assert_eq!(keyed_mutex.len(), 1); + drop(guard); + + assert_eq!(keyed_mutex.len(), 0); + } + + #[tokio::test] + async fn stress_same_key_keeps_single_mutex_and_cleans_up() { + let keyed_mutex = Arc::new(KeyedMutex::new()); + let active = Arc::new(AtomicUsize::new(0)); + let max_active = Arc::new(AtomicUsize::new(0)); + let violations = Arc::new(AtomicUsize::new(0)); + let mut tasks = JoinSet::new(); + + for _ in 0..16 { + let keyed_mutex = Arc::clone(&keyed_mutex); + let active = Arc::clone(&active); + let max_active = Arc::clone(&max_active); + let violations = Arc::clone(&violations); + tasks.spawn(async move { + for _ in 0..64 { + let _guard = keyed_mutex.lock("alpha".to_string()).await; + let current = active.fetch_add(1, Ordering::SeqCst) + 1; + max_active.fetch_max(current, Ordering::SeqCst); + if current != 1 { + violations.fetch_add(1, Ordering::SeqCst); + } + yield_now().await; + active.fetch_sub(1, Ordering::SeqCst); + } + }); + } + + while let Some(result) = tasks.join_next().await { + result.unwrap(); + } + + assert_eq!(violations.load(Ordering::SeqCst), 0); + assert_eq!(max_active.load(Ordering::SeqCst), 1); + assert_eq!(keyed_mutex.len(), 0); + } +} diff --git a/lib/crates/fabro-store/src/keys.rs b/lib/crates/fabro-store/src/keys.rs index d3bcb7d76..44b1a9753 100644 --- a/lib/crates/fabro-store/src/keys.rs +++ b/lib/crates/fabro-store/src/keys.rs @@ -8,17 +8,17 @@ pub(crate) struct SlateKey(String); impl SlateKey { const SEP: char = '\0'; - fn new(segment: impl fmt::Display) -> Self { + pub(crate) fn new(segment: impl fmt::Display) -> Self { Self(segment.to_string()) } - fn with(mut self, segment: impl fmt::Display) -> Self { + pub(crate) fn with(mut self, segment: impl fmt::Display) -> Self { self.0.push(Self::SEP); write!(&mut self.0, "{segment}").expect("write to String cannot fail"); self } - fn into_prefix(mut self) -> Self { + pub(crate) fn into_prefix(mut self) -> Self { self.0.push(Self::SEP); self } @@ -28,7 +28,7 @@ impl SlateKey { &self.0 } - fn segments(raw: &str) -> impl Iterator { + pub(crate) fn segments(raw: &str) -> impl Iterator { raw.split(Self::SEP) } } @@ -41,21 +41,6 @@ impl AsRef<[u8]> for SlateKey { // --- Construction --- -pub(crate) fn runs_index_by_start_prefix() -> SlateKey { - SlateKey::new("runs") - .with("_index") - .with("by-start") - .into_prefix() -} - -pub(crate) fn runs_index_by_start_key(run_id: &RunId) -> SlateKey { - SlateKey::new("runs") - .with("_index") - .with("by-start") - .with(run_id.created_at().format("%Y-%m-%d")) - .with(run_id) -} - pub(crate) fn run_data_prefix(run_id: &RunId) -> SlateKey { SlateKey::new("runs").with(run_id).into_prefix() } @@ -78,30 +63,6 @@ pub(crate) fn blobs_prefix() -> SlateKey { SlateKey::new("blobs").with("sha256").into_prefix() } -pub(crate) fn blob_key(id: &RunBlobId) -> SlateKey { - SlateKey::new("blobs").with("sha256").with(id) -} - -pub(crate) fn auth_code_prefix() -> SlateKey { - SlateKey::new("auth").with("code").into_prefix() -} - -pub(crate) fn auth_code_key(code: &str) -> SlateKey { - SlateKey::new("auth").with("code").with(code) -} - -pub(crate) fn auth_refresh_prefix() -> SlateKey { - SlateKey::new("auth").with("refresh").into_prefix() -} - -pub(crate) fn auth_refresh_key(token_hash: &[u8; 32]) -> SlateKey { - let mut encoded = String::with_capacity(token_hash.len() * 2); - for byte in token_hash { - write!(&mut encoded, "{byte:02x}").expect("write to String cannot fail"); - } - SlateKey::new("auth").with("refresh").with(encoded) -} - // --- Parsing --- pub(crate) fn parse_event_seq(key: &str) -> Option { @@ -129,15 +90,6 @@ pub(crate) fn parse_blob_id(key: &str) -> Option { id.parse().ok() } -pub(crate) fn parse_run_id_from_index_key(key: &str) -> Option { - let mut segments = SlateKey::segments(key); - let _ = segments.next()?; // "runs" - let _ = segments.next()?; // "_index" - let _ = segments.next()?; // "by-start" - let _ = segments.next()?; // date - segments.next()?.parse().ok() -} - #[cfg(test)] mod tests { use fabro_types::RunId; @@ -169,24 +121,10 @@ mod tests { ]); } - #[test] - fn index_key_segments() { - let run_id: RunId = "01JT56VE4Z5NZ814GZN2JZD65A".parse().unwrap(); - let key = runs_index_by_start_key(&run_id); - let segments: Vec<&str> = SlateKey::segments(key.as_str()).collect(); - assert_eq!(segments, [ - "runs", - "_index", - "by-start", - &run_id.created_at().format("%Y-%m-%d").to_string(), - &run_id.to_string(), - ]); - } - #[test] fn blob_key_segments() { let blob_id = RunBlobId::new(b"summary"); - let key = blob_key(&blob_id); + let key = SlateKey::new("blobs").with("sha256").with(blob_id); let segments: Vec<&str> = SlateKey::segments(key.as_str()).collect(); assert_eq!(segments, ["blobs", "sha256", &blob_id.to_string()]); } @@ -208,12 +146,8 @@ mod tests { ); let blob_id = RunBlobId::new(b"summary"); - assert_eq!(parse_blob_id(blob_key(&blob_id).as_str()), Some(blob_id)); - - assert_eq!( - parse_run_id_from_index_key(runs_index_by_start_key(&run_id).as_str()), - Some(run_id) - ); + let key = SlateKey::new("blobs").with("sha256").with(blob_id); + assert_eq!(parse_blob_id(key.as_str()), Some(blob_id)); } #[test] diff --git a/lib/crates/fabro-store/src/lib.rs b/lib/crates/fabro-store/src/lib.rs index 55ff6a427..1bb69834a 100644 --- a/lib/crates/fabro-store/src/lib.rs +++ b/lib/crates/fabro-store/src/lib.rs @@ -2,7 +2,9 @@ use chrono::{DateTime, Utc}; mod artifact_store; mod error; +mod keyed_mutex; mod keys; +mod record; mod run_state; mod slate; mod types; @@ -10,10 +12,11 @@ mod types; pub use artifact_store::{ArtifactStore, NodeArtifact}; pub use error::{Error, Result}; pub use fabro_types::{RunBlobId, StageId}; +pub(crate) use keyed_mutex::KeyedMutex; pub use run_state::{NodeState, PendingInterviewRecord, RunProjection}; pub use slate::{ - AuthCode, ConsumeOutcome, Database, RefreshToken, RunDatabase, Runs, SlateAuthCodeStore, - SlateAuthTokenStore, + AuthCode, AuthCodeStore, Blob, BlobStore, ConsumeOutcome, Database, RefreshToken, + RefreshTokenStore, RunCatalogIndex, RunDatabase, Runs, }; pub use types::{EventEnvelope, EventPayload, RunSummary}; diff --git a/lib/crates/fabro-store/src/record/codec.rs b/lib/crates/fabro-store/src/record/codec.rs new file mode 100644 index 000000000..2a1eef181 --- /dev/null +++ b/lib/crates/fabro-store/src/record/codec.rs @@ -0,0 +1,95 @@ +use bytes::Bytes; +use serde::Serialize; +use serde::de::DeserializeOwned; + +use crate::{Error, Result}; + +pub(crate) trait Codec: Send + Sync + 'static { + fn encode(value: &R) -> Result>; + + fn decode(bytes: &[u8]) -> Result; +} + +pub(crate) struct JsonCodec; + +impl Codec for JsonCodec +where + R: Serialize + DeserializeOwned, +{ + fn encode(value: &R) -> Result> { + serde_json::to_vec(value).map_err(Into::into) + } + + fn decode(bytes: &[u8]) -> Result { + serde_json::from_slice(bytes).map_err(Into::into) + } +} + +pub(crate) struct RawBytesCodec; + +impl Codec for RawBytesCodec +where + R: AsRef<[u8]> + From, +{ + fn encode(value: &R) -> Result> { + Ok(value.as_ref().to_vec()) + } + + fn decode(bytes: &[u8]) -> Result { + Ok(R::from(Bytes::copy_from_slice(bytes))) + } +} + +pub(crate) struct MarkerCodec; + +impl Codec for MarkerCodec +where + R: Default, +{ + fn encode(_: &R) -> Result> { + Ok(Vec::new()) + } + + fn decode(bytes: &[u8]) -> Result { + if bytes.is_empty() { + return Ok(R::default()); + } + Err(Error::Other( + "marker records must decode from an empty byte slice".to_string(), + )) + } +} + +#[cfg(test)] +mod tests { + use chrono::{TimeZone, Utc}; + use serde::{Deserialize, Serialize}; + + use super::{Codec, JsonCodec}; + + #[derive(Debug, Serialize, Deserialize)] + struct SnapshotRecord { + code: String, + issued_at: chrono::DateTime, + expires_at: chrono::DateTime, + attempts: u32, + } + + #[test] + fn json_codec_matches_snapshot() { + let record = SnapshotRecord { + code: "code-123".to_string(), + issued_at: Utc.with_ymd_and_hms(2026, 4, 20, 12, 34, 56).unwrap(), + expires_at: Utc.with_ymd_and_hms(2026, 4, 20, 12, 39, 56).unwrap(), + attempts: 2, + }; + + let encoded = JsonCodec::encode(&record).unwrap(); + let encoded = std::str::from_utf8(&encoded).unwrap(); + + insta::assert_snapshot!( + encoded, + @"{\"code\":\"code-123\",\"issued_at\":\"2026-04-20T12:34:56Z\",\"expires_at\":\"2026-04-20T12:39:56Z\",\"attempts\":2}" + ); + } +} diff --git a/lib/crates/fabro-store/src/record/mod.rs b/lib/crates/fabro-store/src/record/mod.rs new file mode 100644 index 000000000..9c455c205 --- /dev/null +++ b/lib/crates/fabro-store/src/record/mod.rs @@ -0,0 +1,25 @@ +mod codec; +mod record_id; +mod repository; +mod transaction; + +pub(crate) use codec::{Codec, JsonCodec, MarkerCodec, RawBytesCodec}; +pub(crate) use repository::Repository; +pub(crate) use transaction::transaction; + +use crate::Result; + +pub(crate) trait Record: Sized + Send + Sync + 'static { + type Id: RecordId; + type Codec: Codec; + + const PREFIX: &'static str; + + fn id(&self) -> Self::Id; +} + +pub(crate) trait RecordId: Sized { + fn key_segments(&self) -> Vec; + + fn from_key_segments(segs: &[&str]) -> Result; +} diff --git a/lib/crates/fabro-store/src/record/record_id.rs b/lib/crates/fabro-store/src/record/record_id.rs new file mode 100644 index 000000000..21a483a67 --- /dev/null +++ b/lib/crates/fabro-store/src/record/record_id.rs @@ -0,0 +1,97 @@ +use std::fmt::Write; + +use fabro_types::{RunBlobId, RunId}; + +use super::RecordId; +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] + } + + fn from_key_segments(segs: &[&str]) -> Result { + let [segment] = segs else { + return Err(Error::KeyParse(format!( + "expected 1 segment for [u8; 32], got {}", + 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}")))?; + } + Ok(bytes) + } +} + +impl RecordId for String { + fn key_segments(&self) -> Vec { + vec![self.clone()] + } + + fn from_key_segments(segs: &[&str]) -> Result { + let [segment] = segs else { + return Err(Error::KeyParse(format!( + "expected 1 segment for String, got {}", + segs.len() + ))); + }; + Ok((*segment).to_string()) + } +} + +impl RecordId for RunBlobId { + fn key_segments(&self) -> Vec { + vec![self.to_string()] + } + + fn from_key_segments(segs: &[&str]) -> Result { + let [segment] = segs else { + return Err(Error::KeyParse(format!( + "expected 1 segment for RunBlobId, got {}", + segs.len() + ))); + }; + segment + .parse() + .map_err(|err| Error::KeyParse(format!("invalid RunBlobId segment {segment:?}: {err}"))) + } +} + +impl RecordId for RunId { + fn key_segments(&self) -> Vec { + vec![ + self.created_at().format("%Y-%m-%d").to_string(), + self.to_string(), + ] + } + + fn from_key_segments(segs: &[&str]) -> Result { + if segs.len() != 2 { + return Err(Error::KeyParse(format!( + "expected 2 segments for RunId, got {}", + segs.len() + ))); + } + segs[1] + .parse() + .map_err(|err| Error::KeyParse(format!("invalid RunId segment {:?}: {err}", segs[1]))) + } +} diff --git a/lib/crates/fabro-store/src/record/repository.rs b/lib/crates/fabro-store/src/record/repository.rs new file mode 100644 index 000000000..ae77e103e --- /dev/null +++ b/lib/crates/fabro-store/src/record/repository.rs @@ -0,0 +1,504 @@ +use std::marker::PhantomData; +use std::pin::Pin; +use std::sync::Arc; + +use futures::stream::{self}; +use futures::{Stream, StreamExt}; +use slatedb::{Db, KeyValue, WriteBatch}; + +use super::{Codec, Record, RecordId}; +use crate::{Error, Result, keys}; + +pub(crate) struct Repository { + db: Arc, + prefix_segments: Vec<&'static str>, + _record: PhantomData, +} + +impl Repository { + pub(crate) fn new(db: Arc) -> Self { + Self { + db, + prefix_segments: prefix_segments::(), + _record: PhantomData, + } + } + + pub(crate) async fn get(&self, id: &R::Id) -> Result> { + self.db + .get(key_for_id::(id)?) + .await? + .map(|bytes| R::Codec::decode(&bytes)) + .transpose() + } + + pub(crate) async fn put(&self, record: &R) -> Result<()> { + let id = record.id(); + self.put_at(&id, record).await + } + + pub(crate) async fn put_at(&self, id: &R::Id, record: &R) -> Result<()> { + self.db + .put(key_for_id::(id)?, R::Codec::encode(record)?) + .await?; + Ok(()) + } + + pub(crate) async fn delete(&self, id: &R::Id) -> Result<()> { + self.db.delete(key_for_id::(id)?).await?; + Ok(()) + } + + pub(crate) async fn exists(&self, id: &R::Id) -> Result { + Ok(self.db.get(key_for_id::(id)?).await?.is_some()) + } + + #[allow( + dead_code, + reason = "Part of the shared Repository surface; current consumers do not need value scans yet" + )] + pub(crate) fn scan_stream(&self) -> RepositoryStream<'_, (R::Id, R)> { + self.scan_prefix_stream(&[]) + } + + #[allow( + dead_code, + reason = "Part of the shared Repository surface; current consumers do not need value scans by sub-prefix yet" + )] + pub(crate) fn scan_prefix_stream<'a>( + &'a self, + 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)) + }), + ) + } + 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)) + }), + ) + } + Err(err) => Box::pin(stream::once(async move { Err(err) })), + } + } + + pub(crate) async fn gc(&self, predicate: F) -> Result + where + F: Fn(&R) -> bool + Send + Sync, + { + let mut iter = self.db.scan_prefix(prefix_key::(&[])?).await?; + let mut batch = WriteBatch::new(); + let mut deletes = 0_u64; + + while let Some(entry) = iter.next().await? { + let value = R::Codec::decode(&entry.value)?; + if predicate(&value) { + batch.delete(entry.key); + deletes += 1; + } + } + + if deletes > 0 { + self.db.write(batch).await?; + } + + Ok(deletes) + } +} + +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()), + ) +} + +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()), + ) +} + +fn decode_entry(entry: &KeyValue, prefix_segments: &[&str]) -> Result<(R::Id, R)> { + let id = parse_entry_id::(entry, prefix_segments)?; + let value = R::Codec::decode(&entry.value)?; + Ok((id, value)) +} + +fn parse_entry_id(entry: &KeyValue, prefix_segments: &[&str]) -> 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() { + return Err(Error::KeyParse(format!( + "key {raw_key:?} had {} segments, expected at least {} for prefix {}", + segments.len(), + prefix_segments.len(), + R::PREFIX + ))); + } + if segments[..prefix_segments.len()] != prefix_segments[..] { + return Err(Error::KeyParse(format!( + "key {raw_key:?} did not match expected prefix {}", + R::PREFIX + ))); + } + R::Id::from_key_segments(&segments[prefix_segments.len()..]) +} + +fn scan_entries( + db: Arc, + prefix: &keys::SlateKey, +) -> impl Stream> + Send { + enum ScanState { + Opening { db: Arc, prefix: Vec }, + Iterating(Box), + } + + stream::try_unfold( + ScanState::Opening { + db, + prefix: prefix.as_ref().to_vec(), + }, + |state| async move { + let mut iter = match state { + ScanState::Opening { db, prefix } => db.scan_prefix(prefix).await?, + ScanState::Iterating(iter) => *iter, + }; + + match iter.next().await? { + Some(entry) => Ok(Some((entry, ScanState::Iterating(Box::new(iter))))), + None => Ok(None), + } + }, + ) +} + +fn prefix_segments() -> Vec<&'static str> { + debug_assert!( + !R::PREFIX.is_empty() + && !R::PREFIX.starts_with('/') + && !R::PREFIX.ends_with('/') + && R::PREFIX.split('/').all(|segment| !segment.is_empty()), + "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 { + let mut segments = segments.into_iter(); + let first = segments.next().ok_or_else(|| { + Error::Other("record key assembly requires at least one segment".to_string()) + })?; + validate_key_segment(first)?; + + let mut key = keys::SlateKey::new(first); + for segment in segments { + validate_key_segment(segment)?; + key = key.with(segment); + } + Ok(key) +} + +fn prefix_from_segments<'a>(segments: impl IntoIterator) -> Result { + Ok(key_from_segments(segments)?.into_prefix()) +} + +fn validate_key_segment(segment: &str) -> Result<()> { + if segment.as_bytes().contains(&b'\0') { + return Err(Error::InvalidKeySegment { + segment: segment.to_string(), + }); + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use futures::TryStreamExt; + use object_store::memory::InMemory; + use serde::{Deserialize, Serialize}; + + use super::Repository; + use crate::record::{JsonCodec, MarkerCodec, RawBytesCodec, Record, RecordId}; + use crate::{Error, Result}; + + #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] + struct TestRecord { + id: TestId, + payload: String, + delete_me: bool, + } + + #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] + struct TestId { + bucket: String, + name: String, + } + + impl RecordId for TestId { + fn key_segments(&self) -> Vec { + vec![self.bucket.clone(), self.name.clone()] + } + + fn from_key_segments(segs: &[&str]) -> Result { + let [bucket, name] = segs else { + return Err(Error::KeyParse(format!( + "expected 2 segments for TestId, got {}", + segs.len() + ))); + }; + Ok(Self { + bucket: (*bucket).to_string(), + name: (*name).to_string(), + }) + } + } + + impl Record for TestRecord { + type Id = TestId; + type Codec = JsonCodec; + + const PREFIX: &'static str = "test/repository"; + + fn id(&self) -> Self::Id { + self.id.clone() + } + } + + #[derive(Debug, Clone, PartialEq, Eq, Default)] + struct TestMarker; + + impl Record for TestMarker { + type Id = String; + type Codec = MarkerCodec; + + const PREFIX: &'static str = "test/marker"; + + fn id(&self) -> Self::Id { + unreachable!("marker records must use put_at") + } + } + + #[derive(Debug, Clone, PartialEq, Eq)] + struct RawBlob(bytes::Bytes); + + impl AsRef<[u8]> for RawBlob { + fn as_ref(&self) -> &[u8] { + self.0.as_ref() + } + } + + impl From for RawBlob { + fn from(value: bytes::Bytes) -> Self { + Self(value) + } + } + + impl Record for RawBlob { + type Id = String; + type Codec = RawBytesCodec; + + const PREFIX: &'static str = "test/raw"; + + fn id(&self) -> Self::Id { + "blob".to_string() + } + } + + #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] + struct InvalidSegmentRecord { + id: String, + } + + impl Record for InvalidSegmentRecord { + type Id = String; + type Codec = JsonCodec; + + const PREFIX: &'static str = "test/invalid"; + + fn id(&self) -> Self::Id { + self.id.clone() + } + } + + async fn db() -> Arc { + Arc::new( + slatedb::Db::open("repository-tests", Arc::new(InMemory::new())) + .await + .unwrap(), + ) + } + + fn record(bucket: &str, name: &str, delete_me: bool) -> TestRecord { + TestRecord { + id: TestId { + bucket: bucket.to_string(), + name: name.to_string(), + }, + payload: format!("{bucket}/{name}"), + delete_me, + } + } + + #[tokio::test] + async fn put_get_delete_and_scan_round_trip() { + let repo = Repository::::new(db().await); + let saved = record("bucket-a", "alpha", false); + + assert!(repo.get(&saved.id()).await.unwrap().is_none()); + repo.put(&saved).await.unwrap(); + assert_eq!(repo.get(&saved.id()).await.unwrap(), Some(saved.clone())); + + let records = [ + saved.clone(), + record("bucket-a", "beta", false), + record("bucket-b", "alpha", false), + record("bucket-b", "beta", false), + record("bucket-c", "gamma", false), + ]; + for record in &records[1..] { + repo.put(record).await.unwrap(); + } + + let scanned = repo.scan_stream().try_collect::>().await.unwrap(); + assert_eq!(scanned.len(), records.len()); + assert_eq!(scanned[0], (records[0].id(), records[0].clone())); + + let bucket_a = repo + .scan_prefix_stream(&["bucket-a"]) + .try_collect::>() + .await + .unwrap(); + assert_eq!(bucket_a, vec![ + (records[0].id(), records[0].clone()), + (records[1].id(), records[1].clone()), + ]); + + repo.delete(&saved.id()).await.unwrap(); + assert!(repo.get(&saved.id()).await.unwrap().is_none()); + } + + #[tokio::test] + async fn gc_deletes_matching_records() { + let repo = Repository::::new(db().await); + for record in [ + record("bucket-a", "keep", false), + record("bucket-a", "delete", true), + record("bucket-b", "keep", false), + record("bucket-b", "delete", true), + ] { + repo.put(&record).await.unwrap(); + } + + assert_eq!(repo.gc(|record| record.delete_me).await.unwrap(), 2); + + let remaining = repo.scan_stream().try_collect::>().await.unwrap(); + assert_eq!(remaining.len(), 2); + assert!(remaining.iter().all(|(_, record)| !record.delete_me)); + } + + #[tokio::test] + async fn marker_records_use_put_at_exists_and_scan_ids() { + let repo = Repository::::new(db().await); + let marker = TestMarker; + + repo.put_at(&"marker-a".to_string(), &marker).await.unwrap(); + repo.put_at(&"marker-b".to_string(), &marker).await.unwrap(); + + assert!(repo.exists(&"marker-a".to_string()).await.unwrap()); + + let ids = repo + .scan_ids_stream() + .try_collect::>() + .await + .unwrap(); + assert_eq!(ids, vec!["marker-a".to_string(), "marker-b".to_string()]); + + repo.delete(&"marker-a".to_string()).await.unwrap(); + assert!(!repo.exists(&"marker-a".to_string()).await.unwrap()); + } + + #[tokio::test] + async fn raw_bytes_codec_round_trips_bytes() { + let repo = Repository::::new(db().await); + let blob = RawBlob(bytes::Bytes::from_static(b"hello")); + + repo.put(&blob).await.unwrap(); + + assert_eq!(repo.get(&"blob".to_string()).await.unwrap(), Some(blob)); + } + + #[tokio::test] + async fn malformed_bytes_propagate_decode_errors() { + let db = db().await; + let repo = Repository::::new(Arc::clone(&db)); + db.put( + super::key_for_id::(&TestId { + bucket: "bucket-a".to_string(), + name: "broken".to_string(), + }) + .unwrap(), + b"not-json", + ) + .await + .unwrap(); + + let error = repo + .get(&TestId { + bucket: "bucket-a".to_string(), + name: "broken".to_string(), + }) + .await + .unwrap_err(); + assert!(matches!(error, Error::Serde(_))); + } + + #[tokio::test] + async fn invalid_key_segments_return_runtime_error() { + let repo = Repository::::new(db().await); + let error = repo + .put(&InvalidSegmentRecord { + id: "bad\0segment".to_string(), + }) + .await + .unwrap_err(); + + match error { + Error::InvalidKeySegment { segment } => assert_eq!(segment, "bad\0segment"), + other => panic!("expected invalid key segment error, got {other:?}"), + } + } +} diff --git a/lib/crates/fabro-store/src/record/transaction.rs b/lib/crates/fabro-store/src/record/transaction.rs new file mode 100644 index 000000000..a66e7e80f --- /dev/null +++ b/lib/crates/fabro-store/src/record/transaction.rs @@ -0,0 +1,211 @@ +use slatedb::{Db, WriteBatch}; + +use super::repository::key_for_id; +use super::{Codec, Record}; +use crate::Result; + +pub(crate) async fn transaction(db: &Db, f: F) -> Result +where + F: FnOnce(&mut Tx) -> Result, +{ + let mut tx = Tx::new(); + let value = f(&mut tx)?; + if !tx.has_ops { + return Ok(value); + } + db.write(tx.into_batch()).await?; + Ok(value) +} + +pub(crate) struct Tx { + batch: WriteBatch, + has_ops: bool, +} + +impl Tx { + pub(crate) fn new() -> Self { + Self { + batch: WriteBatch::new(), + has_ops: false, + } + } + + pub(crate) fn put(&mut self, record: &R) -> Result<&mut Self> { + let id = record.id(); + self.put_at(&id, record) + } + + pub(crate) fn put_at(&mut self, id: &R::Id, record: &R) -> Result<&mut Self> { + self.batch + .put(key_for_id::(id)?, R::Codec::encode(record)?); + self.has_ops = true; + Ok(self) + } + + #[allow( + dead_code, + reason = "Shared transaction surface; current production callers only use put paths" + )] + pub(crate) fn delete(&mut self, id: &R::Id) -> Result<&mut Self> { + self.batch.delete(key_for_id::(id)?); + self.has_ops = true; + Ok(self) + } + + fn into_batch(self) -> WriteBatch { + self.batch + } +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use object_store::memory::InMemory; + use serde::{Deserialize, Serialize}; + + use super::{Record, Tx, transaction}; + use crate::record::{Codec, JsonCodec, Repository}; + use crate::{Error, Result}; + + #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] + struct TxRecord { + id: String, + payload: String, + poisoned: bool, + } + + impl Record for TxRecord { + type Id = String; + type Codec = JsonCodec; + + const PREFIX: &'static str = "test/transaction"; + + fn id(&self) -> Self::Id { + self.id.clone() + } + } + + #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] + struct FailingRecord { + id: String, + poisoned: bool, + } + + struct FailingCodec; + + impl Codec for FailingCodec { + fn encode(value: &FailingRecord) -> Result> { + if value.poisoned { + return Err(Error::Other( + "poisoned record refused to encode".to_string(), + )); + } + serde_json::to_vec(value).map_err(Into::into) + } + + fn decode(bytes: &[u8]) -> Result { + serde_json::from_slice(bytes).map_err(Into::into) + } + } + + impl Record for FailingRecord { + type Id = String; + type Codec = FailingCodec; + + const PREFIX: &'static str = "test/failing-transaction"; + + fn id(&self) -> Self::Id { + self.id.clone() + } + } + + async fn db() -> Arc { + Arc::new( + slatedb::Db::open("transaction-tests", Arc::new(InMemory::new())) + .await + .unwrap(), + ) + } + + #[tokio::test] + async fn closure_error_short_circuits_without_writing() { + let db = db().await; + let repo = Repository::::new(Arc::clone(&db)); + let record = TxRecord { + id: "record-1".to_string(), + payload: "hello".to_string(), + poisoned: false, + }; + + let error = transaction::<(), _>(&db, |tx| { + tx.put(&record)?; + Err(Error::Other("stop before commit".to_string())) + }) + .await + .unwrap_err(); + + assert_eq!(error.to_string(), "stop before commit"); + assert!(repo.get(&record.id()).await.unwrap().is_none()); + } + + #[tokio::test] + async fn encode_failure_discards_the_entire_batch() { + let db = db().await; + let repo = Repository::::new(Arc::clone(&db)); + let good = FailingRecord { + id: "good".to_string(), + poisoned: false, + }; + let bad = FailingRecord { + id: "bad".to_string(), + poisoned: true, + }; + + let error = transaction::<(), _>(&db, |tx| { + tx.put(&good)?; + tx.put(&bad)?; + Ok(()) + }) + .await + .unwrap_err(); + + assert_eq!(error.to_string(), "poisoned record refused to encode"); + assert!(repo.get(&good.id()).await.unwrap().is_none()); + assert!(repo.get(&bad.id()).await.unwrap().is_none()); + } + + #[tokio::test] + async fn empty_transaction_returns_without_writing() { + let db = db().await; + let repo = Repository::::new(Arc::clone(&db)); + + let value = transaction(&db, |_tx: &mut Tx| Ok::<_, Error>("ok")) + .await + .unwrap(); + + assert_eq!(value, "ok"); + assert!(repo.get(&"missing".to_string()).await.unwrap().is_none()); + } + + #[tokio::test] + async fn delete_operations_are_committed() { + let db = db().await; + let repo = Repository::::new(Arc::clone(&db)); + let record = TxRecord { + id: "delete-me".to_string(), + payload: "hello".to_string(), + poisoned: false, + }; + repo.put(&record).await.unwrap(); + + transaction::<(), _>(&db, |tx| { + tx.delete::(&record.id())?; + Ok(()) + }) + .await + .unwrap(); + + assert!(repo.get(&record.id()).await.unwrap().is_none()); + } +} diff --git a/lib/crates/fabro-store/src/slate/auth_codes.rs b/lib/crates/fabro-store/src/slate/auth_codes.rs index 7171a29cc..f3104c884 100644 --- a/lib/crates/fabro-store/src/slate/auth_codes.rs +++ b/lib/crates/fabro-store/src/slate/auth_codes.rs @@ -1,15 +1,15 @@ use std::sync::Arc; use chrono::{DateTime, Utc}; -use dashmap::DashMap; use fabro_types::IdpIdentity; use serde::{Deserialize, Serialize}; -use tokio::sync::Mutex; -use crate::{Result, keys}; +use crate::record::{JsonCodec, Record, Repository}; +use crate::{KeyedMutex, Result}; #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct AuthCode { + pub code: String, pub identity: IdpIdentity, pub login: String, pub name: String, @@ -19,84 +19,63 @@ pub struct AuthCode { pub expires_at: DateTime, } -pub struct SlateAuthCodeStore { - db: Arc, - code_locks: DashMap>>, -} +impl Record for AuthCode { + type Id = String; + type Codec = JsonCodec; -impl std::fmt::Debug for SlateAuthCodeStore { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.debug_struct("SlateAuthCodeStore").finish_non_exhaustive() + const PREFIX: &'static str = "auth/code"; + + fn id(&self) -> Self::Id { + self.code.clone() } } -impl SlateAuthCodeStore { +pub struct AuthCodeStore { + repo: Repository, + consume_locks: KeyedMutex, +} + +impl std::fmt::Debug for AuthCodeStore { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("AuthCodeStore").finish_non_exhaustive() + } +} + +impl AuthCodeStore { pub(crate) fn new(db: Arc) -> Self { Self { - db, - code_locks: DashMap::new(), + repo: Repository::new(db), + consume_locks: KeyedMutex::new(), } } - pub async fn insert(&self, code: &str, entry: AuthCode) -> Result<()> { - self.db - .put(keys::auth_code_key(code), serde_json::to_vec(&entry)?) - .await?; - Ok(()) + pub async fn insert(&self, entry: AuthCode) -> Result<()> { + self.repo.put(&entry).await } pub async fn consume(&self, code: &str) -> Result> { - let mutex = self - .code_locks - .entry(code.to_string()) - .or_insert_with(|| Arc::new(Mutex::new(()))) - .clone(); - let _guard = mutex.lock().await; - - let key = keys::auth_code_key(code); - let entry = self - .db - .get(&key) - .await? - .map(|bytes| serde_json::from_slice::(&bytes)) - .transpose()?; + let code = code.to_string(); + let _guard = self.consume_locks.lock(code.clone()).await; + let entry = self.repo.get(&code).await?; let result = match entry { Some(entry) if entry.expires_at > Utc::now() => { - self.db.delete(&key).await?; + self.repo.delete(&code).await?; Some(entry) } Some(_) => { - self.db.delete(&key).await?; + self.repo.delete(&code).await?; None } None => None, }; - if Arc::strong_count(&mutex) == 2 { - self.code_locks.remove(code); - } - Ok(result) } pub async fn gc_expired(&self, cutoff: DateTime) -> Result { - let mut iter = self.db.scan_prefix(keys::auth_code_prefix()).await?; - let mut keys_to_delete = Vec::new(); - while let Some(entry) = iter.next().await? { - let auth_code: AuthCode = serde_json::from_slice(&entry.value)?; - if auth_code.expires_at <= cutoff { - keys_to_delete.push( - String::from_utf8(entry.key.to_vec()) - .expect("slatedb keys should be valid utf-8"), - ); - } - } - - for key in &keys_to_delete { - self.db.delete(key).await?; - } - - Ok(keys_to_delete.len() as u64) + self.repo + .gc(|auth_code| auth_code.expires_at <= cutoff) + .await } } @@ -109,10 +88,10 @@ mod tests { use object_store::memory::InMemory; use tokio::task::JoinSet; - use super::{AuthCode, SlateAuthCodeStore}; + use super::{AuthCode, AuthCodeStore}; use crate::Database; - async fn store() -> Arc { + async fn store() -> Arc { let db = Database::new( Arc::new(InMemory::new()), "", @@ -122,8 +101,9 @@ mod tests { db.auth_codes().await.unwrap() } - fn auth_code(expires_at: chrono::DateTime) -> AuthCode { + fn auth_code(code: &str, expires_at: chrono::DateTime) -> AuthCode { AuthCode { + code: code.to_string(), identity: fabro_types::IdpIdentity::new("https://github.com", "12345").unwrap(), login: "octocat".to_string(), name: "The Octocat".to_string(), @@ -138,10 +118,10 @@ mod tests { async fn insert_and_consume_is_single_use() { let store = store().await; store - .insert( + .insert(auth_code( "code-1", - auth_code(chrono::Utc::now() + ChronoDuration::seconds(60)), - ) + chrono::Utc::now() + ChronoDuration::seconds(60), + )) .await .unwrap(); @@ -153,10 +133,10 @@ mod tests { async fn concurrent_consume_has_one_winner() { let store = store().await; store - .insert( + .insert(auth_code( "code-2", - auth_code(chrono::Utc::now() + ChronoDuration::seconds(60)), - ) + chrono::Utc::now() + ChronoDuration::seconds(60), + )) .await .unwrap(); @@ -180,17 +160,17 @@ mod tests { async fn gc_expired_removes_only_expired_codes() { let store = store().await; store - .insert( + .insert(auth_code( "expired", - auth_code(chrono::Utc::now() - ChronoDuration::seconds(1)), - ) + chrono::Utc::now() - ChronoDuration::seconds(1), + )) .await .unwrap(); store - .insert( + .insert(auth_code( "live", - auth_code(chrono::Utc::now() + ChronoDuration::seconds(60)), - ) + chrono::Utc::now() + ChronoDuration::seconds(60), + )) .await .unwrap(); diff --git a/lib/crates/fabro-store/src/slate/auth_tokens.rs b/lib/crates/fabro-store/src/slate/auth_tokens.rs index 91b4213a8..9357dd5b8 100644 --- a/lib/crates/fabro-store/src/slate/auth_tokens.rs +++ b/lib/crates/fabro-store/src/slate/auth_tokens.rs @@ -4,11 +4,10 @@ use chrono::{DateTime, Utc}; use dashmap::DashMap; use fabro_types::IdpIdentity; use serde::{Deserialize, Serialize}; -use slatedb::WriteBatch; -use tokio::sync::Mutex; use uuid::Uuid; -use crate::{Result, keys}; +use crate::record::{JsonCodec, Record, Repository, transaction}; +use crate::{KeyedMutex, Result}; const REPLAY_REVOCATION_TTL_SECONDS: i64 = 60; @@ -27,6 +26,17 @@ pub struct RefreshToken { pub user_agent: String, } +impl Record for RefreshToken { + type Id = [u8; 32]; + type Codec = JsonCodec; + + const PREFIX: &'static str = "auth/refresh"; + + fn id(&self) -> Self::Id { + self.token_hash + } +} + #[derive(Debug, Clone, PartialEq, Eq)] pub enum ConsumeOutcome { Rotated(RefreshToken, Box), @@ -35,45 +45,38 @@ pub enum ConsumeOutcome { NotFound, } -pub struct SlateAuthTokenStore { +pub struct RefreshTokenStore { db: Arc, - refresh_locks: DashMap<[u8; 32], 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. replay_revocations: DashMap<[u8; 32], DateTime>, } -impl std::fmt::Debug for SlateAuthTokenStore { +impl std::fmt::Debug for RefreshTokenStore { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.debug_struct("SlateAuthTokenStore") - .finish_non_exhaustive() + f.debug_struct("RefreshTokenStore").finish_non_exhaustive() } } -impl SlateAuthTokenStore { +impl RefreshTokenStore { pub(crate) fn new(db: Arc) -> Self { Self { + repo: Repository::new(Arc::clone(&db)), db, - refresh_locks: DashMap::new(), + consume_locks: KeyedMutex::new(), replay_revocations: DashMap::new(), } } pub async fn insert_refresh_token(&self, token: RefreshToken) -> Result<()> { - self.db - .put( - keys::auth_refresh_key(&token.token_hash), - serde_json::to_vec(&token)?, - ) - .await?; - Ok(()) + self.repo.put(&token).await } pub async fn find_refresh_token(&self, token_hash: &[u8; 32]) -> Result> { - self.db - .get(keys::auth_refresh_key(token_hash)) - .await? - .map(|bytes| serde_json::from_slice::(&bytes)) - .transpose() - .map_err(Into::into) + self.repo.get(token_hash).await } pub async fn consume_and_rotate( @@ -82,14 +85,9 @@ impl SlateAuthTokenStore { new_token: RefreshToken, now: DateTime, ) -> Result { - let mutex = self - .refresh_locks - .entry(presented_hash) - .or_insert_with(|| Arc::new(Mutex::new(()))) - .clone(); - let _guard = mutex.lock().await; + let _guard = self.consume_locks.lock(presented_hash).await; - let outcome = match self.find_refresh_token(&presented_hash).await? { + let outcome = match self.repo.get(&presented_hash).await? { None => ConsumeOutcome::NotFound, Some(existing) if now >= existing.expires_at => ConsumeOutcome::Expired, Some(existing) if existing.used => ConsumeOutcome::Reused(existing), @@ -98,66 +96,26 @@ impl SlateAuthTokenStore { old_token.used = true; old_token.last_used_at = now; - let mut batch = WriteBatch::new(); - batch.put( - keys::auth_refresh_key(&presented_hash), - serde_json::to_vec(&old_token)?, - ); - batch.put( - keys::auth_refresh_key(&new_token.token_hash), - serde_json::to_vec(&new_token)?, - ); - self.db.write(batch).await?; + transaction(&self.db, |tx| { + tx.put(&old_token)?; + tx.put(&new_token)?; + Ok(()) + }) + .await?; ConsumeOutcome::Rotated(old_token, Box::new(new_token)) } }; - if Arc::strong_count(&mutex) == 2 { - self.refresh_locks.remove(&presented_hash); - } - Ok(outcome) } pub async fn delete_chain(&self, chain_id: Uuid) -> Result { - let mut iter = self.db.scan_prefix(keys::auth_refresh_prefix()).await?; - let mut keys_to_delete = Vec::new(); - while let Some(entry) = iter.next().await? { - let token: RefreshToken = serde_json::from_slice(&entry.value)?; - if token.chain_id == chain_id { - keys_to_delete.push( - String::from_utf8(entry.key.to_vec()) - .expect("slatedb keys should be valid utf-8"), - ); - } - } - - for key in &keys_to_delete { - self.db.delete(key).await?; - } - - Ok(keys_to_delete.len() as u64) + self.repo.gc(|token| token.chain_id == chain_id).await } pub async fn gc_expired(&self, cutoff: DateTime) -> Result { - let mut iter = self.db.scan_prefix(keys::auth_refresh_prefix()).await?; - let mut keys_to_delete = Vec::new(); - while let Some(entry) = iter.next().await? { - let token: RefreshToken = serde_json::from_slice(&entry.value)?; - if token.expires_at <= cutoff { - keys_to_delete.push( - String::from_utf8(entry.key.to_vec()) - .expect("slatedb keys should be valid utf-8"), - ); - } - } - - for key in &keys_to_delete { - self.db.delete(key).await?; - } - - Ok(keys_to_delete.len() as u64) + self.repo.gc(|token| token.expires_at <= cutoff).await } pub fn mark_refresh_token_replay(&self, token_hash: [u8; 32], now: DateTime) { @@ -188,17 +146,17 @@ mod tests { use tokio::task::JoinSet; use uuid::Uuid; - use super::{ConsumeOutcome, RefreshToken, SlateAuthTokenStore}; + use super::{ConsumeOutcome, RefreshToken, RefreshTokenStore}; use crate::Database; - async fn store() -> Arc { + async fn store() -> Arc { let db = Database::new( Arc::new(InMemory::new()), "", Duration::from_millis(1), None, ); - db.auth_tokens().await.unwrap() + db.refresh_tokens().await.unwrap() } fn refresh_token(hash: [u8; 32], chain_id: Uuid, used: bool) -> RefreshToken { diff --git a/lib/crates/fabro-store/src/slate/blob_store.rs b/lib/crates/fabro-store/src/slate/blob_store.rs new file mode 100644 index 000000000..3c058b3c8 --- /dev/null +++ b/lib/crates/fabro-store/src/slate/blob_store.rs @@ -0,0 +1,132 @@ +use std::sync::Arc; + +use bytes::Bytes; +use fabro_types::RunBlobId; + +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 for Blob { + fn from(value: Bytes) -> Self { + Self(value) + } +} + +impl Record for Blob { + type Id = RunBlobId; + type Codec = RawBytesCodec; + + const PREFIX: &'static str = "blobs/sha256"; + + fn id(&self) -> Self::Id { + RunBlobId::new(&self.0) + } +} + +pub struct BlobStore { + repo: Repository, +} + +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) -> Self { + Self { + repo: Repository::new(db), + } + } + + pub async fn write(&self, bytes: &[u8]) -> Result { + let blob = Blob(Bytes::copy_from_slice(bytes)); + let id = blob.id(); + self.repo.put(&blob).await?; + Ok(id) + } + + pub async fn read(&self, id: &RunBlobId) -> Result> { + Ok(self.repo.get(id).await?.map(|blob| blob.0)) + } + + pub async fn exists(&self, id: &RunBlobId) -> Result { + self.repo.exists(id).await + } +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + use std::time::Duration; + + use bytes::Bytes; + use fabro_types::RunBlobId; + use object_store::memory::InMemory; + + use super::BlobStore; + use crate::Database; + use crate::keys::SlateKey; + + async fn store() -> Arc { + let db = Database::new( + Arc::new(InMemory::new()), + "", + Duration::from_millis(1), + None, + ); + db.blobs().await.unwrap() + } + + #[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(&RunBlobId::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 = Arc::new( + slatedb::Db::open("blob-store-tests", Arc::new(InMemory::new())) + .await + .unwrap(), + ); + let store = BlobStore::new(Arc::clone(&raw_db)); + 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); + } +} diff --git a/lib/crates/fabro-store/src/slate/catalog.rs b/lib/crates/fabro-store/src/slate/catalog.rs deleted file mode 100644 index 49181ac08..000000000 --- a/lib/crates/fabro-store/src/slate/catalog.rs +++ /dev/null @@ -1,51 +0,0 @@ -use chrono::{Datelike, Timelike}; -use fabro_types::RunId; -use slatedb::Db; - -use crate::{ListRunsQuery, Result, keys}; - -pub(crate) async fn write_index(db: &Db, run_id: &RunId) -> Result<()> { - db.put(keys::runs_index_by_start_key(run_id), []).await?; - Ok(()) -} - -pub(crate) async fn delete_index(db: &Db, run_id: &RunId) -> Result<()> { - db.delete(keys::runs_index_by_start_key(run_id)).await?; - Ok(()) -} - -pub(crate) async fn list_run_ids(db: &Db, query: &ListRunsQuery) -> Result> { - let mut iter = db.scan_prefix(keys::runs_index_by_start_prefix()).await?; - let mut run_ids = Vec::new(); - while let Some(entry) = iter.next().await? { - let key = String::from_utf8(entry.key.to_vec()) - .map_err(|err| crate::Error::Other(format!("stored key is not valid UTF-8: {err}")))?; - let Some(run_id) = keys::parse_run_id_from_index_key(&key) else { - continue; - }; - let created_at = run_id.created_at(); - if let Some(start) = query.start { - if created_at < start { - continue; - } - } - if let Some(end) = query.end { - if created_at > end { - continue; - } - } - run_ids.push(run_id); - } - run_ids.sort_by_key(|run_id| { - let created_at = run_id.created_at(); - ( - created_at.year(), - created_at.month(), - created_at.day(), - created_at.hour(), - created_at.minute(), - *run_id, - ) - }); - Ok(run_ids) -} diff --git a/lib/crates/fabro-store/src/slate/mod.rs b/lib/crates/fabro-store/src/slate/mod.rs index e368970b0..d14f98aa7 100644 --- a/lib/crates/fabro-store/src/slate/mod.rs +++ b/lib/crates/fabro-store/src/slate/mod.rs @@ -1,6 +1,7 @@ mod auth_codes; mod auth_tokens; -mod catalog; +mod blob_store; +mod run_catalog_index; mod run_store; use std::collections::HashMap; @@ -8,10 +9,12 @@ use std::path::PathBuf; use std::sync::Arc; use std::time::Duration; -pub use auth_codes::{AuthCode, SlateAuthCodeStore}; -pub use auth_tokens::{ConsumeOutcome, RefreshToken, SlateAuthTokenStore}; +pub use auth_codes::{AuthCode, AuthCodeStore}; +pub use auth_tokens::{ConsumeOutcome, RefreshToken, RefreshTokenStore}; +pub use blob_store::{Blob, BlobStore}; use fabro_types::RunId; use object_store::ObjectStore; +pub use run_catalog_index::RunCatalogIndex; pub use run_store::RunDatabase; use run_store::RunDatabaseInner; use slatedb::config::{CompressionCodec, Settings}; @@ -27,8 +30,10 @@ pub struct Database { cache_path: Option, db: Arc>, active_runs: Arc>>>, - auth_codes: Arc>>, - auth_tokens: Arc>>, + blobs: Arc>>, + catalog_index: Arc>>, + auth_codes: Arc>>, + refresh_tokens: Arc>>, } impl std::fmt::Debug for Database { @@ -55,8 +60,10 @@ impl Database { cache_path, db: Arc::new(OnceCell::new()), active_runs: Arc::new(Mutex::new(HashMap::new())), + blobs: Arc::new(OnceCell::new()), + catalog_index: Arc::new(OnceCell::new()), auth_codes: Arc::new(OnceCell::new()), - auth_tokens: Arc::new(OnceCell::new()), + refresh_tokens: Arc::new(OnceCell::new()), } } @@ -116,7 +123,7 @@ impl Database { if run_exists && !active.matches_run(run_id) { return Err(Error::RunAlreadyExists(run_id.to_string())); } - catalog::write_index(&db, run_id).await?; + self.catalog_index().await?.add(run_id).await?; return Ok(active); } @@ -124,7 +131,7 @@ impl Database { return Err(Error::RunAlreadyExists(run_id.to_string())); } - catalog::write_index(&db, run_id).await?; + self.catalog_index().await?.add(run_id).await?; let run_store = RunDatabase::open_writer(*run_id, db).await?; self.cache_active_run(&run_store).await; Ok(run_store) @@ -166,7 +173,7 @@ impl Database { pub async fn list_runs(&self, query: &ListRunsQuery) -> Result> { let db = self.open_db().await?; - let run_ids = catalog::list_run_ids(&db, query).await?; + let run_ids = self.catalog_index().await?.list(query).await?; let mut summaries = Vec::new(); for run_id in run_ids { if let Some(active) = self.get_active_run(&run_id).await { @@ -201,27 +208,49 @@ impl Database { for key in keys_to_delete { db.delete(key).await?; } - catalog::delete_index(&db, run_id).await?; + self.catalog_index().await?.remove(run_id).await?; Ok(()) } - pub async fn auth_codes(&self) -> Result> { + pub async fn auth_codes(&self) -> Result> { let store = self .auth_codes .get_or_try_init(|| async { let db = Arc::new(self.open_db().await?); - Ok::<_, Error>(Arc::new(SlateAuthCodeStore::new(db))) + Ok::<_, Error>(Arc::new(AuthCodeStore::new(db))) }) .await?; Ok(Arc::clone(store)) } - pub async fn auth_tokens(&self) -> Result> { + pub async fn catalog_index(&self) -> Result> { let store = self - .auth_tokens + .catalog_index .get_or_try_init(|| async { let db = Arc::new(self.open_db().await?); - Ok::<_, Error>(Arc::new(SlateAuthTokenStore::new(db))) + Ok::<_, Error>(Arc::new(RunCatalogIndex::new(db))) + }) + .await?; + Ok(Arc::clone(store)) + } + + pub async fn blobs(&self) -> Result> { + let store = self + .blobs + .get_or_try_init(|| async { + let db = Arc::new(self.open_db().await?); + Ok::<_, Error>(Arc::new(BlobStore::new(db))) + }) + .await?; + Ok(Arc::clone(store)) + } + + pub async fn refresh_tokens(&self) -> Result> { + let store = self + .refresh_tokens + .get_or_try_init(|| async { + let db = Arc::new(self.open_db().await?); + Ok::<_, Error>(Arc::new(RefreshTokenStore::new(db))) }) .await?; Ok(Arc::clone(store)) diff --git a/lib/crates/fabro-store/src/slate/run_catalog_index.rs b/lib/crates/fabro-store/src/slate/run_catalog_index.rs new file mode 100644 index 000000000..332c09496 --- /dev/null +++ b/lib/crates/fabro-store/src/slate/run_catalog_index.rs @@ -0,0 +1,150 @@ +use std::sync::Arc; + +use chrono::{Datelike, Timelike}; +use fabro_types::RunId; +use futures::TryStreamExt; + +use crate::record::{MarkerCodec, Record, Repository}; +use crate::{ListRunsQuery, Result}; + +#[derive(Debug, Default)] +pub(crate) struct RunCatalogEntry; + +impl Record for RunCatalogEntry { + type Id = RunId; + type Codec = MarkerCodec; + + const PREFIX: &'static str = "runs/_index/by-start"; + + fn id(&self) -> Self::Id { + unreachable!("marker records must use put_at") + } +} + +pub struct RunCatalogIndex { + repo: Repository, +} + +impl std::fmt::Debug for RunCatalogIndex { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("RunCatalogIndex").finish_non_exhaustive() + } +} + +impl RunCatalogIndex { + pub(crate) fn new(db: Arc) -> Self { + Self { + repo: Repository::new(db), + } + } + + pub async fn add(&self, run_id: &RunId) -> Result<()> { + self.repo.put_at(run_id, &RunCatalogEntry).await + } + + pub async fn remove(&self, run_id: &RunId) -> Result<()> { + self.repo.delete(run_id).await + } + + pub async fn list(&self, query: &ListRunsQuery) -> Result> { + let mut run_ids = self.repo.scan_ids_stream().try_collect::>().await?; + run_ids.retain(|run_id| { + let created_at = run_id.created_at(); + if let Some(start) = query.start { + if created_at < start { + return false; + } + } + if let Some(end) = query.end { + if created_at > end { + return false; + } + } + true + }); + run_ids.sort_by_key(|run_id| { + let created_at = run_id.created_at(); + ( + created_at.year(), + created_at.month(), + created_at.day(), + created_at.hour(), + created_at.minute(), + *run_id, + ) + }); + Ok(run_ids) + } +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use chrono::{Duration as ChronoDuration, TimeZone, Utc}; + use object_store::memory::InMemory; + use ulid::Ulid; + + use super::RunCatalogIndex; + use crate::ListRunsQuery; + + async fn index() -> RunCatalogIndex { + let db = Arc::new( + slatedb::Db::open("run-catalog-index-tests", Arc::new(InMemory::new())) + .await + .unwrap(), + ); + RunCatalogIndex::new(db) + } + + #[tokio::test] + async fn add_list_and_remove_round_trip() { + let index = index().await; + let early = fabro_types::RunId::from(Ulid::from_datetime( + Utc.with_ymd_and_hms(2026, 4, 20, 9, 0, 0).unwrap().into(), + )); + let later = fabro_types::RunId::from(Ulid::from_datetime( + Utc.with_ymd_and_hms(2026, 4, 20, 9, 1, 0).unwrap().into(), + )); + + index.add(&later).await.unwrap(); + index.add(&early).await.unwrap(); + + assert_eq!(index.list(&ListRunsQuery::default()).await.unwrap(), vec![ + early, later + ]); + + index.remove(&early).await.unwrap(); + assert_eq!(index.list(&ListRunsQuery::default()).await.unwrap(), vec![ + later + ]); + } + + #[tokio::test] + async fn list_applies_start_and_end_filters() { + let index = index().await; + let first = fabro_types::RunId::from(Ulid::from_datetime( + Utc.with_ymd_and_hms(2026, 4, 20, 9, 0, 0).unwrap().into(), + )); + let second = fabro_types::RunId::from(Ulid::from_datetime( + Utc.with_ymd_and_hms(2026, 4, 20, 9, 1, 0).unwrap().into(), + )); + let third = fabro_types::RunId::from(Ulid::from_datetime( + Utc.with_ymd_and_hms(2026, 4, 20, 9, 2, 0).unwrap().into(), + )); + for run_id in [first, second, third] { + index.add(&run_id).await.unwrap(); + } + + assert_eq!( + index + .list(&ListRunsQuery { + start: Some(second.created_at()), + end: Some(second.created_at() + ChronoDuration::seconds(1)), + }) + .await + .unwrap(), + vec![second] + ); + } +} diff --git a/lib/crates/fabro-store/src/slate/run_store.rs b/lib/crates/fabro-store/src/slate/run_store.rs index 388e5a188..d1379d3e2 100644 --- a/lib/crates/fabro-store/src/slate/run_store.rs +++ b/lib/crates/fabro-store/src/slate/run_store.rs @@ -10,6 +10,7 @@ use slatedb::{Db, DbRead}; use tokio::sync::{Mutex, broadcast, mpsc}; use tokio_stream::wrappers::UnboundedReceiverStream; +use super::blob_store::BlobStore; use crate::run_state::EventProjectionCache; use crate::{Error, EventEnvelope, EventPayload, Result, RunProjection, RunSummary, keys}; @@ -284,17 +285,15 @@ impl RunDatabase { if self.read_only { return Err(Error::ReadOnly); } - let id = RunBlobId::new(data); - self.inner.db.put(keys::blob_key(&id), data).await?; - Ok(id) + BlobStore::new(Arc::new(self.inner.db.clone())) + .write(data) + .await } pub async fn read_blob(&self, id: &RunBlobId) -> Result> { - let global = self.inner.db.get(keys::blob_key(id)).await?; - if global.is_some() { - return Ok(global); - } - Ok(None) + BlobStore::new(Arc::new(self.inner.db.clone())) + .read(id) + .await } pub async fn list_blobs(&self) -> Result> {