Fix race condition on stop

src/WorkflowCore.Testing/WorkflowTest.cs

@@ -17,6 +17,7 @@ public abstract class WorkflowTest<TWorkflow, TData> : IDisposable
         protected IWorkflowHost Host;
         protected IPersistenceProvider PersistenceProvider;
         protected List<StepError> UnhandledStepErrors = new List<StepError>();
+        private bool isDisposed;
 
         protected virtual void Setup()
         {
@@ -116,9 +117,22 @@ protected TData GetData(string workflowId)
             return (TData)instance.Data;
         }
 
+        protected virtual void Dispose(bool disposing)
+        {
+            if (!isDisposed)
+            {
+                if (disposing)
+                {
+                    Host.Stop();
+                }
+                isDisposed = true;
+            }
+        }
+
         public void Dispose()
         {
-            Host.Stop();
+            Dispose(disposing: true);
+            GC.SuppressFinalize(this);
         }
     }
 

src/WorkflowCore/Services/BackgroundTasks/WorkflowConsumer.cs

@@ -55,7 +55,7 @@ protected override async Task ProcessItem(string itemId, CancellationToken cance
                     finally
                     {
                         WorkflowActivity.Enrich(result);
-                        await _persistenceStore.PersistWorkflow(workflow, result?.Subscriptions, cancellationToken);
+                        await _persistenceStore.PersistWorkflow(workflow, result?.Subscriptions);
                         await QueueProvider.QueueWork(itemId, QueueType.Index);
                         _greylist.Remove($"wf:{itemId}");
                     }

test/Docker.Testify/Docker.Testify.csproj

@@ -1,7 +1,7 @@
 <Project Sdk="Microsoft.NET.Sdk">
 
   <ItemGroup>
-    <PackageReference Include="Docker.DotNet" Version="3.125.2" />
+    <PackageReference Include="Docker.DotNet" Version="3.125.10" />
     <PackageReference Include="System.Diagnostics.Process" Version="4.3.0" />
     <PackageReference Include="System.Net.NetworkInformation" Version="4.3.0" />
     <PackageReference Include="System.Threading.Thread" Version="4.3.0" />

test/WorkflowCore.IntegrationTests/Scenarios/StopScenario.cs

@@ -0,0 +1,47 @@
+using System.Threading;
+using System.Threading.Tasks;
+using FluentAssertions;
+using WorkflowCore.Interface;
+using WorkflowCore.Models;
+using WorkflowCore.Models.LifeCycleEvents;
+using WorkflowCore.Testing;
+using Xunit;
+
+namespace WorkflowCore.IntegrationTests.Scenarios
+{
+    public class StopScenario : WorkflowTest<StopScenario.StopWorkflow, object>
+    {
+        public class StopWorkflow : IWorkflow
+        {
+            public string Id => "StopWorkflow";
+            public int Version => 1;
+            public void Build(IWorkflowBuilder<object> builder)
+            {
+                builder.StartWith(context => ExecutionResult.Next());
+            }
+        }
+
+        public StopScenario() => Setup();
+
+        [Fact]
+        public async Task Scenario()
+        {
+            var tcs = new TaskCompletionSource<object>();
+            Host.OnLifeCycleEvent += async (evt) =>
+            {
+                if (evt is WorkflowCompleted)
+                {
+                    await Host.StopAsync(CancellationToken.None);
+                    tcs.SetResult(default);
+                }
+            };
+
+            var workflowId = StartWorkflow(default);
+            await tcs.Task;
+
+            GetStatus(workflowId).Should().Be(WorkflowStatus.Complete);
+        }
+
+        protected override void Dispose(bool disposing) { }
+    }
+}

test/WorkflowCore.Tests.DynamoDB/Scenarios/DynamoStopScenario.cs

@@ -0,0 +1,18 @@
+using System;
+using Amazon.DynamoDBv2;
+using Microsoft.Extensions.DependencyInjection;
+using WorkflowCore.IntegrationTests.Scenarios;
+using Xunit;
+
+namespace WorkflowCore.Tests.DynamoDB.Scenarios
+{
+    [Collection("DynamoDb collection")]
+    public class DynamoStopScenario : StopScenario
+    {        
+        protected override void ConfigureServices(IServiceCollection services)
+        {
+            var cfg = new AmazonDynamoDBConfig {ServiceURL = DynamoDbDockerSetup.ConnectionString};
+            services.AddWorkflow(x => x.UseAwsDynamoPersistence(DynamoDbDockerSetup.Credentials, cfg, "tests-"));
+        }
+    }
+}

test/WorkflowCore.Tests.MongoDB/Scenarios/MongoStopScenario.cs

@@ -0,0 +1,16 @@
+using System;
+using Microsoft.Extensions.DependencyInjection;
+using WorkflowCore.IntegrationTests.Scenarios;
+using Xunit;
+
+namespace WorkflowCore.Tests.MongoDB.Scenarios
+{
+    [Collection("Mongo collection")]
+    public class MongoStopScenario : StopScenario
+    {        
+        protected override void ConfigureServices(IServiceCollection services)
+        {
+            services.AddWorkflow(x => x.UseMongoDB(MongoDockerSetup.ConnectionString, "integration-tests"));
+        }
+    }
+}

test/WorkflowCore.Tests.MySQL/Scenarios/MysqlStopScenario.cs

@@ -0,0 +1,16 @@
+using System;
+using Microsoft.Extensions.DependencyInjection;
+using WorkflowCore.IntegrationTests.Scenarios;
+using Xunit;
+
+namespace WorkflowCore.Tests.MySQL.Scenarios
+{
+    [Collection("Mysql collection")]
+    public class MysqlStopScenario : StopScenario
+    {
+        protected override void ConfigureServices(IServiceCollection services)
+        {
+            services.AddWorkflow(x => x.UseMySQL(MysqlDockerSetup.ScenarioConnectionString, true, true));
+        }
+    }
+}

test/WorkflowCore.Tests.PostgreSQL/Scenarios/PostgresStopScenario.cs

@@ -0,0 +1,16 @@
+using System;
+using Microsoft.Extensions.DependencyInjection;
+using WorkflowCore.IntegrationTests.Scenarios;
+using Xunit;
+
+namespace WorkflowCore.Tests.PostgreSQL.Scenarios
+{
+    [Collection("Postgres collection")]
+    public class PostgresStopScenario : StopScenario
+    {        
+        protected override void ConfigureServices(IServiceCollection services)
+        {
+            services.AddWorkflow(x => x.UsePostgreSQL(PostgresDockerSetup.ScenarioConnectionString, true, true));
+        }
+    }
+}

test/WorkflowCore.Tests.Redis/Scenarios/RedisStopScenario.cs

@@ -0,0 +1,16 @@
+using System;
+using Microsoft.Extensions.DependencyInjection;
+using WorkflowCore.IntegrationTests.Scenarios;
+using Xunit;
+
+namespace WorkflowCore.Tests.Redis.Scenarios
+{
+    [Collection("Redis collection")]
+    public class RedisStopScenario : StopScenario
+    {        
+        protected override void ConfigureServices(IServiceCollection services)
+        {
+            services.AddWorkflow(x => x.UseRedisPersistence(RedisDockerSetup.ConnectionString, "scenario-"));
+        }
+    }
+}

test/WorkflowCore.Tests.SqlServer/Scenarios/SqlServerStopScenario.cs

@@ -0,0 +1,16 @@
+using System;
+using Microsoft.Extensions.DependencyInjection;
+using WorkflowCore.IntegrationTests.Scenarios;
+using Xunit;
+
+namespace WorkflowCore.Tests.SqlServer.Scenarios
+{
+    [Collection("SqlServer collection")]
+    public class SqlServerStopScenario : StopScenario
+    {        
+        protected override void ConfigureServices(IServiceCollection services)
+        {
+            services.AddWorkflow(x => x.UseSqlServer(SqlDockerSetup.ScenarioConnectionString, true, true));
+        }
+    }
+}

test/WorkflowCore.Tests.Sqlite/Scenarios/SqliteStopScenario.cs

@@ -0,0 +1,17 @@
+using System;
+using Microsoft.Extensions.DependencyInjection;
+using WorkflowCore.IntegrationTests.Scenarios;
+using WorkflowCore.Tests.Sqlite;
+using Xunit;
+
+namespace WorkflowCore.Tests.Sqlite.Scenarios
+{
+    [Collection("Sqlite collection")]
+    public class SqliteStopScenario : StopScenario
+    {
+        protected override void ConfigureServices(IServiceCollection services)
+        {
+            services.AddWorkflow(x => x.UseSqlite(SqliteSetup.ConnectionString, true));
+        }
+    }
+}

test/WorkflowCore.Tests.Sqlite/SqliteCollection.cs

@@ -10,7 +10,7 @@ public class SqliteCollection : ICollectionFixture<SqliteSetup>
 
     public class SqliteSetup : IDisposable
     {
-        public string ConnectionString { get; set; }
+        public static string ConnectionString { get; set; }
 
         public SqliteSetup()
         {

test/WorkflowCore.Tests.Sqlite/SqlitePersistenceProviderFixture.cs

@@ -10,18 +10,15 @@ namespace WorkflowCore.Tests.Sqlite
     [Collection("Sqlite collection")]
     public class SqlitePersistenceProviderFixture : BasePersistenceFixture
     {
-        string _connectionString;
-
         public SqlitePersistenceProviderFixture(SqliteSetup setup)
         {
-            _connectionString = setup.ConnectionString;
         }
 
         protected override IPersistenceProvider Subject
         {
             get
-            {                
-                var db = new EntityFrameworkPersistenceProvider(new SqliteContextFactory(_connectionString), true, false);
+            {
+                var db = new EntityFrameworkPersistenceProvider(new SqliteContextFactory(SqliteSetup.ConnectionString), true, false);
                 db.EnsureStoreExists();
                 return db;
             }