Persistent stream pulling architecture

A persistent stream provider connects Orleans streams to a durable queue technology. Producers enqueue through an adapter. Silo-local pulling agents own queue partitions, read batches, cache them, discover subscriptions, and deliver events through ordinary Orleans calls.

This page describes the runtime mechanism. For stream APIs and provider selection, see the streaming documentation.

Rendering diagram.

Provider composition and lifecycle

PersistentStreamProvider is the common implementation. A provider-specific IQueueAdapterFactory creates:

• an IQueueAdapter for enqueue and receive semantics;
• an IStreamQueueMapper for stream-to-queue mapping;
• an IStreamQueueBalancer for silo ownership;
• an IQueueAdapterCache for per-agent caches; and
• optional failure handlers, filters, and backoff providers.

During lifecycle initialization the provider resolves its named adapter factory and creates the adapter. At the active stage it initializes the pulling manager and starts agents. Shutdown stops agents before the provider closes.

By default, pulling agents start automatically. Explicit grain-based and implicit subscriptions are both enabled.

API: PersistentStreamProvider. Implementation: provider lifecycle and provider options.

Queue mapping and ownership

The queue mapper deterministically assigns a stream identity to a queue. All producers and consumers for a provider must use compatible mapping or events can be written to queues which no intended agent reads.

The queue balancer assigns queues to silos and publishes sequenced ownership changes. PersistentStreamPullingManager is a silo-local system target which serializes those notifications, ignores stale sequences, and starts or stops one pulling agent per owned queue. When membership changes, queues move among managers; agents themselves are not virtual and do not migrate.

Source: PersistentStreamPullingManager.

Pulling-agent loop

Each PersistentStreamPullingAgent is a system target with single-threaded Orleans scheduling. Its loop:

1. asks the adapter receiver for a batch;
2. adds batch containers to its queue cache;
3. groups cached items by stream;
4. resolves and caches pub-sub registrations;
5. advances each subscription’s cursor independently;
6. sends events through Orleans messaging;
7. records delivery progress and failure; and
8. purges only data which the cache says is safe to remove.

The default maximum adapter batch-container batch size is 1 and the empty-poll period is 100 ms. These defaults are runtime behavior, not a universal throughput recommendation.

Shutdown and queue handoff

When an agent stops, it closes admission for new background work and stops its polling timer. It waits for receiver initialization, the active queue pump, and accepted producer registrations, subscription handshakes, and deliveries to finish. Accepted work completes its token bookkeeping and releases registration pins and batch protection while the cache and receiver remain available. Outstanding calls retain their existing messaging timeouts and retry limits while accepted work drains.

The agent then reports final delivery progress to the cache, disposes subscription cursors, and shuts down the receiver so provider-specific checkpoint flushing observes the completed progress. Registrations pending when shutdown starts keep the existing checkpoint, since their subscriber positions are still uncertain. Producer unregistration follows receiver cleanup. When the manager reuses an agent for a reassigned queue, initialization waits for that full cleanup and opens admission for the new run.

Explicit subscription notifications receive an immediate acknowledgement while the agent tracks their asynchronous handshake through completion. This lets the subscribing consumer finish its current call and respond to the handshake.

A completed handshake establishes the subscription’s current cursor and replay position. The latest requested handshake owns reconciliation; responses from superseded requests preserve that ownership. Delivery completions and error handling from an older handshake generation release their work while preserving the replacement position, so final checkpoint progress reflects the accepted rewind.

A failed re-handshake leaves the subscription’s position uncertain even when it was previously registered. The agent retains the stream entry across idle cleanup and keeps the existing checkpoint until a successful handshake reconciles that position.

Subscription removal revokes in-flight handshake and delivery ownership. A terminal pub-sub action issued under valid ownership completes cleanup for that subscription identity, including when a cursor reconciliation overlaps its persistence.

Cache and cursor invariants

An IQueueCache decouples queue reads from consumer delivery. Each subscription has an IQueueCacheCursor, so a slow consumer does not directly block a fast consumer at a later cursor.

The cache tracks the earliest delivery progress across active subscriptions. Purging must not remove an item still needed by any cursor. SimpleQueueCache uses pressure buckets to stop or slow reads as lag grows instead of discarding undelivered events. Its default capacity is 4,096 batch containers.

Rendering diagram.

Cache capacity is not durability. The queue remains the durable boundary, subject to the adapter’s acknowledgement contract.

Pub-sub handshake

The agent registers as a producer for each stream and obtains subscription records from stream pub-sub. It holds a pin cursor while subscription handshakes complete so cache cleanup cannot pass the requested start token. New subscription notifications update the agent’s local pub-sub cache.

Sequence tokens allow a rewindable adapter to start from a supported historical position. An adapter whose IQueueAdapter.IsRewindable property is false must reject unsupported tokens rather than pretending to honor them.

Delivery and failure semantics

The agent normally awaits delivery before advancing a subscription cursor, creating per-subscription backpressure. When delivery fails, it invokes the configured IStreamFailureHandler. Depending on provider policy, an explicit subscription can be faulted and removed.

Persistent streams are not universally exactly once. Semantics depend on:

• when the external queue considers a message acknowledged;
• whether the adapter can redeliver after receiver or silo failure;
• cache checkpoint behavior;
• consumer idempotency; and
• provider-specific sequence tokens.

A queue message can be delivered again after ownership change or failure. Consumers which perform durable side effects should be idempotent.

Extension contracts

Provider authors should keep these responsibilities separate:

• IQueueAdapter defines external queue reads/writes and rewindability.
• IQueueAdapterReceiver defines receive, acknowledgement, and shutdown.
• IStreamQueueMapper defines stable partition mapping.
• IStreamQueueBalancer defines cluster ownership.
• IQueueCache and its cursors define buffering and safe purge.
• IStreamFailureHandler defines delivery failure policy.

See provider authoring for hosting and validation patterns and Azure Queue streams for a concrete adapter.