fix fetched jobs should only update its heartbeat if processing

src/Hangfire.Mongo.Sample.ASPNetCore/Controllers/HomeController.cs

@@ -89,8 +89,8 @@ public ActionResult Delayed(int id)
 
  public ActionResult Recurring()
  {
- RecurringJob.AddOrUpdate("recurring-job",
- () => Recurring($@"Hangfire recurring task started - {Guid.NewGuid()}"), Cron.Minutely);
+ RecurringJob.AddOrUpdate<MyRecurringjob>("my-recurring-job",
+ j => j.Recurring($@"Hangfire recurring task started - {Guid.NewGuid()}"), Cron.Minutely);
 
  return RedirectToAction("Index");
  }
@@ -100,10 +100,5 @@ public static void PrintToDebug(string message)
  Debug.WriteLine(message);
  }
 
- public static void Recurring(string message)
- {
- Thread.Sleep(15000);
- Debug.WriteLine(message);
- }
  }
 }

src/Hangfire.Mongo.Sample.ASPNetCore/MyRecurringjob.cs

@@ -0,0 +1,18 @@
+using System;
+using System.Diagnostics;
+using System.Threading;
+
+namespace Hangfire.Mongo.Sample.ASPNetCore;
+
+[Queue("not-default")]
+[AutomaticRetry(Attempts = 0, LogEvents = true, OnAttemptsExceeded = AttemptsExceededAction.Delete)]
+[SkipWhenPreviousJobIsRunning]
+public class MyRecurringjob
+{
+ // [DisableConcurrentExecution("{0}", 3)]
+ public void Recurring(string message)
+ {
+ Thread.Sleep(TimeSpan.FromMinutes((1)));
+ Console.WriteLine(message);
+ }
+}

src/Hangfire.Mongo.Sample.ASPNetCore/SkipWhenPreviousJobIsRunningAttribute.cs

@@ -0,0 +1,93 @@
+using System;
+using System.Collections.Generic;
+using Hangfire.Client;
+using Hangfire.Common;
+using Hangfire.States;
+using Hangfire.Storage;
+
+namespace Hangfire.Mongo.Sample.ASPNetCore;
+
+// Copied from here: https://gist.github.com/odinserj/a6ad7ba6686076c9b9b2e03fcf6bf74e
+// TODO: Add to Framework-Common's HangfireLocal NuGet package
+public class SkipWhenPreviousJobIsRunningAttribute : JobFilterAttribute, IClientFilter, IApplyStateFilter
+{
+ private const string Running = "Running";
+ private const string RecurringJobParam = "RecurringJobId";
+ private const string KeyPrefix = "recurring-job:";
+ private const string Yes = "yes";
+ private const string No = "no";
+
+ public void OnCreating(CreatingContext context)
+ {
+ Console.WriteLine($"OnCreating: Queue: {context.Job.Queue}, Canceled: {context.Canceled}");
+ // We can't handle old storages
+ if (context.Connection is not JobStorageConnection connection)
+ {
+ return;
+ }
+
+ // We should run this filter only for background jobs based on recurring ones
+ if (!context.Parameters.ContainsKey(RecurringJobParam))
+ {
+ return;
+ }
+
+ var recurringJobId = context.Parameters[RecurringJobParam] as string;
+
+ // RecurringJobId is malformed. This should not happen, but anyway.
+ if (string.IsNullOrWhiteSpace(recurringJobId))
+ {
+ return;
+ }
+
+ var running = connection.GetValueFromHash($"{KeyPrefix}{recurringJobId}", Running);
+ if (running?.Equals(Yes, StringComparison.OrdinalIgnoreCase) == true)
+ {
+ Console.WriteLine($"OnCreating: Setting Canceled: true");
+ context.Canceled = true;
+ }
+ }
+
+ public void OnCreated(CreatedContext filterContext)
+ {
+ }
+
+ public void OnStateApplied(ApplyStateContext context, IWriteOnlyTransaction transaction)
+ {
+ Console.WriteLine($"OnStateApplied: NewState: {context.NewState.Name}");
+ if (context.NewState is EnqueuedState)
+ {
+ var recurringJobId = SerializationHelper.Deserialize<string>(
+ context.Connection.GetJobParameter(context.BackgroundJob.Id, RecurringJobParam));
+ if (string.IsNullOrWhiteSpace(recurringJobId))
+ {
+ return;
+ }
+
+ Console.WriteLine($"OnStateApplied: Setting: {Running}:{Yes}");
+ transaction.SetRangeInHash(
+ $"{KeyPrefix}{recurringJobId}",
+ new[] {new KeyValuePair<string, string>(Running, Yes)});
+ }
+ else if ((context.NewState.IsFinal &&
+ !FailedState.StateName.Equals(context.OldStateName, StringComparison.OrdinalIgnoreCase)) ||
+ (context.NewState is FailedState))
+ {
+ var recurringJobId =
+ SerializationHelper.Deserialize<string>(
+ context.Connection.GetJobParameter(context.BackgroundJob.Id, RecurringJobParam));
+ if (string.IsNullOrWhiteSpace(recurringJobId))
+ {
+ return;
+ }
+ Console.WriteLine($"OnStateApplied: Setting: {Running}:{No}");
+ transaction.SetRangeInHash(
+ $"{KeyPrefix}{recurringJobId}",
+ new[] {new KeyValuePair<string, string>(Running, No)});
+ }
+ }
+
+ public void OnStateUnapplied(ApplyStateContext context, IWriteOnlyTransaction transaction)
+ {
+ }
+}

src/Hangfire.Mongo.Sample.ASPNetCore/Startup.cs

@@ -1,3 +1,4 @@
+using System;
 using Hangfire.Mongo.Migration.Strategies;
 using Hangfire.Mongo.Migration.Strategies.Backup;
 using Microsoft.AspNetCore.Builder;
@@ -49,17 +50,17 @@ public void ConfigureServices(IServiceCollection services)
  MigrationStrategy = new MigrateMongoMigrationStrategy(),
  BackupStrategy = new CollectionMongoBackupStrategy()
  },
- CheckQueuedJobsStrategy = CheckQueuedJobsStrategy.Watch,
+ SlidingInvisibilityTimeout = TimeSpan.FromSeconds(5)
  };
 
  //config.UseLogProvider(new FileLogProvider());
  config.SetDataCompatibilityLevel(CompatibilityLevel.Version_180);
  config.UseMongoStorage(mongoClient, mongoUrlBuilder.DatabaseName, storageOptions)
- .UseColouredConsoleLogProvider(LogLevel.Info);
+ .UseColouredConsoleLogProvider(LogLevel.Trace);
  });
  services.AddHangfireServer(options =>
  {
- options.Queues = new[] { "default", "notDefault" };
+ options.Queues = new[] { "default", "not-default" };
  });
  services.AddMvc(c => c.EnableEndpointRouting = false);
 

src/Hangfire.Mongo.Tests/MongoFetchedJobFacts.cs

@@ -4,6 +4,7 @@
 using Hangfire.Mongo.Database;
 using Hangfire.Mongo.Dto;
 using Hangfire.Mongo.Tests.Utils;
+using Hangfire.States;
 using MongoDB.Bson;
 using MongoDB.Driver;
 using Xunit;
@@ -142,7 +143,7 @@ public void Heartbeat_LonRunningJob_UpdatesFetchedAt()
  var options = new MongoStorageOptions() {SlidingInvisibilityTimeout = TimeSpan.FromSeconds(1)};
  var queue = "default";
  var jobId = ObjectId.GenerateNewId();
- var id = CreateJobQueueRecord(_dbContext, jobId, queue, _fetchedAt);
+ var id = CreateJobQueueRecord(_dbContext, jobId, queue, _fetchedAt, ProcessingState.StateName);
  var initialFetchedAt = DateTime.UtcNow;
 
  // Act
@@ -155,18 +156,24 @@ public void Heartbeat_LonRunningJob_UpdatesFetchedAt()
  Assert.True(job.FetchedAt > initialFetchedAt, "Expected job FetchedAt field to be updated");
  }
 
- private ObjectId CreateJobQueueRecord(HangfireDbContext connection, ObjectId jobId, string queue, DateTime? fetchedAt)
+ private ObjectId CreateJobQueueRecord(
+ HangfireDbContext connection, 
+ ObjectId jobId, 
+ string queue, 
+ DateTime? fetchedAt,
+ string stateName = null)
  {
- var jobQueue = new JobDto
+ var job = new JobDto
  {
  Id = jobId,
  Queue = queue,
- FetchedAt = fetchedAt
+ FetchedAt = fetchedAt,
+ StateName = stateName
  };
 
- connection.JobGraph.InsertOne(jobQueue.Serialize());
+ connection.JobGraph.InsertOne(job.Serialize());
 
- return jobQueue.Id;
+ return job.Id;
  }
  }
 #pragma warning restore 1591

src/Hangfire.Mongo/Database/HangfireDbContext.cs

@@ -1,5 +1,4 @@
 using System;
-using Hangfire.Mongo.Dto;
 using MongoDB.Bson;
 using MongoDB.Driver;