DisableMultipleQueuedItemsFilter.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.

public class DisableMultipleQueuedItemsFilter : JobFilterAttribute, IClientFilter, IServerFilter
{
    private static readonly TimeSpan LockTimeout = TimeSpan.FromSeconds(5);
    private static readonly TimeSpan FingerprintTimeout = TimeSpan.FromHours(1);

    public void OnCreating(CreatingContext filterContext)
    {
        if (!AddFingerprintIfNotExists(filterContext.Connection, filterContext.Job))
        {
            filterContext.Canceled = true;
        }
    }

    public void OnPerformed(PerformedContext filterContext)
    {
        RemoveFingerprint(filterContext.Connection, filterContext.Job);
    }

    private static bool AddFingerprintIfNotExists(IStorageConnection connection, Job job)
    {
        using (connection.AcquireDistributedLock(GetFingerprintLockKey(job), LockTimeout))
        {
            var fingerprint = connection.GetAllEntriesFromHash(GetFingerprintKey(job));

            DateTimeOffset timestamp;

            if (fingerprint != null &&
                fingerprint.ContainsKey("Timestamp") &&
                DateTimeOffset.TryParse(fingerprint["Timestamp"], null, DateTimeStyles.RoundtripKind, out timestamp) &&
                DateTimeOffset.UtcNow <= timestamp.Add(FingerprintTimeout))
            {
                // Actual fingerprint found, returning.
                return false;
            }

            // Fingerprint does not exist, it is invalid (no `Timestamp` key),
            // or it is not actual (timeout expired).
            connection.SetRangeInHash(GetFingerprintKey(job), new Dictionary<string, string>
            {
                { "Timestamp", DateTimeOffset.UtcNow.ToString("o") }
            });

            return true;
        }
    }

    private static void RemoveFingerprint(IStorageConnection connection, Job job)
    {
        using (connection.AcquireDistributedLock(GetFingerprintLockKey(job), LockTimeout))
        using (var transaction = connection.CreateWriteTransaction())
        {
            transaction.RemoveHash(GetFingerprintKey(job));
            transaction.Commit();
        }
    }

    private static string GetFingerprintLockKey(Job job)
    {
        return String.Format("{0}:lock", GetFingerprintKey(job));
    }

    private static string GetFingerprintKey(Job job)
    {
        return String.Format("fingerprint:{0}", GetFingerprint(job));
    }
    
    private static string GetFingerprint(Job job)
    {
        string parameters = string.Empty;
        if (job.Arguments != null)
        {
            parameters = string.Join(".", job.Arguments);
        }
        if (job.Type == null || job.Method == null)
        {
            return string.Empty;
        }
        var fingerprint = String.Format(
            "{0}.{1}.{2}",
            job.Type.FullName,
            job.Method.Name, parameters);

        return fingerprint;
    }

    void IClientFilter.OnCreated(CreatedContext filterContext)
    {
    }

    void IServerFilter.OnPerforming(PerformingContext filterContext)
    {
    }
}

Trying this with 1.7.8 and it does not seem to work.

var fingerprint = connection.GetAllEntriesFromHash(GetFingerprintKey(job));

always seems to return null.

In addition to @dmitry-zaets's change, I added the IApplyStateFilter interface and added this code, to cleanup failed and deleted jobs. This should handle both retried jobs with 0 attempts and restarted servers that interrupts the job

public void OnStateApplied(ApplyStateContext filterContext, IWriteOnlyTransaction transaction)
{
    try
    {
        var failedState = filterContext.NewState is FailedState;
        var deletedState = filterContext.NewState is DeletedState;
        if (failedState || deletedState)
        {
            RemoveFingerprint(filterContext.Connection, filterContext.BackgroundJob.Job);
        }
    }
    catch (Exception)
    {
        // Unhandled exceptions can cause an endless loop.
        // Therefore, catch and ignore them all.
    }
}

The FingerprintTimeout of one hour does however seem counterproductive, as this filter should prevent concurrent jobs. If a job is running for more than an hour, it will be allowed to start. It's however easy to fix, but mayby it shouldn't check the time?

Original GIST should be updated with the latest comments, maybe this should be even added in the official documentation?

Several people seem to need this solution, I am having this problem in an application with hangfire 1.7.x and I'm not sure if this is going to work in production. Reading the comments it seems the original code is not yet production ready

@bokmadsen have you come up with a solution for the one hour FingerprintTimeout limitation (besides simply increasing the TimeSpan :)

Or anyone else have any suggestions / recommendations... it works fine, except like you say the job runs for more than the fingerprint timeout, at which point a duplicate is scheduled...

@mcarter101 ta ha zemren

My use case is that I don't want to enqueue a second job (with same parameter values) until the enqueued job goes to Succeeded or Failed.
From initial testing I am getting the results I want (block successive enqueue until job succeeds or fails) removing the fingerprint in IApplyStateFilter.OnStateApplied. Any feedback?

 public class DisableMultipleQueuedItemsFilter : JobFilterAttribute, IClientFilter, IApplyStateFilter
    {
        private static readonly TimeSpan LockTimeout = TimeSpan.FromSeconds(5);

        private static bool AddFingerprintIfNotExists(IStorageConnection connection, Job job)
        {
            using (connection.AcquireDistributedLock(GetFingerprintLockKey(job), LockTimeout))
            {
                var fingerprint = connection.GetAllEntriesFromHash(GetFingerprintKey(job));

                if (fingerprint != null)
                {
                    // Actual fingerprint found, returning.
                    return false;
                }

                // Fingerprint does not exist, it is invalid (no `Timestamp` key),
                // or it is not actual (timeout expired).
                connection.SetRangeInHash(GetFingerprintKey(job), new Dictionary<string, string>
                {
                    { "Timestamp", DateTimeOffset.UtcNow.ToString("o") }
                });

                return true;
            }
        }

        private static void RemoveFingerprint(IStorageConnection connection, Job job)
        {
            using (connection.AcquireDistributedLock(GetFingerprintLockKey(job), LockTimeout))
            using (var transaction = connection.CreateWriteTransaction())
            {
                transaction.RemoveHash(GetFingerprintKey(job));
                transaction.Commit();
            }
        }

        private static string GetFingerprintLockKey(Job job)
        {
            return string.Format("{0}:lock", GetFingerprintKey(job));
        }

        private static string GetFingerprintKey(Job job)
        {
            return string.Format("fingerprint:{0}", GetFingerprint(job));
        }

        private static string GetFingerprint(Job job)
        {
            string parameters = string.Empty;
            if (job?.Args != null)
            {
                parameters = string.Join(".", job.Args);
            }
            if (job?.Type == null || job.Method == null)
            {
                return string.Empty;
            }

            //https://gist.github.com/odinserj/a8332a3f486773baa009#gistcomment-1898401
            var payload = $"{job.Type.FullName}.{job.Method.Name}.{parameters}";
            var hash = SHA256.Create().ComputeHash(System.Text.Encoding.UTF8.GetBytes(payload));
            var fingerprint = Convert.ToBase64String(hash);
            return fingerprint;
        }

        public void OnCreating(CreatingContext filterContext)
        {
            if (!AddFingerprintIfNotExists(filterContext.Connection, filterContext.Job))
            {
                filterContext.Canceled = true;
            }
        }

        public void OnCreated(CreatedContext filterContext)
        {
            //do nothing
        }


        public void OnStateApplied(ApplyStateContext context, IWriteOnlyTransaction transaction)
        {
            if (context.NewState.Name.Equals(Hangfire.States.SucceededState.StateName)
                || context.NewState.Name.Equals(Hangfire.States.FailedState.StateName))
            {
                RemoveFingerprint(context.Connection, context.BackgroundJob.Job);
            }
        }

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

@mack0196 Your solution is working great for me. I don't have any particular feedback. Thanks for pulling together the final solution from all these comments!

Took me a while to figure out an issue I was having with the above code. When running multiple servers with different cultures AND having jobs with either doubles or dates in the arguments, this is going to generate different fingerprints. The main issue could be that not everybody sets the current culture, but the line

parameters = string.Join(".", job.Args);

will invoke ToString() on all arguments, resulting in this behavior.

So, either always set the CurrentCulture, or use something like below:

private static readonly CultureInfo EnUs = new CultureInfo("en-US");

private static string ConvertArgument(object obj)
{
        switch (obj)
        {
                case null:
                        return string.Empty;
                case string s:
                        return s;
                case DateTime dt:
                        return dt.ToString("o"); // ISO8601 date format
                default:
                        return (string) Convert.ChangeType(obj, typeof(string), EnUs); // And force the rest to US English
        }
}

and change the line above to

parameters = string.Join(".", job.Args.Select(ConvertArgument));

Thanks erwin-faceit. Updated.

using Hangfire.Client;
using Hangfire.Common;
using Hangfire.States;
using Hangfire.Storage;
using System;
using System.Collections.Generic;
using System.Globalization;
using System.Linq;
using System.Security.Cryptography;

namespace Hangfire.Filters
{
    public class DisableMultipleQueuedItemsFilter : JobFilterAttribute, IClientFilter, IApplyStateFilter
    {
        private static readonly TimeSpan LockTimeout = TimeSpan.FromSeconds(5);
        private static readonly CultureInfo EnUs = new CultureInfo("en-US");

        /// <summary>
        /// Convert arguments into a culture-aware string
        /// </summary>
        ///<see cref="https://gist.github.com/odinserj/a8332a3f486773baa009?permalink_comment_id=4048344#gistcomment-4048344"/><see>
        private static string ConvertArgument(object obj)
        {
            switch (obj)
            {
                case null:
                    return string.Empty;
                case string s:
                    return s;
                case DateTime dt:
                    return dt.ToString("o"); // ISO8601 date format
                default:
                    return (string)Convert.ChangeType(obj, typeof(string), EnUs); // And force the rest to US English
            }
        }

        private static bool AddFingerprintIfNotExists(IStorageConnection connection, Job job)
        {
            var fingerprintKey = GetFingerprintKey(job);
            var finterprintLockKey = GetFingerprintLockKey(fingerprintKey);
            var distributedLock = connection.AcquireDistributedLock(finterprintLockKey, LockTimeout);
            using (distributedLock)
            {
                var fingerprint = connection.GetAllEntriesFromHash(fingerprintKey);

                if (fingerprint != null)
                {
                    // Actual fingerprint found, returning.
                    return false;
                }

                // Fingerprint does not exist, it is invalid (no `Timestamp` key),
                // or it is not actual (timeout expired).
                connection.SetRangeInHash(fingerprintKey, new Dictionary<string, string>
                    {
                        { "Timestamp", DateTimeOffset.UtcNow.ToString("o") }
                    });

                return true;
            }
        }

        private static void RemoveFingerprint(IStorageConnection connection, Job job)
        {
            var fingerprintKey = GetFingerprintKey(job);
            var finterprintLockKey = GetFingerprintLockKey(fingerprintKey);
            using (connection.AcquireDistributedLock(finterprintLockKey, LockTimeout))
            using (var transaction = connection.CreateWriteTransaction())
            {
                transaction.RemoveHash(fingerprintKey);
                transaction.Commit();
            }
        }

        private static string GetFingerprintLockKey(string fingerprintKey)
        {
            return string.Format("{0}:lock", fingerprintKey);
        }

        private static string GetFingerprintKey(Job job)
        {
            return string.Format("fingerprint:{0}", GetFingerprint(job));
        }

        private static string GetFingerprint(Job job)
        {
            string parameters = string.Empty;
            if (job?.Args != null)
            {
                parameters = string.Join(".", job.Args.Select(ConvertArgument));
            }
            if (job?.Type == null || job.Method == null)
            {
                return string.Empty;
            }

            //https://gist.github.com/odinserj/a8332a3f486773baa009#gistcomment-1898401
            var payload = $"{job.Type.FullName}.{job.Method.Name}.{parameters}";
            var hash = SHA256.Create().ComputeHash(System.Text.Encoding.UTF8.GetBytes(payload));
            var fingerprint = Convert.ToBase64String(hash);
            return fingerprint;
        }

        public void OnCreating(CreatingContext filterContext)
        {
            if (!AddFingerprintIfNotExists(filterContext.Connection, filterContext.Job))
            {
                filterContext.Canceled = true;
            }
        }

        public void OnCreated(CreatedContext filterContext)
        {
            //do nothing
        }


        public void OnStateApplied(ApplyStateContext context, IWriteOnlyTransaction transaction)
        {
            if (context.NewState.Name.Equals(Hangfire.States.SucceededState.StateName)
                || context.NewState.Name.Equals(Hangfire.States.FailedState.StateName))
            {
                RemoveFingerprint(context.Connection, context.BackgroundJob.Job);
            }
        }

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

⚠️⚠️⚠️ This works for the Enqueue only; Scheduled jobs will duplicate.

⚠️⚠️⚠️ This filter will break ContinueWith scenarios that expect a jobId string returned from BackgroundJobClient.Enqueue.

@mack0196, I believe you should check for the transition to Deleted state as well for fingerprint removal.

This solution sometimes is generating RedisTimeoutException upon fingerprint removal and in the error message it is something about
SUBSCRIBE fingerprint:60684f8fd3ad130e8c60210b41706448:lock:ev

Thanks mcastellano

Hello,
We experienced problem when passing CancellationToken into a job.
https://docs.hangfire.io/en/latest/background-methods/using-cancellation-tokens.html
In our case, when job was completed calculated hash was incorrect as Hangfire replaced it with null in Parameters.
To fix it we changed ConvertArgument method to ignore token.

@afelinczak

To fix it we changed ConvertArgument method to ignore token.

Please share your code changes

@HenrikHoyer Did you ever figure out the code @afelinczak was talking about

Hello, I missed the comment - sorry.
This is the fix we are using.

private static string ConvertArgument(object obj) => obj switch { CancellationToken => String.Empty, _ => JsonConvert.SerializeObject(obj) };

Hangfire replaced it with null in Parameters

Hello @afelinczak. What do you mean? CancellationToken is a struct, it can't be null. We must pass it as the default but it'll be replaced with real token in runtime. So it shouldn't be null anyway.

Btw. There is at least one more "special" parameter: PerformContext. It is not well documented but is is used by popular extension Hangfire.Console

The attribute does not work for me.
I tried applying it to the job's class and to the job's execution method.
Neither seems to work.

By persisting the JobId, the system gains resilience against service restarts. If the service goes down and later recovers, the job can be re-scheduled and processed again. Since the same JobId is preserved, the fingerprint logic recognizes it as the same execution context and allows it to proceed, preventing the system from being stuck in a permanently locked state waiting for the FingerprintTimeout.

Tested with recurring jobs, which is why I couldn’t use OnPerforming, it is not being called.

public class DisableMultipleQueuedItemsFilterAttribute : JobFilterAttribute, IServerFilter, IElectStateFilter
{
    private static readonly ILog log =
        LogManager.GetLogger(System.Reflection.MethodBase.GetCurrentMethod().DeclaringType);

    private static readonly TimeSpan LockTimeout = TimeSpan.FromSeconds(5);
    private static readonly TimeSpan FingerprintTimeout = TimeSpan.FromHours(1);

    private const String TimestampKey = "Timestamp";
    private const String JobIdKey = "JobId";

    private const String FingerprintPrefix = "fingerprint:";
    private const String LockSuffix = ":lock";

    public void OnStateElection(ElectStateContext context)
    {
        log.Debug("Entering method DisableMultipleQueuedItemsFilterAttribute::OnStateElection");

        var backgroundJob = context.BackgroundJob;
        var job = backgroundJob.Job;
        var jobId = backgroundJob.Id;

        if (context.CandidateState is ProcessingState)
        {
            var connection = context.Connection;
            var added = AddFingerprintIfNotExists(connection, job, jobId);

            if (!added)
            {
                var deletedState = new DeletedState
                {
                    Reason = "Active fingerprint: an execution is already in progress."
                };

                context.CandidateState = deletedState;
            }
        }

        log.Debug("Exiting method DisableMultipleQueuedItemsFilterAttribute::OnStateElection");
    }

    public void OnPerformed(PerformedContext filterContext)
    {
        log.Debug("Entering method DisableMultipleQueuedItemsFilterAttribute::OnPerformed");

        var backgroundJob = filterContext.BackgroundJob;
        var job = backgroundJob.Job;
        var connection = filterContext.Connection;

        RemoveFingerprint(connection, job);

        log.Debug("Exiting method DisableMultipleQueuedItemsFilterAttribute::OnPerformed");
    }

    private static Boolean AddFingerprintIfNotExists(
        IStorageConnection connection,
        Job job,
        String backgroundJobId)
    {
        log.Debug("Entering method DisableMultipleQueuedItemsFilterAttribute::AddFingerprintIfNotExists");

        var fingerprintKey = GetFingerprintKey(job);
        var fingerprintLockKey = GetFingerprintLockKey(job);

        using (connection.AcquireDistributedLock(fingerprintLockKey, LockTimeout))
        {
            var fingerprint = connection.GetAllEntriesFromHash(fingerprintKey);

            if (fingerprint != null && fingerprint.ContainsKey(TimestampKey))
            {
                var hasJobId = fingerprint.ContainsKey(JobIdKey);

                if (hasJobId)
                {
                    var storedJobId = fingerprint[JobIdKey];

                    if (storedJobId == backgroundJobId)
                    {
                        UpdateFingerprint(connection, job, backgroundJobId);

                        log.Debug("Fingerprint belongs to the same JobId. Updating.");

                        log.Debug("Exiting method DisableMultipleQueuedItemsFilterAttribute::AddFingerprintIfNotExists");
                        return true;
                    }
                }

                var timestampString = fingerprint[TimestampKey];

                var parsed = DateTimeOffset.TryParse(
                    timestampString,
                    null,
                    DateTimeStyles.RoundtripKind,
                    out var timestamp);

                if (parsed)
                {
                    var expiration = timestamp.Add(FingerprintTimeout);
                    var now = DateTimeOffset.UtcNow;

                    if (now <= expiration)
                    {
                        log.Debug("Active fingerprint found.");

                        log.Debug("Exiting method DisableMultipleQueuedItemsFilterAttribute::AddFingerprintIfNotExists");
                        return false;
                    }
                }
            }

            UpdateFingerprint(connection, job, backgroundJobId);

            log.Debug("Fingerprint created.");

            log.Debug("Exiting method DisableMultipleQueuedItemsFilterAttribute::AddFingerprintIfNotExists");
            return true;
        }
    }

    private static void UpdateFingerprint(
        IStorageConnection connection,
        Job job,
        String backgroundJobId)
    {
        log.Debug("Entering method DisableMultipleQueuedItemsFilterAttribute::UpdateFingerprint");

        var fingerprintKey = GetFingerprintKey(job);

        var values = new Dictionary<String, String>
        {
            { TimestampKey, DateTimeOffset.UtcNow.ToString("o") },
            { JobIdKey, backgroundJobId }
        };

        connection.SetRangeInHash(fingerprintKey, values);

        log.Debug("Exiting method DisableMultipleQueuedItemsFilterAttribute::UpdateFingerprint");
    }

    private static void RemoveFingerprint(
        IStorageConnection connection,
        Job job)
    {
        log.Debug("Entering method DisableMultipleQueuedItemsFilterAttribute::RemoveFingerprint");

        var fingerprintKey = GetFingerprintKey(job);
        var fingerprintLockKey = GetFingerprintLockKey(job);

        using (connection.AcquireDistributedLock(fingerprintLockKey, LockTimeout))
        using (var transaction = connection.CreateWriteTransaction())
        {
            transaction.RemoveHash(fingerprintKey);
            transaction.Commit();
        }

        log.Debug("Exiting method DisableMultipleQueuedItemsFilterAttribute::RemoveFingerprint");
    }

    private static String GetFingerprint(Job job)
    {
        log.Debug("Entering method DisableMultipleQueuedItemsFilterAttribute::GetFingerprint");

        var args = job.Args;
        var parameters = String.Empty;

        if (args != null)
            parameters = String.Join(".", args);

        var jobType = job.Type;
        var jobMethod = job.Method;

        if (jobType == null || jobMethod == null)
            return String.Empty;

        var typeName = jobType.FullName;
        var methodName = jobMethod.Name;

        var payload = $"{typeName}.{methodName}.{parameters}";

        var encoding = System.Text.Encoding.UTF8;
        var payloadBytes = encoding.GetBytes(payload);

        var sha = SHA256.Create();
        var hashBytes = sha.ComputeHash(payloadBytes);

        var fingerprint = Convert.ToBase64String(hashBytes);

        log.Debug("Exiting method DisableMultipleQueuedItemsFilterAttribute::GetFingerprint");

        return fingerprint;
    }

    private static String GetFingerprintLockKey(Job job)
    {
        var fingerprintKey = GetFingerprintKey(job);
        var lockKey = String.Format("{0}{1}", fingerprintKey, LockSuffix);

        return lockKey;
    }

    private static String GetFingerprintKey(Job job)
    {
        var fingerprint = GetFingerprint(job);
        var key = String.Format("{0}{1}", FingerprintPrefix, fingerprint);

        return key;
    }

    public void OnPerforming(PerformingContext filterContext)
    {
    }
}