Implement aggregate snapshot cache storage

README.md

@@ -156,8 +156,8 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
 - **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. 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
 
-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};
@@ -1355,9 +1355,9 @@ 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 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`.
 
 ## Event Upcasting / Versioning
 

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.

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.
+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
+retain last `N` or time-based cache records, but they must never prune aggregate
+events.
 
 ## Commit Semantics
 

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 (

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 (

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<T: serde::Serialize>(&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();

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

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)

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<SnapshotRecord, RepositoryError> {
+    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<SnapshotRecord, RepositoryError> {
                 .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))?,
+        )?,
     })
 }