Add FlushlessUpserts method to EffectResults and Effect and use it in QueueManager

Author: stidsborg

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

@@ -153,4 +153,8 @@ public override Task UtcNowEffectSunshineTest()
     [TestMethod]
     public override Task FlushlessUpsertIsNotStoredUntilFlushed()
         => FlushlessUpsertIsNotStoredUntilFlushed(FunctionStoreFactory.Create());
+
+    [TestMethod]
+    public override Task FlushlessUpsertsAreNotStoredUntilFlushed()
+        => FlushlessUpsertsAreNotStoredUntilFlushed(FunctionStoreFactory.Create());
 }

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

@@ -1537,4 +1537,55 @@ public async Task FlushlessUpsertIsNotStoredUntilFlushed(Task<IFunctionStore> st
         storedEffect.EffectId.ShouldBe(effectId);
         ((int)DefaultSerializer.Instance.Deserialize(storedEffect.Result!, typeof(int))).ShouldBe(42);
     }
+
+    public abstract Task FlushlessUpsertsAreNotStoredUntilFlushed();
+    public async Task FlushlessUpsertsAreNotStoredUntilFlushed(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 effectId1 = new EffectId([1]);
+        var effectId2 = new EffectId([2]);
+        effectResults.FlushlessUpserts(
+        [
+            new EffectResult(effectId1, 42, Alias: null),
+            new EffectResult(effectId2, "hello", Alias: null),
+        ]);
+
+        // before flush nothing should be stored
+        var fetchedResults = await effectStore.GetEffectResults(storedId);
+        fetchedResults.Count.ShouldBe(0);
+
+        // after flush the values should be persisted
+        await effectResults.Flush();
+
+        fetchedResults = await effectStore.GetEffectResults(storedId);
+        fetchedResults.Count.ShouldBe(2);
+        ((int)DefaultSerializer.Instance.Deserialize(
+            fetchedResults.Single(r => r.EffectId == effectId1).Result!, typeof(int)
+        )).ShouldBe(42);
+        ((string)DefaultSerializer.Instance.Deserialize(
+            fetchedResults.Single(r => r.EffectId == effectId2).Result!, typeof(string)
+        )).ShouldBe("hello");
+    }
 }

Core/Cleipnir.ResilientFunctions/Domain/Effect.cs

@@ -69,6 +69,8 @@ internal async Task Upsert<T>(string alias, T value, bool flush = true)
 
     internal Task Upserts(IEnumerable<EffectResult> values, bool flush)
         => effectResults.Upserts(values, flush);
+    internal void FlushlessUpserts(IEnumerable<EffectResult> values)
+        => effectResults.FlushlessUpserts(values);
 
     internal bool TryGet<T>(string alias, out T? value)
     {

Core/Cleipnir.ResilientFunctions/Domain/EffectResults.cs

@@ -175,6 +175,20 @@ internal async Task Upserts(IEnumerable<EffectResult> values, bool flush)
         if (flush)
             await Flush();
     }
+
+    internal void FlushlessUpserts(IEnumerable<EffectResult> values)
+    {
+        var storedEffects = values
+            .Select(t =>
+            {
+                var bytes = _serializer.Serialize(t.Value!, t.Value?.GetType() ?? typeof(object));
+                return new { Id = t.Id, Bytes = bytes, Alias = t.Alias };
+            })
+            .Select(a => StoredEffect.CreateCompleted(a.Id, a.Bytes, a.Alias))
+            .ToList();
+
+        AddToPending(storedEffects);
+    }
 
     public bool TryGet<T>(EffectId effectId, out T? value)
     {

Core/Cleipnir.ResilientFunctions/Queuing/QueueManager.cs

@@ -263,7 +263,7 @@ private void PulseAll()
             if (matched != null)
             {
                 var toRemoveId = new EffectId([-1, 0, positionToRemoveIndex]);
-                await effect.Upserts(
+                effect.FlushlessUpserts(
                     new List<EffectResult>(
                     [
                         new EffectResult(_toRemoveNextIndex, positionToRemoveIndex, Alias: null),
@@ -274,8 +274,7 @@ await effect.Upserts(
                         new EffectResult(senderId, matched.Sender, Alias: null),
                     ]).Concat(matched.IdempotencyKeyResult == null
                         ? []
-                        : [matched.IdempotencyKeyResult]),
-                    flush: false
+                        : [matched.IdempotencyKeyResult])
                 );
 
                 timeouts.RemoveTimeout(timeoutId);

Stores/MariaDB/Cleipnir.ResilientFunctions.MariaDB.Tests/RFunctionTests/EffectTests.cs

@@ -152,4 +152,8 @@ public override Task UtcNowEffectSunshineTest()
     [TestMethod]
     public override Task FlushlessUpsertIsNotStoredUntilFlushed()
         => FlushlessUpsertIsNotStoredUntilFlushed(FunctionStoreFactory.Create());
+
+    [TestMethod]
+    public override Task FlushlessUpsertsAreNotStoredUntilFlushed()
+        => FlushlessUpsertsAreNotStoredUntilFlushed(FunctionStoreFactory.Create());
 }

Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL.Tests/RFunctionTests/EffectTests.cs

@@ -153,4 +153,8 @@ public override Task UtcNowEffectSunshineTest()
     [TestMethod]
     public override Task FlushlessUpsertIsNotStoredUntilFlushed()
         => FlushlessUpsertIsNotStoredUntilFlushed(FunctionStoreFactory.Create());
+
+    [TestMethod]
+    public override Task FlushlessUpsertsAreNotStoredUntilFlushed()
+        => FlushlessUpsertsAreNotStoredUntilFlushed(FunctionStoreFactory.Create());
 }

Stores/SqlServer/Cleipnir.ResilientFunctions.SqlServer.Tests/EffectTests.cs

@@ -153,4 +153,8 @@ public override Task UtcNowEffectSunshineTest()
     [TestMethod]
     public override Task FlushlessUpsertIsNotStoredUntilFlushed()
         => FlushlessUpsertIsNotStoredUntilFlushed(FunctionStoreFactory.Create());
+
+    [TestMethod]
+    public override Task FlushlessUpsertsAreNotStoredUntilFlushed()
+        => FlushlessUpsertsAreNotStoredUntilFlushed(FunctionStoreFactory.Create());
 }