Correct way to reset job queue on startup

ServiceStack
1

We have a number of background tasks (Commands) that need to run on a frequent basis.
Each command should run every x seconds / minutes but not run if the previous execution is still running.

We set up BackgroundJobs for this purpose but because jobs persist across application restarts when we were using jobs.RecurringCommand we realised we after multiple restarts we had a large number of commands queued.

We need new jobs scheduling on app restart because the frequency with which these jobs should execute might have changed.

We’ve tried running jobs.CancelJobs() on app startup, before we use a single jobs.ScheduleCommand() to start the process, with a try catch finally block in our command to reschedule itself x seconds later.

This doesn’t work however, with multiple jobs queued and running again after application restart.

Should jobs.CancelJobs work how we expect? We can’t find a way of programmatically identifying the any jobs and cancelling them.

What is the recommended way of programmatically removing jobs from the queue before we add the jobs on app startup?
Or are we misunderstanding how BackgroundJobs should be used?

2

How did you call CancelJobs? It requires either a BackgroundJobState or a worker to cancel the jobs of.

E.g. If the jobs are queued, you’d cancel them with:

jobs.CancelJobs(state:BackgroundJobState.Queued);

CancelJob without any arguments is a NoOp, what behavior would you prefer instead: to cancel all jobs or throw an Exception?

4

Ok, thanks, so there is no way to simply remove all queued jobs.
So I execute jobs.CancelJobs(BackgroundJobState.Queued) and get…
image

But (after a few restarts) I now have 6 of the same commands queued.

Am I misusing the BackgroundJobs feature?

            var jobs = services.GetRequiredService<IBackgroundJobs>();

            jobs.CancelJobs(BackgroundJobState.Queued);
            jobs.EnqueueCommand<QueueAuditCommand>(
                new BackgroundJobOptions { RunAfter = DateTime.UtcNow.AddSeconds(5) });

[Worker("queue")]
public class QueueAuditCommand(
    IBackgroundJobs jobs,
    ILogger<QueueAuditCommand> log) 
    : IAsyncCommand<NoArgs>
{
    private static readonly SemaphoreSlim ExecutionLock = new(1, 1);

    public async Task ExecuteAsync(NoArgs request)
    {
        // Try to enter the semaphore immediately. If not available, skip execution.
        if (!await ExecutionLock.WaitAsync(0))
        {
            log.LogWarning("Audit queueing is already in progress. Skipping overlapping execution.");
            return;
        }
        
        try
        {
            if (statusChangeMonitor.SyncStatus != ProcessStatus.Active)
            {
                log.LogInformation("Sync is not active, skipping audit queueing.");
                return;
            }

            await PerformSyncExecutionAsync();
        }
        finally
        {
            // Always release the semaphore.
            ExecutionLock.Release();

            jobs.EnqueueCommand<QueueAuditCommand>(
                new BackgroundJobOptions { RunAfter = DateTime.UtcNow.AddSeconds(5) });
        }
    }

    ....
5

I’ve added the jobs.CancelJobs to the HostedJobsService but it fails to cancel or remove the queued jobs.

It feels like I should be doing something else to clear everything down on app startup, any ideas @mythz ?


public class JobsHostedService(ILogger<JobsHostedService> log, IBackgroundJobs jobs) : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        jobs.CancelJobs(BackgroundJobState.Queued, "queue");
        
        await jobs.StartAsync(stoppingToken);
        
        using var timer = new PeriodicTimer(TimeSpan.FromSeconds(3));
        while (!stoppingToken.IsCancellationRequested && await timer.WaitForNextTickAsync(stoppingToken))
        {
            await jobs.TickAsync();
        }
    }
}
6

What state were the incomplete jobs in? You may need to cancel Started jobs as well:

jobs.CancelJobs(state:BackgroundJobState.Queued);
jobs.CancelJobs(state:BackgroundJobState.Started);

I’ve also just added support for cancelling all pending background jobs when called without a filter, e.g:

jobs.CancelJobs();

This change is now available in the pre release packages.

7

Thanks @mythz.

I cannot find a way of using BackgroundJobs that fulfills ALL of the following requirements:

  1. The task should be scheduled to run every X seconds, where X is configurable.
  2. The task can never fire more frequently than X seconds.
  3. If a previously executed task is still executing, any new executions should be skipped.
  4. If the task fails due to database availability etc. queued tasks should not accrue.
  5. Ideally at any point in time there should only be one task in the queue.

My biggest issues are points 2 & 4. If I simulate the db becoming unavailable and the task fails, the scheduled task causes a number of queued tasks to build up; 1 new task every X seconds. When the db then becomes available again I get a spamming of task execution despite having a semaphoreslim to prevent parallel execution.

I think my use-case would be fairly common but I’m beginning to think I can’t achieve this with BackgroundJobs.

Is there a way, or am I better off switching to BackgroundMQ for this use-case? I could keep BackgroundJobs in play for tasks like sending emails, where I want persistence across app restarts. And use BackgroundMQ for the scheduled, non-spamming tasks.

8

If you wan’t tasks to occur after the end of the other tasks you would chain them, e.g. by enquing another command at the end of your command using RunAfter to specify the delay, i.e. instead of having them initiated by a scheduled timer.

But honestly it sounds like you’d get by with a simple flag to prevent other commands from executing whilst it’s still running, e.g. something like:

public class MyScheduledCommand(IBackgroundJobs jobs) : AsyncCommand
{
    private static int running = 0;
    protected override async Task RunAsync(CancellationToken token)
    {
        // Only allow 1 command to execute at a time
        if (Interlocked.CompareExchange(ref running, 1, 0) == 0)
        {
            try {
                // do task...
            } finally {
                Interlocked.Decrement(ref running);
            }
        }
        else
        {
            // Record task was skipped 
            jobs.UpdateJobStatus(new(Request.GetBackgroundJob(), 
                status:"Already running, skipping..."));
        }
    }
}

Background Jobs is the only ServiceStack feature with a concept of Scheduled Recurring tasks, Background MQ and all other MQ providers primary purpose is to enable an MQ endpoint for executing your ServiceStack APIs. For scheduling task alternatives you should look at purpose specific libraries like https://www.quartz-scheduler.net or https://www.hangfire.io