Akka.Persistence.Embedded: SQLite journal, snapshot store and read journal (AOT-compatible)

Akka.slnx

@@ -8,6 +8,7 @@
     <Project Path="src/aot/Akka.AOT.App/Akka.AOT.App.csproj" />
     <Project Path="src/aot/Akka.Hosting.AOT.App/Akka.Hosting.AOT.App.csproj" />
     <Project Path="src/aot/Akka.Persistence.AOT.App/Akka.Persistence.AOT.App.csproj" />
+    <Project Path="src/aot/Akka.Persistence.Embedded.AOT.App/Akka.Persistence.Embedded.AOT.App.csproj" />
   </Folder>
   <Folder Name="/Benchmark/">
     <Project Path="src/benchmark/Akka.Benchmarks/Akka.Benchmarks.csproj" />
@@ -83,12 +84,16 @@
     <Project Path="src/contrib/hosting/Akka.Hosting.TestKit/Akka.Hosting.TestKit.csproj" />
     <Project Path="src/contrib/hosting/Akka.Hosting.Tests/Akka.Hosting.Tests.csproj" />
     <Project Path="src/contrib/hosting/Akka.Hosting/Akka.Hosting.csproj" />
+    <Project Path="src/contrib/hosting/Akka.Persistence.Embedded.Hosting.Tests/Akka.Persistence.Embedded.Hosting.Tests.csproj" />
+    <Project Path="src/contrib/hosting/Akka.Persistence.Embedded.Hosting/Akka.Persistence.Embedded.Hosting.csproj" />
     <Project Path="src/contrib/hosting/Akka.Persistence.Hosting.Tests/Akka.Persistence.Hosting.Tests.csproj" />
     <Project Path="src/contrib/hosting/Akka.Persistence.Hosting/Akka.Persistence.Hosting.csproj" />
     <Project Path="src/contrib/hosting/Akka.Remote.Hosting.Tests/Akka.Remote.Hosting.Tests.csproj" />
     <Project Path="src/contrib/hosting/Akka.Remote.Hosting/Akka.Remote.Hosting.csproj" />
   </Folder>
   <Folder Name="/Contrib/Persistence/">
+    <Project Path="src/contrib/persistence/Akka.Persistence.Embedded.Tests/Akka.Persistence.Embedded.Tests.csproj" />
+    <Project Path="src/contrib/persistence/Akka.Persistence.Embedded/Akka.Persistence.Embedded.csproj" />
     <Project Path="src/contrib/persistence/Akka.Persistence.Query.InMemory.Tests/Akka.Persistence.Query.InMemory.Tests.csproj" />
     <Project Path="src/contrib/persistence/Akka.Persistence.Query.InMemory/Akka.Persistence.Query.InMemory.csproj" />
   </Folder>

Directory.Build.props

@@ -39,6 +39,7 @@
     <GrpcToolsVersion>2.82.0</GrpcToolsVersion>
     <BenchmarkDotNetVersion>0.15.8</BenchmarkDotNetVersion>
     <MessagePackVersion>3.1.8</MessagePackVersion>
+    <MicrosoftDataSqliteVersion>10.0.12</MicrosoftDataSqliteVersion>
     <MicrosoftCodeAnalysisCSharpVersion>4.13.0</MicrosoftCodeAnalysisCSharpVersion>
     <NetTestVersion>net10.0</NetTestVersion>
     <FsharpVersion>10.1.302</FsharpVersion>

build-system/pr-validation.yaml

@@ -309,6 +309,55 @@ jobs:
             --repo-root "$(Build.SourcesDirectory)"
         displayName: 'Compare Persistence AOT warnings against the baseline'
 
+      # --- SQLite persistence canary (src/aot/Akka.Persistence.Embedded.AOT.App): same job, runs after the Persistence canary ---
+      - bash: |
+          rm -rf src/aot/Akka.Persistence.Embedded.AOT.App/bin src/aot/Akka.Persistence.Embedded.AOT.App/obj
+        displayName: 'Clean the SQLite canary intermediate output'
+
+      - bash: |
+          set -o pipefail
+          dotnet publish src/aot/Akka.Persistence.Embedded.AOT.App -r linux-x64 -c Release \
+            -o "$(Agent.TempDirectory)/sqlite-aot-publish" 2>&1 | tee "$(Agent.TempDirectory)/sqlite-aot-publish.log"
+          rc=${PIPESTATUS[0]}
+          if [ "$rc" -ne 0 ]; then
+            echo "##vso[task.logissue type=error]The publish of the SQLite AOT canary failed with exit code $rc."
+            exit 1
+          fi
+          if [ ! -f "$(Agent.TempDirectory)/sqlite-aot-publish/libe_sqlite3.so" ]; then
+            echo "##vso[task.logissue type=error]libe_sqlite3.so was not published next to the canary."
+            exit 1
+          fi
+        displayName: 'Publish the SQLite AOT canary'
+
+      # The canary's log watchdog and its assertions are what make '[canary-sqlite] OK' meaningful.
+      - bash: |
+          set -o pipefail
+          timeout 300 "$(Agent.TempDirectory)/sqlite-aot-publish/Akka.Persistence.Embedded.AOT.App" 2>&1 | tee "$(Agent.TempDirectory)/sqlite-aot-run.log"
+          rc=${PIPESTATUS[0]}
+          if [ "$rc" -eq 124 ]; then
+            echo "##vso[task.logissue type=error]The SQLite AOT canary did not exit within 300s."
+            exit 1
+          fi
+          if [ "$rc" -ne 0 ]; then
+            echo "##vso[task.logissue type=error]The SQLite AOT canary exited with code $rc."
+            exit 1
+          fi
+          if ! grep -qF '[canary-sqlite] OK' "$(Agent.TempDirectory)/sqlite-aot-run.log"; then
+            echo "##vso[task.logissue type=error]The SQLite AOT canary exited 0 but never printed '[canary-sqlite] OK'."
+            exit 1
+          fi
+          echo "SQLite AOT canary: exit code 0 and '[canary-sqlite] OK' on stdout."
+        displayName: 'Run the SQLite AOT canary'
+
+      # Zero warnings from the plugin, its Hosting package and Akka.Streams is the bar; the baseline holds only a header.
+      - bash: |
+          dotnet run scripts/CheckAotWarnings.cs -- \
+            --log "$(Agent.TempDirectory)/sqlite-aot-publish.log" \
+            --baseline src/aot/Akka.Persistence.Embedded.AOT.App/aot-warnings.baseline.txt \
+            --scope "src/contrib/persistence/Akka.Persistence.Embedded/,src/contrib/hosting/Akka.Persistence.Embedded.Hosting/,src/core/Akka.Streams/" \
+            --repo-root "$(Build.SourcesDirectory)"
+        displayName: 'Compare SQLite AOT warnings against the baseline'
+
       # Both publish logs and the run log, so a failure can be diagnosed from the artifact without
       # re-running the job. always() because the interesting case is the one where a step above
       # failed.

src/aot/Akka.Persistence.Embedded.AOT.App/Akka.Persistence.Embedded.AOT.App.csproj

@@ -0,0 +1,52 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <OutputType>Exe</OutputType>
+    <TargetFramework>$(NetTestVersion)</TargetFramework>
+    <ImplicitUsings>enable</ImplicitUsings>
+    <Nullable>enable</Nullable>
+
+    <IsPackable>false</IsPackable>
+
+    <!-- Same rationale as Akka.AOT.App.csproj: this app's job is to print ILC's IL2xxx/IL3xxx
+         list, so ILC's own warnings must not fail the publish. The app's own C# still builds with
+         warnings as errors (repo-wide TreatWarningsAsErrors). -->
+    <IlcTreatWarningsAsErrors>false</IlcTreatWarningsAsErrors>
+  </PropertyGroup>
+
+  <!-- PublishAot is gated on a RuntimeIdentifier, same two reasons as Akka.AOT.App: a plain
+       solution build must not restore a RID-specific ILCompiler, and '-p:PublishAot=true' on the
+       command line is a GLOBAL property that would also turn on trim analysis for the library
+       projects, failing their build under repo-wide TreatWarningsAsErrors. -->
+  <PropertyGroup Label="AotSettings" Condition="'$(RuntimeIdentifier)' != ''">
+    <PublishAot>true</PublishAot>
+  </PropertyGroup>
+
+  <PropertyGroup Label="AotSettings">
+    <TrimmerSingleWarn>false</TrimmerSingleWarn>
+    <InvariantGlobalization>true</InvariantGlobalization>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <!-- the log watchdog shared by the canaries that use Akka's own logger -->
+    <Compile Include="..\Shared\LogWatchdogFilter.cs" Link="LogWatchdogFilter.cs" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <!-- Same as Akka.Hosting.AOT.App: the full Microsoft.Extensions.Hosting package is what an application adds. -->
+    <PackageReference Include="Microsoft.Extensions.Hosting" Version="$(MsExtVersion)" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <!-- Brings Akka.Hosting, Akka.Persistence.Hosting and the plugin. -->
+    <ProjectReference Include="..\..\contrib\hosting\Akka.Persistence.Embedded.Hosting\Akka.Persistence.Embedded.Hosting.csproj"/>
+  </ItemGroup>
+
+  <!-- Turns Akka.DynamicTypeLoading off, same as Akka.AOT.App and for the same reason:
+       buildTransitive/Akka.targets only does this automatically for the NuGet package, not for a
+       ProjectReference. With it off, only what the setups register can start. -->
+  <ItemGroup>
+    <RuntimeHostConfigurationOption Include="Akka.DynamicTypeLoading" Value="false" Trim="true" />
+  </ItemGroup>
+
+</Project>

src/aot/Akka.Persistence.Embedded.AOT.App/CanaryTypes.cs

@@ -0,0 +1,167 @@
+//-----------------------------------------------------------------------
+// <copyright file="CanaryTypes.cs" company="Akka.NET Project">
+//     Copyright (C) 2009-2025 Lightbend Inc. <http://www.lightbend.com>
+//     Copyright (C) 2013-2025 .NET Foundation <https://github.com/akkadotnet/akka.net>
+// </copyright>
+//-----------------------------------------------------------------------
+
+using System.Text;
+using Akka.Actor;
+using Akka.Persistence.Journal;
+using Akka.Serialization;
+
+namespace Akka.Persistence.Embedded.AOT.App;
+
+/// <summary>The event the canary persists.</summary>
+public sealed record CanaryEvent(string Value);
+
+/// <summary>The snapshot the canary saves.</summary>
+public sealed record CanarySnapshot(string[] Values);
+
+/// <summary>Persists one event, tagged "red" when <paramref name="Red"/> is set. Replies with the number of events held.</summary>
+public sealed record PersistCmd(string Value, bool Red);
+
+public sealed record SaveNow;
+
+public sealed record DeleteTo(long SequenceNr);
+
+public sealed record GetState;
+
+public sealed record CanaryState(string[] Values, long SnapshotSequenceNr, long LastSequenceNr);
+
+/// <summary>
+/// Tags every event whose value starts with "adapted-". It is added with the typed <c>AddWriteEventAdapter&lt;T&gt;</c>, which
+/// is what lets core build it with <c>Akka.DynamicTypeLoading</c> off.
+/// </summary>
+public sealed class CanaryTagger : IWriteEventAdapter
+{
+    private static readonly string[] Tags = ["adapted"];
+
+    public string Manifest(object evt) => string.Empty;
+
+    public object ToJournal(object evt) => evt is CanaryEvent { Value: { } value } && value.StartsWith("adapted-", StringComparison.Ordinal)
+        ? new Tagged(evt, Tags)
+        : evt;
+}
+
+/// <summary>
+/// Hand-written, reflection-free serializer for <see cref="CanaryEvent"/> and <see cref="CanarySnapshot"/>.
+/// With <c>Akka.DynamicTypeLoading</c> off core registers no fallback serializer, so an AOT application
+/// binds one for each of its own event and snapshot types through <see cref="SerializationSetup"/>.
+/// </summary>
+public sealed class CanarySerializer : SerializerWithStringManifest
+{
+    private const string EventManifest = "E";
+    private const string SnapshotManifest = "S";
+
+    public CanarySerializer(ExtendedActorSystem system) : base(system)
+    {
+    }
+
+    // set here rather than read from akka.actor.serialization-identifiers, which is keyed by type name
+    public override int Identifier => 91002;
+
+    public override string Manifest(object o) => o switch
+    {
+        CanaryEvent => EventManifest,
+        CanarySnapshot => SnapshotManifest,
+        _ => throw new ArgumentException($"{nameof(CanarySerializer)} cannot serialize [{o.GetType()}]")
+    };
+
+    public override byte[] ToBinary(object obj) => obj switch
+    {
+        CanaryEvent e => Encoding.UTF8.GetBytes(e.Value),
+        CanarySnapshot s => Encoding.UTF8.GetBytes(string.Join('\n', s.Values)),
+        _ => throw new ArgumentException($"{nameof(CanarySerializer)} cannot serialize [{obj.GetType()}]")
+    };
+
+    public override object FromBinary(byte[] bytes, string manifest) => manifest switch
+    {
+        EventManifest => new CanaryEvent(Encoding.UTF8.GetString(bytes)),
+        SnapshotManifest => new CanarySnapshot(Encoding.UTF8.GetString(bytes).Split('\n', StringSplitOptions.RemoveEmptyEntries)),
+        _ => throw new System.Runtime.Serialization.SerializationException($"Unknown manifest [{manifest}]")
+    };
+}
+
+/// <summary>
+/// Persists <see cref="CanaryEvent"/>s one at a time, saves snapshots, deletes events and reports
+/// what it recovered. <paramref name="useSnapshots"/> = false recovers from events only.
+/// </summary>
+public sealed class CanaryPersistentActor : UntypedPersistentActor
+{
+    private static readonly string[] RedTag = ["red"];
+
+    private readonly bool _useSnapshots;
+    private readonly List<string> _values = [];
+    private long _snapshotSequenceNr;
+    private IActorRef _requester = ActorRefs.Nobody;
+
+    public CanaryPersistentActor(string persistenceId, bool useSnapshots = true)
+    {
+        PersistenceId = persistenceId;
+        _useSnapshots = useSnapshots;
+    }
+
+    public override string PersistenceId { get; }
+
+    public override Recovery Recovery => _useSnapshots ? Recovery.Default : new Recovery(SnapshotSelectionCriteria.None);
+
+    protected override void OnRecover(object message)
+    {
+        switch (message)
+        {
+            case SnapshotOffer { Snapshot: CanarySnapshot snapshot } offer:
+                _values.Clear();
+                _values.AddRange(snapshot.Values);
+                _snapshotSequenceNr = offer.Metadata.SequenceNr;
+                break;
+            case CanaryEvent e:
+                _values.Add(e.Value);
+                break;
+        }
+    }
+
+    protected override void OnCommand(object message)
+    {
+        switch (message)
+        {
+            case PersistCmd cmd:
+                var requester = Sender;
+                var evt = new CanaryEvent(cmd.Value);
+                if (cmd.Red)
+                    Persist(new Tagged(evt, RedTag), _ => OnPersisted(evt, requester));
+                else
+                    Persist(evt, _ => OnPersisted(evt, requester));
+                break;
+            case SaveNow:
+                _requester = Sender;
+                SaveSnapshot(new CanarySnapshot(_values.ToArray()));
+                break;
+            case SaveSnapshotSuccess success:
+                _requester.Tell(success.Metadata.SequenceNr);
+                break;
+            case SaveSnapshotFailure failure:
+                _requester.Tell(new Status.Failure(failure.Cause));
+                break;
+            case DeleteTo delete:
+                _requester = Sender;
+                DeleteMessages(delete.SequenceNr);
+                break;
+            case DeleteMessagesSuccess success:
+                _requester.Tell(success.ToSequenceNr);
+                break;
+            case DeleteMessagesFailure failure:
+                _requester.Tell(new Status.Failure(failure.Cause));
+                break;
+            case GetState:
+                Sender.Tell(new CanaryState(_values.ToArray(), _snapshotSequenceNr, LastSequenceNr));
+                break;
+        }
+    }
+
+    private void OnPersisted(CanaryEvent evt, IActorRef requester)
+    {
+        _values.Add(evt.Value);
+        requester.Tell(_values.Count);
+    }
+}