Skip to main content

Entity Framework with Nexus queues

If you're using Entity Framework you might want to sync your SaveChangesAsync() calls with calls to EnqueueAsync() on Nexus queues to ensure that they are committed together.

To achieve this you can install the CommerceMind.Nexus.EntityFramework package and initialize it like this in Program.cs:

builder.Services
    .AddNexus()
    .AddEntityFramework<MyDbContext>();

builder.Services.AddDbContext<MyDbContext>((serviceProvider, options) =>
{
    options.UseNexus(serviceProvider);
});

This will install interceptors with Entity Framework that instead of sending a Nexus message directly to the database will keep it in the background until you call context.SaveChangesAsync() and will then use the same database connection and transaction that EF uses for your other data.

Note that if there's any messages to save Nexus will create an explicit transaction if there isn't already one. Which means that your EF data will be saved in the same transaction as your Nexus messages.

The DbContext and Nexus enqueuer must be resolved from the same dependency injection scope. Nexus resolves the configured context from that scope when a message is enqueued, so each request or unit of work keeps its own context and pending messages. The context must be registered as scoped, must not use DbContext pooling, and must call UseNexus() with the same scoped service provider.

Use SaveChangesAsync() and asynchronous transaction commits while queue operations are pending. Nexus rejects their synchronous counterparts because queue persistence is asynchronous.

Messages that are enqueued but never saved with SaveChangesAsync() (or an asynchronous commit) are discarded together with the context, just like unsaved entity changes. If you need to enqueue without saving the context, see Enqueueing outside an Entity Framework unit of work.

Readiness is requested once per ready queue after a successful commit, even if you call EnqueueAsync() many times or call SaveChangesAsync() several times inside one explicit transaction. Separate commits in the same scope each request readiness independently. In-memory enqueue callbacks finish before readiness is requested, so all committed messages are visible to the consumer.

Failed saves and rolled back transactions​

Pending messages follow the same rules as your other pending changes on the context:

  • If SaveChangesAsync() fails, the messages remain pending together with the entity changes, and are saved by the next successful SaveChangesAsync().
  • If SaveChangesAsync() succeeds inside a transaction that you later roll back, the messages are discarded just like the saved entity changes. They are not saved again by a later SaveChangesAsync().

Enqueueing outside an Entity Framework unit of work​

When an infrastructure entry point owns a standalone enqueue operation and must persist it before returning success, wrap it in an ImmediateQueueOperationScope:

using (new ImmediateQueueOperationScope())
{
    await enqueuer.EnqueueAsync(message);
}

Nexus uses this scope internally for API and MCP enqueue endpoints, incoming broker consumers, and hybrid-storage transitions. Application code should normally let enqueue operations join its scoped DbContext instead.

Which queues join the unit of work​

Messages enqueued to database queues and in-memory queues join the Entity Framework unit of work:

  • Database queue messages are saved in the same transaction as your entity changes.
  • In-memory queue messages are added to the queue after the transaction has been committed. They are discarded together with the entity changes if the transaction is rolled back or the context is never saved.

Messages that leave the application can't be part of a database transaction, so these are sent immediately and are not undone if your Entity Framework transaction is rolled back:

  • Messages enqueued with IApiEnqueuer to Nexus running in another application.
  • Messages enqueued to queues using external pending storage, such as Azure Service Bus, RabbitMQ, Kafka or Redis. This includes messages that are stored in the database until they're ready to be processed, so that the queue behaves the same no matter which status a message is enqueued with.

If such a message must only be sent when your entity changes are saved, use an outbox: enqueue the message to a database queue in the same unit of work, and forward it from there with a queue processor or a queue job. Give the message an id and make it idempotent so that a retried forward doesn't create a duplicate.

Only works with EnqueueAsync()​

Note that this sync with SaveChangesAsync() only works with IEnqueuer.EnqueueAsync() and not other Nexus services like IQueueItemUpdater.

If you need that you can always create an explicit transaction yourself and use DbConnectionScope to make sure that any Nexus operations that happen in that scope use your explicit transaction.