Bump Akka.Persistence.Hosting, Akka, Akka.Streams and Akka.Persistence (

#363)

* Bump Akka.Persistence.Hosting, Akka, Akka.Streams and Akka.Persistence

Bumps [Akka.Persistence.Hosting](https://github.com/akkadotnet/Akka.Hosting), [Akka](https://github.com/akkadotnet/akka.net), [Akka.Streams](https://github.com/akkadotnet/akka.net) and [Akka.Persistence](https://github.com/akkadotnet/akka.net). These dependencies needed to be updated together.

Updates `Akka.Persistence.Hosting` from 1.5.27 to 1.5.29
- [Release notes](https://github.com/akkadotnet/Akka.Hosting/releases)
- [Changelog](https://github.com/akkadotnet/Akka.Hosting/blob/dev/RELEASE_NOTES.md)
- [Commits](akkadotnet/Akka.Hosting@1.5.27...1.5.29)

Updates `Akka` from 1.5.27.1 to 1.5.29
- [Release notes](https://github.com/akkadotnet/akka.net/releases)
- [Changelog](https://github.com/akkadotnet/akka.net/blob/dev/RELEASE_NOTES.md)
- [Commits](akkadotnet/akka.net@1.5.27.1...1.5.29)

Updates `Akka.Streams` from 1.5.27.1 to 1.5.29
- [Release notes](https://github.com/akkadotnet/akka.net/releases)
- [Changelog](https://github.com/akkadotnet/akka.net/blob/dev/RELEASE_NOTES.md)
- [Commits](akkadotnet/akka.net@1.5.27.1...1.5.29)

Updates `Akka.Persistence` from 1.5.27.1 to 1.5.29
- [Release notes](https://github.com/akkadotnet/akka.net/releases)
- [Changelog](https://github.com/akkadotnet/akka.net/blob/dev/RELEASE_NOTES.md)
- [Commits](akkadotnet/akka.net@1.5.27.1...1.5.29)

---
updated-dependencies:
- dependency-name: Akka.Persistence.Hosting
  dependency-type: direct:production
  update-type: version-update:semver-patch
- dependency-name: Akka
  dependency-type: direct:production
  update-type: version-update:semver-patch
- dependency-name: Akka.Streams
  dependency-type: direct:production
  update-type: version-update:semver-patch
- dependency-name: Akka.Persistence
  dependency-type: direct:production
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <support@github.com>

* Add Snapshot Metadata timestamp support

(cherry picked from commit e6f18ed)

---------

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
Co-authored-by: Gregorius Soedharmo <arkatufus@yahoo.com>

src/Akka.Persistence.Redis.Tests/RedisSnapshotStoreSaveSnapshotSpec.cs

@@ -0,0 +1,63 @@
+// -----------------------------------------------------------------------
+// <copyright file="RedisSnapshotStoreSaveSnapshotSpec.cs" company="Akka.NET Project">
+//      Copyright (C) 2013-2021 .NET Foundation <https://github.com/akkadotnet/akka.net>
+// </copyright>
+// -----------------------------------------------------------------------
+
+using Akka.Configuration;
+using Akka.Persistence.TCK.Snapshot;
+using Xunit;
+using Xunit.Abstractions;
+
+#nullable enable
+namespace Akka.Persistence.Redis.Tests;
+
+[Collection("RedisSpec")]
+public class RedisSnapshotStoreSaveSnapshotSpec: SnapshotStoreSaveSnapshotSpec
+{
+    public const int Database = 1;
+
+    public static Config Config(RedisFixture fixture, int id)
+    {
+        DbUtils.Initialize(fixture);
+
+        return ConfigurationFactory.ParseString($@"
+                akka.test.single-expect-default = 3s
+                akka.persistence {{
+                    publish-plugin-commands = on
+                    snapshot-store {{
+                        plugin = ""akka.persistence.snapshot-store.redis""
+                        redis {{
+                            class = ""Akka.Persistence.Redis.Snapshot.RedisSnapshotStore, Akka.Persistence.Redis""
+                            configuration-string = ""{fixture.ConnectionString}""
+                            plugin-dispatcher = ""akka.actor.default-dispatcher""
+                            database = ""{id}""
+                        }}
+                    }}
+                }}
+                akka.actor {{
+                    serializers {{
+                        persistence-snapshot = ""Akka.Persistence.Redis.Serialization.PersistentSnapshotSerializer, Akka.Persistence.Redis""
+                    }}
+                    serialization-bindings {{
+                        ""Akka.Persistence.SelectedSnapshot, Akka.Persistence"" = persistence-snapshot
+                    }}
+                    serialization-identifiers {{
+                        ""Akka.Persistence.Redis.Serialization.PersistentSnapshotSerializer, Akka.Persistence.Redis"" = 48
+                    }}
+                }}").WithFallback(RedisPersistence.DefaultConfig());
+    }
+
+    public RedisSnapshotStoreSaveSnapshotSpec(ITestOutputHelper output, RedisFixture fixture)
+        : base(Config(fixture, Database), nameof(RedisSnapshotStoreSpec), output)
+    {
+        RedisPersistence.Get(Sys);
+    }
+
+    protected override void AfterAll()
+    {
+        base.AfterAll();
+        DbUtils.Clean(Database);
+    }
+
+}

src/Akka.Persistence.Redis/Snapshot/RedisSnapshotStore.cs

@@ -80,8 +80,33 @@ protected override Task SaveAsync(SnapshotMetadata metadata, object snapshot)
 
         protected override async Task DeleteAsync(SnapshotMetadata metadata)
         {
-            await Database.SortedSetRemoveRangeByScoreAsync(GetSnapshotKey(metadata.PersistenceId, IsClustered), metadata.SequenceNr,
-                metadata.SequenceNr);
+            if(metadata.Timestamp == DateTime.MinValue)
+            {
+                await Database.SortedSetRemoveRangeByScoreAsync(
+                    GetSnapshotKey(metadata.PersistenceId, IsClustered),
+                    metadata.SequenceNr,
+                    metadata.SequenceNr);
+                return;
+            }
+
+            var snapshots = await Database.SortedSetRangeByScoreAsync(
+                key: GetSnapshotKey(metadata.PersistenceId, IsClustered),
+                start: metadata.SequenceNr,
+                stop: 0L,
+                exclude: Exclude.None,
+                order: Order.Descending);
+
+            var found = snapshots
+                .Select(c => PersistentFromBytes(c))
+                .Where(snapshot => snapshot.Metadata.Timestamp <= metadata.Timestamp &&
+                                   snapshot.Metadata.SequenceNr == metadata.SequenceNr)
+                .Select(s => _database.Value.SortedSetRemoveRangeByScoreAsync(
+                    key: GetSnapshotKey(metadata.PersistenceId, IsClustered),
+                    start: s.Metadata.SequenceNr, 
+                    stop: s.Metadata.SequenceNr))
+                .ToArray();
+
+            await Task.WhenAll(found);
         }
 
         protected override async Task DeleteAsync(string persistenceId, SnapshotSelectionCriteria criteria)

src/Directory.Packages.props

@@ -1,13 +1,13 @@
 <Project>
     <PropertyGroup>
         <ManagePackageVersionsCentrally>true</ManagePackageVersionsCentrally>
-        <AkkaVersion>1.5.27.1</AkkaVersion>
+        <AkkaVersion>1.5.29</AkkaVersion>
     </PropertyGroup>
     <!-- Library dependencies -->
     <ItemGroup>
         <PackageVersion Include="Akka" Version="$(AkkaVersion)" />
         <PackageVersion Include="Akka.Persistence" Version="$(AkkaVersion)" />
-        <PackageVersion Include="Akka.Persistence.Hosting" Version="1.5.27" />
+        <PackageVersion Include="Akka.Persistence.Hosting" Version="1.5.29" />
         <PackageVersion Include="Akka.Persistence.Query" Version="$(AkkaVersion)" />
         <PackageVersion Include="Akka.Streams" Version="$(AkkaVersion)" />
         <PackageVersion Include="StackExchange.Redis" Version="2.8.0" />