feat(store): unified state persistence and projection rebuild (#158)#163
Merged
Conversation
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
…158) Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Split run() into initialize() -> PreparedProjection -> run(shutdown). Three typestates (Resuming, Rebuilding, Starting) enforce supervisor control points at compile time. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
7 tasks: scaffolding, resolve_startup refactor, event loop move, initialize() impl, typestate methods, integration tests, verification. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
- #[expect(dead_code)] on PreparedProjection (fields used by run() in Task 3) - #[must_use] on Initialized enum to catch silently dropped projections Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Extract inline startup logic from run() into a standalone resolve_startup function that returns StartupOutcome<S> instead of a (S, Option<Version>) tuple. The three variants (Resuming, Rebuilding, Starting) map directly to the Initialized enum variants that initialize() will construct. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Extract the event loop into PreparedProjection::run() generic over Mode, and re-extract apply_event as a pure sync function for testability. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
…d enum ProjectionRunner now exposes initialize() instead of run(). The method loads checkpoint + state, resolves the startup decision, and returns an Initialized enum (Starting | Resuming | Rebuilding) wrapping a PreparedProjection. Supervisors match on the variant for mode-specific behavior, or call the convenience Initialized::run() to delegate directly. All 24 integration tests updated to the new two-phase API. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Add mode-specific accessor methods to PreparedProjection for each typestate (Resuming, Rebuilding, Starting) and the force_rebuild transition from Resuming to Rebuilding. All accessors are const fn with #[must_use]. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Update nexus-framework architecture section to document the new phased startup: initialize() -> Initialized -> run(shutdown) with three typestate markers (Resuming, Rebuilding, Starting). Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Model the projection event loop as an explicit Mealy machine with three write-centric states (Idle, Pending, Committed) instead of scattered mut variables and opaque bools. Pure sync transition function in status.rs, async shell in prepared.rs just interprets status changes. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
4-task plan: create status.rs with enum + tests, refactor event loop shell, migrate old tests, update CLAUDE.md. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Introduce status.rs with a three-state FSM (Idle, Pending, Committed) that replaces the scattered mut variables and (State, bool) tuple in the projection event loop. Pure sync apply_event transition function — no IO. 12 unit tests covering all 6 state transitions, multi-step sequences, error propagation, and checkpoint correctness. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Replace 4 mut variables (state, last_persisted_version, current_version, dirty) with a single mut status: ProjectionStatus<S>. The async shell in run() now pattern-matches on Committed (persist) and Pending (flush on shutdown) — no bools, no version tracking drift. Remove old apply_event from prepared.rs (now in status.rs). Trim unused test fixtures; keep typestate tests with minimal SumProjector fixture. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
…dy> cleanup Collapse ProjectionRunner, Initialized, PreparedProjection, and three typestate markers into Projection<Mode> with two phases: Configured and Ready. resolve_startup returns ProjectionStatus::Idle directly — no anonymous tuples, no StartupOutcome enum. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
…itialized Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
…te/ module Foundation for unifying the 4 identical byte-level types (PendingState, PersistedState, PendingSnapshot, PersistedSnapshot) into 2 generic types. These will eventually replace the old byte-locked versions. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Generic trait for versioned state persistence — replaces both the byte-level StateStore (projections) and SnapshotStore (aggregates). Includes () no-op impl and &T delegation. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
- PersistTrigger: unified trigger for both projections and snapshots (replaces identical ProjectionTrigger and SnapshotTrigger) - EveryNEvents + AfterEventTypes implementations - InMemoryStateStore<S>: generic in-memory impl for testing Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Wraps a StateStore<Vec<u8>> + Codec<S> to produce StateStore<S>. Enables byte-level backends (fjall) to serve typed state to consumers. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
…ied state - Snapshotting<R, SS, SC, T> → Snapshotting<R, SS, T>: codec removed, state store now returns typed state directly - Builder: WithSnapshot drops SC, snapshot-json wraps in CodecStateStore internally, .snapshot_codec() method removed - Fjall: implements state::StateStore<Vec<u8>> instead of SnapshotStore - All use state::PersistTrigger instead of SnapshotTrigger Test files not yet updated (expected breakage in snapshot_tests, snapshot_integration_tests, fjall/snapshot_tests). Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
- Collapse PendingState/PersistedState into single State<S> — the two types were structurally identical with no ownership asymmetry; the read/write direction is expressed by trait method names (load/save) - Move repository/ from store/repository/ to top-level crate module - Move subscription + checkpoint from store/subscription/ to store/ - Remove snapshot/mod.rs indirection (snapshotting.rs → repository/snapshot.rs) Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This was referenced May 5, 2026
joeldsouzax
added a commit
that referenced
this pull request
May 19, 2026
…ance suite (#165) * docs: refresh CLAUDE.md architecture; clean stale type ref in test Update the architecture section to reflect the reorg landed in PRs #161, #162, and #163: nexus-store now has five module directories (codec, envelope, upcasting, store, repository, state, projection) with a clean responsibility split. Document the new Subscription and CheckpointStore traits, the unified StateStore<S>/State<S> persistence model that replaces the separate snapshot and projection-state stores, and the projection runner's move from nexus-store to nexus-framework. Refresh the fjall adapter section for the new partition.rs and subscription_stream.rs files. Also fix a stale comment in projection_tests.rs referencing the removed NoStatePersistence type. Mandatory rules, conventions, and build commands are unchanged. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * refactor(store): flatten projection/ into projection.rs Move projection/projector.rs to projection.rs and remove the now-empty mod.rs scaffolding. The Projector trait now lives directly in the projection module file, matching the flat layout being adopted across nexus-store. Public path nexus_store::projection::Projector is unchanged. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * refactor(store): flatten upcasting/ into upcasting.rs Merge upcasting/morsel.rs (EventMorsel) and upcasting/upcaster.rs (Upcaster trait + no-op () impl) into a single upcasting.rs at the crate root. The two types form one cohesive concept — the schema-migration pipeline — and splitting them across files served no purpose now that the directory has no other contents. Public paths nexus_store::upcasting::EventMorsel and ::Upcaster unchanged. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * refactor(store): flatten envelope/ into envelope.rs Merge envelope/pending.rs and envelope/persisted.rs into a single envelope.rs at the crate root. Write-path PendingEnvelope and read-path PersistedEnvelope form one conceptual unit — the on-the-wire event representation — and live together now. Public paths nexus_store::envelope::{PendingEnvelope, PersistedEnvelope, pending_envelope, WithVersion, WithEventType, WithPayload} unchanged. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * refactor(store): flatten codec/ into codec.rs Collapse codec/{owning, borrowing}.rs and the codec/serde/ subdirectory into a single codec.rs at the crate root. Codec and BorrowingCodec traits live at the top; the feature-gated serde adapter (SerdeFormat, SerdeCodec) sits in an inline `pub mod serde` block, with Json and JsonCodec in a nested `pub mod json` inside that. Public paths unchanged: - nexus_store::codec::{Codec, BorrowingCodec} - nexus_store::codec::serde::{SerdeCodec, SerdeFormat} - nexus_store::codec::serde::json::{Json, JsonCodec} Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * refactor(store): flatten store/ into store.rs Concatenate store/{store, raw, stream, subscription, checkpoint}.rs into a single store.rs at the crate root. All five primitives (Store wrapper, RawEventStore trait, EventStream + EventStreamExt, Subscription, CheckpointStore) belong together as the "what adapters implement" surface. Update internal callers in repository/{event_store, zero_copy, builder}.rs that previously imported via the now-removed submodule paths (crate::store::raw::, crate::store::store::, crate::store::stream::) to import from crate::store::* directly. Public paths unchanged: - nexus_store::store::{Store, RawEventStore, EventStream, EventStreamExt, Subscription, CheckpointStore} Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * refactor(store): flatten state/ into state.rs Concatenate state/{state, store, trigger, codec_adapter, testing}.rs into a single state.rs at the crate root. All five pieces form one cohesive "versioned state persistence" concept — State payload, StateStore trait, PersistTrigger policy, CodecStateStore bridge, and the InMemoryStateStore testing fake. The testing fake lives in an inline `#[cfg(feature = "testing")] mod testing { ... }` block, matching the codec.rs `pub mod serde` pattern used for feature-gated nested content. Public paths unchanged: - nexus_store::state::{State, StateStore, CodecStateStore, CodecStateStoreError, PersistTrigger, EveryNEvents, AfterEventTypes} - nexus_store::state::InMemoryStateStore (feature = "testing") Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * refactor(store): split repository/ into repository.rs, builder.rs, snapshot.rs Promote three crate-root modules out of the repository/ subdirectory: - repository.rs (~475 lines): Repository<A> trait + EventStore facade + ZeroCopyEventStore facade + pub(crate) ReplayFrom trait. The four files that were repository/{repository, event_store, zero_copy, replay}.rs. - builder.rs (~390 lines): RepositoryBuilder typestate + NeedsCodec marker + NoSnapshot/WithSnapshot snapshot slot markers. Pure construction scaffolding, kept separate from the trait + impls. - snapshot.rs (~150 lines): Snapshotting<R, SS, T> decorator (feature = "snapshot"). Wraps a Repository with transparent state-store load/save and a PersistTrigger policy. Three files committed together because Rust's module resolution forbids the intermediate state where repository.rs at root and repository/mod.rs coexist — the move must be atomic. Public crate-root paths unchanged via lib.rs re-exports: - nexus_store::{Repository, EventStore, ZeroCopyEventStore} - nexus_store::{RepositoryBuilder, NeedsCodec, NoSnapshot} - nexus_store::{Snapshotting, WithSnapshot} (feature = "snapshot") Sub-paths changed (no external consumer was using them): - nexus_store::repository::RepositoryBuilder -> nexus_store::builder::RepositoryBuilder - nexus_store::repository::Snapshotting -> nexus_store::snapshot::Snapshotting Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * docs: update CLAUDE.md for flat nexus-store layout After the per-directory flatten (commits 774dff6 through 9573b93), the nexus-store crate is now flat: one file per concept, no module subdirectories. Replace the seven-directory description with the eleven-file flat layout, noting that the boundary that matters is the crate boundary, not the inside of nexus-store. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * feat(store): typed decoder pipeline for event streams Add a single primitive that pairs an `EventStream` with a codec and (optionally) an upcaster, then folds decoded events through a closure receiving `(state, Version, event)`. Replaces the hand-rolled decode loops duplicated across `EventStore::replay_from`, `ZeroCopyEventStore::replay_from`, and the framework projection runner — existing call sites are not yet migrated. New module `nexus-store/src/stream.rs` holds the full stream surface: `EventStream`/`EventStreamExt` (moved from `store.rs` to keep the concept-split clean) plus the new types: - `DecoderBuilder<S, CodecState, UpcasterState>` — typestate builder forking between owning and borrowing codecs. - `DecodedStream<'a, S, C, U, M>` — owning-codec wrapper. - `BorrowedDecodedStream<'a, S, C, U, M>` — zero-copy wrapper. - `DecodeStreamError<S, C, U>` (in `error.rs`) — convenience error enum spanning stream / codec / upcaster failures. The wrapper types borrow codec + upcaster (per the borrow-first rule); the no-op `()` upcaster is supplied via a `static` so callers never write `&()` themselves. Cross-crate import paths shift from `nexus_store::store::EventStream` to `nexus_store::stream::EventStream`; the top-level re-export remains stable at `nexus_store::EventStream`. 6 new tests in `tests/stream_tests.rs` cover the happy path, version threading, upcaster ordering, empty stream, and closure-error short-circuit. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * test(store): exhaustive stream.rs coverage + reusable adapter conformance suite Bulk: ~75 unit and property tests in tests/stream_tests.rs covering the EventStream contract, EventStreamExt combinators (try_fold/for_each/ collect_map/count), DecoderBuilder typestate, and both DecodedStream and BorrowedDecodedStream pipelines including error-variant routing (Stream/Upcast/Codec) and closure short-circuit semantics. Novel testing techniques added in dedicated sections: - Allocation counter (tests/stream_alloc_tests.rs, separate test binary with a #[global_allocator]) proves BorrowedDecodedStream allocates O(1) bytes regardless of stream length; a Codec<String> negative control proves the harness can detect O(n) growth. - Drop tracking via DropProbe in envelope metadata: each envelope drops exactly once at the closure boundary, even on error short-circuit. - Differential proptest: owning and borrowing decoder arms observe byte-identical sequences for any input. - Failure-injection proptest: for any rows + arbitrary fail_at, fold returns Err iff fail_at <= rows.len() and the closure was called exactly min(fail_at, rows.len()) times. - Model-based proptest: a PerfectModel asserts all four combinators agree on count, payloads, event_types, and versions for any input. - Static Send/Sync assertions via static_assertions on the wrapper types. Four compile-fail fixtures lock in structural invariants: - stream_decoder_build_without_codec.rs: typestate refuses .build() until a codec is bound. - stream_double_next_borrow.rs: GAT lending rejects holding two envelopes from one cursor. - stream_not_send_cannot_impl.rs: EventStream::next()'s `+ Send` makes !Send streams unrepresentable at the trait level — discovered while writing the Send assertions. New crate `nexus-store-testing` publishes `assert_event_stream_conformance`, a reusable 10-check suite (empty / single / N-events-then-fused / strictly monotonic versions / event_type round-trip / schema_version round-trip / payload round-trip / insertion order / 1024-event drain / accessor idempotency). InMemoryStore (in nexus-store/tests/inmemory_conformance.rs) and FjallStore (in nexus-fjall/tests/fjall_conformance.rs) each call it from a 5-line test. Future adapters inherit the full regression net by adding one call. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
5 tasks
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
PendingState/PersistedStateinto a singleState<S>type, addedStateStore<S>trait withCodecStateStoreadapter for byte-level backends,PersistTriggerstrategies (EveryNEvents,AfterEventTypes), andInMemoryStateStorefor testingschema_versionchanges, resets to initial and replays from the beginning — withProjectionStatusFSM driving the event loop and typestateProjection<Configured>→Projection<Ready>APIrepository/to top-level crate module, inlinedsubscription/andcheckpointintostore/, removed redundantsnapshot/andprojection/modules fromnexus-storeTest plan
cargo test --all-features)nix flake checkgreen (clippy, fmt, nextest, hakari, deny, audit, toml-fmt, doc)🤖 Generated with Claude Code