Streaming

Streaming calls are opt-in - they use the same policy as everything else, through the RunAsync and TryRunAsync overloads that take an IAsyncEnumerable<T> source. Retry, deadlines, attempt ceilings, the classifier, the breaker, and the retry budget all compose; the only thing that changes is what an attempt is.

An attempt over a stream ends at the first element. Before the first element, a stream is indistinguishable from a call: a connection reset, a throttling reply, or a deadline all arrive before anything is yielded, and that window is exactly what the existing machinery classifies. After the first element the call has succeeded - a retry would duplicate or drop work the consumer has already acted on - so the rest of the enumeration passes to the caller untouched.

// The source is cold: each attempt re-invokes it, exactly as the HTTP handler
// builds a fresh request per attempt. Pass the token into whatever you call.
var api = Resilience.Default;

var received = new List<int>();

await foreach (var item in api.RunAsync(ct => streams.Next(ct)))
    received.Add(item);

The overloads are the same two shapes every other execution surface uses, so a static lambda can avoid a closure:

// The static lambda takes the stream as caller state, so it allocates no closure.
await foreach (var item in Resilience.Default.RunAsync(
                   static (source, ct) => source.Next(ct),
                   streams))
{
    Consume(item);
}

What the policy judges

The first element is the one verdict point. It is classified like any result - OnResult<T> works, where T is the element type - and a non-Ok verdict on it is retryable; the consumer never sees that element.

// `OnResult<T>` judges the first element like any result. A verdict the policy will not
// accept is retried, and on the final attempt it throws CallRejectedException from the
// first MoveNextAsync - the consumer never receives an element the classifier refused.
await foreach (var item in api.RunAsync(ct => streams.Next(ct)))
    received.Add(item);

Elements after the first pass through unclassified, because the call already succeeded and re-judging mid-stream data would be a second policy nobody configured.

A stream the policy could not start successfully throws from the first MoveNextAsync. If the attempts run out on an element the classifier kept refusing, the classifier calls a verdict Permanent, or a guard refuses the retry, the consumer receives nothing: an element does not self-describe its failure the way a response with a status code does, so a one-element stream completing normally would be indistinguishable from success. The verdict, the stop reason, and the attempt log travel on the exception instead - CallRejectedException, DeadlineExceededException, AttemptTimeoutException, or whatever the source threw - exactly the exceptions a failed call throws.

Two outcomes are successes without a verdict point:

• An empty source that completes is a success - no element, nothing to judge. The consumer's enumeration yields nothing.
• A caller who stops pulling is the consumer's business, as with any enumerable.

Handle the outcome without exceptions

TryRunAsync is the streaming form of the non-throwing entry point, and it awaits to the first element: by the time it returns, every attempt, every backoff and every guard has already run. That is the only point at which a stream can be asked whether it worked - before it, nothing has been tried; after it, the policy is no longer in the loop.

// TryRunAsync awaits to the first element: by the time it returns, every attempt,
// every backoff and every guard has already run. IsSuccess answers the only question
// a stream can be asked before it is consumed - did the policy get it started?
var result = await api.TryRunAsync(ct => streams.Next(ct));

if (result.TryGetValue(out var stream))
{
    await foreach (var item in stream)
        received.Add(item);
}
else
    refused = result.Reason;

IsSuccess is false for exactly the outcomes a failed RunAsync would have thrown from the first MoveNextAsync: a first element the classifier refused, a guard's rejection, a deadline, an exhausted attempt count, or whatever the source threw on its last attempt. The Reason, the Exception and the attempt log are all on the result.

// The outcomes a failed RunAsync would have thrown from the first MoveNextAsync arrive
// here instead: a refused first element, a guard's rejection, a deadline, an exhausted
// attempt count. There is no value to read and nothing to dispose.
var result = await api.TryRunAsync(ct => streams.Next(ct));

if (!result.IsSuccess)
{
    // StopReason.Permanent here, carrying the CallRejectedException RunAsync would
    // have thrown, and the log of the one attempt a permanent verdict allows.
    stopped = result.Reason;
    failure = result.Exception;
}

A successful result owns a live enumerator, because its first element has already been pulled. Enumerate it once, or dispose it:

// A successful result owns a live enumerator, because the first element has already
// been pulled. Enumerate it once, or dispose it if you decide not to.
var result = await api.TryRunAsync(ct => streams.Next(ct));

if (result.IsSuccess && !stillWanted)
    await ((IAsyncDisposable)result.Value!).DisposeAsync();

This is the one value in the library a caller has to remember to release. It is also the reason the value is enumerable once - the elements behind it are a started enumeration, not a source that can be re-run. To run the policy a second time, call TryRunAsync again.

NOTE

A fault after the first element is not part of the result. It throws from MoveNextAsync under TryRunAsync exactly as it does under RunAsync, because the result was decided before that element existed - see below.

What belongs to the consumer

A fault after the first element propagates out of MoveNextAsync verbatim: unclassified, no event raised, nothing recorded against the breaker or the budget. The call succeeded; what the source does afterward is the consumer's exception, same as any other enumerable.

// A fault after the first element propagates out of MoveNextAsync verbatim:
// unclassified, no event raised, nothing recorded against the breaker. The call
// succeeded; what the source does afterwards is the consumer's exception.
try
{
    await foreach (var item in Resilience.Default.RunAsync(ct => streams.Next(ct)))
        received.Add(item);
}
catch (InvalidOperationException e)
{
    fault = e;
}

NOTE

What the breaker samples is time to the first element. A stream that opens in 2 ms and dies at minute nine is a fast success to the breaker, because the attempt genuinely ended at the first element. If a dependency always fails at element two, the breaker will not notice - see the deep dive for why that is the honest choice rather than an accident.

Attempt ceilings and deadlines

AttemptTimeout bounds time to the first element only. Once the element is in hand the ceiling is disarmed, so a slow middle of a stream never loses the enumeration. Deadline and Backoff work between attempts exactly as for calls, and the first MoveNextAsync throws the exception a failed call would have thrown: DeadlineExceededException, CallRejectedException, AttemptTimeoutException, or the original exception, with the attempt log attached.

Checkpointed resume

A checkpoint is the exception to "a fault after the first element belongs to the consumer": when the caller can say where a stream picks up, retrying it no longer duplicates or drops work. The RunAsync and TryRunAsync overloads that take a checkpoint restart the source from that point instead of stopping.

// The source takes the checkpoint to resume from. Here the checkpoint is the value of
// the last element itself, so a failure right after element 2 re-invokes the source with
// 2, and it resumes at 3 - nothing already delivered is delivered again.
await foreach (var item in api.RunAsync(
                   (from, ct) => events.ReadFrom(from, ct),
                   start: 0,
                   checkpoint: static value => value))
{
    received.Add(item);
}

Three things make this different from the plain overloads above:

• Before the first element, it is the ordinary retry loop. Attempts, Backoff and the classifier apply exactly as they do for any stream; the checkpoint is not needed until something has been delivered.
• After the first element, a failure restarts rather than stops. The source is re-invoked with checkpoint read off the last element this method yielded, up to Resilience.Restarts times - a bound separate from Attempts, because a restart is a fresh attempt sequence with its own.
• A restart is a retry, not a second call nobody is tracking. It goes through the same Budget and the same Breaker, and it is bounded by a Deadline clamped to what is left of the whole operation's own - not a fresh window per restart.

A restart raises CallEventKind.StreamResumed, so a listener can tell "this call needed to reconnect" from the ordinary attempt log. See the events reference.

HttpResilienceOptions.ResumeDownloads is the same idea for a response body a caller reads directly: a stalled download resumes with an HTTP Range request instead of ending the read. See Resuming a stalled download.

What composes, what is refused

Everything composes except hedging. A hedge is a concurrent second copy of a value-returning attempt; two interleaved enumerables are a buffering problem, not a hedge. The streaming overloads refuse a hedged policy at the RunAsync call rather than silently doing nothing. The same policy still runs calls.

// A hedge is a concurrent second copy of a value-returning attempt; two interleaved
// enumerables is a buffering problem, not a hedge. The streaming overloads refuse a
// hedged policy at the RunAsync call, and the same policy still runs calls.
Assert.Throws<ResilienceConfigurationException>(() => hedged.RunAsync<int>(static ct => Empty(ct)));

Admit runs before the first pull and is classified exactly as for calls; OnEvent hears stream attempts like any other; UseAmbientDeadline composes unchanged.

Test a streaming policy

ScriptedStream is to a streaming source what Sequence is to a callback: a script of stream-shaped outcomes served one per attempt, with counters for which attempts started, which were abandoned, and which survived.

// ScriptedStream is to a streaming source what Sequence is to a callback: a
// script of stream-shaped outcomes, served one per attempt. The counters prove
// which attempts started, which were abandoned, and which survived.
var policy = Resilience.Default with
{
    AttemptTimeout = TimeSpan.FromSeconds(1),
    Backoff = Backoff.None,
};

await foreach (var item in policy.RunAsync(ct => streams.Next(ct)))
    received.Add(item);

See the testing reference for the full surface.

gRPC server streaming

A gRPC server-streaming call is this feature with the plumbing already done: AddGrpcResilience() wraps it for you on the same first-message semantic. See gRPC streaming.

Go deeper

The streaming path in the executor deep dive covers the design point that makes streaming different: the surviving attempt's enumerator and token outlive the loop that produced them, and the timer that armed its ceiling is never returned to the pool.