Distributed queue processors

Distributed queue processors are an opt-in alternative to queue jobs for high-throughput queues. A queue job has one active run across the Nexus cluster. A distributed processor runs a worker pool on every application instance and relies on Nexus's atomic message leases to divide the work safely.

Use a distributed processor when messages can be processed independently and adding application instances should increase throughput. Continue to use a queue job when you need job runs, schedules, parameters, mutexes, progress, or strict single-run behavior.

Registration

Register queues first and jobs before processors if the application uses both:

services
    .AddNexus()
    .AddQueues()
    .AddJobs()
    .AddQueueProcessors()
    .AddSqlServerConnection()
    .Build();

AddQueueProcessors() scans the entry assembly. Additional assemblies and filtering can be configured with QueueProcessorOptions.

Processor bindings are created at Build() after all queue and job contributions. A discovered processor also binds to matching concrete queues contributed by later libraries. Repeated calls share options and do not duplicate bindings. See library contributions.

Each concrete queue can have either one queue job or one distributed processor. Nexus rejects conflicting registrations during startup. In-memory queues are intentionally not supported because their leases cannot be shared across application instances. Database and external-pending queues use their existing atomic lease implementation.

Processing one message at a time

[Queue("order_updates")]
public class OrderUpdate : IQueueMessage
{
    public string OrderId { get; set; } = "";
}

public class OrderUpdateProcessor : IQueueProcessor<OrderUpdate>
{
    public int ConcurrencyPerInstance => 8;
    public TimeSpan MessageLeaseTime => TimeSpan.FromMinutes(5);

    public async Task<ProcessResults> ProcessMessageAsync(
        OrderUpdate message,
        CancellationToken cancellationToken)
    {
        await ApplyUpdateAsync(message, cancellationToken);
        return ProcessResults.Processed();
    }
}

ConcurrencyPerInstance is multiplied by the number of running application instances. For example, eight workers on three instances allows up to 24 leased chunks to make progress concurrently.

A processor instance is created in a fresh dependency-injection scope for each leased chunk. That instance is called sequentially and is never invoked concurrently.

Single-message processors lease one message per worker by default. Override LeaseSize to opt into larger chunks. Batch processors keep a default lease size of 50, capped by BatchSize.

Batch processing

public class SearchIndexProcessor : IBatchQueueProcessor<SearchIndexMessage>
{
    public int ConcurrencyPerInstance => 2;
    public int BatchSize => 100;

    public async Task ProcessMessagesAsync(
        QueueMessageBatch<SearchIndexMessage> batch,
        CancellationToken cancellationToken)
    {
        await IndexAsync(batch.Messages, cancellationToken);
        batch.Processed(batch.Messages);
    }
}

Batch processors use the same result methods as batch queue jobs. Messages without an explicit result are released back to Pending.

Delivery and ordering

Distributed processing is at-least-once. A message can be processed again after a worker crash or an expired lease, so handlers should be idempotent where possible. No ordering is guaranteed across workers or instances.

Set MessageLeaseTime longer than the longest expected processing time for one leased chunk or batch. A lease that expires while processing can allow another instance to receive the same message.

Administration

The new Admin UI shows a distributed processor on its queue rather than in the jobs list. Its Processing tab includes configuration, aggregate cluster activity, and the recent state of each instance. The queue page can pause, resume, or wake the processor.

Pausing is cluster-wide and durable. Workers finish or release their current work and stop taking new leases. Resuming wakes every instance through the normal durable queue-activation mechanism.

Named queues and runtime settings

One processor class can handle several named queues, with independent execution per queue. Use the ProcessMessageAsync overload accepting QueueProcessingContext to read context.QueueName; batches expose batch.QueueName. Worker, lease, and batch settings can be overridden per queue through metadata and the Processing tab in the new Admin UI. Changes finish existing work before replacing the worker pool and apply across all instances. See processing settings for the supported fields, replacement/reset behavior, and API controls.