Fixed reliability of WorkflowConsumer when persisting workflow

viktor-shevchenko-ideals · viktor-shevchenko-ideals · commit 202e7d36826b · 2022-08-05T14:36:33.000+03:00
diff --git a/src/WorkflowCore/Interface/Persistence/IWorkflowRepository.cs b/src/WorkflowCore/Interface/Persistence/IWorkflowRepository.cs
@@ -12,6 +12,8 @@ public interface IWorkflowRepository
 
         Task PersistWorkflow(WorkflowInstance workflow, CancellationToken cancellationToken = default);
 
+        Task PersistWorkflow(WorkflowInstance workflow, List<EventSubscription> subscriptions, CancellationToken cancellationToken = default);
+
         Task<IEnumerable<string>> GetRunnableInstances(DateTime asAt, CancellationToken cancellationToken = default);
 
         [Obsolete]
diff --git a/src/WorkflowCore/Services/BackgroundTasks/WorkflowConsumer.cs b/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, cancellationToken);
+                        await _persistenceStore.PersistWorkflow(workflow, result.Subscriptions, cancellationToken);
                         await QueueProvider.QueueWork(itemId, QueueType.Index);
                         _greylist.Remove($"wf:{itemId}");
                     }
@@ -100,10 +100,6 @@ await _persistenceStore.ScheduleCommand(new ScheduledCommand()
 
         private async Task SubscribeEvent(EventSubscription subscription, IPersistenceProvider persistenceStore, CancellationToken cancellationToken)
         {
-            //TODO: move to own class
-            Logger.LogDebug("Subscribing to event {0} {1} for workflow {2} step {3}", subscription.EventName, subscription.EventKey, subscription.WorkflowId, subscription.StepId);
-
-            await persistenceStore.CreateEventSubscription(subscription, cancellationToken);
             if (subscription.EventName != Event.EventTypeActivity)
             {
                 var events = await persistenceStore.GetEvents(subscription.EventName, subscription.EventKey, subscription.SubscribeAsOf, cancellationToken);
diff --git a/src/WorkflowCore/Services/DefaultProviders/MemoryPersistenceProvider.cs b/src/WorkflowCore/Services/DefaultProviders/MemoryPersistenceProvider.cs
@@ -46,6 +46,25 @@ public async Task PersistWorkflow(WorkflowInstance workflow, CancellationToken _
             }
         }
 
+        public async Task PersistWorkflow(WorkflowInstance workflow, List<EventSubscription> subscriptions, CancellationToken cancellationToken = default)
+        {
+            lock (_instances)
+            {
+                var existing = _instances.First(x => x.Id == workflow.Id);
+                _instances.Remove(existing);
+                _instances.Add(workflow);
+
+                lock (_subscriptions)
+                {
+                    foreach (var subscription in subscriptions)
+                    {
+                        subscription.Id = Guid.NewGuid().ToString();
+                        _subscriptions.Add(subscription);
+                    }
+                }
+            }
+        }
+
         public async Task<IEnumerable<string>> GetRunnableInstances(DateTime asAt, CancellationToken _ = default)
         {
             lock (_instances)
diff --git a/src/WorkflowCore/Services/DefaultProviders/TransientMemoryPersistenceProvider.cs b/src/WorkflowCore/Services/DefaultProviders/TransientMemoryPersistenceProvider.cs
@@ -50,6 +50,16 @@ public TransientMemoryPersistenceProvider(ISingletonMemoryProvider innerService)
 
         public Task PersistWorkflow(WorkflowInstance workflow, CancellationToken _ = default) => _innerService.PersistWorkflow(workflow);
 
+        public async Task PersistWorkflow(WorkflowInstance workflow, List<EventSubscription> subscriptions, CancellationToken cancellationToken = default)
+        {
+            await PersistWorkflow(workflow, cancellationToken);
+
+            foreach(var subscription in subscriptions)
+            {
+                await CreateEventSubscription(subscription, cancellationToken);
+            }
+        }
+
         public Task TerminateSubscription(string eventSubscriptionId, CancellationToken _ = default) => _innerService.TerminateSubscription(eventSubscriptionId);
         public Task<EventSubscription> GetSubscription(string eventSubscriptionId, CancellationToken _ = default) => _innerService.GetSubscription(eventSubscriptionId);
 
diff --git a/src/providers/WorkflowCore.Persistence.EntityFramework/Services/EntityFrameworkPersistenceProvider.cs b/src/providers/WorkflowCore.Persistence.EntityFramework/Services/EntityFrameworkPersistenceProvider.cs
@@ -151,6 +151,32 @@ public async Task PersistWorkflow(WorkflowInstance workflow, CancellationToken c
                 await db.SaveChangesAsync(cancellationToken);
             }
         }
+		
+		public async Task PersistWorkflow(WorkflowInstance workflow, List<EventSubscription> subscriptions, CancellationToken cancellationToken = default)
+        {
+            using (var db = ConstructDbContext())
+            {
+                var uid = new Guid(workflow.Id);
+                var existingEntity = await db.Set<PersistedWorkflow>()
+                    .Where(x => x.InstanceId == uid)
+                    .Include(wf => wf.ExecutionPointers)
+                    .ThenInclude(ep => ep.ExtensionAttributes)
+                    .Include(wf => wf.ExecutionPointers)
+                    .AsTracking()
+                    .FirstAsync(cancellationToken);
+
+                var workflowPersistable = workflow.ToPersistable(existingEntity);
+
+                foreach (var subscription in subscriptions)
+                {
+                    subscription.Id = Guid.NewGuid().ToString();
+                    var subscriptionPersistable = subscription.ToPersistable();
+                    db.Set<PersistedSubscription>().Add(subscriptionPersistable);
+                }
+
+                await db.SaveChangesAsync(cancellationToken);
+            }
+        }
 
         public async Task TerminateSubscription(string eventSubscriptionId, CancellationToken cancellationToken = default)
         {
diff --git a/src/providers/WorkflowCore.Persistence.MongoDB/Services/MongoPersistenceProvider.cs b/src/providers/WorkflowCore.Persistence.MongoDB/Services/MongoPersistenceProvider.cs
@@ -149,6 +149,17 @@ public async Task PersistWorkflow(WorkflowInstance workflow, CancellationToken c
             await WorkflowInstances.ReplaceOneAsync(x => x.Id == workflow.Id, workflow, cancellationToken: cancellationToken);
         }
 
+        public async Task PersistWorkflow(WorkflowInstance workflow, List<EventSubscription> subscriptions, CancellationToken cancellationToken = default)
+        {
+            using (var session = await _database.Client.StartSessionAsync())
+            {
+                session.StartTransaction();
+                await PersistWorkflow(workflow, cancellationToken);
+                await EventSubscriptions.InsertManyAsync(subscriptions, cancellationToken: cancellationToken);
+                await session.CommitTransactionAsync();
+            }
+        }
+
         public async Task<IEnumerable<string>> GetRunnableInstances(DateTime asAt, CancellationToken cancellationToken = default)
         {
             var now = asAt.ToUniversalTime().Ticks;
diff --git a/src/providers/WorkflowCore.Persistence.RavenDB/Services/RavendbPersistenceProvider.cs b/src/providers/WorkflowCore.Persistence.RavenDB/Services/RavendbPersistenceProvider.cs
@@ -1,6 +1,7 @@
 ﻿using Raven.Client.Documents;
 using Raven.Client.Documents.Linq;
 using Raven.Client.Documents.Operations;
+using Raven.Client.Documents.Session;
 using System;
 using System.Collections.Generic;
 using System.Linq;
@@ -56,21 +57,41 @@ public async Task PersistWorkflow(WorkflowInstance workflow, CancellationToken c
 		{
 			using (var session = _database.OpenAsyncSession())
 			{
-				session.Advanced.Patch<WorkflowInstance, string>(workflow.Id, x => x.WorkflowDefinitionId, workflow.WorkflowDefinitionId);
-				session.Advanced.Patch<WorkflowInstance, int>(workflow.Id, x => x.Version, workflow.Version);
-				session.Advanced.Patch<WorkflowInstance, string>(workflow.Id, x => x.Description, workflow.Description);
-				session.Advanced.Patch<WorkflowInstance, string>(workflow.Id, x => x.Reference, workflow.Reference);
-				session.Advanced.Patch<WorkflowInstance, ExecutionPointerCollection>(workflow.Id, x => x.ExecutionPointers, workflow.ExecutionPointers);
-				session.Advanced.Patch<WorkflowInstance, long?>(workflow.Id, x => x.NextExecution, workflow.NextExecution);
-				session.Advanced.Patch<WorkflowInstance, WorkflowStatus>(workflow.Id, x => x.Status, workflow.Status);
-				session.Advanced.Patch<WorkflowInstance, object>(workflow.Id, x => x.Data, workflow.Data);
-				session.Advanced.Patch<WorkflowInstance, DateTime>(workflow.Id, x => x.CreateTime, workflow.CreateTime);
-				session.Advanced.Patch<WorkflowInstance, DateTime?>(workflow.Id, x => x.CompleteTime, workflow.CompleteTime);
+				PatchSession(session, workflow);
+				await session.SaveChangesAsync(cancellationToken);
+			}
+		}
+
+		public async Task PersistWorkflow(WorkflowInstance workflow, List<EventSubscription> subscriptions, CancellationToken cancellationToken = default)
+		{
+			using (var session = _database.OpenAsyncSession())
+			{
+				PatchSession(session, workflow);
+
+                foreach (var subscription in subscriptions)
+				{
+					await session.StoreAsync(subscription, cancellationToken);
+				}
 
 				await session.SaveChangesAsync(cancellationToken);
 			}
 		}
 
+		private void PatchSession(IAsyncDocumentSession session, WorkflowInstance workflow)
+        {
+			session.Advanced.Patch<WorkflowInstance, string>(workflow.Id, x => x.WorkflowDefinitionId, workflow.WorkflowDefinitionId);
+			session.Advanced.Patch<WorkflowInstance, int>(workflow.Id, x => x.Version, workflow.Version);
+			session.Advanced.Patch<WorkflowInstance, string>(workflow.Id, x => x.Description, workflow.Description);
+			session.Advanced.Patch<WorkflowInstance, string>(workflow.Id, x => x.Reference, workflow.Reference);
+			session.Advanced.Patch<WorkflowInstance, ExecutionPointerCollection>(workflow.Id, x => x.ExecutionPointers, workflow.ExecutionPointers);
+			session.Advanced.Patch<WorkflowInstance, long?>(workflow.Id, x => x.NextExecution, workflow.NextExecution);
+			session.Advanced.Patch<WorkflowInstance, WorkflowStatus>(workflow.Id, x => x.Status, workflow.Status);
+			session.Advanced.Patch<WorkflowInstance, object>(workflow.Id, x => x.Data, workflow.Data);
+			session.Advanced.Patch<WorkflowInstance, DateTime>(workflow.Id, x => x.CreateTime, workflow.CreateTime);
+			session.Advanced.Patch<WorkflowInstance, DateTime?>(workflow.Id, x => x.CompleteTime, workflow.CompleteTime);
+
+		}
+
 		public async Task<IEnumerable<string>> GetRunnableInstances(DateTime asAt, CancellationToken cancellationToken = default)
 		{
 			var now = asAt.ToUniversalTime().Ticks;
diff --git a/src/providers/WorkflowCore.Providers.AWS/Services/DynamoPersistenceProvider.cs b/src/providers/WorkflowCore.Providers.AWS/Services/DynamoPersistenceProvider.cs
@@ -61,6 +61,43 @@ public async Task PersistWorkflow(WorkflowInstance workflow, CancellationToken c
             var response = await _client.PutItemAsync(request, cancellationToken);
         }
 
+        public async Task PersistWorkflow(WorkflowInstance workflow, List<EventSubscription> subscriptions, CancellationToken cancellationToken = default)
+        {
+            var transactionWriteItemsRequest = new TransactWriteItemsRequest()
+            {
+                TransactItems = new List<TransactWriteItem>()
+                {
+                    { 
+                        new TransactWriteItem() 
+                        { 
+                            Put = new Put() 
+                            {
+                                TableName = $"{_tablePrefix}-{WORKFLOW_TABLE}",
+                                Item = workflow.ToDynamoMap()
+                            } 
+                        } 
+                    }
+                }
+            };
+
+            foreach(var subscription in subscriptions)
+            {
+                subscription.Id = Guid.NewGuid().ToString();
+
+                transactionWriteItemsRequest.TransactItems.Add(new TransactWriteItem() 
+                {
+                    Put = new Put()
+                    {
+                        TableName = $"{_tablePrefix}-{SUBCRIPTION_TABLE}",
+                        Item = subscription.ToDynamoMap(),
+                        ConditionExpression = "attribute_not_exists(id)"
+                    }
+                });
+            }
+
+            await _client.TransactWriteItemsAsync(transactionWriteItemsRequest, cancellationToken);
+        }
+
         public async Task<IEnumerable<string>> GetRunnableInstances(DateTime asAt, CancellationToken cancellationToken = default)
         {
             var result = new List<string>();
diff --git a/src/providers/WorkflowCore.Providers.Azure/Services/CosmosDbPersistenceProvider.cs b/src/providers/WorkflowCore.Providers.Azure/Services/CosmosDbPersistenceProvider.cs
@@ -227,6 +227,16 @@ public async Task PersistWorkflow(WorkflowInstance workflow, CancellationToken c
             await _workflowContainer.Value.UpsertItemAsync(PersistedWorkflow.FromInstance(workflow), cancellationToken: cancellationToken);
         }
 
+        public async Task PersistWorkflow(WorkflowInstance workflow, List<EventSubscription> subscriptions, CancellationToken cancellationToken = default)
+        {
+            await PersistWorkflow(workflow, cancellationToken);
+
+            foreach(var subscription in subscriptions)
+            {
+                await CreateEventSubscription(subscription, cancellationToken);
+            }
+        }
+
         public Task ProcessCommands(DateTimeOffset asOf, Func<ScheduledCommand, Task> action, CancellationToken cancellationToken = default)
         {
             throw new NotImplementedException();
diff --git a/src/providers/WorkflowCore.Providers.Redis/Services/RedisPersistenceProvider.cs b/src/providers/WorkflowCore.Providers.Redis/Services/RedisPersistenceProvider.cs
@@ -47,6 +47,16 @@ public async Task<string> CreateNewWorkflow(WorkflowInstance workflow, Cancellat
             return workflow.Id;
         }
 
+        public async Task PersistWorkflow(WorkflowInstance workflow, List<EventSubscription> subscriptions, CancellationToken cancellationToken = default)
+        {
+            await PersistWorkflow(workflow, cancellationToken);
+
+            foreach (var subscription in subscriptions)
+            {
+                await CreateEventSubscription(subscription, cancellationToken);
+            }
+        }
+
         public async Task PersistWorkflow(WorkflowInstance workflow, CancellationToken _ = default)
         {
             var str = JsonConvert.SerializeObject(workflow, _serializerSettings);
diff --git a/test/WorkflowCore.UnitTests/BasePersistenceFixture.cs b/test/WorkflowCore.UnitTests/BasePersistenceFixture.cs
@@ -185,6 +185,72 @@ public void PersistWorkflow()
             var current = Subject.GetWorkflowInstance(workflowId).Result;
             current.ShouldBeEquivalentTo(newWorkflow);
         }
+		
+		[Fact]
+        public void PersistWorkflow_with_subscriptions()
+        {
+            var workflow = new WorkflowInstance
+            {
+                Data = new TestData { Value1 = 7 },
+                Description = "My Description",
+                Status = WorkflowStatus.Runnable,
+                NextExecution = 0,
+                Version = 1,
+                WorkflowDefinitionId = "My Workflow",
+                CreateTime = new DateTime(2000, 1, 1).ToUniversalTime(),
+                ExecutionPointers = new ExecutionPointerCollection(),
+                Reference = Guid.NewGuid().ToString()
+            };
+
+            workflow.ExecutionPointers.Add(new ExecutionPointer
+            {
+                Id = Guid.NewGuid().ToString(),
+                Active = true,
+                StepId = 0,
+                Scope = new List<string> { "1", "2", "3", "4" },
+                EventName = "Event1"
+            });
+
+            workflow.ExecutionPointers.Add(new ExecutionPointer
+            {
+                Id = Guid.NewGuid().ToString(),
+                Active = true,
+                StepId = 1,
+                Scope = new List<string> { "1", "2", "3", "4" },
+                EventName = "Event2",
+            });
+
+            var workflowId = Subject.CreateNewWorkflow(workflow).Result;
+            workflow.NextExecution = 0;
+
+            List<EventSubscription> subscriptions = new List<EventSubscription>();
+            foreach (var pointer in workflow.ExecutionPointers)
+            {
+                var subscription = new EventSubscription()
+                {
+                    WorkflowId = workflowId,
+                    StepId = pointer.StepId,
+                    ExecutionPointerId = pointer.Id,
+                    EventName = pointer.EventName,
+                    EventKey = workflowId,
+                    SubscribeAsOf = DateTime.UtcNow,
+                    SubscriptionData = "data"
+                };
+
+                subscriptions.Add(subscription);
+            }
+
+            Subject.PersistWorkflow(workflow, subscriptions).Wait();
+
+            var current = Subject.GetWorkflowInstance(workflowId).Result;
+            current.ShouldBeEquivalentTo(workflow);
+
+            foreach (var pointer in workflow.ExecutionPointers)
+            {
+                subscriptions = Subject.GetSubscriptions(pointer.EventName, workflowId, DateTime.UtcNow).Result.ToList();
+                subscriptions.Should().HaveCount(1);
+            }
+        }
 
         [Fact]
         public void ConcurrentPersistWorkflow()

Original file line number	Diff line number	Diff line change
`@@ -55,7 +55,7 @@ protected override async Task ProcessItem(string itemId, CancellationToken cance`
`55`	`55`	`finally`
`56`	`56`	`{`
`57`	`57`	`WorkflowActivity.Enrich(result);`
`58`		`- await _persistenceStore.PersistWorkflow(workflow, cancellationToken);`
	`58`	`+ await _persistenceStore.PersistWorkflow(workflow, result.Subscriptions, cancellationToken);`
`59`	`59`	`await QueueProvider.QueueWork(itemId, QueueType.Index);`
`60`	`60`	`_greylist.Remove($"wf:{itemId}");`
`61`	`61`	`}`
`@@ -100,10 +100,6 @@ await _persistenceStore.ScheduleCommand(new ScheduledCommand()`
`100`	`100`
`101`	`101`	`private async Task SubscribeEvent(EventSubscription subscription, IPersistenceProvider persistenceStore, CancellationToken cancellationToken)`
`102`	`102`	`{`
`103`		`- //TODO: move to own class`
`104`		`- Logger.LogDebug("Subscribing to event {0} {1} for workflow {2} step {3}", subscription.EventName, subscription.EventKey, subscription.WorkflowId, subscription.StepId);`
`105`		`-`
`106`		`- await persistenceStore.CreateEventSubscription(subscription, cancellationToken);`
`107`	`103`	`if (subscription.EventName != Event.EventTypeActivity)`
`108`	`104`	`{`
`109`	`105`	`var events = await persistenceStore.GetEvents(subscription.EventName, subscription.EventKey, subscription.SubscribeAsOf, cancellationToken);`