Merge pull request #936 from danielgerlag/scheduled-commands

Scheduled commands

ReleaseNotes/3.6.0.md

@@ -0,0 +1,8 @@
+# Workflow Core 3.6.0
+
+## Scheduled Commands
+
+Introduces the ability to schedule delayed commands to process a workflow or event, by persisting them to storage.
+This is the first step toward removing constant polling of the DB.  It also filters out duplicate work items on the queue which is the current problem the greylist tries to solve.
+Initial implementation is supported by MongoDb, SQL Server, PostgeSQL, MySQL and SQLite.
+Additional support from the other persistence providers to follow.

WorkflowCore.sln

@@ -104,6 +104,8 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "ReleaseNotes", "ReleaseNote
 		ReleaseNotes\3.0.0.md = ReleaseNotes\3.0.0.md
 		ReleaseNotes\3.1.0.md = ReleaseNotes\3.1.0.md
 		ReleaseNotes\3.3.0.md = ReleaseNotes\3.3.0.md
+		ReleaseNotes\3.4.0.md = ReleaseNotes\3.4.0.md
+		ReleaseNotes\3.6.0.md = ReleaseNotes\3.6.0.md
 	EndProjectSection
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "WorkflowCore.Sample14", "src\samples\WorkflowCore.Sample14\WorkflowCore.Sample14.csproj", "{6BC66637-B42A-4334-ADFB-DBEC9F29D293}"

src/Directory.Build.props

@@ -4,10 +4,10 @@
     <PackageLicenseUrl>https://github.com/danielgerlag/workflow-core/blob/master/LICENSE.md</PackageLicenseUrl>
     <RepositoryType>git</RepositoryType>
     <RepositoryUrl>https://github.com/danielgerlag/workflow-core.git</RepositoryUrl>
-    <Version>3.5.7</Version>
-    <AssemblyVersion>3.5.7.0</AssemblyVersion>
-    <FileVersion>3.5.7.0</FileVersion>
+    <Version>3.6.0</Version>
+    <AssemblyVersion>3.6.0.0</AssemblyVersion>
+    <FileVersion>3.6.0.0</FileVersion>
     <PackageIconUrl>https://github.com/danielgerlag/workflow-core/raw/master/src/logo.png</PackageIconUrl>
-    <PackageVersion>3.5.7</PackageVersion>
+    <PackageVersion>3.6.0</PackageVersion>
   </PropertyGroup>
 </Project>

src/WorkflowCore/Interface/Persistence/IPersistenceProvider.cs

@@ -6,7 +6,7 @@
 
 namespace WorkflowCore.Interface
 {
-    public interface IPersistenceProvider : IWorkflowRepository, ISubscriptionRepository, IEventRepository
+    public interface IPersistenceProvider : IWorkflowRepository, ISubscriptionRepository, IEventRepository, IScheduledCommandRepository
     {        
 
         Task PersistErrors(IEnumerable<ExecutionError> errors, CancellationToken cancellationToken = default);

src/WorkflowCore/Interface/Persistence/IScheduledCommandRepository.cs

@@ -0,0 +1,18 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using WorkflowCore.Models;
+
+namespace WorkflowCore.Interface
+{
+    public interface IScheduledCommandRepository
+    {
+        bool SupportsScheduledCommands { get; }
+
+        Task ScheduleCommand(ScheduledCommand command);
+
+        Task ProcessCommands(DateTimeOffset asOf, Func<ScheduledCommand, Task> action, CancellationToken cancellationToken = default);
+    }
+}

src/WorkflowCore/Models/ScheduledCommand.cs

@@ -0,0 +1,16 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+
+namespace WorkflowCore.Models
+{
+    public class ScheduledCommand
+    {
+        public const string ProcessWorkflow = "ProcessWorkflow";
+        public const string ProcessEvent = "ProcessEvent";
+
+        public string CommandName { get; set; }
+        public string Data { get; set; }
+        public long ExecuteTime { get; set; }
+    }
+}

src/WorkflowCore/Services/BackgroundTasks/RunnablePoller.cs

@@ -1,6 +1,7 @@
 using System;
 using System.Linq;
 using System.Threading;
+using System.Threading.Tasks;
 using Microsoft.Extensions.Logging;
 using WorkflowCore.Interface;
 using WorkflowCore.Models;
@@ -48,17 +49,41 @@ public void Stop()
         /// Poll the persistence store for stashed unpublished events
         /// </summary>        
         private async void PollRunnables(object target)
+        {
+            await PollWorkflows();
+            await PollEvents();
+            await PollCommands();
+        }
+
+        private async Task PollWorkflows()
         {
             try
             {
                 if (await _lockProvider.AcquireLock("poll runnables", new CancellationToken()))
                 {
                     try
                     {
-                        _logger.LogInformation("Polling for runnable workflows");                        
+                        _logger.LogInformation("Polling for runnable workflows");
                         var runnables = await _persistenceStore.GetRunnableInstances(_dateTimeProvider.Now);
                         foreach (var item in runnables)
                         {
+                            if (_persistenceStore.SupportsScheduledCommands)
+                            {
+                                try
+                                {
+                                    await _persistenceStore.ScheduleCommand(new ScheduledCommand()
+                                    {
+                                        CommandName = ScheduledCommand.ProcessWorkflow,
+                                        Data = item,
+                                        ExecuteTime = _dateTimeProvider.UtcNow.Ticks
+                                    });
+                                    continue;
+                                }
+                                catch (Exception ex)
+                                {
+                                    _logger.LogError(ex, ex.Message);
+                                }
+                            }
                             if (_greylist.Contains($"wf:{item}"))
                             {
                                 _logger.LogDebug($"Got greylisted workflow {item}");
@@ -79,17 +104,37 @@ private async void PollRunnables(object target)
             {
                 _logger.LogError(ex, ex.Message);
             }
+        }
 
+        private async Task PollEvents()
+        {
             try
             {
                 if (await _lockProvider.AcquireLock("unprocessed events", new CancellationToken()))
                 {
                     try
                     {
-                        _logger.LogInformation("Polling for unprocessed events");                        
+                        _logger.LogInformation("Polling for unprocessed events");
                         var events = await _persistenceStore.GetRunnableEvents(_dateTimeProvider.Now);
                         foreach (var item in events.ToList())
                         {
+                            if (_persistenceStore.SupportsScheduledCommands)
+                            {
+                                try
+                                {
+                                    await _persistenceStore.ScheduleCommand(new ScheduledCommand()
+                                    {
+                                        CommandName = ScheduledCommand.ProcessEvent,
+                                        Data = item,
+                                        ExecuteTime = _dateTimeProvider.UtcNow.Ticks
+                                    });
+                                    continue;
+                                }
+                                catch (Exception ex)
+                                {
+                                    _logger.LogError(ex, ex.Message);
+                                }
+                            }
                             if (_greylist.Contains($"evt:{item}"))
                             {
                                 _logger.LogDebug($"Got greylisted event {item}");
@@ -111,5 +156,39 @@ private async void PollRunnables(object target)
                 _logger.LogError(ex, ex.Message);
             }
         }
+
+        private async Task PollCommands()
+        {
+            try
+            {
+                if (await _lockProvider.AcquireLock("poll-commands", new CancellationToken()))
+                {
+                    try
+                    {
+                        _logger.LogInformation("Polling for scheduled commands");
+                        await _persistenceStore.ProcessCommands(new DateTimeOffset(_dateTimeProvider.UtcNow), async (command) =>
+                        {
+                            switch (command.CommandName)
+                            {
+                                case ScheduledCommand.ProcessWorkflow:
+                                    await _queueProvider.QueueWork(command.Data, QueueType.Workflow);
+                                    break;
+                                case ScheduledCommand.ProcessEvent:
+                                    await _queueProvider.QueueWork(command.Data, QueueType.Event);
+                                    break;
+                            }
+                        });
+                    }
+                    finally
+                    {
+                        await _lockProvider.ReleaseLock("poll-commands");
+                    }
+                }
+            }
+            catch (Exception ex)
+            {
+                _logger.LogError(ex, ex.Message);
+            }
+        }
     }
 }

src/WorkflowCore/Services/BackgroundTasks/WorkflowConsumer.cs

@@ -69,11 +69,25 @@ protected override async Task ProcessItem(string itemId, CancellationToken cance
 
                     await _persistenceStore.PersistErrors(result.Errors, cancellationToken);
 
-                    var readAheadTicks = _datetimeProvider.UtcNow.Add(Options.PollInterval).Ticks;
-
-                    if ((workflow.Status == WorkflowStatus.Runnable) && workflow.NextExecution.HasValue && workflow.NextExecution.Value < readAheadTicks)
+                    if ((workflow.Status == WorkflowStatus.Runnable) && workflow.NextExecution.HasValue)
                     {
-                        new Task(() => FutureQueue(workflow, cancellationToken)).Start();
+                        var readAheadTicks = _datetimeProvider.UtcNow.Add(Options.PollInterval).Ticks;
+                        if (workflow.NextExecution.Value < readAheadTicks)
+                        {
+                            new Task(() => FutureQueue(workflow, cancellationToken)).Start();
+                        }
+                        else
+                        {
+                            if (_persistenceStore.SupportsScheduledCommands)
+                            {
+                                await _persistenceStore.ScheduleCommand(new ScheduledCommand()
+                                {
+                                    CommandName = ScheduledCommand.ProcessWorkflow,
+                                    Data = workflow.Id,
+                                    ExecuteTime = workflow.NextExecution.Value
+                                });
+                            }
+                        }
                     }
                 }
             }

src/WorkflowCore/Services/DefaultProviders/MemoryPersistenceProvider.cs

@@ -24,6 +24,8 @@ public class MemoryPersistenceProvider : ISingletonMemoryProvider
         private readonly List<Event> _events = new List<Event>();
         private readonly List<ExecutionError> _errors = new List<ExecutionError>();
 
+        public bool SupportsScheduledCommands => false;
+
         public async Task<string> CreateNewWorkflow(WorkflowInstance workflow, CancellationToken _ = default)
         {
             lock (_instances)
@@ -255,6 +257,16 @@ public async Task PersistErrors(IEnumerable<ExecutionError> errors, Cancellation
                 _errors.AddRange(errors);
             }
         }
+
+        public Task ScheduleCommand(ScheduledCommand command)
+        {
+            throw new NotImplementedException();
+        }
+
+        public Task ProcessCommands(DateTimeOffset asOf, Func<ScheduledCommand, Task> action, CancellationToken cancellationToken = default)
+        {
+            throw new NotImplementedException();
+        }
     }
 
 #pragma warning restore CS1998 // Async method lacks 'await' operators and will run synchronously

src/WorkflowCore/Services/DefaultProviders/TransientMemoryPersistenceProvider.cs

@@ -11,6 +11,8 @@ public class TransientMemoryPersistenceProvider : IPersistenceProvider
     {
         private readonly ISingletonMemoryProvider _innerService;
 
+        public bool SupportsScheduledCommands => false;
+
         public TransientMemoryPersistenceProvider(ISingletonMemoryProvider innerService)
         {
             _innerService = innerService;
@@ -56,5 +58,15 @@ public TransientMemoryPersistenceProvider(ISingletonMemoryProvider innerService)
         public Task<bool> SetSubscriptionToken(string eventSubscriptionId, string token, string workerId, DateTime expiry, CancellationToken _ = default) => _innerService.SetSubscriptionToken(eventSubscriptionId, token, workerId, expiry);
 
         public Task ClearSubscriptionToken(string eventSubscriptionId, string token, CancellationToken _ = default) => _innerService.ClearSubscriptionToken(eventSubscriptionId, token);
+
+        public Task ScheduleCommand(ScheduledCommand command)
+        {
+            throw new NotImplementedException();
+        }
+
+        public Task ProcessCommands(DateTimeOffset asOf, Func<ScheduledCommand, Task> action, CancellationToken cancellationToken = default)
+        {
+            throw new NotImplementedException();
+        }
     }
 }

src/WorkflowCore/WorkflowCore.csproj

@@ -26,6 +26,7 @@
     <PackageReference Include="System.Reflection.TypeExtensions" Version="4.7.0" />
     <PackageReference Include="System.Threading.Thread" Version="4.3.0" />
     <PackageReference Include="System.Linq.Queryable" Version="4.3.0" />
+	<InternalsVisibleTo Include="WorkflowCore.IntegrationTests" />
   </ItemGroup>
 
   <ItemGroup>

src/providers/WorkflowCore.Persistence.EntityFramework/ExtensionMethods.cs

@@ -123,6 +123,16 @@ internal static PersistedEvent ToPersistable(this Event instance)
             return result;
         }
 
+        internal static PersistedScheduledCommand ToPersistable(this ScheduledCommand instance)
+        {
+            var result = new PersistedScheduledCommand();
+            result.CommandName = instance.CommandName;
+            result.Data = instance.Data;
+            result.ExecuteTime = instance.ExecuteTime;
+
+            return result;
+        }
+
         internal static WorkflowInstance ToWorkflowInstance(this PersistedWorkflow instance)
         {
             WorkflowInstance result = new WorkflowInstance();
@@ -219,5 +229,15 @@ internal static Event ToEvent(this PersistedEvent instance)
 
             return result;
         }
+
+        internal static ScheduledCommand ToScheduledCommand(this PersistedScheduledCommand instance)
+        {
+            var result = new ScheduledCommand();
+            result.CommandName = instance.CommandName;
+            result.Data = instance.Data;
+            result.ExecuteTime = instance.ExecuteTime;
+
+            return result;
+        }
     }
 }

src/providers/WorkflowCore.Persistence.EntityFramework/Models/PersistedScheduledCommand.cs

@@ -0,0 +1,20 @@
+using System;
+using System.ComponentModel.DataAnnotations;
+using System.Linq;
+
+namespace WorkflowCore.Persistence.EntityFramework.Models
+{    
+    public class PersistedScheduledCommand
+    {
+        [Key]
+        public long PersistenceId { get; set; }
+
+        [MaxLength(200)]
+        public string CommandName { get; set; }
+
+        [MaxLength(500)]
+        public string Data { get; set; }
+
+        public long ExecuteTime { get; set; }
+    }
+}