Move message deletion from QueueManager into MessageWatchdog.RemoveMessages (#172)

QueueManager no longer calls IMessageStore.DeleteMessages directly. Instead
RemoveMessages now owns deletion: it deletes the handled positions from the
store and then trims them from the watchdog's ignore-set (delete-before-trim so
a no-longer-ignored position can't be re-fetched before it is gone from the
store). The three QueueManager delete sites (Initialize replay, idempotency
duplicate, AfterFlush) collapse to a single awaited RemoveMessages call.

IMessageWatchdog.RemoveMessages gains the StoredId and becomes async; the
hand-rolled QueueManager tests keep a one-line no-op stub.

Author: stidsborg

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

@@ -18,13 +18,13 @@ namespace Cleipnir.ResilientFunctions.Tests.Messaging.TestTemplates;
 
 public abstract class MessagesSubscriptionTests
 {
-    // These tests hand-roll a QueueManager, which depends on IMessageWatchdog only to report deleted message
-    // positions. They don't exercise that path, so a no-op stub suffices.
+    // These tests hand-roll a QueueManager, which delegates message deletion to IMessageWatchdog. They don't
+    // assert on store cleanup, so a no-op stub suffices.
     private static readonly IMessageWatchdog StubMessageWatchdog = new NoopMessageWatchdog();
 
     private sealed class NoopMessageWatchdog : IMessageWatchdog
     {
-        public void RemoveMessages(IReadOnlyList<long> positions) { }
+        public Task RemoveMessages(StoredId storedId, IReadOnlyList<long> positions) => Task.CompletedTask;
     }
 
     public abstract Task EventsSubscriptionSunshineScenario();

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

@@ -1,13 +1,15 @@
 using System.Collections.Generic;
+using System.Threading.Tasks;
+using Cleipnir.ResilientFunctions.Storage;
 
 namespace Cleipnir.ResilientFunctions.CoreRuntime.Watchdogs;
 
 /// <summary>
-/// The slice of <see cref="MessageWatchdog"/> a QueueManager depends on: reporting message positions it has
-/// deleted from the store so the watchdog can drop them from its ignore-set. Exists so tests that hand-roll a
+/// The slice of <see cref="MessageWatchdog"/> a QueueManager depends on: deleting handled messages from the
+/// store and dropping their positions from the watchdog's ignore-set. Exists so tests that hand-roll a
 /// QueueManager can pass a no-op stub instead of a fully wired watchdog.
 /// </summary>
 internal interface IMessageWatchdog
 {
-    void RemoveMessages(IReadOnlyList<long> positions);
+    Task RemoveMessages(StoredId storedId, IReadOnlyList<long> positions);
 }

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

@@ -7,6 +7,7 @@
 using Cleipnir.ResilientFunctions.Domain.Exceptions;
 using Cleipnir.ResilientFunctions.Helpers;
 using Cleipnir.ResilientFunctions.Messaging;
+using Cleipnir.ResilientFunctions.Storage;
 
 namespace Cleipnir.ResilientFunctions.CoreRuntime.Watchdogs;
 
@@ -22,9 +23,9 @@ internal class MessageWatchdog : IMessageWatchdog
     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. A QueueManager calls RemoveMessages once it has deleted the corresponding messages
-    // from the store, trimming this set so it does not grow without bound. Guarded by _pushedPositionsLock
-    // because RemoveMessages runs on flow threads concurrently with the watchdog loop.
+    // on subsequent ticks. RemoveMessages (called by a QueueManager to delete handled messages) trims this set
+    // so it does not grow without bound. Guarded by _pushedPositionsLock because RemoveMessages runs on flow
+    // threads concurrently with the watchdog loop.
     private readonly HashSet<long> _pushedPositions = new();
     private readonly Lock _pushedPositionsLock = new();
 
@@ -98,14 +99,17 @@ public async Task Start()
     }
 
     /// <summary>
-    /// Called by a QueueManager once it has deleted the given message positions from the store, so the
-    /// watchdog drops them from its ignore-set instead of carrying them forever.
+    /// Deletes the given handled message positions from the store on a QueueManager's behalf, then drops them
+    /// from the ignore-set instead of carrying them forever. Deleting before trimming avoids re-fetching a
+    /// position that is no longer ignored but not yet gone from the store.
     /// </summary>
-    public void RemoveMessages(IReadOnlyList<long> positions)
+    public async Task RemoveMessages(StoredId storedId, IReadOnlyList<long> positions)
     {
         if (positions.Count == 0)
             return;
 
+        await _messageStore.DeleteMessages(storedId, positions);
+
         lock (_pushedPositionsLock)
             foreach (var position in positions)
                 _pushedPositions.Remove(position);

Core/Cleipnir.ResilientFunctions/Queuing/QueueManager.cs

@@ -89,8 +89,7 @@ private async Task Initialize()
 
             if (_effect.TryGet<List<long>>(DeliveredPositionsId, out var positions) && positions is { Count: > 0 })
             {
-                await _messageStore.DeleteMessages(_storedId, positions);
-                _messageWatchdog.RemoveMessages(positions);
+                await _messageWatchdog.RemoveMessages(_storedId, positions);
                 positions.Clear();
                 _effect.FlushlessUpsert(DeliveredPositionsId, positions, alias: null);
             }
@@ -228,8 +227,7 @@ private async Task ProcessMessages(IReadOnlyList<StoredMessage> messages)
 
                 if (idempotencyKey != null && !_idempotencyKeys.Add(idempotencyKey, position))
                 {
-                    await _messageStore.DeleteMessages(_storedId, [position]);
-                    _messageWatchdog.RemoveMessages([position]);
+                    await _messageWatchdog.RemoveMessages(_storedId, [position]);
                     continue;
                 }
 
@@ -287,8 +285,7 @@ public async Task AfterFlush()
             if (deliveredPositions.Count == 0 || _effect.IsDirty(DeliveredPositionsId))
                 return;
 
-            await _messageStore.DeleteMessages(_storedId, deliveredPositions);
-            _messageWatchdog.RemoveMessages(deliveredPositions);
+            await _messageWatchdog.RemoveMessages(_storedId, deliveredPositions);
             deliveredPositions.Clear();
             _effect.FlushlessUpsert(DeliveredPositionsId, deliveredPositions, alias: null);
         }