Egil.Orleans.Messaging.Streams.EventHubs

Egil.Orleans.Messaging

Composable messaging infrastructure for Microsoft Orleans grains.

Egil.Orleans.Messaging provides building blocks for grains that need durable state changes and durable message handoff to move together:

• IStateManager<T> wraps IPersistentState<T> so a grain does not keep observing uncommitted state after ambiguous write failures.
• Outbox<T> stores messages alongside grain state and assigns durable sender sequence tokens.
• OutboxProcessor<T> dispatches pending outbox items through registered postmen, with retry, reminder forwarding, failure reconciliation, and telemetry.
• MessageTracker records receiver-side high-water marks for outbox messages and Orleans streams.
• StreamManager gives grains a fluent subscription facade with resume-token and handler-error support.

Install

dotnet add package Egil.Orleans.Messaging

Provider-specific integrations are shipped as companion packages:

dotnet add package Egil.Orleans.Messaging.Streams.EventHubs
dotnet add package Egil.Orleans.Messaging.State.AzureStorage

Use the capability namespaces for the tools you need:

using Egil.Orleans.Messaging.Outboxes;
using Egil.Orleans.Messaging.State;
using Egil.Orleans.Messaging.Streams;
using Egil.Orleans.Messaging.Tracking;

Registration extension members live with the Orleans, hosting, and DI types they extend:

using Microsoft.Extensions.DependencyInjection;
using Orleans;
using Orleans.Hosting;

State Manager

Register the default state manager factory on the silo:

siloBuilder.AddDefaultStateManager("state");

For Orleans Azure Table or Blob grain storage, install and configure the Orleans storage provider separately. The Messaging companion works through IPersistentState<T> and Azure SDK exceptions; it does not select or install the underlying provider. Install Egil.Orleans.Messaging.State.AzureStorage and register the Azure-aware factory instead:

siloBuilder.AddAzureStorageStateManager("state");

The Azure-aware manager uses Azure SDK RequestFailedException.Status and ErrorCode values to decide recovery. Optimistic-concurrency and rejected request failures such as HTTP 412, 409, 404, authentication/authorization failures, and payload/validation failures are treated as definite non-persistence, so writes and clears fail fast without an unnecessary recovery read. Ambiguous or transient outcomes, including HTTP 503 ServerBusy, HTTP 500 OperationTimedOut, HTTP 429 throttling, no-response failures, and timeout exceptions, still use read-back recovery.

Then wrap the Orleans persistent state facet during activation:

public sealed class OrderGrain(
    [PersistentState("state", "Default")] IPersistentState<OrderState> storage)
    : Grain, IOrderGrain
{
    private IStateManager<OrderState> state = default!;

    private OrderState CurrentState =>
        state.State ?? throw new InvalidOperationException("Order state is not initialized.");

    public override Task OnActivateAsync(CancellationToken cancellationToken)
    {
        state = this.RegisterStateManager("state", storage);
        return Task.CompletedTask;
    }

    public async Task RenameAsync(string name)
    {
        await state.WriteAsync(CurrentState with { Name = name });
    }
}

State is nullable because persistent storage can expose no value, including after ClearAsync. Initialize missing state or guard it before dereferencing, as CurrentState does above. Later snippets use state.State! only to keep their focus narrow and assume the surrounding grain has already established that invariant.

State types must be reference types and implement IEquatable<T>. For non-trivial state graphs, inherit from VersionedState so the recovery path compares a library-stamped version rather than relying on structural collection equality.

Outbox

Store an Outbox<T> on the grain state and commit messages with the business state change:

[GenerateSerializer]
public sealed record OrderState : VersionedState
{
    [Id(0)] public string? Name { get; init; }

    [Id(1)] public Outbox<IOrderEvent> Outbox { get; init; } =
        Outbox<IOrderEvent>.Create(GrainId.Create("order", "example"));
}

public async Task SubmitAsync()
{
    var next = state.State! with
    {
        Outbox = state.State!.Outbox.Add(new OrderSubmitted())
    };

    await state.WriteAsync(next);
    await outboxProcessor.PostInBackgroundAsync();
}

Add(message) uses system UTC. When a grain uses an injected clock, sample it at the call site and pass the instant with Add(message, timeProvider.GetUtcNow()). The persisted outbox never retains the provider, so serialization and state rehydration need no clock re-registration.

Use OutboxProcessor<T> to dispatch pending items and acknowledge only the items that were posted successfully. Use OutboxMessageEnvelope<T> as the processor item type so the acknowledgement callback can remove exactly the posted items by token:

public sealed class OrderGrain : Grain, IOutboxGrain
{
    private OutboxProcessor<OutboxMessageEnvelope<IOrderEvent>> outboxProcessor = default!;

    public override Task OnActivateAsync(CancellationToken cancellationToken)
    {
        outboxProcessor = this.RegisterOutboxProcessor(new OutboxProcessorOptions<OutboxMessageEnvelope<IOrderEvent>>
        {
            PendingItems = () => [.. state.State!.Outbox],
            AcknowledgePostedAsync = async (items, ct) =>
            {
                await state.WriteAsync(state.State! with
                {
                    Outbox = state.State!.Outbox.RemoveRange(items.Select(item => item.Token))
                });
            },
            ReconcileFailedAsync = (_, _) => ValueTask.CompletedTask,
        })
        .AddPostman<OutboxMessageEnvelope<IOrderEvent>>(
            envelope => PublishSubmittedAsync(envelope.Message));

        return Task.CompletedTask;
    }
}

AcknowledgePostedAsync receives exactly the items that posted successfully — not necessarily a contiguous prefix of the pending list. Different message types dispatch through different postmen concurrently, and an item without a matching postman fails in place while later items can still succeed. Never acknowledge by position (for example outbox.Take(items.Length)); that can remove a failed item and lose it. When the processor item type is a bare message type instead of the envelope, the callback must map each received item back to its token — which is only reliable when payloads are unique — so prefer the envelope form above unless per-message-type postman registration (shown below) is required.

IOutboxGrain forwards reminder ticks to the single attached processor. Register exactly one processor per grain activation; a second registration throws. Add multiple postmen to that processor when item subtypes need different delivery behavior. The grain remains responsible for its own message contracts, posting target, and dead-letter policy. Postman matching is first-match-wins: register specific message types before base interfaces or catch-all handlers.

Failed dispatches are reported through ReconcileFailedAsync. That callback is where the owning grain applies retry, dead-letter, max-depth, or trimming policy, because the grain owns the durable outbox state. The attempt counts passed to the callback are in-memory per activation (and pruned once an item is no longer pending), so policies that must survive activation restarts need to persist their own counters on the items or grain state.

The outbox tools do not require the state manager. When persisting the outbox with plain IPersistentState<T> writes, the pipeline stays at-least-once on its own: items only leave durable state when the grain removes them in AcknowledgePostedAsync after a successful post, so a failed or ambiguous state write leaves them pending and at worst causes duplicate delivery, never loss. Be aware that Outbox<T>.Equals is an O(1) fingerprint (sender, sequence metadata, count, and first/last pending tokens), not deep payload equality — safe for dirty-checks and write recovery, but not a substitute for comparing message contents item by item.

If a post run fails before reconciliation completes — for example when the run exceeds ProcessingTimeout or an acknowledgement callback throws — the processor arms its retry timer and durable reminder before rethrowing, so pending items are retried without requiring another explicit post. Successful posts never pay reminder I/O: PostInBackgroundAsync schedules an in-memory grain timer only, and the durable reminder is registered lazily when a run fails or leaves items pending.

Background outbox postage allows unrelated grain calls to continue while postmen await I/O by default. IPostman<T> services should be state-free with respect to the owning grain. Inline lambda postmen may read activation-local state, but should not write it; durable changes belong in AcknowledgePostedAsync or ReconcileFailedAsync. Postmen run on Orleans' activation scheduler, not on the .NET thread pool. Acknowledgement and failure callbacks are non-interleaving by default, so they do not interleave with normal grain calls unless InterleaveReconciliationCallbacks is enabled. Reentrant grains can still interleave according to Orleans' normal scheduling rules. Pending items in a post run are dispatched concurrently. Successful items are still acknowledged as one ordered batch after all dispatches complete, and failed items are reconciled as one batch.

For reusable delivery code, implement and register keyed postman services:

[OutboxPostman("orders")]
public sealed class OrderEventPostman : IPostman<OrderSubmitted>
{
    public async ValueTask PostAsync(OrderSubmitted message, CancellationToken ct)
    {
        await publisher.PublishAsync(message, ct);
    }
}

services.AddOutboxPostman<OrderEventPostman>();

Then resolve the postman by name from the grain activation service provider:

outboxProcessor = this.RegisterOutboxProcessor(options)
    .AddPostman<OrderSubmitted>("orders");

For common Orleans targets, use the built-in helpers instead of writing the callback by hand:

outboxProcessor = this.RegisterOutboxProcessor(options)
    .AddStreamPostman<OrderSubmitted>(
        "order-streams",
        message => StreamId.Create("submitted-orders", message.OrderId));

outboxProcessor = this.RegisterOutboxProcessor(options)
    .AddGrainPostman<OrderSubmitted, IOrderProjectionGrain>(
        (message, grainFactory) => grainFactory.GetGrain<IOrderProjectionGrain>(message.OrderId),
        (grain, message) => grain.ApplyAsync(message));

Receiver Dedup

MessageTracker accepts a message only when its stream token, stream cursor, or outbox token advances the stored high-water mark:

if (!state.State!.Tracker.ProcessMessage("prices", token, out var tracker))
{
    return;
}

await state.WriteAsync(state.State! with { Tracker = tracker });

Use LatestStreamSequenceToken("prices") when all you need is the previous resume token. Keep using LatestStream("prices") when you need the full cursor or must distinguish "no stream tracked" from "tracked stream with a null token".

The tracker can also evict old sender or stream entries when your retention policy allows it.

Streams

Use StreamManager to configure stream subscriptions from OnActivateAsync. Pass a tracker snapshot when you want persisted resume tokens, or omit it when the grain does not track stream positions:

streamManager = this.RegisterStreamManager(state.State!.Tracker)
    .ConfigureExplicitSubscription<PriceChanged>(
        "StreamProvider",
        "prices",
        async (message, cursor) =>
        {
            if (!state.State!.Tracker.ProcessMessage(cursor, out var tracker))
            {
                return;
            }

            await state.WriteAsync(state.State! with { Tracker = tracker });
        });

await streamManager.EnsureExplicitSubscriptionsAsync(cancellationToken);

The string namespace overload derives a stream id from the complete receiving GrainId, including its grain type and compound-key extension. Publishers must use the same helper with the target grain identity:

var customer = grainFactory.GetGrain<ICustomerGrain>(customerId);
var streamId = StreamManager.CreateStreamId("prices", customer.GetGrainId());
var stream = streamProvider.GetStream<PriceChanged>(streamId);

This convention follows the grain type, so renaming that type changes the derived stream id. Use the StreamId overload for an application-owned id that must survive grain-type changes, or when a custom grain identity cannot round-trip through Orleans' textual GrainId representation:

streamManager = this.RegisterStreamManager(state.State!.Tracker)
    .ConfigureExplicitSubscription<PriceChanged>(
        "StreamProvider",
        StreamId.Create("prices", customerId),
        HandlePriceChangedAsync);

The previous key-only convention is not compatible with these full-identity stream ids. Recreate existing durable subscriptions and update publishers together, or preserve the previous id through the explicit StreamId overload.

Tracked resume tokens are a per-subscription choice. The default is to pass the previous token when a tracker snapshot is supplied. Opt out when a subscription should attach without a resume token:

streamManager = this.RegisterStreamManager(state.State!.Tracker)
    .ConfigureExplicitSubscription<PriceChanged>(
        "StreamProvider",
        "prices",
        HandlePriceChangedAsync,
        useTrackedResumeToken: false);

Orleans 10.3 lets [StatelessWorker] grains consume streams, but such consumers use provider-managed live delivery and reject any non-null resume token. When a stateless worker registers a stream manager with a tracker snapshot, set useTrackedResumeToken: false on its subscriptions, or omit the snapshot, or Orleans throws InvalidOperationException during attach.

this.RegisterStreamManager()
    .ConfigureImplicitSubscription<PriceChanged>(
        "prices",
        async (message, cursor) => await UpdateProjectionAsync(message));

Install Egil.Orleans.Messaging.Streams.EventHubs when using Orleans Event Hubs streams and the enriched adapter/token support:

using Egil.Orleans.Messaging.Streams.EventHubs;
using Orleans.Hosting;

Registering the enriched adapter also registers Event Hubs sequence-token JSON converters, so MessageTracker and StreamCursor can persist and restore EnrichedEventHubSequenceToken without downcasting it to the Orleans base event token:

siloBuilder.AddEventHubStreams("event-hubs", configurator =>
{
    configurator.UseEnrichedDataAdapter();
});

The core package can consume provider-specific token metadata through IStreamSequenceTokenMetadata without taking a direct Event Hubs dependency. Custom stream providers that expose custom StreamSequenceToken types should register a JsonConverter<TToken> with StreamSequenceTokenJsonConverters during startup.

JSON Grain Storage

Outbox<T>, OutboxMessageEnvelope<T>, OutboxSequenceToken, MessageTracker, and StreamCursor carry [JsonConverter] attributes, so they round-trip through any System.Text.Json-based grain storage — including the Orleans 10.3 siloBuilder.UseSystemTextJsonGrainStorageSerializer() — without extra JsonSerializerOptions configuration. Orleans' own System.Text.Json StreamSequenceToken converter only handles EventSequenceToken/EventSequenceTokenV2; tokens stored inside MessageTracker or StreamCursor bypass it and use the StreamSequenceTokenJsonConverters registry instead, so provider tokens such as EnrichedEventHubSequenceToken persist correctly.

Orleans' default Newtonsoft.Json storage serializer is not supported by these converters. All library state types are [GenerateSerializer], so they pass the Orleans 10.3 JSON $type allow-list, but the payload shape is not guaranteed; use a System.Text.Json serializer or the Orleans binary serializer.

Scope

This package is messaging infrastructure, not an event-sourcing or CQRS framework. It wraps Orleans state, outbox dispatch, receiver deduplication, and stream subscription management while leaving domain modeling, read models, transport targets, and operational policy to the application.