Skip to main content

Jobs

An ordinary scheduled job is a C# class that implements IScheduledJob. Queue jobs use IQueueJob<T> or IBatchQueueJob<T> and can start when messages are ready instead of using a schedule. Both are registered through AddJobs(). Distributed queue processors are a separate alternative without job runs.

The simplest possible implementation of a job would look like this:

public class MyJob : IScheduledJob
{
    public string DefaultSchedule => CronSchedule.TimesPerMinute(10);

    public async Task<JobResults> ExecuteAsync(CancellationToken cancellationToken)
    {
        Console.WriteLine("The job ran");

        return JobResults.Completed();
    }
}

By default the name used as the key for the job in the database is its type/class name. You can use the [Job] attribute to give it a stable name and display name:

[Job("myjob", DisplayName = "My really great job")]
public class MyJob : IScheduledJob
{
}

Now we'll use myjob as the key in the database and the admin UI will display it as My really great job. Now you can also rename the job class without loosing the meta data.

IScheduledJob.DefaultSchedule​

This property tells the job engine what the default schedule of the job is, and in the above example we're scheduling it to run ten times per minute. But DefaultSchedule is just a string that should contain a valid cron schedule (including seconds) so you can return any valid cron expression. CronSchedule here is just a helper class for generating such schedules.

The engine is using Cronos to parse cron expressions, so any expression that is valid in Cronos will work.

The reason for the property being called DefaultSchedule rather than Schedule is because it will only be read once by the system and that is the first time the engine registers the job. After that the schedule is saved to the database and can be changed through the UI and the API.

If you set DefaultSchedule to an empty string (or use CronSchedule.NotScheduled()) the job will get registered but the engine will never execute it unless you start it manually through the UI or the API.

Schedule timezone​

Job cron schedules are evaluated in UTC by default. To run jobs at a fixed local time, configure ScheduleTimeZone:

builder.Services.AddNexus().AddJobs(options =>
{
    options.ScheduleTimeZone = TimeZoneInfo.FindSystemTimeZoneById("Europe/Stockholm");
});

With this setting, 0 0 9 * * * runs at 09:00 in Stockholm, corresponding to 08:00 UTC in winter and 07:00 UTC in summer.

ScheduleTimeZone accepts a TimeZoneInfo object. Use TimeZoneInfo.Utc to preserve the default behavior, TimeZoneInfo.Local for the server's timezone, or resolve a named timezone with TimeZoneInfo.FindSystemTimeZoneById. Named IDs must be supported on the host; invalid IDs throw when that lookup is called during configuration. You can also supply a custom TimeZoneInfo.

The setting applies to all job cron schedules, including schedules changed through the UI or API. Configure the same effective timezone on every instance that runs these jobs; TimeZoneInfo.Local requires matching server timezone settings. Stored timestamps and API dates, including extra runs, remain UTC. New jobs still run immediately the first time they are registered.

Daylight-saving transitions follow Cronos's behavior: a daily time in the spring clock gap runs at the next valid local time, and a daily time in the autumn clock overlap runs once.

IScheduledJob.ExecuteAsync()​

This method is where you'll place the code that should be performed by the job. The example just contains a simple Console.WriteLine() which won't be very useful in a real scenario.

The job instance is created inside a dependency injection scope, so it can request scoped application and Nexus services through constructor injection. See dependency injection.

Instantiation

Note that the job might be instantiated for other reasons than to call ExecuteAsync() so don't do any work in the constructor, you should only set instance fields.

CancellationToken​

The ExecuteAsync() method is passed a CancellationToken by the engine which the job should respect. If you've never worked with cancellation tokens before it's a way for the outside to signal into the job that someone wants to cancel the job execution.

It's up to every job to respect the cancellation token passed in and if you don't do that the job can't be cancelled. It'll either always run to completion or be forcefully killed if the job service is killed.

The cancellation token passed in is a combination of the general service cancellation token and a cancellation token for that specific job run. The service cancellation token can be triggered by a deploy or a machine restart whereas the cancellation token for a specific job is only triggered if someone manually cancels the job through the UI or the API. The difference here is not something a job needs to think about, it only have to concern itself with the cancellation token passed to ExecuteAsync().

Here's an example of how you would use the cancellation token:

public async Task<JobResults> ExecuteAsync(CancellationToken cancellationToken)
{
    foreach (var item in await _service.GetBigListOfThingsAsync())
    {
        cancellationToken.ThrowIfCancellationRequested();

        // Do stuff with the item
    }

    return JobResults.Completed();
}

You add calls to cancellationToken.ThrowIfCancellationRequested(); inside loops or between chunks of code that are time consuming. A general rule is that the job should never execute code for more than one second before calling cancellationToken.ThrowIfCancellationRequested(); again.

ThrowIfCancellationRequested() will throw an OperationCanceledException which the job engine will catch and report that job run as cancelled.

JobResults​

The return type of a job is JobResults which is a class that represents the outcome of a job run. JobResults contains static methods such as Completed() that tells the engine if the job ran successfully.

The Completed() method takes an optional string describing the outcome of the job which is visible under the historical runs of the job in the UI. Something like "Exported 5 orders, 2 of which failed".

JobResults.NothingToDo()​

If your job starts up and you see that there's no work to do, you should return JobResults.NothingToDo() rather than JobResults.Completed(). Returning NothingToDo() signals to the engine that this run isn't interesting, so it won't get saved to the historical runs. This helps to not clutter up the job history in the UI with runs that didn't perform any interesting work.

JobResults.RetryLater()​

Sometimes your job detects that it can't do what it's supposed to do right now. It might be that an external system is down or that the data it needs isn't available right now. For jobs that run less frequently this can be an annoying thing. Lets say that you schedule a job to run every night but external systems might have downtime during the night so now you have to wait a whole day for it to run again.

Instead of waiting that whole day your job can return eg JobResults.RetryLater(TimeSpan.FromMinutes(10)) to tell the engine to schedule an extra run of that job in ten minutes. More on extra runs down below.

JobResults.Failed()​

JobResults contains a method called Failed() which you don't need to call manually. Nexus will catch any exceptions that occur and mark the run as failed. This method is useful if you want to return parameters to the next run of the job. Read more about jobs with parameters here.

JobResults.CompletedWithWarnings()​

Sometimes your job does what it's supposed to do but detects that some things aren't really what they should. Something might be taking a longer time than expected, or some import or exports succeeds but generates warnings. In these cases you can return JobResults.CompletedWithWarnings() and pass a descriptive text of the warnings.

The jobs health checks state will now be Degraded (rather than Healthy or Unhealthy) and will show with an orange warning sign in the Admin UI. You should use standard logging inside your job with additional details about the warnings to be able to resolve it by logging at the job logs.

The recommendation is to use this for things that needs to be looked into but isn't urgent or critical enough to use JobResults.Failed(). Don't use it for things that aren't actionable since it'll only create noise.

Max job duration​

Most jobs have an average expected duration and can sometimes take a much longer time to complete than expected. For some jobs this doesn't matter and the duration can vary a lot. Such as a queue job that sometimes needs to process a lot of messages.

But if you have jobs where you want to be notified if the job runs for too long you can set the max duration either in the Admin UI under the More button in the job details, or by setting MaxDuration on the [Job] attribute like this:

[Job("myjob", MaxDuration = "00:10")]
public class MyJob : IScheduledJob
{
    ...
}

The value should be any string that can be passed to TimeSpan.Parse(). In the above example the max duration is set to 10 minutes.

When a job has been running for a longer time than MaxDuration the Admin UI will show a warning message about the jobs duration, and the health check for the job will report a Degraded health status. Or be cancelled if automatic cancellation is enabled. Read more about automatic cancellation below.

Automatic cancellation​

Sometimes you might have jobs that aren't allowed to run for a longer period of time than the defined max duration. In such cases you can enable automatic cancellation either in the Admin UI under the More button in the job details, or through the job attribute like this:

[Job("myjob", MaxDuration = "00:10", AutoCancel = true)]
public class MyJob : IScheduledJob
{
    ...
}

In the above case the jobs cancellation token will be triggered after ten minutes.

Disabling a job​

A job can be disabled using the Admin UI or the API. This means that the job won't start on it's schedule, or when queue processing is requested. It will only start if you manually start it through the UI or API. This is very useful for when external systems that the job depends on are having issues, or when the job isn't behaving as it should.

In order to not forget about enabling important jobs again you can say that the health check for the job should get a Degraded status when it's disabled:

[Job("myjob", DegradedWhenDisabled = true)]
public class MyJob : IScheduledJob
{
    ...
}

You can also create a new job directly in a disabled state. This can be useful when the job does senstive work where you need to be in control of when it starts. Any new job will have its first run directly when its deployed for the first time which may not be ideal in all cases. You can set the job to be initially disabled like this:

[Job("myjob", InitiallyDisabled = true)]
public class MyJob : IScheduledJob
{
    ...
}

Pausing on error​

Sometimes a job can be very sensitive to errors and in those cases you might want to prevent a job from running again until the error has been resolved. Nexus lets you handle this by setting PauseOnError in the [Job] attribute like this:

[Job("myjob", PauseOnError = true)]
public class MyJob : IScheduledJob
{
    ...
}

Previously this could be done by implementing a bool StopProcessingOnError { get; } on queue jobs. That property was removed in Nexus 3, and Nexus refuses to start if a queue job still has it.

Note that this can be overriden in the Admin UI/API under the More button in the job details page.

How to register jobs​

The job engine discovers concrete IJobBase implementations, including queue jobs and ordinary scheduled jobs implementing the IScheduledJob or IScheduledJob<TParameters> interfaces and automatically register them. This is only done automatically for the entry assembly so if you have multiple .NET projects that contains job classes you need to register the assemblies with the service.

Call AddNexus().AddJobs() to scan additional assemblies for jobs:

Repeated calls share JobsOptions and accumulate discovered jobs. Libraries should append assemblies and GeneratedJobs definitions to the existing lists. Call Build() once after all libraries have contributed: queue jobs then bind to the complete queue set, including queues contributed after discovery. Each generated job's queue must be declared before its AddJobs() call. See library contributions for ordering and conflict rules.

builder.Services.AddNexus().AddJobs(options =>
{
    options.AssembliesToScanForJobs.Add(typeof(MyJob).Assembly);
});

If an assembly contains jobs you do not want to register, use JobFilter:

builder.Services.AddNexus().AddJobs(options =>
{
    options.JobFilter = jobType => jobType != typeof(JobThatIDontWant);
});

Log retention​

By default the job history is stored for 30 days but you can configure for how long job history and logs should be stored like this:

builder.Services.AddNexus().AddJobs(options =>
{
    options.DefaultHistoricalRunsRetention = TimeSpan.FromDays(10);
});

You can also specify this per job with the [Job] attribute like this:

[Job("MyJob", HistoricalRunsRetention = "30.00:00")]
public class MyJob : IScheduledJob
{
    ...
}

The string you set is any string that can be passed to TimeSpan.Parse() or the special string forever to indicate that Nexus should never delete the job history.

Note that this value can be updated in the Admin UI under the More button on the job details page.

Extra job runs​

A job typically starts either by its schedule saying that it should start or by someone manually starting the job. But a job can also have extra runs outside of it's normal schedule.

An extra run can be scheduled either using the API endpoint POST /api/jobs/{jobName}/start?startAt=2022-08-19T12:00:00 (date should be in UTC) or using IJobMetaDataRepository.ScheduleExtraRunAsync(). Extra runs are persisted in the job database and claimed across processing instances. Use them when a start must survive until a processing instance is available, especially for jobs without a schedule:

await jobMetaDataRepository.ScheduleExtraRunAsync(nameof(MyJob), TimeSpan.Zero, JobRunInitiator.Manual);

Starting a job programatically​

Use IJobStartRequester for a fast, best-effort trigger, for example to bring forward the next run of a job that already has a schedule. Use IJobMetaDataRepository.ScheduleExtraRunAsync() or the API start endpoint with an explicit UTC startAt query parameter for a durable extra run. An immediate API start without parameters can use the same best-effort event path.

When you request a start without parameters or a start time, the request is sent as an instance event. Only instances running the continuous job loop handle it. Delivery and exactly one run are not guaranteed: requests may be coalesced, lost while processing instances are unavailable or restarting, or cause successive runs on different instances. A running instance records a request to rerun the job after its current run completes, coalescing further requests received during that run. The normal protection against concurrent runs still applies.

Calls with a non-null parameter object or an explicit delay/start time currently schedule a durable extra run in the job database. Passing null parameters without a start time still uses the best-effort event path. Call the job repository directly when durable scheduling is required so that the intent is explicit.

IJobStartRequester doesn't require the job to be registered in the application that requests the start, and it's available as soon as an instance event broker is configured. An API-only application that doesn't call AddJobs() can request job starts that are handled by the processing instances, as long as they share instance events. Without AddJobs() only starts without parameters and start time are supported, since the other overloads schedule an extra run in the job database. In that case it's up to the requesting application to use the correct job names. RequestJobStartAsync<TJob>() uses the name from [Job("...")] or the class name when the job isn't registered locally, so use the overloads that take a job name for queue jobs that are bound to several queues and for generated jobs.

Job parallelism and mutexes​

The engine guarantees that a single job never executes multiple times in parallel. Even if it's scheduled to run for eg ten times per minute it will only run once per minute if it takes 60 seconds to complete.

Only one instance is allowed to have the job running at any given time. This is done because many jobs communicate with external data sources and having the same job doing that in parallel can often cause bugs. For queue jobs, NumberOfProcessingThreads enables parallelism within the active run. Use a distributed queue processor instead when every instance should process independent messages concurrently.

If you have two or more jobs that you want to ensure never run at the same time you can use the [RequireJobMutex] attribute on your job. Here's an example:

[RequireJobMutex("SomeMutex")]
[RequireJobMutex("SomeOtherMutex")]
public class Job1 : IScheduledJob
{
    // Code omitted for brevity
}

[RequireJobMutex("SomeMutex")]
public class Job2 : IScheduledJob
{
    // Code omitted for brevity
}

[RequireJobMutex("SomeOtherMutex")]
public class Job3 : IScheduledJob
{
    // Code omitted for brevity
}

In this example Job1 can never run at the same time as Job2 and Job3. Job2 can run at the same time as Job3 but not as Job1 since Job2 and Job3 needs different mutexes.

For application resources shared by otherwise independent jobs, you can acquire mutexes dynamically during execution. V3 allows one processing job per concrete queue; use separate named/generated queues when work needs independent filters and processing settings:

public class MyJob(IJobMetaDataRepository jobMetaDataRepository) : IScheduledJob
{
    public string DefaultSchedule => CronSchedule.NotScheduled();

    public async Task<JobResults> ExecuteAsync(CancellationToken cancellationToken)
    {
        await using var _ = await jobMetaDataRepository.AcquireMutexesAsync(["some-mutex-name"], cancellationToken: cancellationToken);

        // The mutex has been acquired

        return JobResults.Completed();
    }
}

The mutexes are decentralized and safe to use with multiple Nexus servers/instances running your jobs. If you're only using a single server you're probably better of using SemaphoreSlim to handle locking.

Job middlewares​

In some cases you want to wrap the execution of a job. You might want to add additional log context properties or instrument the job execution in some way. To achieve this you can register one or more IJobMiddleware instances in the service collection. Eg:

public class MyJobMiddleware(ILogger<MyJobMiddleware> logger) : IJobMiddleware
{
    public async Task<JobResults> ExecuteAsync(JobDescriptor jobDescriptor, Func<Task<JobResults>> executeJob)
    {
        using var logScope = logger.BeginScope(new Dictionary<string, object> {{ "JobName", jobDescriptor.Name }});
        return await executeJob();
    }
}

// Register the middleware
builder.Services.AddScoped<IJobMiddleware, MyJobMiddleware>();

Organizing the Admin UI​

If you have a lot of jobs you can group jobs in the Admin UI by a category, just like you can with queues. Set a category in the Job attribute like this:

[Job("myjob", Category = "My category")]
public class MyJob : IScheduledJob
{
}

Or set the category through the Admin UI under the More button on the job details page.

Job loop delay​

Readiness events can wake the job loop immediately. Queue maintenance runs independently; see queue maintenance for retry/lease timing and cleanup.

By default Nexus will check once per second if there are any new jobs to start since it's not possible to schedule jobs to run more often than that. But in some situations you might want to lower that delay, eg if you use schedule extra runs frequently or if your queue jobs use MaxNumberOfMessagesPerRun and you want the job to restart as fast as possible. In such cases you can configure the job loop delay like this:

builder.Services.AddNexus().AddJobs(options =>
{
    options.JobLoopDelay = TimeSpan.FromSeconds(0.5);
});