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

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. Decide produces events and effects; Evolve computes the state from each persisted event. Issuing an invoice needs an answer from a recipient directory, so Decide records that verification started and issues a VerifyRecipient effect. The directory's answer comes back as a command, RecipientVerified or RecipientRejected, and a second arm finishes the job.

using InvoiceContext = DecisionContext<InvoiceState, InvoiceCommand, InvoiceEvent>;

public static class Invoice
{
    public static readonly EntityDefinition<InvoiceState, InvoiceCommand, InvoiceEvent> Definition =
        new(new EntityTypeName("invoices"), _ => new Draft(), Decide, Evolve);

    public static Decision<InvoiceCommand, InvoiceEvent> Decide(
        InvoiceState state,
        InvoiceCommand command,
        InvoiceContext ctx) =>
        (state, command) switch
        {
            // Needs an outside answer: remember that we asked, then ask.
            (Draft, IssueInvoice c) when c.Total > 0 =>
                ctx.Persist(new VerificationStarted(c))
                    .Execute(new VerifyRecipient(c.Recipient)),

            // The answer arrives as a command. The request kept in state fixes the reply type.
            (Verifying v, 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)),
            (Verifying v, RecipientRejected r) =>
                ctx.Persist(new VerificationFailed(r.Reason))
                    .Reply(v.Request, new InvoiceRejected(r.Reason)),
            (_, RecipientVerified or RecipientRejected) => ctx.Ignore(),

            (Issued s, RecordPayment p) when p.Amount == s.Total =>
                ctx.Persist(new PaymentRecorded(p.Amount))
                    .CancelTimer("reminder")
                    .Reply(p, new PaymentAccepted()),

            // A lost reminder mail is acceptable, so the light, unrecorded Run is enough.
            (Issued, SendReminder) =>
                ctx.None().Run<IInvoiceMailer>((mailer, ct) => mailer.RemindAsync(ct)),
            (_, SendReminder) => ctx.Ignore(),

            (_, IssueInvoice c) =>
                ctx.None().Reply(c, new InvoiceRejected("Invalid invoice operation")),
            (_, RecordPayment p) =>
                ctx.None().Reply(p, new InvoiceRejected("Invalid invoice operation")),
        };

    public static InvoiceState Evolve(InvoiceState state, InvoiceEvent @event) =>
        (state, @event) switch
        {
            (Draft, VerificationStarted e) => new Verifying(e.Request),
            (Verifying, VerificationFailed) => new Draft(),
            (Verifying, InvoiceIssued e) => new Issued(e.Total, e.Recipient),
            (Issued s, PaymentRecorded) => new Paid(s.Total),
            _ => throw new InvalidOperationException($"Cannot apply {@event} to {state}."),
        };
}

VerifyRecipient is the effect: plain data, journaled in the same write as VerificationStarted. It says which command Decide gets when the directory gives no answer:

[AkkaSerializable(Manifest = "verify-recipient-v1")]
public sealed record VerifyRecipient([property: AkkaField(0)] string Recipient)
    : IInvoiceProtocol, IEffect<InvoiceCommand>
{
    public InvoiceCommand GaveUp(WorkGaveUp g) =>
        new RecipientRejected($"directory: {g.Reason}");
}

The only impure code is its 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: