Port akkadotnet/akka.net#7313 (#448)

akkadotnet · Sep 4, 2024 · 64bdd34 · 64bdd34
1 parent ce5bc63
commit 64bdd34
Show file tree

Hide file tree

Showing 5 changed files with 44 additions and 10 deletions.
diff --git a/src/Akka.Persistence.Sql.Tests/Sqlite/SystemDataSqliteSnapshotSpec.cs b/src/Akka.Persistence.Sql.Tests/Sqlite/SystemDataSqliteSnapshotSpec.cs
@@ -0,0 +1,23 @@
+// -----------------------------------------------------------------------
+//  <copyright file="SystemDataSqliteSnapshotSpec.cs" company="Akka.NET Project">
+//      Copyright (C) 2013-2023 .NET Foundation <https://github.com/akkadotnet/akka.net>
+//  </copyright>
+// -----------------------------------------------------------------------
+
+using Akka.Persistence.Sql.Tests.Common.Containers;
+using Akka.Persistence.TCK.Snapshot;
+using Xunit;
+using Xunit.Abstractions;
+
+namespace Akka.Persistence.Sql.Tests.Sqlite
+{
+    [Collection(nameof(SqlitePersistenceSpec))]
+    public class SystemDataSqliteSnapshotSpec: SnapshotStoreSpec
+    {
+        public SystemDataSqliteSnapshotSpec(ITestOutputHelper output, SqliteContainer fixture)
+            : base(SqliteSnapshotSpecConfig.Create(fixture), nameof(SystemDataSqliteSnapshotSpec), output)
+        {
+            Initialize();
+        }
+    }
+}
diff --git a/src/Akka.Persistence.Sql/Snapshot/ByteArraySnapshotDao.cs b/src/Akka.Persistence.Sql/Snapshot/ByteArraySnapshotDao.cs
@@ -357,6 +357,7 @@ public async Task<Option<SelectedSnapshot>> SnapshotForMaxSequenceNrAndMaxTimest
         public async Task DeleteAsync(
             string persistenceId,
             long sequenceNr,
+            DateTime timestamp,
             CancellationToken cancellationToken = default)
         {
             var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, _shutdownCts.Token);
@@ -367,22 +368,31 @@ await _connectionFactory.ExecuteWithTransactionAsync(
                 {
                     if (connection.UseDateTime)
                     {
-                        await connection
-                            .GetTable<DateTimeSnapshotRow>()
-                            .Where(
-                                r =>
-                                    r.PersistenceId == persistenceId &&
-                                    r.SequenceNumber == sequenceNr)
+                        var query = connection.GetTable<DateTimeSnapshotRow>()
+                                .Where(
+                                    r =>
+                                        r.PersistenceId == persistenceId &&
+                                        r.SequenceNumber == sequenceNr);
+                        
+                        if (timestamp > DateTime.MinValue)
+                            query = query.Where(r => r.Created <= timestamp);
+                        
+                        await query
                             .DeleteAsync(token);
                     }
                     else
                     {
-                        await connection
+                        var query = connection
                             .GetTable<LongSnapshotRow>()
                             .Where(
                                 r =>
                                     r.PersistenceId == persistenceId &&
-                                    r.SequenceNumber == sequenceNr)
+                                    r.SequenceNumber == sequenceNr);
+                        
+                        if (timestamp > DateTime.MinValue)
+                            query = query.Where(r => r.Created <= timestamp.Ticks);
+                        
+                        await query
                             .DeleteAsync(token);
                     }
                 });

diff --git a/src/Akka.Persistence.Sql/Snapshot/ISnapshotDao.cs b/src/Akka.Persistence.Sql/Snapshot/ISnapshotDao.cs
@@ -56,6 +56,7 @@ Task<Option<SelectedSnapshot>> SnapshotForMaxSequenceNrAndMaxTimestampAsync(
         Task DeleteAsync(
             string persistenceId,
             long sequenceNr,
+            DateTime timestamp,
             CancellationToken cancellationToken = default);
 
         Task SaveAsync(

diff --git a/src/Akka.Persistence.Sql/Snapshot/SqlSnapshotStore.cs b/src/Akka.Persistence.Sql/Snapshot/SqlSnapshotStore.cs
@@ -114,7 +114,7 @@ protected override async Task SaveAsync(SnapshotMetadata metadata, object snapsh
             => await _dao.SaveAsync(metadata, snapshot);
 
         protected override async Task DeleteAsync(SnapshotMetadata metadata)
-            => await _dao.DeleteAsync(metadata.PersistenceId, metadata.SequenceNr);
+            => await _dao.DeleteAsync(metadata.PersistenceId, metadata.SequenceNr, metadata.Timestamp);
 
         protected override async Task DeleteAsync(string persistenceId, SnapshotSelectionCriteria criteria)
         {

diff --git a/src/Directory.Packages.props b/src/Directory.Packages.props
@@ -1,7 +1,7 @@
 <Project>
   <PropertyGroup>
     <ManagePackageVersionsCentrally>true</ManagePackageVersionsCentrally>
-    <AkkaVersion>1.5.27.1</AkkaVersion>
+    <AkkaVersion>1.5.28-beta1</AkkaVersion>
     <AkkaHostingVersion>1.5.27</AkkaHostingVersion>
     <FluentMigratorVersion>5.2.0</FluentMigratorVersion>
     <SqliteVersion>1.0.118</SqliteVersion>