Skip to main content

External pending queue storage

The great thing about Nexus is the control and insight you get into queue messages from using an SQL based queue storage. But for some cases the message throughput is just too high for it to make sense to use SQL based queues for everything.

This is where external pending queue storage in Nexus comes in, specifically the IExternalPendingStorage interface. It lets you use Nexus without forcing you to store pending messages in an SQL-backed queue. Pending messages stay in Redis, RabbitMQ, Kafka, or Azure Service Bus. A queue job or distributed queue processor leases them from the processing channel; failures and other non-pending states move into the SQL-backed Nexus queue for inspection and management.

This is very useful for cases like event ingestion where you get tons of messages/events but don't really care about individual messages when they're in the happy path of Pending -> getting processed -> Processed/deleted. But you do care about failures and want the insights you get from using Nexus.

Idempotency, message id, etc​

There are some Nexus features that you may not be able to use with external pending queue storage such as idempotency, queue message history, messages with id getting deduplicated etc. How well these features can be supported depends on the feature set of the external queue storage.

The Redis package supports idempotency, message ids and messages scheduled for future processing, so the Redis package is the best option if you're just looking to maximize message throughput and don't have any requirements on which underlying queue storage to use.

Supported packages​

At the time of writing the Nexus.Redis, Nexus.RabbitMQ, Nexus.Kafka and Nexus.Azure packages offer external pending storage out-of-the-box.

Use the extension method from the provider package, for example:

builder.Services
    .AddNexus()
    .AddRedisExternalPendingStorage(options =>
    {
        options.ConnectionString = redisConnectionString;
    });

And in your queue message set it as external like this:

[Queue("export_order", StorageMode = QueueStorageMode.ExternalPending)]
public class ExportOrderQueueMessage : IQueueMessageWithId
{ }

Readiness bridges and direct producers​

Every external provider registration also starts an IExternalQueueReadinessBridge. The bridge observes a durable notification path, requests activation, and acknowledges that notification only after activation is durable. It does not lease or settle the processing message. A burst of notifications is coalesced by the configured activation store.

Each registered bridge also contributes a Nexus health check named NexusExternalQueueReadiness_{ProviderId}. The check is unhealthy while a consumer has not started or while a provider connection, notification, or acknowledgement operation is failing, and returns to healthy after that queue's consumer succeeds again. Providers with no matching external-pending queues report healthy without opening a consumer.

Producers that call a Nexus IEnqueuer automatically use the correct topology. Direct producers must use the same v3 topology:

  • Redis: store the envelope in nexus:queue:{queue-name} (or the configured prefix) and append the queue name to nexus:queue:{queue-name}:readiness in the same Lua script. Delayed messages go to the :delayed sorted set; the bridge promotes due entries and appends the readiness record atomically. The braces are required so the keys share a Redis Cluster slot.
  • RabbitMQ: publish to the fan-out exchange {ExchangeNamePrefix}{queue-name}. Nexus binds both {QueueNamePrefix}{queue-name} and its configured readiness-suffix queue to that exchange. Do not publish directly to the processing queue.
  • Kafka: publish to {TopicNamePrefix}{queue-name}. The processing and readiness consumers use different consumer groups, so readiness commits never advance processing offsets.
  • Azure Service Bus: publish to the topic {QueueNamePrefix}{queue-name}. Nexus receives from the configured processing and readiness subscriptions independently. Do not publish to a legacy queue entity.

The broker message body/member must use the Nexus external-pending envelope. MessageJson is the serialized queue-message JSON, not an escaped object supplied separately by the broker:

{
  "MessageJson": "{\"OrderId\":\"123\"}",
  "MessageId": "123",
  "Priority": 0
}

For Redis, an immediately ready direct enqueue must add that complete envelope to the ready sorted set and append one readiness record atomically. With the default prefix, all keys use the same Redis Cluster hash tag:

-- KEYS[1] = nexus:queue:{queue-name}
-- KEYS[2] = nexus:queue:{queue-name}:readiness
-- ARGV[1] = priority, ARGV[2] = envelope JSON, ARGV[3] = logical queue name
redis.call('ZADD', KEYS[1], ARGV[1], ARGV[2])
redis.call('XADD', KEYS[2], '*', 'queue', ARGV[3])
return 1

For a batch, add every envelope and append only one readiness record after at least one ready message was added. Delayed envelopes belong in {KeyPrefix}{queue-name}:delayed with their Unix timestamp in seconds as the score; the readiness bridge atomically promotes due values and publishes the readiness record. Direct producers that need Nexus's Redis message-ID replacement or idempotency semantics should use IEnqueuer rather than the minimal script above.

These are breaking topology changes for RabbitMQ and Azure Service Bus, and Redis now uses different physical keys with cluster hash tags. Drain or explicitly migrate old resources before cutover; Nexus database migrations do not move broker messages. Create the v3 topology and switch direct producers before enabling readiness-driven processing. Centralized reconciliation recovers missed readiness notifications, but it cannot make publications to an old physical queue observable by the bridge.

Using multiple different external storages​

Note that you can use both eg Redis and RabbitMQ at the same time on different queues like this:

builder.Services
    .AddNexus()
    .AddRedisExternalPendingStorage(options => options.ConnectionString = redisConnectionString)
    .AddRabbitMqExternalPendingStorage(options => options.ConnectionString = rabbitMqConnectionString);
[Queue("export_order", StorageMode = QueueStorageMode.ExternalPending, ExternalProvider = "redis")]
public class ExportOrderQueueMessage : IQueueMessageWithId
{ }

[Queue("import_order", StorageMode = QueueStorageMode.ExternalPending, ExternalProvider = "rabbitmq")]
public class ImportOrderQueueMessage : IQueueMessageWithId
{ }