Extract MessageClearer and add a positions-only DeleteMessages overload

Add IMessageStore.DeleteMessages(IEnumerable<long> positions) - a position-only
delete (positions are globally unique identity values) so handled messages
across many flows can be removed in one query. Implemented inline in every store
(InMemory, PostgreSQL, MariaDB, SqlServer); the existing per-StoredId overload is
untouched.

Extract the handled-message lifecycle out of MessageWatchdog into a dedicated
MessageClearer: it owns the not-yet-cleared ignore-set (MarkPushed /
NonClearedPositions) and the coalescing, retry-until-success delete pipeline
(Clear). The watchdog becomes a pure poll/push loop with no shared mutable state.
QueueManager now depends on the narrow IMessageClearer instead of IMessageWatchdog
and calls Clear(positions) (the dead storedId argument is gone). A failed delete
is retried until it lands - callers are told a message is gone only once it truly
is - notifying the unhandled-exception handler on each failure.

Tests: MessageClearerTests covers coalescing, retry-on-failure, concurrent load,
and ignore-set trimming against a bare MessageClearer; the hand-rolled
QueueManager tests use a one-line IMessageClearer stub.

Author: stidsborg

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

@@ -0,0 +1,152 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using Cleipnir.ResilientFunctions.CoreRuntime;
+using Cleipnir.ResilientFunctions.CoreRuntime.Watchdogs;
+using Cleipnir.ResilientFunctions.Domain;
+using Cleipnir.ResilientFunctions.Domain.Exceptions;
+using Cleipnir.ResilientFunctions.Messaging;
+using Cleipnir.ResilientFunctions.Storage;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using Shouldly;
+
+namespace Cleipnir.ResilientFunctions.Tests.Messaging.InMemoryTests;
+
+[TestClass]
+public class MessageClearerTests
+{
+    private static readonly TimeSpan MaxWait = TimeSpan.FromSeconds(10);
+
+    [TestMethod]
+    public async Task ClearCoalescesCallsArrivingWhileADeleteIsInFlight()
+    {
+        var firstDeleteReached = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var releaseFirstDelete = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var callCount = 0;
+
+        var store = new ControllableMessageStore(async _ =>
+        {
+            if (Interlocked.Increment(ref callCount) == 1)
+            {
+                firstDeleteReached.SetResult();
+                await releaseFirstDelete.Task;
+            }
+        });
+        var clearer = CreateClearer(store);
+
+        // First call starts the drain and blocks inside DeleteMessages.
+        var first = clearer.Clear([1]);
+        await firstDeleteReached.Task.WaitAsync(MaxWait);
+
+        // These arrive mid-flight, so they should be batched into a single follow-up delete.
+        var second = clearer.Clear([2]);
+        var third = clearer.Clear([3]);
+
+        releaseFirstDelete.SetResult();
+        await Task.WhenAll(first, second, third).WaitAsync(MaxWait);
+
+        store.DeletedBatches.Count.ShouldBe(2);
+        store.DeletedBatches[0].ShouldBe(new long[] { 1 });
+        store.DeletedBatches[1].OrderBy(p => p).ShouldBe(new long[] { 2, 3 });
+    }
+
+    [TestMethod]
+    public async Task ClearRetriesUntilDeleteSucceedsAndNotifiesEachFailure()
+    {
+        var unhandledLock = new Lock();
+        var unhandled = new List<FrameworkException>();
+        var failuresRemaining = 3;
+
+        var store = new ControllableMessageStore(_ =>
+            Interlocked.Decrement(ref failuresRemaining) >= 0
+                ? throw new InvalidOperationException("boom")
+                : Task.CompletedTask
+        );
+        var clearer = CreateClearer(
+            store,
+            onUnhandledException: e => { lock (unhandledLock) unhandled.Add(e); },
+            retryDelay: TimeSpan.FromMilliseconds(10)
+        );
+
+        // Despite the first three deletes throwing, the caller's task completes (it is never faulted).
+        await clearer.Clear([1]).WaitAsync(MaxWait);
+
+        store.DeletedPositions.ShouldContain(1L);
+        lock (unhandledLock)
+        {
+            unhandled.Count.ShouldBe(3);
+            unhandled.ShouldAllBe(e => e.InnerException is InvalidOperationException);
+        }
+    }
+
+    [TestMethod]
+    public async Task ClearCompletesEveryCallerUnderConcurrentLoad()
+    {
+        var store = new ControllableMessageStore(_ => Task.CompletedTask);
+        var clearer = CreateClearer(store);
+
+        var tasks = Enumerable
+            .Range(0, 200)
+            .Select(i => clearer.Clear([i]))
+            .ToArray();
+
+        await Task.WhenAll(tasks).WaitAsync(MaxWait);
+
+        store.DeletedPositions.OrderBy(p => p).ShouldBe(Enumerable.Range(0, 200).Select(i => (long)i));
+    }
+
+    [TestMethod]
+    public async Task ClearRemovesPositionsFromIgnoreSetOnceDeleted()
+    {
+        var store = new ControllableMessageStore(_ => Task.CompletedTask);
+        var clearer = CreateClearer(store);
+
+        clearer.MarkPushed([1, 2, 3]);
+        clearer.NonClearedPositions().OrderBy(p => p).ShouldBe(new long[] { 1, 2, 3 });
+
+        await clearer.Clear([2]).WaitAsync(MaxWait);
+
+        clearer.NonClearedPositions().OrderBy(p => p).ShouldBe(new long[] { 1, 3 });
+    }
+
+    private static MessageClearer CreateClearer(
+        IMessageStore messageStore,
+        Action<FrameworkException>? onUnhandledException = null,
+        TimeSpan? retryDelay = null)
+        => new(
+            messageStore,
+            new UnhandledExceptionHandler(onUnhandledException ?? (_ => { })),
+            retryDelay ?? TimeSpan.FromSeconds(1)
+        );
+
+    // Minimal IMessageStore that only implements the positions-only DeleteMessages (the sole method
+    // MessageClearer touches); every other member is irrelevant to these tests.
+    private sealed class ControllableMessageStore(Func<IReadOnlyList<long>, Task> onDelete) : IMessageStore
+    {
+        private readonly Lock _lock = new();
+        public List<long[]> DeletedBatches { get; } = new();
+        public IEnumerable<long> DeletedPositions => DeletedBatches.SelectMany(b => b);
+
+        public async Task DeleteMessages(IEnumerable<long> positions)
+        {
+            var batch = positions.ToArray();
+            lock (_lock)
+                DeletedBatches.Add(batch);
+            await onDelete(batch);
+        }
+
+        public Task Initialize() => throw new NotSupportedException();
+        public Task AppendMessages(IReadOnlyList<StoredIdAndMessage> messages) => throw new NotSupportedException();
+        public Task<bool> ReplaceMessage(StoredId storedId, long position, StoredMessage storedMessage) => throw new NotSupportedException();
+        public Task DeleteMessages(StoredId storedId, IEnumerable<long> positions) => throw new NotSupportedException();
+        public Task Truncate(StoredId storedId) => throw new NotSupportedException();
+        public Task<IReadOnlyList<StoredMessage>> GetMessages(StoredId storedId) => throw new NotSupportedException();
+        public Task<IReadOnlyList<StoredMessage>> GetMessages(StoredId storedId, IReadOnlyList<long> skipPositions) => throw new NotSupportedException();
+        public Task<Dictionary<StoredId, List<StoredMessage>>> GetMessages(IEnumerable<StoredId> storedIds) => throw new NotSupportedException();
+        public Task<List<StoredMessages>> GetMessagesForReplica(ReplicaId replicaId, IReadOnlyList<long> ignorePositions) => throw new NotSupportedException();
+        public Task<List<StoredIdAndPosition>> GetCrashedReplicaMessages(IReadOnlySet<ReplicaId> liveReplicas) => throw new NotSupportedException();
+        public Task SetReplica(IEnumerable<long> positions, ReplicaId newReplica, ReplicaId expectedReplica) => throw new NotSupportedException();
+    }
+}

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 delegates message deletion to IMessageWatchdog. They don't
+    // These tests hand-roll a QueueManager, which delegates message deletion to IMessageClearer. They don't
     // assert on store cleanup, so a no-op stub suffices.
-    private static readonly IMessageWatchdog StubMessageWatchdog = new NoopMessageWatchdog();
+    private static readonly IMessageClearer StubMessageClearer = new NoopMessageClearer();
 
-    private sealed class NoopMessageWatchdog : IMessageWatchdog
+    private sealed class NoopMessageClearer : IMessageClearer
     {
-        public Task RemoveMessages(StoredId storedId, IReadOnlyList<long> positions) => Task.CompletedTask;
+        public Task Clear(IReadOnlyList<long> positions) => Task.CompletedTask;
     }
 
     public abstract Task EventsSubscriptionSunshineScenario();
@@ -582,7 +582,7 @@ protected async Task QueueManagerFailsOnMessageDeserializationError(Task<IFuncti
                     flowTimeouts,
                     () => DateTime.UtcNow,
                     SettingsWithDefaults.Default,
-                    StubMessageWatchdog
+                    StubMessageClearer
                 );
 
                 var queueClient = await queueManager.CreateQueueClient();
@@ -646,7 +646,7 @@ protected async Task RegisteredTimeoutIsRemovedWhenPullingMessage(Task<IFunction
                     minimumTimeout,
                     () => DateTime.UtcNow,
                     SettingsWithDefaults.Default,
-                    StubMessageWatchdog
+                    StubMessageClearer
                 );
 
 
@@ -708,7 +708,7 @@ protected async Task PullEnvelopeReturnsEnvelopeWithReceiverAndSender(Task<IFunc
                     flowTimeouts,
                     () => DateTime.UtcNow,
                     SettingsWithDefaults.Default,
-                    StubMessageWatchdog
+                    StubMessageClearer
                 );
 
                 var queueClient = await queueManager.CreateQueueClient();

Core/Cleipnir.ResilientFunctions/CoreRuntime/Invocation/InvocationHelper.cs

@@ -26,12 +26,12 @@ internal class InvocationHelper<TParam, TReturn>
     private readonly StoredType _storedType;
     private readonly ReplicaId _replicaId;
     private readonly ResultBusyWaiter<TReturn> _resultBusyWaiter;
-    private readonly MessageWatchdog _messageWatchdog;
+    private readonly IMessageClearer _messageClearer;
     public UtcNow UtcNow { get; }
 
     private ISerializer Serializer { get; }
 
-    public InvocationHelper(FlowType flowType, StoredType storedType, ReplicaId replicaId, bool isParamlessFunction, SettingsWithDefaults settings, IFunctionStore functionStore, ShutdownCoordinator shutdownCoordinator, ISerializer serializer, UtcNow utcNow, bool clearChildren, MessageWatchdog messageWatchdog)
+    public InvocationHelper(FlowType flowType, StoredType storedType, ReplicaId replicaId, bool isParamlessFunction, SettingsWithDefaults settings, IFunctionStore functionStore, ShutdownCoordinator shutdownCoordinator, ISerializer serializer, UtcNow utcNow, bool clearChildren, IMessageClearer messageClearer)
     {
         _flowType = flowType;
         _isParamlessFunction = isParamlessFunction;
@@ -44,7 +44,7 @@ public InvocationHelper(FlowType flowType, StoredType storedType, ReplicaId repl
         _storedType = storedType;
         _replicaId = replicaId;
         _functionStore = functionStore;
-        _messageWatchdog = messageWatchdog;
+        _messageClearer = messageClearer;
         _resultBusyWaiter = new ResultBusyWaiter<TReturn>(_functionStore, Serializer);
     }
 
@@ -413,7 +413,7 @@ public async Task<ExistingEffects> CreateExistingEffects(FlowId flowId)
     public ExistingMessages CreateExistingMessages(FlowId flowId) => new(MapToStoredId(flowId), _functionStore.MessageStore, Serializer);
 
     public QueueManager CreateQueueManager(FlowId flowId, StoredId storedId, Effect effect, FlowState flowState, FlowTimeouts timeouts, UnhandledExceptionHandler unhandledExceptionHandler)
-        => new(flowId, storedId, _functionStore.MessageStore, Serializer, effect, flowState, unhandledExceptionHandler, timeouts, UtcNow, _settings, _messageWatchdog);
+        => new(flowId, storedId, _functionStore.MessageStore, Serializer, effect, flowState, unhandledExceptionHandler, timeouts, UtcNow, _settings, _messageClearer);
 
     public StoredId MapToStoredId(FlowId flowId) => StoredId.Create(_storedType, flowId.Instance.Value);
 

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

@@ -0,0 +1,14 @@
+using System.Collections.Generic;
+using System.Threading.Tasks;
+
+namespace Cleipnir.ResilientFunctions.CoreRuntime.Watchdogs;
+
+/// <summary>
+/// The slice of <see cref="MessageClearer"/> 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 clearer.
+/// </summary>
+internal interface IMessageClearer
+{
+    Task Clear(IReadOnlyList<long> positions);
+}

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

@@ -0,0 +1,135 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using Cleipnir.ResilientFunctions.Domain.Exceptions;
+using Cleipnir.ResilientFunctions.Messaging;
+
+namespace Cleipnir.ResilientFunctions.CoreRuntime.Watchdogs;
+
+/// <summary>
+/// Tracks the positions this replica has pushed to its flows but not yet cleared from the store, and clears them
+/// (deleting the messages) once a QueueManager reports them handled. The MessageWatchdog uses the ignore-set to
+/// avoid re-fetching pushed messages; QueueManagers call <see cref="Clear"/> to delete handled ones.
+/// </summary>
+internal sealed class MessageClearer(
+    IMessageStore messageStore,
+    UnhandledExceptionHandler unhandledExceptionHandler,
+    TimeSpan retryDelay)
+    : IMessageClearer
+{
+    // Positions pushed to this replica's flows but not yet cleared from the store. Handed to the MessageWatchdog
+    // as the ignore-set so they are not re-fetched; trimmed by Clear once their messages are deleted.
+    private readonly HashSet<long> _pushedPositions = new();
+    private readonly Lock _pushedPositionsLock = new();
+
+    // Coalescing delete pipeline (see Clear): the first caller starts the drain, callers that arrive while a
+    // delete is in flight batch up and are flushed together. Guarded by _deleteLock; the drain trims
+    // _pushedPositions under _pushedPositionsLock after each batch's delete lands.
+    private readonly Lock _deleteLock = new();
+    private List<long> _pendingDeletes = new();
+    private TaskCompletionSource _pendingDeletesTcs = new();
+    private bool _draining;
+
+    /// <summary>Records positions just pushed to flows so they are excluded from the next fetch.</summary>
+    public void MarkPushed(IEnumerable<long> positions)
+    {
+        lock (_pushedPositionsLock)
+            foreach (var position in positions)
+                _pushedPositions.Add(position);
+    }
+
+    /// <summary>Snapshot of the not-yet-cleared positions, passed to the store as the fetch ignore-set.</summary>
+    public IReadOnlyList<long> NonClearedPositions()
+    {
+        lock (_pushedPositionsLock)
+            return _pushedPositions.ToList();
+    }
+
+    /// <summary>
+    /// 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. The returned task completes only once that has
+    /// happened for the caller's positions, so the caller knows it will not receive those messages again.
+    ///
+    /// Calls are coalesced: the first one starts the drain and runs immediately, while calls that arrive while a
+    /// delete is in flight are batched and flushed together - collapsing a burst of small deletes into one query.
+    /// Deleting before trimming avoids re-fetching a position that is no longer ignored but not yet gone from the
+    /// store.
+    /// </summary>
+    public Task Clear(IReadOnlyList<long> positions)
+    {
+        if (positions.Count == 0)
+            return Task.CompletedTask;
+
+        Task completion;
+        bool startDraining;
+        lock (_deleteLock)
+        {
+            _pendingDeletes.AddRange(positions);
+            completion = _pendingDeletesTcs.Task;
+            startDraining = !_draining;
+            if (startDraining)
+                _draining = true;
+        }
+
+        if (startDraining)
+            _ = Task.Run(DrainPendingDeletes);
+
+        return completion;
+    }
+
+    // Drains the pending-delete queue one batch at a time until it is empty. Only the drain that set _draining
+    // runs at a time; calls arriving mid-flight just enqueue and are picked up by a later batch. A failing batch
+    // is retried until it lands (callers are told a message is gone only once it truly is), notifying the
+    // unhandled-exception handler on each failure.
+    private async Task DrainPendingDeletes()
+    {
+        while (true)
+        {
+            List<long> pendingDeletes;
+            TaskCompletionSource completionTcs;
+            lock (_deleteLock)
+            {
+                if (_pendingDeletes.Count == 0)
+                {
+                    _draining = false;
+                    return;
+                }
+
+                pendingDeletes = _pendingDeletes;
+                _pendingDeletes = new List<long>();
+                completionTcs = _pendingDeletesTcs;
+                _pendingDeletesTcs = new TaskCompletionSource();
+            }
+
+            // Retry until the delete lands - a failed delete must not fault the callers (they are told the
+            // message is gone only once it truly is). Each failure notifies the unhandled-exception handler.
+            while (true)
+            {
+                try
+                {
+                    await messageStore.DeleteMessages(pendingDeletes);
+                    break;
+                }
+                catch (Exception exception)
+                {
+                    unhandledExceptionHandler.Invoke(
+                        new FrameworkException(
+                            $"{nameof(MessageClearer)} failed to delete handled messages - retrying",
+                            innerException: exception
+                        )
+                    );
+                    await Task.Delay(retryDelay);
+                }
+            }
+
+            lock (_pushedPositionsLock)
+                foreach (var position in pendingDeletes)
+                    _pushedPositions.Remove(position);
+
+            // Offload to the pool so awaiting callers' continuations do not run on the drain loop's thread.
+            _ = Task.Run(completionTcs.SetResult);
+        }
+    }
+}