diff --git a/CLAUDE.md b/CLAUDE.md index c87b5348..8e6a57bf 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -80,13 +80,13 @@ Flat layout — one file per concept, no module subdirectories. The earlier `cod - **`decoded.rs`** — Consumer-side typed view over the raw subscription / `read_stream` / `read_all` streams (#249). Keeps the deliberate raw multi-consumer contract; adds an ergonomic layer that **reuses the configured codec** instead of a hand-rolled `from_slice`. `Decoded` is one box (event + `version` + `metadata`) generic over the payload — owned `E` on the stream path, the borrowed window `Decode::Output<'a>` inside a fold. `DecodedStreamExt` adds two methods over any `Stream>` where `I: RawItem` (sealed; impl'd for `PersistedEnvelope` → `Decoded` and `(AllPosition, PersistedEnvelope)` → `(AllPosition, Decoded)`, tag preserved): `.decoded(codec)` — **owning codecs only** (`for<'a> Output<'a> = E`, so a zero-copy codec cannot satisfy it and the compiler steers to the fold) → a stream of carry-away `Decoded`; and `.for_each_decoded(codec, f)` — **owning and zero-copy** (rkyv/bytemuck) → internal-iteration fold handing the borrowed window to a closure, so no lending stream is reintroduced. `for_each_decoded` surfaces the per-stream `version` but not the `$all` position tag (positioned `$all` folds use `.decoded()` or a raw `subscribe_all` loop with `codec.decode`). Error domains stay distinct (rule 3): `DecodeStreamError` (`Read`/`Decode`), `FoldDecodedError` (`Read`/`Decode`/`Handler`). Not subscription-gated — also sugars finite reads. **`StepStreamExt`** (#250) adds the phase-aware surface over a `Step`-tagged stream (what `subscribe`/`subscribe_all` yield): `.events()` drops the phase (removes `CaughtUp`, unwraps `Event`) → a bare raw stream the full `DecodedStreamExt` then applies to (the events-only path; also the public form of the old internal `live` filter); `.decoded(codec)` **keeps** the phase, decoding each `Event` → `Step>` (the projection consumption path — tells catch-up from live *and* hands you typed events, owning codecs only, same steer as `DecodedStreamExt::decoded`). `Step` is deliberately **not** a `RawItem` (the `CaughtUp` marker carries no envelope), so the two `.decoded()` methods are non-overlapping impls sharing one name. Composition table: `subscribe` (raw+phase) · `.events()` (raw, no phase) · `.events().decoded()` (typed, no phase) · `.decoded()` (typed+phase) · `.events().for_each_decoded()` (zero-copy fold, no phase). - **`wire.rs`** — One canonical frame builder used by every adapter: `wire::encode_frame(global_seq, schema_version, &event_type, &payload, metadata) -> Result` and `wire::decode_frame(&[u8]) -> Result`. Layout: `[u8 frame_format_version][u64 LE global_seq][u32 LE schema_version][u16 LE event_type_len][u32 LE meta_len][event_type][metadata?][padding][payload]` where `meta_len == u32::MAX` is the absent-metadata sentinel. The **leading `frame_format_version` byte (offset 0) is read first**: `decode_frame` decodes it via the typed `FrameFormatVersion` enum and branches on layout (`match header.format_version { V1 => decode_frame_v1(..) }`), rejecting an unknown byte as a typed `DecodeError::UnsupportedFrameVersion` rather than misparsing. It is the on-disk escape hatch that makes the frame **evolvable** (a future v2 can change alignment / add a CRC / store the payload length without a full data migration) — added in the pre-1.0-freeze window because the frame is irreversible once real data exists (#205); `HEADER_FIXED_SIZE` is `19` (version byte + the four fixed fields). The padding makes the payload pointer land on a **16-byte boundary** inside the returned `Bytes` — a wire-format invariant zero-copy decoders (rkyv default alignment, flatbuffers, `#[repr(C)]` POD up to 16) rely on; the version byte shifts the fixed fields one byte later but `align_padding` recomputes the payload position so the 16-byte boundary holds. Backing buffer built via `aligned_vec::AVec>` → `bytes::Bytes::from_owner(avec)` for zero-copy ownership transfer that preserves alignment. - **`state.rs`** — Unified snapshot persistence. One trait powers both projection state and aggregate snapshots. - - `SnapshotStore` trait: atomic persistence of derived state plus the position it was folded to. `hydrate(id, schema_version) -> Option<(P, S)>` and `commit(id, schema_version, position, &state)` — state and position are saved and loaded *together*, never separately, so a half-write is unrepresentable. Generic over the position type `P` (`Version` for a single stream, `GlobalSeq` for a multi-stream projection). + - `SnapshotStore` trait: atomic persistence of derived state plus the position it was folded to. `hydrate(id, schema_version) -> Hydrated` and `commit(id, schema_version, position, &state)` — state and position are saved and loaded *together*, never separately, so a half-write is unrepresentable. Generic over the position type `P` (`Version` for a single stream, `GlobalSeq` for a multi-stream projection). **`hydrate` returns a three-state `Hydrated` (`Absent` | `Stale { stored_schema }` | `Found { position, state }`), not `Option`** — the extra state keeps "nothing saved" distinct from "saved but under a different schema version". That distinction is invisible to an aggregate snapshot (both mean "replay the stream", and the decorator collapses them via `Hydrated::into_found`), but load-bearing for a projection: `Stale` means a schema bump invalidated the saved state and the next step is a **full re-fold of the whole `$all` stream** — on a mobile/IoT host a long, battery-heavy operation the host must see coming. `Stale` deliberately carries no stale bytes (derived state has no upcasting path — a schema change forces a rebuild, never a migration). Aggregate/projection adapters validate the stored schema is `NonZeroU32` (a zero is corruption, not staleness) before branching `Stale` vs `Found`. - `CodecSnapshotStore`: bridges a byte-level `SnapshotStore, P>` (what fjall implements) to a typed `SnapshotStore` via `Decode` / `Encode`. The position `P` passes through untouched. `CodecSnapshotStoreError::Wire` covers the rare wire-build failure when synthesizing an envelope from snapshot bytes. - `PersistTrigger` trait: `EveryNEvents(N)` (bucket-crossing), `AfterEventTypes(&[&str])` (semantic). Used by both projection runners and the snapshot decorator. - Inline `#[cfg(feature = "testing")] mod testing` block: `InMemorySnapshotStore`. - **`projection.rs`** — Feature-gated under `projection`. Slim by design. - `Projector` trait: pure fallible fold function (`initial()` + `apply(state, &event) -> Result`). Fallibility is intentional: projections may do checked arithmetic where aggregates do not. Recovery policy (skip/fail/dead-letter) is the framework's concern, not the projector's. nexus ships no runner; the IO-driven loop is the consumer's. See `examples/projection-tokio`. - - `Projection` (#255) — an inert per-event **stepper** that *assembles* the four projection primitives (`Projector` + `PersistTrigger` + `SnapshotStore`, plus a `Subscription` outside) with no runner. `load` (five named inputs, no codec, no `for<'a>`) hydrates and returns `(stepper, starting_state)`; `advance(state, Decoded) -> state` folds one event and commits `(state, position)` if the trigger fires; `flush(&state)` persists the folded-but-unpersisted tail on shutdown. Owns no loop — the host (tokio `while let`, Agency/Bombay actor mailbox) drives it. **State flows through the caller, not the stepper**, because `Projector::apply` consumes by value: keeping it inside would force `Clone` (hot-path) or an `Option`/poison edge (rule 4). The codec is absent by construction — decode with `.decoded()` before `advance`, so the assembly never names a codec nor restates the owning-codec bound. `ProjectionError::{Apply, Commit}` keeps the fold and persist domains distinct (rule 3). Prior art: Akka Projection (inert value, host spawns). The old 8-arg/6-generic hand-wired `run_projection` (which leaked the HRTB) is gone; `examples/projection-tokio` now builds its loop on the stepper. + - `Projection` (#255) — an inert per-event **stepper** that *assembles* the four projection primitives (`Projector` + `PersistTrigger` + `SnapshotStore`, plus a `Subscription` outside) with no runner. `load` (five named inputs, no codec, no `for<'a>`) hydrates and returns `(stepper, starting_state)`; `advance(state, Decoded) -> state` folds one event and commits `(state, position)` if the trigger fires; `flush(&state)` persists the folded-but-unpersisted tail on shutdown. On a `Hydrated::Stale` hydrate `load` starts from `initial()` **and** records the discarded schema version, surfaced via `rebuilding_from() -> Option` — so a host distinguishes a costly schema-bump rebuild (`Some(old)`) from an ordinary fresh start (`None`, disambiguated from a resume via `checkpoint()`). Deliberately **no second `Start` enum** duplicating `Hydrated`'s trichotomy: `load`'s signature stays `(stepper, state)` and the one genuinely-new bit rides on a single accessor. Owns no loop — the host (tokio `while let`, Agency/Bombay actor mailbox) drives it. **State flows through the caller, not the stepper**, because `Projector::apply` consumes by value: keeping it inside would force `Clone` (hot-path) or an `Option`/poison edge (rule 4). The codec is absent by construction — decode with `.decoded()` before `advance`, so the assembly never names a codec nor restates the owning-codec bound. `ProjectionError::{Apply, Commit}` keeps the fold and persist domains distinct (rule 3). Prior art: Akka Projection (inert value, host spawns). The old 8-arg/6-generic hand-wired `run_projection` (which leaked the HRTB) is gone; `examples/projection-tokio` now builds its loop on the stepper. - **`repository.rs`** — Aggregate-facing trait + its single facade impl. - `Repository` trait: high-level `load` and `save` for aggregates. - `EventStore` facade — **one terminal for both owning and borrowing codecs** (#248 `5b7dc29` unified the former `EventStore`/`ZeroCopyEventStore` into one `.build()`). Bound: `for<'a> C: Encode + Decode: Borrow> + 'static`. The owning-vs-borrowing distinction is *inferred from the `Decode::Output` GAT*, not restated: `std`'s `Borrow for T` and `Borrow for &T` make `Output<'a>: Borrow` accept both an owning codec (`Output = E`, serde — one alloc per event) and a zero-copy codec (`Output = &'a E`, POD/rkyv — zero alloc); the decoded value is fed to `replay` via `out.borrow()` in either arm (consumed in place, never escapes → sidesteps the GAT `'static` implication). **`OwningCodec` (the `Output = E` *equality* alias) deliberately does NOT apply here** — it would exclude the zero-copy codecs this facade exists to serve. The alias fits only genuine *owning-output* sites (`.decoded()`, snapshot decode); the facade is a *borrow-and-consume-in-place* site, hence the more general `Borrow` bound. The bound is maintainer-facing (on impl blocks); callers use the builder and never spell it, so there is no HRTB-hiding win to chase here. The aggregate `A` is a phantom type parameter (`PhantomData A>`) so the facade implements `Repository` for **exactly one** `A` — with `A` fixed on the type, `load`/`save` infer the aggregate with no per-call annotation. `A` is named once at `Store::repository::()` (#243). @@ -115,7 +115,7 @@ The workspace pins `fjall = { version = "3", features = ["bytes_1"] }`. That fea After the cursor unification, `nexus-fjall` holds **only fjall-specific code**: the catch-up-then-live-tail loop now lives generically in `nexus-store` (see the subscription machinery above), so the four bespoke cursor files (`stream.rs`, `all_stream.rs`, `subscription_stream.rs`, `all_subscription_stream.rs` — `FjallStream` / `FjallAllStream` / `FjallSubscriptionStream` / `FjallAllSubscriptionStream`) and the old `VecDeque`/batch refill machinery were deleted. -- **`store.rs`** — `FjallStore` implements `RawEventStore` (`append` + `read_stream` + `read_all`, the latter two returning a `ScanCursor`), `WakeSource` (delegating `register`/`wake` to its `StreamNotifiers`; `wake` bumps both the per-stream and `$all` paths), under the `snapshot` feature `SnapshotStore, Version>`, and under the `import`/`export` features `AtomicAppend` + `StreamLister` (#220). Partitions: `streams` (id bytes → version counter, point-read optimized), `events` (event rows, scan-optimized, LZ4 compressed), `global` (one key holding the store-wide `GlobalSeq` counter), and `snapshots` (under the `snapshot` feature). All writes go through one atomic fjall `write_tx`; `append` claims its `GlobalSeq` range inside the same transaction as the event writes. `append` is decomposed into helpers (`read_current_version`, `check_optimistic`, `read_current_global`) and uses running `checked_add` counters for the version sequence and `GlobalSeq` — no `unwrap_or(u64::MAX)` sentinels (rule 2). The `read_current_version`/`read_current_global` helpers each wrap a bare `read_*_raw` core (returning `FjallError` not `AppendError`) so the `AtomicAppend` path can consume them in its own error domain without inventing an impossible variant. `atomic_append_many` mirrors `InMemoryStore`'s two-phase discipline (validate every run's head against a **running projected head** → stage + assign `GlobalSeq` → commit) entirely inside **one** `write_tx` spanning all four partitions: any conflict/overflow/encode failure drops the tx uncommitted, so **nothing** lands across **any** partition (rule 1 — the freeze-gating guarantee, proven by in-crate white-box per-partition rollback tests). `StreamLister::list_streams` is a lazy snapshot-pinned `fjall::Iter` over the `streams` partition's keys (key-only `Guard::key()`, zero-copy `Slice → Bytes`), not an eager `collect()`. Each event row is built via `nexus_store::wire::encode_frame` so the on-disk payload bytes land on a 16-byte boundary. (Adapter limitation: fjall/lsm-tree rejects an **empty key**, so an empty stream id is unsupported at `append` — its `streams` key is the raw id bytes.) +- **`store.rs`** — `FjallStore` implements `RawEventStore` (`append` + `read_stream` + `read_all`, the latter two returning a `ScanCursor`), `WakeSource` (delegating `register`/`wake` to its `StreamNotifiers`; `wake` bumps both the per-stream and `$all` paths), under the `snapshot` feature `SnapshotStore, Version>` (aggregate snapshots), under the `projection` feature `SnapshotStore, GlobalSeq>` (projection state — the folded state plus its `$all` `GlobalSeq` resume position, committed together; #164), and under the `import`/`export` features `AtomicAppend` + `StreamLister` (#220). The two `SnapshotStore` impls are **distinct trait instantiations** (`P = Version` vs `P = GlobalSeq`), so both coexist on `FjallStore` with no coherence conflict, each on its **own** point-read partition so a projection id can never collide with an aggregate-snapshot id sharing the same bytes. `commit` is a single-key insert of one combined `[u32 schema][u64 position][payload]` value; the guarantee is **structural, not fsync-dependent** — because state and position are one value, a persisted checkpoint can never be *ahead* of the state it describes, so a lost write costs a re-fold (idempotent), never a skipped event. That is why no `write_tx` is needed (same reasoning the snapshot path relies on). Partitions: `streams` (id bytes → version counter, point-read optimized), `events` (event rows, scan-optimized, LZ4 compressed), `global` (one key holding the store-wide `GlobalSeq` counter), `snapshots` (under the `snapshot` feature), and `projections` (under the `projection` feature, config `projection_defaults` = point-read blocks + bloom **plus all-levels LZ4**). The `projections` LZ4 choice and the separate-partition choice were both **measured** (rule 9), by the reproducible `benches/projection_storage`: (1) projection state is larger and more compressible than a snapshot, and all-levels LZ4 shrinks compressible state **10–100×** on disk vs the plain point-read config (fjall's default `[None,None,Lz4]` barely engages) with no cost on incompressible payloads — a one-sided flash/storage win on the `IoT`/mobile target; (2) a separate `projections` partition costs **+0.0%** on disk vs a single key-prefixed partition, so the split is free and buys collision-safety by construction plus independent per-partition config (which #1 shows matters), at the price of only one extra memtable's RAM — measured → kept separate. All writes go through one atomic fjall `write_tx`; `append` claims its `GlobalSeq` range inside the same transaction as the event writes. `append` is decomposed into helpers (`read_current_version`, `check_optimistic`, `read_current_global`) and uses running `checked_add` counters for the version sequence and `GlobalSeq` — no `unwrap_or(u64::MAX)` sentinels (rule 2). The `read_current_version`/`read_current_global` helpers each wrap a bare `read_*_raw` core (returning `FjallError` not `AppendError`) so the `AtomicAppend` path can consume them in its own error domain without inventing an impossible variant. `atomic_append_many` mirrors `InMemoryStore`'s two-phase discipline (validate every run's head against a **running projected head** → stage + assign `GlobalSeq` → commit) entirely inside **one** `write_tx` spanning all four partitions: any conflict/overflow/encode failure drops the tx uncommitted, so **nothing** lands across **any** partition (rule 1 — the freeze-gating guarantee, proven by in-crate white-box per-partition rollback tests). `StreamLister::list_streams` is a lazy snapshot-pinned `fjall::Iter` over the `streams` partition's keys (key-only `Guard::key()`, zero-copy `Slice → Bytes`), not an eager `collect()`. Each event row is built via `nexus_store::wire::encode_frame` so the on-disk payload bytes land on a 16-byte boundary. (Adapter limitation: fjall/lsm-tree rejects an **empty key**, so an empty stream id is unsupported at `append` — its `streams` key is the raw id bytes.) - **`scan.rs`** — fjall-private bounded read cursor (NOT exported; no other adapter shares fjall's on-disk key layout). `ScanStrategy` trait factors the only parts that differ between the per-stream (`Version`-keyed) and `$all` (`GlobalSeq`-keyed) reads — the keyset bound bytes (`lower_key`/`upper_key`) and how a stored row decodes into a `PersistedEnvelope`; `StreamScan` / `GlobalScan` are the two impls. `ScanCursor` is ONE bounded cursor over a SINGLE lazy `fjall::Iter` (`keyspace.range(lower..=upper)`): fjall's range iterator is already a lazy k-way merge that pulls the next LSM block only when the current drains, so a single `Iter` bounds memory with no batching/refill needed. **Snapshot-consistent at `open`** — it pins one repeatable-read view as of open, so events appended after `open()` are not observed mid-scan (which is why a never-ending subscription opens a fresh `ScanCursor` per refill rather than holding one). Poisons (yields `None`) after the first error rather than silently skipping corrupt rows. `S: Unpin` bound (relied on by the generic live loop). - **`builder.rs`** — `FjallStoreBuilder`: `FjallStore::builder(path).streams_config(fn).events_config(fn).open()`. The `batch_size` knob was removed — the single lazy `ScanCursor` makes a resident-row cap meaningless. - **`partition.rs`** — Sealed `KeyspaceConfig` trait. Only `()` (defaults) and `FnOnce(KeyspaceCreateOptions) -> KeyspaceCreateOptions` closures implement it. Eliminates dynamic dispatch from the builder by monomorphising keyspace configuration at compile time. diff --git a/crates/nexus-fjall/Cargo.toml b/crates/nexus-fjall/Cargo.toml index 7046cb15..9646b302 100644 --- a/crates/nexus-fjall/Cargo.toml +++ b/crates/nexus-fjall/Cargo.toml @@ -13,6 +13,7 @@ categories = ["database"] [features] snapshot = ["nexus-store/snapshot"] +projection = ["nexus-store/projection"] export = ["nexus-store/export"] import = ["nexus-store/import"] @@ -36,7 +37,7 @@ insta = { workspace = true } # default features only). Same unification trick `nexus-store` uses for its # testing/export/import/cbor tests. `cargo add` refuses a self-dependency, so # this is a deliberate manual entry. -nexus-fjall = { path = ".", features = ["export", "import"] } +nexus-fjall = { path = ".", features = ["export", "import", "snapshot", "projection"] } # Pulls the CBOR backup box (`nexus_store::cbor`) into the test build; `cbor` # implies nexus-store's `export`+`import`, so the end-to-end round-trip tests # can encode/decode chunks. Separate from the line above because `cbor` is a @@ -52,5 +53,12 @@ tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } name = "fjall_benchmarks" harness = false +# Storage-cost measurement for the projection-state partition (#164 follow-up). +# A deterministic reporting binary (not criterion): prints on-disk footprint +# tables for the partition-config and separate-vs-prefixed forks. +[[bench]] +name = "projection_storage" +harness = false + [lints] workspace = true diff --git a/crates/nexus-fjall/benches/projection_storage.rs b/crates/nexus-fjall/benches/projection_storage.rs new file mode 100644 index 00000000..dfa85903 --- /dev/null +++ b/crates/nexus-fjall/benches/projection_storage.rs @@ -0,0 +1,342 @@ +//! Reproducible storage-cost benchmark for the projection-state partition +//! (issues #164 follow-up; CLAUDE rule 9 — "measure the fork, don't assert it"). +//! +//! Run: +//! +//! ```sh +//! nix develop -c cargo bench -p nexus-fjall --bench projection_storage +//! ``` +//! +//! It is deterministic (seeded xorshift for "random" payloads, fixed sizes and +//! counts), so the on-disk numbers reproduce run to run; only the wall-clock +//! timings drift and are reported as indicative, not load-bearing. The decision +//! metric is `disk_space()` — fjall's SSTable footprint — which is what a +//! flash-constrained `IoT`/mobile device actually pays. +//! +//! ## Fork 1 — projection-partition config +//! +//! `nexus_fjall`'s `projections` partition uses `point_read_defaults`: 4 KiB +//! data blocks + a 15-bit bloom filter, and — because it sets no compression +//! policy — fjall's default `[None, None, Lz4]` (L0/L1 uncompressed, L2+ LZ4). +//! It is tuned for small point-read metadata, but projection *state* can be +//! large. This measures three configs across payload sizes × compressibility: +//! +//! - `current` — 4 KiB blocks, default `[None,None,Lz4]`, bloom (today's config) +//! - `scan_lz4` — 32 KiB blocks, all-levels LZ4 (the `events` scan config) +//! - `pr_lz4` — 4 KiB blocks, all-levels LZ4, bloom (point-read + full LZ4) +//! +//! ## Fork 2 — separate partition vs key-prefix +//! +//! `snapshots` and `projections` are separate fjall partitions (separate LSM +//! trees). The alternative is one partition with `s:`/`p:` key prefixes. This +//! measures the on-disk delta and reports the fixed per-partition cost (each +//! partition carries its own active memtable up to `max_memtable_size`). + +#![allow(clippy::unwrap_used, reason = "bench code")] +#![allow(clippy::expect_used, reason = "bench code")] +#![allow(clippy::missing_panics_doc, reason = "bench code")] +#![allow( + clippy::as_conversions, + clippy::cast_precision_loss, + reason = "ratio reporting divides u64 byte counts as f64" +)] +#![allow(clippy::shadow_reuse, reason = "bench-local rebinding is fine")] +#![allow( + clippy::print_stdout, + reason = "reporting bench: emitting the result tables is the point" +)] +#![allow(clippy::doc_markdown, reason = "bench doc is prose, not API")] +#![allow( + clippy::type_complexity, + reason = "the config table is a small slice of (name, fn) pairs" +)] +#![allow( + clippy::items_after_statements, + reason = "bench-local consts sit next to their use site" +)] + +use std::time::Instant; + +use fjall::config::{ + BlockSizePolicy, BloomConstructionPolicy, CompressionPolicy, FilterPolicy, FilterPolicyEntry, +}; +use fjall::{CompressionType, KeyspaceCreateOptions, SingleWriterTxDatabase}; +use tempfile::TempDir; + +/// Small memtable so writing a few MB rotates several times and the bulk of the +/// data lands in (compressed) SSTables rather than the active memtable — so +/// `disk_space()` reflects the on-disk, post-compression cost. +const MEMTABLE_BYTES: u64 = 256 * 1024; + +/// The 12-byte snapshot/projection value header +/// (`[u32 LE schema_version][u64 BE position]`), mirrored so measured values are +/// the same shape the real `write_projection` stores. +const VALUE_HEADER: usize = 12; + +// ── partition configs (mirror `nexus_fjall::partition`) ───────────────────── + +fn bloom() -> FilterPolicy { + FilterPolicy::all(FilterPolicyEntry::Bloom( + BloomConstructionPolicy::BitsPerKey(15.0), + )) +} + +/// Today's `projections` config: `point_read_defaults` (no explicit compression +/// → fjall default `[None, None, Lz4]`). +fn current_config() -> KeyspaceCreateOptions { + KeyspaceCreateOptions::default() + .max_memtable_size(MEMTABLE_BYTES) + .data_block_size_policy(BlockSizePolicy::all(4_096)) + .filter_policy(bloom()) + .expect_point_read_hits(true) +} + +/// The `events` scan config: 32 KiB blocks, all-levels LZ4, no bloom. +fn scan_lz4_config() -> KeyspaceCreateOptions { + KeyspaceCreateOptions::default() + .max_memtable_size(MEMTABLE_BYTES) + .data_block_size_policy(BlockSizePolicy::all(32_768)) + .data_block_compression_policy(CompressionPolicy::all(CompressionType::Lz4)) +} + +/// Hybrid: point-read 4 KiB blocks + bloom, but LZ4 on every level. +fn pr_lz4_config() -> KeyspaceCreateOptions { + KeyspaceCreateOptions::default() + .max_memtable_size(MEMTABLE_BYTES) + .data_block_size_policy(BlockSizePolicy::all(4_096)) + .data_block_compression_policy(CompressionPolicy::all(CompressionType::Lz4)) + .filter_policy(bloom()) + .expect_point_read_hits(true) +} + +// ── deterministic payloads ────────────────────────────────────────────────── + +/// Structured, highly-compressible payload — a repeated record, standing in for +/// real projection state (JSON/struct with repeated keys). +fn structured(size: usize) -> Vec { + let unit = br#"{"id":1234567,"count":42,"total":1000000,"status":"active"},"#; + let mut v = Vec::with_capacity(size + unit.len()); + while v.len() < size { + v.extend_from_slice(unit); + } + v.truncate(size); + v +} + +/// Incompressible payload — a seeded xorshift stream (deterministic). +fn random(size: usize, state: &mut u64) -> Vec { + let mut v = Vec::with_capacity(size + 8); + while v.len() < size { + let mut x = *state; + x ^= x << 13; + x ^= x >> 7; + x ^= x << 17; + *state = x; + v.extend_from_slice(&x.to_le_bytes()); + } + v.truncate(size); + v +} + +fn value(payload: &[u8]) -> Vec { + let mut buf = Vec::with_capacity(VALUE_HEADER + payload.len()); + buf.extend_from_slice(&1u32.to_le_bytes()); + buf.extend_from_slice(&1u64.to_be_bytes()); + buf.extend_from_slice(payload); + buf +} + +// ── measurement ───────────────────────────────────────────────────────────── + +struct SizeCase { + label: &'static str, + bytes: usize, + count: usize, +} + +/// Payload sizes × blob counts, chosen so each config writes a few–tens of MB +/// (enough to trigger memtable rotations) while the bench stays quick. +const SIZE_CASES: &[SizeCase] = &[ + SizeCase { + label: "128 B", + bytes: 128, + count: 8_000, + }, + SizeCase { + label: "1 KiB", + bytes: 1_024, + count: 4_000, + }, + SizeCase { + label: "16 KiB", + bytes: 16_384, + count: 512, + }, + SizeCase { + label: "256 KiB", + bytes: 262_144, + count: 96, + }, +]; + +struct Sample { + disk: u64, + logical: u64, + write_ms: u128, +} + +/// Write `count` blobs of `bytes` each under `config`, flush to SSTables, and +/// return the on-disk footprint, the logical (pre-storage) byte total, and the +/// write wall-clock. +fn measure(config: KeyspaceCreateOptions, case: &SizeCase, compressible: bool) -> Sample { + let dir = TempDir::new().expect("tempdir"); + let db = SingleWriterTxDatabase::builder(dir.path().join("db")) + .open() + .expect("open db"); + let ks = db.keyspace("bench", || config).expect("keyspace"); + + let mut seed = 0x9E37_79B9_7F4A_7C15_u64; + let mut logical = 0u64; + let start = Instant::now(); + for i in 0..case.count { + let payload = if compressible { + structured(case.bytes) + } else { + random(case.bytes, &mut seed) + }; + let v = value(&payload); + logical += v.len() as u64; + ks.insert(format!("proj-{i:08}").as_bytes(), v) + .expect("insert"); + } + let write_ms = start.elapsed().as_millis(); + + // Force the active memtable to SSTable so compression is realised on disk. + ks.inner().rotate_memtable_and_wait().expect("flush"); + + Sample { + disk: ks.inner().disk_space(), + logical, + write_ms, + } +} + +fn fork1_partition_config() { + println!("\n## Fork 1 — projection-partition config\n"); + println!("Disk = fjall `disk_space()` (SSTable bytes) after flush. `×` = disk / logical.\n"); + println!("| payload | data | config | logical | disk | ×raw | write |"); + println!("|--------:|:-----|:-------|--------:|-----:|-----:|------:|"); + + let configs: &[(&str, fn() -> KeyspaceCreateOptions)] = &[ + ("current", current_config), + ("scan_lz4", scan_lz4_config), + ("pr_lz4", pr_lz4_config), + ]; + + for case in SIZE_CASES { + for (compressible, kind) in [(true, "structured"), (false, "random")] { + for (name, cfg) in configs { + let s = measure(cfg(), case, compressible); + println!( + "| {} | {} | {} | {} | {} | {:.2} | {} ms |", + case.label, + kind, + name, + mib(s.logical), + mib(s.disk), + s.disk as f64 / s.logical as f64, + s.write_ms, + ); + } + } + } +} + +fn fork2_partition_vs_prefix() { + println!("\n## Fork 2 — separate partitions vs one key-prefixed partition\n"); + + // Representative mixed workload: small aggregate snapshots + larger + // projection state, equal counts. + const SNAP_BYTES: usize = 256; + const PROJ_BYTES: usize = 4_096; + const COUNT: usize = 3_000; + + // (a) two separate partitions. + let sep_dir = TempDir::new().unwrap(); + let sep_db = SingleWriterTxDatabase::builder(sep_dir.path().join("db")) + .open() + .unwrap(); + let snaps = sep_db.keyspace("snapshots", current_config).unwrap(); + let projs = sep_db.keyspace("projections", current_config).unwrap(); + for i in 0..COUNT { + snaps + .insert( + format!("a-{i:08}").as_bytes(), + value(&structured(SNAP_BYTES)), + ) + .unwrap(); + projs + .insert( + format!("p-{i:08}").as_bytes(), + value(&structured(PROJ_BYTES)), + ) + .unwrap(); + } + snaps.inner().rotate_memtable_and_wait().unwrap(); + projs.inner().rotate_memtable_and_wait().unwrap(); + let sep_disk = snaps.inner().disk_space() + projs.inner().disk_space(); + + // (b) one partition, `s:` / `p:` key prefixes. + let pre_dir = TempDir::new().unwrap(); + let pre_db = SingleWriterTxDatabase::builder(pre_dir.path().join("db")) + .open() + .unwrap(); + let one = pre_db.keyspace("state", current_config).unwrap(); + for i in 0..COUNT { + one.insert( + format!("s:a-{i:08}").as_bytes(), + value(&structured(SNAP_BYTES)), + ) + .unwrap(); + one.insert( + format!("p:p-{i:08}").as_bytes(), + value(&structured(PROJ_BYTES)), + ) + .unwrap(); + } + one.inner().rotate_memtable_and_wait().unwrap(); + let pre_disk = one.inner().disk_space(); + + println!( + "Workload: {COUNT} × {SNAP_BYTES} B snapshots + {COUNT} × {PROJ_BYTES} B projections.\n" + ); + println!("| layout | partitions (memtables) | disk |"); + println!("|:-------|:----------------------:|-----:|"); + println!("| separate | 2 | {} |", mib(sep_disk)); + println!("| prefixed | 1 | {} |", mib(pre_disk)); + println!( + "\nOn-disk delta: {:+.1}% (prefixed vs separate). Each separate partition also \ + holds its own active memtable (up to {} KiB resident), so N partitions = N write \ + buffers — the fixed RAM cost that dominates on a constrained device.", + (pre_disk as f64 / sep_disk as f64 - 1.0) * 100.0, + MEMTABLE_BYTES / 1024, + ); +} + +fn mib(bytes: u64) -> String { + format!("{:.2} MiB", bytes as f64 / (1024.0 * 1024.0)) +} + +fn main() { + println!("# Projection storage benchmark (fjall {})", fjall_version()); + println!( + "\nDeterministic; memtable cap {} KiB. Decision metric = on-disk `disk_space()`.", + MEMTABLE_BYTES / 1024 + ); + fork1_partition_config(); + fork2_partition_vs_prefix(); +} + +const fn fjall_version() -> &'static str { + "3.1.x" +} diff --git a/crates/nexus-fjall/src/builder.rs b/crates/nexus-fjall/src/builder.rs index 96d9abb7..023a84c2 100644 --- a/crates/nexus-fjall/src/builder.rs +++ b/crates/nexus-fjall/src/builder.rs @@ -1,4 +1,6 @@ use crate::error::FjallError; +#[cfg(feature = "projection")] +use crate::partition::projection_defaults; use crate::partition::{AllIndex, KeyspaceConfig, Partitions, point_read_defaults, scan_defaults}; use crate::store::FjallStore; use fjall::KeyspaceCreateOptions; @@ -121,17 +123,27 @@ impl FjallStoreBuilder { #[cfg(feature = "snapshot")] let snapshots = db.keyspace("snapshots", point_read_defaults)?; + // Projection state lives in its own point-read keyspace, distinct from + // `snapshots`, so a projection id and an aggregate-snapshot id with the + // same bytes cannot collide. It adds all-levels LZ4 (see + // `projection_defaults`): projection state is larger and more compressible + // than a snapshot, and the measurement showed it shrinks 10–100× with no + // cost on incompressible data. + #[cfg(feature = "projection")] + let projections = db.keyspace("projections", projection_defaults)?; + let global = db.keyspace("global", point_read_defaults)?; Ok(FjallStore { db, partitions: Partitions::new( streams, - events, - events_global, + (events, events_global), global, #[cfg(feature = "snapshot")] snapshots, + #[cfg(feature = "projection")] + projections, ) .with_all_index(self.all_index), notifiers: StreamNotifiers::new(), diff --git a/crates/nexus-fjall/src/lib.rs b/crates/nexus-fjall/src/lib.rs index 356502ba..7ecd3f4b 100644 --- a/crates/nexus-fjall/src/lib.rs +++ b/crates/nexus-fjall/src/lib.rs @@ -66,7 +66,7 @@ mod global_seq; mod partition; mod plan; mod scan; -#[cfg(feature = "snapshot")] +#[cfg(any(feature = "snapshot", feature = "projection"))] mod snapshot; mod store; mod subscription_id; diff --git a/crates/nexus-fjall/src/partition.rs b/crates/nexus-fjall/src/partition.rs index c9c52a31..ca19704e 100644 --- a/crates/nexus-fjall/src/partition.rs +++ b/crates/nexus-fjall/src/partition.rs @@ -71,6 +71,24 @@ pub fn scan_defaults() -> KeyspaceCreateOptions { .data_block_compression_policy(CompressionPolicy::all(CompressionType::Lz4)) } +/// Options for the projection-state partition: the point-read metadata tuning +/// (4 KiB blocks + 15-bit bloom, because projection state is point-read by id) +/// **plus all-levels LZ4**. +/// +/// Projection state, unlike a small aggregate snapshot or counter, can be large +/// and is typically structured (records, JSON) — i.e. highly compressible. The +/// reproducible `benches/projection_storage` measurement showed the plain +/// `point_read_defaults` barely compresses this workload (fjall's default +/// `[None, None, Lz4]` only engages at deep levels), while all-levels LZ4 shrinks +/// compressible state **10–100×** on disk with no measurable cost on +/// incompressible payloads. Flash-write volume and storage are the binding +/// constraints on the `IoT`/mobile target, so LZ4-on is the safe default. +#[cfg(feature = "projection")] +pub fn projection_defaults() -> KeyspaceCreateOptions { + point_read_defaults() + .data_block_compression_policy(CompressionPolicy::all(CompressionType::Lz4)) +} + /// The single key under which the store-global sequence counter is kept in the /// `global` partition. const GLOBAL_SEQ_KEY: &[u8] = b"global_seq"; @@ -123,6 +141,8 @@ pub struct Partitions { global: SingleWriterTxKeyspace, #[cfg(feature = "snapshot")] snapshots: SingleWriterTxKeyspace, + #[cfg(feature = "projection")] + projections: SingleWriterTxKeyspace, /// Whether the `$all` index (`events_global`) is maintained — gates the /// second write in [`stage_event`](Self::stage_event) and `read_all`. mode: AllIndex, @@ -132,12 +152,21 @@ impl Partitions { /// Assemble from the keyspaces the builder opened, with the default /// (`Denormalized`) `$all` index. Use [`with_all_index`](Self::with_all_index) /// to select a different mode. - pub const fn new( + /// + /// The two event-frame partitions are passed as one `(events, events_global)` + /// pair — they always travel together (normalized + `$all` denormalized copy, + /// configured from the same `events_config` closure), so grouping them keeps + /// the arity within the project's five-argument ceiling. + /// + /// Not `const`: destructuring the event-frame tuple would require dropping + /// the pair at compile time, which const eval forbids. `new` is only called + /// at runtime from the builder's `open`, so `const` bought nothing here. + pub fn new( streams: SingleWriterTxKeyspace, - events: SingleWriterTxKeyspace, - events_global: SingleWriterTxKeyspace, + (events, events_global): (SingleWriterTxKeyspace, SingleWriterTxKeyspace), global: SingleWriterTxKeyspace, #[cfg(feature = "snapshot")] snapshots: SingleWriterTxKeyspace, + #[cfg(feature = "projection")] projections: SingleWriterTxKeyspace, ) -> Self { Self { streams, @@ -146,6 +175,8 @@ impl Partitions { global, #[cfg(feature = "snapshot")] snapshots, + #[cfg(feature = "projection")] + projections, mode: AllIndex::Denormalized, } } @@ -266,6 +297,33 @@ impl Partitions { .map_err(FjallError::Io) } + // ----- projections (separate keyspace from snapshots) --------------- + + /// Point-read a projection state blob by id from the `projections` keyspace + /// — a distinct partition from `snapshots`, so a projection id can never + /// collide with an aggregate-snapshot id that shares the same bytes. + #[cfg(feature = "projection")] + pub fn read_projection(&self, id: &[u8]) -> Result, FjallError> { + self.projections.get(id).map_err(FjallError::Io) + } + + /// Write a projection state blob for id into the `projections` keyspace. + /// + /// The blob bundles `(schema_version, position, state)` into **one value + /// under one key**, so the state and the `GlobalSeq` it was folded up to are + /// never stored apart. The load-bearing guarantee is structural, not a claim + /// about fjall's fsync: because the two are one value, a persisted checkpoint + /// can **never be ahead of the state it describes**. So even if this write is + /// lost on a crash (`IoT` power-loss), the host resumes from the last *durable* + /// checkpoint and re-folds forward — idempotent, never skipping events. That + /// is why projection checkpointing needs no cross-partition transaction. + #[cfg(feature = "projection")] + pub fn write_projection(&self, id: &[u8], bytes: &[u8]) -> Result<(), FjallError> { + self.projections + .insert(id, Slice::from(bytes)) + .map_err(FjallError::Io) + } + // ----- white-box test access ---------------------------------------- /// The `streams` keyspace. `#[cfg(test)]` — white-box tests inspect the @@ -281,4 +339,11 @@ impl Partitions { pub const fn global(&self) -> &SingleWriterTxKeyspace { &self.global } + + /// The `projections` keyspace. `#[cfg(test)]` — white-box tests write raw + /// (possibly corrupt) bytes directly to exercise the decode error path. + #[cfg(all(test, feature = "projection"))] + pub const fn projections(&self) -> &SingleWriterTxKeyspace { + &self.projections + } } diff --git a/crates/nexus-fjall/src/snapshot.rs b/crates/nexus-fjall/src/snapshot.rs index 44e38b10..02fca5f1 100644 --- a/crates/nexus-fjall/src/snapshot.rs +++ b/crates/nexus-fjall/src/snapshot.rs @@ -1,8 +1,11 @@ -//! Snapshot value codec (feature `snapshot`). +//! Snapshot / projection value codec (features `snapshot`, `projection`). //! -//! The snapshot blob layout is fjall-private and distinct from the event -//! wire format: `[u32 LE schema_version][u64 BE version][payload]`. Used only -//! by the `SnapshotStore, Version>` impl in [`crate::store`]. +//! The blob layout is fjall-private and distinct from the event wire format: +//! `[u32 LE schema_version][u64 BE position][payload]`. It backs both +//! `SnapshotStore` impls in [`crate::store`] — the `u64` `position` field is an +//! aggregate `Version` for `SnapshotStore, Version>` and a `GlobalSeq` +//! for `SnapshotStore, GlobalSeq>` (the projection resume position); the +//! codec sees only the underlying `u64`, so one shape serves both. use crate::wire_key::DecodeError; diff --git a/crates/nexus-fjall/src/store.rs b/crates/nexus-fjall/src/store.rs index 14ba38b2..bee042bb 100644 --- a/crates/nexus-fjall/src/store.rs +++ b/crates/nexus-fjall/src/store.rs @@ -222,7 +222,7 @@ mod snapshot_impl { use super::{ErrorId, FjallError, FjallStore, Version}; use crate::snapshot::{decode_snapshot_value, encode_snapshot_value}; use nexus::Id; - use nexus_store::state::SnapshotStore; + use nexus_store::state::{Hydrated, SnapshotStore}; use std::num::NonZeroU32; impl SnapshotStore, Version> for FjallStore { @@ -232,10 +232,10 @@ mod snapshot_impl { &self, id: &impl Id, schema_version: NonZeroU32, - ) -> Result)>, FjallError> { + ) -> Result, Version>, FjallError> { // Snapshot key is the ID bytes directly. let Some(bytes) = self.partitions.read_snapshot(id.as_ref())? else { - return Ok(None); + return Ok(Hydrated::Absent); }; let (schema_version_raw, version_raw, payload) = decode_snapshot_value(&bytes) @@ -244,17 +244,26 @@ mod snapshot_impl { version: None, })?; - // Schema mismatch → treat as absent (avoids cloning a stale payload). - if schema_version_raw != schema_version.get() { - return Ok(None); + // A stored schema version is always >= 1 (commit writes a + // `NonZeroU32`); a zero is corruption, not a stale snapshot. + let stored_schema = + NonZeroU32::new(schema_version_raw).ok_or_else(|| FjallError::CorruptValue { + stream_id: ErrorId::from_display(id), + version: None, + })?; + if stored_schema != schema_version { + return Ok(Hydrated::Stale { stored_schema }); } - let version = Version::new(version_raw).ok_or_else(|| FjallError::CorruptValue { + let position = Version::new(version_raw).ok_or_else(|| FjallError::CorruptValue { stream_id: ErrorId::from_display(id), version: None, })?; - Ok(Some((version, payload.to_vec()))) + Ok(Hydrated::Found { + position, + state: payload.to_vec(), + }) } async fn commit( @@ -271,6 +280,89 @@ mod snapshot_impl { } } +// ═══════════════════════════════════════════════════════════════════════════ +// SnapshotStore, GlobalSeq> implementation — projection state +// ═══════════════════════════════════════════════════════════════════════════ + +/// Projection-state persistence: the projection's folded state plus the +/// `$all` [`GlobalSeq`] it was folded up to, committed together into one value +/// under one key in the dedicated `projections` partition. +/// +/// A separate trait instantiation from the aggregate-snapshot impl +/// (`P = GlobalSeq` vs `P = Version`), so both coexist on `FjallStore` with no +/// coherence conflict. The value reuses the snapshot blob layout — `position` +/// is the `GlobalSeq`'s underlying `u64` — so no new encoder shape is needed. +/// +/// State and position ride in one value under one key, so the checkpoint is +/// never ahead of the state it describes; a lost `commit` costs a re-fold, never +/// a skipped event (the `write_projection` docs spell out why). +#[cfg(feature = "projection")] +mod projection_impl { + use super::{ErrorId, FjallError, FjallStore, GlobalSeq}; + use crate::snapshot::{decode_snapshot_value, encode_snapshot_value}; + use nexus::Id; + use nexus_store::state::{Hydrated, SnapshotStore}; + use std::num::NonZeroU32; + + impl SnapshotStore, GlobalSeq> for FjallStore { + type Error = FjallError; + + async fn hydrate( + &self, + id: &impl Id, + schema_version: NonZeroU32, + ) -> Result, GlobalSeq>, FjallError> { + // Projection key is the id bytes directly, in the `projections` + // partition (never the `snapshots` one). + let Some(bytes) = self.partitions.read_projection(id.as_ref())? else { + return Ok(Hydrated::Absent); + }; + + let (schema_version_raw, position_raw, payload) = decode_snapshot_value(&bytes) + .map_err(|_| FjallError::CorruptValue { + stream_id: ErrorId::from_display(id), + version: None, + })?; + + // A stored schema version is always >= 1 (commit writes a + // `NonZeroU32`); a zero is corruption, not a stale checkpoint. A + // mismatch is `Stale` — the host must re-fold from the beginning. + let stored_schema = + NonZeroU32::new(schema_version_raw).ok_or_else(|| FjallError::CorruptValue { + stream_id: ErrorId::from_display(id), + version: None, + })?; + if stored_schema != schema_version { + return Ok(Hydrated::Stale { stored_schema }); + } + + // A stored `GlobalSeq` is always >= 1; a zero means corrupt bytes. + let position = + GlobalSeq::new(position_raw).ok_or_else(|| FjallError::CorruptValue { + stream_id: ErrorId::from_display(id), + version: None, + })?; + + Ok(Hydrated::Found { + position, + state: payload.to_vec(), + }) + } + + async fn commit( + &self, + id: &impl Id, + schema_version: NonZeroU32, + position: GlobalSeq, + state: &Vec, + ) -> Result<(), FjallError> { + let mut buf = Vec::new(); + encode_snapshot_value(&mut buf, schema_version.get(), position.as_u64(), state); + self.partitions.write_projection(id.as_ref(), &buf) + } + } +} + // ═══════════════════════════════════════════════════════════════════════════ // AtomicAppend — commit N per-stream runs in ONE cross-partition transaction // (export/import, issue #145 + #220). The primitive `Atomicity::WholeChunk` @@ -1261,6 +1353,71 @@ mod tests { assert_eq!(store.partitions.events_global().inner().iter().count(), 1); assert!(store.read_all(None).await.is_ok()); } + + // ---- projection SnapshotStore, GlobalSeq> white-box tests ---- + + /// Defensive boundary (rule 7.3): a value in the `projections` partition + /// that is shorter than the 12-byte header must decode to a typed + /// `CorruptValue` error, never a panic or a silent wrong result. Written as + /// a white-box test because planting raw corrupt bytes needs direct access + /// to the `projections` keyspace, which the public API does not expose. + #[cfg(feature = "projection")] + #[tokio::test] + async fn projection_hydrate_corrupt_value_is_corrupt_not_panic() { + use nexus_store::state::SnapshotStore; + use std::num::NonZeroU32; + + let (store, _dir) = temp_store(); + let id = sk("proj-corrupt"); + + // Plant a too-short value (5 bytes < the 12-byte header) directly. + store + .partitions + .projections() + .insert(id.as_ref(), Slice::from(&b"short"[..])) + .unwrap(); + + let err = SnapshotStore::, GlobalSeq>::hydrate(&store, &id, NonZeroU32::MIN) + .await + .expect_err("corrupt bytes must surface an error, not None or a panic"); + assert!( + matches!(err, FjallError::CorruptValue { .. }), + "expected CorruptValue, got {err:?}" + ); + } + + /// A stored `GlobalSeq` position of 0 is impossible (`GlobalSeq >= 1`); a + /// well-formed header carrying a zero position is corruption, and must map + /// to `CorruptValue` rather than being resurrected as a valid position. + #[cfg(feature = "projection")] + #[tokio::test] + async fn projection_hydrate_zero_position_is_corrupt() { + use nexus_store::state::SnapshotStore; + use std::num::NonZeroU32; + + let (store, _dir) = temp_store(); + let id = sk("proj-zero"); + + // `[u32 LE schema_version=1][u64 BE position=0][payload]` — a valid + // header shape but an impossible GlobalSeq. + let mut value = Vec::new(); + value.extend_from_slice(&1u32.to_le_bytes()); + value.extend_from_slice(&0u64.to_be_bytes()); + value.push(0xAB); + store + .partitions + .projections() + .insert(id.as_ref(), Slice::from(&value[..])) + .unwrap(); + + let err = SnapshotStore::, GlobalSeq>::hydrate(&store, &id, NonZeroU32::MIN) + .await + .expect_err("zero GlobalSeq must be corruption"); + assert!( + matches!(err, FjallError::CorruptValue { .. }), + "expected CorruptValue, got {err:?}" + ); + } } // ═══════════════════════════════════════════════════════════════════════════ diff --git a/crates/nexus-fjall/tests/projection_tests.rs b/crates/nexus-fjall/tests/projection_tests.rs new file mode 100644 index 00000000..3ea4a5c7 --- /dev/null +++ b/crates/nexus-fjall/tests/projection_tests.rs @@ -0,0 +1,278 @@ +#![cfg(feature = "projection")] +#![allow( + clippy::unwrap_used, + clippy::expect_used, + clippy::panic, + clippy::shadow_reuse, + clippy::shadow_same, + clippy::shadow_unrelated, + clippy::as_conversions, + reason = "test harness — relaxed lints for test code" +)] + +//! `SnapshotStore, GlobalSeq>` — the fjall-backed projection-state +//! adapter (issue #164). Projection state and the `$all` [`GlobalSeq`] it was +//! folded up to are committed together into the dedicated `projections` +//! partition. These are the public-API tests; the corrupt-bytes defensive test +//! is a white-box unit test in `src/store.rs` (it needs raw partition access). + +use std::num::NonZeroU32; + +use nexus_fjall::{FjallStore, GlobalSeq}; +use nexus_store::StreamKey; +use nexus_store::state::{Hydrated, SnapshotStore}; +// Only the cross-partition collision test (both features) needs `Version`. +#[cfg(feature = "snapshot")] +use nexus::Version; + +const SV1: NonZeroU32 = NonZeroU32::MIN; + +fn pk(s: &str) -> StreamKey { + StreamKey::from_slice(s.as_bytes()) +} + +const fn gs(v: u64) -> GlobalSeq { + GlobalSeq::new(v).unwrap() +} + +fn temp_store() -> (FjallStore, tempfile::TempDir) { + let dir = tempfile::tempdir().unwrap(); + let store = FjallStore::builder(dir.path().join("db")).open().unwrap(); + (store, dir) +} + +// ── 1. Sequence/Protocol Tests ───────────────────────────────────── + +#[tokio::test] +async fn commit_then_hydrate_roundtrips() { + let (store, _dir) = temp_store(); + let id = pk("proj-1"); + + SnapshotStore::, GlobalSeq>::commit(&store, &id, SV1, gs(7), &vec![1, 2, 3]) + .await + .unwrap(); + + let (pos, state) = SnapshotStore::, GlobalSeq>::hydrate(&store, &id, SV1) + .await + .unwrap() + .into_found() + .unwrap(); + assert_eq!(pos, gs(7)); + assert_eq!(state, vec![1, 2, 3]); +} + +#[tokio::test] +async fn commit_overwrites_previous_projection() { + let (store, _dir) = temp_store(); + let id = pk("proj-1"); + + SnapshotStore::, GlobalSeq>::commit(&store, &id, SV1, gs(5), &vec![1]) + .await + .unwrap(); + + let sv2 = NonZeroU32::new(2).unwrap(); + SnapshotStore::, GlobalSeq>::commit(&store, &id, sv2, gs(42), &vec![2, 3]) + .await + .unwrap(); + + let (pos, state) = SnapshotStore::, GlobalSeq>::hydrate(&store, &id, sv2) + .await + .unwrap() + .into_found() + .unwrap(); + assert_eq!(pos, gs(42)); + assert_eq!(state, vec![2, 3]); + + // Old schema version is filtered at the store level → absent. + assert!( + SnapshotStore::, GlobalSeq>::hydrate(&store, &id, SV1) + .await + .unwrap() + .into_found() + .is_none() + ); +} + +// ── 2. Lifecycle Tests ───────────────────────────────────────────── + +#[tokio::test] +async fn projection_persists_across_reopen() { + let dir = tempfile::tempdir().unwrap(); + let db_path = dir.path().join("db"); + let id = pk("proj-1"); + + // First session: commit projection state at GlobalSeq 9. + { + let store = FjallStore::builder(&db_path).open().unwrap(); + SnapshotStore::, GlobalSeq>::commit(&store, &id, SV1, gs(9), &vec![42, 43, 44]) + .await + .unwrap(); + } + + // Second session: reopen and hydrate the exact same (GlobalSeq, state). + { + let store = FjallStore::builder(&db_path).open().unwrap(); + let (pos, state) = SnapshotStore::, GlobalSeq>::hydrate(&store, &id, SV1) + .await + .unwrap() + .into_found() + .unwrap(); + assert_eq!(pos, gs(9)); + assert_eq!(state, vec![42, 43, 44]); + } +} + +// ── 3. Defensive Boundary Tests ──────────────────────────────────── + +#[tokio::test] +async fn hydrate_unknown_id_is_absent() { + let (store, _dir) = temp_store(); + let result = SnapshotStore::, GlobalSeq>::hydrate(&store, &pk("nope"), SV1) + .await + .unwrap(); + assert_eq!(result, Hydrated::Absent); +} + +#[tokio::test] +async fn hydrate_schema_mismatch_is_stale_not_absent() { + let (store, _dir) = temp_store(); + let id = pk("proj-1"); + SnapshotStore::, GlobalSeq>::commit(&store, &id, SV1, gs(3), &vec![9]) + .await + .unwrap(); + + // A checkpoint at a *different* schema version reports `Stale` (carrying the + // stored schema), not `Absent` — so a host can tell a schema-bump rebuild + // from a fresh projection instead of silently re-folding all of `$all`. + let other = NonZeroU32::new(2).unwrap(); + let result = SnapshotStore::, GlobalSeq>::hydrate(&store, &id, other) + .await + .unwrap(); + assert_eq!(result, Hydrated::Stale { stored_schema: SV1 }); +} + +#[tokio::test] +async fn commit_at_initial_global_seq_roundtrips() { + let (store, _dir) = temp_store(); + let id = pk("proj-1"); + // Boundary: the very first GlobalSeq (1). + SnapshotStore::, GlobalSeq>::commit(&store, &id, SV1, GlobalSeq::INITIAL, &vec![0]) + .await + .unwrap(); + + let (pos, state) = SnapshotStore::, GlobalSeq>::hydrate(&store, &id, SV1) + .await + .unwrap() + .into_found() + .unwrap(); + assert_eq!(pos, GlobalSeq::INITIAL); + assert_eq!(state, vec![0]); +} + +#[tokio::test] +async fn commit_at_max_global_seq_roundtrips() { + let (store, _dir) = temp_store(); + let id = pk("proj-1"); + // Boundary: the largest GlobalSeq (u64::MAX). + SnapshotStore::, GlobalSeq>::commit(&store, &id, SV1, gs(u64::MAX), &vec![7]) + .await + .unwrap(); + + let (pos, state) = SnapshotStore::, GlobalSeq>::hydrate(&store, &id, SV1) + .await + .unwrap() + .into_found() + .unwrap(); + assert_eq!(pos, gs(u64::MAX)); + assert_eq!(state, vec![7]); +} + +#[tokio::test] +async fn commit_empty_state_roundtrips() { + let (store, _dir) = temp_store(); + let id = pk("proj-1"); + SnapshotStore::, GlobalSeq>::commit(&store, &id, SV1, gs(4), &Vec::new()) + .await + .unwrap(); + + let (pos, state) = SnapshotStore::, GlobalSeq>::hydrate(&store, &id, SV1) + .await + .unwrap() + .into_found() + .unwrap(); + assert_eq!(pos, gs(4)); + assert!(state.is_empty()); +} + +// ── 4. Isolation Tests ───────────────────────────────────────────── + +#[tokio::test] +async fn different_projections_are_independent() { + let (store, _dir) = temp_store(); + let id1 = pk("proj-1"); + let id2 = pk("proj-2"); + + SnapshotStore::, GlobalSeq>::commit(&store, &id1, SV1, gs(5), &vec![1]) + .await + .unwrap(); + SnapshotStore::, GlobalSeq>::commit(&store, &id2, SV1, gs(10), &vec![2]) + .await + .unwrap(); + + let (pos1, state1) = SnapshotStore::, GlobalSeq>::hydrate(&store, &id1, SV1) + .await + .unwrap() + .into_found() + .unwrap(); + let (pos2, state2) = SnapshotStore::, GlobalSeq>::hydrate(&store, &id2, SV1) + .await + .unwrap() + .into_found() + .unwrap(); + + assert_eq!((pos1, state1), (gs(5), vec![1])); + assert_eq!((pos2, state2), (gs(10), vec![2])); +} + +/// The reason the `projections` partition is separate from `snapshots`: an +/// aggregate snapshot and a projection checkpoint stored under the **same id +/// bytes** must not clobber each other. This is the one test that defends the +/// partition split — it FAILS if the two `SnapshotStore` impls ever share a +/// keyspace (a merged partition would let the second `commit` overwrite the +/// first, so one `hydrate` would decode the other's value). Requires both +/// features, hence the `snapshot` gate on top of the file's `projection` gate. +#[cfg(feature = "snapshot")] +#[tokio::test] +async fn snapshot_and_projection_with_same_id_do_not_collide() { + let (store, _dir) = temp_store(); + let id = pk("shared-id"); + + // Same id bytes, two distinct stores, two distinct position types. + SnapshotStore::, Version>::commit( + &store, + &id, + SV1, + Version::new(5).unwrap(), + &vec![1, 2, 3], + ) + .await + .unwrap(); + SnapshotStore::, GlobalSeq>::commit(&store, &id, SV1, gs(9), &vec![9, 9, 9]) + .await + .unwrap(); + + // Each hydrates its own value untouched — proof the keyspaces are disjoint. + let (v, snap) = SnapshotStore::, Version>::hydrate(&store, &id, SV1) + .await + .unwrap() + .into_found() + .unwrap(); + let (g, proj) = SnapshotStore::, GlobalSeq>::hydrate(&store, &id, SV1) + .await + .unwrap() + .into_found() + .unwrap(); + + assert_eq!((v, snap), (Version::new(5).unwrap(), vec![1, 2, 3])); + assert_eq!((g, proj), (gs(9), vec![9, 9, 9])); +} diff --git a/crates/nexus-fjall/tests/snapshot_tests.rs b/crates/nexus-fjall/tests/snapshot_tests.rs index da21b3f6..6bfab97a 100644 --- a/crates/nexus-fjall/tests/snapshot_tests.rs +++ b/crates/nexus-fjall/tests/snapshot_tests.rs @@ -15,7 +15,7 @@ use std::num::NonZeroU32; use nexus::Version; use nexus_fjall::FjallStore; use nexus_store::StreamKey; -use nexus_store::state::SnapshotStore; +use nexus_store::state::{Hydrated, SnapshotStore}; const SV1: NonZeroU32 = NonZeroU32::MIN; @@ -60,7 +60,11 @@ async fn commit_then_hydrate_roundtrips() { .await .unwrap(); - let (version, state) = store.hydrate(&id, SV1).await.unwrap().unwrap(); + let (version, state) = SnapshotStore::, Version>::hydrate(&store, &id, SV1) + .await + .unwrap() + .into_found() + .unwrap(); assert_eq!(version, Version::new(5).unwrap()); assert_eq!(state, vec![1, 2, 3]); } @@ -82,12 +86,22 @@ async fn commit_overwrites_previous_snapshot() { .await .unwrap(); - let (version, state) = store.hydrate(&id, sv2).await.unwrap().unwrap(); + let (version, state) = SnapshotStore::, Version>::hydrate(&store, &id, sv2) + .await + .unwrap() + .into_found() + .unwrap(); assert_eq!(version, Version::new(10).unwrap()); assert_eq!(state, vec![2, 3]); // Old schema version should return None (filtered at store level). - assert!(store.hydrate(&id, SV1).await.unwrap().is_none()); + assert!( + SnapshotStore::, Version>::hydrate(&store, &id, SV1) + .await + .unwrap() + .into_found() + .is_none() + ); } // ── 2. Lifecycle Tests ───────────────────────────────────────────── @@ -111,7 +125,11 @@ async fn snapshot_persists_across_reopen() { // Second session: reopen + verify snapshot { let store = FjallStore::builder(&db_path).open().unwrap(); - let (version, state) = store.hydrate(&id, SV1).await.unwrap().unwrap(); + let (version, state) = SnapshotStore::, Version>::hydrate(&store, &id, SV1) + .await + .unwrap() + .into_found() + .unwrap(); assert_eq!(version, Version::new(5).unwrap()); assert_eq!(state, vec![42, 43, 44]); } @@ -122,8 +140,10 @@ async fn snapshot_persists_across_reopen() { #[tokio::test] async fn hydrate_unknown_id_returns_none() { let (store, _dir) = temp_store(); - let result = store.hydrate(&sk("nope"), SV1).await.unwrap(); - assert!(result.is_none()); + let result = SnapshotStore::, Version>::hydrate(&store, &sk("nope"), SV1) + .await + .unwrap(); + assert_eq!(result, Hydrated::Absent); } #[tokio::test] @@ -132,8 +152,10 @@ async fn hydrate_id_without_snapshot_returns_none() { let id = sk("agg-1"); setup_stream(&store, &id, 3).await; - let result = store.hydrate(&id, SV1).await.unwrap(); - assert!(result.is_none()); + let result = SnapshotStore::, Version>::hydrate(&store, &id, SV1) + .await + .unwrap(); + assert_eq!(result, Hydrated::Absent); } #[tokio::test] @@ -146,7 +168,11 @@ async fn commit_without_event_stream_is_persisted() { .unwrap(); // And hydrate reads it back regardless of stream existence. - let (version, state) = store.hydrate(&sk("nope"), SV1).await.unwrap().unwrap(); + let (version, state) = SnapshotStore::, Version>::hydrate(&store, &sk("nope"), SV1) + .await + .unwrap() + .into_found() + .unwrap(); assert_eq!(version, Version::new(1).unwrap()); assert_eq!(state, vec![1]); } @@ -170,8 +196,16 @@ async fn different_streams_have_separate_snapshots() { .await .unwrap(); - let (version1, state1) = store.hydrate(&id1, SV1).await.unwrap().unwrap(); - let (version2, state2) = store.hydrate(&id2, SV1).await.unwrap().unwrap(); + let (version1, state1) = SnapshotStore::, Version>::hydrate(&store, &id1, SV1) + .await + .unwrap() + .into_found() + .unwrap(); + let (version2, state2) = SnapshotStore::, Version>::hydrate(&store, &id2, SV1) + .await + .unwrap() + .into_found() + .unwrap(); assert_eq!(version1, Version::new(5).unwrap()); assert_eq!(state1, vec![1]); diff --git a/crates/nexus-store/src/lib.rs b/crates/nexus-store/src/lib.rs index b8393aa0..b5927c0a 100644 --- a/crates/nexus-store/src/lib.rs +++ b/crates/nexus-store/src/lib.rs @@ -173,8 +173,8 @@ pub use snapshot::Snapshotting; #[cfg(feature = "testing")] pub use state::InMemorySnapshotStore; pub use state::{ - AfterEventTypes, CodecSnapshotStore, CodecSnapshotStoreError, EveryNEvents, PersistTrigger, - SnapshotStore, + AfterEventTypes, CodecSnapshotStore, CodecSnapshotStoreError, EveryNEvents, Hydrated, + PersistTrigger, SnapshotStore, }; pub use step::Step; pub use store::{AllPosition, RawEventStore, Store}; diff --git a/crates/nexus-store/src/projection.rs b/crates/nexus-store/src/projection.rs index 59dd0d81..f5dab987 100644 --- a/crates/nexus-store/src/projection.rs +++ b/crates/nexus-store/src/projection.rs @@ -4,7 +4,7 @@ use core::num::NonZeroU32; use nexus::{DomainEvent, Id, Version}; use crate::decoded::Decoded; -use crate::state::{PersistTrigger, SnapshotStore}; +use crate::state::{Hydrated, PersistTrigger, SnapshotStore}; /// A pure fold function over domain events. /// @@ -105,6 +105,11 @@ pub struct Projection { checkpoint: Option, /// Folded-but-not-yet-persisted tail position, flushed on shutdown. pending: Option, + /// `Some(old_schema)` iff `load` discarded a snapshot under a different + /// schema version — the projection is re-folding from scratch. Surfaced via + /// [`rebuilding_from`](Projection::rebuilding_from) so a host can distinguish + /// a costly schema-bump rebuild from an ordinary fresh start. + rebuilt_from: Option, } impl Projection @@ -117,12 +122,13 @@ where /// Assemble and hydrate the stepper, returning it alongside the starting /// state. /// - /// Resolves `(state, checkpoint)` from the snapshot store atomically. If a - /// snapshot exists for `id` at `schema_version`, its state and position are - /// restored; otherwise the state is [`Projector::initial`] and the - /// checkpoint is `None` (subscribe from the beginning). A snapshot saved - /// under a *different* schema version is invisible — a schema bump forces a - /// full replay from `initial()`. + /// Resolves `(state, checkpoint)` from the snapshot store atomically: + /// - [`Hydrated::Found`] → restore its state and position (resume). + /// - [`Hydrated::Absent`] → [`Projector::initial`], no checkpoint (fresh). + /// - [`Hydrated::Stale`] → `initial()`, no checkpoint, and + /// [`rebuilding_from`](Self::rebuilding_from) reports the discarded schema + /// version — a bump invalidated the saved state, so the projection re-folds + /// the whole stream. The host sees that instead of a silent full replay. /// /// # Errors /// @@ -134,9 +140,13 @@ where snapshot_store: SS, schema_version: NonZeroU32, ) -> Result<(Self, P::State), SS::Error> { - let (state, checkpoint) = match snapshot_store.hydrate(&id, schema_version).await? { - Some((position, state)) => (state, Some(position)), - None => (projector.initial(), None), + let (state, checkpoint, rebuilt_from) = match snapshot_store + .hydrate(&id, schema_version) + .await? + { + Hydrated::Found { position, state } => (state, Some(position), None), + Hydrated::Absent => (projector.initial(), None, None), + Hydrated::Stale { stored_schema } => (projector.initial(), None, Some(stored_schema)), }; Ok(( Self { @@ -147,11 +157,23 @@ where schema_version, checkpoint, pending: None, + rebuilt_from, }, state, )) } + /// `Some(old_schema)` when [`load`](Self::load) discarded a snapshot written + /// under a different schema version — i.e. this projection is re-folding + /// from the beginning because of a schema bump, not because it is new. + /// `None` means it resumed from a checkpoint or started genuinely fresh + /// (disambiguate those two via [`checkpoint`](Self::checkpoint)). A host on a + /// constrained device can use this to warn/defer/throttle the rebuild. + #[must_use] + pub const fn rebuilding_from(&self) -> Option { + self.rebuilt_from + } + /// The id this projection is bound to — pass to `subscribe`. pub const fn id(&self) -> &I { &self.id @@ -261,7 +283,7 @@ mod tests { use super::{Projection, ProjectionError, Projector}; use crate::decoded::Decoded; - use crate::state::{EveryNEvents, InMemorySnapshotStore, SnapshotStore}; + use crate::state::{EveryNEvents, Hydrated, InMemorySnapshotStore, SnapshotStore}; // ── fixtures ──────────────────────────────────────────────────────────── @@ -379,7 +401,12 @@ mod tests { ); // Persisted together, atomically. - let (pos, st) = ss.hydrate(&id, NonZeroU32::MIN).await.unwrap().unwrap(); + let (pos, st) = ss + .hydrate(&id, NonZeroU32::MIN) + .await + .unwrap() + .into_found() + .unwrap(); assert_eq!(pos, version!(3)); assert_eq!( st, @@ -509,11 +536,19 @@ mod tests { .await .unwrap(); assert_eq!(p.checkpoint(), None, "trigger must not have fired"); - assert!(ss.hydrate(&id, NonZeroU32::MIN).await.unwrap().is_none()); + assert_eq!( + ss.hydrate(&id, NonZeroU32::MIN).await.unwrap(), + Hydrated::Absent + ); p.flush(&state).await.unwrap(); assert_eq!(p.checkpoint(), Some(version!(2))); - let (pos, st) = ss.hydrate(&id, NonZeroU32::MIN).await.unwrap().unwrap(); + let (pos, st) = ss + .hydrate(&id, NonZeroU32::MIN) + .await + .unwrap() + .into_found() + .unwrap(); assert_eq!(pos, version!(2)); assert_eq!( st, @@ -541,13 +576,16 @@ mod tests { // Never advanced: flush must not write a spurious snapshot. p.flush(&state).await.unwrap(); assert_eq!(p.checkpoint(), None); - assert!(ss.hydrate(&id, NonZeroU32::MIN).await.unwrap().is_none()); + assert_eq!( + ss.hydrate(&id, NonZeroU32::MIN).await.unwrap(), + Hydrated::Absent + ); } - // ── defensive: a schema-mismatched snapshot is invisible → start fresh ─── + // ── defensive: a schema-mismatched snapshot starts fresh, but flags a rebuild ─ #[tokio::test] - async fn load_ignores_stale_schema_and_starts_from_initial() { + async fn load_stale_schema_starts_from_initial_and_signals_rebuild() { let ss = store(); let id = TestId("s"); // Pre-commit a v1 snapshot. @@ -563,7 +601,7 @@ mod tests { .await .unwrap(); - // Load with schema v2 — the v1 snapshot must be invisible. + // Load with schema v2 — the v1 snapshot must not be restored... let (p, state) = Projection::load( id, CountingProjector, @@ -575,5 +613,25 @@ mod tests { .unwrap(); assert_eq!(p.checkpoint(), None); assert_eq!(state, CountState { count: 0, total: 0 }); + // ...but the rebuild is *visible*: a host can tell this apart from a + // fresh start (which reports `None`) — the whole point of the Stale + // signal. The invalidated schema version (v1) is reported. + assert_eq!(p.rebuilding_from(), Some(NonZeroU32::MIN)); + } + + #[tokio::test] + async fn load_fresh_does_not_signal_rebuild() { + let ss = store(); + let (p, _state) = Projection::load( + TestId("s"), + CountingProjector, + EveryNEvents(NonZeroU64::MIN), + &ss, + NonZeroU32::MIN, + ) + .await + .unwrap(); + // A genuinely-new projection is not a rebuild. + assert_eq!(p.rebuilding_from(), None); } } diff --git a/crates/nexus-store/src/snapshot.rs b/crates/nexus-store/src/snapshot.rs index f698ba7e..8e890e71 100644 --- a/crates/nexus-store/src/snapshot.rs +++ b/crates/nexus-store/src/snapshot.rs @@ -129,11 +129,14 @@ where A: Aggregate, SS: state::SnapshotStore, { + // Aggregate snapshots treat absent and stale identically — replay the + // stream either way — so `into_found` collapses both to `None`. let (version, typed_state) = self .snapshot_store .hydrate(id, self.schema_version) .await - .ok()??; + .ok()? + .into_found()?; let root = AggregateRoot::::restore(id.clone(), typed_state, version); let next = version.next()?; Some((root, next)) diff --git a/crates/nexus-store/src/state.rs b/crates/nexus-store/src/state.rs index 542ada7a..47fbc42d 100644 --- a/crates/nexus-store/src/state.rs +++ b/crates/nexus-store/src/state.rs @@ -9,6 +9,57 @@ use crate::codec::{Decode, Encode, OwningCodec}; // SnapshotStore — atomic state + position persistence // ═══════════════════════════════════════════════════════════════════════════ +/// Outcome of [`SnapshotStore::hydrate`] — a three-state answer that keeps +/// "nothing saved" distinct from "saved, but stale". +/// +/// The distinction is invisible to an aggregate snapshot (both mean "replay the +/// stream"), but load-bearing for a projection: [`Absent`](Self::Absent) is a +/// brand-new projection expected to start empty, whereas [`Stale`](Self::Stale) +/// means an existing projection was invalidated by a schema bump and the very +/// next thing that happens is a **full re-fold of the whole `$all` stream**. On +/// a mobile/`IoT` host that re-fold can be a long, battery-heavy operation, so +/// the host must be able to see it coming (warn, defer to Wi-Fi/charging, +/// throttle) — collapsing both into `None` would hide it. +/// +/// There is deliberately no `stored_state` on `Stale`: derived state has no +/// upcasting path (a schema change forces a rebuild, it cannot be migrated), so +/// carrying the old bytes would only invite a migration that does not exist. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Hydrated { + /// Nothing has ever been saved for this id — start from scratch. + Absent, + /// A snapshot exists but under a *different* schema version; it cannot be + /// decoded into the requested shape, so the caller must rebuild from the + /// log. Carries the schema version that was found, for observability. + Stale { + /// The schema version the stored snapshot was written under. + stored_schema: NonZeroU32, + }, + /// A snapshot at the requested schema version. + Found { + /// The position the state was folded up to. + position: P, + /// The restored state. + state: S, + }, +} + +impl Hydrated { + /// The restored `(position, state)` when a snapshot at the requested schema + /// version was found; `None` for `Absent` or `Stale` (both mean "rebuild"). + /// + /// A convenience for callers that treat absent and stale identically (e.g. + /// the aggregate-snapshot decorator, which replays the stream either way). + /// Callers that must tell the two apart — projection hosts — match the enum. + #[must_use] + pub fn into_found(self) -> Option<(P, S)> { + match self { + Self::Found { position, state } => Some((position, state)), + Self::Absent | Self::Stale { .. } => None, + } + } +} + /// Atomic persistence of a snapshot — derived state plus the position it /// was folded up to. /// @@ -32,8 +83,11 @@ pub trait SnapshotStore: Send + Sync { /// Load the saved state and position from a single consistent snapshot. /// - /// Returns `None` if nothing has been saved for `id`, or if the saved - /// state's schema version does not match `schema_version`. + /// Returns [`Hydrated::Found`] with the state and position when a snapshot + /// at `schema_version` exists; [`Hydrated::Stale`] when one exists under a + /// different schema version (caller must rebuild); [`Hydrated::Absent`] when + /// nothing has been saved. `Stale` and `Absent` are kept distinct so a + /// projection host can tell a fresh start from a schema-bump rebuild. /// /// # Errors /// @@ -42,7 +96,7 @@ pub trait SnapshotStore: Send + Sync { &self, id: &impl Id, schema_version: NonZeroU32, - ) -> impl Future, Self::Error>> + Send; + ) -> impl Future, Self::Error>> + Send; /// Save state and position together, in a single transaction. /// @@ -76,7 +130,7 @@ where &self, id: &impl Id, schema_version: NonZeroU32, - ) -> impl Future, Self::Error>> + Send { + ) -> impl Future, Self::Error>> + Send { (**self).hydrate(id, schema_version) } @@ -196,14 +250,19 @@ where &self, id: &impl Id, schema_version: NonZeroU32, - ) -> Result, Self::Error> { - let Some((position, bytes)) = self + ) -> Result, Self::Error> { + // Absent/Stale pass through untouched — only `Found` carries bytes to + // decode. The position `P` and the schema-version signal are opaque to + // the bridge. + let (position, bytes) = match self .store .hydrate(id, schema_version) .await .map_err(CodecSnapshotStoreError::Store)? - else { - return Ok(None); + { + Hydrated::Absent => return Ok(Hydrated::Absent), + Hydrated::Stale { stored_schema } => return Ok(Hydrated::Stale { stored_schema }), + Hydrated::Found { position, state } => (position, state), }; let label = id.to_label(); @@ -216,7 +275,7 @@ where let state = >::decode(&self.codec, &env).map_err(CodecSnapshotStoreError::Decode)?; - Ok(Some((position, state))) + Ok(Hydrated::Found { position, state }) } async fn commit( @@ -274,7 +333,7 @@ mod testing { use nexus::Id; use tokio::sync::RwLock; - use super::SnapshotStore; + use super::{Hydrated, SnapshotStore}; /// In-memory snapshot store for tests. #[derive(Debug, Default)] @@ -302,12 +361,20 @@ mod testing { &self, id: &impl Id, schema_version: NonZeroU32, - ) -> Result, Infallible> { + ) -> Result, Infallible> { let snapshots = self.snapshots.read().await; - Ok(snapshots - .get(&id.to_string()) - .filter(|(stored_schema, _, _)| *stored_schema == schema_version) - .map(|(_, position, state)| (position.clone(), state.clone()))) + Ok(match snapshots.get(&id.to_string()) { + None => Hydrated::Absent, + Some((stored_schema, _, _)) if *stored_schema != schema_version => { + Hydrated::Stale { + stored_schema: *stored_schema, + } + } + Some((_, position, state)) => Hydrated::Found { + position: position.clone(), + state: state.clone(), + }, + }) } async fn commit( diff --git a/crates/nexus-store/tests/projection_tests.rs b/crates/nexus-store/tests/projection_tests.rs index e0cb8a25..c542ceb5 100644 --- a/crates/nexus-store/tests/projection_tests.rs +++ b/crates/nexus-store/tests/projection_tests.rs @@ -18,7 +18,7 @@ use std::num::{NonZeroU32, NonZeroU64}; use nexus::Version; use nexus_store::Projector; -use nexus_store::state::{AfterEventTypes, EveryNEvents, PersistTrigger, SnapshotStore}; +use nexus_store::state::{AfterEventTypes, EveryNEvents, Hydrated, PersistTrigger, SnapshotStore}; const SV1: NonZeroU32 = NonZeroU32::MIN; @@ -130,7 +130,7 @@ mod in_memory_tests { async fn hydrate_returns_none_when_empty() { let store = InMemorySnapshotStore::, Version>::new(); let result = store.hydrate(&TestId("proj-1".into()), SV1).await.unwrap(); - assert!(result.is_none()); + assert_eq!(result, Hydrated::Absent); } #[tokio::test] @@ -143,7 +143,8 @@ mod in_memory_tests { .commit(&id, NonZeroU32::new(1).unwrap(), version, &vec![1, 2, 3]) .await .unwrap(); - let (loaded_version, loaded_state) = store.hydrate(&id, SV1).await.unwrap().unwrap(); + let (loaded_version, loaded_state) = + store.hydrate(&id, SV1).await.unwrap().into_found().unwrap(); assert_eq!(loaded_version, version); assert_eq!(loaded_state, vec![1, 2, 3]); @@ -174,7 +175,8 @@ mod in_memory_tests { .await .unwrap(); - let (loaded_version, loaded_state) = store.hydrate(&id, SV1).await.unwrap().unwrap(); + let (loaded_version, loaded_state) = + store.hydrate(&id, SV1).await.unwrap().into_found().unwrap(); assert_eq!(loaded_version, Version::new(20).unwrap()); assert_eq!(loaded_state, vec![2]); } @@ -207,11 +209,13 @@ mod in_memory_tests { .hydrate(&TestId("proj-1".into()), SV1) .await .unwrap() + .into_found() .unwrap(); let (version2, _) = store .hydrate(&TestId("proj-2".into()), SV1) .await .unwrap() + .into_found() .unwrap(); assert_eq!(version1, Version::new(5).unwrap()); assert_eq!(version2, Version::new(10).unwrap()); @@ -232,19 +236,25 @@ mod in_memory_tests { .await .unwrap(); - // Matching schema version returns state + // Matching schema version returns the found state. let loaded = store .hydrate(&id, NonZeroU32::new(1).unwrap()) .await .unwrap(); - assert!(loaded.is_some()); + assert!(matches!(loaded, Hydrated::Found { .. })); - // Mismatched schema version returns None + // Mismatched schema version reports Stale (not Absent), carrying the + // stored schema version so a host can tell a rebuild from a fresh start. let loaded = store .hydrate(&id, NonZeroU32::new(2).unwrap()) .await .unwrap(); - assert!(loaded.is_none()); + assert_eq!( + loaded, + Hydrated::Stale { + stored_schema: NonZeroU32::new(1).unwrap() + } + ); } } diff --git a/crates/nexus-store/tests/snapshot_integration_tests.rs b/crates/nexus-store/tests/snapshot_integration_tests.rs index 8873ea07..49dc08fa 100644 --- a/crates/nexus-store/tests/snapshot_integration_tests.rs +++ b/crates/nexus-store/tests/snapshot_integration_tests.rs @@ -27,7 +27,7 @@ fn sv2() -> NonZeroU32 { use nexus::*; use nexus_store::Store; use nexus_store::state::{ - AfterEventTypes, EveryNEvents, InMemorySnapshotStore, PersistTrigger, SnapshotStore, + AfterEventTypes, EveryNEvents, Hydrated, InMemorySnapshotStore, PersistTrigger, SnapshotStore, }; use nexus_store::testing::InMemoryStore; use nexus_store::{Repository, Snapshotting}; @@ -303,9 +303,12 @@ async fn lazy_snapshot_on_read_after_full_replay() { assert_eq!(loaded.state().value, 3); // Verify snapshot was created by hydrating directly - let snap = snap_store.hydrate(&id, SV1).await.unwrap(); - assert!(snap.is_some()); - let (snap_version, _) = snap.unwrap(); + let (snap_version, _) = snap_store + .hydrate(&id, SV1) + .await + .unwrap() + .into_found() + .expect("snapshot should exist"); assert_eq!(snap_version, Version::new(3).unwrap()); } @@ -480,9 +483,21 @@ async fn sequence_snapshot_invalidation_then_new_snapshot() { assert_eq!(loaded.version(), Some(Version::new(2).unwrap())); // Verify the snapshot has schema v2: hydrating at v2 hits, at v1 misses. - let (snap_version, _) = snap_store.hydrate(&id, sv2()).await.unwrap().unwrap(); + let (snap_version, _) = snap_store + .hydrate(&id, sv2()) + .await + .unwrap() + .into_found() + .unwrap(); assert_eq!(snap_version, Version::new(2).unwrap()); - assert!(snap_store.hydrate(&id, SV1).await.unwrap().is_none()); + assert!( + snap_store + .hydrate(&id, SV1) + .await + .unwrap() + .into_found() + .is_none() + ); } // ═══════════════════════════════════════════════════════════════════════════ @@ -545,7 +560,14 @@ async fn lifecycle_lazy_snapshot_then_subsequent_load_uses_it() { .unwrap(); // No snapshot yet - assert!(snap_store.hydrate(&id, SV1).await.unwrap().is_none()); + assert!( + snap_store + .hydrate(&id, SV1) + .await + .unwrap() + .into_found() + .is_none() + ); // Load with on-read → creates lazy snapshot let inner2 = store.repository().build(); @@ -559,7 +581,12 @@ async fn lifecycle_lazy_snapshot_then_subsequent_load_uses_it() { let _loaded: AggregateRoot = repo_on_read.load(id.clone()).await.unwrap(); // Snapshot now exists - let (snap_version, _) = snap_store.hydrate(&id, SV1).await.unwrap().unwrap(); + let (snap_version, _) = snap_store + .hydrate(&id, SV1) + .await + .unwrap() + .into_found() + .unwrap(); assert_eq!(snap_version, Version::new(5).unwrap()); // Second load uses snapshot (we can't directly prove partial replay, @@ -646,7 +673,7 @@ async fn defensive_snapshot_store_load_error_falls_back_to_full_replay() { &self, _id: &impl nexus::Id, _schema_version: NonZeroU32, - ) -> Result, Self::Error> { + ) -> Result, Self::Error> { Err(std::io::Error::new( std::io::ErrorKind::Other, "disk on fire", @@ -708,8 +735,8 @@ async fn defensive_snapshot_save_failure_does_not_fail_event_save() { &self, _id: &impl nexus::Id, _schema_version: NonZeroU32, - ) -> Result, Self::Error> { - Ok(None) + ) -> Result, Self::Error> { + Ok(Hydrated::Absent) } async fn commit( &self, diff --git a/crates/nexus-store/tests/snapshot_tests.rs b/crates/nexus-store/tests/snapshot_tests.rs index 92031336..a2d3588b 100644 --- a/crates/nexus-store/tests/snapshot_tests.rs +++ b/crates/nexus-store/tests/snapshot_tests.rs @@ -21,7 +21,7 @@ use std::num::{NonZeroU32, NonZeroU64}; use nexus::Version; const SV1: NonZeroU32 = NonZeroU32::MIN; -use nexus_store::state::{AfterEventTypes, EveryNEvents, PersistTrigger, SnapshotStore}; +use nexus_store::state::{AfterEventTypes, EveryNEvents, Hydrated, PersistTrigger, SnapshotStore}; #[derive(Debug, Clone, Hash, PartialEq, Eq)] struct TestId(String); @@ -136,7 +136,7 @@ mod in_memory_tests { async fn hydrate_returns_none_when_empty() { let store = InMemorySnapshotStore::, Version>::new(); let result = store.hydrate(&TestId("agg-1".into()), SV1).await.unwrap(); - assert!(result.is_none()); + assert_eq!(result, Hydrated::Absent); } #[tokio::test] @@ -149,7 +149,8 @@ mod in_memory_tests { .commit(&id, NonZeroU32::new(1).unwrap(), version, &vec![1, 2, 3]) .await .unwrap(); - let (loaded_version, loaded_state) = store.hydrate(&id, SV1).await.unwrap().unwrap(); + let (loaded_version, loaded_state) = + store.hydrate(&id, SV1).await.unwrap().into_found().unwrap(); assert_eq!(loaded_version, version); assert_eq!(loaded_state, vec![1, 2, 3]); @@ -180,7 +181,8 @@ mod in_memory_tests { .await .unwrap(); - let (loaded_version, loaded_state) = store.hydrate(&id, SV1).await.unwrap().unwrap(); + let (loaded_version, loaded_state) = + store.hydrate(&id, SV1).await.unwrap().into_found().unwrap(); assert_eq!(loaded_version, Version::new(20).unwrap()); assert_eq!(loaded_state, vec![2]); } @@ -213,11 +215,13 @@ mod in_memory_tests { .hydrate(&TestId("agg-1".into()), SV1) .await .unwrap() + .into_found() .unwrap(); let (version2, _) = store .hydrate(&TestId("agg-2".into()), SV1) .await .unwrap() + .into_found() .unwrap(); assert_eq!(version1, Version::new(5).unwrap()); assert_eq!(version2, Version::new(10).unwrap()); diff --git a/examples/projection-tokio/tests/loop_tests.rs b/examples/projection-tokio/tests/loop_tests.rs index 675f7770..a29a0e6b 100644 --- a/examples/projection-tokio/tests/loop_tests.rs +++ b/examples/projection-tokio/tests/loop_tests.rs @@ -294,6 +294,7 @@ async fn runner_processes_events_and_checkpoints() { .hydrate(&stream_id, NonZeroU32::MIN) .await .unwrap() + .into_found() .unwrap(); assert_eq!(position, version!(3)); assert_eq!( @@ -367,6 +368,7 @@ async fn runner_resumes_from_checkpoint() { .hydrate(&stream_id, NonZeroU32::MIN) .await .unwrap() + .into_found() .unwrap(); assert_eq!(position, version!(5)); assert_eq!( @@ -420,6 +422,7 @@ async fn runner_trigger_controls_checkpoint_frequency() { .hydrate(&stream_id, NonZeroU32::MIN) .await .unwrap() + .into_found() .unwrap(); assert_eq!(position, version!(5)); assert_eq!( @@ -499,6 +502,7 @@ async fn runner_resumes_normally_after_rebuild_completes() { .hydrate(&stream_id, NonZeroU32::new(2).unwrap()) .await .unwrap() + .into_found() .unwrap(); assert_eq!( state, @@ -545,7 +549,7 @@ async fn runner_immediate_shutdown_with_no_events() { .hydrate(&stream_id, NonZeroU32::MIN) .await .unwrap(); - assert!(loaded.is_none()); + assert!(loaded.into_found().is_none()); } #[tokio::test] @@ -586,6 +590,7 @@ async fn runner_rebuild_is_idempotent_after_crash_before_trigger() { .hydrate(&stream_id, NonZeroU32::MIN) .await .unwrap() + .into_found() .unwrap(); assert_eq!(position, version!(3)); @@ -634,6 +639,7 @@ async fn runner_rebuild_is_idempotent_after_crash_before_trigger() { .hydrate(&stream_id, NonZeroU32::new(2).unwrap()) .await .unwrap() + .into_found() .unwrap(); assert_eq!( state, @@ -681,6 +687,7 @@ async fn runner_graceful_shutdown_flushes_dirty_state() { .hydrate(&stream_id, NonZeroU32::MIN) .await .unwrap() + .into_found() .unwrap(); assert_eq!( state, @@ -741,6 +748,7 @@ async fn runner_stale_state_falls_back_to_initial() { .hydrate(&stream_id, NonZeroU32::new(2).unwrap()) .await .unwrap() + .into_found() .unwrap(); assert_eq!( state, @@ -776,7 +784,10 @@ async fn runner_first_run_with_state_persistence_is_not_rebuild() { .hydrate(&stream_id, NonZeroU32::MIN) .await .unwrap(); - assert!(before.is_none(), "expected no snapshot before first run"); + assert!( + before.into_found().is_none(), + "expected no snapshot before first run" + ); run( stream_id.clone(), @@ -797,6 +808,7 @@ async fn runner_first_run_with_state_persistence_is_not_rebuild() { .hydrate(&stream_id, NonZeroU32::MIN) .await .unwrap() + .into_found() .unwrap(); assert_eq!( state, @@ -937,6 +949,7 @@ async fn runner_catches_up_and_processes_all_existing_events() { .hydrate(&stream_id, NonZeroU32::MIN) .await .unwrap() + .into_found() .unwrap(); assert_eq!(position, version!(5)); assert_eq!( @@ -988,6 +1001,7 @@ async fn runner_resumes_from_checkpoint_on_second_run() { .hydrate(&stream_id, NonZeroU32::MIN) .await .unwrap() + .into_found() .unwrap(); assert_eq!(cp1, version!(1)); assert_eq!( @@ -1021,6 +1035,7 @@ async fn runner_resumes_from_checkpoint_on_second_run() { .hydrate(&stream_id, NonZeroU32::MIN) .await .unwrap() + .into_found() .unwrap(); assert_eq!(cp2, version!(2)); // count:2, total:15 — not count:1/total:5 (fresh) or count:2/total:20 (double-fold) @@ -1064,7 +1079,7 @@ async fn schema_bump_resolves_to_fresh() { .hydrate(&stream_id, NonZeroU32::MIN) .await .unwrap(); - assert!(v1.is_some(), "expected v1 snapshot to exist"); + assert!(v1.into_found().is_some(), "expected v1 snapshot to exist"); // Second run with schema v2 — the schema-mismatched snapshot is invisible. // hydrate returns None → state starts from initial() → full replay. @@ -1088,6 +1103,7 @@ async fn schema_bump_resolves_to_fresh() { .hydrate(&stream_id, NonZeroU32::new(2).unwrap()) .await .unwrap() + .into_found() .unwrap(); assert_eq!(pos2, version!(1)); assert_eq!(