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) });
        }
    }
}

private void DeleteJobsFromOldServers()
{
    var monitoringApi = JobStorage.Current.GetMonitoringApi();
    var processingJobs = monitoringApi.ProcessingJobs(0, 10);
    var currentServers = monitoringApi.Servers();

    var currentServersNames = currentServers.Select(server => server.Name).ToHashSet();

    var oldServerJobs = processingJobs
        .Where(job => currentServersNames.Contains(job.Value.ServerId) is false)
        .Select(job => job.Key)
        .ToList();

    foreach (var jobId in oldServerJobs)
    {
        BackgroundJob.Delete(jobId);
    }
}

Before var running = connection.GetValueFromHash($"recurring-job:{recurringJobId}", "Running");