Skip to content

Commit 961b95a

Browse files
stidsborgstidsborg
authored
Restart flows via empty-message pokes; restarts no longer pull messages (#184)
1 parent c493b44 commit 961b95a

40 files changed

Lines changed: 968 additions & 949 deletions

File tree

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

Lines changed: 20 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -21,4 +21,23 @@ public override Task FunctionIsSuspendedWhenAwaitedMessageDoesNotAlreadyExist()
2121
[TestMethod]
2222
public override Task ScheduleInvocationWithPublishResultToSpecifiedFunctionId()
2323
=> ScheduleInvocationWithPublishResultToSpecifiedFunctionId(FunctionStoreFactory.Create());
24-
}
24+
25+
[TestMethod]
26+
public override Task EmptyMessagesRestartSuspendedFlowsWithoutDeliveryAndAreRemovedAfterwards()
27+
=> EmptyMessagesRestartSuspendedFlowsWithoutDeliveryAndAreRemovedAfterwards(FunctionStoreFactory.Create());
28+
29+
[TestMethod]
30+
public override Task EmptyMessageIsNotDeliveredToRestartedFlowWhileNonEmptyMessageIs()
31+
=> EmptyMessageIsNotDeliveredToRestartedFlowWhileNonEmptyMessageIs(FunctionStoreFactory.Create());
32+
33+
[TestMethod]
34+
public override Task EmptyMessageIsNotDeliveredWhenFlowIsRestartedViaControlPanel()
35+
=> EmptyMessageIsNotDeliveredWhenFlowIsRestartedViaControlPanel(FunctionStoreFactory.Create());
36+
[TestMethod]
37+
public override Task PendingMessageIsDeliveredWhenCompletedFlowIsPostponedAndRestartedByWatchdog()
38+
=> PendingMessageIsDeliveredWhenCompletedFlowIsPostponedAndRestartedByWatchdog(FunctionStoreFactory.Create());
39+
40+
[TestMethod]
41+
public override Task PendingMessageIsDeliveredWhenCompletedFlowIsRestartedOnDifferentReplica()
42+
=> PendingMessageIsDeliveredWhenCompletedFlowIsRestartedOnDifferentReplica(FunctionStoreFactory.Create());
43+
}

‎Core/Cleipnir.ResilientFunctions.Tests/InMemoryTests/StoreCrudTests.cs‎

Lines changed: 2 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -56,25 +56,13 @@ public override Task StateCanBeUpdatedOnExistingFunction()
5656
public override Task ParameterAndStateAreNotUpdatedWhenEpochDoesNotMatch()
5757
=> ParameterAndStateAreNotUpdatedWhenEpochDoesNotMatch(FunctionStoreFactory.Create());
5858

59-
[TestMethod]
60-
public override Task RestartExecutionsRestartsMultipleUnownedFlows()
61-
=> RestartExecutionsRestartsMultipleUnownedFlows(FunctionStoreFactory.Create());
6259

6360
[TestMethod]
64-
public override Task RestartExecutionsReturnsEmptyDictionaryWhenNoFlowsAreEligible()
65-
=> RestartExecutionsReturnsEmptyDictionaryWhenNoFlowsAreEligible(FunctionStoreFactory.Create());
61+
public override Task RestartExecutionsWithoutMessagesReturnsEmptyDictionaryWhenNoFlowsAreEligible()
62+
=> RestartExecutionsWithoutMessagesReturnsEmptyDictionaryWhenNoFlowsAreEligible(FunctionStoreFactory.Create());
6663

67-
[TestMethod]
68-
public override Task RestartExecutionsRestartsOnlyUnownedFlows()
69-
=> RestartExecutionsRestartsOnlyUnownedFlows(FunctionStoreFactory.Create());
7064

71-
[TestMethod]
72-
public override Task RestartExecutionsReturnsEmptyDictionaryForEmptyInput()
73-
=> RestartExecutionsReturnsEmptyDictionaryForEmptyInput(FunctionStoreFactory.Create());
7465

75-
[TestMethod]
76-
public override Task RestartExecutionsIncludesExistingEffectsAndMessages()
77-
=> RestartExecutionsIncludesExistingEffectsAndMessages(FunctionStoreFactory.Create());
7866

7967
[TestMethod]
8068
public override Task RestartExecutionsWithoutMessagesRestartsMultipleUnownedFlows()

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

Lines changed: 4 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -171,12 +171,12 @@ public override Task FunctionCanBeCreatedWithEffectsOnly()
171171
=> FunctionCanBeCreatedWithEffectsOnly(FunctionStoreFactory.Create());
172172

173173
[TestMethod]
174-
public override Task RestartExecutionReturnsEffectsAndMessages()
175-
=> RestartExecutionReturnsEffectsAndMessages(FunctionStoreFactory.Create());
174+
public override Task RestartExecutionReturnsEffects()
175+
=> RestartExecutionReturnsEffects(FunctionStoreFactory.Create());
176176

177177
[TestMethod]
178-
public override Task RestartExecutionWorksWithEmptyEffectsAndMessages()
179-
=> RestartExecutionWorksWithEmptyEffectsAndMessages(FunctionStoreFactory.Create());
178+
public override Task RestartExecutionWorksWithEmptyEffects()
179+
=> RestartExecutionWorksWithEmptyEffects(FunctionStoreFactory.Create());
180180

181181
[TestMethod]
182182
public override Task FunctionOwnedByReplicaIsPostponedAfterRescheduleFunctionsInvocation()
@@ -210,9 +210,6 @@ public override Task GetResultsReturnsEmptyDictionaryForEmptyInput()
210210
public override Task GetResultsReturnsOnlyExistingFunctionResults()
211211
=> GetResultsReturnsOnlyExistingFunctionResults(FunctionStoreFactory.Create());
212212

213-
[TestMethod]
214-
public override Task RestartExecutionsDoesNotReturnFlowClaimedByPreviousCall()
215-
=> RestartExecutionsDoesNotReturnFlowClaimedByPreviousCall(FunctionStoreFactory.Create());
216213

217214
[TestMethod]
218215
public override Task RestartExecutionsWithoutMessagesDoesNotReturnFlowClaimedByPreviousCall()

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

Lines changed: 289 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,12 @@
11
using System;
2+
using System.Linq;
23
using System.Threading.Tasks;
34
using Cleipnir.ResilientFunctions.CoreRuntime.Invocation;
5+
using Cleipnir.ResilientFunctions.CoreRuntime.Serialization;
46
using Cleipnir.ResilientFunctions.Domain;
57
using Cleipnir.ResilientFunctions.Domain.Exceptions;
68
using Cleipnir.ResilientFunctions.Helpers;
9+
using Cleipnir.ResilientFunctions.Messaging;
710
using Cleipnir.ResilientFunctions.Storage;
811
using Cleipnir.ResilientFunctions.Tests.Utils;
912
using Shouldly;
@@ -114,8 +117,293 @@ await BusyWait.Until(async () =>
114117
controlPanel.Result.ShouldNotBeNull();
115118
var functionCompletion = controlPanel.Result;
116119
functionCompletion.ShouldBe("hello world");
117-
120+
118121
unhandledExceptionHandler.ShouldNotHaveExceptions();
119122
}
120123

124+
public abstract Task EmptyMessagesRestartSuspendedFlowsWithoutDeliveryAndAreRemovedAfterwards();
125+
public async Task EmptyMessagesRestartSuspendedFlowsWithoutDeliveryAndAreRemovedAfterwards(Task<IFunctionStore> functionStore)
126+
{
127+
var store = await functionStore;
128+
129+
var flowType = TestFlowId.Create().Type;
130+
var unhandledExceptionHandler = new UnhandledExceptionCatcher();
131+
using var functionsRegistry = new FunctionsRegistry(
132+
store,
133+
new Settings(unhandledExceptionHandler.Catch, messagesPullFrequency: TimeSpan.FromMilliseconds(100))
134+
);
135+
136+
var invocations = new SyncedCounter();
137+
var registration = functionsRegistry.RegisterParamless(
138+
flowType,
139+
inner: async Task (workflow) =>
140+
{
141+
invocations.Increment();
142+
await workflow.Message<string>();
143+
}
144+
);
145+
146+
await registration.Schedule("instance1");
147+
await registration.Schedule("instance2");
148+
149+
var controlPanel1 = await registration.ControlPanel("instance1");
150+
var controlPanel2 = await registration.ControlPanel("instance2");
151+
controlPanel1.ShouldNotBeNull();
152+
controlPanel2.ShouldNotBeNull();
153+
await controlPanel1.BusyWaitUntil(c => c.Status == Status.Suspended);
154+
await controlPanel2.BusyWaitUntil(c => c.Status == Status.Suspended);
155+
invocations.Current.ShouldBe(2);
156+
157+
var storedId1 = registration.MapToStoredId("instance1".ToFlowInstance());
158+
var storedId2 = registration.MapToStoredId("instance2".ToFlowInstance());
159+
var replicaId = functionsRegistry.ClusterInfo.ReplicaId;
160+
await store.MessageStore.AppendMessages([
161+
new StoredIdAndMessage(storedId1, StoredMessage.CreateEmpty(replicaId)),
162+
new StoredIdAndMessage(storedId2, StoredMessage.CreateEmpty(replicaId))
163+
]);
164+
165+
// The empty messages restart both flows without being delivered...
166+
await BusyWait.Until(() => invocations.Current == 4);
167+
168+
// ...so both flows suspend on the awaited message again...
169+
await controlPanel1.BusyWaitUntil(c => c.Status == Status.Suspended);
170+
await controlPanel2.BusyWaitUntil(c => c.Status == Status.Suspended);
171+
172+
// ...and the empty messages are deleted from the store once the restarts have happened.
173+
await BusyWait.Until(async () =>
174+
(await store.MessageStore.GetMessages(storedId1)).Count == 0 &&
175+
(await store.MessageStore.GetMessages(storedId2)).Count == 0
176+
);
177+
178+
invocations.Current.ShouldBe(4);
179+
unhandledExceptionHandler.ShouldNotHaveExceptions();
180+
}
181+
182+
public abstract Task EmptyMessageIsNotDeliveredToRestartedFlowWhileNonEmptyMessageIs();
183+
public async Task EmptyMessageIsNotDeliveredToRestartedFlowWhileNonEmptyMessageIs(Task<IFunctionStore> functionStore)
184+
{
185+
var store = await functionStore;
186+
187+
var flowId = TestFlowId.Create();
188+
var unhandledExceptionHandler = new UnhandledExceptionCatcher();
189+
using var functionsRegistry = new FunctionsRegistry(
190+
store,
191+
new Settings(unhandledExceptionHandler.Catch, messagesPullFrequency: TimeSpan.FromMilliseconds(100))
192+
);
193+
194+
var registration = functionsRegistry.RegisterFunc(
195+
flowId.Type,
196+
inner: async Task<string> (string _, Workflow workflow) => await workflow.Message<string>()
197+
);
198+
199+
await registration.Schedule(flowId.Instance.Value, "");
200+
201+
var controlPanel = await registration.ControlPanel(flowId.Instance);
202+
controlPanel.ShouldNotBeNull();
203+
await controlPanel.BusyWaitUntil(c => c.Status == Status.Suspended);
204+
205+
var storedId = registration.MapToStoredId(flowId.Instance);
206+
var replicaId = functionsRegistry.ClusterInfo.ReplicaId;
207+
var serializer = DefaultSerializer.Instance;
208+
await store.MessageStore.AppendMessages([
209+
new StoredIdAndMessage(storedId, StoredMessage.CreateEmpty(replicaId)),
210+
new StoredIdAndMessage(
211+
storedId,
212+
new StoredMessage(
213+
serializer.Serialize("hello world", typeof(string)),
214+
serializer.SerializeType(typeof(string)),
215+
Position: 0,
216+
Replica: replicaId
217+
)
218+
)
219+
]);
220+
221+
// The batch restarts the flow with only the non-empty message delivered.
222+
await controlPanel.WaitForCompletion(allowPostponeAndSuspended: true);
223+
await controlPanel.Refresh();
224+
controlPanel.Status.ShouldBe(Status.Succeeded);
225+
controlPanel.Result.ShouldBe("hello world");
226+
227+
// Both the delivered and the empty message are eventually deleted from the store.
228+
await BusyWait.Until(async () => (await store.MessageStore.GetMessages(storedId)).Count == 0);
229+
230+
unhandledExceptionHandler.ShouldNotHaveExceptions();
231+
}
232+
233+
public abstract Task EmptyMessageIsNotDeliveredWhenFlowIsRestartedViaControlPanel();
234+
public async Task EmptyMessageIsNotDeliveredWhenFlowIsRestartedViaControlPanel(Task<IFunctionStore> functionStore)
235+
{
236+
var store = await functionStore;
237+
238+
var flowId = TestFlowId.Create();
239+
var unhandledExceptionHandler = new UnhandledExceptionCatcher();
240+
using var functionsRegistry = new FunctionsRegistry(
241+
store,
242+
new Settings(unhandledExceptionHandler.Catch, messagesPullFrequency: TimeSpan.FromMilliseconds(100))
243+
);
244+
245+
var awaitMessage = new SyncedFlag();
246+
var registration = functionsRegistry.RegisterFunc(
247+
flowId.Type,
248+
inner: async Task<string> (string _, Workflow workflow) =>
249+
awaitMessage.IsRaised ? await workflow.Message<string>() : "no message awaited"
250+
);
251+
252+
// Complete the flow without it consuming any messages.
253+
await registration.Run(flowId.Instance.Value, "");
254+
255+
var storedId = registration.MapToStoredId(flowId.Instance);
256+
var replicaId = functionsRegistry.ClusterInfo.ReplicaId;
257+
var serializer = DefaultSerializer.Instance;
258+
await store.MessageStore.AppendMessages([
259+
new StoredIdAndMessage(storedId, StoredMessage.CreateEmpty(replicaId)),
260+
new StoredIdAndMessage(
261+
storedId,
262+
new StoredMessage(
263+
serializer.Serialize("hello world", typeof(string)),
264+
serializer.SerializeType(typeof(string)),
265+
Position: 0,
266+
Replica: replicaId
267+
)
268+
)
269+
]);
270+
271+
// The pending messages must not resurrect the completed flow.
272+
await Task.Delay(500);
273+
var controlPanel = await registration.ControlPanel(flowId.Instance);
274+
controlPanel.ShouldNotBeNull();
275+
controlPanel.Status.ShouldBe(Status.Succeeded);
276+
controlPanel.Result.ShouldBe("no message awaited");
277+
278+
// An explicit restart does not pull messages itself - the MessageWatchdog delivers the pending non-empty
279+
// message to the restarted flow, while the empty message is never delivered.
280+
awaitMessage.Raise();
281+
await controlPanel.ScheduleRestart();
282+
await controlPanel.WaitForCompletion(allowPostponeAndSuspended: true);
283+
await controlPanel.Refresh();
284+
controlPanel.Status.ShouldBe(Status.Succeeded);
285+
controlPanel.Result.ShouldBe("hello world");
286+
287+
// Both rows end up deleted: the non-empty message is inlined into the completed flow's effect state (and
288+
// its row removed), while the empty poke is simply deleted - a completed flow needs no restart.
289+
await BusyWait.Until(async () => (await store.MessageStore.GetMessages(storedId)).Count == 0);
290+
291+
unhandledExceptionHandler.ShouldNotHaveExceptions();
292+
}
293+
294+
public abstract Task PendingMessageIsDeliveredWhenCompletedFlowIsPostponedAndRestartedByWatchdog();
295+
public async Task PendingMessageIsDeliveredWhenCompletedFlowIsPostponedAndRestartedByWatchdog(Task<IFunctionStore> functionStore)
296+
{
297+
var store = await functionStore;
298+
299+
var flowId = TestFlowId.Create();
300+
var unhandledExceptionHandler = new UnhandledExceptionCatcher();
301+
using var functionsRegistry = new FunctionsRegistry(
302+
store,
303+
new Settings(
304+
unhandledExceptionHandler.Catch,
305+
messagesPullFrequency: TimeSpan.FromMilliseconds(100),
306+
watchdogCheckFrequency: TimeSpan.FromMilliseconds(100)
307+
)
308+
);
309+
310+
var awaitMessage = new SyncedFlag();
311+
var registration = functionsRegistry.RegisterFunc(
312+
flowId.Type,
313+
inner: async Task<string> (string _, Workflow workflow) =>
314+
awaitMessage.IsRaised ? await workflow.Message<string>() : "no message awaited"
315+
);
316+
317+
// Complete the flow without it consuming any messages.
318+
await registration.Run(flowId.Instance.Value, "");
319+
320+
var storedId = registration.MapToStoredId(flowId.Instance);
321+
var replicaId = functionsRegistry.ClusterInfo.ReplicaId;
322+
var serializer = DefaultSerializer.Instance;
323+
await store.MessageStore.AppendMessages([
324+
new StoredIdAndMessage(
325+
storedId,
326+
new StoredMessage(
327+
serializer.Serialize("hello world", typeof(string)),
328+
serializer.SerializeType(typeof(string)),
329+
Position: 0,
330+
Replica: replicaId
331+
)
332+
)
333+
]);
334+
335+
// The message is inlined into the completed flow's effect state and its row deleted.
336+
await BusyWait.Until(async () => (await store.MessageStore.GetMessages(storedId)).Count == 0);
337+
338+
// Resurrect the completed flow via Postpone - the PostponedWatchdog's restart path must also hand the
339+
// inlined message over (it travels in the effect snapshot, not through any restart-specific channel).
340+
awaitMessage.Raise();
341+
var controlPanel = await registration.ControlPanel(flowId.Instance);
342+
controlPanel.ShouldNotBeNull();
343+
await controlPanel.Postpone(DateTime.UtcNow);
344+
345+
await controlPanel.BusyWaitUntil(c => c.Status == Status.Succeeded);
346+
controlPanel.Result.ShouldBe("hello world");
347+
348+
unhandledExceptionHandler.ShouldNotHaveExceptions();
349+
}
350+
351+
public abstract Task PendingMessageIsDeliveredWhenCompletedFlowIsRestartedOnDifferentReplica();
352+
public async Task PendingMessageIsDeliveredWhenCompletedFlowIsRestartedOnDifferentReplica(Task<IFunctionStore> functionStore)
353+
{
354+
var store = await functionStore;
355+
356+
var flowId = TestFlowId.Create();
357+
var unhandledExceptionHandler = new UnhandledExceptionCatcher();
358+
var awaitMessage = new SyncedFlag();
359+
360+
Func<FunctionsRegistry> createRegistry = () => new FunctionsRegistry(
361+
store,
362+
new Settings(unhandledExceptionHandler.Catch, messagesPullFrequency: TimeSpan.FromMilliseconds(100))
363+
);
364+
Func<FunctionsRegistry, FuncRegistration<string, string>> register = registry => registry.RegisterFunc(
365+
flowId.Type,
366+
inner: async Task<string> (string _, Workflow workflow) =>
367+
awaitMessage.IsRaised ? await workflow.Message<string>() : "no message awaited"
368+
);
369+
370+
using var publisherRegistry = createRegistry();
371+
var publisherRegistration = register(publisherRegistry);
372+
373+
// Complete the flow on the publisher replica without it consuming any messages.
374+
await publisherRegistration.Run(flowId.Instance.Value, "");
375+
376+
// Append a message stamped with the publisher replica - its watchdog fetches it, finds the flow
377+
// completed and inlines it into the flow's effect state.
378+
var storedId = publisherRegistration.MapToStoredId(flowId.Instance);
379+
var serializer = DefaultSerializer.Instance;
380+
await store.MessageStore.AppendMessages([
381+
new StoredIdAndMessage(
382+
storedId,
383+
new StoredMessage(
384+
serializer.Serialize("hello world", typeof(string)),
385+
serializer.SerializeType(typeof(string)),
386+
Position: 0,
387+
Replica: publisherRegistry.ClusterInfo.ReplicaId
388+
)
389+
)
390+
]);
391+
await BusyWait.Until(async () => (await store.MessageStore.GetMessages(storedId)).Count == 0);
392+
393+
// Restart the flow from a DIFFERENT replica: the inlined message travels in the effect snapshot the
394+
// restart hands over, so delivery does not depend on the publisher replica's in-memory state.
395+
using var restartingRegistry = createRegistry();
396+
var restartingRegistration = register(restartingRegistry);
397+
398+
awaitMessage.Raise();
399+
var controlPanel = await restartingRegistration.ControlPanel(flowId.Instance);
400+
controlPanel.ShouldNotBeNull();
401+
await controlPanel.ScheduleRestart();
402+
await controlPanel.WaitForCompletion(allowPostponeAndSuspended: true);
403+
await controlPanel.Refresh();
404+
controlPanel.Status.ShouldBe(Status.Succeeded);
405+
controlPanel.Result.ShouldBe("hello world");
406+
407+
unhandledExceptionHandler.ShouldNotHaveExceptions();
408+
}
121409
}

0 commit comments

Comments
 (0)