Route QueueManager message fetch through MessageWatchdog

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

@@ -574,7 +574,7 @@ protected async Task QueueManagerFailsOnMessageDeserializationError(Task<IFuncti
                 var queueManager = new QueueManager(
                     workflow.FlowId,
                     workflow.StoredId,
-                    functionStore.MessageStore,
+                    functionStore.MessageStore.GetMessages,
                     exceptionThrowingSerializer,
                     workflow.Effect,
                     flowState,
@@ -638,7 +638,7 @@ protected async Task RegisteredTimeoutIsRemovedWhenPullingMessage(Task<IFunction
                 var queueManager = new QueueManager(
                     workflow.FlowId,
                     workflow.StoredId,
-                    functionStore.MessageStore,
+                    functionStore.MessageStore.GetMessages,
                     DefaultSerializer.Instance,
                     workflow.Effect,
                     flowState,
@@ -700,7 +700,7 @@ protected async Task PullEnvelopeReturnsEnvelopeWithReceiverAndSender(Task<IFunc
                 var queueManager = new QueueManager(
                     workflow.FlowId,
                     workflow.StoredId,
-                    functionStore.MessageStore,
+                    functionStore.MessageStore.GetMessages,
                     DefaultSerializer.Instance,
                     workflow.Effect,
                     flowState,

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

@@ -27,11 +27,12 @@ internal class InvocationHelper<TParam, TReturn>
     private readonly ReplicaId _replicaId;
     private readonly ResultBusyWaiter<TReturn> _resultBusyWaiter;
     private readonly IMessageClearer _messageClearer;
+    private readonly MessageFetcher _fetchMessages;
     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, IMessageClearer messageClearer)
+    public InvocationHelper(FlowType flowType, StoredType storedType, ReplicaId replicaId, bool isParamlessFunction, SettingsWithDefaults settings, IFunctionStore functionStore, ShutdownCoordinator shutdownCoordinator, ISerializer serializer, UtcNow utcNow, bool clearChildren, IMessageClearer messageClearer, MessageFetcher fetchMessages)
     {
         _flowType = flowType;
         _isParamlessFunction = isParamlessFunction;
@@ -45,6 +46,7 @@ public InvocationHelper(FlowType flowType, StoredType storedType, ReplicaId repl
         _replicaId = replicaId;
         _functionStore = functionStore;
         _messageClearer = messageClearer;
+        _fetchMessages = fetchMessages;
         _resultBusyWaiter = new ResultBusyWaiter<TReturn>(_functionStore, Serializer);
     }
 
@@ -413,7 +415,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, _messageClearer);
+        => new(flowId, storedId, _fetchMessages, Serializer, effect, flowState, unhandledExceptionHandler, timeouts, UtcNow, _settings, _messageClearer);
 
     public StoredId MapToStoredId(FlowId flowId) => StoredId.Create(_storedType, flowId.Instance.Value);
 

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

@@ -1,10 +1,12 @@
 using System;
+using System.Collections.Generic;
 using System.Linq;
 using System.Threading.Tasks;
 using Cleipnir.ResilientFunctions.Domain;
 using Cleipnir.ResilientFunctions.Domain.Exceptions;
 using Cleipnir.ResilientFunctions.Helpers;
 using Cleipnir.ResilientFunctions.Messaging;
+using Cleipnir.ResilientFunctions.Storage;
 
 namespace Cleipnir.ResilientFunctions.CoreRuntime.Watchdogs;
 
@@ -19,6 +21,16 @@ internal class MessageWatchdog(
     TimeSpan delayStartUp,
     UtcNow utcNow)
 {
+    /// <summary>
+    /// On-demand fetch for a single flow, used by its <see cref="Queuing.QueueManager"/> instead of reaching into
+    /// the message store directly - so all <see cref="IMessageStore"/> access stays owned by the watchdog. Unlike
+    /// the poll loop below this does not consult the clearer's pushed-set: the caller passes its own already-fetched
+    /// positions as <paramref name="skipPositions"/>, which keeps the subscribe-time and restart-from-replay fetch
+    /// behaviour identical to the previous in-QueueManager fetch.
+    /// </summary>
+    public Task<IReadOnlyList<StoredMessage>> FetchMessages(StoredId storedId, IReadOnlyList<long> skipPositions)
+        => messageStore.GetMessages(storedId, skipPositions);
+
     public async Task Start()
     {
         await Task.Delay(delayStartUp);

Core/Cleipnir.ResilientFunctions/FunctionsRegistry.cs

@@ -255,7 +255,8 @@ public FuncRegistration<TParam, TReturn> RegisterFunc<TParam, TReturn>(
                 serializer,
                 _settings.UtcNow,
                 settings?.ClearChildrenAfterCapture ?? true,
-                _messageClearer
+                _messageClearer,
+                _messageWatchdog.FetchMessages
             );
             var invoker = new Invoker<TParam, TReturn>(
                 flowType,
@@ -339,7 +340,8 @@ private ParamlessRegistration RegisterParamless(
                 serializer,
                 _settings.UtcNow,
                 settings?.ClearChildrenAfterCapture ?? true,
-                _messageClearer
+                _messageClearer,
+                _messageWatchdog.FetchMessages
             );
             var invoker = new Invoker<Unit, Unit>(
                 flowType,
@@ -423,7 +425,8 @@ public ActionRegistration<TParam> RegisterAction<TParam>(
                 serializer,
                 _settings.UtcNow,
                 settings?.ClearChildrenAfterCapture ?? true,
-                _messageClearer
+                _messageClearer,
+                _messageWatchdog.FetchMessages
             );
             var rActionInvoker = new Invoker<TParam, Unit>(
                 flowType,

Core/Cleipnir.ResilientFunctions/Messaging/MessageFetcher.cs

@@ -0,0 +1,12 @@
+using System.Collections.Generic;
+using System.Threading.Tasks;
+using Cleipnir.ResilientFunctions.Storage;
+
+namespace Cleipnir.ResilientFunctions.Messaging;
+
+/// <summary>
+/// Fetches the not-yet-skipped messages for a single flow. Implemented by the MessageWatchdog so that all
+/// IMessageStore access is owned by the watchdog: the QueueManager pulls on-demand through this delegate
+/// instead of reaching into the message store itself.
+/// </summary>
+public delegate Task<IReadOnlyList<StoredMessage>> MessageFetcher(StoredId storedId, IReadOnlyList<long> skipPositions);

Core/Cleipnir.ResilientFunctions/Queuing/QueueManager.cs

@@ -24,7 +24,7 @@ internal class QueueManager : IDisposable
 
     private readonly FlowId _flowId;
     private readonly StoredId _storedId;
-    private readonly IMessageStore _messageStore;
+    private readonly MessageFetcher _fetchMessages;
     private readonly ISerializer _serializer;
     private readonly Effect _effect;
     private readonly FlowState _flowState;
@@ -49,7 +49,7 @@ internal class QueueManager : IDisposable
     public QueueManager(
         FlowId flowId,
         StoredId storedId,
-        IMessageStore messageStore,
+        MessageFetcher fetchMessages,
         ISerializer serializer,
         Effect effect,
         FlowState flowState,
@@ -63,7 +63,7 @@ public QueueManager(
     {
         _flowId = flowId;
         _storedId = storedId;
-        _messageStore = messageStore;
+        _fetchMessages = fetchMessages;
         _serializer = serializer;
         _effect = effect;
         _flowState = flowState;
@@ -199,7 +199,7 @@ private async Task FetchAndNotify()
             lock (_lock)
                 skipPositions = _fetchedPositions.ToList();
 
-            var messages = await _messageStore.GetMessages(_storedId, skipPositions);
+            var messages = await _fetchMessages(_storedId, skipPositions);
             await ProcessMessages(messages);
         }
         finally