From ac9f34649392bd9cc7ac6ca7142eba1eda6739f4 Mon Sep 17 00:00:00 2001 From: Patrick Lee Scott Date: Tue, 26 May 2026 22:31:49 -0500 Subject: [PATCH 1/3] Implement aggregate snapshot cache storage --- README.md | 12 +- docs/async-repositories.md | 5 +- docs/postgres-event-store.md | 32 ++- migrations/postgres/0001_initial.sql | 14 +- migrations/sqlite/0001_initial.sql | 14 +- src/entity/entity.rs | 31 --- src/hashmap_repo/repository.rs | 9 +- src/lib.rs | 6 +- src/postgres_repo/mod.rs | 84 +++++- src/snapshot/in_memory.rs | 72 +++-- src/snapshot/mod.rs | 4 +- src/snapshot/repository.rs | 263 +++++++++++++++--- src/snapshot/snapshottable.rs | 9 +- src/snapshot/store.rs | 116 +++++++- src/sqlite_repo/mod.rs | 78 +++++- src/sqlx_repo/mod.rs | 8 +- tests/async_repository/main.rs | 150 +++++++++- .../scenario.rs | 50 ++-- tests/postgres_repository/main.rs | 50 ++-- tests/snapshots/main.rs | 5 +- tests/sqlite_repository/main.rs | 37 ++- tests/upcasting/main.rs | 13 +- 22 files changed, 839 insertions(+), 223 deletions(-) diff --git a/README.md b/README.md index 3f51f603..e83943bb 100644 --- a/README.md +++ b/README.md @@ -156,8 +156,8 @@ fn main() -> Result<(), Box> { - **HashMapRepository**: In-memory repository for tests and examples. - **QueuedRepository**: Wraps any repository and adds per-entity queue locking. - **EventUpcaster**: A pure, stateless transformation that converts event payloads from one version to another at read time. -- **Snapshottable**: Opt-in trait for aggregates that support periodic snapshots for fast hydration. Use `#[derive(Snapshot)]` to auto-generate the snapshot struct and trait impl. -- **SnapshotAggregateRepository**: Wraps an `AggregateRepository` to transparently create and load snapshots. +- **Snapshottable**: Opt-in trait for aggregates that produce state snapshot payload DTOs. Use `#[derive(Snapshot)]` to auto-generate the payload struct and trait impl. +- **SnapshotAggregateRepository**: Wraps an `AggregateRepository` to transparently create and load rebuildable snapshot cache records. - **OutboxMessage**: A durable publication work item for a domain event, integration event, command, or generic transport message. Supports optional `destination` for point-to-point routing and metadata propagation. - **Outbox Worker**: Publishes outbox messages to external systems. `spawn` for fan-out, `spawn_routed` for point-to-point routing. - **ReadModel**: Query-optimized projection state for UI/API reads. Read models may be updated atomically with a command or eventually from published messages. @@ -1259,11 +1259,11 @@ See [`docs/read-models.md`](docs/read-models.md) for the full guide, including r ## Snapshots -As aggregates accumulate events, replaying from scratch gets expensive. Snapshots let you periodically capture an aggregate's state and restore from it, replaying only the events that came after. +As aggregates accumulate events, replaying from scratch gets expensive. Distributed keeps aggregate events as the durable source of truth and stores repository snapshots as a rebuildable hydration cache. A snapshot cache record can be deleted and rebuilt from events without changing aggregate correctness. ### Making an Aggregate Snapshottable -Add `#[derive(Snapshot)]` to your aggregate struct. This generates a `TodoSnapshot` struct, a `fn snapshot()` method, and the full `impl Snapshottable` — no boilerplate needed: +Add `#[derive(Snapshot)]` to your aggregate struct. This generates a state snapshot payload DTO such as `TodoSnapshot`, a `fn snapshot()` method, and the full `impl Snapshottable` — no boilerplate needed: ```rust use sourced_rust::{Entity, Snapshot}; @@ -1356,8 +1356,8 @@ let Some(todo) = repo.get("todo-1")? else { ### How It Works - **On commit**: If `entity.version() >= snapshot_version + frequency`, the aggregate's state is serialized via `create_snapshot()` and saved to the snapshot store. -- **On load**: If a snapshot exists, the aggregate is restored from it and only events with `sequence > snapshot.version` are replayed. If no snapshot exists, full replay is used as a fallback. -- **Storage**: Snapshots are stored separately from the event stream. `HashMapRepository` embeds an `InMemorySnapshotStore`; for production, implement the `SnapshotStore` trait for your backend. +- **On load**: If a usable snapshot cache record exists, the aggregate is restored from its payload and only events with `sequence > snapshot.version` are replayed. If no snapshot exists or the cache record is incompatible, full replay is used as a fallback. +- **Storage**: Snapshot cache records are stored separately from the event stream. They carry aggregate type, aggregate ID, covered event version, snapshot payload type/version, payload codec metadata, cache metadata, and timestamp. `HashMapRepository` embeds an `InMemorySnapshotStore`; durable async backends implement `AsyncSnapshotStore`. ## Event Upcasting / Versioning diff --git a/docs/async-repositories.md b/docs/async-repositories.md index c0b04714..473a762a 100644 --- a/docs/async-repositories.md +++ b/docs/async-repositories.md @@ -20,7 +20,10 @@ persistence should override it with an explicit durable name through - `AsyncReadModelStore`, `AsyncReadModelSessionStore`, and `AsyncRelationalReadModelQueryStore` mirror the current document and relational read-model surfaces for async adapters. -- `AsyncSnapshotStore` keys snapshots by full stream identity. +- `AsyncSnapshotStore` keys rebuildable snapshot cache records by full stream + identity. The record envelope carries stream identity, covered event version, + snapshot payload type/version, payload codec metadata, cache metadata, and + timestamp. - `AsyncOutboxStore` exposes async claim/update operations for durable outbox table stores. Aggregate repositories commit outbox rows transactionally, but workers do not hydrate outbox messages through aggregate repositories. diff --git a/docs/postgres-event-store.md b/docs/postgres-event-store.md index ae080d28..78520477 100644 --- a/docs/postgres-event-store.md +++ b/docs/postgres-event-store.md @@ -142,42 +142,46 @@ Recommended table name: `aggregate_snapshots`. | `aggregate_type` | `text` | Same stable aggregate type as events; `NOT NULL`. | | `aggregate_id` | `text` | Same aggregate ID as events; `NOT NULL`. | | `version` | `bigint` | Stream sequence covered by this snapshot; `NOT NULL`. | -| `payload` | `bytea` | Encoded snapshot payload bytes; `NOT NULL`. | +| `snapshot_type` | `text` | State snapshot payload type; `NOT NULL`. | +| `snapshot_version` | `integer` | State snapshot payload version; `NOT NULL`. | +| `payload` | `bytea` | Encoded state snapshot payload bytes; `NOT NULL`. | | `payload_codec` | `text` | Codec label; `NOT NULL`. | | `payload_codec_version` | `integer` | Codec metadata; `NOT NULL`. | +| `metadata` | `jsonb` | Cache metadata; `NOT NULL`, default `{}`. | | `recorded_at` | `timestamptz` | UTC instant for the snapshot; `NOT NULL`. | The DDL must declare `aggregate_type` and `aggregate_id` as `NOT NULL`; the -checks below also reject empty strings. `version`, `payload`, `payload_codec`, -`payload_codec_version`, and `recorded_at` must also be `NOT NULL`. +checks below also reject empty strings. `version`, `snapshot_type`, +`snapshot_version`, `payload`, `payload_codec`, `payload_codec_version`, +`metadata`, and `recorded_at` must also be `NOT NULL`. Required constraints and indexes: ```sql -PRIMARY KEY (aggregate_type, aggregate_id, version); +PRIMARY KEY (aggregate_type, aggregate_id); CHECK (aggregate_type <> ''); CHECK (aggregate_id <> ''); CHECK (version > 0); +CHECK (snapshot_type <> ''); +CHECK (snapshot_version > 0); CHECK (payload_codec <> ''); CHECK (payload_codec_version > 0); -CREATE INDEX aggregate_snapshots_latest - ON aggregate_snapshots (aggregate_type, aggregate_id, version DESC); ``` -Hydration should load the newest snapshot for the stream, then replay event rows -where `sequence > snapshot.version` ordered ascending. If no snapshot exists, -hydrate from sequence `1`. +The first implementation is latest-only: writing a snapshot cache record +upserts the `(aggregate_type, aggregate_id)` row. Hydration should load that +record, then replay event rows where `sequence > snapshot.version` ordered +ascending. If no usable snapshot exists, hydrate from sequence `1`. If the newest snapshot version exceeds the current maximum event sequence for the stream, the implementation should reject the load with `RepositoryError::Model`. That fail-fast behavior is preferred over continuing from an impossible snapshot tail because it surfaces data corruption early. -Snapshot retention is implementation-specific but must be explicit. The -contract permits multiple snapshots per stream. A Postgres implementation must -document whether it retains all snapshots, only the latest snapshot, last `N` -snapshots, or a time-based retention window. The first implementation should -prefer retaining all snapshots until a pruning policy and tests exist. +Snapshot retention is implementation-specific but must be explicit. The current +SQL adapters retain only the latest cache record per stream. Future adapters may +retain last `N` or time-based cache records, but they must never prune aggregate +events. ## Commit Semantics diff --git a/migrations/postgres/0001_initial.sql b/migrations/postgres/0001_initial.sql index a3da6de1..5f72ec62 100644 --- a/migrations/postgres/0001_initial.sql +++ b/migrations/postgres/0001_initial.sql @@ -28,12 +28,22 @@ CREATE TABLE IF NOT EXISTS aggregate_snapshots ( aggregate_type text NOT NULL, aggregate_id text NOT NULL, version bigint NOT NULL, - data bytea NOT NULL, + snapshot_type text NOT NULL, + snapshot_version integer NOT NULL, + payload bytea NOT NULL, + payload_codec text NOT NULL, + payload_codec_version integer NOT NULL, + metadata jsonb NOT NULL DEFAULT '{}'::jsonb, + recorded_at timestamptz NOT NULL DEFAULT now(), updated_at timestamptz NOT NULL DEFAULT now(), PRIMARY KEY (aggregate_type, aggregate_id), CHECK (aggregate_type <> ''), CHECK (aggregate_id <> ''), - CHECK (version > 0) + CHECK (version > 0), + CHECK (snapshot_type <> ''), + CHECK (snapshot_version > 0), + CHECK (payload_codec <> ''), + CHECK (payload_codec_version > 0) ); CREATE TABLE IF NOT EXISTS outbox_messages ( diff --git a/migrations/sqlite/0001_initial.sql b/migrations/sqlite/0001_initial.sql index 0929ed2a..f32e4038 100644 --- a/migrations/sqlite/0001_initial.sql +++ b/migrations/sqlite/0001_initial.sql @@ -49,12 +49,22 @@ CREATE TABLE IF NOT EXISTS aggregate_snapshots ( aggregate_type TEXT NOT NULL, aggregate_id TEXT NOT NULL, version INTEGER NOT NULL, - data BLOB NOT NULL, + snapshot_type TEXT NOT NULL, + snapshot_version INTEGER NOT NULL, + payload BLOB NOT NULL, + payload_codec TEXT NOT NULL, + payload_codec_version INTEGER NOT NULL, + metadata TEXT NOT NULL DEFAULT '{}', + recorded_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (aggregate_type, aggregate_id), CHECK (aggregate_type <> ''), CHECK (aggregate_id <> ''), - CHECK (version > 0) + CHECK (version > 0), + CHECK (snapshot_type <> ''), + CHECK (snapshot_version > 0), + CHECK (payload_codec <> ''), + CHECK (payload_codec_version > 0) ); CREATE TABLE IF NOT EXISTS outbox_messages ( diff --git a/src/entity/entity.rs b/src/entity/entity.rs index f69f5443..6ef48942 100644 --- a/src/entity/entity.rs +++ b/src/entity/entity.rs @@ -254,19 +254,6 @@ impl Entity { pub fn set_replaying(&mut self, replaying: bool) { self.replaying = replaying; } - - /// Replace all events with a single snapshot event. - /// Used by read models to store current state. - pub fn set_snapshot(&mut self, data: &T) -> SourcedResult { - let payload = BitcodePayloadCodec::encode(data).map_err(EventRecordError::encode)?; - self.events.clear(); - let record = EventRecord::new("Snapshot", payload, 1); - self.events.push(record); - self.version = 1; - self.committed_version = self.events.len() as u64; - self.timestamp = SystemTime::now(); - Ok(()) - } } #[cfg(test)] @@ -460,24 +447,6 @@ mod tests { assert_eq!(entity.events().len(), 2); } - #[test] - fn set_snapshot_resets_committed_version_to_snapshot_event_len() { - let mut source = Entity::new(); - source.digest("e1", &"a").unwrap(); - source.digest("e2", &"b").unwrap(); - - let mut entity = Entity::new(); - entity.load_from_history(source.events().to_vec()); - assert_eq!(entity.committed_version(), 2); - - entity.set_snapshot(&"snapshot").unwrap(); - - assert_eq!(entity.events().len(), 1); - assert_eq!(entity.version(), 1); - assert_eq!(entity.committed_version(), 1); - assert!(entity.new_events().is_empty()); - } - #[test] fn digest_propagates_metadata_to_event_record() { let mut entity = Entity::new(); diff --git a/src/hashmap_repo/repository.rs b/src/hashmap_repo/repository.rs index f723fb55..814853a3 100644 --- a/src/hashmap_repo/repository.rs +++ b/src/hashmap_repo/repository.rs @@ -345,6 +345,7 @@ impl TransactionalCommit for HashMapRepository { for write in batch.snapshots { match write { SnapshotWrite::Save(record) => { + record.validate()?; staged_snapshots.insert(record.aggregate_id.clone(), record); } } @@ -465,13 +466,7 @@ fn validate_snapshot_identity( identity: &StreamIdentity, record: &SnapshotRecord, ) -> Result<(), RepositoryError> { - if record.aggregate_id != identity.aggregate_id() { - return Err(RepositoryError::Model(format!( - "snapshot aggregate id `{}` does not match stream identity `{}`", - record.aggregate_id, identity - ))); - } - Ok(()) + record.validate_for_identity(identity) } fn reject_duplicate_streams(entities: &[&mut Entity]) -> Result<(), RepositoryError> { diff --git a/src/lib.rs b/src/lib.rs index 0536f20d..a30942a0 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -144,10 +144,10 @@ pub use commit_builder::{ CommitBuilder, CommitBuilderExt, ReadModelSessionCommitExt, StagedCommitBuilder, }; -// Snapshot: periodic aggregate snapshots for fast hydration +// Snapshot: state snapshot payloads and rebuildable cache records for hydration pub use snapshot::{ - hydrate_from_snapshot, InMemorySnapshotStore, SnapshotAggregateRepository, SnapshotRecord, - SnapshotStore, Snapshottable, + hydrate_from_snapshot, AsyncSnapshotAggregateRepository, InMemorySnapshotStore, + SnapshotAggregateRepository, SnapshotRecord, SnapshotStore, Snapshottable, }; // Re-export the EventEmitter from the event_emitter_rs crate (requires "emitter" feature) diff --git a/src/postgres_repo/mod.rs b/src/postgres_repo/mod.rs index 75318f38..c3c1b4c3 100644 --- a/src/postgres_repo/mod.rs +++ b/src/postgres_repo/mod.rs @@ -604,7 +604,16 @@ impl AsyncSnapshotStore for PostgresRepository { async move { let row = sqlx::query( r#" - SELECT aggregate_id, version, data + SELECT aggregate_type, + aggregate_id, + version, + snapshot_type, + snapshot_version, + payload, + payload_codec, + payload_codec_version, + metadata::text AS metadata, + EXTRACT(EPOCH FROM recorded_at)::double precision AS recorded_at_epoch FROM aggregate_snapshots WHERE aggregate_type = $1 AND aggregate_id = $2 "#, @@ -1108,11 +1117,28 @@ async fn save_snapshot_in_tx( sqlx::query( r#" - INSERT INTO aggregate_snapshots (aggregate_type, aggregate_id, version, data) - VALUES ($1, $2, $3, $4) + INSERT INTO aggregate_snapshots ( + aggregate_type, + aggregate_id, + version, + snapshot_type, + snapshot_version, + payload, + payload_codec, + payload_codec_version, + metadata, + recorded_at + ) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::jsonb, to_timestamp($10)) ON CONFLICT(aggregate_type, aggregate_id) DO UPDATE SET version = excluded.version, - data = excluded.data, + snapshot_type = excluded.snapshot_type, + snapshot_version = excluded.snapshot_version, + payload = excluded.payload, + payload_codec = excluded.payload_codec, + payload_codec_version = excluded.payload_codec_version, + metadata = excluded.metadata, + recorded_at = excluded.recorded_at, updated_at = now() "#, ) @@ -1124,7 +1150,18 @@ async fn save_snapshot_in_tx( "snapshot version", BIGINT_STORAGE, )?) - .bind(record.data) + .bind(&record.snapshot_type) + .bind(sqlx_repository_i32_from_u64( + POSTGRES_BACKEND, + record.snapshot_version, + "snapshot payload version", + INTEGER_STORAGE, + )?) + .bind(&record.payload) + .bind(&record.payload_codec) + .bind(i32::from(record.payload_codec_version)) + .bind(serialize_event_metadata(&record.metadata)?) + .bind(system_time_to_epoch_secs(record.recorded_at)?) .execute(&mut **tx) .await .map_err(|err| repository_storage_error("save snapshot", err))?; @@ -1133,7 +1170,13 @@ async fn save_snapshot_in_tx( } fn snapshot_from_row(row: PgRow) -> Result { + let metadata_json: String = row + .try_get("metadata") + .map_err(|err| repository_storage_error("decode snapshot metadata row", err))?; Ok(SnapshotRecord { + aggregate_type: row + .try_get("aggregate_type") + .map_err(|err| repository_storage_error("decode snapshot aggregate type row", err))?, aggregate_id: row .try_get("aggregate_id") .map_err(|err| repository_storage_error("decode snapshot aggregate id row", err))?, @@ -1143,9 +1186,34 @@ fn snapshot_from_row(row: PgRow) -> Result { .map_err(|err| repository_storage_error("decode snapshot version row", err))?, "snapshot version", )?, - data: row - .try_get("data") - .map_err(|err| repository_storage_error("decode snapshot data row", err))?, + snapshot_type: row + .try_get("snapshot_type") + .map_err(|err| repository_storage_error("decode snapshot type row", err))?, + snapshot_version: sqlx_repository_u64_from_i32( + POSTGRES_BACKEND, + row.try_get("snapshot_version").map_err(|err| { + repository_storage_error("decode snapshot payload version row", err) + })?, + "snapshot payload version", + )?, + payload_codec: row + .try_get("payload_codec") + .map_err(|err| repository_storage_error("decode snapshot payload codec row", err))?, + payload_codec_version: sqlx_repository_u16_from_i32( + POSTGRES_BACKEND, + row.try_get("payload_codec_version").map_err(|err| { + repository_storage_error("decode snapshot payload codec version row", err) + })?, + "snapshot payload codec version", + )?, + payload: row + .try_get("payload") + .map_err(|err| repository_storage_error("decode snapshot payload row", err))?, + metadata: deserialize_event_metadata(&metadata_json)?, + recorded_at: system_time_from_epoch_secs( + row.try_get("recorded_at_epoch") + .map_err(|err| repository_storage_error("decode snapshot recorded_at row", err))?, + )?, }) } diff --git a/src/snapshot/in_memory.rs b/src/snapshot/in_memory.rs index e1c0e79a..d4cfd8e0 100644 --- a/src/snapshot/in_memory.rs +++ b/src/snapshot/in_memory.rs @@ -44,6 +44,7 @@ impl SnapshotStore for InMemorySnapshotStore { } fn save_snapshot(&self, record: SnapshotRecord) -> Result<(), RepositoryError> { + record.validate()?; let mut storage = self .storage .write() @@ -81,6 +82,7 @@ impl AsyncSnapshotStore for InMemorySnapshotStore { record: SnapshotRecord, ) -> impl Future> + Send + 'a { async move { + record.validate_for_identity(identity)?; let mut storage = self .storage .write() @@ -111,16 +113,20 @@ mod tests { #[test] fn save_and_get() { let store = InMemorySnapshotStore::new(); - let record = SnapshotRecord { - aggregate_id: "agg-1".into(), - version: 5, - data: vec![1, 2, 3], - }; + let record = SnapshotRecord::new( + "test.aggregate", + "agg-1", + 5, + "TestSnapshot", + 1, + vec![1, 2, 3], + ); store.save_snapshot(record).unwrap(); let loaded = store.get_snapshot("agg-1").unwrap().unwrap(); assert_eq!(loaded.version, 5); - assert_eq!(loaded.data, vec![1, 2, 3]); + assert_eq!(loaded.payload, vec![1, 2, 3]); + assert_eq!(loaded.snapshot_type, "TestSnapshot"); } #[test] @@ -133,34 +139,43 @@ mod tests { fn save_overwrites() { let store = InMemorySnapshotStore::new(); store - .save_snapshot(SnapshotRecord { - aggregate_id: "agg-1".into(), - version: 1, - data: vec![1], - }) + .save_snapshot(SnapshotRecord::new( + "test.aggregate", + "agg-1", + 1, + "TestSnapshot", + 1, + vec![1], + )) .unwrap(); store - .save_snapshot(SnapshotRecord { - aggregate_id: "agg-1".into(), - version: 5, - data: vec![5], - }) + .save_snapshot(SnapshotRecord::new( + "test.aggregate", + "agg-1", + 5, + "TestSnapshot", + 1, + vec![5], + )) .unwrap(); let loaded = store.get_snapshot("agg-1").unwrap().unwrap(); assert_eq!(loaded.version, 5); - assert_eq!(loaded.data, vec![5]); + assert_eq!(loaded.payload, vec![5]); } #[test] fn delete_existing() { let store = InMemorySnapshotStore::new(); store - .save_snapshot(SnapshotRecord { - aggregate_id: "agg-1".into(), - version: 1, - data: vec![1], - }) + .save_snapshot(SnapshotRecord::new( + "test.aggregate", + "agg-1", + 1, + "TestSnapshot", + 1, + vec![1], + )) .unwrap(); assert!(store.delete_snapshot("agg-1").unwrap()); assert!(store.get_snapshot("agg-1").unwrap().is_none()); @@ -177,11 +192,14 @@ mod tests { let store = InMemorySnapshotStore::new(); let clone = store.clone(); store - .save_snapshot(SnapshotRecord { - aggregate_id: "agg-1".into(), - version: 3, - data: vec![3], - }) + .save_snapshot(SnapshotRecord::new( + "test.aggregate", + "agg-1", + 3, + "TestSnapshot", + 1, + vec![3], + )) .unwrap(); let loaded = clone.get_snapshot("agg-1").unwrap().unwrap(); diff --git a/src/snapshot/mod.rs b/src/snapshot/mod.rs index 3c63001a..5da3ebaa 100644 --- a/src/snapshot/mod.rs +++ b/src/snapshot/mod.rs @@ -4,6 +4,8 @@ mod snapshottable; mod store; pub use in_memory::InMemorySnapshotStore; -pub use repository::{hydrate_from_snapshot, SnapshotAggregateRepository}; +pub use repository::{ + hydrate_from_snapshot, AsyncSnapshotAggregateRepository, SnapshotAggregateRepository, +}; pub use snapshottable::Snapshottable; pub use store::{SnapshotRecord, SnapshotStore}; diff --git a/src/snapshot/repository.rs b/src/snapshot/repository.rs index 0bbb119f..2724f266 100644 --- a/src/snapshot/repository.rs +++ b/src/snapshot/repository.rs @@ -1,16 +1,38 @@ -use crate::aggregate::{hydrate, AggregateRepository}; +use crate::aggregate::{hydrate, AggregateRepository, AsyncAggregateRepository}; use crate::entity::{upcast_events, Entity}; use crate::queued_repo::{GetAllWithOpts, GetWithOpts, ReadOpts, UnlockableRepository}; -use crate::repository::{CommitBatch, Get, RepositoryError, SnapshotWrite, TransactionalCommit}; +use crate::repository::{ + AsyncCommitBatch, AsyncGetStream, AsyncSnapshotStore, AsyncSnapshotWrite, AsyncStreamWrite, + AsyncTransactionalCommit, CommitBatch, Get, RepositoryError, SnapshotWrite, StreamIdentity, + TransactionalCommit, +}; use super::snapshottable::Snapshottable; use super::store::{SnapshotRecord, SnapshotStore}; -/// Hydrate an aggregate from a snapshot, replaying only events after the snapshot version. +#[derive(Debug, PartialEq, Eq)] +enum SnapshotHydrationError { + Cache(String), + Replay(String), +} + +/// Hydrate an aggregate from a snapshot cache record, replaying only events +/// after the snapshot version. pub fn hydrate_from_snapshot( entity: Entity, snapshot: SnapshotRecord, ) -> Result { + try_hydrate_from_snapshot::(entity, snapshot).map_err(|err| match err { + SnapshotHydrationError::Cache(message) | SnapshotHydrationError::Replay(message) => { + RepositoryError::Replay(message) + } + }) +} + +fn try_hydrate_from_snapshot( + entity: Entity, + snapshot: SnapshotRecord, +) -> Result { let mut agg = A::new_empty(); *agg.entity_mut() = entity; @@ -18,8 +40,15 @@ pub fn hydrate_from_snapshot( agg.entity_mut().set_snapshot_version(snapshot.version); // Restore aggregate state from snapshot - let snap: A::Snapshot = bitcode::deserialize(&snapshot.data) - .map_err(|e| RepositoryError::Replay(format!("snapshot deserialize: {e}")))?; + if !snapshot.has_supported_payload_codec() { + return Err(SnapshotHydrationError::Cache(format!( + "unsupported snapshot payload codec `{}` version {}", + snapshot.payload_codec, snapshot.payload_codec_version + ))); + } + + let snap: A::Snapshot = bitcode::deserialize(&snapshot.payload) + .map_err(|e| SnapshotHydrationError::Cache(format!("snapshot deserialize: {e}")))?; agg.restore_from_snapshot(snap); // Replay only events AFTER the snapshot @@ -37,20 +66,67 @@ pub fn hydrate_from_snapshot( post_snapshot } else { upcast_events(post_snapshot, upcasters) - .map_err(|err| RepositoryError::Replay(err.to_string()))? + .map_err(|err| SnapshotHydrationError::Replay(err.to_string()))? }; agg.entity_mut().set_replaying(true); for event in &events { if let Err(err) = agg.replay_event(event) { agg.entity_mut().set_replaying(false); - return Err(RepositoryError::Replay(err.to_string())); + return Err(SnapshotHydrationError::Replay(err.to_string())); } } agg.entity_mut().set_replaying(false); Ok(agg) } +fn snapshot_type_name() -> String { + std::any::type_name::().to_string() +} + +fn snapshot_record_for(aggregate: &A) -> Result { + let payload = bitcode::serialize(&aggregate.create_snapshot()) + .map_err(|e| RepositoryError::Replay(format!("snapshot serialize: {e}")))?; + + Ok(SnapshotRecord::new( + A::aggregate_type(), + aggregate.entity().id(), + aggregate.entity().version(), + snapshot_type_name::(), + SnapshotRecord::DEFAULT_SNAPSHOT_VERSION, + payload, + )) +} + +fn hydrate_with_optional_snapshot( + entity: Entity, + snapshot: Option, +) -> Result { + let Some(snapshot) = snapshot else { + return hydrate::(entity); + }; + + if snapshot.aggregate_id != entity.id() || snapshot.aggregate_type != A::aggregate_type() { + return hydrate::(entity); + } + + if snapshot.version > entity.version() { + return Err(RepositoryError::Model(format!( + "snapshot cache version {} exceeds stream version {} for {}:{}", + snapshot.version, + entity.version(), + snapshot.aggregate_type, + snapshot.aggregate_id + ))); + } + + match try_hydrate_from_snapshot::(entity.clone(), snapshot) { + Ok(aggregate) => Ok(aggregate), + Err(SnapshotHydrationError::Cache(_)) => hydrate::(entity), + Err(SnapshotHydrationError::Replay(message)) => Err(RepositoryError::Replay(message)), + } +} + /// A repository wrapper that provides snapshot-aware get and commit for a specific aggregate type. pub struct SnapshotAggregateRepository { inner: AggregateRepository, @@ -68,6 +144,147 @@ impl SnapshotAggregateRepository { } } +/// Async repository wrapper that treats aggregate snapshots as rebuildable +/// hydration cache records. +pub struct AsyncSnapshotAggregateRepository { + inner: AsyncAggregateRepository, + frequency: u64, +} + +impl AsyncSnapshotAggregateRepository { + pub fn new(inner: AsyncAggregateRepository, frequency: u64) -> Self { + Self { inner, frequency } + } + + pub fn repo(&self) -> &AsyncAggregateRepository { + &self.inner + } +} + +impl AsyncAggregateRepository { + /// Wrap this async repository with snapshot cache support at the given event + /// frequency. + pub fn with_snapshots(self, frequency: u64) -> AsyncSnapshotAggregateRepository { + AsyncSnapshotAggregateRepository::new(self, frequency) + } +} + +impl AsyncSnapshotAggregateRepository +where + R: AsyncGetStream + AsyncSnapshotStore, + A: Snapshottable + Send, +{ + pub async fn get(&self, id: &str) -> Result, RepositoryError> { + let identity = StreamIdentity::new(A::aggregate_type(), id)?; + let entity = self.inner.repo().get_stream(&identity).await?; + let Some(entity) = entity else { + return Ok(None); + }; + let snapshot = self.inner.repo().get_snapshot_async(&identity).await?; + Ok(Some(hydrate_with_optional_snapshot::(entity, snapshot)?)) + } + + pub async fn get_all(&self, ids: &[&str]) -> Result, RepositoryError> { + let identities = ids + .iter() + .map(|id| StreamIdentity::new(A::aggregate_type(), *id)) + .collect::, _>>()?; + let entities = self.inner.repo().get_streams(&identities).await?; + let mut aggregates = Vec::with_capacity(entities.len()); + for entity in entities { + let identity = StreamIdentity::new(A::aggregate_type(), entity.id())?; + let snapshot = self.inner.repo().get_snapshot_async(&identity).await?; + aggregates.push(hydrate_with_optional_snapshot::(entity, snapshot)?); + } + Ok(aggregates) + } +} + +impl AsyncSnapshotAggregateRepository +where + R: AsyncTransactionalCommit, + A: Snapshottable + Send, +{ + pub async fn commit(&self, aggregate: &mut A) -> Result<(), RepositoryError> { + let snapshot = self.snapshot_record(aggregate)?; + let snapshot_version = snapshot.as_ref().map(|record| record.version); + let identity = StreamIdentity::new(A::aggregate_type(), aggregate.entity().id())?; + let snapshots = snapshot + .into_iter() + .map(|record| AsyncSnapshotWrite::Save { + identity: identity.clone(), + record, + }) + .collect(); + + self.inner + .repo() + .commit_batch_async(AsyncCommitBatch { + streams: vec![AsyncStreamWrite::new(identity, aggregate.entity_mut())], + outbox_messages: Vec::new(), + read_model_plans: Vec::new(), + snapshots, + }) + .await?; + + if let Some(version) = snapshot_version { + aggregate.entity_mut().set_snapshot_version(version); + } + Ok(()) + } + + pub async fn commit_all(&self, aggregates: &mut [&mut A]) -> Result<(), RepositoryError> { + let mut snapshot_versions = Vec::with_capacity(aggregates.len()); + let mut snapshots = Vec::new(); + for aggregate in aggregates.iter() { + let snapshot = self.snapshot_record(*aggregate)?; + snapshot_versions.push(snapshot.as_ref().map(|record| record.version)); + if let Some(record) = snapshot { + snapshots.push(AsyncSnapshotWrite::Save { + identity: StreamIdentity::new( + A::aggregate_type(), + record.aggregate_id.as_str(), + )?, + record, + }); + } + } + + let mut streams = Vec::with_capacity(aggregates.len()); + for aggregate in aggregates.iter_mut() { + let identity = StreamIdentity::new(A::aggregate_type(), (*aggregate).entity().id())?; + streams.push(AsyncStreamWrite::new(identity, (*aggregate).entity_mut())); + } + + self.inner + .repo() + .commit_batch_async(AsyncCommitBatch { + streams, + outbox_messages: Vec::new(), + read_model_plans: Vec::new(), + snapshots, + }) + .await?; + + for (aggregate, snapshot_version) in aggregates.iter_mut().zip(snapshot_versions) { + if let Some(version) = snapshot_version { + aggregate.entity_mut().set_snapshot_version(version); + } + } + Ok(()) + } + + fn snapshot_record(&self, aggregate: &A) -> Result, RepositoryError> { + let version = aggregate.entity().version(); + let snap_version = aggregate.entity().snapshot_version(); + + if version >= snap_version + self.frequency { + return snapshot_record_for(aggregate).map(Some); + } + Ok(None) + } +} + // ============================================================================ // get / get_all — snapshot-aware hydration // ============================================================================ @@ -103,12 +320,7 @@ where entity: Entity, snapshot: Option, ) -> Result { - match snapshot { - Some(snap) if snap.version <= entity.version() => { - hydrate_from_snapshot::(entity, snap) - } - _ => hydrate::(entity), - } + hydrate_with_optional_snapshot::(entity, snapshot) } } @@ -176,15 +388,7 @@ where let snap_version = aggregate.entity().snapshot_version(); if version >= snap_version + self.frequency { - let snap = aggregate.create_snapshot(); - let data = bitcode::serialize(&snap) - .map_err(|e| RepositoryError::Replay(format!("snapshot serialize: {e}")))?; - - return Ok(Some(SnapshotRecord { - aggregate_id: aggregate.entity().id().to_string(), - version, - data, - })); + return snapshot_record_for(aggregate).map(Some); } Ok(None) } @@ -216,12 +420,7 @@ where return Ok(None); }; let snapshot = self.inner.repo().get_snapshot(id)?; - match snapshot { - Some(snap) if snap.version <= entity.version() => { - Ok(Some(hydrate_from_snapshot::(entity, snap)?)) - } - _ => Ok(Some(hydrate::(entity)?)), - } + Ok(Some(hydrate_with_optional_snapshot::(entity, snapshot)?)) } } @@ -236,13 +435,7 @@ where let mut aggregates = Vec::with_capacity(entities.len()); for entity in entities { let snapshot = self.inner.repo().get_snapshot(entity.id())?; - let agg = match snapshot { - Some(snap) if snap.version <= entity.version() => { - hydrate_from_snapshot::(entity, snap)? - } - _ => hydrate::(entity)?, - }; - aggregates.push(agg); + aggregates.push(hydrate_with_optional_snapshot::(entity, snapshot)?); } Ok(aggregates) } diff --git a/src/snapshot/snapshottable.rs b/src/snapshot/snapshottable.rs index 5170cdb5..09d8e371 100644 --- a/src/snapshot/snapshottable.rs +++ b/src/snapshot/snapshottable.rs @@ -2,10 +2,13 @@ use serde::{de::DeserializeOwned, Serialize}; use crate::aggregate::Aggregate; -/// Opt-in trait for aggregates that support snapshot-based hydration. +/// Opt-in trait for aggregates that can produce state snapshot payloads. /// -/// Aggregates implementing this trait can have their state serialized at a point -/// in time and restored later, skipping costly full event replay. +/// Aggregates implementing this trait can have their state captured as a DTO at +/// a point in time and restored later. Repositories may serialize that payload +/// into a snapshot cache record to skip costly full event replay, but the +/// payload itself is not an aggregate event and is not durable history by +/// itself. /// /// The associated `Snapshot` type is a separate struct (e.g., `TodoSnapshot`) /// that captures the aggregate's current state. diff --git a/src/snapshot/store.rs b/src/snapshot/store.rs index 8681705e..a06e96d9 100644 --- a/src/snapshot/store.rs +++ b/src/snapshot/store.rs @@ -1,14 +1,120 @@ -use crate::repository::RepositoryError; +use std::collections::HashMap; +use std::time::SystemTime; -/// A stored snapshot record: aggregate ID, version at time of snapshot, and serialized data. -#[derive(Clone, Debug)] +use crate::entity::{BITCODE_PAYLOAD_CODEC, BITCODE_PAYLOAD_CODEC_VERSION}; +use crate::repository::{RepositoryError, StreamIdentity}; + +/// Stored aggregate snapshot cache record. +/// +/// This is a repository-owned cache envelope, not an aggregate event and not +/// the user-defined state snapshot payload itself. The payload bytes usually +/// come from `Snapshottable::create_snapshot()`. +#[derive(Clone, Debug, PartialEq)] pub struct SnapshotRecord { + pub aggregate_type: String, pub aggregate_id: String, + /// Aggregate event sequence covered by this cache record. pub version: u64, - pub data: Vec, + pub snapshot_type: String, + pub snapshot_version: u64, + pub payload_codec: String, + pub payload_codec_version: u16, + pub payload: Vec, + pub metadata: HashMap, + pub recorded_at: SystemTime, +} + +impl SnapshotRecord { + pub const DEFAULT_SNAPSHOT_VERSION: u64 = 1; + + pub fn new( + aggregate_type: impl Into, + aggregate_id: impl Into, + version: u64, + snapshot_type: impl Into, + snapshot_version: u64, + payload: Vec, + ) -> Self { + Self { + aggregate_type: aggregate_type.into(), + aggregate_id: aggregate_id.into(), + version, + snapshot_type: snapshot_type.into(), + snapshot_version, + payload_codec: BITCODE_PAYLOAD_CODEC.to_string(), + payload_codec_version: BITCODE_PAYLOAD_CODEC_VERSION, + payload, + metadata: HashMap::new(), + recorded_at: SystemTime::now(), + } + } + + pub fn validate_for_identity(&self, identity: &StreamIdentity) -> Result<(), RepositoryError> { + self.validate()?; + if self.aggregate_type != identity.aggregate_type() { + return Err(RepositoryError::Model(format!( + "snapshot aggregate type `{}` does not match stream identity `{}`", + self.aggregate_type, identity + ))); + } + if self.aggregate_id != identity.aggregate_id() { + return Err(RepositoryError::Model(format!( + "snapshot aggregate id `{}` does not match stream identity `{}`", + self.aggregate_id, identity + ))); + } + Ok(()) + } + + pub fn validate(&self) -> Result<(), RepositoryError> { + if self.aggregate_type.trim().is_empty() { + return Err(RepositoryError::Model( + "snapshot aggregate type must not be empty".into(), + )); + } + if self.aggregate_id.trim().is_empty() { + return Err(RepositoryError::Model( + "snapshot aggregate id must not be empty".into(), + )); + } + if self.version == 0 { + return Err(RepositoryError::Model( + "snapshot version must be greater than zero".into(), + )); + } + if self.snapshot_type.trim().is_empty() { + return Err(RepositoryError::Model( + "snapshot type must not be empty".into(), + )); + } + if self.snapshot_version == 0 { + return Err(RepositoryError::Model( + "snapshot payload version must be greater than zero".into(), + )); + } + if self.payload_codec.trim().is_empty() { + return Err(RepositoryError::Model( + "snapshot payload codec must not be empty".into(), + )); + } + if self.payload_codec_version == 0 { + return Err(RepositoryError::Model( + "snapshot payload codec version must be greater than zero".into(), + )); + } + Ok(()) + } + + pub fn has_supported_payload_codec(&self) -> bool { + self.payload_codec == BITCODE_PAYLOAD_CODEC + && self.payload_codec_version == BITCODE_PAYLOAD_CODEC_VERSION + } } -/// Trait for snapshot persistence. One snapshot per aggregate ID (latest wins). +/// Trait for ID-only snapshot persistence. One snapshot per aggregate ID (latest wins). +/// +/// Durable async repositories should prefer `AsyncSnapshotStore`, which keys +/// cache records by full `StreamIdentity`. pub trait SnapshotStore: Send + Sync { /// Load the latest snapshot for the given aggregate ID. fn get_snapshot(&self, id: &str) -> Result, RepositoryError>; diff --git a/src/sqlite_repo/mod.rs b/src/sqlite_repo/mod.rs index ce44cfe5..178fe166 100644 --- a/src/sqlite_repo/mod.rs +++ b/src/sqlite_repo/mod.rs @@ -852,7 +852,9 @@ impl AsyncSnapshotStore for SqliteRepository { async move { let row = sqlx::query( r#" - SELECT aggregate_id, version, data + SELECT aggregate_type, aggregate_id, version, snapshot_type, + snapshot_version, payload, payload_codec, + payload_codec_version, metadata, recorded_at FROM aggregate_snapshots WHERE aggregate_type = ? AND aggregate_id = ? "#, @@ -1551,11 +1553,28 @@ async fn save_snapshot_in_tx( sqlx::query( r#" - INSERT INTO aggregate_snapshots (aggregate_type, aggregate_id, version, data) - VALUES (?, ?, ?, ?) + INSERT INTO aggregate_snapshots ( + aggregate_type, + aggregate_id, + version, + snapshot_type, + snapshot_version, + payload, + payload_codec, + payload_codec_version, + metadata, + recorded_at + ) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(aggregate_type, aggregate_id) DO UPDATE SET version = excluded.version, - data = excluded.data, + snapshot_type = excluded.snapshot_type, + snapshot_version = excluded.snapshot_version, + payload = excluded.payload, + payload_codec = excluded.payload_codec, + payload_codec_version = excluded.payload_codec_version, + metadata = excluded.metadata, + recorded_at = excluded.recorded_at, updated_at = CURRENT_TIMESTAMP "#, ) @@ -1567,7 +1586,18 @@ async fn save_snapshot_in_tx( "snapshot version", SIGNED_INTEGER_STORAGE, )?) - .bind(record.data) + .bind(&record.snapshot_type) + .bind(sqlx_repository_i64_from_u64( + SQLITE_BACKEND, + record.snapshot_version, + "snapshot payload version", + SIGNED_INTEGER_STORAGE, + )?) + .bind(&record.payload) + .bind(&record.payload_codec) + .bind(i64::from(record.payload_codec_version)) + .bind(serialize_event_metadata(&record.metadata)?) + .bind(system_time_to_storage(record.recorded_at)?) .execute(&mut **tx) .await .map_err(|err| repository_storage_error("save snapshot", err))?; @@ -1576,7 +1606,13 @@ async fn save_snapshot_in_tx( } fn snapshot_from_row(row: sqlx::sqlite::SqliteRow) -> Result { + let metadata_json: String = row + .try_get("metadata") + .map_err(|err| repository_storage_error("decode snapshot metadata row", err))?; Ok(SnapshotRecord { + aggregate_type: row + .try_get("aggregate_type") + .map_err(|err| repository_storage_error("decode snapshot aggregate type row", err))?, aggregate_id: row .try_get("aggregate_id") .map_err(|err| repository_storage_error("decode snapshot aggregate id row", err))?, @@ -1586,9 +1622,35 @@ fn snapshot_from_row(row: sqlx::sqlite::SqliteRow) -> Result("recorded_at") + .map_err(|err| repository_storage_error("decode snapshot recorded_at row", err))? + .as_str(), + ), }) } diff --git a/src/sqlx_repo/mod.rs b/src/sqlx_repo/mod.rs index 6cd5a982..09ad03ce 100644 --- a/src/sqlx_repo/mod.rs +++ b/src/sqlx_repo/mod.rs @@ -113,13 +113,7 @@ pub(crate) fn validate_snapshot_identity( identity: &StreamIdentity, record: &SnapshotRecord, ) -> Result<(), RepositoryError> { - if record.aggregate_id != identity.aggregate_id() { - return Err(RepositoryError::Model(format!( - "snapshot aggregate id `{}` does not match stream identity `{}`", - record.aggregate_id, identity - ))); - } - Ok(()) + record.validate_for_identity(identity) } pub(crate) fn repository_i64_from_u64( diff --git a/tests/async_repository/main.rs b/tests/async_repository/main.rs index 83140ba7..d7af3056 100644 --- a/tests/async_repository/main.rs +++ b/tests/async_repository/main.rs @@ -6,7 +6,8 @@ use sourced_rust::{ AsyncOutboxStore, AsyncReadModelSessionStore, AsyncReadModelStore, AsyncSnapshotStore, AsyncStreamWrite, AsyncTransactionalCommit, ClaimOutboxMessages, Entity, EventRecord, HashMapRepository, InMemorySnapshotStore, OutboxMessage, ProcessedMessageMark, ReadModel, - ReadModelSession, ReadModelWritePlan, RepositoryError, SnapshotRecord, StreamIdentity, + ReadModelSession, ReadModelWritePlan, RepositoryError, SnapshotRecord, Snapshottable, + StreamIdentity, }; #[derive(Default)] @@ -50,6 +51,46 @@ impl BetaAggregate { impl_aggregate!(BetaAggregate, entity, replay, aggregate_type = "async.beta"); +#[derive(Default)] +struct SnapshotCounter { + entity: Entity, + value: i32, +} + +impl SnapshotCounter { + fn increment(&mut self, id: &str, by: i32) { + self.entity.set_id(id); + self.entity.digest("Incremented", &by).unwrap(); + self.value += by; + } + + fn replay(&mut self, event: &EventRecord) -> Result<(), String> { + if event.event_name == "Incremented" { + self.value += event.decode::().map_err(|err| err.to_string())?; + } + Ok(()) + } +} + +impl_aggregate!( + SnapshotCounter, + entity, + replay, + aggregate_type = "async.snapshot_counter" +); + +impl Snapshottable for SnapshotCounter { + type Snapshot = i32; + + fn create_snapshot(&self) -> Self::Snapshot { + self.value + } + + fn restore_from_snapshot(&mut self, snapshot: Self::Snapshot) { + self.value = snapshot; + } +} + #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] struct TestView { id: String, @@ -179,22 +220,14 @@ async fn async_snapshot_store_uses_full_stream_identity() { store .save_snapshot_async( &alpha, - SnapshotRecord { - aggregate_id: "same-id".into(), - version: 1, - data: vec![1], - }, + SnapshotRecord::new("async.alpha", "same-id", 1, "AlphaSnapshot", 1, vec![1]), ) .await .unwrap(); store .save_snapshot_async( &beta, - SnapshotRecord { - aggregate_id: "same-id".into(), - version: 2, - data: vec![2], - }, + SnapshotRecord::new("async.beta", "same-id", 2, "BetaSnapshot", 1, vec![2]), ) .await .unwrap(); @@ -204,6 +237,101 @@ async fn async_snapshot_store_uses_full_stream_identity() { assert_eq!(loaded_alpha.version, 1); assert_eq!(loaded_beta.version, 2); + assert_eq!(loaded_alpha.aggregate_type, "async.alpha"); + assert_eq!(loaded_beta.aggregate_type, "async.beta"); +} + +#[tokio::test] +async fn async_snapshot_repository_writes_cache_without_event_record() { + let repo = HashMapRepository::new(); + let snapshot_repo = repo + .clone() + .async_aggregate::() + .with_snapshots(2); + let id = "snapshot-counter-1"; + + let mut counter = SnapshotCounter::default(); + counter.increment(id, 2); + snapshot_repo.commit(&mut counter).await.unwrap(); + counter.increment(id, 3); + snapshot_repo.commit(&mut counter).await.unwrap(); + + let identity = StreamIdentity::new(SnapshotCounter::aggregate_type(), id).unwrap(); + let stream = repo.get_stream(&identity).await.unwrap().unwrap(); + let snapshot = repo.get_snapshot_async(&identity).await.unwrap().unwrap(); + + assert_eq!(stream.events().len(), 2); + assert_eq!(stream.events()[0].event_name, "Incremented"); + assert_eq!(stream.events()[1].event_name, "Incremented"); + assert_eq!(snapshot.version, 2); + assert_eq!(snapshot.aggregate_type, SnapshotCounter::aggregate_type()); + assert_eq!(snapshot.payload, bitcode::serialize(&5_i32).unwrap()); +} + +#[tokio::test] +async fn async_snapshot_repository_ignores_invalid_cache_and_replays_events() { + let repo = HashMapRepository::new(); + let aggregate_repo = repo.clone().async_aggregate::(); + let snapshot_repo = repo + .clone() + .async_aggregate::() + .with_snapshots(10); + let id = "snapshot-counter-invalid"; + + let mut counter = SnapshotCounter::default(); + counter.increment(id, 4); + counter.increment(id, 6); + aggregate_repo.commit(&mut counter).await.unwrap(); + + let identity = StreamIdentity::new(SnapshotCounter::aggregate_type(), id).unwrap(); + let mut invalid = SnapshotRecord::new( + SnapshotCounter::aggregate_type(), + id, + 1, + std::any::type_name::(), + 1, + vec![0xff], + ); + invalid.payload_codec = "json".into(); + repo.save_snapshot_async(&identity, invalid).await.unwrap(); + + let loaded = snapshot_repo.get(id).await.unwrap().unwrap(); + assert_eq!(loaded.value, 10); + assert_eq!(loaded.entity().snapshot_version(), 0); +} + +#[tokio::test] +async fn async_snapshot_repository_rejects_cache_past_stream_version() { + let repo = HashMapRepository::new(); + let aggregate_repo = repo.clone().async_aggregate::(); + let snapshot_repo = repo + .clone() + .async_aggregate::() + .with_snapshots(10); + let id = "snapshot-counter-ahead"; + + let mut counter = SnapshotCounter::default(); + counter.increment(id, 4); + aggregate_repo.commit(&mut counter).await.unwrap(); + + let identity = StreamIdentity::new(SnapshotCounter::aggregate_type(), id).unwrap(); + let record = SnapshotRecord::new( + SnapshotCounter::aggregate_type(), + id, + 2, + std::any::type_name::(), + 1, + bitcode::serialize(&4_i32).unwrap(), + ); + repo.save_snapshot_async(&identity, record).await.unwrap(); + + let err = match snapshot_repo.get(id).await { + Err(err) => err, + Ok(_) => panic!("snapshot cache past stream version should fail"), + }; + assert!( + matches!(err, RepositoryError::Model(message) if message.contains("exceeds stream version")) + ); } #[tokio::test] diff --git a/tests/persistent_repository_conformance/scenario.rs b/tests/persistent_repository_conformance/scenario.rs index c892a1b6..2105ac9a 100644 --- a/tests/persistent_repository_conformance/scenario.rs +++ b/tests/persistent_repository_conformance/scenario.rs @@ -187,11 +187,14 @@ where read_model_plans: Vec::new(), snapshots: vec![AsyncSnapshotWrite::Save { identity: checkout_identity.clone(), - record: SnapshotRecord { - aggregate_id: checkout_id, - version: 1, - data: vec![7], - }, + record: SnapshotRecord::new( + CheckoutSaga::aggregate_type(), + checkout_id, + 1, + "CheckoutSnapshot", + 1, + vec![7], + ), }], }) .await @@ -327,21 +330,27 @@ where repo.save_snapshot_async( &seat_identity, - SnapshotRecord { - aggregate_id: id.clone(), - version: 1, - data: vec![1], - }, + SnapshotRecord::new( + Seat::aggregate_type(), + id.clone(), + 1, + "SeatSnapshot", + 1, + vec![1], + ), ) .await .expect("seat snapshot should save"); repo.save_snapshot_async( &checkout_identity, - SnapshotRecord { - aggregate_id: id, - version: 2, - data: vec![2], - }, + SnapshotRecord::new( + CheckoutSaga::aggregate_type(), + id, + 2, + "CheckoutSnapshot", + 1, + vec![2], + ), ) .await .expect("checkout snapshot should save"); @@ -358,9 +367,16 @@ where .expect("checkout snapshot should exist"); assert_eq!(loaded_seat.version, 1); - assert_eq!(loaded_seat.data, vec![1]); + assert_eq!(loaded_seat.aggregate_type, Seat::aggregate_type()); + assert_eq!(loaded_seat.snapshot_type, "SeatSnapshot"); + assert_eq!(loaded_seat.payload, vec![1]); assert_eq!(loaded_checkout.version, 2); - assert_eq!(loaded_checkout.data, vec![2]); + assert_eq!( + loaded_checkout.aggregate_type, + CheckoutSaga::aggregate_type() + ); + assert_eq!(loaded_checkout.snapshot_type, "CheckoutSnapshot"); + assert_eq!(loaded_checkout.payload, vec![2]); } async fn add_seat( diff --git a/tests/postgres_repository/main.rs b/tests/postgres_repository/main.rs index 5168a084..38eb421d 100644 --- a/tests/postgres_repository/main.rs +++ b/tests/postgres_repository/main.rs @@ -205,11 +205,14 @@ async fn optimistic_conflict_rolls_back_other_stream_and_snapshot() { read_model_plans: Vec::new(), snapshots: vec![sourced_rust::AsyncSnapshotWrite::Save { identity: other_identity.clone(), - record: SnapshotRecord { - aggregate_id: other_id.clone(), - version: 1, - data: vec![1], - }, + record: SnapshotRecord::new( + CounterProjection::aggregate_type(), + other_id.clone(), + 1, + "CounterProjectionSnapshot", + 1, + vec![1], + ), }], }) .await @@ -294,21 +297,27 @@ async fn snapshots_persist_by_full_stream_identity() { repo.save_snapshot_async( &counter, - SnapshotRecord { - aggregate_id: id.clone(), - version: 1, - data: vec![1], - }, + SnapshotRecord::new( + "postgres.counter", + id.clone(), + 1, + "CounterSnapshot", + 1, + vec![1], + ), ) .await .unwrap(); repo.save_snapshot_async( &projection, - SnapshotRecord { - aggregate_id: id, - version: 2, - data: vec![2], - }, + SnapshotRecord::new( + "postgres.counter_projection", + id, + 2, + "ProjectionSnapshot", + 1, + vec![2], + ), ) .await .unwrap(); @@ -317,9 +326,16 @@ async fn snapshots_persist_by_full_stream_identity() { let loaded_projection = repo.get_snapshot_async(&projection).await.unwrap().unwrap(); assert_eq!(loaded_counter.version, 1); - assert_eq!(loaded_counter.data, vec![1]); + assert_eq!(loaded_counter.aggregate_type, "postgres.counter"); + assert_eq!(loaded_counter.snapshot_type, "CounterSnapshot"); + assert_eq!(loaded_counter.payload, vec![1]); assert_eq!(loaded_projection.version, 2); - assert_eq!(loaded_projection.data, vec![2]); + assert_eq!( + loaded_projection.aggregate_type, + "postgres.counter_projection" + ); + assert_eq!(loaded_projection.snapshot_type, "ProjectionSnapshot"); + assert_eq!(loaded_projection.payload, vec![2]); } #[tokio::test] diff --git a/tests/snapshots/main.rs b/tests/snapshots/main.rs index 1a41e4ea..c8956fb2 100644 --- a/tests/snapshots/main.rs +++ b/tests/snapshots/main.rs @@ -1,7 +1,7 @@ mod aggregate; use aggregate::Todo; -use sourced_rust::{AggregateBuilder, HashMapRepository, Queueable, SnapshotStore}; +use sourced_rust::{Aggregate, AggregateBuilder, HashMapRepository, Queueable, SnapshotStore}; #[test] fn snapshot_created_at_frequency_threshold() { @@ -27,6 +27,9 @@ fn snapshot_created_at_frequency_threshold() { assert!(snap.is_some()); let snap = snap.unwrap(); assert_eq!(snap.version, 2); + assert_eq!(snap.aggregate_type, Todo::aggregate_type()); + assert!(snap.snapshot_type.ends_with("TodoSnapshot")); + assert_eq!(snap.payload_codec, sourced_rust::BITCODE_PAYLOAD_CODEC); // Reload and verify state let loaded = repo.get("t1").unwrap().unwrap(); diff --git a/tests/sqlite_repository/main.rs b/tests/sqlite_repository/main.rs index 2c23db08..93fa2620 100644 --- a/tests/sqlite_repository/main.rs +++ b/tests/sqlite_repository/main.rs @@ -259,21 +259,27 @@ async fn snapshots_persist_by_full_stream_identity() { repo.save_snapshot_async( &counter, - SnapshotRecord { - aggregate_id: "same-id".into(), - version: 1, - data: vec![1], - }, + SnapshotRecord::new( + "sqlite.counter", + "same-id", + 1, + "CounterSnapshot", + 1, + vec![1], + ), ) .await .unwrap(); repo.save_snapshot_async( &projection, - SnapshotRecord { - aggregate_id: "same-id".into(), - version: 2, - data: vec![2], - }, + SnapshotRecord::new( + "sqlite.counter_projection", + "same-id", + 2, + "ProjectionSnapshot", + 1, + vec![2], + ), ) .await .unwrap(); @@ -282,9 +288,16 @@ async fn snapshots_persist_by_full_stream_identity() { let loaded_projection = repo.get_snapshot_async(&projection).await.unwrap().unwrap(); assert_eq!(loaded_counter.version, 1); - assert_eq!(loaded_counter.data, vec![1]); + assert_eq!(loaded_counter.aggregate_type, "sqlite.counter"); + assert_eq!(loaded_counter.snapshot_type, "CounterSnapshot"); + assert_eq!(loaded_counter.payload, vec![1]); assert_eq!(loaded_projection.version, 2); - assert_eq!(loaded_projection.data, vec![2]); + assert_eq!( + loaded_projection.aggregate_type, + "sqlite.counter_projection" + ); + assert_eq!(loaded_projection.snapshot_type, "ProjectionSnapshot"); + assert_eq!(loaded_projection.payload, vec![2]); } #[tokio::test] diff --git a/tests/upcasting/main.rs b/tests/upcasting/main.rs index ab9297e9..96762fef 100644 --- a/tests/upcasting/main.rs +++ b/tests/upcasting/main.rs @@ -381,10 +381,13 @@ fn snapshot_repo_with_v1_events_upcasted_on_hydrate() { #[test] fn hydrate_from_snapshot_returns_replay_error_when_post_snapshot_upcaster_decode_fails() { - let snapshot = SnapshotRecord { - aggregate_id: "t1".to_string(), - version: 1, - data: bitcode::serialize(&aggregate::TodoV2Snapshot { + let snapshot = SnapshotRecord::new( + TodoV2::aggregate_type(), + "t1", + 1, + std::any::type_name::(), + 1, + bitcode::serialize(&aggregate::TodoV2Snapshot { id: "t1".to_string(), user_id: "iris".to_string(), task: "Plan".to_string(), @@ -392,7 +395,7 @@ fn hydrate_from_snapshot_returns_replay_error_when_post_snapshot_upcaster_decode completed: false, }) .unwrap(), - }; + ); let mut invalid_event = EventRecord::new("Initialized", vec![0xff], 2); invalid_event.sequence = 2; From 44e8ca45663c7b826b86fb900da2ba5c299d1bdc Mon Sep 17 00:00:00 2001 From: Patrick Lee Scott Date: Tue, 26 May 2026 23:41:28 -0500 Subject: [PATCH 2/3] test: compare snapshot hydration with full replay --- tests/snapshots/main.rs | 116 +++++++++++++++++++++++++++++++++++++++- 1 file changed, 115 insertions(+), 1 deletion(-) diff --git a/tests/snapshots/main.rs b/tests/snapshots/main.rs index c8956fb2..a0697170 100644 --- a/tests/snapshots/main.rs +++ b/tests/snapshots/main.rs @@ -1,7 +1,63 @@ mod aggregate; use aggregate::Todo; -use sourced_rust::{Aggregate, AggregateBuilder, HashMapRepository, Queueable, SnapshotStore}; +use serde::{Deserialize, Serialize}; +use sourced_rust::{ + impl_aggregate, Aggregate, AggregateBuilder, Entity, EventRecord, HashMapRepository, Queueable, + SnapshotRecord, SnapshotStore, Snapshottable, +}; + +#[derive(Default)] +struct ReplayCounter { + entity: Entity, + total: i32, +} + +impl ReplayCounter { + fn add(&mut self, id: &str, amount: i32) { + if self.entity.id().is_empty() { + self.entity.set_id(id); + } + self.entity.digest("Added", &amount).unwrap(); + self.total += amount; + } + + fn replay(&mut self, event: &EventRecord) -> Result<(), String> { + if event.event_name == "Added" { + self.total += event.decode::().map_err(|err| err.to_string())?; + } + Ok(()) + } +} + +impl_aggregate!( + ReplayCounter, + entity, + replay, + aggregate_type = "snapshot.replay_counter" +); + +#[derive(Serialize, Deserialize)] +struct ReplayCounterSnapshot { + id: String, + total: i32, +} + +impl Snapshottable for ReplayCounter { + type Snapshot = ReplayCounterSnapshot; + + fn create_snapshot(&self) -> Self::Snapshot { + ReplayCounterSnapshot { + id: self.entity.id().to_string(), + total: self.total, + } + } + + fn restore_from_snapshot(&mut self, snapshot: Self::Snapshot) { + self.entity.set_id(&snapshot.id); + self.total = snapshot.total; + } +} #[test] fn snapshot_created_at_frequency_threshold() { @@ -110,6 +166,64 @@ fn snapshot_plus_newer_events() { assert_eq!(loaded.entity.snapshot_version(), 2); } +#[test] +fn snapshot_hydration_replays_every_event_after_snapshot_version() { + let base_repo = HashMapRepository::new(); + let full_replay_repo = base_repo.clone().aggregate::(); + let snapshot_repo = base_repo + .clone() + .aggregate::() + .with_snapshots(100); + + let mut counter = ReplayCounter::default(); + counter.add("counter-1", 10); + full_replay_repo.commit(&mut counter).unwrap(); + + let payload = bitcode::serialize(&ReplayCounterSnapshot { + id: "counter-1".into(), + total: 10, + }) + .unwrap(); + base_repo + .save_snapshot(SnapshotRecord::new( + ReplayCounter::aggregate_type(), + "counter-1", + 1, + std::any::type_name::(), + 1, + payload, + )) + .unwrap(); + + let mut counter = snapshot_repo.get("counter-1").unwrap().unwrap(); + counter.add("counter-1", 5); + snapshot_repo.commit(&mut counter).unwrap(); + + let mut counter = snapshot_repo.get("counter-1").unwrap().unwrap(); + counter.add("counter-1", 7); + snapshot_repo.commit(&mut counter).unwrap(); + + let loaded = snapshot_repo.get("counter-1").unwrap().unwrap(); + let replayed = full_replay_repo.get("counter-1").unwrap().unwrap(); + + assert_eq!(loaded.total, 22); + assert_eq!(loaded.total, replayed.total); + assert_eq!(loaded.entity.version(), replayed.entity.version()); + assert_eq!(loaded.entity.snapshot_version(), 1); + assert_eq!(replayed.entity.snapshot_version(), 0); + assert_eq!(loaded.entity.events().len(), 3); + assert_eq!(loaded.entity.events(), replayed.entity.events()); + assert_eq!( + loaded + .entity + .events() + .iter() + .map(|event| event.sequence) + .collect::>(), + vec![1, 2, 3] + ); +} + #[test] fn no_snapshot_falls_back_to_full_replay() { let repo = HashMapRepository::new() From 317f15f7c4b6277eef17eaca3cbf17d06bcb7d7f Mon Sep 17 00:00:00 2001 From: Patrick Lee Scott Date: Tue, 26 May 2026 23:55:35 -0500 Subject: [PATCH 3/3] fix: address snapshot review comments --- README.md | 4 +- docs/postgres-event-store.md | 6 +- src/snapshot/repository.rs | 174 ++++++++++++++++++++++++++------- tests/async_repository/main.rs | 15 ++- tests/upcasting/main.rs | 2 +- 5 files changed, 148 insertions(+), 53 deletions(-) diff --git a/README.md b/README.md index e83943bb..5ed093ba 100644 --- a/README.md +++ b/README.md @@ -1259,7 +1259,7 @@ See [`docs/read-models.md`](docs/read-models.md) for the full guide, including r ## Snapshots -As aggregates accumulate events, replaying from scratch gets expensive. Distributed keeps aggregate events as the durable source of truth and stores repository snapshots as a rebuildable hydration cache. A snapshot cache record can be deleted and rebuilt from events without changing aggregate correctness. +As aggregates accumulate events, replaying from scratch gets expensive. The framework keeps aggregate events as the durable source of truth and stores repository snapshots as a rebuildable hydration cache. A snapshot cache record can be deleted and rebuilt from events without changing aggregate correctness. ### Making an Aggregate Snapshottable @@ -1355,7 +1355,7 @@ let Some(todo) = repo.get("todo-1")? else { ### How It Works -- **On commit**: If `entity.version() >= snapshot_version + frequency`, the aggregate's state is serialized via `create_snapshot()` and saved to the snapshot store. +- **On commit**: If `entity.version().saturating_sub(snapshot_version) >= frequency`, the aggregate's state is serialized via `create_snapshot()` and saved to the snapshot store. - **On load**: If a usable snapshot cache record exists, the aggregate is restored from its payload and only events with `sequence > snapshot.version` are replayed. If no snapshot exists or the cache record is incompatible, full replay is used as a fallback. - **Storage**: Snapshot cache records are stored separately from the event stream. They carry aggregate type, aggregate ID, covered event version, snapshot payload type/version, payload codec metadata, cache metadata, and timestamp. `HashMapRepository` embeds an `InMemorySnapshotStore`; durable async backends implement `AsyncSnapshotStore`. diff --git a/docs/postgres-event-store.md b/docs/postgres-event-store.md index 78520477..6ce2a381 100644 --- a/docs/postgres-event-store.md +++ b/docs/postgres-event-store.md @@ -174,9 +174,9 @@ record, then replay event rows where `sequence > snapshot.version` ordered ascending. If no usable snapshot exists, hydrate from sequence `1`. If the newest snapshot version exceeds the current maximum event sequence for -the stream, the implementation should reject the load with -`RepositoryError::Model`. That fail-fast behavior is preferred over continuing -from an impossible snapshot tail because it surfaces data corruption early. +the stream, the implementation should ignore that cache record and hydrate from +sequence `1`. Snapshot cache fallback should be observable when tracing exists, +but it should not turn a recoverable cache miss into command failure. Snapshot retention is implementation-specific but must be explicit. The current SQL adapters retain only the latest cache record per stream. Future adapters may diff --git a/src/snapshot/repository.rs b/src/snapshot/repository.rs index 2724f266..5931c978 100644 --- a/src/snapshot/repository.rs +++ b/src/snapshot/repository.rs @@ -16,30 +16,69 @@ enum SnapshotHydrationError { Replay(String), } +fn snapshot_hydration_error_to_repository_error(err: SnapshotHydrationError) -> RepositoryError { + match err { + SnapshotHydrationError::Cache(message) | SnapshotHydrationError::Replay(message) => { + RepositoryError::Replay(message) + } + } +} + +fn snapshot_due(version: u64, snapshot_version: u64, frequency: u64) -> bool { + version.saturating_sub(snapshot_version) >= frequency +} + /// Hydrate an aggregate from a snapshot cache record, replaying only events /// after the snapshot version. pub fn hydrate_from_snapshot( entity: Entity, snapshot: SnapshotRecord, ) -> Result { - try_hydrate_from_snapshot::(entity, snapshot).map_err(|err| match err { - SnapshotHydrationError::Cache(message) | SnapshotHydrationError::Replay(message) => { - RepositoryError::Replay(message) - } - }) + let snapshot_payload = prepare_snapshot::(&entity, &snapshot) + .map_err(snapshot_hydration_error_to_repository_error)?; + hydrate_prepared_snapshot::(entity, &snapshot, snapshot_payload) + .map_err(snapshot_hydration_error_to_repository_error) } -fn try_hydrate_from_snapshot( - entity: Entity, - snapshot: SnapshotRecord, -) -> Result { - let mut agg = A::new_empty(); - *agg.entity_mut() = entity; +fn entity_stream_version(entity: &Entity) -> u64 { + entity + .events() + .iter() + .map(|event| event.sequence) + .max() + .unwrap_or_else(|| entity.version()) +} - // Set snapshot_version so frequency check works on next commit - agg.entity_mut().set_snapshot_version(snapshot.version); +fn validate_snapshot_for_entity( + entity: &Entity, + snapshot: &SnapshotRecord, +) -> Result<(), SnapshotHydrationError> { + if snapshot.aggregate_id != entity.id() || snapshot.aggregate_type != A::aggregate_type() { + return Err(SnapshotHydrationError::Cache(format!( + "snapshot cache identity {}:{} does not match aggregate {}:{}", + snapshot.aggregate_type, + snapshot.aggregate_id, + A::aggregate_type(), + entity.id() + ))); + } - // Restore aggregate state from snapshot + let stream_version = entity_stream_version(entity); + if snapshot.version > stream_version { + return Err(SnapshotHydrationError::Cache(format!( + "snapshot cache version {} exceeds stream version {} for {}:{}", + snapshot.version, stream_version, snapshot.aggregate_type, snapshot.aggregate_id + ))); + } + + Ok(()) +} + +fn prepare_snapshot( + entity: &Entity, + snapshot: &SnapshotRecord, +) -> Result { + validate_snapshot_for_entity::(entity, snapshot)?; if !snapshot.has_supported_payload_codec() { return Err(SnapshotHydrationError::Cache(format!( "unsupported snapshot payload codec `{}` version {}", @@ -47,9 +86,23 @@ fn try_hydrate_from_snapshot( ))); } - let snap: A::Snapshot = bitcode::deserialize(&snapshot.payload) - .map_err(|e| SnapshotHydrationError::Cache(format!("snapshot deserialize: {e}")))?; - agg.restore_from_snapshot(snap); + bitcode::deserialize(&snapshot.payload) + .map_err(|e| SnapshotHydrationError::Cache(format!("snapshot deserialize: {e}"))) +} + +fn hydrate_prepared_snapshot( + entity: Entity, + snapshot: &SnapshotRecord, + snapshot_payload: A::Snapshot, +) -> Result { + let mut agg = A::new_empty(); + *agg.entity_mut() = entity; + + // Set snapshot_version so frequency check works on next commit + agg.entity_mut().set_snapshot_version(snapshot.version); + + // Restore aggregate state from snapshot + agg.restore_from_snapshot(snapshot_payload); // Replay only events AFTER the snapshot let post_snapshot: Vec = agg @@ -106,25 +159,16 @@ fn hydrate_with_optional_snapshot( return hydrate::(entity); }; - if snapshot.aggregate_id != entity.id() || snapshot.aggregate_type != A::aggregate_type() { - return hydrate::(entity); - } - - if snapshot.version > entity.version() { - return Err(RepositoryError::Model(format!( - "snapshot cache version {} exceeds stream version {} for {}:{}", - snapshot.version, - entity.version(), - snapshot.aggregate_type, - snapshot.aggregate_id - ))); - } + let snapshot_payload = match prepare_snapshot::(&entity, &snapshot) { + Ok(snapshot_payload) => snapshot_payload, + Err(SnapshotHydrationError::Cache(_)) => return hydrate::(entity), + Err(SnapshotHydrationError::Replay(message)) => { + return Err(RepositoryError::Replay(message)) + } + }; - match try_hydrate_from_snapshot::(entity.clone(), snapshot) { - Ok(aggregate) => Ok(aggregate), - Err(SnapshotHydrationError::Cache(_)) => hydrate::(entity), - Err(SnapshotHydrationError::Replay(message)) => Err(RepositoryError::Replay(message)), - } + hydrate_prepared_snapshot::(entity, &snapshot, snapshot_payload) + .map_err(snapshot_hydration_error_to_repository_error) } /// A repository wrapper that provides snapshot-aware get and commit for a specific aggregate type. @@ -278,7 +322,7 @@ where let version = aggregate.entity().version(); let snap_version = aggregate.entity().snapshot_version(); - if version >= snap_version + self.frequency { + if snapshot_due(version, snap_version, self.frequency) { return snapshot_record_for(aggregate).map(Some); } Ok(None) @@ -387,7 +431,7 @@ where let version = aggregate.entity().version(); let snap_version = aggregate.entity().snapshot_version(); - if version >= snap_version + self.frequency { + if snapshot_due(version, snap_version, self.frequency) { return snapshot_record_for(aggregate).map(Some); } Ok(None) @@ -498,7 +542,7 @@ where #[cfg(test)] mod tests { use super::*; - use crate::{impl_aggregate, AggregateRepository, Entity, EventRecord}; + use crate::{impl_aggregate, Aggregate, AggregateRepository, Entity, EventRecord}; use std::cell::RefCell; #[derive(Default)] @@ -571,4 +615,58 @@ mod tests { assert_eq!(aggregate.entity.snapshot_version(), 0); assert_eq!(aggregate.entity.new_events().len(), 1); } + + #[test] + fn snapshot_due_uses_saturating_version_distance() { + assert!(snapshot_due(5, 2, 3)); + assert!(!snapshot_due(5, 3, 3)); + assert!(!snapshot_due(0, u64::MAX, 1)); + assert!(snapshot_due(u64::MAX, u64::MAX - 1, 1)); + } + + #[test] + fn hydrate_from_snapshot_rejects_identity_mismatch() { + let mut entity = Entity::with_id("snap-1"); + entity.load_from_history(vec![EventRecord::new("Touched", vec![], 1)]); + let snapshot = SnapshotRecord::new( + TestAggregate::aggregate_type(), + "other", + 1, + std::any::type_name::(), + 1, + bitcode::serialize(&1_u32).unwrap(), + ); + + let err = match hydrate_from_snapshot::(entity, snapshot) { + Err(err) => err, + Ok(_) => panic!("expected identity mismatch error"), + }; + + assert!( + matches!(err, RepositoryError::Replay(message) if message.contains("does not match")) + ); + } + + #[test] + fn hydrate_from_snapshot_rejects_snapshot_ahead_of_stream() { + let mut entity = Entity::with_id("snap-1"); + entity.load_from_history(vec![EventRecord::new("Touched", vec![], 1)]); + let snapshot = SnapshotRecord::new( + TestAggregate::aggregate_type(), + "snap-1", + 2, + std::any::type_name::(), + 1, + bitcode::serialize(&1_u32).unwrap(), + ); + + let err = match hydrate_from_snapshot::(entity, snapshot) { + Err(err) => err, + Ok(_) => panic!("expected future snapshot error"), + }; + + assert!( + matches!(err, RepositoryError::Replay(message) if message.contains("exceeds stream version")) + ); + } } diff --git a/tests/async_repository/main.rs b/tests/async_repository/main.rs index d7af3056..c3be416a 100644 --- a/tests/async_repository/main.rs +++ b/tests/async_repository/main.rs @@ -301,7 +301,7 @@ async fn async_snapshot_repository_ignores_invalid_cache_and_replays_events() { } #[tokio::test] -async fn async_snapshot_repository_rejects_cache_past_stream_version() { +async fn async_snapshot_repository_ignores_cache_past_stream_version_and_replays_events() { let repo = HashMapRepository::new(); let aggregate_repo = repo.clone().async_aggregate::(); let snapshot_repo = repo @@ -321,17 +321,14 @@ async fn async_snapshot_repository_rejects_cache_past_stream_version() { 2, std::any::type_name::(), 1, - bitcode::serialize(&4_i32).unwrap(), + bitcode::serialize(&999_i32).unwrap(), ); repo.save_snapshot_async(&identity, record).await.unwrap(); - let err = match snapshot_repo.get(id).await { - Err(err) => err, - Ok(_) => panic!("snapshot cache past stream version should fail"), - }; - assert!( - matches!(err, RepositoryError::Model(message) if message.contains("exceeds stream version")) - ); + let loaded = snapshot_repo.get(id).await.unwrap().unwrap(); + assert_eq!(loaded.value, 4); + assert_eq!(loaded.entity().version(), 1); + assert_eq!(loaded.entity().snapshot_version(), 0); } #[tokio::test] diff --git a/tests/upcasting/main.rs b/tests/upcasting/main.rs index 96762fef..196c4ed4 100644 --- a/tests/upcasting/main.rs +++ b/tests/upcasting/main.rs @@ -399,7 +399,7 @@ fn hydrate_from_snapshot_returns_replay_error_when_post_snapshot_upcaster_decode let mut invalid_event = EventRecord::new("Initialized", vec![0xff], 2); invalid_event.sequence = 2; - let mut entity = Entity::new(); + let mut entity = Entity::with_id("t1"); entity.load_from_history(vec![invalid_event]); match hydrate_from_snapshot::(entity, snapshot) {