Skip to content

Refactored Suspend-functionality #93

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 2 commits into from
Oct 16, 2025
Merged

Refactored Suspend-functionality #93

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
0066b60
fix postpone
tsydelegate Oct 15, 2025
c32628f
Refactored suspend functionality
tsydelegate Oct 15, 2025
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
2 changes: 0 additions & 2 deletions Core/Cleipnir.ResilientFunctions.Tests/InMemoryTests/RFunctionTests/DelayedStartUpTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -111,7 +111,6 @@ await store.PostponeFunction(
registration.MapToStoredId(functionId.Instance),
postponeUntil: 0,
timestamp: DateTime.UtcNow.Ticks,
ignoreInterrupted: true,
expectedReplica: ReplicaId.Empty,
effects: null,
messages: null,
Expand Down Expand Up @@ -152,7 +151,6 @@ await store.PostponeFunction(
registration.MapToStoredId(functionId.Instance),
postponeUntil: 0,
timestamp: DateTime.UtcNow.Ticks,
ignoreInterrupted: true,
expectedReplica: ReplicaId.Empty,
effects: null,
messages: null,
Expand Down
1 change: 0 additions & 1 deletion ...silientFunctions.Tests/InMemoryTests/ShutdownCoordinationTests/RFunctionsShutdownTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -183,7 +183,6 @@ await store.PostponeFunction(
registration.MapToStoredId(functionId.Instance),
postponeUntil: DateTime.UtcNow.AddDays(-1).Ticks,
timestamp: DateTime.UtcNow.Ticks,
ignoreInterrupted: true,
Guid.Empty.ToReplicaId(),
effects: null,
messages: null,
Expand Down
4 changes: 2 additions & 2 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 @@ -175,8 +175,8 @@ public override Task MultipleFunctionsStatusCanBeFetched()
=> MultipleFunctionsStatusCanBeFetched(FunctionStoreFactory.Create());

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

[TestMethod]
public override Task InterruptNothingWorks()
Expand Down
1 change: 0 additions & 1 deletion 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 @@ -559,7 +559,6 @@ await store.PostponeFunction(
storedId,
postponeUntil: DateTime.UtcNow.AddDays(-1).Ticks,
timestamp: DateTime.UtcNow.Ticks,
ignoreInterrupted: false,
expectedReplica: ReplicaId.Empty,
effects: null,
messages: null,
Expand Down
28 changes: 12 additions & 16 deletions Core/Cleipnir.ResilientFunctions.Tests/TestTemplates/StoreTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -235,7 +235,6 @@ await store.PostponeFunction(
functionId,
postponeUntil: nowTicks,
timestamp: DateTime.UtcNow.Ticks,
ignoreInterrupted: false,
expectedReplica: ReplicaId.Empty,
effects: null,
messages: null,
Expand Down Expand Up @@ -273,7 +272,6 @@ await store.PostponeFunction(
functionId,
postponeUntil: nowTicks,
timestamp: DateTime.UtcNow.Ticks,
ignoreInterrupted: false,
expectedReplica: ReplicaId.Empty,
effects: null,
messages: null,
Expand Down Expand Up @@ -311,7 +309,6 @@ await store.PostponeFunction(
functionId,
postponeUntil: nowTicks,
timestamp: DateTime.UtcNow.Ticks,
ignoreInterrupted: false,
expectedReplica: ReplicaId.NewId(),
effects: null,
messages: null,
Expand Down Expand Up @@ -802,7 +799,6 @@ await store.PostponeFunction(
functionId,
postponeUntil: DateTime.UtcNow.Ticks,
timestamp: DateTime.UtcNow.Ticks,
ignoreInterrupted: false,
expectedReplica: ReplicaId.Empty,
effects: null,
messages: null,
Expand Down Expand Up @@ -905,11 +901,12 @@ await store.SuspendFunction(
effects: null,
messages: null,
storageSession: null
).ShouldBeFalseAsync();
).ShouldBeTrueAsync();

var storedFunction = await store.GetFunction(functionId);
storedFunction.ShouldNotBeNull();
(storedFunction.Status is Status.Executing).ShouldBeTrue();
storedFunction.Status.ShouldBe(Status.Postponed);
storedFunction.Expires.ShouldBe(0);
}

public abstract Task FunctionIsStillExecutingOnSuspensionAndInterruptCountMismatch();
Expand Down Expand Up @@ -945,11 +942,12 @@ await store.MessageStore.AppendMessage(
storageSession: null
);

success.ShouldBeFalse();
success.ShouldBeTrue();

var storedFunction = await store.GetFunction(functionId);
storedFunction.ShouldNotBeNull();
storedFunction.Status.ShouldBe(Status.Executing);
storedFunction.Status.ShouldBe(Status.Postponed);
storedFunction.Expires.ShouldBe(0);
}

public abstract Task InterruptCountCanBeIncrementedForExecutingFunction();
Expand Down Expand Up @@ -1284,8 +1282,8 @@ await store.SucceedFunction(
statusAndEpoch2.Status.ShouldBe(Status.Succeeded);
}

public abstract Task InterruptedFunctionIsNotPostponedWhenFlagIsSet();
protected async Task InterruptedFunctionIsNotPostponedWhenFlagIsSet(Task<IFunctionStore> storeTask)
public abstract Task InterruptedFunctionIsNotPostponedToZeroWhenInterrupted();
protected async Task InterruptedFunctionIsNotPostponedToZeroWhenInterrupted(Task<IFunctionStore> storeTask)
{
var storedId = TestStoredId.Create();
var store = await storeTask;
Expand All @@ -1306,17 +1304,17 @@ await store.CreateFunction(
var success = await store.PostponeFunction(
storedId,
postponeUntil: 0,
timestamp: 0,
ignoreInterrupted: false,
timestamp: 100,
expectedReplica: ReplicaId.Empty,
effects: null,
messages: null,
storageSession: null
);
success.ShouldBeFalse();
success.ShouldBeTrue();

var sf = await store.GetFunction(storedId).ShouldNotBeNullAsync();
sf.Status.ShouldBe(Status.Executing);
sf.Status.ShouldBe(Status.Postponed);
sf.Expires.ShouldBe(0);
}

public abstract Task InterruptNothingWorks();
Expand Down Expand Up @@ -1349,7 +1347,6 @@ await store.CreateFunction(
storedId,
postponeUntil: 0,
timestamp: 0,
ignoreInterrupted: true,
expectedReplica: ReplicaId.Empty,
effects: null,
messages: null,
Expand Down Expand Up @@ -1803,7 +1800,6 @@ await store.CreateFunction(
await store.PostponeFunction(
functionId,
postponeUntil: DateTime.UtcNow.Ticks,
ignoreInterrupted: false,
timestamp: DateTime.UtcNow.Ticks,
expectedReplica: ReplicaId.Empty,
effects: null,
Expand Down
3 changes: 1 addition & 2 deletions .../Cleipnir.ResilientFunctions.Tests/TestTemplates/WatchDogsTests/CrashableFunctionStore.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -137,7 +137,6 @@ public async Task<bool> PostponeFunction(
StoredId storedId,
long postponeUntil,
long timestamp,
bool ignoreInterrupted,
ReplicaId expectedReplica,
IReadOnlyList<StoredEffect>? effects,
IReadOnlyList<StoredMessage>? messages,
Expand All @@ -147,7 +146,7 @@ public async Task<bool> PostponeFunction(
if (_crashed)
throw new TimeoutException();

var result = await _inner.PostponeFunction(storedId, postponeUntil, timestamp, ignoreInterrupted, expectedReplica, effects, messages, storageSession);
var result = await _inner.PostponeFunction(storedId, postponeUntil, timestamp, expectedReplica, effects, messages, storageSession);
AfterPostponeFunctionFlag.Raise();

return result;
Expand Down
2 changes: 0 additions & 2 deletions Core/Cleipnir.ResilientFunctions/CoreRuntime/Invocation/InvocationHelper.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -131,7 +131,6 @@ public async Task<PersistResultOutcome> PersistResult(StoredId storedId, Result<
storedId,
postponeUntil: result.Postpone!.DateTime.Ticks,
timestamp: UtcNow().Ticks,
ignoreInterrupted: true,
_replicaId,
effects: null,
messages: null,
Expand Down Expand Up @@ -520,7 +519,6 @@ public async Task<bool> Reschedule(StoredId id, TParam param)
id,
postponeUntil: 0,
timestamp: UtcNow().Ticks,
ignoreInterrupted: true,
_replicaId,
effects: null,
messages: null,
Expand Down
1 change: 0 additions & 1 deletion Core/Cleipnir.ResilientFunctions/Storage/IFunctionStore.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -85,7 +85,6 @@ Task<bool> PostponeFunction(
StoredId storedId,
long postponeUntil,
long timestamp,
bool ignoreInterrupted,
ReplicaId expectedReplica,
IReadOnlyList<StoredEffect>? effects,
IReadOnlyList<StoredMessage>? messages,
Expand Down
11 changes: 3 additions & 8 deletions Core/Cleipnir.ResilientFunctions/Storage/InMemoryFunctionStore.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -280,7 +280,6 @@ public Task<bool> PostponeFunction(
StoredId storedId,
long postponeUntil,
long timestamp,
bool ignoreInterrupted,
ReplicaId? expectedReplica,
IReadOnlyList<StoredEffect>? effects,
IReadOnlyList<StoredMessage>? messages,
Expand All @@ -293,9 +292,6 @@ public Task<bool> PostponeFunction(
var state = _states[storedId];
if (state.Owner != expectedReplica) return false.ToTask();

if (!ignoreInterrupted && state.Interrupted)
return false.ToTask();

state.Status = Status.Postponed;
state.Expires = state.Interrupted ? 0 : postponeUntil;
state.Interrupted = false;
Expand Down Expand Up @@ -348,12 +344,11 @@ public Task<bool> SuspendFunction(
if (state.Owner != expectedReplica)
return false.ToTask();

if (state.Interrupted)
return false.ToTask();

state.Status = Status.Suspended;
state.Status = state.Interrupted ? Status.Postponed : Status.Suspended;
state.Expires = 0;
state.Timestamp = timestamp;
state.Owner = null;
state.Interrupted = false;

return true.ToTask();
}
Expand Down
3 changes: 1 addition & 2 deletions Samples/Sample.ConsoleApp/Utils/CrashableFunctionStore.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -124,14 +124,13 @@ public Task<bool> PostponeFunction(
StoredId storedId,
long postponeUntil,
long timestamp,
bool ignoreInterrupted,
ReplicaId expectedReplica,
IReadOnlyList<StoredEffect>? effects,
IReadOnlyList<StoredMessage>? messages,
IStorageSession? storageSession
) => _crashed
? Task.FromException<bool>(new TimeoutException())
: _inner.PostponeFunction(storedId, postponeUntil, timestamp, ignoreInterrupted, expectedReplica, effects, messages, storageSession);
: _inner.PostponeFunction(storedId, postponeUntil, timestamp, expectedReplica, effects, messages, storageSession);

public Task<bool> FailFunction(
StoredId storedId,
Expand Down
4 changes: 2 additions & 2 deletions Stores/MariaDB/Cleipnir.ResilientFunctions.MariaDB.Tests/StoreTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -164,8 +164,8 @@ public override Task MultipleFunctionsStatusCanBeFetched()
=> MultipleFunctionsStatusCanBeFetched(FunctionStoreFactory.Create());

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

[TestMethod]
public override Task InterruptNothingWorks()
Expand Down
3 changes: 1 addition & 2 deletions Stores/MariaDB/Cleipnir.ResilientFunctions.MariaDB/MariaDbFunctionStore.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -357,15 +357,14 @@ public async Task<bool> PostponeFunction(
StoredId storedId,
long postponeUntil,
long timestamp,
bool ignoreInterrupted,
ReplicaId expectedReplica,
IReadOnlyList<StoredEffect>? effects,
IReadOnlyList<StoredMessage>? messages,
IStorageSession? storageSession)
{
await using var conn = await CreateOpenConnection(_connectionString);
await using var command = _sqlGenerator
.PostponeFunction(storedId, postponeUntil, timestamp, ignoreInterrupted, expectedReplica)
.PostponeFunction(storedId, postponeUntil, timestamp, expectedReplica)
.ToSqlCommand(conn);

var affectedRows = await command.ExecuteNonQueryAsync();
Expand Down
21 changes: 8 additions & 13 deletions Stores/MariaDB/Cleipnir.ResilientFunctions.MariaDB/SqlGenerator.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -239,7 +239,6 @@ public StoreCommand PostponeFunction(
StoredId storedId,
long postponeUntil,
long timestamp,
bool ignoreInterrupted,
ReplicaId expectedReplica)
{
_postponedFunctionSql ??= $@"
Expand All @@ -251,15 +250,10 @@ public StoreCommand PostponeFunction(
interrupted = 0
WHERE
id = ? AND
owner = ? AND
interrupted=0";

var sql = _postponedFunctionSql;
if (ignoreInterrupted)
sql = sql.Replace("interrupted=0", "1 = 1");
owner = ?";

return StoreCommand.Create(
sql,
_postponedFunctionSql,
values: [
postponeUntil,
timestamp,
Expand Down Expand Up @@ -299,11 +293,12 @@ public StoreCommand SuspendFunction(StoredId storedId, long timestamp, ReplicaId
{
_suspendFunctionSql ??= $@"
UPDATE {tablePrefix}
SET status = {(int) Status.Suspended}, timestamp = ?, owner = NULL
WHERE
id = ? AND
owner = ? AND
NOT interrupted";
SET status = CASE WHEN interrupted = 1 THEN {(int) Status.Postponed} ELSE {(int) Status.Suspended} END,
expires = 0,
timestamp = ?,
owner = NULL,
interrupted = 0
WHERE id = ? AND owner = ?";

return StoreCommand.Create(
_suspendFunctionSql,
Expand Down
4 changes: 2 additions & 2 deletions Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL.Tests/StoreTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -167,8 +167,8 @@ public override Task MultipleFunctionsStatusCanBeFetched()
=> MultipleFunctionsStatusCanBeFetched(FunctionStoreFactory.Create());

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

[TestMethod]
public override Task InterruptNothingWorks()
Expand Down
2 changes: 0 additions & 2 deletions Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL/PostgreSqlFunctionStore.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -390,7 +390,6 @@ public async Task<bool> PostponeFunction(
StoredId storedId,
long postponeUntil,
long timestamp,
bool ignoreInterrupted,
ReplicaId expectedReplica,
IReadOnlyList<StoredEffect>? effects,
IReadOnlyList<StoredMessage>? messages,
Expand All @@ -401,7 +400,6 @@ public async Task<bool> PostponeFunction(
storedId,
postponeUntil,
timestamp,
ignoreInterrupted,
expectedReplica
).ToNpgsqlCommand(conn);

Expand Down
20 changes: 8 additions & 12 deletions Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL/SqlGenerator.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -231,7 +231,6 @@ public StoreCommand PostponeFunction(
StoredId storedId,
long postponeUntil,
long timestamp,
bool ignoreInterrupted,
ReplicaId expectedReplica)
{
_postponeFunctionSql ??= $@"
Expand All @@ -243,15 +242,10 @@ public StoreCommand PostponeFunction(
interrupted = FALSE
WHERE
id = $3 AND
owner = $4 AND
NOT interrupted";

var sql = _postponeFunctionSql;
if (ignoreInterrupted)
sql = sql.Replace("NOT interrupted", "1 = 1");
owner = $4";

return StoreCommand.Create(
sql,
_postponeFunctionSql,
values: [
postponeUntil,
timestamp,
Expand Down Expand Up @@ -289,10 +283,12 @@ public StoreCommand SuspendFunction(StoredId storedId, long timestamp, ReplicaId
{
_suspendFunctionSql ??= $@"
UPDATE {tablePrefix}
SET status = {(int)Status.Suspended}, timestamp = $1, owner = NULL
WHERE id = $2 AND
owner = $3 AND
NOT interrupted;";
SET status = CASE WHEN interrupted THEN {(int) Status.Postponed} ELSE {(int) Status.Suspended} END,
expires = 0,
timestamp = $1,
owner = NULL,
interrupted = FALSE
WHERE id = $2 AND owner = $3";

return StoreCommand.Create(
_suspendFunctionSql,
Expand Down
4 changes: 2 additions & 2 deletions Stores/SqlServer/Cleipnir.ResilientFunctions.SqlServer.Tests/StoreTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -167,8 +167,8 @@ public override Task MultipleFunctionsStatusCanBeFetched()
=> MultipleFunctionsStatusCanBeFetched(FunctionStoreFactory.Create());

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

[TestMethod]
public override Task InterruptNothingWorks()
Expand Down
Loading