Update ResilientFunctions to the CreateAndStart lifecycle and fix wrapper suspension

Cleipnir.Tests.AspNet/BulkPublishTests.cs

@@ -25,7 +25,7 @@ public Task MultipleMessageCanBePublishedUsingSqlServer()
 
     private async Task MultipleMessageCanBePublished(Task<IFunctionStore> storeTask)
     {
-        using var container = FlowsContainer.Create(functionStore: await storeTask);
+        using var container = await FlowsContainer.CreateAndStart(functionStore: await storeTask);
         var flows = container.RegisterAnonymousFlow(() => new TestFlow(), flowName: $"TestFlow#{Guid.NewGuid()}");
 
         var instances = Enumerable.Range(0, 100)

Cleipnir.Tests/Flows/FlowFactoryTests.cs

@@ -9,7 +9,7 @@ public class FlowFactoryTests
     [TestMethod]
     public async Task FlowFactoryForTestFlowWithResultIsUsedWhenProvided()
     {
-        var flowsContainer = FlowsContainer.Create();
+        var flowsContainer = await FlowsContainer.CreateAndStart();
         var flows = flowsContainer.RegisterAnonymousFlow<TestFlowWithResult, string, string>(flowFactory: () => new TestFlowWithResult());
         var result = await flows.Run("SomeInstance", "hallo world");
         result.ShouldBe("HALLO WORLD");
@@ -30,7 +30,7 @@ public override Task<string> Run(string param)
     [TestMethod]
     public async Task FlowFactoryForTestFlowWithParamIsUsedWhenProvided()
     {
-        var flowsContainer = FlowsContainer.Create();
+        var flowsContainer = await FlowsContainer.CreateAndStart();
         var flows = flowsContainer.RegisterAnonymousFlow<TestFlowWithParam, string>(flowFactory: () => new TestFlowWithParam());
         await flows.Run("SomeInstance", "hallo world");
 
@@ -55,7 +55,7 @@ public override Task Run(string param)
     [TestMethod]
     public async Task FlowFactoryForTestFlowWithoutParamIsUsedWhenProvided()
     {
-        var flowsContainer = FlowsContainer.Create();
+        var flowsContainer = await FlowsContainer.CreateAndStart();
         var flows = flowsContainer.RegisterAnonymousFlow(flowFactory: () => new TestFlowWithoutParam());
         await flows.Run("SomeInstance");
 

Cleipnir.Tests/Flows/FlowsWithResultTests.cs

@@ -17,7 +17,7 @@ public async Task SimpleFlowCompletesSuccessfully()
         serviceCollection.AddTransient<SimpleFuncFlow>();
 
         var flowStore = new InMemoryFunctionStore();
-        var flowsContainer = new FlowsContainer(
+        var flowsContainer = await FlowsContainer.CreateAndStart(
             flowStore,
             serviceCollection.BuildServiceProvider()
         );
@@ -63,7 +63,7 @@ public async Task CompletionOfScheduledFlowCanBeAwaited()
         serviceCollection.AddTransient<SimpledDelayedFlow>();
 
         var flowStore = new InMemoryFunctionStore();
-        var flowsContainer = new FlowsContainer(
+        var flowsContainer = await FlowsContainer.CreateAndStart(
             flowStore,
             serviceCollection.BuildServiceProvider(),
             new Settings(watchdogCheckFrequency: TimeSpan.FromMilliseconds(100))
@@ -100,7 +100,7 @@ public async Task EventDrivenFlowCompletesSuccessfully()
         serviceCollection.AddTransient<MessageDrivenFuncFlow>();
 
         var flowStore = new InMemoryFunctionStore();
-        var flowsContainer = new FlowsContainer(
+        var flowsContainer = await FlowsContainer.CreateAndStart(
             flowStore,
             serviceCollection.BuildServiceProvider(),
             new Settings(watchdogCheckFrequency: TimeSpan.FromMilliseconds(100))
@@ -148,7 +148,7 @@ public async Task FailingFlowCompletesWithError()
         serviceCollection.AddTransient<FailingFuncFlow>();
 
         var flowStore = new InMemoryFunctionStore();
-        var flowsContainer = new FlowsContainer(
+        var flowsContainer = await FlowsContainer.CreateAndStart(
             flowStore,
             serviceCollection.BuildServiceProvider()
         );

Cleipnir.Tests/Flows/FunctionRegistrationTests.cs

@@ -11,7 +11,7 @@ public class FunctionRegistrationTests
     [TestMethod]
     public async Task FunctionCanBeRegisteredAndInvoked()
     {
-        using var container = FlowsContainer.Create();
+        using var container = await FlowsContainer.CreateAndStart();
         var registration = container.Functions.RegisterFunc(
             "TestFlow",
             inner: Task<string> (string param) => param.ToUpper().ToTask()
@@ -24,7 +24,7 @@ public async Task FunctionCanBeRegisteredAndInvoked()
     [TestMethod]
     public async Task FlowCanBeCreatedWithInitialState()
     {
-        var flowsContainer = FlowsContainer.Create();
+        var flowsContainer = await FlowsContainer.CreateAndStart();
         var flow = new InitialStateFlow();
         var flows = flowsContainer.RegisterAnonymousFlow<InitialStateFlow, string, string>(
             flowFactory: () => flow

Cleipnir.Tests/Flows/MultipleRegistrationTests.cs

@@ -1,4 +1,5 @@
 using Cleipnir.ResilientFunctions.Storage;
+using System.Threading.Tasks;
 using Cleipnir.ResilientFunctions.Domain;
 using Microsoft.Extensions.DependencyInjection;
 using Shouldly;
@@ -9,21 +10,21 @@ namespace Cleipnir.Flows.Tests.Flows;
 public class MultipleRegistrationTests
 {
     [TestMethod]
-    public void SameFlowTypeWithSameNameCanBeRegisteredSeveralTimes()
+    public async Task SameFlowTypeWithSameNameCanBeRegisteredSeveralTimes()
     {
         var serviceCollection = new ServiceCollection();
         var store = new InMemoryFunctionStore();
-        var flowsContainer = new FlowsContainer(store, serviceCollection.BuildServiceProvider());
+        var flowsContainer = await FlowsContainer.CreateAndStart(store, serviceCollection.BuildServiceProvider());
         _ = new TestFlowType1s(flowsContainer);
         _ = new TestFlowType1s(flowsContainer);
     }
 
     [TestMethod]
-    public void DifferentFlowTypesWithSameNameCannotBeRegisteredSeveralTimes()
+    public async Task DifferentFlowTypesWithSameNameCannotBeRegisteredSeveralTimes()
     {
         var serviceCollection = new ServiceCollection();
         var store = new InMemoryFunctionStore();
-        var flowsContainer = new FlowsContainer(store, serviceCollection.BuildServiceProvider());
+        var flowsContainer = await FlowsContainer.CreateAndStart(store, serviceCollection.BuildServiceProvider());
         _ = new TestFlowType1s(flowsContainer);
         Should.Throw<InvalidOperationException>(() =>
             new TestFlowType2s(flowsContainer)

Cleipnir.Tests/Flows/OptionsTests.cs

@@ -47,7 +47,7 @@ await Should.ThrowAsync<InvocationSuspendedException>(
         // Flow may be Executing (waiting for message) or Suspended depending on timing
         controlPanel.Status.ShouldBeOneOf(Status.Executing, Status.Suspended);
 
-        await controlPanel.Messages.Append(new StringWrapper("Hello"));
+        await flowsWithDefaultProvidedOptions.MessageWriter("Id").AppendMessage(new StringWrapper("Hello"));
 
         await controlPanel.WaitForCompletion(allowPostponeAndSuspended: true);        
     }

Cleipnir.Tests/Flows/ParamlessFlowsTests.cs

@@ -17,7 +17,7 @@ public async Task SimpleFlowCompletesSuccessfully()
         serviceCollection.AddTransient<SimpleParamlessFlow>();
 
         var flowStore = new InMemoryFunctionStore();
-        var flowsContainer = new FlowsContainer(
+        var flowsContainer = await FlowsContainer.CreateAndStart(
             flowStore,
             serviceCollection.BuildServiceProvider()
         );
@@ -56,7 +56,7 @@ public async Task EventDrivenFlowCompletesSuccessfully()
         serviceCollection.AddTransient<EventDrivenParamlessFlow>();
 
         var flowStore = new InMemoryFunctionStore();
-        var flowsContainer = new FlowsContainer(
+        var flowsContainer = await FlowsContainer.CreateAndStart(
             flowStore,
             serviceCollection.BuildServiceProvider(),
             new Settings(watchdogCheckFrequency: TimeSpan.FromMilliseconds(100))
@@ -102,7 +102,7 @@ public async Task FailingFlowCompletesWithError()
         serviceCollection.AddTransient<FailingParamlessFlow>();
 
         var flowStore = new InMemoryFunctionStore();
-        var flowsContainer = new FlowsContainer(
+        var flowsContainer = await FlowsContainer.CreateAndStart(
             flowStore,
             serviceCollection.BuildServiceProvider()
         );
@@ -146,7 +146,7 @@ public override Task Run()
     [TestMethod]
     public async Task FlowCanBeCreatedWithInitialState()
     {
-        var flowsContainer = FlowsContainer.Create();
+        var flowsContainer = await FlowsContainer.CreateAndStart();
         var flow = new InitialStateFlow();
         var flows = flowsContainer.RegisterAnonymousFlow(
             flowFactory: () => flow

Cleipnir.Tests/Flows/UnitFlowsTests.cs

@@ -18,7 +18,7 @@ public async Task SimpleFlowCompletesSuccessfully()
         serviceCollection.AddTransient<SimpleUnitFlow>();
 
         var flowStore = new InMemoryFunctionStore();
-        var flowsContainer = new FlowsContainer(
+        var flowsContainer = await FlowsContainer.CreateAndStart(
             flowStore,
             serviceCollection.BuildServiceProvider()
         );
@@ -60,7 +60,7 @@ public async Task EventDrivenFlowCompletesSuccessfully()
         serviceCollection.AddTransient<EventDrivenUnitFlow>();
 
         var flowStore = new InMemoryFunctionStore();
-        var flowsContainer = new FlowsContainer(
+        var flowsContainer = await FlowsContainer.CreateAndStart(
             flowStore,
             serviceCollection.BuildServiceProvider(),
             new Settings(watchdogCheckFrequency: TimeSpan.FromMilliseconds(100))
@@ -107,7 +107,7 @@ public async Task FailingActionFlowCompletesWithError()
         serviceCollection.AddTransient<FailingUnitActionFlow>();
 
         var flowStore = new InMemoryFunctionStore();
-        var flowsContainer = new FlowsContainer(
+        var flowsContainer = await FlowsContainer.CreateAndStart(
             flowStore,
             serviceCollection.BuildServiceProvider()
         );
@@ -156,7 +156,7 @@ public async Task FailingFuncFlowCompletesWithError()
         serviceCollection.AddTransient<FailingUnitFuncFlow>();
 
         var flowStore = new InMemoryFunctionStore();
-        var flowsContainer = new FlowsContainer(
+        var flowsContainer = await FlowsContainer.CreateAndStart(
             flowStore,
             serviceCollection.BuildServiceProvider()
         );
@@ -205,7 +205,7 @@ public async Task FailingParamlessFlowCompletesWithError()
         serviceCollection.AddTransient<FailingUnitParamlessFlow>();
 
         var flowStore = new InMemoryFunctionStore();
-        var flowsContainer = new FlowsContainer(
+        var flowsContainer = await FlowsContainer.CreateAndStart(
             flowStore,
             serviceCollection.BuildServiceProvider()
         );
@@ -250,7 +250,7 @@ public override Task Run()
     [TestMethod]
     public async Task FlowCanBeCreatedWithInitialState()
     {
-        var flowsContainer = FlowsContainer.Create();
+        var flowsContainer = await FlowsContainer.CreateAndStart();
         var flow = new InitialStateFlow();
         var flows = flowsContainer.RegisterAnonymousFlow<InitialStateFlow, string>(
             flowFactory: () => flow

Cleipnir.Tests/Flows/UtcNowTests.cs

@@ -15,7 +15,7 @@ public async Task ProvidedUtcNowDelegateIsUsed()
 
         var serviceCollection = new ServiceCollection();
         serviceCollection.AddSingleton<UtcNowTestFlow>();
-        var flowsContainer = new FlowsContainer(
+        var flowsContainer = await FlowsContainer.CreateAndStart(
             new InMemoryFunctionStore(),
             serviceCollection.BuildServiceProvider(),
             new Settings(utcNow: () => now, watchdogCheckFrequency: TimeSpan.FromMilliseconds(100))

Cleipnir/AspNet/FlowsModule.cs

@@ -20,8 +20,16 @@ public static IServiceCollection AddFlows(this IServiceCollection services, Func
             services.AddSingleton(new Settings());
         else
             services.AddSingleton(configurator.OptionsFunc);
-
-        services.AddSingleton<FlowsContainer>();
+
+        services.AddSingleton<FlowsContainer>(sp =>
+            {
+                var store = sp.GetRequiredService<IFunctionStore>();
+                var settings = configurator.OptionsFunc?.Invoke(sp) ?? sp.GetService<Settings>();
+                // Resolved once during host start-up; blocking here mirrors the old constructor, which
+                // initialized the replica watchdog synchronously.
+                return FlowsContainer.CreateAndStart(store, sp, settings).GetAwaiter().GetResult();
+            }
+        );
         services.AddTransient<Workflow>(_ => CurrentFlow.Workflow ?? throw new InvalidOperationException("Workflow is not present outside Flow"));
         services.AddTransient<Effect>(_ => CurrentFlow.Workflow?.Effect ?? throw new InvalidOperationException("Effect is not present outside Flow"));
 

Cleipnir/FlowOptions.cs

@@ -8,19 +8,16 @@ public class FlowOptions
     public static FlowOptions Default { get; } = new();
 
     internal TimeSpan? RetentionPeriod { get; }
-    internal bool? EnableWatchdogs { get; }
     internal int? MaxParallelRetryInvocations { get; }
     internal TimeSpan? MessagesDefaultMaxWaitForCompletion { get; }
 
     public FlowOptions(
         TimeSpan? retentionPeriod = null,
-        bool? enableWatchdogs = null,
         TimeSpan? messagesDefaultMaxWaitForCompletion = null,
         int? maxParallelRetryInvocations = null
     )
     {
         RetentionPeriod = retentionPeriod;
-        EnableWatchdogs = enableWatchdogs;
         MessagesDefaultMaxWaitForCompletion = messagesDefaultMaxWaitForCompletion;
         MaxParallelRetryInvocations = maxParallelRetryInvocations;
     }
@@ -29,12 +26,11 @@ public FlowOptions Merge(Settings settings)
     {
         return new FlowOptions(
             RetentionPeriod ?? settings.RetentionPeriod,
-            EnableWatchdogs ?? settings.EnableWatchdogs,
             MessagesDefaultMaxWaitForCompletion ?? settings.MessagesDefaultMaxWaitForCompletion,
             MaxParallelRetryInvocations ?? settings.MaxParallelRetryInvocations
         );
     }
 
     internal LocalSettings MapToLocalSettings()
-        => new(RetentionPeriod, EnableWatchdogs, MessagesDefaultMaxWaitForCompletion, MaxParallelRetryInvocations);
+        => new(RetentionPeriod, MessagesDefaultMaxWaitForCompletion, MaxParallelRetryInvocations);
 }