Skip to content

Commit e18b535

Browse files
stidsborgstidsborg
authored
Add message-driven RestartExecutions to FlowsManager (#179)
Move RestartExecutions into the per-type FlowsManager so it can claim and resume flows targeted by fetched messages: - Restart unowned flows via RestartExecutionsWithoutMessages and resume each through an IFlowRestarter (implemented by Invoker) using the in-hand messages - no re-claim or extra store round-trip. - Reopen the message-clearer positions of flows that could not be claimed (dropped from the ignore-set without deleting them from the store). - Add ReopenPositions to IMessageClearer; inject MessageClearer and ClusterInfo (non-nullable) into FlowsManager; wire the restarter in FunctionsRegistry after each Invoker is built.
1 parent d61c3c0 commit e18b535

8 files changed

Lines changed: 152 additions & 29 deletions

File tree

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

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -21,10 +21,12 @@ public abstract class MessagesSubscriptionTests
2121
// These tests hand-roll a QueueManager, which delegates message deletion to IMessageClearer. They don't
2222
// assert on store cleanup, so a no-op stub suffices.
2323
private static readonly IMessageClearer StubMessageClearer = new NoopMessageClearer();
24+
private static readonly ClusterInfo StubClusterInfo = new(ReplicaId.NewId());
2425

2526
private sealed class NoopMessageClearer : IMessageClearer
2627
{
2728
public Task Clear(IReadOnlyList<long> positions) => Task.CompletedTask;
29+
public void ReopenPositions(IEnumerable<long> positions) { }
2830
}
2931

3032
public abstract Task EventsSubscriptionSunshineScenario();
@@ -569,7 +571,7 @@ protected async Task QueueManagerFailsOnMessageDeserializationError(Task<IFuncti
569571
{
570572

571573
var flowTimeouts = new FlowTimeouts();
572-
var flowsManager = new FlowsManager(functionStore);
574+
var flowsManager = new FlowsManager(functionStore, StubMessageClearer, StubClusterInfo);
573575
var flowState = flowsManager.CreateFlowState(workflow.StoredId, flowTimeouts, completed: ForeverTask.Instance);
574576
var queueManager = new QueueManager(
575577
workflow.FlowId,
@@ -633,7 +635,7 @@ protected async Task RegisteredTimeoutIsRemovedWhenPullingMessage(Task<IFunction
633635
{
634636
storedId = workflow.StoredId;
635637
var minimumTimeout = new FlowTimeouts();
636-
var flowsManager = new FlowsManager(functionStore);
638+
var flowsManager = new FlowsManager(functionStore, StubMessageClearer, StubClusterInfo);
637639
var flowState = flowsManager.CreateFlowState(workflow.StoredId, minimumTimeout, completed: ForeverTask.Instance);
638640
var queueManager = new QueueManager(
639641
workflow.FlowId,
@@ -695,7 +697,7 @@ protected async Task PullEnvelopeReturnsEnvelopeWithReceiverAndSender(Task<IFunc
695697
{
696698

697699
var flowTimeouts = new FlowTimeouts();
698-
var flowsManager = new FlowsManager(functionStore);
700+
var flowsManager = new FlowsManager(functionStore, StubMessageClearer, StubClusterInfo);
699701
var flowState = flowsManager.CreateFlowState(workflow.StoredId, flowTimeouts, completed: ForeverTask.Instance);
700702
var queueManager = new QueueManager(
701703
workflow.FlowId,

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

Lines changed: 68 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,9 @@
22
using System.Linq;
33
using System.Threading;
44
using System.Threading.Tasks;
5+
using Cleipnir.ResilientFunctions.CoreRuntime.Invocation;
6+
using Cleipnir.ResilientFunctions.CoreRuntime.Watchdogs;
7+
using Cleipnir.ResilientFunctions.Domain;
58
using Cleipnir.ResilientFunctions.Messaging;
69
using Cleipnir.ResilientFunctions.Storage;
710

@@ -11,9 +14,26 @@ public class FlowsManager
1114
{
1215
private readonly Dictionary<StoredId, FlowExecutionState> _dict = new();
1316
private readonly IFunctionStore _functionStore;
17+
private readonly IMessageClearer _messageClearer;
18+
private readonly ClusterInfo _clusterInfo;
19+
private IFlowRestarter? _restarter;
1420
private readonly Lock _lock = new();
1521

16-
public FlowsManager(IFunctionStore functionStore) => _functionStore = functionStore;
22+
internal FlowsManager(
23+
IFunctionStore functionStore,
24+
IMessageClearer messageClearer,
25+
ClusterInfo clusterInfo)
26+
{
27+
_functionStore = functionStore;
28+
_messageClearer = messageClearer;
29+
_clusterInfo = clusterInfo;
30+
}
31+
32+
/// <summary>
33+
/// Supplies the per-type flow restarter (the Invoker). Late-bound because the Invoker is constructed after this
34+
/// manager during registration.
35+
/// </summary>
36+
internal void SetRestarter(IFlowRestarter restarter) => _restarter = restarter;
1737

1838
public FlowExecutionState CreateFlowState(StoredId id, FlowTimeouts timeouts, Task completed)
1939
{
@@ -53,6 +73,53 @@ public Task Push(IReadOnlyList<StoredMessages> messagesByFlow)
5373
return Task.WhenAll(tasks);
5474
}
5575

76+
/// <summary>
77+
/// Restarts (claims for this replica) the targeted flows that are not already owned, then hands each restarted
78+
/// flow - together with the in-hand messages - to the <see cref="ScheduleRestartFromWatchdog"/> delegate so it
79+
/// resumes executing. Flows that could not be claimed have their positions reopened in the message clearer
80+
/// (dropped from the ignore-set without deleting them from the store, since their actual owner still needs them).
81+
/// </summary>
82+
public async Task RestartExecutions(IEnumerable<StoredMessages> messages)
83+
{
84+
var groups = messages
85+
.GroupBy(m => m.StoredId)
86+
.ToDictionary(g => g.Key, g => g.ToList());
87+
88+
var results = await _functionStore
89+
.RestartExecutionsWithoutMessages(groups.Keys.ToList(), _clusterInfo.ReplicaId);
90+
91+
// Flows that could not be restarted (already owned/running elsewhere, or no longer present) were never
92+
// delivered to here, yet the MessageWatchdog optimistically marked their positions as pushed. Drop those
93+
// positions from the clearer's ignore-set so they can be re-fetched - without deleting them from the store,
94+
// since the actual owner still needs them.
95+
var notRestartedPositions = groups
96+
.Where(kv => !results.ContainsKey(kv.Key))
97+
.SelectMany(kv => kv.Value)
98+
.SelectMany(storedMessages => storedMessages.Messages)
99+
.Select(message => message.Position)
100+
.ToList();
101+
_messageClearer.ReopenPositions(notRestartedPositions);
102+
103+
// Resume each restarted flow, supplying the messages we already hold so it does not re-fetch them. The
104+
// claim + flow snapshot returned by RestartExecutionsWithoutMessages is everything the delegate needs, so
105+
// no further store round-trip or re-claim is performed.
106+
foreach (var (storedId, storedFlowWithEffects) in results)
107+
{
108+
var inHandMessages = groups[storedId]
109+
.SelectMany(storedMessages => storedMessages.Messages)
110+
.ToList();
111+
112+
var restartedFunction = new RestartedFunction(
113+
storedFlowWithEffects.StoredFlow,
114+
storedFlowWithEffects.Effects,
115+
inHandMessages,
116+
storedFlowWithEffects.StorageSession
117+
);
118+
119+
await _restarter!.ScheduleRestart(storedId, restartedFunction, onCompletion: () => { });
120+
}
121+
}
122+
56123
/// <summary>
57124
/// Wakes the target flow so it consumes the just-published message by applying the durable interrupt
58125
/// (interrupted flag + expires=0) - the suspend-race guard and watchdog backstop, so the message is never

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

Lines changed: 29 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,8 @@
22
using System.Linq;
33
using System.Threading;
44
using System.Threading.Tasks;
5+
using Cleipnir.ResilientFunctions.CoreRuntime.Watchdogs;
6+
using Cleipnir.ResilientFunctions.Domain;
57
using Cleipnir.ResilientFunctions.Messaging;
68
using Cleipnir.ResilientFunctions.Storage;
79

@@ -16,10 +18,19 @@ namespace Cleipnir.ResilientFunctions.CoreRuntime;
1618
public class FlowsManagers
1719
{
1820
private readonly Dictionary<StoredType, FlowsManager> _managers = new();
21+
1922
private readonly IFunctionStore _functionStore;
23+
private readonly MessageClearer _messageClearer;
24+
private readonly ClusterInfo _clusterInfo;
25+
2026
private readonly Lock _lock = new();
2127

22-
public FlowsManagers(IFunctionStore functionStore) => _functionStore = functionStore;
28+
internal FlowsManagers(IFunctionStore functionStore, MessageClearer messageClearer, ClusterInfo clusterInfo)
29+
{
30+
_functionStore = functionStore;
31+
_messageClearer = messageClearer;
32+
_clusterInfo = clusterInfo;
33+
}
2334

2435
public FlowsManager GetOrCreate(StoredType storedType)
2536
{
@@ -28,7 +39,7 @@ public FlowsManager GetOrCreate(StoredType storedType)
2839
if (_managers.TryGetValue(storedType, out var existing))
2940
return existing;
3041

31-
return _managers[storedType] = new FlowsManager(_functionStore);
42+
return _managers[storedType] = new FlowsManager(_functionStore, _messageClearer, _clusterInfo);
3243
}
3344
}
3445

@@ -38,14 +49,24 @@ public FlowsManager GetOrCreate(StoredType storedType)
3849
return _managers.GetValueOrDefault(storedType);
3950
}
4051

41-
public Task Push(IReadOnlyList<StoredMessages> messagesByFlow)
52+
public Task Push(IReadOnlyList<StoredMessages> messages)
4253
{
43-
List<Task> tasks = new();
44-
foreach (var group in messagesByFlow.GroupBy(m => m.StoredId.Type))
45-
if (TryGet(group.Key) is { } manager)
46-
tasks.Add(manager.Push(group.ToList()));
54+
List<Task> messageDeliveries;
55+
lock (_lock)
56+
{
57+
var notRunning = messages
58+
.Where(msg => !_managers.ContainsKey(msg.StoredId.Type));
4759

48-
return Task.WhenAll(tasks);
60+
var running = messages
61+
.Where(msg => _managers.ContainsKey(msg.StoredId.Type))
62+
.GroupBy(msg => msg.StoredId.Type)
63+
.Select(g => _managers[g.Key].Push(g.ToList()))
64+
.ToList();
65+
66+
messageDeliveries = running;
67+
}
68+
69+
return Task.WhenAll(messageDeliveries);
4970
}
5071

5172
public IReadOnlyList<StoredId> FilterOwned(IEnumerable<StoredId> ids)

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

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,15 @@
1+
using System;
2+
using System.Threading.Tasks;
3+
using Cleipnir.ResilientFunctions.Storage;
4+
5+
namespace Cleipnir.ResilientFunctions.CoreRuntime.Invocation;
6+
7+
/// <summary>
8+
/// Resumes execution of an already-claimed flow from the snapshot returned by the store's restart call. Implemented
9+
/// by <see cref="Invoker{TParam,TReturn}"/> (which owns the user function and drives the execution loop) and held by
10+
/// the non-generic <see cref="FlowsManager"/> so it can restart flows without knowing their parameter/return types.
11+
/// </summary>
12+
internal interface IFlowRestarter
13+
{
14+
Task ScheduleRestart(StoredId storedId, RestartedFunction restartedFunction, Action onCompletion);
15+
}

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

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,7 @@
1111

1212
namespace Cleipnir.ResilientFunctions.CoreRuntime.Invocation;
1313

14-
public class Invoker<TParam, TReturn>
14+
public class Invoker<TParam, TReturn> : IFlowRestarter
1515
{
1616
private readonly FlowType _flowType;
1717
private readonly StoredType _storedType;
@@ -236,7 +236,10 @@ internal async Task ScheduleRestart(StoredId storedId, RestartedFunction rf, Act
236236
finally{ _flowsManager.RemoveFlow(storedId, flowState); onCompletion(); }
237237
});
238238
}
239-
239+
240+
Task IFlowRestarter.ScheduleRestart(StoredId storedId, RestartedFunction restartedFunction, Action onCompletion)
241+
=> ScheduleRestart(storedId, restartedFunction, onCompletion);
242+
240243
private async Task<PreparedInvocation> PrepareForInvocation(FlowId flowId, StoredId storedId, TParam param, StoredId? parent, InitialState? initialState, Task completed)
241244
{
242245
var disposables = new List<IDisposable>(capacity: 3);

‎Core/Cleipnir.ResilientFunctions/CoreRuntime/Watchdogs/IMessageClearer.cs‎

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -4,11 +4,14 @@
44
namespace Cleipnir.ResilientFunctions.CoreRuntime.Watchdogs;
55

66
/// <summary>
7-
/// The slice of <see cref="MessageClearer"/> a QueueManager depends on: deleting handled messages from the store
8-
/// and dropping their positions from the watchdog's ignore-set. Exists so tests that hand-roll a QueueManager can
9-
/// pass a no-op stub instead of a fully wired clearer.
7+
/// The slice of <see cref="MessageClearer"/> its collaborators depend on: the QueueManager deletes handled messages
8+
/// from the store (and drops their positions from the watchdog's ignore-set) via <see cref="Clear"/>, while the
9+
/// FlowsManager drops the positions of flows it could not restart back out of the ignore-set - without deleting them
10+
/// from the store - via <see cref="ReopenPositions"/>. Exists so tests can pass a no-op stub instead of a fully
11+
/// wired clearer.
1012
/// </summary>
1113
internal interface IMessageClearer
1214
{
1315
Task Clear(IReadOnlyList<long> positions);
16+
void ReopenPositions(IEnumerable<long> positions);
1417
}

‎Core/Cleipnir.ResilientFunctions/CoreRuntime/Watchdogs/MessageClearer.cs‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,13 @@ public void MarkPushed(IEnumerable<long> positions)
4040
_pushedPositions.Add(position);
4141
}
4242

43+
public void ReopenPositions(IEnumerable<long> positions)
44+
{
45+
lock (_pushedPositionsLock)
46+
foreach (var position in positions)
47+
_pushedPositions.Remove(position);
48+
}
49+
4350
/// <summary>Snapshot of the not-yet-cleared positions, passed to the store as the fetch ignore-set.</summary>
4451
public IReadOnlyList<long> NonClearedPositions()
4552
{

‎Core/Cleipnir.ResilientFunctions/FunctionsRegistry.cs‎

Lines changed: 17 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -43,10 +43,15 @@ public FunctionsRegistry(IFunctionStore functionStore, Settings? settings = null
4343
_shutdownCoordinator = new ShutdownCoordinator();
4444
_settings = SettingsWithDefaults.Default.Merge(settings);
4545
var utcNow = _settings.UtcNow;
46-
_flowsManagers = new FlowsManagers(_functionStore);
47-
46+
_messageClearer = new MessageClearer(
47+
_functionStore.MessageStore,
48+
_settings.UnhandledExceptionHandler,
49+
_settings.WatchdogCheckFrequency
50+
);
4851
ClusterInfo = new ClusterInfo(ReplicaId.NewId());
49-
52+
53+
_flowsManagers = new FlowsManagers(_functionStore, _messageClearer, ClusterInfo);
54+
5055
_postponedWatchdog = new PostponedWatchdog(
5156
_functionStore,
5257
_shutdownCoordinator,
@@ -75,12 +80,6 @@ public FunctionsRegistry(IFunctionStore functionStore, Settings? settings = null
7580
utcNow
7681
);
7782

78-
_messageClearer = new MessageClearer(
79-
_functionStore.MessageStore,
80-
_settings.UnhandledExceptionHandler,
81-
_settings.WatchdogCheckFrequency
82-
);
83-
8483
_messageWatchdog = new MessageWatchdog(
8584
_functionStore.MessageStore,
8685
_flowsManagers,
@@ -257,15 +256,17 @@ public FuncRegistration<TParam, TReturn> RegisterFunc<TParam, TReturn>(
257256
settings?.ClearChildrenAfterCapture ?? true,
258257
_messageClearer
259258
);
259+
var flowsManager = _flowsManagers.GetOrCreate(storedType);
260260
var invoker = new Invoker<TParam, TReturn>(
261261
flowType,
262262
storedType,
263263
inner,
264264
invocationHelper,
265265
settingsWithDefaults.UnhandledExceptionHandler,
266266
ClusterInfo.ReplicaId,
267-
_flowsManagers.GetOrCreate(storedType)
267+
flowsManager
268268
);
269+
flowsManager.SetRestarter(invoker);
269270

270271
WatchDogsFactory.CreateAndStart(
271272
flowType,
@@ -341,15 +342,17 @@ private ParamlessRegistration RegisterParamless(
341342
settings?.ClearChildrenAfterCapture ?? true,
342343
_messageClearer
343344
);
345+
var flowsManager = _flowsManagers.GetOrCreate(storedType);
344346
var invoker = new Invoker<Unit, Unit>(
345347
flowType,
346348
storedType,
347349
inner,
348350
invocationHelper,
349351
settingsWithDefaults.UnhandledExceptionHandler,
350352
ClusterInfo.ReplicaId,
351-
_flowsManagers.GetOrCreate(storedType)
353+
flowsManager
352354
);
355+
flowsManager.SetRestarter(invoker);
353356

354357
WatchDogsFactory.CreateAndStart(
355358
flowType,
@@ -425,15 +428,17 @@ public ActionRegistration<TParam> RegisterAction<TParam>(
425428
settings?.ClearChildrenAfterCapture ?? true,
426429
_messageClearer
427430
);
431+
var flowsManager = _flowsManagers.GetOrCreate(storedType);
428432
var rActionInvoker = new Invoker<TParam, Unit>(
429433
flowType,
430434
storedType,
431435
inner,
432436
invocationHelper,
433437
settingsWithDefaults.UnhandledExceptionHandler,
434438
ClusterInfo.ReplicaId,
435-
_flowsManagers.GetOrCreate(storedType)
439+
flowsManager
436440
);
441+
flowsManager.SetRestarter(rActionInvoker);
437442

438443
WatchDogsFactory.CreateAndStart(
439444
flowType,

0 commit comments

Comments
 (0)