Depend on IMessageWatchdog so QueueManager can be stubbed in tests

Extract an IMessageWatchdog interface (just RemoveMessages) that MessageWatchdog
implements, and have QueueManager depend on it. The hand-rolled QueueManager
tests now pass a one-line no-op stub instead of constructing a fully wired
MessageWatchdog. Production still injects the real watchdog (non-null).

Author: stidsborg

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

@@ -18,19 +18,14 @@ 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
-        );
+    // 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.
+    private static readonly IMessageWatchdog StubMessageWatchdog = new NoopMessageWatchdog();
+
+    private sealed class NoopMessageWatchdog : IMessageWatchdog
+    {
+        public void RemoveMessages(IReadOnlyList<long> positions) { }
+    }
 
     public abstract Task EventsSubscriptionSunshineScenario();
     protected async Task EventsSubscriptionSunshineScenario(Task<IFunctionStore> functionStoreTask)
@@ -587,7 +582,7 @@ protected async Task QueueManagerFailsOnMessageDeserializationError(Task<IFuncti
                     flowTimeouts,
                     () => DateTime.UtcNow,
                     SettingsWithDefaults.Default,
-                    CreateMessageWatchdog(functionStore, unhandledExceptionHandler)
+                    StubMessageWatchdog
                 );
 
                 var queueClient = await queueManager.CreateQueueClient();
@@ -651,7 +646,7 @@ protected async Task RegisteredTimeoutIsRemovedWhenPullingMessage(Task<IFunction
                     minimumTimeout,
                     () => DateTime.UtcNow,
                     SettingsWithDefaults.Default,
-                    CreateMessageWatchdog(functionStore, unhandledExceptionHandler)
+                    StubMessageWatchdog
                 );
 
 
@@ -713,7 +708,7 @@ protected async Task PullEnvelopeReturnsEnvelopeWithReceiverAndSender(Task<IFunc
                     flowTimeouts,
                     () => DateTime.UtcNow,
                     SettingsWithDefaults.Default,
-                    CreateMessageWatchdog(functionStore, unhandledExceptionHandler)
+                    StubMessageWatchdog
                 );
 
                 var queueClient = await queueManager.CreateQueueClient();

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

@@ -0,0 +1,13 @@
+using System.Collections.Generic;
+
+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
+/// QueueManager can pass a no-op stub instead of a fully wired watchdog.
+/// </summary>
+internal interface IMessageWatchdog
+{
+    void RemoveMessages(IReadOnlyList<long> positions);
+}

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

@@ -10,7 +10,7 @@
 
 namespace Cleipnir.ResilientFunctions.CoreRuntime.Watchdogs;
 
-internal class MessageWatchdog
+internal class MessageWatchdog : IMessageWatchdog
 {
     private readonly IMessageStore _messageStore;
     private readonly FlowsManagers _flowsManagers;

Core/Cleipnir.ResilientFunctions/Queuing/QueueManager.cs

@@ -32,7 +32,7 @@ internal class QueueManager : IDisposable
     private readonly FlowTimeouts _timeouts;
     private readonly UtcNow _utcNow;
     private readonly SettingsWithDefaults _settings;
-    private readonly MessageWatchdog _messageWatchdog;
+    private readonly IMessageWatchdog _messageWatchdog;
     private readonly IdempotencyKeys _idempotencyKeys;
 
     private readonly SemaphoreSlim _initializeSemaphore = new(1);
@@ -57,7 +57,7 @@ public QueueManager(
         FlowTimeouts timeouts,
         UtcNow utcNow,
         SettingsWithDefaults settings,
-        MessageWatchdog messageWatchdog,
+        IMessageWatchdog messageWatchdog,
         int maxIdempotencyKeyCount = 100,
         TimeSpan? maxIdempotencyKeyTtl = null)
     {