Skip to content

Commit d61c3c0

Browse files
stidsborgstidsborg
authored
Add RestartExecutionsWithoutMessages to IFunctionStore (#178)
Adds a lighter variant of RestartExecutions that restarts flows and returns their effects without fetching messages. Introduces the StoredFlowWithEffects record and implements the method across the in-memory, PostgreSQL, MariaDB and SqlServer stores (reusing the existing restart/effects helpers), plus the CrashableFunctionStore decorators. Covered by store-level tests in StoreCrudTests.
1 parent 1d509c7 commit d61c3c0

13 files changed

Lines changed: 451 additions & 1 deletion

File tree

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

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -75,4 +75,20 @@ public override Task RestartExecutionsReturnsEmptyDictionaryForEmptyInput()
7575
[TestMethod]
7676
public override Task RestartExecutionsIncludesExistingEffectsAndMessages()
7777
=> RestartExecutionsIncludesExistingEffectsAndMessages(FunctionStoreFactory.Create());
78+
79+
[TestMethod]
80+
public override Task RestartExecutionsWithoutMessagesRestartsMultipleUnownedFlows()
81+
=> RestartExecutionsWithoutMessagesRestartsMultipleUnownedFlows(FunctionStoreFactory.Create());
82+
83+
[TestMethod]
84+
public override Task RestartExecutionsWithoutMessagesRestartsOnlyUnownedFlows()
85+
=> RestartExecutionsWithoutMessagesRestartsOnlyUnownedFlows(FunctionStoreFactory.Create());
86+
87+
[TestMethod]
88+
public override Task RestartExecutionsWithoutMessagesReturnsEmptyDictionaryForEmptyInput()
89+
=> RestartExecutionsWithoutMessagesReturnsEmptyDictionaryForEmptyInput(FunctionStoreFactory.Create());
90+
91+
[TestMethod]
92+
public override Task RestartExecutionsWithoutMessagesIncludesExistingEffects()
93+
=> RestartExecutionsWithoutMessagesIncludesExistingEffects(FunctionStoreFactory.Create());
7894
}

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

Lines changed: 203 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -555,4 +555,207 @@ await store.CreateFunction(
555555
flow2.Messages[0].MessageContent.ShouldBe("message2".ToUtf8Bytes());
556556
flow2.Messages[0].MessageType.ShouldBe("Type2".ToUtf8Bytes());
557557
}
558+
559+
public abstract Task RestartExecutionsWithoutMessagesRestartsMultipleUnownedFlows();
560+
public async Task RestartExecutionsWithoutMessagesRestartsMultipleUnownedFlows(Task<IFunctionStore> storeTask)
561+
{
562+
var store = await storeTask;
563+
var storedId1 = TestStoredId.Create();
564+
var storedId2 = TestStoredId.Create();
565+
var storedId3 = TestStoredId.Create();
566+
var owner = ReplicaId.NewId();
567+
568+
// Create 3 functions without owners (postponed)
569+
await store.CreateFunction(
570+
storedId1,
571+
"instance1",
572+
Param.ToUtf8Bytes(),
573+
postponeUntil: DateTime.UtcNow.Ticks,
574+
timestamp: DateTime.UtcNow.Ticks,
575+
parent: null,
576+
owner: null
577+
);
578+
await store.CreateFunction(
579+
storedId2,
580+
"instance2",
581+
Param.ToUtf8Bytes(),
582+
postponeUntil: DateTime.UtcNow.Ticks,
583+
timestamp: DateTime.UtcNow.Ticks,
584+
parent: null,
585+
owner: null
586+
);
587+
await store.CreateFunction(
588+
storedId3,
589+
"instance3",
590+
Param.ToUtf8Bytes(),
591+
postponeUntil: DateTime.UtcNow.Ticks,
592+
timestamp: DateTime.UtcNow.Ticks,
593+
parent: null,
594+
owner: null
595+
);
596+
597+
// Restart all three
598+
var result = await store.RestartExecutionsWithoutMessages([storedId1, storedId2, storedId3], owner);
599+
600+
// All three should be restarted
601+
result.Count.ShouldBe(3);
602+
result.ContainsKey(storedId1).ShouldBeTrue();
603+
result.ContainsKey(storedId2).ShouldBeTrue();
604+
result.ContainsKey(storedId3).ShouldBeTrue();
605+
606+
result[storedId1].StoredFlow.OwnerId.ShouldBe(owner);
607+
result[storedId1].StoredFlow.Status.ShouldBe(Status.Executing);
608+
result[storedId2].StoredFlow.OwnerId.ShouldBe(owner);
609+
result[storedId2].StoredFlow.Status.ShouldBe(Status.Executing);
610+
result[storedId3].StoredFlow.OwnerId.ShouldBe(owner);
611+
result[storedId3].StoredFlow.Status.ShouldBe(Status.Executing);
612+
}
613+
614+
public abstract Task RestartExecutionsWithoutMessagesRestartsOnlyUnownedFlows();
615+
public async Task RestartExecutionsWithoutMessagesRestartsOnlyUnownedFlows(Task<IFunctionStore> storeTask)
616+
{
617+
var store = await storeTask;
618+
var storedId1 = TestStoredId.Create();
619+
var storedId2 = TestStoredId.Create();
620+
var storedId3 = TestStoredId.Create();
621+
var existingOwner = ReplicaId.NewId();
622+
var newOwner = ReplicaId.NewId();
623+
624+
// Create 2 functions without owners and 1 with owner
625+
await store.CreateFunction(
626+
storedId1,
627+
"instance1",
628+
Param.ToUtf8Bytes(),
629+
postponeUntil: DateTime.UtcNow.Ticks,
630+
timestamp: DateTime.UtcNow.Ticks,
631+
parent: null,
632+
owner: null
633+
);
634+
await store.CreateFunction(
635+
storedId2,
636+
"instance2",
637+
Param.ToUtf8Bytes(),
638+
postponeUntil: null,
639+
timestamp: DateTime.UtcNow.Ticks,
640+
parent: null,
641+
owner: existingOwner
642+
);
643+
await store.CreateFunction(
644+
storedId3,
645+
"instance3",
646+
Param.ToUtf8Bytes(),
647+
postponeUntil: DateTime.UtcNow.Ticks,
648+
timestamp: DateTime.UtcNow.Ticks,
649+
parent: null,
650+
owner: null
651+
);
652+
653+
// Restart all three - only 1 and 3 should succeed
654+
var result = await store.RestartExecutionsWithoutMessages([storedId1, storedId2, storedId3], newOwner);
655+
656+
result.Count.ShouldBe(2);
657+
result.ContainsKey(storedId1).ShouldBeTrue();
658+
result.ContainsKey(storedId2).ShouldBeFalse();
659+
result.ContainsKey(storedId3).ShouldBeTrue();
660+
661+
result[storedId1].StoredFlow.OwnerId.ShouldBe(newOwner);
662+
result[storedId3].StoredFlow.OwnerId.ShouldBe(newOwner);
663+
}
664+
665+
public abstract Task RestartExecutionsWithoutMessagesReturnsEmptyDictionaryForEmptyInput();
666+
public async Task RestartExecutionsWithoutMessagesReturnsEmptyDictionaryForEmptyInput(Task<IFunctionStore> storeTask)
667+
{
668+
var store = await storeTask;
669+
var owner = ReplicaId.NewId();
670+
671+
var result = await store.RestartExecutionsWithoutMessages([], owner);
672+
673+
result.Count.ShouldBe(0);
674+
}
675+
676+
public abstract Task RestartExecutionsWithoutMessagesIncludesExistingEffects();
677+
public async Task RestartExecutionsWithoutMessagesIncludesExistingEffects(Task<IFunctionStore> storeTask)
678+
{
679+
var store = await storeTask;
680+
var storedId1 = TestStoredId.Create();
681+
var storedId2 = TestStoredId.Create();
682+
var owner = ReplicaId.NewId();
683+
684+
// Create effects
685+
var effect1 = new StoredEffect(
686+
EffectId: "effect1".GetHashCode().ToEffectId(),
687+
WorkStatus: WorkStatus.Completed,
688+
Result: "result1".ToUtf8Bytes(),
689+
StoredException: null,
690+
Alias: null
691+
);
692+
var effect2 = new StoredEffect(
693+
EffectId: "effect2".GetHashCode().ToEffectId(),
694+
WorkStatus: WorkStatus.Completed,
695+
Result: "result2".ToUtf8Bytes(),
696+
StoredException: null,
697+
Alias: null
698+
);
699+
700+
// Create messages - these must NOT be fetched by RestartExecutionsWithoutMessages
701+
var message1 = new StoredMessage(
702+
MessageContent: "message1".ToUtf8Bytes(),
703+
MessageType: "Type1".ToUtf8Bytes(),
704+
Position: 0,
705+
Replica: ReplicaId.Empty,
706+
IdempotencyKey: null
707+
);
708+
var message2 = new StoredMessage(
709+
MessageContent: "message2".ToUtf8Bytes(),
710+
MessageType: "Type2".ToUtf8Bytes(),
711+
Position: 1,
712+
Replica: ReplicaId.Empty,
713+
IdempotencyKey: null
714+
);
715+
716+
// Create 2 functions with effects and messages
717+
await store.CreateFunction(
718+
storedId1,
719+
"instance1",
720+
Param.ToUtf8Bytes(),
721+
postponeUntil: DateTime.UtcNow.Ticks,
722+
timestamp: DateTime.UtcNow.Ticks,
723+
parent: null,
724+
owner: null,
725+
effects: [effect1],
726+
messages: [message1]
727+
);
728+
await store.CreateFunction(
729+
storedId2,
730+
"instance2",
731+
Param.ToUtf8Bytes(),
732+
postponeUntil: DateTime.UtcNow.Ticks,
733+
timestamp: DateTime.UtcNow.Ticks,
734+
parent: null,
735+
owner: null,
736+
effects: [effect2],
737+
messages: [message2]
738+
);
739+
740+
// Restart both
741+
var result = await store.RestartExecutionsWithoutMessages([storedId1, storedId2], owner);
742+
743+
// Verify both flows returned with their effects
744+
result.Count.ShouldBe(2);
745+
746+
result.ContainsKey(storedId1).ShouldBeTrue();
747+
result.ContainsKey(storedId2).ShouldBeTrue();
748+
749+
// Verify flow 1 has correct effects
750+
var flow1 = result[storedId1];
751+
flow1.Effects.Count.ShouldBe(1);
752+
flow1.Effects[0].EffectId.ShouldBe("effect1".GetHashCode().ToEffectId());
753+
flow1.Effects[0].Result.ShouldBe("result1".ToUtf8Bytes());
754+
755+
// Verify flow 2 has correct effects
756+
var flow2 = result[storedId2];
757+
flow2.Effects.Count.ShouldBe(1);
758+
flow2.Effects[0].EffectId.ShouldBe("effect2".GetHashCode().ToEffectId());
759+
flow2.Effects[0].Result.ShouldBe("result2".ToUtf8Bytes());
760+
}
558761
}

‎Core/Cleipnir.ResilientFunctions.Tests/TestTemplates/WatchDogsTests/CrashableFunctionStore.cs‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -81,6 +81,11 @@ public Task<Dictionary<StoredId, StoredFlowWithEffectsAndMessages>> RestartExecu
8181
? Task.FromException<Dictionary<StoredId, StoredFlowWithEffectsAndMessages>>(new TimeoutException())
8282
: _inner.RestartExecutions(storedIds, owner);
8383

84+
public Task<Dictionary<StoredId, StoredFlowWithEffects>> RestartExecutionsWithoutMessages(IReadOnlyList<StoredId> storedIds, ReplicaId owner)
85+
=> _crashed
86+
? Task.FromException<Dictionary<StoredId, StoredFlowWithEffects>>(new TimeoutException())
87+
: _inner.RestartExecutionsWithoutMessages(storedIds, owner);
88+
8489
public Task<IReadOnlyList<StoredId>> GetExpiredFunctions(long isEligibleBefore)
8590
=> _crashed
8691
? Task.FromException<IReadOnlyList<StoredId>>(new TimeoutException())

‎Core/Cleipnir.ResilientFunctions/Storage/IFunctionStore.cs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@ Task<int> BulkScheduleFunctions(
3333

3434
Task<StoredFlowWithEffectsAndMessages?> RestartExecution(StoredId storedId, ReplicaId owner);
3535
Task<Dictionary<StoredId, StoredFlowWithEffectsAndMessages>> RestartExecutions(IReadOnlyList<StoredId> storedIds, ReplicaId owner);
36+
Task<Dictionary<StoredId, StoredFlowWithEffects>> RestartExecutionsWithoutMessages(IReadOnlyList<StoredId> storedIds, ReplicaId owner);
3637

3738
Task<IReadOnlyList<StoredId>> GetExpiredFunctions(long expiresBefore);
3839
Task<IReadOnlyList<StoredId>> GetSucceededFunctions(long completedBefore);

‎Core/Cleipnir.ResilientFunctions/Storage/InMemoryFunctionStore.cs‎

Lines changed: 58 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -201,7 +201,64 @@ public virtual async Task<Dictionary<StoredId, StoredFlowWithEffectsAndMessages>
201201

202202
return result;
203203
}
204-
204+
205+
public virtual async Task<Dictionary<StoredId, StoredFlowWithEffects>> RestartExecutionsWithoutMessages(
206+
IReadOnlyList<StoredId> storedIds,
207+
ReplicaId owner)
208+
{
209+
if (storedIds.Count == 0)
210+
return new Dictionary<StoredId, StoredFlowWithEffects>();
211+
212+
var restartedIds = new List<StoredId>();
213+
214+
// Restart eligible flows
215+
lock (_sync)
216+
{
217+
foreach (var storedId in storedIds)
218+
{
219+
if (!_states.ContainsKey(storedId))
220+
continue;
221+
222+
var state = _states[storedId];
223+
if (state.Owner != null)
224+
continue; // Skip already owned flows
225+
226+
// Restart this flow
227+
state.Status = Status.Executing;
228+
state.Expires = 0;
229+
state.Interrupted = false;
230+
state.Owner = owner;
231+
232+
restartedIds.Add(storedId);
233+
}
234+
}
235+
236+
var effectsDict = await EffectsStore.GetEffectResults(storedIds);
237+
238+
// Build result dictionary - only for restarted flows
239+
var result = new Dictionary<StoredId, StoredFlowWithEffects>();
240+
foreach (var storedId in restartedIds)
241+
{
242+
var sf = await GetFunction(storedId);
243+
if (sf == null) continue;
244+
245+
var effects = effectsDict.TryGetValue(storedId, out var effs)
246+
? effs
247+
: new List<StoredEffect>();
248+
249+
var session = new SnapshotStorageSession(owner);
250+
foreach (var effect in effects)
251+
session.Effects[effect.EffectId] = effect;
252+
253+
session.Version = _effectsStore.GetVersion(storedId);
254+
session.RowExists = effects.Any();
255+
256+
result[storedId] = new StoredFlowWithEffects(sf, effects, session);
257+
}
258+
259+
return result;
260+
}
261+
205262
public virtual Task<IReadOnlyList<StoredId>> GetExpiredFunctions(long expiresBefore)
206263
{
207264
lock (_sync)

‎Core/Cleipnir.ResilientFunctions/Storage/Types.cs‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -182,6 +182,12 @@ public static StoredEffect Deserialize(byte[] bytes)
182182

183183
public record IdWithParam(StoredId StoredId, string HumanInstanceId, byte[]? Param);
184184

185+
public record StoredFlowWithEffects(
186+
StoredFlow StoredFlow,
187+
IReadOnlyList<StoredEffect> Effects,
188+
IStorageSession StorageSession
189+
);
190+
185191
public record StoredFlowWithEffectsAndMessages(
186192
StoredFlow StoredFlow,
187193
IReadOnlyList<StoredEffect> Effects,

‎Samples/Sample.ConsoleApp/Utils/CrashableFunctionStore.cs‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -63,6 +63,11 @@ public Task<Dictionary<StoredId, StoredFlowWithEffectsAndMessages>> RestartExecu
6363
? Task.FromException<Dictionary<StoredId, StoredFlowWithEffectsAndMessages>>(new TimeoutException())
6464
: _inner.RestartExecutions(storedIds, owner);
6565

66+
public Task<Dictionary<StoredId, StoredFlowWithEffects>> RestartExecutionsWithoutMessages(IReadOnlyList<StoredId> storedIds, ReplicaId owner)
67+
=> _crashed
68+
? Task.FromException<Dictionary<StoredId, StoredFlowWithEffects>>(new TimeoutException())
69+
: _inner.RestartExecutionsWithoutMessages(storedIds, owner);
70+
6671
public Task<IReadOnlyList<StoredId>> GetExpiredFunctions(long expiresBefore)
6772
=> _crashed
6873
? Task.FromException<IReadOnlyList<StoredId>>(new TimeoutException())

‎Stores/MariaDB/Cleipnir.ResilientFunctions.MariaDB.Tests/StoreCrudTests.cs‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -72,4 +72,20 @@ public override Task RestartExecutionsReturnsEmptyDictionaryForEmptyInput()
7272
[TestMethod]
7373
public override Task RestartExecutionsIncludesExistingEffectsAndMessages()
7474
=> RestartExecutionsIncludesExistingEffectsAndMessages(FunctionStoreFactory.Create());
75+
76+
[TestMethod]
77+
public override Task RestartExecutionsWithoutMessagesRestartsMultipleUnownedFlows()
78+
=> RestartExecutionsWithoutMessagesRestartsMultipleUnownedFlows(FunctionStoreFactory.Create());
79+
80+
[TestMethod]
81+
public override Task RestartExecutionsWithoutMessagesRestartsOnlyUnownedFlows()
82+
=> RestartExecutionsWithoutMessagesRestartsOnlyUnownedFlows(FunctionStoreFactory.Create());
83+
84+
[TestMethod]
85+
public override Task RestartExecutionsWithoutMessagesReturnsEmptyDictionaryForEmptyInput()
86+
=> RestartExecutionsWithoutMessagesReturnsEmptyDictionaryForEmptyInput(FunctionStoreFactory.Create());
87+
88+
[TestMethod]
89+
public override Task RestartExecutionsWithoutMessagesIncludesExistingEffects()
90+
=> RestartExecutionsWithoutMessagesIncludesExistingEffects(FunctionStoreFactory.Create());
7591
}

0 commit comments

Comments
 (0)