Take IReadOnlySet<ReplicaId> for GetCrashedReplicaMessages liveReplicas

A set is the natural type for the membership test and matches how live
replicas are already represented elsewhere (ReplicaWatchdog builds a
HashSet). The in-memory store now uses the set's Contains directly instead
of copying into a local HashSet.

Author: stidsborg

Core/Cleipnir.ResilientFunctions.Tests/Messaging/TestTemplates/MessageStoreTests.cs

@@ -1229,7 +1229,7 @@ protected async Task CrashedReplicaMessagesAreFetched(Task<IFunctionStore> funct
         var bPosition = flow1Messages.Single(m => m.Replica == crashedReplica1).Position;
         var cPosition = flow2Messages.Single(m => m.Replica == crashedReplica2).Position;
 
-        var crashed = await messageStore.GetCrashedReplicaMessages([liveReplica]);
+        var crashed = await messageStore.GetCrashedReplicaMessages(new HashSet<ReplicaId> { liveReplica });
 
         crashed.Count.ShouldBe(2);
         crashed.ShouldContain(t => t.StoredId == flow1 && t.Position == bPosition);

Core/Cleipnir.ResilientFunctions/Messaging/IMessageStore.cs

@@ -37,7 +37,7 @@ public interface IMessageStore
     /// longer alive (its replica is not contained in <paramref name="liveReplicas"/>).
     /// Used to detect messages stranded by crashed replicas so they can be re-assigned to a live replica via <see cref="SetReplica"/>.
     /// </summary>
-    Task<List<StoredIdAndPosition>> GetCrashedReplicaMessages(IEnumerable<ReplicaId> liveReplicas);
+    Task<List<StoredIdAndPosition>> GetCrashedReplicaMessages(IReadOnlySet<ReplicaId> liveReplicas);
 
     /// <summary>
     /// Re-assigns the messages at the provided positions to <paramref name="newReplica"/>,

Core/Cleipnir.ResilientFunctions/Storage/InMemoryFunctionStore.cs

@@ -688,15 +688,14 @@ public Task<Dictionary<StoredId, List<StoredMessage>>> GetMessagesForReplica(Rep
         }
     }
 
-    public Task<List<StoredIdAndPosition>> GetCrashedReplicaMessages(IEnumerable<ReplicaId> liveReplicas)
+    public Task<List<StoredIdAndPosition>> GetCrashedReplicaMessages(IReadOnlySet<ReplicaId> liveReplicas)
     {
-        var live = liveReplicas.ToHashSet();
         lock (_sync)
         {
             var result = new List<StoredIdAndPosition>();
             foreach (var (storedId, messages) in _messages)
                 foreach (var (position, message) in messages.OrderBy(kv => kv.Key))
-                    if (!live.Contains(message.Replica))
+                    if (!liveReplicas.Contains(message.Replica))
                         result.Add(new StoredIdAndPosition(storedId, position));
 
             return result.ToTask();

Stores/MariaDB/Cleipnir.ResilientFunctions.MariaDB/MariaDbMessageStore.cs

@@ -212,7 +212,7 @@ public async Task<Dictionary<StoredId, List<StoredMessage>>> GetMessagesForRepli
         return storedMessages;
     }
 
-    public async Task<List<StoredIdAndPosition>> GetCrashedReplicaMessages(IEnumerable<ReplicaId> liveReplicas)
+    public async Task<List<StoredIdAndPosition>> GetCrashedReplicaMessages(IReadOnlySet<ReplicaId> liveReplicas)
     {
         await using var conn = await DatabaseHelper.CreateOpenConnection(_connectionString);
         await using var command = _sqlGenerator

Stores/MariaDB/Cleipnir.ResilientFunctions.MariaDB/SqlGenerator.cs

@@ -671,7 +671,7 @@ public StoreCommand GetMessagesForReplica(ReplicaId replicaId)
         return StoreCommand.Create(sql, values: [ replicaId.AsGuid.ToString("N") ]);
     }
 
-    public StoreCommand GetCrashedReplicaMessages(IEnumerable<ReplicaId> liveReplicas)
+    public StoreCommand GetCrashedReplicaMessages(IReadOnlySet<ReplicaId> liveReplicas)
     {
         var replicas = liveReplicas.Select(r => $"'{r.AsGuid:N}'").ToList();
         var sql = @$"

Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL/PostgreSqlMessageStore.cs

@@ -211,7 +211,7 @@ public async Task<Dictionary<StoredId, List<StoredMessage>>> GetMessagesForRepli
         return storedMessages;
     }
 
-    public async Task<List<StoredIdAndPosition>> GetCrashedReplicaMessages(IEnumerable<ReplicaId> liveReplicas)
+    public async Task<List<StoredIdAndPosition>> GetCrashedReplicaMessages(IReadOnlySet<ReplicaId> liveReplicas)
     {
         await using var conn = await CreateConnection();
         await using var command = sqlGenerator.GetCrashedReplicaMessages(liveReplicas).ToNpgsqlCommand(conn);

Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL/SqlGenerator.cs

@@ -595,7 +595,7 @@ public StoreCommand GetMessagesForReplica(ReplicaId replicaId)
         return StoreCommand.Create(sql, values: [ replicaId.AsGuid ]);
     }
 
-    public StoreCommand GetCrashedReplicaMessages(IEnumerable<ReplicaId> liveReplicas)
+    public StoreCommand GetCrashedReplicaMessages(IReadOnlySet<ReplicaId> liveReplicas)
     {
         var sql = @$"
             SELECT id, position

Stores/SqlServer/Cleipnir.ResilientFunctions.SqlServer/SqlGenerator.cs

@@ -675,7 +675,7 @@ public StoreCommand GetMessagesForReplica(ReplicaId replicaId)
         return command;
     }
 
-    public StoreCommand GetCrashedReplicaMessages(IEnumerable<ReplicaId> liveReplicas)
+    public StoreCommand GetCrashedReplicaMessages(IReadOnlySet<ReplicaId> liveReplicas)
     {
         var replicas = liveReplicas.ToList();
         var sql = @$"

Stores/SqlServer/Cleipnir.ResilientFunctions.SqlServer/SqlServerMessageStore.cs

@@ -221,7 +221,7 @@ public async Task<Dictionary<StoredId, List<StoredMessage>>> GetMessagesForRepli
         return storedMessages;
     }
 
-    public async Task<List<StoredIdAndPosition>> GetCrashedReplicaMessages(IEnumerable<ReplicaId> liveReplicas)
+    public async Task<List<StoredIdAndPosition>> GetCrashedReplicaMessages(IReadOnlySet<ReplicaId> liveReplicas)
     {
         await using var conn = await CreateConnection();
         await using var cmd = _sqlGenerator.GetCrashedReplicaMessages(liveReplicas).ToSqlCommand(conn);