JournaledGrain snapshotting: replay only post-snapshot events (#144)

samples/Persistence/Bank.Grains/BankStateCopiers.cs

@@ -4,24 +4,27 @@
 namespace Bank.Grains;
 
 /// <summary>
-///     Registers the deep copiers that the storage providers use to snapshot durable state.
+///     Registers the deep copiers that storage providers and the <c>ISnapshotStore</c> use to
+///     snapshot durable/journaled state.
 ///     <para>
 ///         The code generator emits an internal <c>{StateType}Copier</c> for every
 ///         <c>[GenerateSerializer]</c> type. They are <c>internal</c> to the assembly that declares
-///         the state, so this helper — living in the same assembly — wires them into DI. State that
-///         is never written through <c>IGrainStorage</c> (e.g. the event-sourced
-///         <see cref="LedgerState" />) does not need a copier.
+///         the state, so this helper — living in the same assembly — wires them into DI. This
+///         includes <see cref="LedgerState" />, whose copier lets the in-memory
+///         <c>ISnapshotStore</c> snapshot the event-sourced ledger projection.
 ///     </para>
 /// </summary>
 public static class BankStateCopiers
 {
-    /// <summary>Registers <c>IDeepCopier&lt;T&gt;</c> for every storage-backed Bank state type.</summary>
+    /// <summary>Registers <c>IDeepCopier&lt;T&gt;</c> for every storage-backed or snapshotted Bank state type.</summary>
     public static IServiceCollection AddBankStateCopiers(this IServiceCollection services)
     {
         services.AddSingleton<IDeepCopier<AccountState>>(
             sp => new AccountStateCopier(sp.GetRequiredService<ICopierProvider>()));
         services.AddSingleton<IDeepCopier<ProfileState>>(
             sp => new ProfileStateCopier(sp.GetRequiredService<ICopierProvider>()));
+        services.AddSingleton<IDeepCopier<LedgerState>>(
+            sp => new LedgerStateCopier(sp.GetRequiredService<ICopierProvider>()));
         return services;
     }
 }

samples/Persistence/Bank.Grains/LedgerBehavior.cs

@@ -17,12 +17,16 @@ namespace Bank.Grains;
 public sealed class LedgerBehavior : JournaledGrain<LedgerState, LedgerEvent>, ILedgerGrain
 {
     // The code generator registers IActivationMemory<JournaledGrainState<LedgerState, LedgerEvent>>
-    // for this constructor; ICallContext and ILogStorage come from the runtime / DI.
+    // for this constructor; ICallContext, ILogStorage, and ISnapshotStore come from the runtime / DI.
     public LedgerBehavior(
         IActivationMemory<JournaledGrainState<LedgerState, LedgerEvent>> memory,
         ICallContext ctx,
-        ILogStorage? log = null)
-        : base(memory, ctx, log) { }
+        ILogStorage? log = null,
+        ISnapshotStore? snapshot = null)
+        : base(memory, ctx, log, snapshot) { }
+
+    // Snapshot every 5 confirmed events so long-lived ledgers don't replay their whole history.
+    protected override int SnapshotInterval => 5;
 
     // Pure function: how each event mutates the projection. Used for both live updates and replay.
     protected override void TransitionState(LedgerState state, LedgerEvent @event)

samples/Persistence/Bank.Grains/LedgerState.cs

@@ -1,14 +1,17 @@
+using Quark.Serialization.Abstractions.Attributes;
+
 namespace Bank.Grains;
 
 /// <summary>
-///     In-memory projection for <see cref="LedgerBehavior" />. This is never persisted directly —
-///     it is rebuilt by replaying <see cref="LedgerEvent" />s from the log. No serializer is needed
-///     because the projection lives only in the activation shell.
+///     Projection for <see cref="LedgerBehavior" />, rebuilt by replaying <see cref="LedgerEvent" />s.
+///     <c>[GenerateSerializer]</c> lets the in-memory <c>ISnapshotStore</c> deep-copy it so activation
+///     can replay only post-snapshot events instead of the whole log.
 /// </summary>
+[GenerateSerializer]
 public sealed class LedgerState
 {
-    public decimal Balance { get; set; }
-    public List<string> History { get; } = [];
+    [Id(0)] public decimal Balance { get; set; }
+    [Id(1)] public List<string> History { get; set; } = [];
 }
 
 /// <summary>Base type for ledger events. Events are the source of truth, persisted to the log.</summary>

samples/Persistence/Bank.Server/Program.cs

@@ -25,6 +25,9 @@
         // Event log — backs the JournaledGrain ledger.
         silo.Services.AddSingleton<ILogStorage, InMemoryLogStorage>();
 
+        // Snapshot store — lets the JournaledGrain ledger replay only post-snapshot events.
+        silo.Services.AddInMemorySnapshotStore();
+
         // Deep copiers for the storage-backed state types ([GenerateSerializer]).
         silo.Services.AddBankStateCopiers();
 

src/Quark.Persistence.Abstractions/Journaling/ISnapshotStore.cs

@@ -0,0 +1,68 @@
+using Quark.Core.Abstractions.Identity;
+
+namespace Quark.Persistence.Abstractions.Journaling;
+
+/// <summary>
+///     Optional snapshot store for <see cref="JournaledGrain{TState,TEvent}" />. A snapshot is a
+///     replay-shortcut only; the event log remains the source of truth. A missing snapshot is
+///     normal (activation full-replays). A present-but-corrupt snapshot must surface as a
+///     <see cref="CorruptSnapshotException" /> rather than silently producing wrong state.
+/// </summary>
+public interface ISnapshotStore
+{
+    /// <summary>
+    ///     Reads the latest snapshot for <paramref name="grainId" />, or <c>null</c> if none exists.
+    ///     Durable providers throw <see cref="CorruptSnapshotException" /> when a stored snapshot
+    ///     cannot be deserialized into <typeparamref name="TState" />.
+    ///     Implementations MUST return a deep/isolated copy of the state: the caller assigns the
+    ///     returned <see cref="SnapshotEnvelope{TState}.State" /> directly into activation state and
+    ///     mutates it in place while replaying post-snapshot events.
+    /// </summary>
+    Task<SnapshotEnvelope<TState>?> ReadSnapshotAsync<TState>(
+        GrainId grainId, CancellationToken ct = default) where TState : class;
+
+    /// <summary>
+    ///     Writes (replaces) the snapshot for <paramref name="grainId" />. Implementations MUST store
+    ///     an isolated copy of <see cref="SnapshotEnvelope{TState}.State" />: the caller may continue
+    ///     mutating its own state object after this call returns.
+    /// </summary>
+    Task WriteSnapshotAsync<TState>(
+        GrainId grainId, SnapshotEnvelope<TState> snapshot, CancellationToken ct = default)
+        where TState : class;
+
+    /// <summary>Deletes any stored snapshot for <paramref name="grainId" /> (recovery path).</summary>
+    Task ClearSnapshotAsync(GrainId grainId, CancellationToken ct = default);
+}
+
+/// <summary>A point-in-time projection of grain state and the log version it folds up to.</summary>
+public sealed class SnapshotEnvelope<TState>
+{
+    public SnapshotEnvelope(int version, TState state)
+    {
+        Version = version;
+        State = state;
+    }
+
+    /// <summary>Number of events folded into <see cref="State" /> — i.e. the index of the next event.</summary>
+    public int Version { get; }
+
+    /// <summary>State after applying events <c>[0, Version)</c>.</summary>
+    public TState State { get; }
+}
+
+/// <summary>Thrown when a present snapshot is unusable (undeserializable or inconsistent with the log).</summary>
+public sealed class CorruptSnapshotException : Exception
+{
+    public CorruptSnapshotException(GrainId grainId, int snapshotVersion, string message, Exception? inner = null)
+        : base(message, inner)
+    {
+        GrainId = grainId;
+        SnapshotVersion = snapshotVersion;
+    }
+
+    /// <summary>The grain whose snapshot is corrupt.</summary>
+    public GrainId GrainId { get; }
+
+    /// <summary>The version stamped on the offending snapshot.</summary>
+    public int SnapshotVersion { get; }
+}

src/Quark.Persistence.Abstractions/Journaling/JournaledGrain.cs

@@ -16,15 +16,18 @@ public abstract class JournaledGrain<TState, TEvent> : IGrainBehavior, IActivati
     private readonly IActivationMemory<JournaledGrainState<TState, TEvent>> _memory;
     private readonly ICallContext _ctx;
     private ILogStorage? _logStorage;
+    private ISnapshotStore? _snapshotStore;
 
     protected JournaledGrain(
         IActivationMemory<JournaledGrainState<TState, TEvent>> memory,
         ICallContext ctx,
-        ILogStorage? logStorage = null)
+        ILogStorage? logStorage = null,
+        ISnapshotStore? snapshotStore = null)
     {
         _memory = memory;
         _ctx = ctx;
         _logStorage = logStorage;
+        _snapshotStore = snapshotStore;
     }
 
     /// <summary>The grain identity for this call.</summary>
@@ -36,6 +39,13 @@ protected JournaledGrain(
     /// <summary>The current in-memory state (includes staged but not-yet-confirmed events).</summary>
     protected TState State => _memory.Value.State;
 
+    /// <summary>
+    ///     Number of confirmed events between automatic snapshots. Override per grain type.
+    ///     <c>0</c> disables automatic snapshotting for this grain type. Default: 100.
+    ///     Automatic snapshots require a registered <see cref="ISnapshotStore" />.
+    /// </summary>
+    protected virtual int SnapshotInterval => 100;
+
     /// <inheritdoc />
     public async Task OnActivateAsync(CancellationToken ct)
     {
@@ -80,6 +90,28 @@ await _logStorage.AppendEntriesAsync(GrainId, st.ConfirmedVersion, entries, canc
             .ConfigureAwait(false);
         st.ConfirmedVersion += st.StagedEvents.Count;
         st.StagedEvents.Clear();
+
+        if (_snapshotStore is not null && SnapshotInterval > 0 &&
+            st.ConfirmedVersion - st.LastSnapshotVersion >= SnapshotInterval)
+        {
+            await WriteSnapshotCoreAsync(cancellationToken).ConfigureAwait(false);
+        }
+    }
+
+    /// <summary>
+    ///     Writes a snapshot of the current confirmed state to the registered
+    ///     <see cref="ISnapshotStore" />. No-op when no snapshot store is registered.
+    /// </summary>
+    protected Task WriteSnapshotAsync(CancellationToken cancellationToken = default) =>
+        _snapshotStore is null ? Task.CompletedTask : WriteSnapshotCoreAsync(cancellationToken);
+
+    private async Task WriteSnapshotCoreAsync(CancellationToken ct)
+    {
+        JournaledGrainState<TState, TEvent> st = _memory.Value;
+        await _snapshotStore!
+            .WriteSnapshotAsync(GrainId, new SnapshotEnvelope<TState>(st.ConfirmedVersion, st.State), ct)
+            .ConfigureAwait(false);
+        st.LastSnapshotVersion = st.ConfirmedVersion;
     }
 
     /// <summary>Retrieves confirmed events in the range [<paramref name="fromVersion"/>, <paramref name="toVersion"/>).</summary>
@@ -96,9 +128,44 @@ await _logStorage.ReadEntriesAsync(GrainId, fromVersion, toVersion, cancellation
     private async Task ReloadFromLogAsync(CancellationToken ct)
     {
         JournaledGrainState<TState, TEvent> st = _memory.Value;
+        st.ConfirmedVersion = 0;
+        st.LastSnapshotVersion = 0;
+
+        if (_snapshotStore is not null)
+        {
+            SnapshotEnvelope<TState>? snap =
+                await _snapshotStore.ReadSnapshotAsync<TState>(GrainId, ct).ConfigureAwait(false);
+
+            if (snap is not null && snap.Version > 0)
+            {
+                // Boundary probe: read from snap.Version - 1 so we can confirm the log actually
+                // contains >= snap.Version contiguous entries WITHOUT a length API on ILogStorage
+                // (AppendEntriesAsync guarantees version == index, so entry[V-1] existing => 0..V-1 exist).
+                IReadOnlyList<LogEntry> tail = await _logStorage!
+                    .ReadEntriesAsync(GrainId, snap.Version - 1, int.MaxValue, ct).ConfigureAwait(false);
+
+                if (tail.Count == 0 || tail[0].Version != snap.Version - 1)
+                    throw new CorruptSnapshotException(GrainId, snap.Version,
+                        $"Snapshot version {snap.Version} is ahead of the event log for grain {GrainId}.");
+
+                st.State = snap.State;                 // store returned an isolated copy
+                st.ConfirmedVersion = snap.Version;
+                st.LastSnapshotVersion = snap.Version;
+
+                for (int i = 1; i < tail.Count; i++)   // skip the boundary entry (already in the snapshot)
+                {
+                    TransitionState(st.State, (TEvent)tail[i].Event);
+                    st.ConfirmedVersion = tail[i].Version + 1;
+                }
+
+                return;
+            }
+        }
+
+        // No usable snapshot: full replay from 0 (original behavior).
+        st.State = new TState();
         IReadOnlyList<LogEntry> all =
             await _logStorage!.ReadEntriesAsync(GrainId, 0, int.MaxValue, ct).ConfigureAwait(false);
-        st.State = new TState();
         foreach (LogEntry entry in all)
         {
             TransitionState(st.State, (TEvent)entry.Event);

src/Quark.Persistence.Abstractions/Journaling/JournaledGrainState.cs

@@ -10,4 +10,7 @@ public sealed class JournaledGrainState<TState, TEvent>
     public TState State { get; set; } = new();
     public List<TEvent> StagedEvents { get; } = [];
     public int ConfirmedVersion { get; set; }
+
+    /// <summary>The <see cref="ConfirmedVersion" /> captured by the most recent snapshot write.</summary>
+    public int LastSnapshotVersion { get; set; }
 }

src/Quark.Persistence.InMemory/InMemorySnapshotStore.cs

@@ -0,0 +1,53 @@
+using System.Collections.Concurrent;
+using Quark.Core.Abstractions.Identity;
+using Quark.Persistence.Abstractions.Journaling;
+using Quark.Serialization.Abstractions.Abstractions;
+
+namespace Quark.Persistence.InMemory;
+
+/// <summary>
+///     In-memory <see cref="ISnapshotStore" /> for development and tests. State is deep-copied on
+///     both write and read to isolate the stored snapshot from the grain's live, still-mutating
+///     state (the same isolation <see cref="InMemoryGrainStorage" /> applies). Not durable across
+///     process restarts, so it never produces the undeserializable-snapshot failure mode.
+/// </summary>
+public sealed class InMemorySnapshotStore : ISnapshotStore
+{
+    private readonly ConcurrentDictionary<GrainId, (int Version, object State)> _snapshots = new();
+    private readonly ICopierProvider _copiers;
+
+    /// <summary>Initializes the in-memory snapshot store.</summary>
+    public InMemorySnapshotStore(ICopierProvider copiers) => _copiers = copiers;
+
+    /// <inheritdoc />
+    public Task WriteSnapshotAsync<TState>(
+        GrainId grainId, SnapshotEnvelope<TState> snapshot, CancellationToken ct = default)
+        where TState : class
+    {
+        ct.ThrowIfCancellationRequested();
+        TState isolated = _copiers.GetRequiredCopier<TState>().DeepCopy(snapshot.State, new CopyContext());
+        _snapshots[grainId] = (snapshot.Version, isolated);
+        return Task.CompletedTask;
+    }
+
+    /// <inheritdoc />
+    public Task<SnapshotEnvelope<TState>?> ReadSnapshotAsync<TState>(
+        GrainId grainId, CancellationToken ct = default)
+        where TState : class
+    {
+        ct.ThrowIfCancellationRequested();
+        if (!_snapshots.TryGetValue(grainId, out (int Version, object State) entry))
+            return Task.FromResult<SnapshotEnvelope<TState>?>(null);
+
+        TState copy = _copiers.GetRequiredCopier<TState>().DeepCopy((TState)entry.State, new CopyContext());
+        return Task.FromResult<SnapshotEnvelope<TState>?>(new SnapshotEnvelope<TState>(entry.Version, copy));
+    }
+
+    /// <inheritdoc />
+    public Task ClearSnapshotAsync(GrainId grainId, CancellationToken ct = default)
+    {
+        ct.ThrowIfCancellationRequested();
+        _snapshots.TryRemove(grainId, out _);
+        return Task.CompletedTask;
+    }
+}

src/Quark.Persistence.InMemory/InMemorySnapshotStoreServiceCollectionExtensions.cs

@@ -0,0 +1,20 @@
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.DependencyInjection.Extensions;
+using Quark.Persistence.Abstractions.Journaling;
+
+namespace Quark.Persistence.InMemory;
+
+/// <summary>Service registration helpers for the in-memory snapshot store.</summary>
+public static class InMemorySnapshotStoreServiceCollectionExtensions
+{
+    /// <summary>
+    ///     Registers the in-memory <see cref="ISnapshotStore" />. Once registered, every
+    ///     <see cref="JournaledGrain{TState,TEvent}" /> with a positive <c>SnapshotInterval</c>
+    ///     writes snapshots and replays only post-snapshot events on activation.
+    /// </summary>
+    public static IServiceCollection AddInMemorySnapshotStore(this IServiceCollection services)
+    {
+        services.TryAddSingleton<ISnapshotStore, InMemorySnapshotStore>();
+        return services;
+    }
+}