Queue processing jobs
There's two interfaces for queue processing jobs included in the SDK. One implementation that process messages one by one and one that process messages in batch.
Accessing the previous message
Queue jobs and distributed processors can both read the current and previously successfully processed version of a message. Such as a message for product information updates where you want to compare the previous and the current message to know what changed. You access it using the IPreviousMessageProvider service from inside your job. Read more here.
Queue jobs run once at a time across the cluster. For worker pools on every instance, see distributed processors. Queue activation explains readiness-driven versus scheduled-only starts.
Processing messages one by one
This is the easiest and safest way to process a message since processing one by one never risks missing a message, it will either complete or fail.
You implement such a job by creating a class that implements IQueueJob<TMessage>. The job executor takes care of leasing messages and all you have to do is implement ProcessMessageAsync():
public class ExampleQueueJob : IQueueJob<ExampleQueueMessage>
{
public async Task<ProcessResults> ProcessMessageAsync(ExampleQueueMessage message, CancellationToken cancellationToken)
{
await DoSomethingInteresting(message);
return ProcessResults.Completed();
}
}
The job will run until there are no more Pending messages in the queue and will then return a short summary of how many messages that got processed and how many of them that failed as the job results.
Note that even though the ProcessResults class has a Failed() method you don't have to catch exceptions yourself. The executor will catch any exceptions that happen in ProcessMessageAsync() and mark that message with the Error status together with the exception details in its processing log.
Updating message during processing
Sometimes you want to store meta data about the processing of a message and use that meta data the next time you process it. An example being syncing entities to another system and then storing the id the entity got in the other system so the next time we process the message we can use that id when communicating with the external system. Or you want to store error details that the next run can use when retrying a message.
This can be achieved by returning an updated message as part of the returned ProcessResult. Eg:
public class ExampleQueueJob(ILogger<ExampleQueueJob> logger) : IQueueJob<ExampleQueueMessage>
{
public async Task<ProcessResults> ProcessMessageAsync(ExampleQueueMessage message, CancellationToken cancellationToken)
{
var id = await SendToExternalSystemAsync(message);
message.IdInExternalSystem = id;
return ProcessResults.Processed(updatedMessage: message);
}
}
Note that if a new version of the message is enqueued the default behavior is to replace the message, which means that your updated message with the id is cleared. You can work around that by using partial messages since that won't replace the message, only replace the properties that the new message contains.
Pausing on error
If you have a queue and a job where the processing of a message is dependent on that previous messages have been successfully processed you can tell Nexus to pause processing when an error occurs. Nexus will by default process messages in the order they are created, but will continue to process the next message if a previous message fails.
To pause processing when an error happens, set PauseOnError to true in the [Job] attribute on your job.
This means that as soon as any message has either Error or AwaitingRetry status in the queue the processing will be paused and won't continue until those messages gets another status.
Use this with caution as you risk pausing processing indefinitely unless you monitor the queue for errors.
[Job(PauseOnError = true)]
public class ExampleQueueJob(ILogger<ExampleQueueJob> logger) : IQueueJob<ExampleQueueMessage>
{
public async Task<ProcessResults> ProcessMessageAsync(ExampleQueueMessage message, CancellationToken cancellationToken)
{
// Important to wrap the entire job in a try-catch with a retry. Otherwise you risk exceptions
// stopping the queue processing until someone notices it.
try
{
await DoSomethingInteresting(message);
return ProcessResults.Processed();
}
catch (Exception e)
{
logger.LogError(e, "An error occured");
return ProcessResults.RetryLater(TimeSpan.FromMinutes(10));
}
}
}
Note that this can be overriden in the Admin UI/API under the More button in the job details page.
Processing messages in batches
The other interface is IBatchQueueJob<TMessage>, where you're passed a batch of messages instead of getting them one by one.
When you're creating a queue processing job that will read data from an external system it's often more efficient to batch load such data rather than requesting it one by one. Such as having a queue for when product data is updated and your job needs to read the full product object.
Processing messages in a batch is a bit harder to do correctly than processing one by one, but the executor will take care of most of the complexity for you.
Here's an example implementation:
public class ExampleQueueJob : IBatchQueueJob<ExampleQueueMessage>
{
public async Task ProcessMessagesAsync(QueueMessageBatch<ExampleQueueMessage> batch, CancellationToken cancellationToken)
{
foreach (var message in batch.Messages)
{
cancellationToken.ThrowIfCancellationRequested();
try
{
await DoSomethingInteresting(message);
batch.Processed(message);
}
catch (Exception e)
{
batch.Failed(message, e);
}
}
}
}
You are responsible for letting the system know which messages you've completed and which failed. If you didn't have the try-catch and there was an exception when processing the fifth message in a batch of ten the system will mark all messages as failed that hasn't explicitly been passed to batch.Processed(message).
Queue jobs have no schedule by default, so EnqueueAsync() is enough to start processing. Implement DefaultSchedule with a valid cron expression when the queue should instead be processed only on that schedule. Clearing or setting the schedule in the API or Admin UI switches between the two behaviors.
For batch workloads, configure ActivationDelay on the queue to collect a short burst before the first run starts:
[Queue(ActivationDelay = "250ms")]
public class ExampleQueueMessage : IQueueMessage
{
}
The activation delay is a fixed collection window. New messages do not keep extending it. It is different from ProcessingDelay, which controls when each message becomes eligible for processing.
Filtering messages
Use generated queues and jobs when a job should process only a filtered subset of a message type. A generated queue job can provide a configured MessageFilter expression. Job parameters do not currently derive or modify this filter. V3 removed the old filtered queue-job interfaces.
Returning a custom status
Sometimes it's not specific enough to say that a queue message was either processed or failed. You might have different actions you need to take for different outcomes that you want to specifically monitor. Nexus allows you to process a message and set it directly in a custom status rather than one of the built in statuses.
In a batched queue job you call the CustomStatus method on the batch:
var processingLog = "... something descriptive of what happened ...";
batch.CustomStatus(message, "MyCustomStatus", processingLog, isError: true);
And in a regular queue job you return ProcessResults.CustomStatus():
var processingLog = "... something descriptive of what happened ...";
return ProcessResults.CustomStatus("MyCustomStatus", processingLog, isError: true);
The isError argument lets you determine if the job should be marked as failed or not because of this custom status.
Note that it's up to you to monitor queues for these custom statuses and take action. If you want to be notified about it you can implement a custom healthcheck that checks if any queue has such a status.
Batch size
By default the batch size of a job is 100. You can control this by implementing the BatchSize property from IBatchQueueJob. Like this:
public class MyBatchJob : IBatchQueueJob<SomeMessage>
{
public int BatchSize => 200;
}
Number of messages per job run
By default a Nexus queue job will run until the queue is empty. Sometimes that can be too aggressive. It might put too much pressure on other systems that the job is communicating with, and digging into the logs of that single run can be hard since it'll contain tons of logs. Nexus lets you set a max number of messages that a single job run can process by implementing the property MaxNumberOfMessagesPerRun in your queue job. Like this:
public class MyQueueJob : IQueueJob<SomeMessage> // Or IBatchQueueJob<SomeMessage>
{
public int MaxNumberOfMessagesPerRun => 1000;
}
Automatic progress reporting
Queue jobs do not report progress automatically by default. This lets processing start with a lease immediately instead of first counting every pending message. You can opt into automatic progress by implementing AutomaticProgressMode:
public class MyQueueJob : IQueueJob<SomeMessage> // Or IBatchQueueJob<SomeMessage>
{
public QueueJobProgressMode AutomaticProgressMode => QueueJobProgressMode.ProcessedCount;
}
The available modes are:
None(the default) does not count pending messages or report automatic progress.ProcessedCountreports how many messages have been handled, without counting the queue first. The Admin UI shows an indeterminate progress bar and the processed count.Percentagecounts pending messages before processing, then reports both the handled count and total. This preserves percentage-based progress at the cost of an additional queue query before every job run.
Progress updates are aggregated per lease or batch instead of being written for every individual message. Code in the job can still inject and use IJobProgress in any mode, including None.
For queues using external pending storage, the count used by Percentage can be approximate and may require an additional broker management operation or permission. The queue can also change after it is counted. Use ProcessedCount when avoiding that overhead is more important than showing a percentage.
Multithreading
The interfaces IBatchQueueJob<TMessage> and IQueueJob<TMessage> have a property called NumberOfProcessingThreads which by default returns 1. If you want your job to process messages in multiple threads in parallel you can implement the method in your job to increase parallelism like this:
public class ExampleQueueJob : IQueueJob<ExampleQueueMessage>
{
public int NumberOfProcessingThreads => 10;
// Other methods and properties excluded for brevity
}
Make sure not to store any instance state that isn't thread safe when you turn on multi threading. The same instance of your job will be used by all threads so if you need to store instance state make sure to use concurrent collections and/or locks.
Controlling the job result message
By default Nexus will create a job result message based on the number of messages processed by the job run. To create your own result message, implement IQueueJobBase.GetResultsText():
public class ExampleQueueJob : IQueueJob<ExampleQueueMessage>
{
public async Task<ProcessResults> ProcessMessageAsync(ExampleQueueMessage message, CancellationToken cancellationToken) => ProcessResults.Completed();
public string GetResultsText(QueueJobProcessResult result)
{
if (result.FailedCount == 0)
{
return null; // returning null means we'll get the default result message
}
return $"ERROR! {result.FailedCount} messages failed";
}
}
Writing your own queue processing
You don't have to use the built-in executor. A custom job can use IQueueReader<TMessage>.LeaseAsync(), IQueueItemUpdater<TMessage>.ProcessedAsync(), and IQueueItemUpdater<TMessage>.FailedAsync() directly.
The job should implement both IScheduledJob (with its own ExecuteAsync and schedule) and IQueueJobBase<TMessage> so it has a custom execution entry point and the UI/API can identify its queue. Implementing only the marker base does not supply an executable job handler.
Scheduling a processed message to be processed again
Sometimes when you're using idempotent messages to update external systems you might have dates in your message that affects the outcome of the processing. Such as a product message which contains a list of prices that have a valid from and to date. At the time of processing you pick the price that is valid right now, but you want to run the processing again when the next price becomes valid.
For a batch processing job you can call:
batch.Processed(message, processAgainAtUtc: DateTime.UtcNow.AddHours(1));
And for a job that process messages one by one you do:
return ProcessResults.Completed(processAgainAtUtc: DateTime.UtcNow.AddHours(1));
This will mark the queue message as Processed together with the date for when it should be processed again. Note that it's not marked as Sleeping like unprocessed, future messages. This to be able to differentiate between messages that have been processed from ones that are just scheduled for processing.
The message status is then changed to Pending again at the date you set, and the queue processing job is called again with the same message.
If a change is enqueued for the same product/message before the date you set the status for the message becomes Pending directly and it's up to the job to recalculate when/if the message should get processed again.
Named queues and processing settings
One queue job implementation can process several named queues, with independent job names, runs, parameters, and state. Use the ProcessMessageAsync overload with QueueProcessingContext or batch.QueueName to identify the queue. The Processing tab in the new Admin UI can override processing threads, lease time, and batch size per queue; active runs retain their snapshot and use updated settings on their next run. See processing settings for supported overrides and reset behavior.
For example, a single-message handler can take the context explicitly:
public Task<ProcessResults> ProcessMessageAsync(
ExampleQueueMessage message, QueueProcessingContext context,
CancellationToken cancellationToken)
{
var queueName = context.QueueName;
return Task.FromResult(ProcessResults.Processed());
}
Existing handlers without the context argument continue to work. A batch belongs to one queue and exposes both batch.QueueName and batch.Context.