5.5.1.1. Streaming

Activity Streaming

Activity streaming lets a long-running data-flow network accept more tokens and steering events while its nodes are still executing. It is an Orleans-only Stateflows extension to Activity execution, not a new UML node type. The same graph, routing rules, and behavior identity are used in both ordinary and streaming execution.

Ordinary run-to-completion processing remains the default. There is no definition-level streaming switch: a client starts or joins a streaming busy period by sending a token-input event with the Stream header, normally through IActivityBehavior.StreamInputAsync.

Use streaming for continuous processing, progress reporting, and externally steered work when weaker durability and isolation are acceptable. Use ordinary processing when the caller must wait for its whole data-flow execution and retain ordinary transaction boundaries.

Ordinary processing and streaming compared

Concern	Ordinary Activity processing	During a streaming busy period
Behavior instance	One behavior instance identified by tenant and ActivityId	The same activation-owned instance, executor, and network
Regular events	Processed in inbox order, one regular transaction at a time	Still processed in inbox order, one regular transaction at a time; stream admission can interleave
Completion of a regular send	Waits for its run-to-completion work	Waits for its transactional work, but not unrelated streamed work or descendants handed off to it
Stream input	Not used unless explicitly requested	Returns after admission, not after token processing
Values	Ordinary successful-call commit and failed-call rollback	Shared live values; no rollback or isolation across sequences of operations
Values and network marking	Stored at ordinary processing boundaries	Stored together at natural quiescence or an explicit checkpoint
Publications	Normally buffered until successful commit	Regular-only branches remain buffered; streamed/mixed branches publish immediately
Failure	Ordinary transaction semantics	Already-applied live effects remain; unrelated admitted work can continue

These are not two independent Activities. A regular event already executing when the first stream input arrives shares its executor and state with the new input. Its pure transactional branches keep their provenance, but its value rollback semantics change to those of the live streaming busy period.

Start or join a stream

The following examples assume that an IActivityBehavior activity has been located. The graph needs input pins, declared with AddInputPin<TToken>, and flows for the supplied token types.

using Stateflows.Activities;
using Stateflows.Common;

await using var watcher = await activity.WatchOutputAsync<ProcessedItem>(
    item => Console.WriteLine(item));

var admission = await activity.StreamInputAsync(input => input
    .Add(new WorkItem("item-42"))
    .AddRange(new WorkItem("item-43"), new WorkItem("item-44")));

if (admission.Status != EventStatus.Accepted)
{
    throw new InvalidOperationException(
        $"Input was not admitted: {admission.Status}.");
}

StreamInputAsync constructs a TokensInput batch, adds the Stream header, and uses send, not request/response. It copies an optional caller-supplied header dictionary rather than changing it. The equivalent low-level operation is:

var admission = await activity.SendAsync(
    new TokensInput().Add(new WorkItem("item-42")),
    new Dictionary<string, EventHeader>
    {
        [nameof(Stateflows.Common.Stream)] = new Stateflows.Common.Stream()
    });

EventStatus.Accepted acknowledges responsibility for the entire batch in the live activation. It does not mean the tokens have finished, the network is idle, output is available, or the batch is durable. Register output/notification watches before sending input that can finish quickly. A client watch is volatile; it is not a durable processing ledger.

The Stream header is valid only for token-input sends to Orleans Activities. Other runtimes, non-Activity behaviors, and non-token payloads reject it. A TokensInput object implements a request interface, but using SendAsync on it is still a send; using RequestAsync is a different operation and is rejected with Stream. Regular token requests, including SendInputAsync, are also rejected while streaming is active. Use streamed token sends, regular token sends, output watches, or later output queries instead.

Initialization and admission limits

In a pin-only input scope, every payload in the batch must match a declared input pin by runtime-type assignability. An unsupported batch returns Rejected before cold initialization or admission; no supported prefix is injected. Typed pins route matching tokens independently and preserve the same streamed provenance. They do not change checkpoint, publication, or regular-event ordering semantics.

For an uninitialized Activity with no typed initializers, its default/implicit initializer must succeed before admission returns Accepted. Initial/Input work belonging to this cold stream admission is streamed. If typed initializers are configured, initialize the Activity explicitly before submitting streamed tokens; cold admission does not guess which typed initializer to use.

StateflowsRuntimeOptions.MaxQueuedActivityStreamEvents bounds queued batches (including batches held for the next checkpoint epoch) plus in-flight batch executions per activation. The default is 128. When full, the runtime rejects the whole new batch with Overloaded; it does not admit a prefix of that batch. A slot is released when that batch's admission execution completes, or when queued work is discarded. This is not necessarily when all business work derived from the batch finishes.

In particular, a later arrival can transfer its tokens into an already active .Interleave() invocation and finish its batch execution while the invocation still has queued or running partitions for those tokens. The admission slot is then available for another batch. MaxQueuedActivityStreamEvents is therefore not an end-to-end interleaved backlog limit.

This is a batch-execution bound, not a bound on tokens, bytes, partitions, retained marking, or buffered output. Large batches, waiting joins, data stores, and interleaved partition queues can still retain substantial memory. Parallel maxConcurrency limits active partitions, not the number waiting. Choose batch sizes and retention policies deliberately; for an end-to-end bound, keep an application-owned limit on outstanding work and release it on an explicit processing-completion signal, not merely on Accepted. Structural boundary output is aggregated until the invocation completes; use explicit notifications or an application-owned completion record when per-item completion is needed.

Orleans resource permits follow the same admitted batch-execution lifetime: they are released on execution completion/discard, which can precede completion of transferred interleaved work. Resource limits are separate from the per-Activity batch bound and do not cap that downstream backlog either. Awaited dependency chains need sufficient capacity or separate resources; see Orleans Runtime.

Stream admission has no durable event receipt. After an uncertain transport failure, blind resubmission can duplicate effects; after a crash, accepted but uncheckpointed work can be lost. Use application-level work identities and idempotent external operations. Accepted alone is not an exactly-once or durable-job guarantee.

Steer a running network

Regular events can execute while streamed work is awaiting asynchronous operations. They enter through their normal accepting nodes; there is no bypass of Activity routing or of the regular inbox.

For example, an accept-event action can update configuration used by processing actions:

public sealed record SetThreshold(decimal Value);

Inside the Activity definition:

.AddInitial(initial => initial.AddControlFlow("Configure"))
.AddAcceptEventAction<SetThreshold>(
    "Configure",
    async context =>
        await context.Behavior.Values.SetAsync(
            "processing.threshold", context.Event.Value))
.AddInputPin<WorkItem>(pin => pin.AddFlow("Process"))
.AddAction("Process", async context =>
{
    var threshold = await context.Behavior.Values.GetOrDefaultAsync(
        "processing.threshold", 10m);
    // Use threshold and the routed WorkItem tokens in application work.
})

The client can then send new SetThreshold(20m) while streaming is live. The accepting node must be active, and the event still waits behind an earlier regular transaction. A node reads the latest live value when its value operation runs; a value already read into a local variable does not change automatically.

Streaming permits overlap; it does not preempt synchronous user code or make every node reentrant. An action blocked without yielding can still delay other work.

Token provenance and transaction handoff

The engine tracks each routed token occurrence as transactional or streamed. This color is execution metadata, not an application payload type.

Source or operation	Provenance
Regular input, Initial work from regular initialization, accepted regular events, and time events	Transactional
Explicit stream input and Initial/Input work from cold stream admission	Streamed
A node consuming only transactional inputs	Transactional
A node consuming any streamed input, including a streamed control input	Streamed, including all outputs
Transactional arrivals at a Data Store during streaming	Converted to streamed
Idle transactional marking after its regular transaction closes	Recolored as streamed

Accept-event and time-event nodes are event sources: their firing is owned by the triggering event, rather than acquiring streamed provenance merely because enabling tokens were streamed. A later consumer can still mix their outputs with streamed input.

Provenance survives guards, transformations, control and zero-weight flows, exception recovery, and nested structured/iterative/parallel execution. Forked routing occurrences can share a logical token identity while keeping independent color; one route does not recolor another route's envelope.

A regular transaction remains open while its own firings, flow pipelines, and enabled transactional successors execute. Tokens waiting for a missing input do not keep it open.

Consider a join receiving a streamed WorkItem and a transactional control token from a regular event:

1. The regular event produces its transactional control token.
2. The join now has all required inputs, including the streamed item.
3. The consumer and its descendants become streamed; the regular transaction hands them off instead of waiting for their completion.

Consequently, a successful regular send is not a promise that all downstream business work has finished. Report that completion explicitly through outputs, notifications, or an external durable domain record.

The same handoff works within nested Activity nodes. The engine records the enclosing structural continuation, finishes the regular call's pure branches, and lets streaming continue the remaining work without repeating structural initialization.

Shared values and publication boundaries

Both transactional and streamed branches can read and write context.Behavior.Values. Individual collection operations are synchronized, but several operations together are not an isolated transaction. Prefer UpdateAsync over separate read/set operations for a single-key read-modify-write:

await context.Behavior.Values.UpdateAsync<int>(
    "processing.count", count => count + 1, defaultValue: 0);

During streaming, even a failed regular event leaves its already-applied value writes visible to other branches. Values are not durably stored with every regular receipt. Values and marking become durable together at a checkpoint or natural quiescence. After the streaming busy period ends, later ordinary calls again have ordinary rollback semantics.

Publications from streamed or mixed branches are immediate, without needing ImmediatePublish. They can reach observers before the corresponding marking or values become durable and cannot be retracted. Publications from regular-only branches remain associated with that regular call's successful commit and are discarded on failure, unless the application explicitly requests immediate publication.

A regular publication commit does not imply that the whole busy network has been checkpointed. External database writes, network calls, and other application effects are not made atomic with a Stateflows checkpoint.

Root output pins publish each token as a raw notification. Their defaults (Retain = false, TTL 0) provide live delivery without notification history or subscription retries, suitable for continuous progress streams. Configure positive TTL to keep bounded history, or Retain = true to keep only the latest value. Dispatch timing and retention are independent; immediate streaming publication does not imply durable history. See notification retention.

Execution context and event metadata

The common injectable IExecutionContext exposes ExecutionMode.Transactional or ExecutionMode.Streamed for the current invocation. A live stream does not make a concurrent pure transactional branch streamed.

For streamed/mixed node firings, ExecutionTrigger, ExecutionTriggerId, and Headers are null. Accepted tokens, potentially from several batches or event sources, enable the firing; the engine does not choose one arbitrary event as their cause. Checkpoint-resumed streamed work also has no event trigger. No synthetic trigger or full event lineage is provided.

Pure transactional invocations retain their event payload, delivery ID, and headers. Actual accept-event/time-event sources remain transactional and expose their triggering event during streaming. Use token APIs for a node's input data and typed event contexts for actual incoming events. Metadata is invocation-scoped and stable across concurrent admissions; CancellationToken remains available in either mode.

ExecutionTriggerId is nullable event-delivery metadata, not a node-firing identity. Existing implementations of IExecutionContext must add the mode property and change their trigger ID to Guid?; consumers must handle null event metadata. See Dependency Injection.

Node concurrency and invocation scope

By default, firings of the same node in the same execution thread scope are serialized. Different eligible nodes and independent nested partitions can execute concurrently. The default per-node gate does not make the entire streaming Activity an ordinary single-call actor turn.

Opt in to bounded overlapping action firings only when the implementation is safe:

.AddAction(
    "Process",
    async context =>
    {
        foreach (var item in context.GetTokensOfType<WorkItem>())
            await processor.ProcessAsync(item, context.CancellationToken);
    },
    node => node.Reentrant())

.Reentrant() sets NodeOptions.Reentrant; it does not allow overlapping regular transactions. Reentrant structured/iterative/parallel nodes are not supported. Each invocation retains its own input snapshot, trigger, cancellation context, and DI scope. Application singleton services, external resources, and mutable objects shared through tokens must still be concurrency-safe. The engine preserves routing metadata, not arbitrary payload-object isolation.

Stream admission, reentrancy, interleaving, and partition concurrency

These controls act at different boundaries; enabling one does not enable the others:

Control	What it controls	What it does not do
StreamInputAsync / Stream header	Admit a token batch into the live Orleans Activity network without waiting for processing completion	Make actions reentrant, make structural arrivals share an invocation, or guarantee durability
Action .Reentrant()	Allow bounded overlapping firings of the same action in one execution thread scope	Merge invocations, overlap regular transactions, or configure parallel partition concurrency
Structural .Interleave()	Let later arrivals join an active structured, iterative, or parallel invocation without repeating initialization	Make inner actions reentrant, publish aggregate boundary output early, or keep a drained invocation open
Parallel maxConcurrency	Bound active partitions per parallel invocation across all its arrivals	Bound queued partitions, admission batches, total Activity concurrency, or CPU threads

On a structured node, .Interleave() feeds the existing nested graph. On an iterative node, it appends sequential partitions; on a parallel node, it appends partitions subject to maxConcurrency. Each arrival is chunked separately and retains its own common inputs. Without .Interleave(), later arrivals do not join the active structural invocation.

.Interleave() is structural-only and cannot be combined with .Reentrant() on the same node. An action inside an interleaved structure may independently opt into reentrancy. Separate parallel partitions already have independent execution thread scopes, so processing one action in several partitions does not itself require .Reentrant().

Interleaving is a graph execution option, not a definition-level streaming switch. A streaming producer still uses the Stream header. Once an invocation drains, later input starts a fresh invocation; arrivals during finalization wait for finalization and completion-output routing. See admission limits for why neither interleaving nor partition concurrency provides end-to-end backpressure.

Quiescence, timers, and lifetime

A streaming busy period ends when no ready, scheduled-to-run, or in-flight network work remains. Idle tokens waiting for missing inputs, active accept-event nodes waiting for an event, and timers scheduled for the future do not prevent quiescence.

Quiescence is not Final Node completion. The Activity can remain initialized with waiting nodes and retained tokens. Orleans persists its values and marking, can release the live session for deactivation, and later resumes ordinary processing until another stream is admitted.

The quiescence/checkpoint commit is an epoch boundary: new inputs arriving during it belong to subsequent work rather than mutating the snapshot being written.

While the stream is live, due time events can steer the same network without waiting for a checkpoint. Their source is transactional. Schedules created during the busy period are volatile until checkpointed; do not assume that a newly scheduled timer survives a crash immediately. After durable persistence, Orleans reminders support later delivery.

Failures, cancellation, and reset

Graph exception handlers remain the mechanism for local recovery. Unhandled streamed invocation failures are logged and publish an immediate ActivityStreamFailure notification:

using Stateflows.Activities.Streams;

await using var failures = await activity.WatchAsync<ActivityStreamFailure>(
    failure => Console.Error.WriteLine(
        $"{failure.BehaviorId}: {failure.EventId} ({failure.EventName}): " +
        string.Join(", ", failure.ExceptionTypes)));

The notification identifies the triggering event and exception types, not a durable per-token failure ledger. The failed invocation is isolated from unrelated admitted work; effects already applied to live values or external systems are not undone.

Operation while streaming	Behavior
FinalizeAsync(FinalizationMode.Queued)	Waits behind earlier regular transactions, then stops admission, discards queued stream batches, cancels active scopes, and drains before finalization
FinalizeAsync(FinalizationMode.Immediate)	Acknowledges the cancellation request with Cancelled before active user code necessarily drains; finalization follows drain
Reset with a Checkpoint header	Drains active bodies to a barrier, clears Activity values and marking, and discards ready successors

Queued finalization does not wait for natural stream exhaustion. It requests cancellation after earlier regular calls finish. Pass context.CancellationToken to cancellable operations; arbitrary non-cooperative code cannot be forcibly stopped.

Late engine values, marking, publications, and outputs from cancelled/replaced work are fenced. Value mutations through a cancelled Activity scope are ignored; pre-cancellation values remain visible to the finalizer. External effects already started, or deliberately ignoring cancellation, cannot be fenced by the engine.

Neither an immediate Cancelled acknowledgement nor a checkpoint is a promise that external side effects have stopped or committed.

Further reading

• Activity Checkpoints: standalone checkpoints, checkpoint headers, timeout, and recovery.
• Activities: graph definition and client APIs.
• Behavior Values: value semantics during a busy period.
• Orleans Runtime: hosting and operational configuration.
• Streaming integration tests: executable examples for barriers, handoff, lifecycle, and reactivation.