Stream with Amazon SQS

The Microsoft.Orleans.Streaming.SQS package connects Orleans persistent streams to Amazon Simple Queue Service. Orleans maps streams across a configurable set of SQS queues, creates a queue when its mapped partition is first used, receives batches through persistent-stream pulling agents, and deletes messages after successful delivery.

SQS streams provide at-least-once delivery. A message becomes visible again when its visibility timeout expires before Orleans acknowledges it, so consumers must handle duplicates. The provider assigns receiver-local sequence tokens as messages arrive and doesn’t support rewind to an earlier token.

Configure a standard queue provider

Install the package and register a named provider on the silo:

siloBuilder
    .AddDynamoDBGrainStorage("PubSubStore", options =>
    {
        options.Service = "us-east-1";
        options.TableName = "OrdersPubSub";
    })
    .AddSqsStreams("Orders", options =>
    {
        options.ConnectionString = "Service=us-east-1";
        options.ReceiveWaitTimeSeconds = 20;
        options.VisibilityTimeoutSeconds = 60;
    });

PubSubStore persists explicit Orleans stream subscriptions. SQS retains event messages independently, while the configured grain storage retains subscription metadata.

Configure each Orleans client which publishes through the provider with the same provider name, connection, and partition count:

clientBuilder.AddSqsStreams("Orders", options =>
{
    options.ConnectionString = "Service=us-east-1";
});

The Service connection value accepts an AWS region such as us-east-1 or an SQS-compatible endpoint such as http://localhost:4566. With a region, the provider uses the AWS SDK for .NET credential resolution chain when the connection string contains no explicit credentials. Prefer workload credentials such as an IAM role. A protected connection string can supply AccessKey, SecretKey, and SessionToken when the deployment requires explicit temporary credentials.

Preserve per-stream order with FIFO queues

Set FifoQueue to use SQS FIFO queues:

siloBuilder.AddSqsStreams("Orders", streams =>
{
    streams.ConfigurePartitioning(16);
    streams.ConfigureSqs(options => options.Configure(sqs =>
    {
        sqs.ConnectionString = "Service=us-east-1";
        sqs.FifoQueue = true;
        sqs.ReceiveWaitTimeSeconds = 20;
        sqs.VisibilityTimeoutSeconds = 60;
    }));
});

The provider appends the .fifo suffix to its queue names and configures each new queue for FIFO throughput. It derives the SQS message group from the complete StreamId, so SQS preserves publication order within one Orleans stream while processing different streams independently. Each publication receives a unique deduplication ID, preserving repeated events with identical payloads.

Received FIFO batches expose SQSFIFOSequenceToken. The token carries the SQS sequence number for same-stream ordering and a receiver-local sequence number for Orleans cache progress. FIFO delivery can still repeat a message after acknowledgement failure or visibility timeout, so processing remains idempotent.

Use the same FifoQueue value and partition count on silos and publishing clients. Changing either value creates a new queue topology and requires an explicit cutover which drains the previous queues.

Use an application wire format

The default SQSDataAdapter serializes Orleans batches into an Orleans-specific message body. Implement ISQSDataAdapter when external applications need to produce or consume the messages, or when the application requires a versioned payload contract.

Register the same adapter on every silo and publishing client:

siloBuilder.AddSqsStreams("Orders", streams =>
{
    streams.ConfigureSqs(options => options.Configure(sqs =>
    {
        sqs.ConnectionString = "Service=us-east-1";
        sqs.ReceiveMessageAttributes = ["SchemaVersion", "ContentType"];
    }));
    streams.UseDataAdapter((services, _) =>
        ActivatorUtilities.CreateInstance<ApplicationSqsDataAdapter>(services));
});

clientBuilder.AddSqsStreams("Orders", streams =>
{
    streams.ConfigureSqs(options => options.Configure(sqs =>
    {
        sqs.ConnectionString = "Service=us-east-1";
    }));
    streams.UseDataAdapter((services, _) =>
        ActivatorUtilities.CreateInstance<ApplicationSqsDataAdapter>(services));
});

The adapter converts between Amazon.SQS.Model.Message and IBatchContainer. On send, the provider uses the adapter’s message body and application-defined message attributes, then supplies the queue URL and FIFO transport fields. On receive, the adapter gets the SQS message and a receiver-local sequence number.

List every application-defined attribute required by the decoder in ReceiveMessageAttributes. List required SQS system attributes in ReceiveMessageSystemAttributes; FIFO mode requests the SQS sequence number automatically. Keep stream identity, event ordering within a batch, schema versioning, and request-context handling stable across producers and consumers. See Customize persistent-stream data formats for wire-contract and rollout guidance.

Tune queue and cache behavior

Setting	Runtime behavior
ConfigurePartitioning	Sets the physical SQS queue count. More queues increase pulling-agent parallelism and create more queues to operate. Keep the value identical on silos and clients.
ConfigureCache	Sets the bounded in-memory cache size used by each pulling agent. Size it for event rate, consumer lag, and available silo memory.
ReceiveWaitTimeSeconds	Enables SQS long polling for receive requests and sets the queue default when Orleans creates the queue. Long polling reduces empty receives and request cost.
VisibilityTimeoutSeconds	Sets the queue visibility timeout when Orleans creates the queue. Choose a value longer than expected delivery and acknowledgement latency; expiration makes the message eligible for redelivery.
ReceiveMessageAttributes	Requests the application-defined attributes consumed by a custom data adapter.
ReceiveMessageSystemAttributes	Requests SQS system attributes consumed by the provider or application adapter.

Queue-creation settings apply when the queue is absent. Manage changes to retention, visibility, encryption, access policy, and dead-letter redrive policy through SQS administration for existing queues.

Serialized Orleans batches must fit within the SQS message quotas. Bound batch size before messages approach the service limit, and include custom envelope and message-attribute overhead in that calculation.

Permissions and operations

The runtime looks up and creates mapped queues, sends and receives messages, and deletes delivered messages in batches. Grant the application the corresponding sqs:GetQueueUrl, sqs:CreateQueue, sqs:SendMessage, sqs:ReceiveMessage, and sqs:DeleteMessage permissions for its queue-name scope. The sqs:DeleteMessage action authorizes both single and batch deletion APIs. Administrative cleanup through DeleteAllUsedQueues also requires sqs:DeleteQueue.

Monitor SQS queue depth, age of the oldest message, receive count, empty receives, deletion failures, and dead-letter movement together with Orleans streaming metrics. Rising oldest-message age indicates that pulling agents or consumers aren’t keeping pace. Repeated receives indicate processing failures, acknowledgement failures, or a visibility timeout shorter than end-to-end delivery latency.