Skip to content

Commit 06f0bf4

Browse files
stidsborgstidsborg
committed
Make QueueManager push-only: remove its message-store dependency
The QueueManager no longer pulls from the message store; it is fed solely by pushed messages (the MessageWatchdog poll and the restart hand-over). - QueueManager: drop _messageStore and FetchAndNotify; Subscribe, Interrupt and FetchMessagesOnce are now deliver-only (re-evaluate already-pushed messages). - FlowsManager.Push routes not-live (suspended / removed-from-_dict) flows through RestartExecutions instead of silently dropping their messages. - FlowExecutionState.Push reopens the clearer positions when a push lands on a suspended flow, so they are re-fetched and restarted rather than orphaned. - InvocationHelper.CreateQueueManager no longer passes the message store. - Workflow exposes its production QueueManager (internal) so tests can reach it. Tests: - Rewrite the three hand-rolled QueueManager tests to use the production path: registry-level serializer for the deserialization-failure case; observe message-timeout removal via the resulting postpone duration; pull the envelope through the flow's own QueueManager. - Convert the initial-state tests to Schedule + tolerant Completion (messages now arrive via watchdog restart rather than a synchronous first-invocation fetch). - Give WorkflowMessageIsIdempotentAcrossRestarts a test-local messagesDefaultMaxWaitForCompletion so the first invocation consumes the pre-sent message before suspending. - Remove AwaitMessageAfterAppendShouldNotCauseSuspension: a self-append can no longer be delivered synchronously under push-only.
1 parent 3759ad6 commit 06f0bf4

13 files changed

Lines changed: 80 additions & 176 deletions

File tree

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

Lines changed: 0 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -88,10 +88,6 @@ public override Task ExecutingFlowIsReExecutedWhenSuspendedAfterInterrupt()
8888
public override Task InterruptSuspendedFlows()
8989
=> InterruptSuspendedFlows(FunctionStoreFactory.Create());
9090

91-
[TestMethod]
92-
public override Task AwaitMessageAfterAppendShouldNotCauseSuspension()
93-
=> AwaitMessageAfterAppendShouldNotCauseSuspension(FunctionStoreFactory.Create());
94-
9591
[TestMethod]
9692
public override Task DelayedFlowIsRestartedOnce()
9793
=> DelayedFlowIsRestartedOnce(FunctionStoreFactory.Create());

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

Lines changed: 19 additions & 88 deletions
Original file line numberDiff line numberDiff line change
@@ -558,42 +558,17 @@ protected async Task QueueManagerFailsOnMessageDeserializationError(Task<IFuncti
558558
{
559559
var functionStore = await functionStoreTask;
560560
var unhandledExceptionCatcher = new UnhandledExceptionCatcher();
561-
var unhandledExceptionHandler = new UnhandledExceptionHandler(unhandledExceptionCatcher.Catch);
562561
var exceptionThrowingSerializer = new ExceptionThrowingEventSerializer(typeof(BadMessage));
563562
using var functionsRegistry = new FunctionsRegistry(
564563
functionStore,
565-
new Settings(unhandledExceptionCatcher.Catch)
564+
new Settings(unhandledExceptionCatcher.Catch, serializer: exceptionThrowingSerializer)
566565
);
567566

568567
var rFunc = functionsRegistry.RegisterFunc(
569568
nameof(QueueManagerFailsOnMessageDeserializationError),
570569
inner: async Task<string> (string _, Workflow workflow) =>
571570
{
572-
573-
var flowTimeouts = new FlowTimeouts();
574-
var flowsManager = new FlowsManager(functionStore, StubMessageClearer, StubClusterInfo);
575-
var flowState = flowsManager.CreateFlowState(workflow.StoredId, flowTimeouts, completed: ForeverTask.Instance);
576-
var queueManager = new QueueManager(
577-
workflow.FlowId,
578-
workflow.StoredId,
579-
functionStore.MessageStore,
580-
exceptionThrowingSerializer,
581-
workflow.Effect,
582-
flowState,
583-
unhandledExceptionHandler,
584-
flowTimeouts,
585-
() => DateTime.UtcNow,
586-
SettingsWithDefaults.Default,
587-
StubMessageClearer
588-
);
589-
590-
var queueClient = await queueManager.CreateQueueClient();
591-
592-
var message = await queueClient.Pull<GoodMessage>(
593-
workflow,
594-
workflow.Effect.CreateNextImplicitId()
595-
);
596-
571+
var message = await workflow.Message<GoodMessage>();
597572
return message.Value;
598573
}
599574
);
@@ -622,61 +597,36 @@ protected async Task RegisteredTimeoutIsRemovedWhenPullingMessage(Task<IFunction
622597
{
623598
var functionStore = await functionStoreTask;
624599
var unhandledExceptionCatcher = new UnhandledExceptionCatcher();
625-
var unhandledExceptionHandler = new UnhandledExceptionHandler(unhandledExceptionCatcher.Catch);
626600
using var functionsRegistry = new FunctionsRegistry(
627601
functionStore,
628602
new Settings(unhandledExceptionCatcher.Catch)
629603
);
630604

631-
StoredId? storedId = null;
632605
var rFunc = functionsRegistry.RegisterFunc(
633606
nameof(RegisteredTimeoutIsRemovedWhenPullingMessage),
634607
inner: async Task<string> (string _, Workflow workflow) =>
635608
{
636-
storedId = workflow.StoredId;
637-
var minimumTimeout = new FlowTimeouts();
638-
var flowsManager = new FlowsManager(functionStore, StubMessageClearer, StubClusterInfo);
639-
var flowState = flowsManager.CreateFlowState(workflow.StoredId, minimumTimeout, completed: ForeverTask.Instance);
640-
var queueManager = new QueueManager(
641-
workflow.FlowId,
642-
workflow.StoredId,
643-
functionStore.MessageStore,
644-
DefaultSerializer.Instance,
645-
workflow.Effect,
646-
flowState,
647-
unhandledExceptionHandler,
648-
minimumTimeout,
649-
() => DateTime.UtcNow,
650-
SettingsWithDefaults.Default,
651-
StubMessageClearer
652-
);
653-
654-
655-
var queueClient = await queueManager.CreateQueueClient();
656-
657-
// Verify timeout is not set before pull
658-
minimumTimeout.MinimumTimeout.ShouldBeNull();
659-
660-
var message = await queueClient.Pull<string>(
661-
workflow,
662-
workflow.Effect.CreateNextImplicitId(),
663-
timeout: TimeSpan.FromMinutes(5)
664-
);
665-
666-
// Verify timeout is removed after successful pull
667-
minimumTimeout.MinimumTimeout.ShouldBeNull();
668-
609+
// Pull a message with a 1-hour timeout (which registers the timeout), then suspend on a 1-day delay.
610+
// Once the message is delivered the 1-hour timeout must be removed, leaving the 1-day delay as the
611+
// only (minimum) timeout - so the flow is postponed ~1 day out. If the message timeout lingered the
612+
// postpone would instead be ~1 hour.
613+
var message = await workflow.Message<string>(TimeSpan.FromHours(1));
614+
await workflow.Delay(TimeSpan.FromDays(1));
669615
return message!;
670616
}
671617
);
672618

673-
var scheduled = await rFunc.Schedule("instanceId", "");
619+
await rFunc.Schedule("instanceId", "");
674620
var messageWriter = rFunc.MessageWriters.For("instanceId".ToFlowInstance());
675621
await messageWriter.AppendMessage("test message");
676622

677-
var result = await scheduled.Completion(timeout: TimeSpan.FromSeconds(5));
678-
result.ShouldBe("test message");
623+
var controlPanel = await rFunc.ControlPanel("instanceId").ShouldNotBeNullAsync();
624+
await controlPanel.BusyWaitUntil(
625+
c => c.Status == Status.Postponed && c.PostponedUntil > DateTime.UtcNow + TimeSpan.FromHours(12),
626+
maxWait: TimeSpan.FromSeconds(10)
627+
);
679628

629+
controlPanel.PostponedUntil!.Value.ShouldBeGreaterThan(DateTime.UtcNow + TimeSpan.FromHours(23));
680630
unhandledExceptionCatcher.ShouldNotHaveExceptions();
681631
}
682632

@@ -685,7 +635,6 @@ protected async Task PullEnvelopeReturnsEnvelopeWithReceiverAndSender(Task<IFunc
685635
{
686636
var functionStore = await functionStoreTask;
687637
var unhandledExceptionCatcher = new UnhandledExceptionCatcher();
688-
var unhandledExceptionHandler = new UnhandledExceptionHandler(unhandledExceptionCatcher.Catch);
689638
using var functionsRegistry = new FunctionsRegistry(
690639
functionStore,
691640
new Settings(unhandledExceptionCatcher.Catch)
@@ -695,27 +644,9 @@ protected async Task PullEnvelopeReturnsEnvelopeWithReceiverAndSender(Task<IFunc
695644
nameof(PullEnvelopeReturnsEnvelopeWithReceiverAndSender),
696645
inner: async Task<string> (string _, Workflow workflow) =>
697646
{
698-
699-
var flowTimeouts = new FlowTimeouts();
700-
var flowsManager = new FlowsManager(functionStore, StubMessageClearer, StubClusterInfo);
701-
var flowState = flowsManager.CreateFlowState(workflow.StoredId, flowTimeouts, completed: ForeverTask.Instance);
702-
var queueManager = new QueueManager(
703-
workflow.FlowId,
704-
workflow.StoredId,
705-
functionStore.MessageStore,
706-
DefaultSerializer.Instance,
707-
workflow.Effect,
708-
flowState,
709-
unhandledExceptionHandler,
710-
flowTimeouts,
711-
() => DateTime.UtcNow,
712-
SettingsWithDefaults.Default,
713-
StubMessageClearer
714-
);
715-
716-
var queueClient = await queueManager.CreateQueueClient();
717-
718-
// Pull envelope for specific receiver
647+
// Use the flow's own (production) queue manager - not a hand-rolled instance - to pull the envelope
648+
// and read its receiver/sender metadata.
649+
var queueClient = await workflow.QueueManager.CreateQueueClient();
719650
var envelope = await queueClient.PullEnvelope<string>(
720651
workflow,
721652
workflow.Effect.CreateNextImplicitId(),
@@ -730,7 +661,7 @@ protected async Task PullEnvelopeReturnsEnvelopeWithReceiverAndSender(Task<IFunc
730661
var messageWriter = rFunc.MessageWriters.For("instanceId".ToFlowInstance());
731662
await messageWriter.AppendMessage("test message", receiver: "receiver1", sender: "sender1");
732663

733-
var result = await scheduled.Completion(timeout: TimeSpan.FromSeconds(5));
664+
var result = await scheduled.Completion(timeout: TimeSpan.FromSeconds(10));
734665
result.ShouldBe("test message|receiver1|sender1");
735666

736667
unhandledExceptionCatcher.ShouldNotHaveExceptions();

‎Core/Cleipnir.ResilientFunctions.Tests/TestTemplates/FunctionTests/SunshineTests.cs‎

Lines changed: 18 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -351,15 +351,19 @@ public async Task FuncCanBeCreatedWithInitialState(Task<IFunctionStore> storeTas
351351
});
352352

353353

354-
await registration.Run(
354+
var scheduled = await registration.Schedule(
355355
flowInstance: "hello",
356356
param: "hello",
357357
initialState: new InitialState(
358358
Messages: [new MessageAndIdempotencyKey("InitialMessage")],
359359
Effects: [new InitialEffect(Id: 0.ToEffectId(), Value: "InitialEffectValue", Exception: null, Alias: "InitialEffectId")]
360360
)
361361
);
362-
362+
363+
// The initial message is delivered via the MessageWatchdog after a suspend/restart, so wait for completion
364+
// tolerantly instead of Run (which throws on the first suspension).
365+
await scheduled.Completion(TimeSpan.FromSeconds(10));
366+
363367
initialEffectValue.ShouldBe("InitialEffectValue");
364368
initialMessageValue.ShouldBe("InitialMessage");
365369

@@ -389,15 +393,19 @@ public async Task ActionCanBeCreatedWithInitialState(Task<IFunctionStore> storeT
389393
});
390394

391395

392-
await registration.Run(
396+
var scheduled = await registration.Schedule(
393397
flowInstance: "hello",
394398
param: "hello",
395399
initialState: new InitialState(
396400
Messages: [new MessageAndIdempotencyKey("InitialMessage")],
397401
Effects: [new InitialEffect(Id: 0.ToEffectId(), Value: "InitialEffectValue", Exception: null, Alias: "InitialEffectId")]
398402
)
399403
);
400-
404+
405+
// The initial message is delivered via the MessageWatchdog after a suspend/restart, so wait for completion
406+
// tolerantly instead of Run (which throws on the first suspension).
407+
await scheduled.Completion(TimeSpan.FromSeconds(10));
408+
401409
initialEffectValue.ShouldBe("InitialEffectValue");
402410
initialMessageValue.ShouldBe("InitialMessage");
403411

@@ -427,14 +435,18 @@ public async Task ParamlessCanBeCreatedWithInitialState(Task<IFunctionStore> sto
427435
});
428436

429437

430-
await registration.Run(
438+
var scheduled = await registration.Schedule(
431439
flowInstance: "hello",
432440
initialState: new InitialState(
433441
Messages: [new MessageAndIdempotencyKey("InitialMessage")],
434442
Effects: [new InitialEffect(Id: 0.ToEffectId(), Value: "InitialEffectValue", Exception: null, Alias: "InitialEffectId")]
435443
)
436444
);
437-
445+
446+
// The initial message is delivered via the MessageWatchdog after a suspend/restart, so wait for completion
447+
// tolerantly instead of Run (which throws on the first suspension).
448+
await scheduled.Completion(TimeSpan.FromSeconds(10));
449+
438450
initialEffectValue.ShouldBe("InitialEffectValue");
439451
initialMessageValue.ShouldBe("InitialMessage");
440452

‎Core/Cleipnir.ResilientFunctions.Tests/TestTemplates/FunctionTests/SuspensionTests.cs‎

Lines changed: 0 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -721,36 +721,6 @@ protected async Task InterruptSuspendedFlows(Task<IFunctionStore> storeTask)
721721
unhandledExceptionHandler.ShouldNotHaveExceptions();
722722
}
723723

724-
public abstract Task AwaitMessageAfterAppendShouldNotCauseSuspension();
725-
protected async Task AwaitMessageAfterAppendShouldNotCauseSuspension(Task<IFunctionStore> storeTask)
726-
{
727-
var store = await storeTask;
728-
var id = TestFlowId.Create();
729-
var (flowType, flowInstance) = id;
730-
731-
var unhandledExceptionHandler = new UnhandledExceptionCatcher();
732-
using var functionsRegistry = new FunctionsRegistry
733-
(
734-
store,
735-
new Settings(unhandledExceptionHandler.Catch)
736-
);
737-
738-
var registration = functionsRegistry.RegisterFunc<string, string>(
739-
flowType,
740-
inner: async Task<string> (param, workflow) =>
741-
{
742-
await workflow.AppendMessage(param);
743-
744-
return await workflow.Message<string>();
745-
}
746-
);
747-
748-
var result = await registration.Run("SomeInstance", "Hello World");
749-
result.ShouldBe("Hello World");
750-
751-
unhandledExceptionHandler.ShouldNotHaveExceptions();
752-
}
753-
754724
public abstract Task DelayedFlowIsRestartedOnce();
755725
protected async Task DelayedFlowIsRestartedOnce(Task<IFunctionStore> storeTask)
756726
{

‎Core/Cleipnir.ResilientFunctions.Tests/TestTemplates/FunctionTests/WorkflowMessageTests.cs‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -422,7 +422,9 @@ public async Task WorkflowMessageIsIdempotentAcrossRestarts(Task<IFunctionStore>
422422
var unhandledExceptionHandler = new UnhandledExceptionCatcher();
423423
using var functionsRegistry = new FunctionsRegistry(
424424
store,
425-
new Settings(unhandledExceptionHandler.Catch)
425+
// Keep the first invocation running long enough for the watchdog poll to deliver the pre-sent message
426+
// before it suspends, so it consumes the message and fails on invocation 1 as the test expects.
427+
new Settings(unhandledExceptionHandler.Catch, messagesDefaultMaxWaitForCompletion: TimeSpan.FromSeconds(1))
426428
);
427429

428430
var invocationCount = 0;

‎Core/Cleipnir.ResilientFunctions/CoreRuntime/FlowExecutionState.cs‎

Lines changed: 15 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,6 @@
1+
using System;
12
using System.Collections.Generic;
3+
using System.Linq;
24
using System.Threading;
35
using System.Threading.Tasks;
46
using Cleipnir.ResilientFunctions.Helpers;
@@ -19,6 +21,7 @@ public class FlowExecutionState
1921
{
2022
private readonly Lock _lock = new();
2123
private readonly TaskCompletionSource _suspendedTcs = new();
24+
private readonly Action<IEnumerable<long>>? _reopenPositions;
2225

2326
public StoredId Id { get; }
2427
public int Subflows { get; private set; }
@@ -49,13 +52,15 @@ public FlowExecutionState(
4952
int subflows,
5053
int waitingSubflows,
5154
FlowTimeouts timeouts,
52-
Task completed)
55+
Task completed,
56+
Action<IEnumerable<long>>? reopenPositions = null)
5357
{
5458
Id = id;
5559
Subflows = subflows;
5660
WaitingSubflows = waitingSubflows;
5761
Timeouts = timeouts;
5862
SuspendedTask = _suspendedTcs.Task;
63+
_reopenPositions = reopenPositions;
5964

6065
_ = completed.ContinueWith(_ => Status = FlowStatus.Completed);
6166
}
@@ -103,7 +108,15 @@ public void Interrupt()
103108

104109
public Task Push(IReadOnlyList<StoredMessage> messages)
105110
{
106-
if (Suspended) return Task.CompletedTask;
111+
if (Suspended)
112+
{
113+
// Lost the suspend race: this flow is suspended but still in _dict, so the push was routed here rather
114+
// than to RestartExecutions. Reopen the positions so the watchdog re-fetches them once the flow has left
115+
// _dict - they then route through RestartExecutions and get delivered instead of being orphaned.
116+
_reopenPositions?.Invoke(messages.Select(m => m.Position));
117+
return Task.CompletedTask;
118+
}
119+
107120
return QueueManager?.Push(messages) ?? Task.CompletedTask;
108121
}
109122

‎Core/Cleipnir.ResilientFunctions/CoreRuntime/FlowsManager.cs‎

Lines changed: 10 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -38,7 +38,7 @@ internal FlowsManager(
3838
public FlowExecutionState CreateFlowState(StoredId id, FlowTimeouts timeouts, Task completed)
3939
{
4040
lock (_lock)
41-
return _dict[id] = new FlowExecutionState(id, subflows: 1, waitingSubflows: 0, timeouts, completed);
41+
return _dict[id] = new FlowExecutionState(id, subflows: 1, waitingSubflows: 0, timeouts, completed, _messageClearer.ReopenPositions);
4242
}
4343

4444
public void RemoveFlow(StoredId id, FlowExecutionState flowExecutionState)
@@ -65,10 +65,16 @@ public void Interrupt(IReadOnlyList<StoredId> ids)
6565
public Task Push(IReadOnlyList<StoredMessages> messagesByFlow)
6666
{
6767
List<Task> tasks = new();
68+
List<StoredMessages> notLive = new();
6869
lock (_lock)
69-
foreach (var (id, messages) in messagesByFlow)
70-
if (_dict.TryGetValue(id, out var flowState))
71-
tasks.Add(flowState.Push(messages));
70+
foreach (var storedMessages in messagesByFlow)
71+
if (_dict.TryGetValue(storedMessages.StoredId, out var flowState))
72+
tasks.Add(flowState.Push(storedMessages.Messages));
73+
else
74+
notLive.Add(storedMessages); // suspended or already removed from _dict - restart to deliver
75+
76+
if (notLive.Count > 0)
77+
tasks.Add(RestartExecutions(notLive));
7278

7379
return Task.WhenAll(tasks);
7480
}

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -413,7 +413,7 @@ public async Task<ExistingEffects> CreateExistingEffects(FlowId flowId)
413413
public ExistingMessages CreateExistingMessages(FlowId flowId) => new(MapToStoredId(flowId), _functionStore.MessageStore, Serializer);
414414

415415
public QueueManager CreateQueueManager(FlowId flowId, StoredId storedId, Effect effect, FlowExecutionState flowExecutionState, FlowTimeouts timeouts, UnhandledExceptionHandler unhandledExceptionHandler)
416-
=> new(flowId, storedId, _functionStore.MessageStore, Serializer, effect, flowExecutionState, unhandledExceptionHandler, timeouts, UtcNow, _settings, _messageClearer);
416+
=> new(flowId, storedId, Serializer, effect, flowExecutionState, unhandledExceptionHandler, timeouts, UtcNow, _settings, _messageClearer);
417417

418418
public StoredId MapToStoredId(FlowId flowId) => StoredId.Create(_storedType, flowId.Instance.Value);
419419

‎Core/Cleipnir.ResilientFunctions/CoreRuntime/Invocation/Workflow.cs‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,8 @@ public class Workflow
1919
private readonly UtcNow _utcNow;
2020
private MessageWriter MessageWriter { get; }
2121

22+
internal QueueManager QueueManager => _queueManager;
23+
2224

2325
internal Workflow(FlowId flowId, StoredId storedId, Effect effect, QueueManager queueManager, UtcNow utcNow, MessageWriter messageWriter)
2426
{

0 commit comments

Comments
 (0)