diff --git a/src/Akka.Persistence.Sql.Tests/Sqlite/SystemDataSqliteSnapshotSpec.cs b/src/Akka.Persistence.Sql.Tests/Sqlite/SystemDataSqliteSnapshotSpec.cs new file mode 100644 index 00000000..6091e00f --- /dev/null +++ b/src/Akka.Persistence.Sql.Tests/Sqlite/SystemDataSqliteSnapshotSpec.cs @@ -0,0 +1,23 @@ +// ----------------------------------------------------------------------- +// +// Copyright (C) 2013-2023 .NET Foundation +// +// ----------------------------------------------------------------------- + +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 index 2ad8f233..dab5541f 100644 --- a/src/Akka.Persistence.Sql/Snapshot/ByteArraySnapshotDao.cs +++ b/src/Akka.Persistence.Sql/Snapshot/ByteArraySnapshotDao.cs @@ -357,6 +357,7 @@ public async Task> 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() - .Where( - r => - r.PersistenceId == persistenceId && - r.SequenceNumber == sequenceNr) + var query = connection.GetTable() + .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() .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 index 37b08649..362c5f6c 100644 --- a/src/Akka.Persistence.Sql/Snapshot/ISnapshotDao.cs +++ b/src/Akka.Persistence.Sql/Snapshot/ISnapshotDao.cs @@ -56,6 +56,7 @@ Task> 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 index 9e1a4937..32c2a9a5 100644 --- 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 index fd3062bf..b18faf00 100644 --- a/src/Directory.Packages.props +++ b/src/Directory.Packages.props @@ -1,7 +1,7 @@ true - 1.5.27.1 + 1.5.28-beta1 1.5.27 5.2.0 1.0.118