Write a custom persistent-stream queue adapter

Use a custom queue adapter to connect the Orleans persistent-stream runtime to a queue technology with its own transport and delivery semantics. The adapter translates between Orleans stream batches and the external queue. Orleans supplies the persistent stream provider, pulling agents, subscription routing, queue balancing, and cache management.

When Azure Queue Storage or Azure Event Hubs already provides the required transport behavior, a data adapter customizes its wire format while retaining the built-in queue adapter.

Register an IQueueAdapterFactory with AddPersistentStreams; Orleans creates and hosts PersistentStreamProvider.

Before implementing an adapter, understand the persistent stream pulling architecture. In particular, decide the adapter’s delivery, acknowledgement, ordering, partitioning, and rewind semantics.

Implement the transport boundary

Keep the queue SDK and wire-format logic behind a transport abstraction. The transport must:

• serialize the stream ID, event payloads, and request context into an evolvable envelope;
• assign a monotonically ordered sequence number within each queue partition;
• return only messages from the requested partition;
• acknowledge or delete messages only after Orleans calls the completion method; and
• surface queue failures instead of returning a successful empty read.

public interface ICustomQueueTransport
{
    Task SendAsync(
        QueueId queueId,
        StreamId streamId,
        object[] events,
        Dictionary<string, object>? requestContext,
        CancellationToken cancellationToken);

    Task<IReadOnlyList<CustomQueueMessage>> ReceiveAsync(
        QueueId queueId,
        int maxCount,
        CancellationToken cancellationToken);

    Task CompleteAsync(
        QueueId queueId,
        IReadOnlyList<CustomQueueMessage> messages,
        CancellationToken cancellationToken);
}

The sequence number must remain stable when the queue redelivers a message. The example assigns sequence numbers in the transport, so it rejects caller-supplied sequence tokens when producing events.

Implement the adapter and receiver

IQueueAdapter handles writes and creates one receiver per queue partition. The stream-to-queue mapper used for writes must be the same mapper returned by the factory.

public sealed class CustomQueueAdapter(
    string name,
    ICustomQueueTransport transport,
    IStreamQueueMapper mapper) : IQueueAdapter
{
    public string Name => name;

    public bool IsRewindable => false;

    public StreamProviderDirection Direction => StreamProviderDirection.ReadWrite;

    public Task QueueMessageBatchAsync<T>(
        StreamId streamId,
        IEnumerable<T> events,
        StreamSequenceToken? token,
        Dictionary<string, object>? requestContext)
    {
        if (token is not null)
        {
            throw new ArgumentException(
                "This adapter doesn't support caller-supplied sequence tokens.",
                nameof(token));
        }

        var queueId = mapper.GetQueueForStream(streamId);
        return transport.SendAsync(
            queueId,
            streamId,
            events.Cast<object>().ToArray(),
            requestContext,
            CancellationToken.None);
    }

    public IQueueAdapterReceiver CreateReceiver(QueueId queueId) =>
        new CustomQueueReceiver(queueId, transport);
}

IQueueAdapterReceiver reads queue messages and acknowledges them after every consumer has processed them. If the queue uses visibility leases, renew them while Orleans retains the message and make shutdown cancel outstanding reads.

public sealed class CustomQueueReceiver(
    QueueId queueId,
    ICustomQueueTransport transport) : IQueueAdapterReceiver
{
    public Task Initialize(TimeSpan timeout) => Task.CompletedTask;

    [Obsolete("Use the overload which accepts a CancellationToken.")]
    public Task<IList<IBatchContainer>> GetQueueMessagesAsync(int maxCount) =>
        GetQueueMessagesAsync(maxCount, CancellationToken.None);

    public async Task<IList<IBatchContainer>> GetQueueMessagesAsync(
        int maxCount,
        CancellationToken cancellationToken)
    {
        var messages = await transport.ReceiveAsync(
            queueId,
            maxCount,
            cancellationToken);

        return messages.Cast<IBatchContainer>().ToList();
    }

    [Obsolete("Use the overload which accepts a CancellationToken.")]
    public Task MessagesDeliveredAsync(IList<IBatchContainer> messages) =>
        MessagesDeliveredAsync(messages, CancellationToken.None);

    public Task MessagesDeliveredAsync(
        IList<IBatchContainer> messages,
        CancellationToken cancellationToken) =>
        transport.CompleteAsync(
            queueId,
            messages.Cast<CustomQueueMessage>().ToArray(),
            cancellationToken);

    public Task Shutdown(TimeSpan timeout) => Task.CompletedTask;
}

The batch container restores the stream identity, per-event sequence tokens, and request context when Orleans delivers the batch.

public sealed class CustomQueueMessage(
    StreamId streamId,
    object[] events,
    long sequenceNumber,
    Dictionary<string, object>? requestContext) : IBatchContainer
{
    private readonly EventSequenceToken _sequenceToken = new(sequenceNumber);

    public StreamId StreamId { get; } = streamId;

    public StreamSequenceToken SequenceToken => _sequenceToken;

    public IEnumerable<Tuple<T, StreamSequenceToken>> GetEvents<T>() =>
        events.OfType<T>().Select(
            (item, index) => Tuple.Create<T, StreamSequenceToken>(
                item,
                _sequenceToken.CreateSequenceTokenForEvent(index)));

    public bool ImportRequestContext()
    {
        if (requestContext is null)
        {
            return false;
        }

        RequestContextExtensions.Import(requestContext);
        return true;
    }
}

Implement the factory

The factory composes the adapter with queue mapping, caching, and failure handling. Use named options because one process can register multiple providers with different names.

public sealed class CustomQueueAdapterFactory : IQueueAdapterFactory
{
    private readonly string _name;
    private readonly ICustomQueueTransport _transport;
    private readonly IStreamQueueMapper _mapper;
    private readonly IQueueAdapterCache _cache;

    public CustomQueueAdapterFactory(
        string name,
        ICustomQueueTransport transport,
        HashRingStreamQueueMapperOptions mapperOptions,
        SimpleQueueCacheOptions cacheOptions,
        ILoggerFactory loggerFactory)
    {
        _name = name;
        _transport = transport;
        _mapper = new HashRingBasedStreamQueueMapper(mapperOptions, name);
        _cache = new SimpleQueueAdapterCache(cacheOptions, name, loggerFactory);
    }

    [Obsolete("Use the overload which accepts a CancellationToken.")]
    public Task<IQueueAdapter> CreateAdapter() =>
        CreateAdapter(CancellationToken.None);

    public Task<IQueueAdapter> CreateAdapter(CancellationToken cancellationToken) =>
        Task.FromResult<IQueueAdapter>(
            new CustomQueueAdapter(_name, _transport, _mapper));

    public IQueueAdapterCache GetQueueAdapterCache() => _cache;

    public IStreamQueueMapper GetStreamQueueMapper() => _mapper;

    public Task<IStreamFailureHandler> GetDeliveryFailureHandler(QueueId queueId) =>
        Task.FromResult<IStreamFailureHandler>(
            new NoOpStreamDeliveryFailureHandler());

    public static IQueueAdapterFactory Create(IServiceProvider services, string name) =>
        new CustomQueueAdapterFactory(
            name,
            services.GetRequiredService<ICustomQueueTransport>(),
            services.GetOptionsByName<HashRingStreamQueueMapperOptions>(name),
            services.GetOptionsByName<SimpleQueueCacheOptions>(name),
            services.GetRequiredService<ILoggerFactory>());
}

SimpleQueueAdapterCache is suitable for a non-rewindable adapter whose queue remains the durability boundary. A rewindable adapter usually needs a cache and sequence-token implementation which can position cursors at retained historical messages.

Custom pooled caches use Compare to position cursors against cached messages. The default implementation compares SequenceNumber and EventIndex, preserving the numeric ordering used by existing providers. Override it when the authoritative provider position is encoded in Segment; return an order consistent with the token produced by GetSequenceToken so cache bounds, block selection, and cache-miss detection use the same position contract.

AddPersistentStreams leaves checkpointing to the adapter. The non-rewindable example acknowledges completed messages through its receiver and therefore has no independent checkpoint. For a retained-log transport, implement an IStreamQueueCheckpointerFactory, have the receiver or cache load and update the per-partition position, and register it as a named component with ConfigureComponent. Persist a checkpoint only after all consumers have advanced beyond the corresponding cached messages. A no-op checkpointer is suitable only when replay position is deliberately disposable.

For a durable custom checkpoint backend, implement IStreamCheckpointStore and pass it to StreamQueueCheckpointer with StreamQueueCheckpointerOptions. Load the checkpointer before processing the partition. Each store update receives the expected backend version and returns the resulting StreamCheckpointStoreState, so a version conflict supplies the authoritative checkpoint and version for retry or reconciliation.

StreamQueueCheckpointer limits writes to the configured persistence interval, coalesces pending updates to the latest checkpoint, and flushes the latest position on demand. Set CheckpointComparer when checkpoint values have an ordering contract; the checkpointer then keeps progress monotonic across local updates and concurrent store writers.

Register the provider

Register the transport client in dependency injection, then pass the factory’s Create method to AddPersistentStreams. Configure queue count and cache capacity through the provider configurator.

public static void ConfigureSilo(
    ISiloBuilder builder,
    ICustomQueueTransport transport)
{
    builder.Services.AddSingleton(transport);
    builder.AddPersistentStreams(
        ProviderName,
        CustomQueueAdapterFactory.Create,
        streams =>
        {
            streams.Configure<HashRingStreamQueueMapperOptions>(
                options => options.Configure(
                    value => value.TotalQueueCount = 8));
            streams.Configure<SimpleQueueCacheOptions>(
                options => options.Configure(
                    value => value.CacheSize = 4_096));
        });
}

Register the same provider name and compatible mapping on Orleans clients which directly produce or consume streams:

public static void ConfigureClient(
    IClientBuilder builder,
    ICustomQueueTransport transport)
{
    builder.Services.AddSingleton(transport);
    builder.AddPersistentStreams(
        ProviderName,
        CustomQueueAdapterFactory.Create,
        streams => streams.Configure<HashRingStreamQueueMapperOptions>(
            options => options.Configure(
                value => value.TotalQueueCount = 8)));
}

Keep the provider name and partition count stable. Changing either can map an existing stream to a different queue and strand previously enqueued messages. Configure durable PubSubStore grain storage for explicit subscriptions in production; PubSubStore preserves subscription records independently from queue durability.

Recover persisted token positions during an upgrade

Earlier inherited CreateSequenceTokenForEvent implementations returned an exact EventSequenceToken or EventSequenceTokenV2, including when the batch token had a custom subtype. Current factories preserve the concrete subtype and its metadata while changing the event index.

Orleans recovers these saved positions according to the provider’s token contract:

Token contract	Persisted positions supported during recovery
Generic V1/V2 and custom subclasses explicitly selecting the generic compatibility domain	Exact V1 and V2 positions compare with current tokens by sequence number and event index, with matching equality and hashes across the family.
Event Hubs V1/V2 and custom subclasses explicitly selecting the Event Hubs compatibility domain	Exact V1 positions from the earlier inherited factory are normalized within Event Hubs recovery comparisons. The sequence number and event index identify the position; current delivered tokens retain their Event Hubs offset and custom metadata.
Kinesis	Persisted Kinesis tokens retain the numeric shard offset as the authoritative position, followed by event index, across receiver restarts.
Redis	Persisted Redis tokens retain the entry ID, per-millisecond sequence number, and event index.

Derived tokens are isolated by default, which preserves the identity contract of adapters compiled before this compatibility hook existed. A custom token which uses the complete generic numeric contract overrides SequenceTokenCompatibilityDomain to return typeof(EventSequenceToken). Related token versions select the same stable domain type. A custom token which adds position identity keeps a distinct domain and implements equality, ordering, and hashing consistently for that domain.

Event Hubs, Kinesis, and Redis each retain their provider-specific public equality contract. Event Hubs recovery uses a sequence-only token with an empty offset for a legacy position whose factory omitted the offset; offset-bearing tokens come from the current provider data. Keep stream identity and partition mapping stable when replaying persisted positions.

Validate failure behavior

Test the adapter against the real queue service, including:

1. batches containing multiple event types and request-context values;
2. empty reads, cancellation, transient errors, throttling, and shutdown;
3. producer, receiver, and silo failure before and after acknowledgement;
4. queue ownership moving between silos during membership changes;
5. duplicate delivery and consumer idempotency;
6. stable stream-to-partition mapping across restarts and upgrades;
7. sustained load beyond cache capacity to verify backpressure and queue retention; and
8. sequence-token equality, ordering, and hashing in both comparison directions.

Monitor queue depth and oldest-message age by partition, receive and acknowledgement latency, redelivery count, throttling, pulling-agent errors, and consumer delivery failures. Alert before retention or visibility limits can cause data loss or a redelivery storm.