Using Background Jobs

Background Jobs are queued with the IBackgroundJobs dependency, which is the same for both RDBMS and SQLite Background Jobs, so everything on this page applies whichever provider your App uses.

You can queue your existing ServiceStack Request DTOs as a Background Job in addition to Commands, the recommended way for encapsulating units of logic into internal invokable, inspectable and auto-retryable building blocks.

Queue Commands​

Any API, Controller or Minimal API can execute jobs with the IBackgroundJobs dependency, e.g. here's how you can run a background job to send a new email when an API is called in any new Identity Auth template:

class MyService(IBackgroundJobs jobs) : Service 
{
    public object Any(MyOrder request)
    {
        var jobRef = jobs.EnqueueCommand<SendEmailCommand>(new SendEmail {
            To = "my@email.com",
            Subject = $"Received New Order {request.Id}",
            BodyText = $"""
                       Order Details:
                       {request.OrderDetails.DumptTable()}
                       """,
        });
        //...
    }
}

Which records and immediately executes a worker to execute the SendEmailCommand with the specified SendEmail Request argument. It also returns a reference to a Job which can be used later to query and track the execution of a job.

Queue APIs​

Alternatively a SendEmail API could be executed with just the Request DTO:

var jobRef = jobs.EnqueueApi(new SendEmail {
    To = "my@email.com",
    Subject = $"Received New Order {request.Id}",
    BodyText = $"""
               Order Details:
               {request.OrderDetails.DumptTable()}
               """,
});

Although Sending Emails is typically not an API you want to make externally available and would want to Restrict access or limit usage to specified users.

In both cases the SendEmail Request is persisted into the Jobs database for durability that gets updated as it progresses through the queue.

For execution the API or command is resolved from the IOC before being invoked with the Request. APIs are executed via the MQ Request Pipeline and commands executed using the Commands Feature where they'll also be visible in the Commands Admin UI.

TIP

With RDBMS Background Jobs you can also pass your own IDbConnection to queue a Job in the same transaction as your data.

Background Job Options​

The behavior for each Enqueue* method for executing background jobs can be customized with the following options:

When and where it runs

  • Queue - Run the job on a named queue with its own concurrency, e.g. emails
  • Priority - Higher priority jobs are started first within their queue
  • Worker - Serially process job using a named worker thread
  • ConcurrencyKey - Jobs sharing a key run one at a time, while different keys run in parallel
  • RunAfter - Queue jobs that are only run after a specified date
  • ExpiresAt / ExpiresIn - Cancel the job instead of running it late
  • UserId - Execute within an Authenticated User Context

Workflows

  • DependsOn - Execute jobs after completion of a parent job
  • DependsOnBatch - Execute after every job in a batch has finished
  • BatchId - Add the job to a Job Batch to track its progress
  • Callback - Invoke another command with the result of a successful job
  • ReplyTo - Deliver the result to a URL or MQ queue when the job completes

Failure handling

  • RetryLimit - Override default retry limit for how many attempts should be made to execute a job
  • RetryBackoff, RetryDelay, MaxRetryDelay - How long to wait between retries
  • TimeoutSecs / Timeout - Override default timeout for how long a job should run before being cancelled

Preventing duplicates

Metadata

  • Tag - Group related jobs under a user specified tag
  • TenantId - The tenant the job belongs to, for filtering and reporting
  • CreatedBy - Optional field for capturing the owner of a job
  • Args - Optional String Dictionary of Arguments that can be attached to a Job

Queue Job to run after a specified date​

Using RunAfter lets you queue jobs that are only executed after a specified DateTime, useful for executing resource intensive tasks at low traffic times, e.g:

var jobRef = jobs.EnqueueCommand<AggregateMonthlyDataCommand>(new Aggregate {
       Month = DateTime.UtcNow
   }, 
   new() {
       RunAfter = DateTime.UtcNow.Date.AddDays(1)
   });

Executing jobs with an Authorized User Context​

If you have logic dependent on an Authenticated ClaimsPrincipal or ServiceStack IAuthSession you can have your APIs and Commands also be executed with that user context by specifying the UserId the job should be executed as:

var openAiRequest = new CreateOpenAiChat {
   Request = new() {
       Model = "gpt-4",
       Messages = [
           new() {
               Content = request.Question
           }
       ]
   },
}; 

// Example executing API Job with User Context
jobs.EnqueueApi(openAiRequest, 
    new() {
      UserId = Request.GetClaimsPrincipal().GetUserId(),
      CreatedBy = Request.GetClaimsPrincipal().GetUserName(),
   });

// Example executing Command Job with User Context
jobs.EnqueueCommand<CreateOpenAiChatCommand>(openAiRequest, 
    new() {
      UserId = Request.GetClaimsPrincipal().GetUserId(),
      CreatedBy = Request.GetClaimsPrincipal().GetUserName(),
   });

Inside your API or Command you access the populated User ClaimsPrincipal or ServiceStack IAuthSession using the same APIs that you'd use inside your ServiceStack APIs, e.g:

public class CreateOpenAiChatCommand(IBackgroundJobs jobs) 
    : AsyncCommand<CreateOpenAiChat>
{
    protected override async Task RunAsync(
        CreateOpenAiChat request, CancellationToken token)
    {
        var user = Request.GetClaimsPrincipal();
        var session = Request.GetSession();
        //...
    }
}

Attach Metadata to Jobs​

All above Background Job Options have an effect on when and how Jobs are executed. There are also a number of properties that can be attached to a Job that can be useful in background job processing despite not having any effect on how jobs are executed.

These properties can be accessed by commands or APIs executing the Job and are visible and can be filtered in the Jobs Admin UI to help find and analyze executed jobs.

var jobRef = jobs.EnqueueCommand<CreateOpenAiChatCommand>(openAiRequest, 
   new() {
      // Group related jobs under a common tag
      Tag = "ai",

      // A User-specified or system generated unique Id to track the job
      RefId = request.RefId,
      
      // Capture who created the job
      CreatedBy = Request.GetClaimsPrincipal().GetUserName(),
      
      // Link jobs together that are sent together in a batch
      BatchId = batchId,
      
      // Tenant the job belongs to
      TenantId = tenantId,
      
      // Additional properties about the job that aren't in the Request  
      Args = new() {
          ["Additional"] = "Metadata"
      }
   });

Serially Execute Jobs with named Workers​

By default jobs are executed by the Workers of their queue, which run up to MaxConcurrentJobs (default: the number of CPU cores) jobs at a time. We can also change the behavior to instead execute jobs one-by-one in a serial queue by specifying them to use the same named worker, e.g. new() { Worker = "smtp" }.

Alternatively you can annotate the command with the [Worker] attribute if you always want all jobs executing the command to use the same worker:

[Worker("smtp")]
public class SendEmailCommand(IBackgroundJobs jobs) : SyncCommand<SendEmail>
{
    //...
}

Executing non-durable jobs​

IBackgroundJobs also supports RunCommand methods to be able to execute jobs transiently (i.e. non-durable), which is useful for commands that want to be serially executed by a named worker but don't need to be persisted.

You could use this to queue system emails to be sent by the same smtp worker and are happy to avoid tracking its state and execution history in the Jobs database.

var job = jobs.RunCommand<SendEmailCommand>(new SendEmail { ... }, 
    new() {
        Worker = "smtp"
    });

In this case RunCommand returns the actual BackgroundJob instance that will be updated by the worker.

You can also use RunCommandAsync if you prefer to wait until the job has been executed. Instead of a Job it returns the Result of the command if it returned one.

var result = await jobs.RunCommandAsync<SendEmailCommand>(new SendEmail {...}, 
    new() {
        Worker = "smtp"
    });

Job Execution Limits​

Default Retry and Timeout Limits can be configured on the Backgrounds Job plugin (or DatabaseJobFeature for RDBMS Background Jobs):

services.AddPlugin(new BackgroundsJobFeature
{
   DefaultRetryLimit = 2,                // retries after the first attempt
   DefaultTimeoutSecs = 10 * 60,         // 10 mins
   DefaultRetryBackoff = RetryBackoff.ExponentialJitter,
   DefaultRetryDelayMs = 5_000,          // delay before the first retry
   DefaultMaxRetryDelayMs = 300_000,     // longest delay between retries
});

These limits are also overridable on a per-job basis, e.g:

var jobRef = jobs.EnqueueCommand<AggregateMonthlyDataCommand>(new Aggregate {
       Month = DateTime.UtcNow
   }, 
   new() {
      RetryLimit = 3,
      Timeout = TimeSpan.FromMinutes(30),
   });

See Retries & Reliability for how retries are spaced out, the history kept for each failed attempt, and expiring jobs that shouldn't run late.

Use Callbacks to process the results of Commands​

Callbacks can be used to extend the lifetime of a job to include processing a callback to process its results. This is useful where you would like to reuse the the same command but handle the results differently, e.g. the same command can email results or post them to Slack by using a callback:

jobs.EnqueueCommand<CheckUrlsCommand>(new CheckUrls { Urls = allUrls },
    new() {
        Callback = nameof(EmailUrlResultsCommand),
    });

jobs.EnqueueCommand<CheckUrlsCommand>(new CheckUrls { Urls = criticalUrls },
    new() {
        Callback = nameof(SlackUrlResultsCommand),
    });

Callbacks that fail are auto-retried the same number of times as their jobs, which if they all fail then the entire job is also marked as failed.

Run Job dependent on successful completion of parent​

Jobs can be queued to only run after the successful completion of another job, this is useful for when you need to kick off multiple jobs after a long running task has finished like generating monthly reports after monthly data has been aggregated, e.g:

var jobRef = jobs.EnqueueCommand<AggregateMonthlyDataCommand>(new Aggregate {
    Month = DateTime.UtcNow
});

jobs.EnqueueCommand<GenerateSalesReportCommand>(new () {
   DependsOn = jobRef.Id,
});

jobs.EnqueueCommand<GenerateExpenseReportCommand>(new () {
   DependsOn = jobRef.Id,
});

Inside your command you can get a reference to your current job with Request.GetBackgroundJob() which will have its ParentId populated with the parent job Id and job.ParentJob containing a reference to the completed Parent Job where you can access its Request, Results and other job information:

public class GenerateSalesReportCommand(ILogger<MyCommandNoArgs> log) 
    : SyncCommand
{
    protected override void Run()
    {
        var job = Request.GetBackgroundJob();
        var parentJob = job.ParentJob;
    }
}

Atomic Batching Behavior​

We can also use DependsOn to implement atomic batching behavior where from inside our executing command we can queue new jobs that are dependent on the successful execution of the current job, e.g:

public class CheckUrlsCommand(IHttpClientFactory factory, IBackgroundJobs jobs)
    : AsyncCommand<CheckUrls>
{
    protected override async Task RunAsync(CheckUrls req, CancellationToken ct)
    {
        var job = Request.GetBackgroundJob();

        var batchId = Guid.NewGuid().ToString("N");
        using var client = factory.CreateClient();
        foreach (var url in req.Urls)
        {
            var msg = new HttpRequestMessage(HttpMethod.Get, url);
            var response = await client.SendAsync(msg, ct);
            response.EnsureSuccessStatusCode();
      
            jobs.EnqueueCommand<SendEmailCommand>(new SendEmail {
                To = "my@email.com",
                Subject = $"{new Uri(url).Host} status",
                BodyText = $"{url} is up",
            }, new() {
                DependsOn = job.Id,
                BatchId = batchId,
            });
        }
    }
}

Where any dependent jobs are only executed if the job was successfully completed. If instead an exception was thrown during execution, the job will be failed and all its dependent jobs cancelled and removed from the queue.

TIP

See Workflows & Batches for running a job however its parent finished, tracking the progress of a batch of jobs and running a job once a whole batch has finished.

Logging, Cancellation and Status Updates​

We'll use the command for checking multiple URLs to demonstrate some recommended patterns and how to enlist different job processing features.

public class CheckUrlsCommand(
    ILogger<CheckUrlsCommand> logger,
    IBackgroundJobs jobs,
    IHttpClientFactory clientFactory) : AsyncCommandWithResult<CheckUrls,CheckUrlsResult>
{
    protected override async Task<CheckUrlsResult> RunAsync(CheckUrls req, CancellationToken ct)
    {
        // 1. Create Logger that Logs and maintains logging in Jobs DB
        var log = Request.CreateJobLogger(jobs,logger);

        // 2. Get Current Executing Job
        var job = Request.GetBackgroundJob();

        var result = new CheckUrlsResult {
            Statuses = new()
        };
        using var client = clientFactory.CreateClient();
        for (var i = 0; i < req.Urls.Count; i++)
        {
            // 3. Stop processing Job if it's been cancelled 
            ct.ThrowIfCancellationRequested();

            var url = req.Urls[i];
            try
            {
                var msg = new HttpRequestMessage(HttpMethod.Get,url);
                var response = await client.SendAsync(msg, ct);

                result.Statuses[url] = response.IsSuccessStatusCode;
                log.LogInformation("{Url} is {Status}",
                    url, response.IsSuccessStatusCode ? "up" : "down");

                // 4. Optional: Maintain explicit progress and status updates
                log.UpdateStatus(i/(double)req.Urls.Count,$"Checked {i} URLs");
            }
            catch (Exception e)
            {
                log.LogError(e, "Error checking {Url}", url);
                result.Statuses[url] = false;
            }
        }

        // 5. Return the result, which is delivered to the job's ReplyTo if it has one
        return result;
    }
}

We'll cover some of the notable parts useful when executing Jobs:

1. Job Logger​

We can use a Job logger to enable database logging that can be monitored in real-time in the Admin Jobs UI. Creating it with both BackgroundJobs and ILogger will return a combined logger that both Logs to standard output and to the Jobs database:

var log = Request.CreateJobLogger(jobs,logger);

Or just use Request.CreateJobLogger(jobs) to only save logs to the database.

2. Resolve Executing Job​

If needed the currently executing job can be accessed with:

var job = Request.GetBackgroundJob();

Where you'll be able to access all the metadata the jobs were created with including ReplyTo and Args.

3. Check if Job has been cancelled​

To be able to cancel a long running job you'll need to periodically check if a Cancellation has been requested and throw a TaskCanceledException if it has to short-circuit the command which can be done with:

ct.ThrowIfCancellationRequested();

You'll typically want to call this at the start of any loops to prevent it from doing any more work.

4. Optionally record progress and status updates​

By default Background Jobs looks at the last API or Command run and worker used to estimate the duration and progress for how long a running job will take.

If preferred your command can explicitly set a more precise progress and optional status update that should be used instead, e.g:

log.UpdateStatus(progress:i/(double)req.Urls.Count, $"Checked {i} URLs");

Although generally the estimated duration and live logs provide a good indication for the progress of a job.

5. Notify completion of Job​

Calling a Web Hook is a good way to notify externally initiated job requests of the completion of a job. Instead of calling it from within your command, return the result and queue the job with a ReplyTo - once the job completes its result is delivered automatically:

jobs.EnqueueCommand<CheckUrlsCommand>(new CheckUrls { Urls = urls },
    new() {
        ReplyTo = $"https://api.example.com/callback?refId={refId}",
    });

This has a few benefits over sending the callback yourself:

  • Frees up the worker as soon as the command has finished
  • A failed delivery is logged without failing a job that has already succeeded
  • X-Job-Id, X-Job-RefId, X-Job-BatchId, X-Job-Tag and X-Job-State HTTP Headers let the receiver correlate the result without parsing the body

A ReplyTo that's an http:// or https:// URL receives the result as a JSON POST, anything else is treated as an MQ queue name the result is published to. See Deliver results to ReplyTo to customize how results are delivered, or restrict where they can be sent.

Callback URLs​

Typically URLs will contain a reference Id so external clients can correlate a callback with the internal process that initiated the job. If the callback API is publicly available you'll want to use an internal Id that can't be guessed (like a Guid) so the callback can't be spoofed, e.g:

$"https://api.example.com/callback?refId={RefId}"

If you need more control over how a callback is sent, the SendJsonCallbackAsync extension method supports a number of formats for customizing the HTTP Request, which you can use from your own command or from a custom OnJobReplyTo handler:

await clientFactory.SendJsonCallbackAsync(callbackUrl, result, token);

If the URL contains a space, the text before the space is treated as the HTTP method:

"PUT https://api.example.com/callback"

If the auth part contains a colon : it's treated as Basic Auth:

"username:password@https://api.example.com/callback"

If name starts with http. sends a HTTP Header

"http.X-API-Key:myApiKey@https://api.example.com/callback"

Otherwise it's sent as a Bearer Token:

"myToken123@https://api.example.com/callback"

Bearer Token or HTTP Headers starting with $ is substituted with Environment Variable if exists:

"$API_TOKEN@https://api.example.com/callback"

When needed headers, passwords and tokens can be URL encoded if they contain any delimiter characters.

Querying a Job​

A job can be queried by either it's auto-incrementing Id Primary Key or by a unique RefId that can be user-specified.

var jobResult = jobs.GetJob(jobRef.Id);

var jobResult = jobs.GetJobByRefId(jobRef.RefId);

At a minimum a JobResult will contain the Summary Information about a Job as well as the full information about a job depending on where it's located:

class JobResult
{
    // Summary Metadata about a Job in the JobSummary Table 
    JobSummary Summary
    // Job that's still in the BackgroundJob Queue
    BackgroundJob? Queued
    // Full Job information in Monthly DB CompletedJob Table
    CompletedJob? Completed
    // Full Job information in Monthly DB FailedJob Table
    FailedJob? Failed
    // Helper to access full Job Information
    BackgroundJobBase? Job => Queued ?? Completed ?? Failed 
}

Next steps​

Edit this page on GitHub