Aaron.Akka.Durable.TestKit 0.1.0-alpha.7

Akka.Durable

Akka.Durable gives you event-sourced entities on Akka.NET. Define their behavior with pure Decide and Evolve functions, keep the impure work in effect handlers, then send commands through typed references. The runtime handles persistence, recovery, and routing in local or clustered deployments. Callers can wait for an entity's state, or watch it, without polling.

Inject it. Send a message.

Entity reference factories are .NET DI services. Call AddDurableEntities() at startup, then inject IEntityRefs<InvoiceCommand> into any controller, hosted service, or application service:

public sealed class BillingService(IEntityRefs<InvoiceCommand> invoices)
{
    public Task<IssueReply> IssueAsync(EntityId id, CancellationToken ct) =>
        invoices.Get(id).Ask(new IssueInvoice(250m, "billing@example.com"), ct);

    public void Remind(EntityId id) =>
        invoices.Get(id).Tell(new SendReminder());
}

That's the caller API: choose an ID, send a command, get its declared reply type. The first message creates or recovers the entity. The same calls work with local hosting and Cluster Sharding. Register and host the entity once at startup.

Define its behavior

Here is the Invoice behind those calls. Issuing an invoice runs a VerifyRecipient effect, and its answer comes back as a command. This is the state that handles it:

public static InvoiceDecision Decide(InvoiceState state, InvoiceCommand command, InvoiceContext ctx) =>
    state switch
    {
        // ... Draft: IssueInvoice persists VerificationStarted and issues the VerifyRecipient effect.
        Verifying v => command switch
        {
            // The answer to VerifyRecipient arrives as a command.
            // v.Request is the IssueInvoice we are still answering.
            RecipientVerified =>
                ctx.Persist(new InvoiceIssued(v.Request.Total, v.Request.Recipient, ctx.UtcNow))
                    .Schedule("reminder", TimeSpan.FromDays(30), new SendReminder())
                    .Reply(v.Request, new InvoiceAccepted(ctx.EntityId)),
            RecipientRejected r =>
                ctx.Persist(new VerificationFailed(r.Reason))
                    .Reply(v.Request, new InvoiceRejected(r.Reason)),
            _ => NotHere(command, ctx),   // refused or ignored: it does not belong here
        },
        // ... Issued and Paid have the same shape.
    };

Every state looks like this: see the complete invoice.

VerifyRecipient is the effect: a plain record, journaled in the same write as VerificationStarted. The definition puts the invoice together. Its GaveUp is the command Decide gets when the directory gives no answer:

[AkkaSerializable(Manifest = "verify-recipient-v1")]
public sealed record VerifyRecipient([property: AkkaField(0)] string Recipient) : IInvoiceProtocol;

public static readonly EntityDefinition<InvoiceState, InvoiceCommand, InvoiceEvent> Definition =
    new(new EntityTypeName("invoices"), _ => new Draft(), Decide, Evolve,
        GaveUp: g => new RecipientRejected($"directory: {g.Reason}"));

The only impure code is the effect's handler, which Akka.Durable resolves from a fresh DI scope when the effect runs. The handler's second type argument says what kind of answer it produces, here a command that goes back through Decide:

public sealed class VerifyRecipientHandler(IRecipientDirectory directory)
    : IEffectHandler<VerifyRecipient, InvoiceCommand>
{
    public async Task<InvoiceCommand> HandleAsync(
        VerifyRecipient effect,
        EffectDelivery delivery,
        CancellationToken ct) =>
        await directory.LookupAsync(effect.Recipient, ct) switch
        {
            Known => new RecipientVerified(),
            Unknown u => new RecipientRejected(u.Reason),
        };
}

// at startup, next to the other services
services.AddEffectHandler<VerifyRecipient, InvoiceCommand, VerifyRecipientHandler>();

The state, command, event, and reply declarations are in the guide. You write the domain functions and the handlers; Akka.Durable supplies the actor.

The calling surface is small: IEntityRef sends commands, and IDurableEntityRef adds state reads:

public interface IEntityRef<in TCommand>
{
    EntityId Id { get; }

    void Tell(TCommand command, IActorRef? sender = null);

    Task<TReply> Ask<TReply>(
        IReplies<TCommand, TReply> command,
        CancellationToken cancellationToken = default) where TReply : notnull;
}

public interface IDurableEntityRef<in TCommand, TState, TEvent> : IEntityRef<TCommand>
{
    Task<StateSnapshot<TState>> GetStateAsync(CancellationToken cancellationToken = default);

    Task<StateSnapshot<TState>> WaitForStateAsync(
        Func<TState, bool> until,
        CancellationToken cancellationToken);

    IAsyncEnumerable<StateSnapshot<TState>> WatchStateAsync(
        CancellationToken cancellationToken = default);
}

Ask infers its reply type from the command. GetStateAsync returns the committed state and its journal sequence number. Both reference factories can be injected: use IEntityRefs<InvoiceCommand> for messaging, or IDurableEntityRefs<InvoiceCommand, InvoiceState, InvoiceEvent> for messaging plus state reads. Both are registered by AddDurableEntities(). The registry is keyed by command type, so each entity type must have its own command type.

To react to a change, do not read the state in a loop. Wait for the state, or watch it:

// In a request handler: wait at most five seconds for the payment.
using var limit = CancellationTokenSource.CreateLinkedTokenSource(ct);
limit.CancelAfter(TimeSpan.FromSeconds(5));

// Throws OperationCanceledException at the limit.
var paid = await invoices.Get(id).WaitForStateAsync(s => s is Paid, limit.Token);

// In a status stream: the current state first, then each new one.
await foreach (var snapshot in invoices.Get(id).WatchStateAsync(ct))
    await WriteStatusAsync(snapshot.State, snapshot.SequenceNr, ct);

You get the current state first, then a snapshot after each write that holds events, once its Run effects have finished. Every snapshot carries the journal's sequence number as the version of the state, and a slow reader of the stream gets the newest one. This is a live, best-effort push held in memory: nothing is journaled, a state can be missed while the entity restarts or moves, and a reaction that must happen belongs in an effect. Workflow refs have the same two calls, and WaitForOutcome is built on the wait. See Waiting for a state, and watching it.

You do not need to build an outbox. Akka.Durable has one.

// In Decide:
ctx.Persist(new PaymentRecorded(p.Amount))
    .Execute(new PublishPayment(ctx.EntityId, p.Amount));

services.AddEffectHandler<PublishPayment, PublishPaymentHandler>(
    Retry.UntilAcknowledged(
        TimeSpan.FromSeconds(1),
        TimeSpan.FromMinutes(5)));

public sealed class PublishPaymentHandler(IBus bus)
    : IEffectHandler<PublishPayment>
{
    /* ... */
}

PublishPayment is recorded in the same journal write as PaymentRecorded, and its handler runs at least once, after a crash too, until it returns without throwing; it can run more than once, so it must be safe to repeat.

See Effects for the other retry policies, and for Execute without a policy, which is attempted once and tells Decide how it ended. A workflow's work takes the same policies from the same registration: see Workflows.

Getting started

Install the .NET 11 SDK pinned in global.json, then run:

dotnet restore
dotnet build
dotnet test

The solution targets net11.0 and enables preview C# features (C# 15 unions are used throughout). Package versions are managed centrally; Akka.NET packages are pinned to 1.6.0-beta7 (Aaron.Akka.Reminders to 1.6.0-beta5).

Guide

The rest of the documentation is in docs/, one topic per page. Read it in this order: