Give QueueManager a MessageWatchdog reference to trim handled messages

Thread the registry's MessageWatchdog through InvocationHelper into each
QueueManager via constructor injection (no cyclic dependency: the watchdog is
built once before any InvocationHelper/QueueManager and only reaches a
QueueManager at runtime via FlowsManagers -> FlowState -> Push).

QueueManager now calls MessageWatchdog.RemoveMessages once it deletes messages
from the store (AfterFlush, the Initialize replay, and the idempotency-duplicate
path) so the watchdog drops those positions from its ignore-set instead of
carrying them forever - realizing the existing TODO on _pushedPositions.
RemoveMessages runs on flow threads, so _pushedPositions is now guarded by a lock.

The hand-rolled QueueManager tests in MessagesSubscriptionTests get a shared
CreateMessageWatchdog helper (the watchdog is never started, so RemoveMessages
no-ops against its empty ignore-set).

Author: stidsborg

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

@@ -4,6 +4,7 @@
 using System.Threading.Tasks;
 using Cleipnir.ResilientFunctions.CoreRuntime;
 using Cleipnir.ResilientFunctions.CoreRuntime.Invocation;
+using Cleipnir.ResilientFunctions.CoreRuntime.Watchdogs;
 using Cleipnir.ResilientFunctions.CoreRuntime.Serialization;
 using Cleipnir.ResilientFunctions.Domain;
 using Cleipnir.ResilientFunctions.Helpers;
@@ -17,6 +18,20 @@ namespace Cleipnir.ResilientFunctions.Tests.Messaging.TestTemplates;
 
 public abstract class MessagesSubscriptionTests
 {
+    // These tests hand-roll a QueueManager, which now requires a MessageWatchdog. The watchdog is never
+    // started here, so it only serves QueueManager.RemoveMessages calls (a no-op against its empty ignore-set).
+    private static MessageWatchdog CreateMessageWatchdog(IFunctionStore functionStore, UnhandledExceptionHandler unhandledExceptionHandler)
+        => new(
+            functionStore.MessageStore,
+            new FlowsManagers(functionStore),
+            new ClusterInfo(ReplicaId.NewId()),
+            new ShutdownCoordinator(),
+            unhandledExceptionHandler,
+            checkFrequency: TimeSpan.FromSeconds(1),
+            delayStartUp: TimeSpan.Zero,
+            () => DateTime.UtcNow
+        );
+
     public abstract Task EventsSubscriptionSunshineScenario();
     protected async Task EventsSubscriptionSunshineScenario(Task<IFunctionStore> functionStoreTask)
     {
@@ -571,7 +586,8 @@ protected async Task QueueManagerFailsOnMessageDeserializationError(Task<IFuncti
                     unhandledExceptionHandler,
                     flowTimeouts,
                     () => DateTime.UtcNow,
-                    SettingsWithDefaults.Default
+                    SettingsWithDefaults.Default,
+                    CreateMessageWatchdog(functionStore, unhandledExceptionHandler)
                 );
 
                 var queueClient = await queueManager.CreateQueueClient();
@@ -634,7 +650,8 @@ protected async Task RegisteredTimeoutIsRemovedWhenPullingMessage(Task<IFunction
                     unhandledExceptionHandler,
                     minimumTimeout,
                     () => DateTime.UtcNow,
-                    SettingsWithDefaults.Default
+                    SettingsWithDefaults.Default,
+                    CreateMessageWatchdog(functionStore, unhandledExceptionHandler)
                 );
 
 
@@ -695,7 +712,8 @@ protected async Task PullEnvelopeReturnsEnvelopeWithReceiverAndSender(Task<IFunc
                     unhandledExceptionHandler,
                     flowTimeouts,
                     () => DateTime.UtcNow,
-                    SettingsWithDefaults.Default
+                    SettingsWithDefaults.Default,
+                    CreateMessageWatchdog(functionStore, unhandledExceptionHandler)
                 );
 
                 var queueClient = await queueManager.CreateQueueClient();

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

@@ -4,6 +4,7 @@
 using System.Runtime.ExceptionServices;
 using System.Threading.Tasks;
 using Cleipnir.ResilientFunctions.CoreRuntime.Serialization;
+using Cleipnir.ResilientFunctions.CoreRuntime.Watchdogs;
 using Cleipnir.ResilientFunctions.Domain;
 using Cleipnir.ResilientFunctions.Messaging;
 using Cleipnir.ResilientFunctions.Domain.Exceptions;
@@ -25,11 +26,12 @@ internal class InvocationHelper<TParam, TReturn>
     private readonly StoredType _storedType;
     private readonly ReplicaId _replicaId;
     private readonly ResultBusyWaiter<TReturn> _resultBusyWaiter;
+    private readonly MessageWatchdog _messageWatchdog;
     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)
+    public InvocationHelper(FlowType flowType, StoredType storedType, ReplicaId replicaId, bool isParamlessFunction, SettingsWithDefaults settings, IFunctionStore functionStore, ShutdownCoordinator shutdownCoordinator, ISerializer serializer, UtcNow utcNow, bool clearChildren, MessageWatchdog messageWatchdog)
     {
         _flowType = flowType;
         _isParamlessFunction = isParamlessFunction;
@@ -42,6 +44,7 @@ public InvocationHelper(FlowType flowType, StoredType storedType, ReplicaId repl
         _storedType = storedType;
         _replicaId = replicaId;
         _functionStore = functionStore;
+        _messageWatchdog = messageWatchdog;
         _resultBusyWaiter = new ResultBusyWaiter<TReturn>(_functionStore, Serializer);
     }
 
@@ -410,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);
+        => new(flowId, storedId, _functionStore.MessageStore, Serializer, effect, flowState, unhandledExceptionHandler, timeouts, UtcNow, _settings, _messageWatchdog);
 
     public StoredId MapToStoredId(FlowId flowId) => StoredId.Create(_storedType, flowId.Instance.Value);
 

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

@@ -1,6 +1,7 @@
 using System;
 using System.Collections.Generic;
 using System.Linq;
+using System.Threading;
 using System.Threading.Tasks;
 using Cleipnir.ResilientFunctions.Domain;
 using Cleipnir.ResilientFunctions.Domain.Exceptions;
@@ -21,9 +22,11 @@ internal class MessageWatchdog
     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.
+    // 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.
     private readonly HashSet<long> _pushedPositions = new();
+    private readonly Lock _pushedPositionsLock = new();
 
     public MessageWatchdog(
         IMessageStore messageStore,
@@ -59,12 +62,17 @@ public async Task Start()
                 // Messages destined for flows currently owned by this replica (replica = COALESCE(owner, publisher)).
                 // FlowsManagers.Push routes each group to its flow type's manager and delivers only to live
                 // flows; entries for non-live flows (or unregistered types) are ignored.
-                var messageGroups = await _messageStore.GetMessagesForReplica(_clusterInfo.ReplicaId, _pushedPositions.ToList());
+                List<long> ignorePositions;
+                lock (_pushedPositionsLock)
+                    ignorePositions = _pushedPositions.ToList();
+
+                var messageGroups = await _messageStore.GetMessagesForReplica(_clusterInfo.ReplicaId, ignorePositions);
                 if (messageGroups.Count > 0)
                 {
-                    foreach (var group in messageGroups)
-                        foreach (var message in group.Messages)
-                            _pushedPositions.Add(message.Position);
+                    lock (_pushedPositionsLock)
+                        foreach (var group in messageGroups)
+                            foreach (var message in group.Messages)
+                                _pushedPositions.Add(message.Position);
 
                     await _flowsManagers.Push(messageGroups);
                 }
@@ -88,4 +96,18 @@ public async Task Start()
             goto 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.
+    /// </summary>
+    public void RemoveMessages(IReadOnlyList<long> positions)
+    {
+        if (positions.Count == 0)
+            return;
+
+        lock (_pushedPositionsLock)
+            foreach (var position in positions)
+                _pushedPositions.Remove(position);
+    }
 }

Core/Cleipnir.ResilientFunctions/FunctionsRegistry.cs

@@ -246,7 +246,8 @@ public FuncRegistration<TParam, TReturn> RegisterFunc<TParam, TReturn>(
                 _shutdownCoordinator,
                 serializer,
                 _settings.UtcNow,
-                settings?.ClearChildrenAfterCapture ?? true
+                settings?.ClearChildrenAfterCapture ?? true,
+                _messageWatchdog
             );
             var invoker = new Invoker<TParam, TReturn>(
                 flowType,
@@ -329,7 +330,8 @@ private ParamlessRegistration RegisterParamless(
                 _shutdownCoordinator,
                 serializer,
                 _settings.UtcNow,
-                settings?.ClearChildrenAfterCapture ?? true
+                settings?.ClearChildrenAfterCapture ?? true,
+                _messageWatchdog
             );
             var invoker = new Invoker<Unit, Unit>(
                 flowType,
@@ -412,7 +414,8 @@ public ActionRegistration<TParam> RegisterAction<TParam>(
                 _shutdownCoordinator,
                 serializer,
                 _settings.UtcNow,
-                settings?.ClearChildrenAfterCapture ?? true
+                settings?.ClearChildrenAfterCapture ?? true,
+                _messageWatchdog
             );
             var rActionInvoker = new Invoker<TParam, Unit>(
                 flowType,

Core/Cleipnir.ResilientFunctions/Queuing/QueueManager.cs

@@ -5,6 +5,7 @@
 using System.Threading.Tasks;
 using Cleipnir.ResilientFunctions.CoreRuntime;
 using Cleipnir.ResilientFunctions.CoreRuntime.Serialization;
+using Cleipnir.ResilientFunctions.CoreRuntime.Watchdogs;
 using Cleipnir.ResilientFunctions.Domain;
 using Cleipnir.ResilientFunctions.Domain.Exceptions.Commands;
 using Cleipnir.ResilientFunctions.Helpers;
@@ -31,6 +32,7 @@ internal class QueueManager : IDisposable
     private readonly FlowTimeouts _timeouts;
     private readonly UtcNow _utcNow;
     private readonly SettingsWithDefaults _settings;
+    private readonly MessageWatchdog _messageWatchdog;
     private readonly IdempotencyKeys _idempotencyKeys;
 
     private readonly SemaphoreSlim _initializeSemaphore = new(1);
@@ -55,6 +57,7 @@ public QueueManager(
         FlowTimeouts timeouts,
         UtcNow utcNow,
         SettingsWithDefaults settings,
+        MessageWatchdog messageWatchdog,
         int maxIdempotencyKeyCount = 100,
         TimeSpan? maxIdempotencyKeyTtl = null)
     {
@@ -68,6 +71,7 @@ public QueueManager(
         _timeouts = timeouts;
         _utcNow = utcNow;
         _settings = settings;
+        _messageWatchdog = messageWatchdog;
         _idempotencyKeys = new IdempotencyKeys(IdempotencyKeysRoot, _effect, maxIdempotencyKeyCount, maxIdempotencyKeyTtl, utcNow);
     }
 
@@ -86,6 +90,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);
                 positions.Clear();
                 _effect.FlushlessUpsert(DeliveredPositionsId, positions, alias: null);
             }
@@ -224,6 +229,7 @@ private async Task ProcessMessages(IReadOnlyList<StoredMessage> messages)
                 if (idempotencyKey != null && !_idempotencyKeys.Add(idempotencyKey, position))
                 {
                     await _messageStore.DeleteMessages(_storedId, [position]);
+                    _messageWatchdog.RemoveMessages([position]);
                     continue;
                 }
 
@@ -282,6 +288,7 @@ public async Task AfterFlush()
                 return;
 
             await _messageStore.DeleteMessages(_storedId, deliveredPositions);
+            _messageWatchdog.RemoveMessages(deliveredPositions);
             deliveredPositions.Clear();
             _effect.FlushlessUpsert(DeliveredPositionsId, deliveredPositions, alias: null);
         }