SkipWhenPreviousJobIsRunningAttribute.cs

SkipWhenPreviousJobIsRunningAttribute.cs

// Zero-Clause BSD (more permissive than MIT, doesn't require copyright notice)
// 
// Permission to use, copy, modify, and/or distribute this software for any purpose
// with or without fee is hereby granted.
// 
// THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
// WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
// AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT,
// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS
// OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER
// TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF
// THIS SOFTWARE.

// Hangfire.Core 1.8+ is required, for previous versions please see revision from year 2022.

using System;
using System.Collections.Generic;
using Hangfire.Client;
using Hangfire.Common;
using Hangfire.States;
using Hangfire.Storage;

namespace ConsoleApp28
{
    public class SkipWhenPreviousJobIsRunningAttribute : JobFilterAttribute, IClientFilter, IApplyStateFilter
    {
        public void OnCreating(CreatingContext context)
        {
            // We can't handle old storages
            if (!(context.Connection is JobStorageConnection connection)) return;

            // We should run this filter only for background jobs based on 
            // recurring ones
            if (!context.Parameters.TryGetValue("RecurringJobId", out var parameter)) return;

            var recurringJobId = parameter as string;

            // RecurringJobId is malformed. This should not happen, but anyway.
            if (String.IsNullOrWhiteSpace(recurringJobId)) return;

            var running = connection.GetValueFromHash($"recurring-job:{recurringJobId}", "Running");
            if ("yes".Equals(running, StringComparison.OrdinalIgnoreCase))
            {
                context.Canceled = true;
            }
        }

        public void OnCreated(CreatedContext filterContext)
        {
        }

        public void OnStateApplied(ApplyStateContext context, IWriteOnlyTransaction transaction)
        {
            if (context.NewState is EnqueuedState)
            {
                ChangeRunningState(context, "yes");
            }
            else if ((context.NewState.IsFinal && !FailedState.StateName.Equals(context.OldStateName, StringComparison.OrdinalIgnoreCase)) || 
                     (context.NewState is FailedState))
            {
                ChangeRunningState(context, "no");
            }
        }

        public void OnStateUnapplied(ApplyStateContext context, IWriteOnlyTransaction transaction)
        {
        }

        private static void ChangeRunningState(ApplyStateContext context, string state)
        {
            // We can't handle old storages
            if (!(context.Connection is JobStorageConnection connection)) return;

            // Obtaining a recurring job identifier
            var recurringJobId = context.GetJobParameter<string>("RecurringJobId", allowStale: true);
            if (String.IsNullOrWhiteSpace(recurringJobId)) return;

            if (context.Storage.HasFeature(JobStorageFeatures.Transaction.AcquireDistributedLock))
            {
                // Acquire a lock in newer storages to avoid race conditions
                ((JobStorageTransaction)context.Transaction).AcquireDistributedLock(
                    $"lock:recurring-job:{recurringJobId}",
                    TimeSpan.FromSeconds(5));
            }

            // Checking whether recurring job exists
            var recurringJob = connection.GetValueFromHash($"recurring-job:{recurringJobId}", "Job");
            if (String.IsNullOrEmpty(recurringJob)) return;

            // Changing the running state
            context.Transaction.SetRangeInHash(
                $"recurring-job:{recurringJobId}",
                new[] { new KeyValuePair<string, string>("Running", state) });
        }
    }
}

Am I right, that this attribute would not prevent a recurring job to be run when the job was already manually triggered by a user?

In my case I've had a problem when a concurrent job was stuck in kind of "loop" (so sometimes the job was stuck and didn't run at all for hours,days..), that was because I was making a lot "Task.Run(...)" in other services which are not related to Hangfire, so I simply moved most of my Task.Run to System.Threading.Channels.Channel and all works well now

It's works perfect. Many thanks.

But it not work if you place the attribute in a interface.

This is OK

[SkipWhenPreviousJobIsRunning]
public class JobWithITaskDelay90Sec : ITask<bool>

This not working.

public class JobWithITaskDelay90Sec : ITask<bool>
{
}

[SkipWhenPreviousJobIsRunning]
public interface ITask<TResult>
{
}

Thank you for providing this code. It perfectly matches the functionality I was looking for.

Unfortunately, it doesn't work for me. Initially, it seemed to work fine on my local setup with a single Hangfire instance, although I didn't test it for very long.

Now that I've deployed it to four test instances, I'm seeing multiple instances of the same RecurringJob running simultaneously. I created a dummy job with Task.Delay for 10 minutes, which starts every minute. It ran overnight, and the test instances were restarted by IIS at some point during the night. Now, I have multiple instances of this job running concurrently.

I'm curious about how this differs from Hangfire Ace's concurrency and throttling via mutex?