Simplify GetMessagesForReplica to List<StoredMessages> and add ignorePositions

Return a List<StoredMessages> (StoredId + its messages) instead of a
Dictionary<StoredId, List<StoredMessage>>, and accept an ignorePositions
parameter so already-pushed messages are not re-delivered.

MessageWatchdog (version 1) keeps an ever-growing set of pushed positions
and passes it each tick; this will be refined once the QueueManager reports
the positions it has persisted into its effects.

Stores filter the ignore-set via a stable, parameterized query to avoid
plan-cache churn: Postgres '!= ALL($2)', SQL Server STRING_SPLIT, MariaDB
FIND_IN_SET. Adds a cross-store test covering empty and non-empty ignore sets.

Author: stidsborg

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

@@ -125,4 +125,8 @@ public override Task CrashedReplicaMessagesAreFetched()
     [TestMethod]
     public override Task MessageReplicaCanBeReassigned()
         => MessageReplicaCanBeReassigned(FunctionStoreFactory.Create());
+
+    [TestMethod]
+    public override Task GetMessagesForReplicaExcludesIgnoredPositions()
+        => GetMessagesForReplicaExcludesIgnoredPositions(FunctionStoreFactory.Create());
 }

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

@@ -1269,4 +1269,46 @@ protected async Task MessageReplicaCanBeReassigned(Task<IFunctionStore> function
         afterFlow2.Single(m => m.Position == bPosition).Replica.ShouldBe(newReplica);
         afterFlow2.Single(m => m.Position == cPosition).Replica.ShouldBe(otherReplica);
     }
+
+    public abstract Task GetMessagesForReplicaExcludesIgnoredPositions();
+    protected async Task GetMessagesForReplicaExcludesIgnoredPositions(Task<IFunctionStore> functionStoreTask)
+    {
+        var functionStore = await functionStoreTask;
+        var messageStore = functionStore.MessageStore;
+        var stringType = typeof(string).SimpleQualifiedName().ToUtf8Bytes();
+
+        var replica = ReplicaId.NewId();
+        var otherReplica = ReplicaId.NewId();
+
+        var flow1 = TestStoredId.Create();
+        var flow2 = TestStoredId.Create();
+
+        // idle flows -> the stored replica falls back to the publishing replica we provide
+        await messageStore.AppendMessage(flow1, new StoredMessage("a".ToJson().ToUtf8Bytes(), stringType, Position: 0, Replica: replica));
+        await messageStore.AppendMessage(flow1, new StoredMessage("b".ToJson().ToUtf8Bytes(), stringType, Position: 0, Replica: replica));
+        await messageStore.AppendMessage(flow2, new StoredMessage("c".ToJson().ToUtf8Bytes(), stringType, Position: 0, Replica: replica));
+        // owned by a different replica -> must never be returned
+        await messageStore.AppendMessage(flow2, new StoredMessage("d".ToJson().ToUtf8Bytes(), stringType, Position: 0, Replica: otherReplica));
+
+        // no positions ignored -> all of the replica's messages, grouped by flow and ordered by position
+        var all = await messageStore.GetMessagesForReplica(replica, ignorePositions: []);
+        all.Count.ShouldBe(2);
+        var flow1Group = all.Single(g => g.StoredId == flow1);
+        var flow2Group = all.Single(g => g.StoredId == flow2);
+        flow1Group.Messages.Select(m => (string) m.DefaultDeserialize()).ShouldBe(["a", "b"]);
+        flow2Group.Messages.Select(m => (string) m.DefaultDeserialize()).ShouldBe(["c"]); // "d" belongs to otherReplica
+
+        var aPosition = flow1Group.Messages.Single(m => (string) m.DefaultDeserialize() == "a").Position;
+        var bPosition = flow1Group.Messages.Single(m => (string) m.DefaultDeserialize() == "b").Position;
+
+        // ignoring "b" leaves only "a" in flow1; flow2 is unaffected
+        var ignoredB = await messageStore.GetMessagesForReplica(replica, ignorePositions: [bPosition]);
+        ignoredB.Single(g => g.StoredId == flow1).Messages.Select(m => (string) m.DefaultDeserialize()).ShouldBe(["a"]);
+        ignoredB.Single(g => g.StoredId == flow2).Messages.Select(m => (string) m.DefaultDeserialize()).ShouldBe(["c"]);
+
+        // ignoring every position of a flow drops the whole group
+        var ignoredFlow1 = await messageStore.GetMessagesForReplica(replica, ignorePositions: [aPosition, bPosition]);
+        ignoredFlow1.ShouldAllBe(g => g.StoredId != flow1);
+        ignoredFlow1.Single(g => g.StoredId == flow2).Messages.Single().DefaultDeserialize().ShouldBe("c");
+    }
 }

Core/Cleipnir.ResilientFunctions/CoreRuntime/FlowsManager.cs

@@ -42,7 +42,7 @@ public void Interrupt(IReadOnlyList<StoredId> ids)
                     flowState.Interrupt();
     }
 
-    public Task Push(IReadOnlyDictionary<StoredId, List<StoredMessage>> messagesByFlow)
+    public Task Push(IReadOnlyList<StoredMessages> messagesByFlow)
     {
         List<Task> tasks = new();
         lock (_lock)

Core/Cleipnir.ResilientFunctions/CoreRuntime/Watchdogs/MessageWatchdog.cs

@@ -1,4 +1,6 @@
 using System;
+using System.Collections.Generic;
+using System.Linq;
 using System.Threading.Tasks;
 using Cleipnir.ResilientFunctions.Domain;
 using Cleipnir.ResilientFunctions.Domain.Exceptions;
@@ -18,6 +20,11 @@ internal class MessageWatchdog
     private readonly TimeSpan _delayStartUp;
     private readonly UtcNow _utcNow;
 
+    // Positions already pushed to this replica's flows. Passed as ignore-set so messages are not re-delivered
+    // on subsequent ticks. Version 1: ever-growing - to be refined once the QueueManager reports the positions
+    // it has persisted into its effects.
+    private readonly HashSet<long> _pushedPositions = new();
+
     public MessageWatchdog(
         IMessageStore messageStore,
         FlowsManager flowsManager,
@@ -51,9 +58,15 @@ public async Task Start()
 
                 // Messages destined for flows currently owned by this replica (replica = COALESCE(owner, publisher)).
                 // FlowsManager.Push delivers only to live flows; entries for non-live flows are ignored.
-                var messagesByFlow = await _messageStore.GetMessagesForReplica(_clusterInfo.ReplicaId);
-                if (messagesByFlow.Count > 0)
-                    await _flowsManager.Push(messagesByFlow);
+                var messageGroups = await _messageStore.GetMessagesForReplica(_clusterInfo.ReplicaId, _pushedPositions.ToList());
+                if (messageGroups.Count > 0)
+                {
+                    foreach (var group in messageGroups)
+                        foreach (var message in group.Messages)
+                            _pushedPositions.Add(message.Position);
+
+                    await _flowsManager.Push(messageGroups);
+                }
 
                 var timeElapsed = _utcNow() - now;
                 var delay = (_checkFrequency - timeElapsed).RoundUpToZero();

Core/Cleipnir.ResilientFunctions/Messaging/IMessageStore.cs

@@ -28,9 +28,11 @@ public interface IMessageStore
 
     /// <summary>
     /// Returns the undelivered messages whose replica equals the provided replica, grouped by target flow.
+    /// Messages at any of the <paramref name="ignorePositions"/> are excluded - the MessageWatchdog passes the
+    /// positions it has already pushed so they are not re-delivered on subsequent ticks.
     /// Used by the MessageWatchdog to push messages to live flows owned by this replica.
     /// </summary>
-    Task<Dictionary<StoredId, List<StoredMessage>>> GetMessagesForReplica(ReplicaId replicaId);
+    Task<List<StoredMessages>> GetMessagesForReplica(ReplicaId replicaId, IReadOnlyList<long> ignorePositions);
 
     /// <summary>
     /// Returns the (flow, position) identifiers of the undelivered messages owned by a replica that is no

Core/Cleipnir.ResilientFunctions/Messaging/StoredMessage.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Collections.Generic;
 using System.Text.Json;
 using Cleipnir.ResilientFunctions.Domain;
 using Cleipnir.ResilientFunctions.Helpers;
@@ -12,6 +13,7 @@ public record StoredMessage(byte[] MessageContent, byte[] MessageType, long Posi
 }
 
 public record StoredIdAndMessage(StoredId StoredId, StoredMessage StoredMessage);
+public record StoredMessages(StoredId StoredId, List<StoredMessage> Messages);
 public static class StoredIdAndMessageExtensions
 {
     public static StoredIdAndMessage ToStoredIdAndMessage(this StoredMessage storedMessage, StoredId storedId) 

Core/Cleipnir.ResilientFunctions/Storage/InMemoryFunctionStore.cs

@@ -668,23 +668,24 @@ public async Task<Dictionary<StoredId, List<StoredMessage>>> GetMessages(IEnumer
         return dict;
     }
 
-    public Task<Dictionary<StoredId, List<StoredMessage>>> GetMessagesForReplica(ReplicaId replicaId)
+    public Task<List<StoredMessages>> GetMessagesForReplica(ReplicaId replicaId, IReadOnlyList<long> ignorePositions)
     {
         lock (_sync)
         {
-            var dict = new Dictionary<StoredId, List<StoredMessage>>();
+            var ignore = ignorePositions.ToHashSet();
+            var result = new List<StoredMessages>();
             foreach (var (storedId, messages) in _messages)
             {
                 var list = messages
                     .OrderBy(kv => kv.Key)
-                    .Where(kv => kv.Value.Replica == replicaId)
+                    .Where(kv => kv.Value.Replica == replicaId && !ignore.Contains(kv.Key))
                     .Select(kv => kv.Value with { Position = kv.Key })
                     .ToList();
                 if (list.Count > 0)
-                    dict[storedId] = list;
+                    result.Add(new StoredMessages(storedId, list));
             }
 
-            return dict.ToTask();
+            return result.ToTask();
         }
     }
 

Stores/MariaDB/Cleipnir.ResilientFunctions.MariaDB.Tests/Messaging/MessageStoreTests.cs

@@ -120,4 +120,8 @@ public override Task CrashedReplicaMessagesAreFetched()
     [TestMethod]
     public override Task MessageReplicaCanBeReassigned()
         => MessageReplicaCanBeReassigned(FunctionStoreFactory.Create());
+
+    [TestMethod]
+    public override Task GetMessagesForReplicaExcludesIgnoredPositions()
+        => GetMessagesForReplicaExcludesIgnoredPositions(FunctionStoreFactory.Create());
 }

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

@@ -192,22 +192,23 @@ public async Task<Dictionary<StoredId, List<StoredMessage>>> GetMessages(IEnumer
         return storedMessages;
     }
 
-    public async Task<Dictionary<StoredId, List<StoredMessage>>> GetMessagesForReplica(ReplicaId replicaId)
+    public async Task<List<StoredMessages>> GetMessagesForReplica(ReplicaId replicaId, IReadOnlyList<long> ignorePositions)
     {
         await using var conn = await DatabaseHelper.CreateOpenConnection(_connectionString);
         await using var command = _sqlGenerator
-            .GetMessagesForReplica(replicaId)
+            .GetMessagesForReplica(replicaId, ignorePositions)
             .ToSqlCommand(conn);
 
         await using var reader = await command.ExecuteReaderAsync();
 
         var messages = await _sqlGenerator.ReadStoredIdsMessages(reader);
-        var storedMessages = new Dictionary<StoredId, List<StoredMessage>>();
+        var storedMessages = new List<StoredMessages>();
 
         foreach (var id in messages.Keys)
-            storedMessages[id] = messages[id]
-                .Select(m => ConvertToStoredMessage(m.content, m.position, m.replica))
-                .ToList();
+            storedMessages.Add(new StoredMessages(
+                id,
+                messages[id].Select(m => ConvertToStoredMessage(m.content, m.position, m.replica)).ToList()
+            ));
 
         return storedMessages;
     }

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

@@ -660,15 +660,15 @@ public StoreCommand GetMessages(StoredId storedId, IReadOnlyList<long> skipPosit
         return messages;
     }
 
-    public StoreCommand GetMessagesForReplica(ReplicaId replicaId)
+    public StoreCommand GetMessagesForReplica(ReplicaId replicaId, IReadOnlyList<long> ignorePositions)
     {
         var sql = @$"
             SELECT id, position, content, replica
             FROM {tablePrefix}_messages
-            WHERE replica = ?
+            WHERE replica = ? AND FIND_IN_SET(position, ?) = 0
             ORDER BY position;";
 
-        return StoreCommand.Create(sql, values: [ replicaId.AsGuid.ToString("N") ]);
+        return StoreCommand.Create(sql, values: [ replicaId.AsGuid.ToString("N"), string.Join(",", ignorePositions) ]);
     }
 
     public StoreCommand GetCrashedReplicaMessages(IReadOnlySet<ReplicaId> liveReplicas)