Merge pull request #1128 from eventflow/handfire-bug

Fix: Scheduling bug in Hangfire

Author: rasmus

Source/EventFlow.Hangfire.Tests/Integration/HangfireJobSchedulerTests.cs

@@ -22,6 +22,7 @@
 
 using System;
 using System.Collections.Concurrent;
+using System.Diagnostics;
 using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
@@ -124,40 +125,55 @@ public async Task ScheduleNow()
         [Test]
         public async Task ScheduleAsyncWithDateTime()
         {
-            await ValidateScheduleHappens((j, s) => s.ScheduleAsync(j, DateTimeOffset.Now.AddSeconds(1), CancellationToken.None)).ConfigureAwait(false);
+            var minimumDelay = TimeSpan.FromSeconds(0.75);
+            var runAt = DateTimeOffset.UtcNow.AddSeconds(1);
+            await ValidateScheduleHappens(
+                (j, s) => s.ScheduleAsync(j, runAt, CancellationToken.None),
+                minimumDelay).ConfigureAwait(false);
         }
 
         [Test]
         public async Task ScheduleAsyncWithTimeSpan()
         {
-            await ValidateScheduleHappens((j, s) => s.ScheduleAsync(j, TimeSpan.FromSeconds(1), CancellationToken.None)).ConfigureAwait(false);
+            var minimumDelay = TimeSpan.FromSeconds(0.75);
+            await ValidateScheduleHappens(
+                (j, s) => s.ScheduleAsync(j, TimeSpan.FromSeconds(1), CancellationToken.None),
+                minimumDelay).ConfigureAwait(false);
         }
 
-        private async Task ValidateScheduleHappens(Func<IJob, IJobScheduler, Task<IJobId>> schedule)
+        private async Task ValidateScheduleHappens(Func<IJob, IJobScheduler, Task<IJobId>> schedule, TimeSpan? minimumElapsed = null)
         {
             // Arrange
             var testId = ThingyId.New;
             var pingId = PingId.New;
             var executeCommandJob = PublishCommandJob.Create(new ThingyPingCommand(testId, pingId), ServiceProvider);
 
             // Act
+            // Use a monotonic timer to avoid false negatives from wall-clock drift in CI.
+            var stopwatch = Stopwatch.StartNew();
             var jobId = await schedule(executeCommandJob, _jobScheduler).ConfigureAwait(false);
 
             // Assert
-            var start = DateTimeOffset.Now;
-            while (DateTimeOffset.Now < start + TimeSpan.FromSeconds(10))
+            while (stopwatch.Elapsed < TimeSpan.FromSeconds(10))
             {
                 var testAggregate = await AggregateStore.LoadAsync<ThingyAggregate, ThingyId>(testId, CancellationToken.None).ConfigureAwait(false);
                 if (!testAggregate.IsNew)
                 {
+                    var elapsed = stopwatch.Elapsed;
+                    stopwatch.Stop();
                     await AssertJobIsSuccessfullyAsync(jobId).ConfigureAwait(false);
+                    if (minimumElapsed.HasValue)
+                    {
+                        elapsed.ShouldBeGreaterThanOrEqualTo(minimumElapsed.Value);
+                    }
                     Assert.Contains(pingId, testAggregate.PingsReceived.ToList());
-                    Assert.Pass();
+                    return;
                 }
 
                 await Task.Delay(TimeSpan.FromSeconds(0.2)).ConfigureAwait(false);
             }
 
+            stopwatch.Stop();
             Assert.Fail("Aggregate did not receive the command as expected");
         }
 

Source/EventFlow.Hangfire/Integration/HangfireJobScheduler.cs

@@ -71,7 +71,7 @@ public Task<IJobId> ScheduleAsync(IJob job, DateTimeOffset runAt, CancellationTo
                 cancellationToken,
                 (jobDefinition, json) =>
                     _queueName == null
-                        ? _backgroundJobClient.Enqueue(ExecuteMethodCallExpression(jobDefinition, json))
+                        ? _backgroundJobClient.Schedule(ExecuteMethodCallExpression(jobDefinition, json), runAt)
                         : _backgroundJobClient.Schedule(_queueName, ExecuteMethodCallExpression(jobDefinition, json), runAt));
         }
 
@@ -82,7 +82,7 @@ public Task<IJobId> ScheduleAsync(IJob job, TimeSpan delay, CancellationToken ca
                 cancellationToken,
                 (jobDefinition, json) =>
                     _queueName == null
-                        ? _backgroundJobClient.Enqueue(ExecuteMethodCallExpression(jobDefinition, json))
+                        ? _backgroundJobClient.Schedule(ExecuteMethodCallExpression(jobDefinition, json), delay)
                         : _backgroundJobClient.Schedule(_queueName, ExecuteMethodCallExpression(jobDefinition, json), delay));
         }