Skip to content

Deserialize effects once at the invocation boundary #257

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

Open
stidsborg wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Deserialize effects once at the invocation boundary #257

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
4531555
Remove the unused StoredInputOutput type
stidsborg Aug 9, 2026
1d5b4a5
Deserialize effects once at the invocation boundary
stidsborg Aug 9, 2026
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: 1 addition & 1 deletion Core/Cleipnir.ResilientFunctions.Tests/InMemoryTests/EffectFlushConcurrencyTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@ private static Effect CreateEffect(StoredId storedId, IFunctionStore functionSto
var effectResults = new EffectResults(
TestFlowId.Create(),
storedId,
existingEffects: new List<StoredEffect>(),
existingEffects: [],
functionStore,
DefaultSerializer.Instance,
new TypeMapper(functionStore.TypeStore),
Expand Down
14 changes: 7 additions & 7 deletions Core/Cleipnir.ResilientFunctions.Tests/InMemoryTests/EffectResultTypeTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,12 +17,12 @@ namespace Cleipnir.ResilientFunctions.Tests.InMemoryTests;
[TestClass]
public class EffectResultTypeTests
{
private static Effect CreateEffect(StoredId storedId, IFunctionStore functionStore, IReadOnlyList<StoredEffect>? existingEffects = null)
private static Effect CreateEffect(StoredId storedId, IFunctionStore functionStore, IReadOnlyList<DeserializedEffect>? existingEffects = null)
{
var effectResults = new EffectResults(
TestFlowId.Create(),
storedId,
existingEffects ?? new List<StoredEffect>(),
existingEffects ?? [],
functionStore,
DefaultSerializer.Instance,
CreateTypeMapper(functionStore),
Expand Down Expand Up @@ -120,7 +120,7 @@ public async Task ResultCapturedThroughBaseTypeIsPersistedAndReadBackAsItsActual
// Replaying the same capture against the persisted effect returns the instance that was captured -
// not an Animal-shaped shell of it.
EffectContext.Reset();
var restarted = CreateEffect(storedId, store, existingEffects: [storedEffect]);
var restarted = CreateEffect(storedId, store, existingEffects: [await storedEffect.Deserialize(DefaultSerializer.Instance, CreateTypeMapper(store))]);
var replayed = await restarted.Capture<Animal>(
() => Task.FromException<Animal>(new InvalidOperationException("Work should not be invoked on replay"))
);
Expand All @@ -145,7 +145,7 @@ public async Task LazilyTypedSequenceIsMaterializedBeforeItIsPersisted()
(await ResolveResultType(store, storedEffect)).ShouldBe(typeof(List<string>));

EffectContext.Reset();
var restarted = CreateEffect(storedId, store, existingEffects: [storedEffect]);
var restarted = CreateEffect(storedId, store, existingEffects: [await storedEffect.Deserialize(DefaultSerializer.Instance, CreateTypeMapper(store))]);
var replayed = await restarted.Capture<IEnumerable<string>>(
() => Task.FromException<IEnumerable<string>>(new InvalidOperationException("Work should not be invoked on replay"))
);
Expand All @@ -171,7 +171,7 @@ public async Task LazilyTypedSequenceCapturedAsObjectIsMaterializedBeforeItIsPer
// Without the materialized type the declared type is all there is to go on, and object yields a
// JsonElement rather than the captured sequence.
EffectContext.Reset();
var restarted = CreateEffect(storedId, store, existingEffects: [storedEffect]);
var restarted = CreateEffect(storedId, store, existingEffects: [await storedEffect.Deserialize(DefaultSerializer.Instance, CreateTypeMapper(store))]);
var replayed = await restarted.Capture<object>(
() => Task.FromException<object>(new InvalidOperationException("Work should not be invoked on replay"))
);
Expand Down Expand Up @@ -209,7 +209,7 @@ public async Task DictionaryIsPersistedAsIs()
(await ResolveResultType(store, storedEffect)).ShouldBe(typeof(Dictionary<string, int>));

EffectContext.Reset();
var restarted = CreateEffect(storedId, store, existingEffects: [storedEffect]);
var restarted = CreateEffect(storedId, store, existingEffects: [await storedEffect.Deserialize(DefaultSerializer.Instance, CreateTypeMapper(store))]);
var replayed = await restarted.Capture<IDictionary<string, int>>(
() => Task.FromException<IDictionary<string, int>>(new InvalidOperationException("Work should not be invoked on replay"))
);
Expand All @@ -233,7 +233,7 @@ public async Task NonVisibleReadOnlyDictionaryIsMaterializedIntoADictionaryBefor
(await ResolveResultType(store, storedEffect)).ShouldBe(typeof(Dictionary<string, int>));

EffectContext.Reset();
var restarted = CreateEffect(storedId, store, existingEffects: [storedEffect]);
var restarted = CreateEffect(storedId, store, existingEffects: [await storedEffect.Deserialize(DefaultSerializer.Instance, CreateTypeMapper(store))]);
var replayed = await restarted.Capture<IReadOnlyDictionary<string, int>>(
() => Task.FromException<IReadOnlyDictionary<string, int>>(new InvalidOperationException("Work should not be invoked on replay"))
);
Expand Down
59 changes: 21 additions & 38 deletions Core/Cleipnir.ResilientFunctions.Tests/TestTemplates/FunctionTests/EffectTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -374,7 +374,7 @@ public async Task EffectsCrudTest(Task<IFunctionStore> storeTask)
var effectResults = new EffectResults(
TestFlowId.Create(),
storedId,
await store.GetEffectResults(storedId),
await store.GetDeserializedEffects(storedId),
store,
DefaultSerializer.Instance,
new TypeMapper(store.TypeStore),
Expand Down Expand Up @@ -427,7 +427,7 @@ public async Task ExistingEffectsFuncIsOnlyInvokedAfterGettingValue(Task<IFuncti
var effectResults = new EffectResults(
TestFlowId.Create(),
storedId,
new List<StoredEffect> { existingEffect },
[await existingEffect.Deserialize(DefaultSerializer.Instance, typeMapper)],
store,
DefaultSerializer.Instance,
typeMapper,
Expand Down Expand Up @@ -730,7 +730,7 @@ public async Task DelayedFlushIsReflectedInUnderlyingStoreForSet(Task<IFunctionS
var effectResults = new EffectResults(
TestFlowId.Create(),
storedId,
new List<StoredEffect>(),
[],
effectStore,
DefaultSerializer.Instance,
new TypeMapper(store.TypeStore),
Expand All @@ -740,43 +740,26 @@ public async Task DelayedFlushIsReflectedInUnderlyingStoreForSet(Task<IFunctionS
);

var effectId1 = new EffectId([1]);
var storedEffect1 = new StoredEffect(
effectId1,
WorkStatus.Completed,
Result: "hello world".ToUtf8Bytes(),
ResultType: null,
StoredException: null,
Alias: null
);
await effectResults.Set(storedEffect1, flush: false);
await effectResults.Set(effectId1, alias: "first", flush: false);
await effectStore
.GetEffectResults(storedId)
.SelectAsync(r => r.Count == 0)
.ShouldBeTrueAsync();

var effectId2 = new EffectId([2]);
var storedEffect2 = new StoredEffect(
effectId2,
WorkStatus.Completed,
Result: "hello universe".ToUtf8Bytes(),
ResultType: null,
StoredException: null,
Alias: null
);
await effectResults.Set(storedEffect2, flush: true);

await effectResults.Set(effectId2, alias: "second", flush: true);

var fetchedResults = await effectStore.GetEffectResults(storedId);
fetchedResults.Count.ShouldBe(2);
fetchedResults
.Single(r => r.EffectId == effectId1)
.Result!
.ToStringFromUtf8Bytes()
.ShouldBe("hello world");
.Alias
.ShouldBe("first");
fetchedResults
.Single(r => r.EffectId == effectId2)
.Result!
.ToStringFromUtf8Bytes()
.ShouldBe("hello universe");
.Alias
.ShouldBe("second");
fetchedResults.All(r => r.WorkStatus == WorkStatus.Completed).ShouldBeTrue();
}

public abstract Task CaptureUsingAtLeastOnceWithoutFlushResiliencyDelaysFlush();
Expand All @@ -799,7 +782,7 @@ public async Task CaptureUsingAtLeastOnceWithoutFlushResiliencyDelaysFlush(Task<
var effectResults = new EffectResults(
TestFlowId.Create(),
storedId,
new List<StoredEffect>(),
[],
effectStore,
DefaultSerializer.Instance,
new TypeMapper(store.TypeStore),
Expand Down Expand Up @@ -890,7 +873,7 @@ public async Task UpsertingExistingEffectDoesNotAffectOtherExistingEffects(Task<
var effectResults = new EffectResults(
TestFlowId.Create(),
storedId,
await effectStore.GetEffectResults(storedId),
await effectStore.GetDeserializedEffects(storedId),
effectStore,
DefaultSerializer.Instance,
new TypeMapper(store.TypeStore),
Expand Down Expand Up @@ -1539,7 +1522,7 @@ public async Task GetChildrenReturnsAllChildEffectValues(Task<IFunctionStore> st
var effectResults = new EffectResults(
TestFlowId.Create(),
storedId,
await store.GetEffectResults(storedId),
await store.GetDeserializedEffects(storedId),
store,
DefaultSerializer.Instance,
new TypeMapper(store.TypeStore),
Expand All @@ -1561,7 +1544,7 @@ await store.GetEffectResults(storedId),
var children = new List<string>();
foreach (var childId in childIds)
{
var (success, child) = await effectResults.TryGet<string>(childId);
var (success, child) = effectResults.TryGet<string>(childId);
if (success)
children.Add(child!);
}
Expand Down Expand Up @@ -1590,7 +1573,7 @@ public async Task GetChildrenReturnsEmptyListWhenNoChildren(Task<IFunctionStore>
var effectResults = new EffectResults(
TestFlowId.Create(),
storedId,
await store.GetEffectResults(storedId),
await store.GetDeserializedEffects(storedId),
store,
DefaultSerializer.Instance,
new TypeMapper(store.TypeStore),
Expand Down Expand Up @@ -1624,7 +1607,7 @@ public async Task GetChildrenReturnsAllDescendants(Task<IFunctionStore> storeTas
var effectResults = new EffectResults(
TestFlowId.Create(),
storedId,
await store.GetEffectResults(storedId),
await store.GetDeserializedEffects(storedId),
store,
DefaultSerializer.Instance,
new TypeMapper(store.TypeStore),
Expand All @@ -1648,7 +1631,7 @@ await store.GetEffectResults(storedId),
var children = new List<int>();
foreach (var childId in childIds)
{
var (success, child) = await effectResults.TryGet<int>(childId);
var (success, child) = effectResults.TryGet<int>(childId);
if (success)
children.Add(child);
}
Expand Down Expand Up @@ -1770,7 +1753,7 @@ public async Task FlushlessUpsertIsNotStoredUntilFlushed(Task<IFunctionStore> st
var effectResults = new EffectResults(
TestFlowId.Create(),
storedId,
new List<StoredEffect>(),
[],
effectStore,
DefaultSerializer.Instance,
new TypeMapper(store.TypeStore),
Expand Down Expand Up @@ -1815,7 +1798,7 @@ public async Task FlushlessUpsertsAreNotStoredUntilFlushed(Task<IFunctionStore>
var effectResults = new EffectResults(
TestFlowId.Create(),
storedId,
new List<StoredEffect>(),
[],
effectStore,
DefaultSerializer.Instance,
new TypeMapper(store.TypeStore),
Expand Down
5 changes: 5 additions & 0 deletions Core/Cleipnir.ResilientFunctions.Tests/Utils/FunctionStoreEffectTestExtensions.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using Cleipnir.ResilientFunctions.CoreRuntime.Serialization;
using Cleipnir.ResilientFunctions.Domain;
using Cleipnir.ResilientFunctions.Storage;

Expand All @@ -16,6 +17,10 @@ public static class FunctionStoreEffectTestExtensions
public static async Task<IReadOnlyList<StoredEffect>> GetEffectResults(this IFunctionStore store, StoredId storedId)
=> (await store.GetFunction(storedId))?.Effects ?? [];

public static async Task<IReadOnlyList<DeserializedEffect>> GetDeserializedEffects(this IFunctionStore store, StoredId storedId)
=> await (await store.GetEffectResults(storedId))
.Deserialize(DefaultSerializer.Instance, new TypeMapper(store.TypeStore));

public static async Task<Dictionary<StoredId, List<StoredEffect>>> GetEffectResults(this IFunctionStore store, IEnumerable<StoredId> storedIds)
{
var result = new Dictionary<StoredId, List<StoredEffect>>();
Expand Down
Loading