Merge pull request #723 from DDzia/fix/issue-721

danielgerlag · web-flow · commit 89430ed53a65 · 2021-01-01T08:52:05.000-08:00
added DateTimeProvider to usages #issue-721
diff --git a/src/WorkflowCore/Services/ActivityController.cs b/src/WorkflowCore/Services/ActivityController.cs
@@ -28,10 +28,10 @@ public ActivityController(ISubscriptionRepository subscriptionRepository, IWorkf
         
         public async Task<PendingActivity> GetPendingActivity(string activityName, string workerId, TimeSpan? timeout = null)
         {
-            var endTime = DateTime.UtcNow.Add(timeout ?? TimeSpan.Zero);
+            var endTime = _dateTimeProvider.UtcNow.Add(timeout ?? TimeSpan.Zero);
             var firstPass = true;
             EventSubscription subscription = null;
-            while ((subscription == null && DateTime.UtcNow < endTime) || firstPass)
+            while ((subscription == null && _dateTimeProvider.UtcNow < endTime) || firstPass)
             {
                 if (!firstPass)
                     await Task.Delay(100);
diff --git a/src/WorkflowCore/Services/BackgroundTasks/RunnablePoller.cs b/src/WorkflowCore/Services/BackgroundTasks/RunnablePoller.cs
@@ -15,15 +15,17 @@ internal class RunnablePoller : IBackgroundTask
         private readonly ILogger _logger;
         private readonly IGreyList _greylist;
         private readonly WorkflowOptions _options;
+        private readonly IDateTimeProvider _dateTimeProvider;
         private Timer _pollTimer;
 
-        public RunnablePoller(IPersistenceProvider persistenceStore, IQueueProvider queueProvider, ILoggerFactory loggerFactory, IServiceProvider serviceProvider, IWorkflowRegistry registry, IDistributedLockProvider lockProvider, IGreyList greylist, WorkflowOptions options)
+        public RunnablePoller(IPersistenceProvider persistenceStore, IQueueProvider queueProvider, ILoggerFactory loggerFactory, IServiceProvider serviceProvider, IWorkflowRegistry registry, IDistributedLockProvider lockProvider, IGreyList greylist, IDateTimeProvider dateTimeProvider, WorkflowOptions options)
         {
             _persistenceStore = persistenceStore;
             _greylist = greylist;
             _queueProvider = queueProvider;            
             _logger = loggerFactory.CreateLogger<RunnablePoller>();
             _lockProvider = lockProvider;
+            _dateTimeProvider = dateTimeProvider;
             _options = options;
         }
 
@@ -54,7 +56,7 @@ private async void PollRunnables(object target)
                     try
                     {
                         _logger.LogInformation("Polling for runnable workflows");                        
-                        var runnables = await _persistenceStore.GetRunnableInstances(DateTime.Now);
+                        var runnables = await _persistenceStore.GetRunnableInstances(_dateTimeProvider.Now);
                         foreach (var item in runnables)
                         {
                             if (_greylist.Contains($"wf:{item}"))
@@ -85,7 +87,7 @@ private async void PollRunnables(object target)
                     try
                     {
                         _logger.LogInformation("Polling for unprocessed events");                        
-                        var events = await _persistenceStore.GetRunnableEvents(DateTime.Now);
+                        var events = await _persistenceStore.GetRunnableEvents(_dateTimeProvider.Now);
                         foreach (var item in events.ToList())
                         {
                             if (_greylist.Contains($"evt:{item}"))
diff --git a/src/WorkflowCore/Services/CancellationProcessor.cs b/src/WorkflowCore/Services/CancellationProcessor.cs
@@ -12,11 +12,13 @@ public class CancellationProcessor : ICancellationProcessor
     {
         protected readonly ILogger _logger;
         private readonly IExecutionResultProcessor _executionResultProcessor;
+        private readonly IDateTimeProvider _dateTimeProvider;
 
-        public CancellationProcessor(IExecutionResultProcessor executionResultProcessor, ILoggerFactory logFactory)
+        public CancellationProcessor(IExecutionResultProcessor executionResultProcessor, ILoggerFactory logFactory, IDateTimeProvider dateTimeProvider)
         {
             _executionResultProcessor = executionResultProcessor;
             _logger = logFactory.CreateLogger<CancellationProcessor>();
+            _dateTimeProvider = dateTimeProvider;
         }
 
         public void ProcessCancellations(WorkflowInstance workflow, WorkflowDefinition workflowDef, WorkflowExecutorResult executionResult)
@@ -44,13 +46,13 @@ public void ProcessCancellations(WorkflowInstance workflow, WorkflowDefinition w
                             _executionResultProcessor.ProcessExecutionResult(workflow, workflowDef, ptr, step, ExecutionResult.Next(), executionResult);
                         }
 
-                        ptr.EndTime = DateTime.Now.ToUniversalTime();
+                        ptr.EndTime = _dateTimeProvider.UtcNow;
                         ptr.Active = false;
                         ptr.Status = PointerStatus.Cancelled;
 
                         foreach (var descendent in workflow.ExecutionPointers.FindByScope(ptr.Id).Where(x => x.Status != PointerStatus.Complete && x.Status != PointerStatus.Cancelled))
                         {
-                            descendent.EndTime = DateTime.Now.ToUniversalTime();
+                            descendent.EndTime = _dateTimeProvider.UtcNow;
                             descendent.Active = false;
                             descendent.Status = PointerStatus.Cancelled;
                         }
diff --git a/src/WorkflowCore/Services/GreyList.cs b/src/WorkflowCore/Services/GreyList.cs
@@ -11,27 +11,29 @@ public class GreyList : IGreyList, IDisposable
         private readonly Timer _cycleTimer;
         private readonly ConcurrentDictionary<string, DateTime> _list;
         private readonly ILogger _logger;
+        private readonly IDateTimeProvider _dateTimeProvider;
         private const int CYCLE_TIME = 30;
         private const int TTL = 5;
 
-        public GreyList(ILoggerFactory loggerFactory)
+        public GreyList(ILoggerFactory loggerFactory, IDateTimeProvider dateTimeProvider)
         {
             _logger = loggerFactory.CreateLogger<GreyList>();
+            _dateTimeProvider = dateTimeProvider;
             _list = new ConcurrentDictionary<string, DateTime>();
             _cycleTimer = new Timer(new TimerCallback(Cycle), null, TimeSpan.FromMinutes(CYCLE_TIME), TimeSpan.FromMinutes(CYCLE_TIME));
         }
 
         public void Add(string id)
         {
-            _list.AddOrUpdate(id, DateTime.Now, (key, val) => DateTime.Now);
+            _list.AddOrUpdate(id, _dateTimeProvider.Now, (key, val) => _dateTimeProvider.Now);
         }
 
         public bool Contains(string id)
         {
             if (!_list.TryGetValue(id, out var start))
                 return false;
 
-            var result = start > (DateTime.Now.AddMinutes(-1 * TTL));
+            var result = start > (_dateTimeProvider.Now.AddMinutes(-1 * TTL));
 
             if (!result)
                 _list.TryRemove(id, out var _);
diff --git a/src/WorkflowCore/Services/SyncWorkflowRunner.cs b/src/WorkflowCore/Services/SyncWorkflowRunner.cs
@@ -18,8 +18,9 @@ public class SyncWorkflowRunner : ISyncWorkflowRunner
         private readonly IPersistenceProvider _persistenceStore;
         private readonly IExecutionPointerFactory _pointerFactory;
         private readonly IQueueProvider _queueService;
+        private readonly IDateTimeProvider _dateTimeProvider;
 
-        public SyncWorkflowRunner(IWorkflowHost host, IWorkflowExecutor executor, IDistributedLockProvider lockService, IWorkflowRegistry registry, IPersistenceProvider persistenceStore, IExecutionPointerFactory pointerFactory, IQueueProvider queueService)
+        public SyncWorkflowRunner(IWorkflowHost host, IWorkflowExecutor executor, IDistributedLockProvider lockService, IWorkflowRegistry registry, IPersistenceProvider persistenceStore, IExecutionPointerFactory pointerFactory, IQueueProvider queueService, IDateTimeProvider dateTimeProvider)
         {
             _host = host;
             _executor = executor;
@@ -28,6 +29,7 @@ public SyncWorkflowRunner(IWorkflowHost host, IWorkflowExecutor executor, IDistr
             _persistenceStore = persistenceStore;
             _pointerFactory = pointerFactory;
             _queueService = queueService;
+            _dateTimeProvider = dateTimeProvider;
         }
 
         public async Task<WorkflowInstance> RunWorkflowSync<TData>(string workflowId, int version, TData data, string reference, TimeSpan timeOut, bool persistSate = true)
@@ -46,7 +48,7 @@ public async Task<WorkflowInstance> RunWorkflowSync<TData>(string workflowId, in
                 Data = data,
                 Description = def.Description,
                 NextExecution = 0,
-                CreateTime = DateTime.Now.ToUniversalTime(),
+                CreateTime = _dateTimeProvider.UtcNow,
                 Status = WorkflowStatus.Suspended,
                 Reference = reference
             };
diff --git a/src/WorkflowCore/Services/WorkflowController.cs b/src/WorkflowCore/Services/WorkflowController.cs
@@ -22,8 +22,9 @@ public class WorkflowController : IWorkflowController
         private readonly ILifeCycleEventHub _eventHub;
         private readonly IServiceProvider _serviceProvider;
         private readonly ILogger _logger;
+        private readonly IDateTimeProvider _dateTimeProvider;
 
-        public WorkflowController(IPersistenceProvider persistenceStore, IDistributedLockProvider lockProvider, IWorkflowRegistry registry, IQueueProvider queueProvider, IExecutionPointerFactory pointerFactory, ILifeCycleEventHub eventHub, ILoggerFactory loggerFactory, IServiceProvider serviceProvider)
+        public WorkflowController(IPersistenceProvider persistenceStore, IDistributedLockProvider lockProvider, IWorkflowRegistry registry, IQueueProvider queueProvider, IExecutionPointerFactory pointerFactory, ILifeCycleEventHub eventHub, ILoggerFactory loggerFactory, IServiceProvider serviceProvider, IDateTimeProvider dateTimeProvider)
         {
             _persistenceStore = persistenceStore;
             _lockProvider = lockProvider;
@@ -33,6 +34,7 @@ public WorkflowController(IPersistenceProvider persistenceStore, IDistributedLoc
             _eventHub = eventHub;
             _serviceProvider = serviceProvider;
             _logger = loggerFactory.CreateLogger<WorkflowController>();
+            _dateTimeProvider = dateTimeProvider;
         }
 
         public Task<string> StartWorkflow(string workflowId, object data = null, string reference=null)
@@ -68,7 +70,7 @@ public async Task<string> StartWorkflow<TData>(string workflowId, int? version,
                 Data = data,
                 Description = def.Description,
                 NextExecution = 0,
-                CreateTime = DateTime.Now.ToUniversalTime(),
+                CreateTime = _dateTimeProvider.UtcNow,
                 Status = WorkflowStatus.Runnable,
                 Reference = reference
             };
@@ -94,7 +96,7 @@ public async Task<string> StartWorkflow<TData>(string workflowId, int? version,
             await _queueProvider.QueueWork(id, QueueType.Index);
             await _eventHub.PublishNotification(new WorkflowStarted()
             {
-                EventTimeUtc = DateTime.UtcNow,
+                EventTimeUtc = _dateTimeProvider.UtcNow,
                 Reference = reference,
                 WorkflowInstanceId = id,
                 WorkflowDefinitionId = def.Id,
@@ -111,7 +113,7 @@ public async Task PublishEvent(string eventName, string eventKey, object eventDa
             if (effectiveDate.HasValue)
                 evt.EventTime = effectiveDate.Value.ToUniversalTime();
             else
-                evt.EventTime = DateTime.Now.ToUniversalTime();
+                evt.EventTime = _dateTimeProvider.UtcNow;
 
             evt.EventData = eventData;
             evt.EventKey = eventKey;
@@ -137,7 +139,7 @@ public async Task<bool> SuspendWorkflow(string workflowId)
                     await _queueProvider.QueueWork(workflowId, QueueType.Index);
                     await _eventHub.PublishNotification(new WorkflowSuspended()
                     {
-                        EventTimeUtc = DateTime.UtcNow,
+                        EventTimeUtc = _dateTimeProvider.UtcNow,
                         Reference = wf.Reference,
                         WorkflowInstanceId = wf.Id,
                         WorkflowDefinitionId = wf.WorkflowDefinitionId,
@@ -173,7 +175,7 @@ public async Task<bool> ResumeWorkflow(string workflowId)
                     await _queueProvider.QueueWork(workflowId, QueueType.Index);
                     await _eventHub.PublishNotification(new WorkflowResumed()
                     {
-                        EventTimeUtc = DateTime.UtcNow,
+                        EventTimeUtc = _dateTimeProvider.UtcNow,
                         Reference = wf.Reference,
                         WorkflowInstanceId = wf.Id,
                         WorkflowDefinitionId = wf.WorkflowDefinitionId,
@@ -207,7 +209,7 @@ public async Task<bool> TerminateWorkflow(string workflowId)
                 await _queueProvider.QueueWork(workflowId, QueueType.Index);
                 await _eventHub.PublishNotification(new WorkflowTerminated()
                 {
-                    EventTimeUtc = DateTime.UtcNow,
+                    EventTimeUtc = _dateTimeProvider.UtcNow,
                     Reference = wf.Reference,
                     WorkflowInstanceId = wf.Id,
                     WorkflowDefinitionId = wf.WorkflowDefinitionId,
diff --git a/src/providers/WorkflowCore.Providers.AWS/ServiceCollectionExtensions.cs b/src/providers/WorkflowCore.Providers.AWS/ServiceCollectionExtensions.cs
@@ -21,7 +21,7 @@ public static WorkflowOptions UseAwsSimpleQueueService(this WorkflowOptions opti
 
         public static WorkflowOptions UseAwsDynamoLocking(this WorkflowOptions options, AWSCredentials credentials, AmazonDynamoDBConfig config, string tableName)
         {
-            options.UseDistributedLockManager(sp => new DynamoLockProvider(credentials, config, tableName, sp.GetService<ILoggerFactory>()));
+            options.UseDistributedLockManager(sp => new DynamoLockProvider(credentials, config, tableName, sp.GetService<ILoggerFactory>(), sp.GetService<IDateTimeProvider>()));
             return options;
         }
 
@@ -35,7 +35,7 @@ public static WorkflowOptions UseAwsDynamoPersistence(this WorkflowOptions optio
         public static WorkflowOptions UseAwsKinesis(this WorkflowOptions options, AWSCredentials credentials, RegionEndpoint region, string appName, string streamName)
         {
             options.Services.AddTransient<IKinesisTracker>(sp => new KinesisTracker(credentials, region, "workflowcore_kinesis", sp.GetService<ILoggerFactory>()));
-            options.Services.AddTransient<IKinesisStreamConsumer>(sp => new KinesisStreamConsumer(credentials, region, sp.GetService<IKinesisTracker>(), sp.GetService<IDistributedLockProvider>(), sp.GetService<ILoggerFactory>()));
+            options.Services.AddTransient<IKinesisStreamConsumer>(sp => new KinesisStreamConsumer(credentials, region, sp.GetService<IKinesisTracker>(), sp.GetService<IDistributedLockProvider>(), sp.GetService<ILoggerFactory>(), sp.GetService<IDateTimeProvider>()));
             options.UseEventHub(sp => new KinesisProvider(credentials, region, appName, streamName, sp.GetService<IKinesisStreamConsumer>(), sp.GetService<ILoggerFactory>()));
             return options;
         }
diff --git a/src/providers/WorkflowCore.Providers.AWS/Services/DynamoLockProvider.cs b/src/providers/WorkflowCore.Providers.AWS/Services/DynamoLockProvider.cs
@@ -23,14 +23,16 @@ public class DynamoLockProvider : IDistributedLockProvider
         private Task _heartbeatTask;
         private CancellationTokenSource _cancellationTokenSource;
         private readonly AutoResetEvent _mutex = new AutoResetEvent(true);
+        private readonly IDateTimeProvider _dateTimeProvider;
 
-        public DynamoLockProvider(AWSCredentials credentials, AmazonDynamoDBConfig config, string tableName, ILoggerFactory logFactory)
+        public DynamoLockProvider(AWSCredentials credentials, AmazonDynamoDBConfig config, string tableName, ILoggerFactory logFactory, IDateTimeProvider dateTimeProvider)
         {
             _logger = logFactory.CreateLogger<DynamoLockProvider>();
             _client = new AmazonDynamoDBClient(credentials, config);
             _localLocks = new List<string>();
             _tableName = tableName;
             _nodeId = Guid.NewGuid().ToString();
+            _dateTimeProvider = dateTimeProvider;
         }
 
         public async Task<bool> AcquireLock(string Id, CancellationToken cancellationToken)
@@ -46,7 +48,7 @@ public async Task<bool> AcquireLock(string Id, CancellationToken cancellationTok
                         { "lock_owner", new AttributeValue(_nodeId) },
                         { "expires", new AttributeValue()
                             {
-                                N = Convert.ToString(new DateTimeOffset(DateTime.UtcNow).ToUnixTimeMilliseconds() + _ttl)
+                                N = Convert.ToString(new DateTimeOffset(_dateTimeProvider.UtcNow).ToUnixTimeMilliseconds() + _ttl)
                             }
                         }
                     },
@@ -55,7 +57,7 @@ public async Task<bool> AcquireLock(string Id, CancellationToken cancellationTok
                     {
                         { ":expired", new AttributeValue()
                             {
-                                N = Convert.ToString(new DateTimeOffset(DateTime.UtcNow).ToUnixTimeMilliseconds() + _jitter)
+                                N = Convert.ToString(new DateTimeOffset(_dateTimeProvider.UtcNow).ToUnixTimeMilliseconds() + _jitter)
                             }
                         }
                     }
@@ -154,7 +156,7 @@ private async void SendHeartbeat()
                                         { "lock_owner", new AttributeValue(_nodeId) },
                                         { "expires", new AttributeValue()
                                             {
-                                                N = Convert.ToString(new DateTimeOffset(DateTime.UtcNow).ToUnixTimeMilliseconds() + _ttl)
+                                                N = Convert.ToString(new DateTimeOffset(_dateTimeProvider.UtcNow).ToUnixTimeMilliseconds() + _ttl)
                                             }
                                         }
                                     },
diff --git a/src/providers/WorkflowCore.Providers.AWS/Services/KinesisStreamConsumer.cs b/src/providers/WorkflowCore.Providers.AWS/Services/KinesisStreamConsumer.cs
@@ -23,15 +23,17 @@ public class KinesisStreamConsumer : IKinesisStreamConsumer, IDisposable
         private readonly Task _processTask;
         private readonly int _batchSize = 100;
         private ICollection<ShardSubscription> _subscribers = new HashSet<ShardSubscription>();
+        private readonly IDateTimeProvider _dateTimeProvider;
         
-        public KinesisStreamConsumer(AWSCredentials credentials, RegionEndpoint region, IKinesisTracker tracker, IDistributedLockProvider lockManager, ILoggerFactory logFactory)
+        public KinesisStreamConsumer(AWSCredentials credentials, RegionEndpoint region, IKinesisTracker tracker, IDistributedLockProvider lockManager, ILoggerFactory logFactory, IDateTimeProvider dateTimeProvider)
         {
             _logger = logFactory.CreateLogger(GetType());
             _tracker = tracker;
             _lockManager = lockManager;
             _client = new AmazonKinesisClient(credentials, region);
             _processTask = new Task(Process);
             _processTask.Start();
+            _dateTimeProvider = dateTimeProvider;
         }
 
         public async Task Subscribe(string appName, string stream, Action<Record> action)
@@ -59,7 +61,7 @@ private async void Process()
             {
                 try
                 {
-                    var todo = _subscribers.Where(x => x.Snooze < DateTime.Now).ToList();
+                    var todo = _subscribers.Where(x => x.Snooze < _dateTimeProvider.Now).ToList();
                     foreach (var sub in todo)
                     {
                         if (!await _lockManager.AcquireLock($"{sub.AppName}.{sub.Stream}.{sub.Shard.ShardId}",
@@ -71,7 +73,7 @@ private async void Process()
                             var records = await GetBatch(sub);
 
                             if (records.Records.Count == 0)
-                                sub.Snooze = DateTime.Now.AddSeconds(5);
+                                sub.Snooze = _dateTimeProvider.Now.AddSeconds(5);
 
                             var lastSequence = string.Empty;
 
diff --git a/src/providers/WorkflowCore.Providers.Redis/Services/RedisPersistenceProvider.cs b/src/providers/WorkflowCore.Providers.Redis/Services/RedisPersistenceProvider.cs
@@ -62,7 +62,7 @@ public async Task PersistWorkflow(WorkflowInstance workflow)
         public async Task<IEnumerable<string>> GetRunnableInstances(DateTime asAt)
         {
             var result = new List<string>();
-            var data = await _redis.SortedSetRangeByScoreAsync($"{_prefix}.{WORKFLOW_SET}.{RUNNABLE_INDEX}", -1, DateTime.UtcNow.Ticks);
+            var data = await _redis.SortedSetRangeByScoreAsync($"{_prefix}.{WORKFLOW_SET}.{RUNNABLE_INDEX}", -1, asAt.ToUniversalTime().Ticks);
 
             foreach (var item in data)
                 result.Add(item);

Original file line number	Diff line number	Diff line change
`@@ -28,10 +28,10 @@ public ActivityController(ISubscriptionRepository subscriptionRepository, IWorkf`
`28`	`28`
`29`	`29`	`public async Task<PendingActivity> GetPendingActivity(string activityName, string workerId, TimeSpan? timeout = null)`
`30`	`30`	`{`
`31`		`- var endTime = DateTime.UtcNow.Add(timeout ?? TimeSpan.Zero);`
	`31`	`+ var endTime = _dateTimeProvider.UtcNow.Add(timeout ?? TimeSpan.Zero);`
`32`	`32`	`var firstPass = true;`
`33`	`33`	`EventSubscription subscription = null;`
`34`		`- while ((subscription == null && DateTime.UtcNow < endTime) \|\| firstPass)`
	`34`	`+ while ((subscription == null && _dateTimeProvider.UtcNow < endTime) \|\| firstPass)`
`35`	`35`	`{`
`36`	`36`	`if (!firstPass)`
`37`	`37`	`await Task.Delay(100);`
Original file line number	Diff line number	Diff line change
`@@ -15,15 +15,17 @@ internal class RunnablePoller : IBackgroundTask`
`15`	`15`	`private readonly ILogger _logger;`
`16`	`16`	`private readonly IGreyList _greylist;`
`17`	`17`	`private readonly WorkflowOptions _options;`
	`18`	`+ private readonly IDateTimeProvider _dateTimeProvider;`
`18`	`19`	`private Timer _pollTimer;`
`19`	`20`
`20`		`- public RunnablePoller(IPersistenceProvider persistenceStore, IQueueProvider queueProvider, ILoggerFactory loggerFactory, IServiceProvider serviceProvider, IWorkflowRegistry registry, IDistributedLockProvider lockProvider, IGreyList greylist, WorkflowOptions options)`
	`21`	`+ public RunnablePoller(IPersistenceProvider persistenceStore, IQueueProvider queueProvider, ILoggerFactory loggerFactory, IServiceProvider serviceProvider, IWorkflowRegistry registry, IDistributedLockProvider lockProvider, IGreyList greylist, IDateTimeProvider dateTimeProvider, WorkflowOptions options)`
`21`	`22`	`{`
`22`	`23`	`_persistenceStore = persistenceStore;`
`23`	`24`	`_greylist = greylist;`
`24`	`25`	`_queueProvider = queueProvider;`
`25`	`26`	`_logger = loggerFactory.CreateLogger<RunnablePoller>();`
`26`	`27`	`_lockProvider = lockProvider;`
	`28`	`+ _dateTimeProvider = dateTimeProvider;`
`27`	`29`	`_options = options;`
`28`	`30`	`}`
`29`	`31`
`@@ -54,7 +56,7 @@ private async void PollRunnables(object target)`
`54`	`56`	`try`
`55`	`57`	`{`
`56`	`58`	`_logger.LogInformation("Polling for runnable workflows");`
`57`		`- var runnables = await _persistenceStore.GetRunnableInstances(DateTime.Now);`
	`59`	`+ var runnables = await _persistenceStore.GetRunnableInstances(_dateTimeProvider.Now);`
`58`	`60`	`foreach (var item in runnables)`
`59`	`61`	`{`
`60`	`62`	`if (_greylist.Contains($"wf:{item}"))`
`@@ -85,7 +87,7 @@ private async void PollRunnables(object target)`
`85`	`87`	`try`
`86`	`88`	`{`
`87`	`89`	`_logger.LogInformation("Polling for unprocessed events");`
`88`		`- var events = await _persistenceStore.GetRunnableEvents(DateTime.Now);`
	`90`	`+ var events = await _persistenceStore.GetRunnableEvents(_dateTimeProvider.Now);`
`89`	`91`	`foreach (var item in events.ToList())`
`90`	`92`	`{`
`91`	`93`	`if (_greylist.Contains($"evt:{item}"))`
Original file line number	Diff line number	Diff line change
`@@ -12,11 +12,13 @@ public class CancellationProcessor : ICancellationProcessor`
`12`	`12`	`{`
`13`	`13`	`protected readonly ILogger _logger;`
`14`	`14`	`private readonly IExecutionResultProcessor _executionResultProcessor;`
	`15`	`+ private readonly IDateTimeProvider _dateTimeProvider;`
`15`	`16`
`16`		`- public CancellationProcessor(IExecutionResultProcessor executionResultProcessor, ILoggerFactory logFactory)`
	`17`	`+ public CancellationProcessor(IExecutionResultProcessor executionResultProcessor, ILoggerFactory logFactory, IDateTimeProvider dateTimeProvider)`
`17`	`18`	`{`
`18`	`19`	`_executionResultProcessor = executionResultProcessor;`
`19`	`20`	`_logger = logFactory.CreateLogger<CancellationProcessor>();`
	`21`	`+ _dateTimeProvider = dateTimeProvider;`
`20`	`22`	`}`
`21`	`23`
`22`	`24`	`public void ProcessCancellations(WorkflowInstance workflow, WorkflowDefinition workflowDef, WorkflowExecutorResult executionResult)`
`@@ -44,13 +46,13 @@ public void ProcessCancellations(WorkflowInstance workflow, WorkflowDefinition w`
`44`	`46`	`_executionResultProcessor.ProcessExecutionResult(workflow, workflowDef, ptr, step, ExecutionResult.Next(), executionResult);`
`45`	`47`	`}`
`46`	`48`
`47`		`- ptr.EndTime = DateTime.Now.ToUniversalTime();`
	`49`	`+ ptr.EndTime = _dateTimeProvider.UtcNow;`
`48`	`50`	`ptr.Active = false;`
`49`	`51`	`ptr.Status = PointerStatus.Cancelled;`
`50`	`52`
`51`	`53`	`foreach (var descendent in workflow.ExecutionPointers.FindByScope(ptr.Id).Where(x => x.Status != PointerStatus.Complete && x.Status != PointerStatus.Cancelled))`
`52`	`54`	`{`
`53`		`- descendent.EndTime = DateTime.Now.ToUniversalTime();`
	`55`	`+ descendent.EndTime = _dateTimeProvider.UtcNow;`
`54`	`56`	`descendent.Active = false;`
`55`	`57`	`descendent.Status = PointerStatus.Cancelled;`
`56`	`58`	`}`
Original file line number	Diff line number	Diff line change
`@@ -11,27 +11,29 @@ public class GreyList : IGreyList, IDisposable`
`11`	`11`	`private readonly Timer _cycleTimer;`
`12`	`12`	`private readonly ConcurrentDictionary<string, DateTime> _list;`
`13`	`13`	`private readonly ILogger _logger;`
	`14`	`+ private readonly IDateTimeProvider _dateTimeProvider;`
`14`	`15`	`private const int CYCLE_TIME = 30;`
`15`	`16`	`private const int TTL = 5;`
`16`	`17`
`17`		`- public GreyList(ILoggerFactory loggerFactory)`
	`18`	`+ public GreyList(ILoggerFactory loggerFactory, IDateTimeProvider dateTimeProvider)`
`18`	`19`	`{`
`19`	`20`	`_logger = loggerFactory.CreateLogger<GreyList>();`
	`21`	`+ _dateTimeProvider = dateTimeProvider;`
`20`	`22`	`_list = new ConcurrentDictionary<string, DateTime>();`
`21`	`23`	`_cycleTimer = new Timer(new TimerCallback(Cycle), null, TimeSpan.FromMinutes(CYCLE_TIME), TimeSpan.FromMinutes(CYCLE_TIME));`
`22`	`24`	`}`
`23`	`25`
`24`	`26`	`public void Add(string id)`
`25`	`27`	`{`
`26`		`- _list.AddOrUpdate(id, DateTime.Now, (key, val) => DateTime.Now);`
	`28`	`+ _list.AddOrUpdate(id, _dateTimeProvider.Now, (key, val) => _dateTimeProvider.Now);`
`27`	`29`	`}`
`28`	`30`
`29`	`31`	`public bool Contains(string id)`
`30`	`32`	`{`
`31`	`33`	`if (!_list.TryGetValue(id, out var start))`
`32`	`34`	`return false;`
`33`	`35`
`34`		`- var result = start > (DateTime.Now.AddMinutes(-1 * TTL));`
	`36`	`+ var result = start > (_dateTimeProvider.Now.AddMinutes(-1 * TTL));`
`35`	`37`
`36`	`38`	`if (!result)`
`37`	`39`	`_list.TryRemove(id, out var _);`
Original file line number	Diff line number	Diff line change
`@@ -21,7 +21,7 @@ public static WorkflowOptions UseAwsSimpleQueueService(this WorkflowOptions opti`
`21`	`21`
`22`	`22`	`public static WorkflowOptions UseAwsDynamoLocking(this WorkflowOptions options, AWSCredentials credentials, AmazonDynamoDBConfig config, string tableName)`
`23`	`23`	`{`
`24`		`- options.UseDistributedLockManager(sp => new DynamoLockProvider(credentials, config, tableName, sp.GetService<ILoggerFactory>()));`
	`24`	`+ options.UseDistributedLockManager(sp => new DynamoLockProvider(credentials, config, tableName, sp.GetService<ILoggerFactory>(), sp.GetService<IDateTimeProvider>()));`
`25`	`25`	`return options;`
`26`	`26`	`}`
`27`	`27`
`@@ -35,7 +35,7 @@ public static WorkflowOptions UseAwsDynamoPersistence(this WorkflowOptions optio`
`35`	`35`	`public static WorkflowOptions UseAwsKinesis(this WorkflowOptions options, AWSCredentials credentials, RegionEndpoint region, string appName, string streamName)`
`36`	`36`	`{`
`37`	`37`	`options.Services.AddTransient<IKinesisTracker>(sp => new KinesisTracker(credentials, region, "workflowcore_kinesis", sp.GetService<ILoggerFactory>()));`
`38`		`- options.Services.AddTransient<IKinesisStreamConsumer>(sp => new KinesisStreamConsumer(credentials, region, sp.GetService<IKinesisTracker>(), sp.GetService<IDistributedLockProvider>(), sp.GetService<ILoggerFactory>()));`
	`38`	`+ options.Services.AddTransient<IKinesisStreamConsumer>(sp => new KinesisStreamConsumer(credentials, region, sp.GetService<IKinesisTracker>(), sp.GetService<IDistributedLockProvider>(), sp.GetService<ILoggerFactory>(), sp.GetService<IDateTimeProvider>()));`
`39`	`39`	`options.UseEventHub(sp => new KinesisProvider(credentials, region, appName, streamName, sp.GetService<IKinesisStreamConsumer>(), sp.GetService<ILoggerFactory>()));`
`40`	`40`	`return options;`
`41`	`41`	`}`
Original file line number	Diff line number	Diff line change
`@@ -23,14 +23,16 @@ public class DynamoLockProvider : IDistributedLockProvider`
`23`	`23`	`private Task _heartbeatTask;`
`24`	`24`	`private CancellationTokenSource _cancellationTokenSource;`
`25`	`25`	`private readonly AutoResetEvent _mutex = new AutoResetEvent(true);`
	`26`	`+ private readonly IDateTimeProvider _dateTimeProvider;`
`26`	`27`
`27`		`- public DynamoLockProvider(AWSCredentials credentials, AmazonDynamoDBConfig config, string tableName, ILoggerFactory logFactory)`
	`28`	`+ public DynamoLockProvider(AWSCredentials credentials, AmazonDynamoDBConfig config, string tableName, ILoggerFactory logFactory, IDateTimeProvider dateTimeProvider)`
`28`	`29`	`{`
`29`	`30`	`_logger = logFactory.CreateLogger<DynamoLockProvider>();`
`30`	`31`	`_client = new AmazonDynamoDBClient(credentials, config);`
`31`	`32`	`_localLocks = new List<string>();`
`32`	`33`	`_tableName = tableName;`
`33`	`34`	`_nodeId = Guid.NewGuid().ToString();`
	`35`	`+ _dateTimeProvider = dateTimeProvider;`
`34`	`36`	`}`
`35`	`37`
`36`	`38`	`public async Task<bool> AcquireLock(string Id, CancellationToken cancellationToken)`
`@@ -46,7 +48,7 @@ public async Task<bool> AcquireLock(string Id, CancellationToken cancellationTok`
`46`	`48`	`{ "lock_owner", new AttributeValue(_nodeId) },`
`47`	`49`	`{ "expires", new AttributeValue()`
`48`	`50`	`{`
`49`		`- N = Convert.ToString(new DateTimeOffset(DateTime.UtcNow).ToUnixTimeMilliseconds() + _ttl)`
	`51`	`+ N = Convert.ToString(new DateTimeOffset(_dateTimeProvider.UtcNow).ToUnixTimeMilliseconds() + _ttl)`
`50`	`52`	`}`
`51`	`53`	`}`
`52`	`54`	`},`
`@@ -55,7 +57,7 @@ public async Task<bool> AcquireLock(string Id, CancellationToken cancellationTok`
`55`	`57`	`{`
`56`	`58`	`{ ":expired", new AttributeValue()`
`57`	`59`	`{`
`58`		`- N = Convert.ToString(new DateTimeOffset(DateTime.UtcNow).ToUnixTimeMilliseconds() + _jitter)`
	`60`	`+ N = Convert.ToString(new DateTimeOffset(_dateTimeProvider.UtcNow).ToUnixTimeMilliseconds() + _jitter)`
`59`	`61`	`}`
`60`	`62`	`}`
`61`	`63`	`}`
`@@ -154,7 +156,7 @@ private async void SendHeartbeat()`
`154`	`156`	`{ "lock_owner", new AttributeValue(_nodeId) },`
`155`	`157`	`{ "expires", new AttributeValue()`
`156`	`158`	`{`
`157`		`- N = Convert.ToString(new DateTimeOffset(DateTime.UtcNow).ToUnixTimeMilliseconds() + _ttl)`
	`159`	`+ N = Convert.ToString(new DateTimeOffset(_dateTimeProvider.UtcNow).ToUnixTimeMilliseconds() + _ttl)`
`158`	`160`	`}`
`159`	`161`	`}`
`160`	`162`	`},`
Original file line number	Diff line number	Diff line change
`@@ -62,7 +62,7 @@ public async Task PersistWorkflow(WorkflowInstance workflow)`
`62`	`62`	`public async Task<IEnumerable<string>> GetRunnableInstances(DateTime asAt)`
`63`	`63`	`{`
`64`	`64`	`var result = new List<string>();`
`65`		`- var data = await _redis.SortedSetRangeByScoreAsync($"{_prefix}.{WORKFLOW_SET}.{RUNNABLE_INDEX}", -1, DateTime.UtcNow.Ticks);`
	`65`	`+ var data = await _redis.SortedSetRangeByScoreAsync($"{_prefix}.{WORKFLOW_SET}.{RUNNABLE_INDEX}", -1, asAt.ToUniversalTime().Ticks);`
`66`	`66`
`67`	`67`	`foreach (var item in data)`
`68`	`68`	`result.Add(item);`