Added disable of flush lock to effects and added more retry-policy tests

Author: stidsborg

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

@@ -89,4 +89,12 @@ public override Task CaptureEffectWithRetryPolicy()
     [TestMethod]
     public override Task CaptureEffectWithRetryPolicyWithResult()
         => CaptureEffectWithRetryPolicyWithResult(FunctionStoreFactory.Create());
+
+    [TestMethod]
+    public override Task CaptureEffectWithRetryPolicyWithoutSuspension()
+        => CaptureEffectWithRetryPolicyWithoutSuspension(FunctionStoreFactory.Create());
+
+    [TestMethod]
+    public override Task ExceptionPredicateIsUsedForRetryPolicy()
+        => ExceptionPredicateIsUsedForRetryPolicy(FunctionStoreFactory.Create());
 }

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

@@ -828,4 +828,86 @@ public async Task CaptureEffectWithRetryPolicyWithResult(Task<IFunctionStore> st
 
         syncedCounter.Current.ShouldBe(3);
     }
+    
+    public abstract Task CaptureEffectWithRetryPolicyWithoutSuspension();
+    public async Task CaptureEffectWithRetryPolicyWithoutSuspension(Task<IFunctionStore> storeTask)
+    {
+        var store = await storeTask;
+        var flowId = TestFlowId.Create();
+        using var registry = new FunctionsRegistry(store);
+        var syncedCounter = new SyncedCounter();
+
+        var retryPolicy = RetryPolicy.Create(suspendThreshold: TimeSpan.MaxValue, initialInterval: TimeSpan.FromMilliseconds(100), backoffCoefficient: 1);
+        var registration = registry.RegisterFunc<string, string>(
+            flowType: flowId.Type,
+            async (param, workflow) =>
+            {
+                var effect = workflow.Effect;
+                return await effect.Capture(() =>
+                {
+                    if (syncedCounter.Current < 2)
+                    {
+                        syncedCounter.Increment();
+                        throw new TimeoutException();
+                    } 
+                    
+                    syncedCounter.Increment();
+                    return Task.FromResult(param);
+                }, retryPolicy);
+            }
+        );
+
+        var result = await registration.Invoke(flowId.Instance, "Hello World!");
+        result.ShouldBe("Hello World!");
+        
+        syncedCounter.Current.ShouldBe(3);
+    }
+    
+    public abstract Task ExceptionPredicateIsUsedForRetryPolicy();
+    public async Task ExceptionPredicateIsUsedForRetryPolicy(Task<IFunctionStore> storeTask)
+    {
+        var store = await storeTask;
+        var flowId = TestFlowId.Create();
+        using var registry = new FunctionsRegistry(store);
+        var syncedCounter = new SyncedCounter();
+
+        var retryPolicy = RetryPolicy.Create(
+            suspendThreshold: TimeSpan.MaxValue,
+            initialInterval: TimeSpan.FromMilliseconds(100),
+            backoffCoefficient: 1,
+            shouldRetry: e => e is TimeoutException
+        );
+        var registration = registry.RegisterParamless(
+            flowType: flowId.Type,
+            async workflow =>
+            {
+                var effect = workflow.Effect;
+                await effect.Capture(() =>
+                {
+                    if (syncedCounter.Current == 0)
+                    {
+                        syncedCounter.Increment();
+                        throw new TimeoutException();
+                    }
+                    else
+                    {
+                        syncedCounter.Increment();
+                        throw new InvalidOperationException();
+                    }
+                }, retryPolicy);
+            }
+        );
+
+        try
+        {
+            await registration.Invoke(flowId.Instance);
+            Assert.Fail("Expected InvalidOperationException");
+        }
+        catch (FatalWorkflowException e)
+        {
+            e.ErrorType.ShouldBe(typeof(InvalidOperationException));
+        }
+        
+        syncedCounter.Current.ShouldBe(2);
+    }
 }

Core/Cleipnir.ResilientFunctions/Domain/Effect.cs

@@ -43,6 +43,13 @@ public async Task<bool> Mark(string id)
     public async Task Upsert<T>(string id, T value, bool flush = true) => await Upsert(CreateEffectId(id, EffectType.Effect), value, flush);
     internal Task Upsert<T>(EffectId effectId, T value, bool flush) => effectResults.Upsert(effectId, value, flush);
 
+    /// <summary>
+    /// Take flush lock - thereby posting any new flushes until the lock is released.
+    /// Useful for adding multiple effects without risking only some of the effects are persisted.
+    /// </summary>
+    /// <returns>A disposable instance needed to release the lock</returns>
+    internal Task<IDisposable> DisableFlush() => effectResults.DisableFlush();
+
     public async Task<Option<T>> TryGet<T>(string id) => await TryGet<T>(CreateEffectId(id, EffectType.Effect));
     internal Task<Option<T>> TryGet<T>(EffectId effectId) => effectResults.TryGet<T>(effectId);
     public async Task<T> Get<T>(string id) => await Get<T>(CreateEffectId(id, EffectType.Effect));

Core/Cleipnir.ResilientFunctions/Domain/EffectResults.cs

@@ -5,6 +5,7 @@
 using System.Threading.Tasks;
 using Cleipnir.ResilientFunctions.CoreRuntime.Serialization;
 using Cleipnir.ResilientFunctions.Domain.Exceptions.Commands;
+using Cleipnir.ResilientFunctions.Helpers.Disposables;
 using Cleipnir.ResilientFunctions.Reactive.Utilities;
 using Cleipnir.ResilientFunctions.Storage;
 
@@ -123,6 +124,12 @@ await FlushOrAddToPending(
             delete: false
         );
     }
+
+    public async Task<IDisposable> DisableFlush()
+    {
+        await _flushSync.WaitAsync();
+        return new ActionDisposable(() => _flushSync.Release());
+    }
 
     public async Task<Option<T>> TryGet<T>(EffectId effectId)
     {

Core/Cleipnir.ResilientFunctions/Domain/RetryPolicy.cs

@@ -131,8 +131,11 @@ public async Task<T> Invoke<T>(Func<Task<T>> work, Effect effect, UtcNow utcNow)
 
                 delayUntil = utcNow().Add(delay);
                 iteration += 1;
-                await effect.Upsert(delayUntilId, delayUntil.Ticks, flush: false);
-                await effect.Upsert(iterationId, iteration, flush: false);
+                {
+                    using var @lock = await effect.DisableFlush();
+                    await effect.Upsert(delayUntilId, delayUntil.Ticks, flush: false);
+                    await effect.Upsert(iterationId, iteration, flush: false);    
+                }
 
                 if (iteration >= maximumAttempts)
                     throw;

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

@@ -89,4 +89,11 @@ public override Task CaptureEffectWithRetryPolicy()
     public override Task CaptureEffectWithRetryPolicyWithResult()
         => CaptureEffectWithRetryPolicyWithResult(FunctionStoreFactory.Create());
 
+    [TestMethod]
+    public override Task CaptureEffectWithRetryPolicyWithoutSuspension()
+        => CaptureEffectWithRetryPolicyWithoutSuspension(FunctionStoreFactory.Create());
+    
+    [TestMethod]
+    public override Task ExceptionPredicateIsUsedForRetryPolicy()
+        => ExceptionPredicateIsUsedForRetryPolicy(FunctionStoreFactory.Create());
 }

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

@@ -90,4 +90,11 @@ public override Task CaptureEffectWithRetryPolicy()
     public override Task CaptureEffectWithRetryPolicyWithResult()
         => CaptureEffectWithRetryPolicyWithResult(FunctionStoreFactory.Create());
 
+    [TestMethod]
+    public override Task CaptureEffectWithRetryPolicyWithoutSuspension()
+        => CaptureEffectWithRetryPolicyWithoutSuspension(FunctionStoreFactory.Create());
+    
+    [TestMethod]
+    public override Task ExceptionPredicateIsUsedForRetryPolicy()
+        => ExceptionPredicateIsUsedForRetryPolicy(FunctionStoreFactory.Create());
 }

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

@@ -90,4 +90,11 @@ public override Task CaptureEffectWithRetryPolicy()
     public override Task CaptureEffectWithRetryPolicyWithResult()
         => CaptureEffectWithRetryPolicyWithResult(FunctionStoreFactory.Create());
 
+    [TestMethod]
+    public override Task CaptureEffectWithRetryPolicyWithoutSuspension()
+        => CaptureEffectWithRetryPolicyWithoutSuspension(FunctionStoreFactory.Create());
+    
+    [TestMethod]
+    public override Task ExceptionPredicateIsUsedForRetryPolicy()
+        => ExceptionPredicateIsUsedForRetryPolicy(FunctionStoreFactory.Create());
 }