Make message delivery push-only (#180)

* Make MariaDB batch restart claims precise

RestartExecutions claimed via UPDATE..WHERE owner IS NULL and then
SELECTed all requested rows, filtering by owner = replica. A flow
claimed by an earlier call from the same replica was therefore returned
to a concurrent caller as if it had just been claimed, letting two
claimers (e.g. two watchdogs) restart the same flow concurrently -
surfacing as UnexpectedStateException: concurrent modification.

Run a locking SELECT..FOR UPDATE and the claiming UPDATE inside one
transaction so the caller receives exactly the rows its own call
claimed, mirroring the already-precise UPDATE..RETURNING (PostgreSQL)
and UPDATE..OUTPUT (SqlServer) implementations.

* Only claim postponed/suspended flows in batch restarts

The batch RestartExecutions/RestartExecutionsWithoutMessages claimed any
ownerless flow, so a caller targeting a flow that had meanwhile
completed would resurrect it - e.g. a message arriving after its target
succeeded, once message-driven restarts land. Guard the claim with
status IN (Postponed, Suspended) in all stores. The singular
RestartExecution (manual restart) intentionally keeps claiming
completed flows.

* Deliver restart in-hand messages via QueueManager.Push

PrepareForReInvocation received the messages loaded by the restart
claim but discarded them, so a restarted flow had to re-fetch its own
messages from the store. Push them straight into the queue manager's
delivery pipeline instead. Push now initializes the queue manager
first so the idempotency-key state is loaded before processing.

Initialize additionally seeds the fetched-positions set from the
persisted delivered positions: the in-hand messages are loaded at
claim time - before Initialize clears the delivered positions from
the store - so without the seed a message the previous incarnation
already delivered (but had not yet cleared) would be delivered again.

* Push pending messages immediately on flow initialization

Extract the MessageWatchdog's fetch-and-push cycle into PushOnce and
invoke it when a queue manager initializes through CreateQueueClient,
so a freshly started/resumed flow receives its pending messages
immediately instead of waiting for the next watchdog poll.

The pending-push is skipped when initialization is triggered by an
incoming Push: the pusher already holds an in-flight batch and a
nested fetch would stage newer messages ahead of it. Staged messages
are additionally kept position-sorted so delivery order stays append
order even when concurrent pushers stage batches out of order.

* Hand over crashed replicas' messages to a live replica

GetMessagesForReplica only returns messages assigned to the fetching
replica, so messages published by a crashed replica to ownerless flows
were never fetched again once the replica died. The ReplicaWatchdog now
reassigns a crashed replica's message rows to itself alongside
rescheduling its crashed functions - before deleting the crashed
replica from the replica store, so an interrupted handover is retried
on a later iteration.

* Rework hand-rolled QueueManager tests to production paths

Three tests constructed their own QueueManager inside the flow, which
exercised the store-pull directly and duplicated runtime wiring:
- QueueManagerFailsOnMessageDeserializationError now registers the
  exception-throwing serializer on the registry and uses
  workflow.Message
- RegisteredTimeoutIsRemovedWhenPullingMessage now asserts through the
  observable postpone: a delivered message's timeout must not linger
  and shorten a subsequent delay's postpone
- PullEnvelopeReturnsEnvelopeWithReceiverAndSender uses the flow's own
  queue manager via a new internal Workflow.QueueManager accessor

The tests now stay valid when the queue manager's store-pull is
removed and message delivery becomes push-only.

* Reopen pushed positions that cannot be delivered

A push whose target flow was suspended, not yet queue-manager-attached,
or absent from the manager's dictionary was silently dropped while its
positions stayed in the clearer's ignore-set. Later fetches then only
returned higher positions, so a later-positioned duplicate consumed the
idempotency key before the original message - observed on CI as
MultipleIterationsWithDuplicateIdempotencyKeysProcessCorrectly
delivering the second-copy values.

The queue manager now attaches to its flow state at construction, and
undeliverable pushes reopen their positions so they are re-fetched
instead of stranded.

* Reset EffectContext at the start of every flow invocation

The invocation body runs on a Task that captures the scheduler's
execution context. When a flow is restarted from a watchdog chain that
has itself touched the ambient EffectContext, all invocations spawned
from that chain inherit - and mutate - the same context instance, so
implicit effect ids shift between incarnations and replay diverges:
a restarted flow re-subscribes under a different effect id, filters on
the wrong message and suspends, restarting endlessly.

Give every invocation a fresh context at start so implicit ids are
deterministic regardless of the scheduling context.

* Run MessageWatchdog at the messages-pull frequency

MessagesPullFrequency (default 250ms) lost its consumer when the
per-flow fetch loop was removed. The MessageWatchdog is the message-
delivery loop, so it should run at that frequency - the slower
watchdog check frequency made every push-restarted message exchange
poll-bound.

* Restart not-live flows from FlowsManager.Push with in-hand messages

Messages pushed to a flow that is not live are now delivered by
claiming and restarting the flow (RestartExecutionsWithoutMessages)
with the pushed messages handed to the new invocation - the message
itself becomes the wake-up signal instead of relying on the interrupt
flag. A flow-state entry whose flow has suspended counts as not live:
a suspended flow's parked invocation never reaches RemoveFlow by
design, and treating the lingering entry as live swallowed the wake-up.

Claim failures distinguish retryable targets (reopened for the next
poll) from completed/deleted flows, whose positions stay ignored so
their messages are not refetched forever. A failed claim call reopens
everything.

Positions that reach a dying queue manager are reopened rather than
dropped: pushes hitting a disposed instance, staged-but-undelivered
messages at dispose, and messages for unregistered flow types. A
stranded position otherwise loses the flow's wake-up once the message
watchdog has marked it as pushed.

The tests' thread pool is pre-sized since every test's registry now
runs its message poll at 250ms, which starves the default pool growth
on small CI runners.

* Make QueueManager push-only: remove its message-store pull

Delete FetchAndNotify and the queue manager's message-store dependency.
Messages now arrive exclusively via Push - the MessageWatchdog poll,
the immediate fetch at initialization and the restart hand-over - so
Subscribe, Interrupt and FetchMessagesOnce only re-evaluate the already
staged messages against the current subscriptions.

* Stop interrupting target flows when appending messages

Message appends no longer set the interrupted flag: the message row
itself is the durable wake-up signal - the MessageWatchdog fetches it
and either pushes to the live flow or claims and restarts the parked
one with the message in hand. The public Interrupt API and the
suspend-time interrupted-guard remain.

A claim failure for a flow that does not exist (yet) now reopens the
positions instead of stranding them: messages may legally precede
their flow's creation, and the delivery must be retried once the flow
is scheduled.

* Remove the InterruptedWatchdog

Message delivery to live flows is handled by the MessageWatchdog's
push, and parked flows are woken by push-restarts - the interrupt
polling loop no longer had a purpose. Remove it together with its
supporting plumbing: FlowsManagers/FlowsManager Interrupt and
FilterOwned, FlowExecutionState.Interrupt, QueueManager.Interrupt and
the GetInterruptedFunctions/ResetInterrupted store operations. The
public Interrupt API and the suspend-time interrupted-guard remain.

* Remove unused FlowsManager.Schedule

Message-driven flow wake-up is push-restart based; the interrupt-only
scheduling helper has no callers left.

* Give at-least-once tests a load-tolerant suspension delay

The tests rely on workflow.Delay suspending the flow so the captured
work is re-executed on restart. A 10ms delay can fully elapse before
the suspension check on a loaded machine (an effect-persist roundtrip
sits in between), letting the capture complete inline on the first
attempt. Use 500ms so the suspension reliably happens.

* Remove the immediate fetch-and-push at queue manager initialization

The init-time PushOnce ran on the initializing flow's own async branch,
which made restart chains executable from inside another flow's
execution context - the trigger for the EffectContext id-shift bug -
and coupled queue manager construction to the MessageWatchdog. Message
delivery is now driven solely by the watchdog poll and the restart
in-hand hand-over, simplifying the wiring (no push delegate threaded
through FunctionsRegistry/InvocationHelper/QueueManager).

The trade-off is latency: a flow awaiting an already-present message
suspends once and is restarted by the next poll instead of receiving it
inline. Tests encoding the stronger inline guarantee are adjusted:
initial-state flows are awaited tolerantly via Schedule+Completion,
AwaitMessageAfterAppendShouldNotCauseSuspension is removed, and
WorkflowMessageIsIdempotentAcrossRestarts counts deliveries instead of
delivery-timing-dependent invocations.

* Wake the MessageWatchdog on message appends

Add MessageWatchdog.Notify: appending a message completes a wake signal
that cuts the watchdog's inter-poll sleep short, so locally appended
messages are delivered immediately instead of waiting out the poll
interval. The signal is re-armed before each fetch, so a notify arriving
mid-push is never lost - it simply makes the next wait return at once.

Notify only completes the signal: the fetch-and-push always executes on
the watchdog's own loop, so - unlike the removed initialization-time
push - no flow execution context is ever captured by another flow's
restart. Wired into MessageWriter/MessageWriters appends and the
child-to-parent completion message; reopened positions deliberately do
not notify, as a persistently unclaimable flow would otherwise spin the
poll loop.

The faster delivery exposed a stranding hazard: a push whose message
fails deserialization poisons the queue manager without staging the
positions, and when that races the flow's suspension the message stayed
marked forever and the flow never failed. A poisoned queue manager now
reopens the batch's unstaged positions so a restarted incarnation
receives the message and surfaces the failure.

* Pass MessageWatchdog to InvocationHelper instead of an Action

* Pass IMessageClearer to FlowExecutionState instead of an Action

* wip

* Pass MessageWatchdog to MessageWriter and MessageWriters instead of an Action

Author: stidsborg

Core/Cleipnir.ResilientFunctions.Tests/InMemoryTests/RFunctionTests/SuspensionTests.cs

@@ -88,10 +88,6 @@ public override Task ExecutingFlowIsReExecutedWhenSuspendedAfterInterrupt()
     public override Task InterruptSuspendedFlows()
         => InterruptSuspendedFlows(FunctionStoreFactory.Create());
 
-    [TestMethod]
-    public override Task AwaitMessageAfterAppendShouldNotCauseSuspension()
-        => AwaitMessageAfterAppendShouldNotCauseSuspension(FunctionStoreFactory.Create());
-
     [TestMethod]
     public override Task DelayedFlowIsRestartedOnce()
         => DelayedFlowIsRestartedOnce(FunctionStoreFactory.Create());

Core/Cleipnir.ResilientFunctions.Tests/InMemoryTests/StoreTests.cs

@@ -98,10 +98,6 @@ public override Task RestartingExecutionShouldFailWhenExpectedEpochDoesNotMatch(
     public override Task RestartingFunctionShouldSetInterruptedToFalse()
         => RestartingFunctionShouldSetInterruptedToFalse(FunctionStoreFactory.Create());
 
-    [TestMethod]
-    public override Task ResetInterruptedClearsInterruptedFlag()
-        => ResetInterruptedClearsInterruptedFlag(FunctionStoreFactory.Create());
-
     [TestMethod]
     public override Task MessagesCanBeFetchedAfterFunctionWithInitialMessagesHasBeenCreated()
         => MessagesCanBeFetchedAfterFunctionWithInitialMessagesHasBeenCreated(FunctionStoreFactory.Create());
@@ -239,34 +235,30 @@ public override Task SucceedSetsOwnerToNull()
         => SucceedSetsOwnerToNull(FunctionStoreFactory.Create());
 
     [TestMethod]
-    public override Task GetInterruptedFunctionsReturnsOnlyInterruptedFunctions()
-        => GetInterruptedFunctionsReturnsOnlyInterruptedFunctions(FunctionStoreFactory.Create());
-
-    [TestMethod]
-    public override Task GetInterruptedFunctionsReturnsEmptyListWhenNoneExist()
-        => GetInterruptedFunctionsReturnsEmptyListWhenNoneExist(FunctionStoreFactory.Create());
+    public override Task GetResultsReturnsResultsForExistingFunctions()
+        => GetResultsReturnsResultsForExistingFunctions(FunctionStoreFactory.Create());
 
     [TestMethod]
-    public override Task GetInterruptedFunctionsReturnsEmptyListWhenNoneFunctionsAreInterrupted()
-        => GetInterruptedFunctionsReturnsEmptyListWhenNoneFunctionsAreInterrupted(FunctionStoreFactory.Create());
+    public override Task GetResultsReturnsEmptyDictionaryForEmptyInput()
+        => GetResultsReturnsEmptyDictionaryForEmptyInput(FunctionStoreFactory.Create());
 
     [TestMethod]
-    public override Task GetInterruptedFunctionsReturnsIdOnceWhenInterruptedMultipleTimes()
-        => GetInterruptedFunctionsReturnsIdOnceWhenInterruptedMultipleTimes(FunctionStoreFactory.Create());
+    public override Task GetResultsReturnsOnlyExistingFunctionResults()
+        => GetResultsReturnsOnlyExistingFunctionResults(FunctionStoreFactory.Create());
 
     [TestMethod]
-    public override Task GetInterruptedFunctionsIncludesPostponedInterruptedFunction()
-        => GetInterruptedFunctionsIncludesPostponedInterruptedFunction(FunctionStoreFactory.Create());
+    public override Task RestartExecutionsDoesNotReturnFlowClaimedByPreviousCall()
+        => RestartExecutionsDoesNotReturnFlowClaimedByPreviousCall(FunctionStoreFactory.Create());
 
     [TestMethod]
-    public override Task GetResultsReturnsResultsForExistingFunctions()
-        => GetResultsReturnsResultsForExistingFunctions(FunctionStoreFactory.Create());
+    public override Task RestartExecutionsWithoutMessagesDoesNotReturnFlowClaimedByPreviousCall()
+        => RestartExecutionsWithoutMessagesDoesNotReturnFlowClaimedByPreviousCall(FunctionStoreFactory.Create());
 
     [TestMethod]
-    public override Task GetResultsReturnsEmptyDictionaryForEmptyInput()
-        => GetResultsReturnsEmptyDictionaryForEmptyInput(FunctionStoreFactory.Create());
+    public override Task RestartExecutionsDoesNotClaimSucceededFlow()
+        => RestartExecutionsDoesNotClaimSucceededFlow(FunctionStoreFactory.Create());
 
     [TestMethod]
-    public override Task GetResultsReturnsOnlyExistingFunctionResults()
-        => GetResultsReturnsOnlyExistingFunctionResults(FunctionStoreFactory.Create());
+    public override Task RestartExecutionsClaimsSuspendedFlow()
+        => RestartExecutionsClaimsSuspendedFlow(FunctionStoreFactory.Create());
 }

Core/Cleipnir.ResilientFunctions.Tests/InMemoryTests/WatchDogsTests/ReplicaWatchdogTests.cs

@@ -53,4 +53,8 @@ public override Task WorkIsDividedBetweenReplicas()
     [TestMethod]
     public override Task ReplicaCrashedFunctionIsTakenOverByOtherReplica()
         => ReplicaCrashedFunctionIsTakenOverByOtherReplica(FunctionStoreFactory.Create());
+
+    [TestMethod]
+    public override Task CrashedReplicasMessagesAreTakenOverByLiveReplica()
+        => CrashedReplicasMessagesAreTakenOverByLiveReplica(FunctionStoreFactory.Create());
 }

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

@@ -637,13 +637,12 @@ protected async Task AppendedMultipleMessagesAtOnceCanBeFetchedAgain(Task<IFunct
         id2Msgs[0].MessageType.ShouldBe(stringType);
         id2Msgs[0].MessageContent.ShouldBe(msg1.ToJsonByteArray());
 
+        // appending messages no longer interrupts the target flows - the MessageWatchdog push delivers
         var sf1 = await functionStore.GetFunction(id1).ShouldNotBeNullAsync();
-        sf1.Interrupted.ShouldBeTrue();
-        sf1.Expires.ShouldBe(0);
-        
+        sf1.Interrupted.ShouldBeFalse();
+
         var sf2 = await functionStore.GetFunction(id2).ShouldNotBeNullAsync();
-        sf2.Interrupted.ShouldBeTrue();
-        sf2.Expires.ShouldBe(0);
+        sf2.Interrupted.ShouldBeFalse();
     }   
 
     public abstract Task AppendedBatchedMessageCanBeFetchedAgain();
@@ -676,9 +675,9 @@ [new StoredMessage(msg.ToJsonByteArray(), stringType, Replica: ReplicaId.Empty,
         messages[0].MessageType.ShouldBe(stringType);
         messages[0].MessageContent.ShouldBe(msg.ToJsonByteArray());
 
+        // appending messages no longer interrupts the target flow - the MessageWatchdog push delivers
         var sf = await functionStore.GetFunction(id).ShouldNotBeNullAsync();
-        sf.Interrupted.ShouldBeTrue();
-        sf.Expires.ShouldBe(0);
+        sf.Interrupted.ShouldBeFalse();
     }
 
     public abstract Task AppendedBatchedMessagesWithPositionCanBeFetchedAgain();
@@ -739,11 +738,12 @@ await messageStore.AppendMessages(
         messagesId2[0].MessageType.ShouldBe(stringType);
         messagesId2[0].MessageContent.ToStringFromUtf8Bytes().ShouldBe(msg1String);
 
+        // appending messages no longer interrupts the target flows - the MessageWatchdog push delivers
         var sf1 = await functionStore.GetFunction(id1).ShouldNotBeNullAsync();
-        sf1.Interrupted.ShouldBeTrue();
+        sf1.Interrupted.ShouldBeFalse();
 
         var sf2 = await functionStore.GetFunction(id2).ShouldNotBeNullAsync();
-        sf2.Interrupted.ShouldBeTrue();
+        sf2.Interrupted.ShouldBeFalse();
 
         await messageStore.AppendMessages(
             [
@@ -761,8 +761,7 @@ await messageStore.AppendMessages(
         messagesId2[1].MessageContent.ToStringFromUtf8Bytes().ShouldBe(msg2String);
 
         sf2 = await functionStore.GetFunction(id2).ShouldNotBeNullAsync();
-        sf2.Interrupted.ShouldBeTrue();
-        sf2.Expires.ShouldBe(0);
+        sf2.Interrupted.ShouldBeFalse();
     }  
 
     public abstract Task MessagesForMultipleStoreIdsCanBeFetched();

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

@@ -18,17 +18,6 @@ namespace Cleipnir.ResilientFunctions.Tests.Messaging.TestTemplates;
 
 public abstract class MessagesSubscriptionTests
 {
-    // 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 IMessageClearer StubMessageClearer = new NoopMessageClearer();
-    private static readonly ClusterInfo StubClusterInfo = new(ReplicaId.NewId());
-
-    private sealed class NoopMessageClearer : IMessageClearer
-    {
-        public Task Clear(IReadOnlyList<long> positions) => Task.CompletedTask;
-        public void ReopenPositions(IEnumerable<long> positions) { }
-    }
-
     public abstract Task EventsSubscriptionSunshineScenario();
     protected async Task EventsSubscriptionSunshineScenario(Task<IFunctionStore> functionStoreTask)
     {
@@ -558,42 +547,17 @@ protected async Task QueueManagerFailsOnMessageDeserializationError(Task<IFuncti
     {
         var functionStore = await functionStoreTask;
         var unhandledExceptionCatcher = new UnhandledExceptionCatcher();
-        var unhandledExceptionHandler = new UnhandledExceptionHandler(unhandledExceptionCatcher.Catch);
         var exceptionThrowingSerializer = new ExceptionThrowingEventSerializer(typeof(BadMessage));
         using var functionsRegistry = new FunctionsRegistry(
             functionStore,
-            new Settings(unhandledExceptionCatcher.Catch)
+            new Settings(unhandledExceptionCatcher.Catch, serializer: exceptionThrowingSerializer)
         );
 
         var rFunc = functionsRegistry.RegisterFunc(
             nameof(QueueManagerFailsOnMessageDeserializationError),
             inner: async Task<string> (string _, Workflow workflow) =>
             {
-
-                var flowTimeouts = new FlowTimeouts();
-                var flowsManager = new FlowsManager(functionStore, StubMessageClearer, StubClusterInfo);
-                var flowState = flowsManager.CreateFlowState(workflow.StoredId, flowTimeouts, completed: ForeverTask.Instance);
-                var queueManager = new QueueManager(
-                    workflow.FlowId,
-                    workflow.StoredId,
-                    functionStore.MessageStore,
-                    exceptionThrowingSerializer,
-                    workflow.Effect,
-                    flowState,
-                    unhandledExceptionHandler,
-                    flowTimeouts,
-                    () => DateTime.UtcNow,
-                    SettingsWithDefaults.Default,
-                    StubMessageClearer
-                );
-
-                var queueClient = await queueManager.CreateQueueClient();
-
-                var message = await queueClient.Pull<GoodMessage>(
-                    workflow,
-                    workflow.Effect.CreateNextImplicitId()
-                );
-
+                var message = await workflow.Message<GoodMessage>();
                 return message.Value;
             }
         );
@@ -622,61 +586,36 @@ protected async Task RegisteredTimeoutIsRemovedWhenPullingMessage(Task<IFunction
     {
         var functionStore = await functionStoreTask;
         var unhandledExceptionCatcher = new UnhandledExceptionCatcher();
-        var unhandledExceptionHandler = new UnhandledExceptionHandler(unhandledExceptionCatcher.Catch);
         using var functionsRegistry = new FunctionsRegistry(
             functionStore,
             new Settings(unhandledExceptionCatcher.Catch)
         );
 
-        StoredId? storedId = null;
         var rFunc = functionsRegistry.RegisterFunc(
             nameof(RegisteredTimeoutIsRemovedWhenPullingMessage),
             inner: async Task<string> (string _, Workflow workflow) =>
             {
-                storedId = workflow.StoredId;
-                var minimumTimeout = new FlowTimeouts();
-                var flowsManager = new FlowsManager(functionStore, StubMessageClearer, StubClusterInfo);
-                var flowState = flowsManager.CreateFlowState(workflow.StoredId, minimumTimeout, completed: ForeverTask.Instance);
-                var queueManager = new QueueManager(
-                    workflow.FlowId,
-                    workflow.StoredId,
-                    functionStore.MessageStore,
-                    DefaultSerializer.Instance,
-                    workflow.Effect,
-                    flowState,
-                    unhandledExceptionHandler,
-                    minimumTimeout,
-                    () => DateTime.UtcNow,
-                    SettingsWithDefaults.Default,
-                    StubMessageClearer
-                );
-
-
-                var queueClient = await queueManager.CreateQueueClient();
-
-                // Verify timeout is not set before pull
-                minimumTimeout.MinimumTimeout.ShouldBeNull();
-
-                var message = await queueClient.Pull<string>(
-                    workflow,
-                    workflow.Effect.CreateNextImplicitId(),
-                    timeout: TimeSpan.FromMinutes(5)
-                );
-
-                // Verify timeout is removed after successful pull
-                minimumTimeout.MinimumTimeout.ShouldBeNull();
-
+                // Pull a message with a 1-hour timeout (which registers the timeout), then suspend on a 1-day delay.
+                // Once the message is delivered the 1-hour timeout must be removed, leaving the 1-day delay as the
+                // only (minimum) timeout - so the flow is postponed ~1 day out. If the message timeout lingered the
+                // postpone would instead be ~1 hour.
+                var message = await workflow.Message<string>(TimeSpan.FromHours(1));
+                await workflow.Delay(TimeSpan.FromDays(1));
                 return message!;
             }
         );
 
-        var scheduled = await rFunc.Schedule("instanceId", "");
+        await rFunc.Schedule("instanceId", "");
         var messageWriter = rFunc.MessageWriters.For("instanceId".ToFlowInstance());
         await messageWriter.AppendMessage("test message");
 
-        var result = await scheduled.Completion(timeout: TimeSpan.FromSeconds(5));
-        result.ShouldBe("test message");
+        var controlPanel = await rFunc.ControlPanel("instanceId").ShouldNotBeNullAsync();
+        await controlPanel.BusyWaitUntil(
+            c => c.Status == Status.Postponed && c.PostponedUntil > DateTime.UtcNow + TimeSpan.FromHours(12),
+            maxWait: TimeSpan.FromSeconds(10)
+        );
 
+        controlPanel.PostponedUntil!.Value.ShouldBeGreaterThan(DateTime.UtcNow + TimeSpan.FromHours(23));
         unhandledExceptionCatcher.ShouldNotHaveExceptions();
     }
 
@@ -685,7 +624,6 @@ protected async Task PullEnvelopeReturnsEnvelopeWithReceiverAndSender(Task<IFunc
     {
         var functionStore = await functionStoreTask;
         var unhandledExceptionCatcher = new UnhandledExceptionCatcher();
-        var unhandledExceptionHandler = new UnhandledExceptionHandler(unhandledExceptionCatcher.Catch);
         using var functionsRegistry = new FunctionsRegistry(
             functionStore,
             new Settings(unhandledExceptionCatcher.Catch)
@@ -695,27 +633,9 @@ protected async Task PullEnvelopeReturnsEnvelopeWithReceiverAndSender(Task<IFunc
             nameof(PullEnvelopeReturnsEnvelopeWithReceiverAndSender),
             inner: async Task<string> (string _, Workflow workflow) =>
             {
-
-                var flowTimeouts = new FlowTimeouts();
-                var flowsManager = new FlowsManager(functionStore, StubMessageClearer, StubClusterInfo);
-                var flowState = flowsManager.CreateFlowState(workflow.StoredId, flowTimeouts, completed: ForeverTask.Instance);
-                var queueManager = new QueueManager(
-                    workflow.FlowId,
-                    workflow.StoredId,
-                    functionStore.MessageStore,
-                    DefaultSerializer.Instance,
-                    workflow.Effect,
-                    flowState,
-                    unhandledExceptionHandler,
-                    flowTimeouts,
-                    () => DateTime.UtcNow,
-                    SettingsWithDefaults.Default,
-                    StubMessageClearer
-                );
-
-                var queueClient = await queueManager.CreateQueueClient();
-
-                // Pull envelope for specific receiver
+                // Use the flow's own (production) queue manager - not a hand-rolled instance - to pull the envelope
+                // and read its receiver/sender metadata.
+                var queueClient = await workflow.QueueManager.CreateQueueClient();
                 var envelope = await queueClient.PullEnvelope<string>(
                     workflow,
                     workflow.Effect.CreateNextImplicitId(),

Core/Cleipnir.ResilientFunctions.Tests/TestSetup.cs

@@ -1,3 +1,19 @@
+using System.Runtime.CompilerServices;
+using System.Threading;
 using Microsoft.VisualStudio.TestTools.UnitTesting;
 
 [assembly: Parallelize(Workers = 0, Scope = ExecutionScope.MethodLevel)]
+
+internal static class TestModuleInitializer
+{
+    [ModuleInitializer]
+    public static void Init()
+    {
+        // Every test spins up FunctionsRegistry instances each running several Task.Delay-based watchdogs. Under a
+        // small-core CI runner the thread pool grows only ~1 thread/sec, so these continuations - and the tests' own
+        // timeout timers - queue behind pool growth, spiking message-delivery latency. Pre-size the pool so work is
+        // serviced immediately instead of waiting for the pool to grow.
+        ThreadPool.GetMinThreads(out _, out var completionPortThreads);
+        ThreadPool.SetMinThreads(workerThreads: 200, completionPortThreads);
+    }
+}

Core/Cleipnir.ResilientFunctions.Tests/TestTemplates/FunctionTests/AtLeastOnceWorkStatusAndResultTests.cs

@@ -64,7 +64,7 @@ async Task<string>(string param, Workflow workflow) =>
                             counter.Increment();
                             if (counter.Current == 1)
                             {
-                                await workflow.Delay(TimeSpan.FromMilliseconds(10));
+                                await workflow.Delay(TimeSpan.FromMilliseconds(500));
                                 return "nothing";
                             }
 
@@ -105,7 +105,7 @@ async Task<Person>(string param, Workflow workflow) =>
                             counter.Increment();
                             if (counter.Current == 1)
                             {
-                                await workflow.Delay(TimeSpan.FromMilliseconds(10));
+                                await workflow.Delay(TimeSpan.FromMilliseconds(500));
                                 return null!;
                             }
 

Core/Cleipnir.ResilientFunctions.Tests/TestTemplates/FunctionTests/AtLeastOnceWorkStatusTests.cs

@@ -31,7 +31,7 @@ await workflow.Effect
                         {
                             counter.Increment();
                             if (counter.Current == 1)
-                                await workflow.Delay(TimeSpan.FromMilliseconds(10));
+                                await workflow.Delay(TimeSpan.FromMilliseconds(500));
                         }
                     );
             });
@@ -65,7 +65,7 @@ await workflow.Effect
                         {
                             counter.Increment();
                             if (counter.Current == 1)
-                                await workflow.Delay(TimeSpan.FromMilliseconds(10));
+                                await workflow.Delay(TimeSpan.FromMilliseconds(500));
                             return Task.CompletedTask;
                         }
                     );