Add synchronous UpsertWithoutFlush method to EffectResults and Effect

Author: stidsborg

Core/Cleipnir.ResilientFunctions.Tests/InMemoryTests/RFunctionTests/EffectTests.cs

@@ -149,4 +149,8 @@ public override Task RunParallelleTest()
     [TestMethod]
     public override Task UtcNowEffectSunshineTest()
         => UtcNowEffectSunshineTest(FunctionStoreFactory.Create());
+
+    [TestMethod]
+    public override Task UpsertWithoutFlushIsNotStoredUntilFlushed()
+        => UpsertWithoutFlushIsNotStoredUntilFlushed(FunctionStoreFactory.Create());
 }

Core/Cleipnir.ResilientFunctions.Tests/TestTemplates/FunctionTests/EffectTests.cs

@@ -1494,4 +1494,47 @@ async Task (workflow) =>
         syncedList.Count.ShouldBe(2);
         syncedList[0].ShouldBe(syncedList[1]);
     }
+
+    public abstract Task UpsertWithoutFlushIsNotStoredUntilFlushed();
+    public async Task UpsertWithoutFlushIsNotStoredUntilFlushed(Task<IFunctionStore> storeTask)
+    {
+        var store = await storeTask;
+        var storedId = TestStoredId.Create();
+        var session = await store.CreateFunction(
+            storedId,
+            "SomeInstance",
+            param: null,
+            leaseExpiration: 0,
+            postponeUntil: null,
+            timestamp: 0,
+            parent: null,
+            owner: ReplicaId.NewId()
+        );
+        var effectStore = store.EffectsStore;
+        var effectResults = new EffectResults(
+            TestFlowId.Create(),
+            storedId,
+            new List<StoredEffect>(),
+            effectStore,
+            DefaultSerializer.Instance,
+            session,
+            clearChildren: true
+        );
+
+        var effectId = new EffectId([1]);
+        effectResults.UpsertWithoutFlush(effectId, alias: null, 42);
+
+        // before flush nothing should be stored
+        var fetchedResults = await effectStore.GetEffectResults(storedId);
+        fetchedResults.Count.ShouldBe(0);
+
+        // after flush the value should be persisted
+        await effectResults.Flush();
+
+        fetchedResults = await effectStore.GetEffectResults(storedId);
+        fetchedResults.Count.ShouldBe(1);
+        var storedEffect = fetchedResults.Single();
+        storedEffect.EffectId.ShouldBe(effectId);
+        ((int)DefaultSerializer.Instance.Deserialize(storedEffect.Result!, typeof(int))).ShouldBe(42);
+    }
 }

Core/Cleipnir.ResilientFunctions/Domain/Effect.cs

@@ -65,6 +65,7 @@ internal async Task Upsert<T>(string alias, T value, bool flush = true)
             flush
         );
     internal Task Upsert<T>(EffectId effectId, T value, string? alias, bool flush) => effectResults.Upsert(effectId, alias, value, flush);
+    internal void UpsertWithoutFlush<T>(EffectId effectId, T value, string? alias) => effectResults.UpsertWithoutFlush(effectId, alias, value);
 
     internal Task Upserts(IEnumerable<EffectResult> values, bool flush)
         => effectResults.Upserts(values, flush);

Core/Cleipnir.ResilientFunctions/Domain/EffectResults.cs

@@ -151,6 +151,13 @@ await FlushOrAddToPending(
             clearChildren: false
         );
     }
+
+    internal void UpsertWithoutFlush<T>(EffectId effectId, string? alias, T value)
+    {
+        var serializedValue = _serializer.Serialize(value!, typeof(T));
+        var storedEffect = StoredEffect.CreateCompleted(effectId, serializedValue, alias);
+        AddToPending(storedEffect.EffectId, storedEffect, delete: false, clearChildren: false);
+    }
 
     internal async Task Upserts(IEnumerable<EffectResult> values, bool flush)
     {