Skip to content

Commit e4f46bd

Browse files
stidsborgstidsborg
authored
Remove interrupt from StoredFlow, function stores and public API (#181)
1 parent cb75b78 commit e4f46bd

33 files changed

Lines changed: 115 additions & 1159 deletions

File tree

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

Lines changed: 0 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -53,8 +53,4 @@ public override Task WorkflowDelayInvocationDelaysFunction()
5353
[TestMethod]
5454
public override Task WorkflowDelayWithDateTimeInvocationDelaysFunction()
5555
=> WorkflowDelayWithDateTimeInvocationDelaysFunction(FunctionStoreFactory.Create());
56-
57-
[TestMethod]
58-
public override Task InterruptedFunctionIsRescheduledWhenPostponed()
59-
=> InterruptedFunctionIsRescheduledWhenPostponed(FunctionStoreFactory.Create());
6056
}

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

Lines changed: 0 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -75,27 +75,11 @@ public override Task ParentCanWaitForBulkScheduledChildren()
7575
[TestMethod]
7676
public override Task ChildIsCreatedWithParentsId()
7777
=> ChildIsCreatedWithParentsId(FunctionStoreFactory.Create());
78-
79-
[TestMethod]
80-
public override Task SuspendedFlowIsRestartedAfterInterrupt()
81-
=> SuspendedFlowIsRestartedAfterInterrupt(FunctionStoreFactory.Create());
82-
83-
[TestMethod]
84-
public override Task ExecutingFlowIsReExecutedWhenSuspendedAfterInterrupt()
85-
=> ExecutingFlowIsReExecutedWhenSuspendedAfterInterrupt(FunctionStoreFactory.Create());
86-
87-
[TestMethod]
88-
public override Task InterruptSuspendedFlows()
89-
=> InterruptSuspendedFlows(FunctionStoreFactory.Create());
9078

9179
[TestMethod]
9280
public override Task DelayedFlowIsRestartedOnce()
9381
=> DelayedFlowIsRestartedOnce(FunctionStoreFactory.Create());
9482

95-
[TestMethod]
96-
public override Task InterruptedExecutingFlowIsRestartedOnce()
97-
=> InterruptedExecutingFlowIsRestartedOnce(FunctionStoreFactory.Create());
98-
9983
[TestMethod]
10084
public override Task TwoDelaysFlowCompletesSuccessfully()
10185
=> TwoDelaysFlowCompletesSuccessfully(FunctionStoreFactory.Create());

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

Lines changed: 0 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -94,10 +94,6 @@ public override Task FunctionStatusForNonExistingFunctionIsNull()
9494
public override Task RestartingExecutionShouldFailWhenExpectedEpochDoesNotMatch()
9595
=> RestartingExecutionShouldFailWhenExpectedEpochDoesNotMatch(FunctionStoreFactory.Create());
9696

97-
[TestMethod]
98-
public override Task RestartingFunctionShouldSetInterruptedToFalse()
99-
=> RestartingFunctionShouldSetInterruptedToFalse(FunctionStoreFactory.Create());
100-
10197
[TestMethod]
10298
public override Task MessagesCanBeFetchedAfterFunctionWithInitialMessagesHasBeenCreated()
10399
=> MessagesCanBeFetchedAfterFunctionWithInitialMessagesHasBeenCreated(FunctionStoreFactory.Create());
@@ -122,26 +118,6 @@ public override Task EpochIsNotIncrementedOnFailure()
122118
public override Task EpochIsNotIncrementedOnSuspension()
123119
=> EpochIsNotIncrementedOnSuspension(FunctionStoreFactory.Create());
124120

125-
[TestMethod]
126-
public override Task SuspensionDoesNotSucceedOnExpectedMessagesCountMismatchButPostponesFunction()
127-
=> SuspensionDoesNotSucceedOnExpectedMessagesCountMismatchButPostponesFunction(FunctionStoreFactory.Create());
128-
129-
[TestMethod]
130-
public override Task FunctionIsStillExecutingOnSuspensionAndInterruptCountMismatch()
131-
=> FunctionIsStillExecutingOnSuspensionAndInterruptCountMismatch(FunctionStoreFactory.Create());
132-
133-
[TestMethod]
134-
public override Task InterruptCountCanBeIncrementedForExecutingFunction()
135-
=> InterruptCountCanBeIncrementedForExecutingFunction(FunctionStoreFactory.Create());
136-
137-
[TestMethod]
138-
public override Task NonExecutingFunctionCanBeInterrupted()
139-
=> NonExecutingFunctionCanBeInterrupted(FunctionStoreFactory.Create());
140-
141-
[TestMethod]
142-
public override Task InterruptCountForNonExistingFunctionIsNull()
143-
=> InterruptCountForNonExistingFunctionIsNull(FunctionStoreFactory.Create());
144-
145121
[TestMethod]
146122
public override Task DefaultStateCanSetAndFetchedAfterwards()
147123
=> DefaultStateCanSetAndFetchedAfterwards(FunctionStoreFactory.Create());
@@ -182,18 +158,6 @@ public override Task FlowWithWithoutParentIsReturnsNullParentInSubsequentGetTest
182158
public override Task MultipleFunctionsStatusCanBeFetched()
183159
=> MultipleFunctionsStatusCanBeFetched(FunctionStoreFactory.Create());
184160

185-
[TestMethod]
186-
public override Task InterruptedFunctionIsNotPostponedToZeroWhenInterrupted()
187-
=> InterruptedFunctionIsNotPostponedToZeroWhenInterrupted(FunctionStoreFactory.Create());
188-
189-
[TestMethod]
190-
public override Task InterruptNothingWorks()
191-
=> InterruptNothingWorks(FunctionStoreFactory.Create());
192-
193-
[TestMethod]
194-
public override Task InterruptedFunctionIsPostponedWhenIgnoringInterruptedFunction()
195-
=> InterruptedFunctionIsPostponedWhenIgnoringInterruptedFunction(FunctionStoreFactory.Create());
196-
197161
[TestMethod]
198162
public override Task FunctionCanBeCreatedWithMessagesAndEffects()
199163
=> FunctionCanBeCreatedWithMessagesAndEffects(FunctionStoreFactory.Create());

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

Lines changed: 2 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -636,14 +636,7 @@ protected async Task AppendedMultipleMessagesAtOnceCanBeFetchedAgain(Task<IFunct
636636
id2Msgs[0].IdempotencyKey.ShouldBe("3");
637637
id2Msgs[0].MessageType.ShouldBe(stringType);
638638
id2Msgs[0].MessageContent.ShouldBe(msg1.ToJsonByteArray());
639-
640-
// appending messages no longer interrupts the target flows - the MessageWatchdog push delivers
641-
var sf1 = await functionStore.GetFunction(id1).ShouldNotBeNullAsync();
642-
sf1.Interrupted.ShouldBeFalse();
643-
644-
var sf2 = await functionStore.GetFunction(id2).ShouldNotBeNullAsync();
645-
sf2.Interrupted.ShouldBeFalse();
646-
}
639+
}
647640

648641
public abstract Task AppendedBatchedMessageCanBeFetchedAgain();
649642
protected async Task AppendedBatchedMessageCanBeFetchedAgain(Task<IFunctionStore> functionStoreTask)
@@ -674,10 +667,6 @@ [new StoredMessage(msg.ToJsonByteArray(), stringType, Replica: ReplicaId.Empty,
674667
messages[0].IdempotencyKey.ShouldBe("1");
675668
messages[0].MessageType.ShouldBe(stringType);
676669
messages[0].MessageContent.ShouldBe(msg.ToJsonByteArray());
677-
678-
// appending messages no longer interrupts the target flow - the MessageWatchdog push delivers
679-
var sf = await functionStore.GetFunction(id).ShouldNotBeNullAsync();
680-
sf.Interrupted.ShouldBeFalse();
681670
}
682671

683672
public abstract Task AppendedBatchedMessagesWithPositionCanBeFetchedAgain();
@@ -738,13 +727,6 @@ await messageStore.AppendMessages(
738727
messagesId2[0].MessageType.ShouldBe(stringType);
739728
messagesId2[0].MessageContent.ToStringFromUtf8Bytes().ShouldBe(msg1String);
740729

741-
// appending messages no longer interrupts the target flows - the MessageWatchdog push delivers
742-
var sf1 = await functionStore.GetFunction(id1).ShouldNotBeNullAsync();
743-
sf1.Interrupted.ShouldBeFalse();
744-
745-
var sf2 = await functionStore.GetFunction(id2).ShouldNotBeNullAsync();
746-
sf2.Interrupted.ShouldBeFalse();
747-
748730
await messageStore.AppendMessages(
749731
[
750732
new StoredIdAndMessage(id2, msg2)
@@ -759,10 +741,7 @@ await messageStore.AppendMessages(
759741
messagesId2[1].IdempotencyKey.ShouldBeNull();
760742
messagesId2[1].MessageType.ShouldBe(stringType);
761743
messagesId2[1].MessageContent.ToStringFromUtf8Bytes().ShouldBe(msg2String);
762-
763-
sf2 = await functionStore.GetFunction(id2).ShouldNotBeNullAsync();
764-
sf2.Interrupted.ShouldBeFalse();
765-
}
744+
}
766745

767746
public abstract Task MessagesForMultipleStoreIdsCanBeFetched();
768747
protected async Task MessagesForMultipleStoreIdsCanBeFetched(Task<IFunctionStore> functionStoreTask)

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

Lines changed: 0 additions & 42 deletions
Original file line numberDiff line numberDiff line change
@@ -736,46 +736,4 @@ await Should.ThrowAsync<InvocationPostponedException>(
736736

737737
unhandledExceptionHandler.ShouldNotHaveExceptions();
738738
}
739-
740-
public abstract Task InterruptedFunctionIsRescheduledWhenPostponed();
741-
protected async Task InterruptedFunctionIsRescheduledWhenPostponed(Task<IFunctionStore> storeTask)
742-
{
743-
var store = await storeTask;
744-
var testId = TestFlowId.Create();
745-
var unhandledExceptionHandler = new UnhandledExceptionCatcher();
746-
747-
using var functionsRegistry = new FunctionsRegistry
748-
(
749-
store,
750-
new Settings(unhandledExceptionHandler.Catch)
751-
);
752-
753-
var insideFunctionFlag = new SyncedFlag();
754-
var mayContinueFlag = new SyncedFlag();
755-
var invocations = new SyncedCounter();
756-
757-
var registration = functionsRegistry
758-
.RegisterParamless(
759-
testId.Type,
760-
async workflow =>
761-
{
762-
invocations.Increment();
763-
insideFunctionFlag.Raise();
764-
await mayContinueFlag.WaitForRaised();
765-
await workflow.Delay(TimeSpan.FromDays(1));
766-
});
767-
768-
await registration.Schedule(testId.Instance);
769-
770-
await insideFunctionFlag.WaitForRaised();
771-
await registration.Interrupt([testId.Instance]);
772-
mayContinueFlag.Raise();
773-
774-
var controlPanel = await registration.ControlPanel(testId.Instance);
775-
controlPanel.ShouldNotBeNull();
776-
await controlPanel.BusyWaitUntil(c => c.Status == Status.Postponed);
777-
778-
await BusyWait.Until(() => invocations.Current == 2);
779-
unhandledExceptionHandler.ShouldNotHaveExceptions();
780-
}
781739
}

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

Lines changed: 0 additions & 181 deletions
Original file line numberDiff line numberDiff line change
@@ -584,143 +584,6 @@ await child.Schedule("Child", param).Completion()
584584
childStoredFunction.ParentId.ShouldBe(parentStoredId);
585585
}
586586

587-
public abstract Task SuspendedFlowIsRestartedAfterInterrupt();
588-
protected async Task SuspendedFlowIsRestartedAfterInterrupt(Task<IFunctionStore> storeTask)
589-
{
590-
var store = await storeTask;
591-
var id = TestFlowId.Create();
592-
var (flowType, flowInstance) = id;
593-
594-
var unhandledExceptionHandler = new UnhandledExceptionCatcher();
595-
using var functionsRegistry = new FunctionsRegistry
596-
(
597-
store,
598-
new Settings(unhandledExceptionHandler.Catch)
599-
);
600-
601-
var syncFlag = new SyncedFlag();
602-
var registration = functionsRegistry.RegisterParamless(
603-
flowType,
604-
inner: Task (workflow) =>
605-
{
606-
if (syncFlag.IsRaised)
607-
return Task.CompletedTask;
608-
609-
syncFlag.Raise();
610-
throw new SuspendInvocationException();
611-
}
612-
);
613-
614-
await registration.Schedule(flowInstance);
615-
await syncFlag.WaitForRaised();
616-
617-
var controlPanel = await registration.ControlPanel(flowInstance).ShouldNotBeNullAsync();
618-
619-
await BusyWait.Until(async () =>
620-
{
621-
await controlPanel.Refresh();
622-
return controlPanel.Status == Status.Suspended;
623-
});
624-
625-
await store.Interrupt(registration.MapToStoredId(id.Instance));
626-
627-
await BusyWait.Until(async () =>
628-
{
629-
await controlPanel.Refresh();
630-
return controlPanel.Status == Status.Succeeded;
631-
});
632-
633-
unhandledExceptionHandler.ShouldNotHaveExceptions();
634-
}
635-
636-
public abstract Task ExecutingFlowIsReExecutedWhenSuspendedAfterInterrupt();
637-
protected async Task ExecutingFlowIsReExecutedWhenSuspendedAfterInterrupt(Task<IFunctionStore> storeTask)
638-
{
639-
var store = await storeTask;
640-
var id = TestFlowId.Create();
641-
var (flowType, flowInstance) = id;
642-
643-
var unhandledExceptionHandler = new UnhandledExceptionCatcher();
644-
using var functionsRegistry = new FunctionsRegistry
645-
(
646-
store,
647-
new Settings(unhandledExceptionHandler.Catch)
648-
);
649-
650-
var insideFlowFlag = new SyncedFlag();
651-
var canContinueFlag = new SyncedFlag();
652-
var executingAgainFlag = new SyncedFlag();
653-
654-
var registration = functionsRegistry.RegisterParamless(
655-
flowType,
656-
inner: async Task (workflow) =>
657-
{
658-
if (insideFlowFlag.IsRaised)
659-
{
660-
executingAgainFlag.Raise();
661-
return;
662-
}
663-
664-
insideFlowFlag.Raise();
665-
await canContinueFlag.WaitForRaised();
666-
throw new SuspendInvocationException();
667-
}
668-
);
669-
670-
await registration.Schedule(flowInstance);
671-
await insideFlowFlag.WaitForRaised();
672-
673-
await store.Interrupt(registration.MapToStoredId(id.Instance));
674-
canContinueFlag.Raise();
675-
676-
await executingAgainFlag.WaitForRaised();
677-
678-
unhandledExceptionHandler.ShouldNotHaveExceptions();
679-
}
680-
681-
public abstract Task InterruptSuspendedFlows();
682-
protected async Task InterruptSuspendedFlows(Task<IFunctionStore> storeTask)
683-
{
684-
var store = await storeTask;
685-
var id = TestFlowId.Create();
686-
var (flowType, flowInstance) = id;
687-
688-
var unhandledExceptionHandler = new UnhandledExceptionCatcher();
689-
using var functionsRegistry = new FunctionsRegistry
690-
(
691-
store,
692-
new Settings(unhandledExceptionHandler.Catch)
693-
);
694-
695-
var succeedFlag = new SyncedFlag();
696-
697-
var registration = functionsRegistry.RegisterParamless(
698-
flowType,
699-
inner: Task () =>
700-
{
701-
if (succeedFlag.IsRaised)
702-
return Task.CompletedTask;
703-
704-
throw new SuspendInvocationException();
705-
}
706-
);
707-
708-
var flows = new [] { "Flow#1", "Flow#2", "Flow#3" }.Select(instance => new FlowInstance(instance)).ToList();
709-
await registration.BulkSchedule(flows);
710-
711-
foreach (var flow in flows)
712-
await (await registration.ControlPanel(flow))!.BusyWaitUntil(cp => cp.Status == Status.Suspended);
713-
714-
succeedFlag.Raise();
715-
716-
await registration.Interrupt(flows.Select(f => f.Value.ToStoredId(registration.StoredType)));
717-
718-
foreach (var flow in flows)
719-
await (await registration.ControlPanel(flow))!.BusyWaitUntil(cp => cp.Status == Status.Succeeded);
720-
721-
unhandledExceptionHandler.ShouldNotHaveExceptions();
722-
}
723-
724587
public abstract Task DelayedFlowIsRestartedOnce();
725588
protected async Task DelayedFlowIsRestartedOnce(Task<IFunctionStore> storeTask)
726589
{
@@ -759,50 +622,6 @@ protected async Task DelayedFlowIsRestartedOnce(Task<IFunctionStore> storeTask)
759622
unhandledExceptionHandler.ShouldNotHaveExceptions();
760623
}
761624

762-
public abstract Task InterruptedExecutingFlowIsRestartedOnce();
763-
protected async Task InterruptedExecutingFlowIsRestartedOnce(Task<IFunctionStore> storeTask)
764-
{
765-
var store = await storeTask;
766-
var id = TestFlowId.Create();
767-
var (flowType, flowInstance) = id;
768-
769-
var unhandledExceptionHandler = new UnhandledExceptionCatcher();
770-
using var functionsRegistry = new FunctionsRegistry
771-
(
772-
store,
773-
new Settings(unhandledExceptionHandler.Catch)
774-
);
775-
776-
var syncedCounter = new SyncedCounter();
777-
var insideFlow = new SyncedFlag();
778-
var continueFlow = new SyncedFlag();
779-
var registration = functionsRegistry.RegisterParamless(
780-
flowType,
781-
inner: async Task (workflow) =>
782-
{
783-
syncedCounter.Increment();
784-
insideFlow.Raise();
785-
if (syncedCounter.Current == 1)
786-
{
787-
await continueFlow.WaitForRaised();
788-
throw new SuspendInvocationException();
789-
}
790-
}
791-
);
792-
793-
await registration.Schedule(flowInstance);
794-
await insideFlow.WaitForRaised();
795-
await registration.Interrupt([flowInstance]);
796-
continueFlow.Raise();
797-
798-
var cp = await registration.ControlPanel(flowInstance).ShouldNotBeNullAsync();
799-
await cp.WaitForCompletion(allowPostponeAndSuspended: true);
800-
801-
syncedCounter.Current.ShouldBe(2);
802-
803-
unhandledExceptionHandler.ShouldNotHaveExceptions();
804-
}
805-
806625
public abstract Task TwoDelaysFlowCompletesSuccessfully();
807626
protected async Task TwoDelaysFlowCompletesSuccessfully(Task<IFunctionStore> storeTask)
808627
{

0 commit comments

Comments
 (0)