Unit of work

The life of a command

open the transaction (if the provider supports one, through its execution strategy)
 ├─ handler
 ├─ collect the domain events   ← before the save: a removed entity is still tracked
 ├─ SaveChanges                 ← once, after the handler returns
 ├─ dispatch the events         ← inside this same transaction
 │   └─ integration events go to the outbox, if it is turned on
 ├─ SaveChanges for whatever the event handlers wrote
 └─ commit

Any exception at any point undoes everything. The handler has no transaction try/catch, does not call SaveChanges, and does not know a transaction exists.

Events are collected before the save for a concrete reason: once a real DELETE is written, the entity leaves the change tracker and takes its events with it. Collected beforehand, the OrderCancelled of a deleted order reaches the handlers like any other.

Why save once, and not per operation

The obvious alternative — each ICreate saving on its own — is what a naive auto-save layer does. It looks convenient and delivers two bad things: N round trips, and non-atomic writes.

CreateAsync(product)   → INSERT + COMMIT
CreateAsync(history)   → INSERT + COMMIT
CreateAsync(audit)     → fails

Result: product and history written,
audit not. Inconsistent state, with
nothing in the log saying so.

CreateAsync(product)   → only tracks
CreateAsync(history)   → only tracks
CreateAsync(audit)     → fails

Result: rollback. The database is
exactly as it was before the command.
One round trip on the happy path.

That is why SaveBehavior.UnitOfWork is the default. Immediate exists for data migration scripts and console utilities, where there is no pipeline at all — and its documentation says exactly what you give up.

Transactions and the execution strategy

When there is a transaction, it goes through the provider's execution strategy. That matters for two reasons: a transient failure re-runs the whole operation rather than half of it; and EF Core throws if you open a transaction by hand while a retrying strategy is configured.

// This is what the pipeline calls. You can call it too, if you need a
// transactional block outside a command.
var report = await unitOfWork.ExecuteInTransactionAsync(async token =>
{
    await create.CreateAsync(new Snapshot(), token);
    return await BuildReportAsync(token);
}, ct);

It is re-entrant: if a transaction is already open, the operation joins the existing one instead of nesting.

A retry starts from scratch. If the save fails, the added entities are still tracked and the handler would add them again; if the commit fails, they have already been saved and accepted, and the handler would see "nothing to change" and report success without writing. So before each retry the change tracker of every context is cleared, and the handler runs over clean state.

One case is left that no retry solves on its own: the connection drops during the commit, and there is no way to know whether it happened. Register an ICommitVerifier that answers the question — usually by looking in the database for something the command writes — and Nexo uses the answer instead of replaying the command:

public sealed class OrderCommitVerifier(IQueryable<Order> orders, CurrentOrder current)
    : ICommitVerifier
{
    public Task<bool> HasCommittedAsync(CancellationToken ct) =>
        orders.IgnoreQueryFilters().AnyAsync(o => o.ExternalId == current.Id, ct);
}

services.AddScoped<ICommitVerifier, OrderCommitVerifier>();

A command that should not run in a transaction — a long import that writes in batches, say — says so with an attribute. It is still saved at the end; there is just no transaction around it:

[NoTransaction]
public sealed class ImportCatalog : Command<int>
{
    public required string FileName { get; init; }
}

Re-entrancy: events inside commands

A command handler that calls BroadcastAsync sends the message through the pipeline again. Left alone, the second pass would try to open a nested transaction and save halfway through the command.

Nexo keeps, in an AsyncLocal, which unit of work is in flight. It follows the await chain and does not let a change made in a nested call escape back to the caller. A pass through the pipeline that finds its own unit of work already in flight joins it; only the outermost one owns the save and the commit.

Keeping the instance, rather than just "I am inside a command", matters in one case: a handler that creates a fresh DI scope and sends a command through it. That command has a different DbContext, which the outer unit of work would never save — so it gets a unit of work of its own, and saves and commits by itself.

public async Task HandleAsync(PublishCatalog command, CancellationToken ct)
{
    await create.CreateAsync(new CatalogSnapshot(command.Version), ct);

    // Joins this command's unit of work. No new transaction,
    // no half save.
    await mediator.BroadcastAsync(new CatalogPublished(command.Version), ct);
}

Queries never save

The pipeline saves nothing for a query. If a query handler leaves pending changes behind, it logs a warning naming the query — which is nearly always a write that has leaked into the read side.

warn: Nexo.Pipelines.UnitOfWorkPipeline[1004]
      Query SearchProducts left unsaved changes behind; queries are never saved.
      A write has most likely leaked into the read side.

Turn it off with uow.WarnOnPendingQueryChanges = false if the noise is not worth it to you.

The scope is the boundary

A unit of work is scoped: every operation resolved in the same DI scope shares the same session, and that is what makes ICreate<A> and ICreate<B> add up to a single atomic write.

In ASP.NET Core the scope is the request, and there is nothing to do. In a worker or a queue consumer, create the scope yourself:

while (!stoppingToken.IsCancellationRequested)
{
    var message = await queue.ReceiveAsync(stoppingToken);

    // One scope per message: one DbContext, one unit of work, one transaction.
    await using var scope = serviceProvider.CreateAsyncScope();
    var mediator = scope.ServiceProvider.GetRequiredService<IMediator>();

    await mediator.SendAsync(message.ToCommand(), stoppingToken);
}

When you want the control

IUnitOfWork is there for the rare cases: a long command that needs checkpoints, or code outside the mediator.

public interface IUnitOfWork
{
    bool HasChanges { get; }
    bool SupportsTransactions { get; }
    bool HasActiveTransaction { get; }

    Task<int> SaveChangesAsync(CancellationToken ct = default);
    Task<IDataTransaction> BeginTransactionAsync(CancellationToken ct = default);
    Task<TResult> ExecuteInTransactionAsync<TResult>(
        Func<CancellationToken, Task<TResult>> operation,
        CancellationToken ct = default);
}

await using var transaction = await unitOfWork.BeginTransactionAsync(ct);

await create.CreateAsync(new Product("A1", 10m), ct);

// CommitAsync saves what is pending and commits — no need to remember to save first.
await transaction.CommitAsync(ct);

RollbackAsync clears the change tracker too: leaving the changes tracked would let a later save resurrect them. Called inside a command, BeginTransactionAsync returns a transaction that joins the command's: its CommitAsync does not commit the command's work, and leaving the block without committing does not undo it.