Skip to content

Remove interrupt from StoredFlow, function stores and public API #181

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
stidsborg merged 1 commit into from
Jul 4, 2026
Merged

Remove interrupt from StoredFlow, function stores and public API #181

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 0 additions & 4 deletions Core/Cleipnir.ResilientFunctions.Tests/InMemoryTests/RFunctionTests/PostponedTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,8 +53,4 @@ public override Task WorkflowDelayInvocationDelaysFunction()
[TestMethod]
public override Task WorkflowDelayWithDateTimeInvocationDelaysFunction()
=> WorkflowDelayWithDateTimeInvocationDelaysFunction(FunctionStoreFactory.Create());

[TestMethod]
public override Task InterruptedFunctionIsRescheduledWhenPostponed()
=> InterruptedFunctionIsRescheduledWhenPostponed(FunctionStoreFactory.Create());
}
16 changes: 0 additions & 16 deletions Core/Cleipnir.ResilientFunctions.Tests/InMemoryTests/RFunctionTests/SuspensionTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -75,27 +75,11 @@ public override Task ParentCanWaitForBulkScheduledChildren()
[TestMethod]
public override Task ChildIsCreatedWithParentsId()
=> ChildIsCreatedWithParentsId(FunctionStoreFactory.Create());

[TestMethod]
public override Task SuspendedFlowIsRestartedAfterInterrupt()
=> SuspendedFlowIsRestartedAfterInterrupt(FunctionStoreFactory.Create());

[TestMethod]
public override Task ExecutingFlowIsReExecutedWhenSuspendedAfterInterrupt()
=> ExecutingFlowIsReExecutedWhenSuspendedAfterInterrupt(FunctionStoreFactory.Create());

[TestMethod]
public override Task InterruptSuspendedFlows()
=> InterruptSuspendedFlows(FunctionStoreFactory.Create());

[TestMethod]
public override Task DelayedFlowIsRestartedOnce()
=> DelayedFlowIsRestartedOnce(FunctionStoreFactory.Create());

[TestMethod]
public override Task InterruptedExecutingFlowIsRestartedOnce()
=> InterruptedExecutingFlowIsRestartedOnce(FunctionStoreFactory.Create());

[TestMethod]
public override Task TwoDelaysFlowCompletesSuccessfully()
=> TwoDelaysFlowCompletesSuccessfully(FunctionStoreFactory.Create());
Expand Down
36 changes: 0 additions & 36 deletions Core/Cleipnir.ResilientFunctions.Tests/InMemoryTests/StoreTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -94,10 +94,6 @@ public override Task FunctionStatusForNonExistingFunctionIsNull()
public override Task RestartingExecutionShouldFailWhenExpectedEpochDoesNotMatch()
=> RestartingExecutionShouldFailWhenExpectedEpochDoesNotMatch(FunctionStoreFactory.Create());

[TestMethod]
public override Task RestartingFunctionShouldSetInterruptedToFalse()
=> RestartingFunctionShouldSetInterruptedToFalse(FunctionStoreFactory.Create());

[TestMethod]
public override Task MessagesCanBeFetchedAfterFunctionWithInitialMessagesHasBeenCreated()
=> MessagesCanBeFetchedAfterFunctionWithInitialMessagesHasBeenCreated(FunctionStoreFactory.Create());
Expand All @@ -122,26 +118,6 @@ public override Task EpochIsNotIncrementedOnFailure()
public override Task EpochIsNotIncrementedOnSuspension()
=> EpochIsNotIncrementedOnSuspension(FunctionStoreFactory.Create());

[TestMethod]
public override Task SuspensionDoesNotSucceedOnExpectedMessagesCountMismatchButPostponesFunction()
=> SuspensionDoesNotSucceedOnExpectedMessagesCountMismatchButPostponesFunction(FunctionStoreFactory.Create());

[TestMethod]
public override Task FunctionIsStillExecutingOnSuspensionAndInterruptCountMismatch()
=> FunctionIsStillExecutingOnSuspensionAndInterruptCountMismatch(FunctionStoreFactory.Create());

[TestMethod]
public override Task InterruptCountCanBeIncrementedForExecutingFunction()
=> InterruptCountCanBeIncrementedForExecutingFunction(FunctionStoreFactory.Create());

[TestMethod]
public override Task NonExecutingFunctionCanBeInterrupted()
=> NonExecutingFunctionCanBeInterrupted(FunctionStoreFactory.Create());

[TestMethod]
public override Task InterruptCountForNonExistingFunctionIsNull()
=> InterruptCountForNonExistingFunctionIsNull(FunctionStoreFactory.Create());

[TestMethod]
public override Task DefaultStateCanSetAndFetchedAfterwards()
=> DefaultStateCanSetAndFetchedAfterwards(FunctionStoreFactory.Create());
Expand Down Expand Up @@ -182,18 +158,6 @@ public override Task FlowWithWithoutParentIsReturnsNullParentInSubsequentGetTest
public override Task MultipleFunctionsStatusCanBeFetched()
=> MultipleFunctionsStatusCanBeFetched(FunctionStoreFactory.Create());

[TestMethod]
public override Task InterruptedFunctionIsNotPostponedToZeroWhenInterrupted()
=> InterruptedFunctionIsNotPostponedToZeroWhenInterrupted(FunctionStoreFactory.Create());

[TestMethod]
public override Task InterruptNothingWorks()
=> InterruptNothingWorks(FunctionStoreFactory.Create());

[TestMethod]
public override Task InterruptedFunctionIsPostponedWhenIgnoringInterruptedFunction()
=> InterruptedFunctionIsPostponedWhenIgnoringInterruptedFunction(FunctionStoreFactory.Create());

[TestMethod]
public override Task FunctionCanBeCreatedWithMessagesAndEffects()
=> FunctionCanBeCreatedWithMessagesAndEffects(FunctionStoreFactory.Create());
Expand Down
25 changes: 2 additions & 23 deletions Core/Cleipnir.ResilientFunctions.Tests/Messaging/TestTemplates/MessageStoreTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -636,14 +636,7 @@ protected async Task AppendedMultipleMessagesAtOnceCanBeFetchedAgain(Task<IFunct
id2Msgs[0].IdempotencyKey.ShouldBe("3");
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.ShouldBeFalse();

var sf2 = await functionStore.GetFunction(id2).ShouldNotBeNullAsync();
sf2.Interrupted.ShouldBeFalse();
}
}

public abstract Task AppendedBatchedMessageCanBeFetchedAgain();
protected async Task AppendedBatchedMessageCanBeFetchedAgain(Task<IFunctionStore> functionStoreTask)
Expand Down Expand Up @@ -674,10 +667,6 @@ [new StoredMessage(msg.ToJsonByteArray(), stringType, Replica: ReplicaId.Empty,
messages[0].IdempotencyKey.ShouldBe("1");
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.ShouldBeFalse();
}

public abstract Task AppendedBatchedMessagesWithPositionCanBeFetchedAgain();
Expand Down Expand Up @@ -738,13 +727,6 @@ 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.ShouldBeFalse();

var sf2 = await functionStore.GetFunction(id2).ShouldNotBeNullAsync();
sf2.Interrupted.ShouldBeFalse();

await messageStore.AppendMessages(
[
new StoredIdAndMessage(id2, msg2)
Expand All @@ -759,10 +741,7 @@ await messageStore.AppendMessages(
messagesId2[1].IdempotencyKey.ShouldBeNull();
messagesId2[1].MessageType.ShouldBe(stringType);
messagesId2[1].MessageContent.ToStringFromUtf8Bytes().ShouldBe(msg2String);

sf2 = await functionStore.GetFunction(id2).ShouldNotBeNullAsync();
sf2.Interrupted.ShouldBeFalse();
}
}

public abstract Task MessagesForMultipleStoreIdsCanBeFetched();
protected async Task MessagesForMultipleStoreIdsCanBeFetched(Task<IFunctionStore> functionStoreTask)
Expand Down
42 changes: 0 additions & 42 deletions Core/Cleipnir.ResilientFunctions.Tests/TestTemplates/FunctionTests/PostponedTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -736,46 +736,4 @@ await Should.ThrowAsync<InvocationPostponedException>(

unhandledExceptionHandler.ShouldNotHaveExceptions();
}

public abstract Task InterruptedFunctionIsRescheduledWhenPostponed();
protected async Task InterruptedFunctionIsRescheduledWhenPostponed(Task<IFunctionStore> storeTask)
{
var store = await storeTask;
var testId = TestFlowId.Create();
var unhandledExceptionHandler = new UnhandledExceptionCatcher();

using var functionsRegistry = new FunctionsRegistry
(
store,
new Settings(unhandledExceptionHandler.Catch)
);

var insideFunctionFlag = new SyncedFlag();
var mayContinueFlag = new SyncedFlag();
var invocations = new SyncedCounter();

var registration = functionsRegistry
.RegisterParamless(
testId.Type,
async workflow =>
{
invocations.Increment();
insideFunctionFlag.Raise();
await mayContinueFlag.WaitForRaised();
await workflow.Delay(TimeSpan.FromDays(1));
});

await registration.Schedule(testId.Instance);

await insideFunctionFlag.WaitForRaised();
await registration.Interrupt([testId.Instance]);
mayContinueFlag.Raise();

var controlPanel = await registration.ControlPanel(testId.Instance);
controlPanel.ShouldNotBeNull();
await controlPanel.BusyWaitUntil(c => c.Status == Status.Postponed);

await BusyWait.Until(() => invocations.Current == 2);
unhandledExceptionHandler.ShouldNotHaveExceptions();
}
}
181 changes: 0 additions & 181 deletions Core/Cleipnir.ResilientFunctions.Tests/TestTemplates/FunctionTests/SuspensionTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -584,143 +584,6 @@ await child.Schedule("Child", param).Completion()
childStoredFunction.ParentId.ShouldBe(parentStoredId);
}

public abstract Task SuspendedFlowIsRestartedAfterInterrupt();
protected async Task SuspendedFlowIsRestartedAfterInterrupt(Task<IFunctionStore> storeTask)
{
var store = await storeTask;
var id = TestFlowId.Create();
var (flowType, flowInstance) = id;

var unhandledExceptionHandler = new UnhandledExceptionCatcher();
using var functionsRegistry = new FunctionsRegistry
(
store,
new Settings(unhandledExceptionHandler.Catch)
);

var syncFlag = new SyncedFlag();
var registration = functionsRegistry.RegisterParamless(
flowType,
inner: Task (workflow) =>
{
if (syncFlag.IsRaised)
return Task.CompletedTask;

syncFlag.Raise();
throw new SuspendInvocationException();
}
);

await registration.Schedule(flowInstance);
await syncFlag.WaitForRaised();

var controlPanel = await registration.ControlPanel(flowInstance).ShouldNotBeNullAsync();

await BusyWait.Until(async () =>
{
await controlPanel.Refresh();
return controlPanel.Status == Status.Suspended;
});

await store.Interrupt(registration.MapToStoredId(id.Instance));

await BusyWait.Until(async () =>
{
await controlPanel.Refresh();
return controlPanel.Status == Status.Succeeded;
});

unhandledExceptionHandler.ShouldNotHaveExceptions();
}

public abstract Task ExecutingFlowIsReExecutedWhenSuspendedAfterInterrupt();
protected async Task ExecutingFlowIsReExecutedWhenSuspendedAfterInterrupt(Task<IFunctionStore> storeTask)
{
var store = await storeTask;
var id = TestFlowId.Create();
var (flowType, flowInstance) = id;

var unhandledExceptionHandler = new UnhandledExceptionCatcher();
using var functionsRegistry = new FunctionsRegistry
(
store,
new Settings(unhandledExceptionHandler.Catch)
);

var insideFlowFlag = new SyncedFlag();
var canContinueFlag = new SyncedFlag();
var executingAgainFlag = new SyncedFlag();

var registration = functionsRegistry.RegisterParamless(
flowType,
inner: async Task (workflow) =>
{
if (insideFlowFlag.IsRaised)
{
executingAgainFlag.Raise();
return;
}

insideFlowFlag.Raise();
await canContinueFlag.WaitForRaised();
throw new SuspendInvocationException();
}
);

await registration.Schedule(flowInstance);
await insideFlowFlag.WaitForRaised();

await store.Interrupt(registration.MapToStoredId(id.Instance));
canContinueFlag.Raise();

await executingAgainFlag.WaitForRaised();

unhandledExceptionHandler.ShouldNotHaveExceptions();
}

public abstract Task InterruptSuspendedFlows();
protected async Task InterruptSuspendedFlows(Task<IFunctionStore> storeTask)
{
var store = await storeTask;
var id = TestFlowId.Create();
var (flowType, flowInstance) = id;

var unhandledExceptionHandler = new UnhandledExceptionCatcher();
using var functionsRegistry = new FunctionsRegistry
(
store,
new Settings(unhandledExceptionHandler.Catch)
);

var succeedFlag = new SyncedFlag();

var registration = functionsRegistry.RegisterParamless(
flowType,
inner: Task () =>
{
if (succeedFlag.IsRaised)
return Task.CompletedTask;

throw new SuspendInvocationException();
}
);

var flows = new [] { "Flow#1", "Flow#2", "Flow#3" }.Select(instance => new FlowInstance(instance)).ToList();
await registration.BulkSchedule(flows);

foreach (var flow in flows)
await (await registration.ControlPanel(flow))!.BusyWaitUntil(cp => cp.Status == Status.Suspended);

succeedFlag.Raise();

await registration.Interrupt(flows.Select(f => f.Value.ToStoredId(registration.StoredType)));

foreach (var flow in flows)
await (await registration.ControlPanel(flow))!.BusyWaitUntil(cp => cp.Status == Status.Succeeded);

unhandledExceptionHandler.ShouldNotHaveExceptions();
}

public abstract Task DelayedFlowIsRestartedOnce();
protected async Task DelayedFlowIsRestartedOnce(Task<IFunctionStore> storeTask)
{
Expand Down Expand Up @@ -759,50 +622,6 @@ protected async Task DelayedFlowIsRestartedOnce(Task<IFunctionStore> storeTask)
unhandledExceptionHandler.ShouldNotHaveExceptions();
}

public abstract Task InterruptedExecutingFlowIsRestartedOnce();
protected async Task InterruptedExecutingFlowIsRestartedOnce(Task<IFunctionStore> storeTask)
{
var store = await storeTask;
var id = TestFlowId.Create();
var (flowType, flowInstance) = id;

var unhandledExceptionHandler = new UnhandledExceptionCatcher();
using var functionsRegistry = new FunctionsRegistry
(
store,
new Settings(unhandledExceptionHandler.Catch)
);

var syncedCounter = new SyncedCounter();
var insideFlow = new SyncedFlag();
var continueFlow = new SyncedFlag();
var registration = functionsRegistry.RegisterParamless(
flowType,
inner: async Task (workflow) =>
{
syncedCounter.Increment();
insideFlow.Raise();
if (syncedCounter.Current == 1)
{
await continueFlow.WaitForRaised();
throw new SuspendInvocationException();
}
}
);

await registration.Schedule(flowInstance);
await insideFlow.WaitForRaised();
await registration.Interrupt([flowInstance]);
continueFlow.Raise();

var cp = await registration.ControlPanel(flowInstance).ShouldNotBeNullAsync();
await cp.WaitForCompletion(allowPostponeAndSuspended: true);

syncedCounter.Current.ShouldBe(2);

unhandledExceptionHandler.ShouldNotHaveExceptions();
}

public abstract Task TwoDelaysFlowCompletesSuccessfully();
protected async Task TwoDelaysFlowCompletesSuccessfully(Task<IFunctionStore> storeTask)
{
Expand Down
Loading