Skip to main content
Table of Contents

Write a Command Handler

Derived page. The behaviour described here is specified by the mediator-dispatch and event-sourcing-store capabilities under openspec/specs/. Those specifications are the source; this page explains and illustrates them. Where a specification and this page disagree, the specification is right and this page is a bug.

A command handler implements ICommandHandler<TCommand> (no result) or IQueryHandler<TCommand, TResult> (the unified interface that also handles ICommand<T>). Stratara discovers them via AddCommandHandlersFromAssemblyContaining<T>().

Choose the command shape

Scenario Marker Routing
Mutation without result, fire-and-forget ICommand ICommandOutboxDispatcher (async via outbox)
Mutation with synchronous result (e.g. setup, "create-and-return-id" flows) ICommand<TResult> IMediator (in-process)
Mutation that destroys infrastructure (drop outbox, recreate write-store) ICommand<TResult> IMediator (must stay in-process)
Mutation that signals process-local state ICommand IMediator (in-process)
Read IQuery<TResult> IMediator

See Routing Conventions for the full table.

Define the command + handler

using Stratara.Abstractions.Mediator;

public sealed record DepositCommand(Guid AccountId, decimal Amount) : ICommand;

public sealed class DepositHandler(IAccountRepository repo) : ICommandHandler<DepositCommand>
{
    public async Task HandleAsync(DepositCommand cmd, CancellationToken ct)
    {
        var account = await repo.GetAsync(cmd.AccountId, ct);
        account.Deposit(cmd.Amount);
        await repo.SaveAsync(account, ct);
    }
}

Register

services.AddCommandHandlersFromAssemblyContaining<DepositHandler>();

That's it — the handler is now resolved per-scope and dispatched whenever mediator.HandleAsync(new DepositCommand(...)) is called.

A handler that appends events needs command auditing

Every recorded event carries the identity of the command that caused it, and the command-audit pipeline is what supplies it. A host whose handlers append events registers it:

services.AddCommandAuditing();

It lives in Stratara.EventSourcing.Pipeline.CommandAudit and AddCommandServices() does not register it for you, because where it sits in the pipeline is yours to decide — a host composing the Orleans execution model keeps the aggregate-grain behaviour innermost, so the order matters.

Without it, SaveChangesAsync refuses the append before anything is written, with a message naming this registration. A flow that appends without a command having preceded it — a timer, a sweep — sets a causation id on the session it builds instead.

Which tenant owns the events you append

A handler that writes events appends them through IEventSource and commits them with SaveChangesAsync. Every event is recorded under the tenant that owns it — and that tenant is not simply the one in the caller's session. The store takes, for each event, the first of these that names a tenant:

  1. the subject you stated for that event with AppendOnBehalfOfAsync, or for the stream's first event with CreateOnBehalfOfAsync;
  2. the owner already resolved for the same stream earlier in the same batch;
  3. the owner recorded on the stream's first event — its tenant, and its user where one was recorded;
  4. the TenantId the event carries itself, when it implements IAggregateCreationEvent;
  5. the tenant in the session, together with the session's data-owner user.

If none of them yields a tenant, the append fails with a message that names the event, the stream, and the three ways to supply an owner: an explicit subject, a creation event, or a session tenant.

The stream comes before the session on purpose. A privileged operator whose session names another tenant cannot silently re-home an existing aggregate by appending to it — the stream keeps the owner it was created with, for every aggregate, whether or not the aggregate exposes a tenant property of its own. The user is part of that owner: a stream whose first event was recorded for a user gives every later event the same user, whichever user the session names, and a stream whose first event names no user is given none. A creation event carries a tenant and no user, so a stream it creates has no user. An aggregate whose events carried different owners could not be fully erased, because each tenant's or user's erasure reaches only its own entries, and once one of those keys was shredded it could not be rehydrated at all.

So a new aggregate states its owner on its first event:

public sealed record AccountOpenedInTenant(Guid AccountId, Guid TenantId, decimal InitialBalance)
    : IAggregateCreationEvent;

public sealed record OpenAccountInTenant(Guid AccountId, Guid TenantId, decimal InitialBalance) : ICommand;

public sealed class OpenAccountInTenantHandler(IEventSource events) : ICommandHandler<OpenAccountInTenant>
{
    public async Task HandleAsync(OpenAccountInTenant cmd, CancellationToken ct)
    {
        await events.CreateAsync<Account>(
            cmd.AccountId,
            new AccountOpenedInTenant(cmd.AccountId, cmd.TenantId, cmd.InitialBalance),
            ct);
        await events.SaveChangesAsync(ct);
    }
}

Every later append to that stream lands in the same tenant, whoever's session makes it.

When the event does not carry the owner — an account an operator opens in a customer's tenant, a platform-owned record created under any session — state it at creation instead:

await events.CreateOnBehalfOfAsync<Account>(
    accountId,
    new AccountOpened(accountId, initialBalance),
    new EventSubject(customerTenantId),
    ct);

CreateOnBehalfOfAsync fails like CreateAsync when the stream exists, and like AppendOnBehalfOfAsync when the subject names no tenant. The stated owner is the stream's owner from then on.

When the session would decide a new stream's owner

The fifth candidate is the one that goes wrong quietly. A first event that states no owner and carries none takes the session's tenant — right when a tenant's user creates a record in their own tenant, wrong when an operator creates one for somebody else, and nothing fails either way. A host can decide:

{
  "EventSourcing": {
    "NewStreamOwnerFromSession": "Warn"
  }
}
Value A new stream's first event that would take its owner from the session
Allow (default) is recorded for the session's tenant, as before
Warn is recorded for the session's tenant, and 102_007 (Warning) names the stream, the event type and the tenant — once per stream and batch, again when a failed save is retried
Refuse fails before anything is staged, naming the stream, the event type and the ways to state an owner

The setting touches nothing else: an owner stated with CreateOnBehalfOfAsync or AppendOnBehalfOfAsync, a creation event that carries a tenant, and every append to a stream that already exists behave as without it. Run Warn in a lower environment until the log stays quiet, then Refuse.

Appending on behalf of another subject

When an event genuinely belongs to someone other than the resolved owner, say so — never obtain it by leaving something out. AppendOnBehalfOfAsync takes an EventSubject (a tenant id and an optional user id) and overrides every other source, for that one event only:

public sealed record CorrectBalance(Guid AccountId, Guid OwningTenantId, decimal Amount) : ICommand;

public sealed class CorrectBalanceHandler(IEventSource events) : ICommandHandler<CorrectBalance>
{
    public async Task HandleAsync(CorrectBalance cmd, CancellationToken ct)
    {
        await events.AppendOnBehalfOfAsync<Account>(
            cmd.AccountId,
            new AmountDeposited(cmd.AccountId, cmd.Amount),
            new EventSubject(cmd.OwningTenantId),
            ct);
        await events.SaveChangesAsync(ct);
    }
}

The actor recorded with the event is still the caller's session; only the owner changes. A subject whose tenant id is empty fails the append at once — the message names the event and the stream, nothing is recorded, and the store does not fall back to the stream, the event or the session, because stating a subject already said that none of them applies.

When two writers race on one stream

SaveChangesAsync writes the whole batch or none of it. If another writer has already written the versions this batch is appending, the save throws ConcurrencyException (Stratara.Abstractions.EventSourcing), whose StreamId and AggregateTypeName name the stream and the aggregate type that lost the race. The staged batch is discarded — every stream in it — so a retry starts from a fresh read of the aggregate, not from the events that just failed.

The exception is the same on every database provider the framework ships a store registration for: a version collision the database refuses through its uniqueness constraint is a concurrency conflict on PostgreSQL and on the SQLite store the test-support package registers alike. A save that fails for any other reason propagates unchanged and is never presented as a conflict. Catch ConcurrencyException and nothing broader: a conflict is worth retrying, anything else is not.

Who retries depends on how the command was dispatched:

  • Through ICommandOutboxDispatcher — nothing to write. The command worker treats the conflict as a retry rather than a failure and redelivers the command, up to MessageRetry:MaxConflictRequeues times (default 100); the handler runs again and re-reads. See When a handler cannot take a message.
  • Through IMediator, in process — the exception reaches the caller, who decides:
app.MapPost("/accounts/{accountId:guid}/deposits", async (Guid accountId, decimal amount, IMediator mediator, CancellationToken ct) =>
{
    try
    {
        await mediator.HandleAsync(new DepositCommand(accountId, amount), ct);
        return Results.NoContent();
    }
    catch (ConcurrencyException conflict)
    {
        return Results.Conflict(new { conflict.StreamId, conflict.AggregateTypeName });
    }
});

Each conflict is also counted, on event_source.append.conflicts, tagged with aggregate.type and bucket.id — the partition the stream fell in — so contention on one aggregate type or one hot partition shows up before it shows up as latency.

A write between your read and your append

AppendAsync numbers the event after whatever the stream holds when you append, not after what you read. A handler that rebuilds an aggregate, decides on it, and then appends therefore does not conflict with a write that landed in between: its event is simply recorded after that write. Most of the time that is harmless. It is not when the write in between changed what the decision rests on — the entry was deleted, the order was cancelled — because the handler's fact is then recorded after a fact that contradicts it. Re-reading just before the append narrows the window; it does not close it, because the save that follows still takes time.

Two ways close it:

  • Name the version you decided on. AppendAtVersionAsync<TAggregate>(streamId, expectedVersion, @event) and AppendRangeAtVersionAsync<TAggregate>(streamId, expectedVersion, events) number the events from expectedVersion + 1. If the stream has moved past it, SaveChangesAsync throws the same ConcurrencyException as above and records nothing. Read the version with GetCurrentVersionAsync before you rebuild the aggregate, so a write in between surfaces as a conflict rather than slipping through. A version the stream has not reached yet is refused at the append with ArgumentOutOfRangeException.

  • Let the host apply it to every read-then-write handler. Set the option, and an append to a stream that the handler rebuilt through IAggregationService.AggregateAsync in the same scope is made on the condition of the version that rebuild saw — without touching a handler:

    {
  "EventSourcing": {
    "AppendAgainstAggregatedVersion": true
  }
}

    AddEventSourcing() reads the section. The condition is the version of the first rebuild of a stream in the scope — a second rebuild of the same stream, by a validator or a helper, does not move it. A rebuild bounded with toVersion sets no condition. A rebuild that found no stream sets the condition that the stream still does not exist, so a handler whose AppendAsync writes the first event conflicts with whoever created it first; CreateAsync refuses an existing stream — or one already staged in the batch — with InvalidOperationException. A successful save ends the condition on the streams it wrote and keeps it on those the handler read but has not written; a failed save ends it everywhere, so a handler that runs again reads again.

What the option costs: a handler that appends to a stream another writer touches concurrently now gets a conflict where it used to succeed. Through ICommandOutboxDispatcher that is a redelivery and a fresh read; through IMediator the caller sees the ConcurrencyException. The option is off by default.

When a save committed but could not publish

SaveChangesAsync commits the events and then hands their bundle on to the outbox. On a host without durable bundles, that handover can fail after the commit, when both the bus and the outbox's own table are unavailable. The save then throws CommittedEventsNotPublishedException, which names the committed streams, whatever ended the handover — a cancellation included. The events are recorded; do not append them again. Readers that consume bundles did not receive them: a projection is repaired by a replay, but a saga that reacts to bundles has missed them, and nothing repairs that — which is why a host that cannot lose a bundle stores bundles with the commit. Such a host does not raise the exception at all: its bundle is recorded in the same transaction as the events, and a handover that fails afterwards loses nothing. Nothing in the framework runs the work again because of it: the RabbitMQ and Azure Service Bus transports acknowledge the message and log an error (108_113); on the Orleans execution model a recorded command is completed (117_127), a store reader counts the entry as applied (117_128) and a durable timer counts as fired (117_129), each logged as an error; and the retrying pipelines do not retry it.

Mandatory hygiene

  • Max 7 constructor parameters (this counts as one). If you need more, group them in a sealed record parameter object.
  • No magic numbers — name your constants.
  • No manual retry loops — pull a Polly pipeline from Stratara.Resilience via ResiliencePipelineProvider<string>.
  • No Stopwatch — use ActivitySource.StartActivity() for timing.

Logging

Source-generated only for new code. Per-package extension class under Diagnostics/Extensions/Logger*Extensions.cs:

public static partial class LoggerAccountExtensions
{
    [LoggerMessage(
        EventId = LogEvents.MyApp.DepositApplied,
        Level = LogLevel.Information,
        Message = "Deposited {Amount} into account {AccountId}.")]
    public static partial void LogDepositApplied(this ILogger logger, decimal amount, Guid accountId);
}

Then in the handler:

logger.LogDepositApplied(cmd.Amount, cmd.AccountId);

Never call logger.LogInformation(...) directly. If you need expensive arguments (string-join, LINQ projection), wrap them in a small struct with ToString() — the source-gen formatter will defer evaluation until the channel is enabled. See Stratara.Shared.Diagnostics.Extensions.DistinctEventTypeNames for the canonical pattern.

Tracing

Every dispatch is wrapped in a span named Handle <RequestType> — you never emit one yourself. The spans come from the Stratara.Application activity source, the same one every other framework trace uses, so a host that subscribes to that source sees each command and query as it passes through the pipeline. Nothing has to be registered for this: AddMediator() and AddAuthorizingMediator<T>() work on their own. A host that registers its own OpenTelemetry Tracer keeps it, and the mediator emits through that instead.

Edit this page