Designing an Outbox as an Orleans grain extension

outbox-extension-spike.md

Designing an Outbox as an Orleans grain extension

What the outbox does

Outbox lives on grain state. Same WriteStateAsync persists state + queued items. Processor posts queued items to postmen (stream, gRPC, event hub, other grain). Removes item on ack.

Guarantees:

• At-least-once. Persisted item retried across activations + silo restarts until postman acks.
• In-order, per grain. Insertion order. One postman call at a time. No overlapping post runs per activation.
• No second store. Outbox sits on grain state record. Grain's own WriteStateAsync is the only atomic write.
• Idle-grain liveness. Deactivated grain with pending items → reminder reactivates it. No external poke.
• Stuck-postman containment. Per-post-run timeout cancels hung postmen. One bad downstream can't pin the activation.

Grain author provides via OutboxProcessorOptions:

1. GetPending — snapshot of pending items from state.
2. OnPostCompletedAsync — remove just-posted items from whatever backs GetPending. Persistence timing is the grain's call.
3. OnPostErrorAsync (optional) — receives failed items + their exceptions. Leave them in state to retry; remove them to dead-letter.
4. Postmen — AddPostman<TSub>(...) per outbox item type.

Processor owns: timer + reminder lifecycle, post-run orchestration, postman dispatch, telemetry, default ReceiveReminder.

Proposed API

// The marker interface. Default interface methods on IRemindable hand
// reminders to the processor without the grain author writing ReceiveReminder.
public interface IOutboxGrain : IRemindable
{
    Task IRemindable.ReceiveReminder(string reminderName, TickStatus status)
    {
        // Cast is safe: every Orleans grain implementation (Grain or POCO)
        // implements IGrainBase. See "surprise #1" below for why we cannot
        // declare IOutboxGrain : IGrainBase, IRemindable directly.
        var grainBase = (IGrainBase)this;
        var component = grainBase.GrainContext.GetComponent<IOutboxComponent>();
        return component is null
            ? Task.CompletedTask
            : component.ReceiveReminderAsync(reminderName, status).AsTask();
    }
}

public sealed class OutboxProcessorOptions<TOutbox> where TOutbox : notnull
{
    /// Snapshot of currently pending items. Called once per post run. The sole
    /// source of truth — the processor never caches between post runs.
    public required Func<ImmutableArray<TOutbox>> GetPending { get; init; }

    /// Notification that the supplied items have been posted by every
    /// matching postman. The grain must remove them from whatever backs
    /// GetPending; whether the removal is persisted immediately or batched
    /// with other state changes is up to the grain.
    public required Func<ImmutableArray<TOutbox>, CancellationToken, ValueTask> OnPostCompletedAsync { get; init; }

    /// Optional notification for items whose dispatch failed. The grain
    /// receives each failed item paired with the first exception observed
    /// and decides: leave it in the backing collection to retry on the
    /// next post run, or remove it (dead-letter) when the exception indicates
    /// a permanent failure. If null, failed items remain and are retried
    /// without any grain-side observation.
    public Func<ImmutableArray<(TOutbox Item, Exception Error)>, CancellationToken, ValueTask>? OnPostErrorAsync { get; init; }

    /// Max time spent dispatching on a single post run. Set comfortably below
    /// the grain's response timeout so an in-progress post run never starves
    /// the inbound call that triggered it.
    public TimeSpan ProcessingTimeout { get; init; } = TimeSpan.FromSeconds(20);

    /// Period for both the in-process retry timer and the durable reminder.
    /// Tune to expected postman recovery time. Orleans reminders fire at
    /// most once per minute regardless of value.
    public TimeSpan RetryDelay { get; init; } = TimeSpan.FromMinutes(2);
}

// C# 14 extension members.
public static class OutboxGrainExtensions
{
    extension<TGrain>(TGrain grain) where TGrain : IOutboxGrain, IGrainBase
    {
        public OutboxProcessor<TOutbox> InitializeOutboxProcessor<TOutbox>(
            OutboxProcessorOptions<TOutbox> options)
            where TOutbox : notnull
        {
            ArgumentNullException.ThrowIfNull(options);
            var services = grain.GrainContext.ActivationServices;
            var processor = new OutboxProcessor<TOutbox>(
                grain,
                options,
                services.GetRequiredService<ILoggerFactory>().CreateLogger($"OutboxProcessor<{typeof(TOutbox).Name}>"),
                services.GetService<TimeProvider>() ?? TimeProvider.System,
                services.GetRequiredService<IReminderRegistry>());
            processor.AttachToGrain();    // registers IOutboxComponent on the activation's component bag
            return processor;
        }
    }
}

/// <summary>
/// Owns timer + reminder lifecycle, postmen dispatch, and the post-run
/// orchestration. Created and registered exclusively via
/// <see cref="OutboxGrainExtensions.InitializeOutboxProcessor"/>.
/// </summary>
/// <remarks>
/// Assumes the owning grain uses Orleans' default non-reentrant concurrency
/// model — the runtime serialises turns on the grain's task scheduler, so the
/// processor adds no internal reentrancy/concurrency guards and is not safe
/// on grains marked <c>[Reentrant]</c> or with interleaving methods.
/// </remarks>
public sealed partial class OutboxProcessor<TOutbox> : IOutboxComponent where TOutbox : notnull
{
    /// <summary>
    /// Registers a postman responsible for outbox items assignable to
    /// <typeparamref name="TSub"/>. Matching is first-registered-wins,
    /// evaluated against the item's runtime type (think of it as a
    /// <c>switch</c> on the instance: first arm whose pattern matches handles
    /// the item). A postman registered for a base type or interface therefore
    /// catches all derived/implementing items not already claimed by an
    /// earlier, more specific registration — order them most-specific first.
    /// Registering twice for the same <typeparamref name="TSub"/> throws.
    /// Items whose runtime type matches no postman are reported as failed via
    /// <see cref="OutboxProcessorOptions{TOutbox}.OnPostErrorAsync"/> with a
    /// <see cref="NoPostmanRegisteredException"/> so the grain can dead-letter
    /// them (or leave them in state for a later activation that registers the
    /// missing postman). Returns <c>this</c> for fluent chaining.
    /// </summary>
    public OutboxProcessor<TOutbox> AddPostman<TSub>(Func<TSub, ValueTask> postman) where TSub : TOutbox;

    /// <inheritdoc cref="AddPostman{TSub}(Func{TSub, ValueTask})" />
    public OutboxProcessor<TOutbox> AddPostman<TSub>(Func<TSub, Task> postman) where TSub : TOutbox;

    /// <inheritdoc cref="AddPostman{TSub}(Func{TSub, ValueTask})" />
    /// <remarks>
    /// Cancellation-token overload. The token is cancelled after
    /// <see cref="OutboxProcessorOptions{TOutbox}.ProcessingTimeout"/> elapses;
    /// postmen that honour it short-circuit cleanly and leave the item in the
    /// outbox for the next post run.
    /// </remarks>
    public OutboxProcessor<TOutbox> AddPostman<TSub>(Func<TSub, CancellationToken, Task> postman) where TSub : TOutbox;

    /// <summary>
    /// Posts pending outbox items: dispatches each pending item to its matching postman
    /// and notifies the grain via <c>OnPostCompletedAsync</c> /
    /// <c>OnPostErrorAsync</c>. Safe to call from inside the grain's task
    /// scheduler (e.g. from a handler that just enqueued an item). After a post run, the reminder is unregistered if the outbox is empty, or the
    /// in-process retry timer is (re-)armed if items remain.
    /// </summary>
    /// <remarks>
    /// Per-item postman exceptions are surfaced through <c>OnPostErrorAsync</c>
    /// and do NOT escape. This method only throws <see cref="TimeoutException"/>
    /// (per-post-run <c>ProcessingTimeout</c> elapsed),
    /// <see cref="OperationCanceledException"/> (caller token cancelled), or
    /// whatever the grain's own callbacks throw. The retry timer + reminder
    /// are always armed before the exception leaves.
    /// </remarks>
    public ValueTask PostAsync(CancellationToken cancellationToken = default);

    /// <summary>
    /// Called by the default <c>ReceiveReminder</c> implementation on
    /// <see cref="IOutboxGrain"/> when Orleans' reminder service fires.
    /// No-ops for reminder names the processor did not register.
    /// </summary>
    public ValueTask ReceiveReminderAsync(string reminderName, TickStatus status);
}

internal interface IOutboxComponent
{
    ValueTask ReceiveReminderAsync(string reminderName, TickStatus status);
}

Grain author experience

public sealed class MyGrain : Grain, IMyGrain, IOutboxGrain
{
    private readonly IPersistentState<MyState> state;
    private OutboxProcessor<IMyOutbox> outbox = default!;

    public MyGrain([PersistentState("state")] IPersistentState<MyState> state)
    {
        this.state = state;
    }

    public override Task OnActivateAsync(CancellationToken ct)
    {
        outbox = this.InitializeOutboxProcessor(new OutboxProcessorOptions<IMyOutbox>
        {
            GetPending = () => state.State.Outbox,
            OnPostCompletedAsync = async (posted, token) =>
            {
                state.State = state.State with { Outbox = state.State.Outbox.RemoveRange(posted) };
                await state.WriteStateAsync(token);
            },
        });
        outbox.AddPostman<MyEventA>(evt => myStream.OnNextAsync(evt));
        outbox.AddPostman<MyEventB>(evt => otherGrain.AcceptAsync(evt));
        return Task.CompletedTask;
    }

    // No ReceiveReminder needed. The DIM on IOutboxGrain handles it.

    public async Task EnqueueAsync(MyEventA payload)
    {
        state.State = state.State with { Outbox = state.State.Outbox.Add(payload) };
        await state.WriteStateAsync();        // atomic enqueue
        await outbox.PostAsync();         // best-effort immediate flush
    }
}

Two obligations on the grain author, both enforced by the type system:

1. Implement IOutboxGrain. Compiler-enforced opt-in.
2. Call InitializeOutboxProcessor(...) exactly once in OnActivateAsync. The extension's generic constraint requires IOutboxGrain, so it won't compile on any other type.

The user never inherits a base class, writes ReceiveReminder (unless they have other reminders), manages reminder/timer lifecycle, or writes outbox telemetry.

Escape hatch — grains that have their own reminders

A grain that needs ReceiveReminder for non-outbox work implements it explicitly, which shadows the DIM:

public async Task ReceiveReminder(string name, TickStatus status)
{
    if (name == MyOwnReminder) { await DoMyReminderWork(); return; }
    await outbox.ReceiveReminderAsync(name, status);
}