NResilience
Blog
BlogDocs
11 min readPower features

Streams that fail mid-flight

Learn how NResilience handles IAsyncEnumerable sources, protecting the connection start, bounding a quiet stream, and resuming from a checkpoint when you have one.

streamingcancellationcheckpointing

TL;DR: When you run a stream through a resilience policy, "an attempt" ends at the first element. NResilience protects the start of the stream - retrying if the connection fails - but once the first element arrives, the retry loop is over and mid-stream faults are yours to handle, unless you can tell the policy where the stream picked up. Two things survive the handoff: a stream that goes quiet for longer than the AttemptTimeout is cut off rather than left to hang, and a stream you can checkpoint restarts itself from the last element it delivered instead of ending your enumeration.

The Streaming Dilemma

Resilience is easy for a single value. You call a method, you get a result, or you get an exception. You can retry that whole cycle because the operation is atomic.

But a stream - an IAsyncEnumerable<T> - is not atomic. It's a conversation.

If you have a stream that yields 1,000 items, and the connection drops on item 500, what does "retry" even mean? Do you restart the entire stream from item 1? Do you try to resume from item 501? Unless the dependency supports a cursor or an offset, "retrying" a mid-flight failure is impossible for the library to do on your behalf - it has no way to know that offset 501 exists, let alone how to ask for it. Below is what NResilience does with the failure it can handle safely, and then the escape hatch for the one it can't guess: telling it the cursor yourself.

The "First Element" Rule

NResilience handles this by defining the boundary of an attempt as the time-to-first-element.

When you use RunAsync over a stream, the executor treats the process of obtaining the first element as the unit of work.

  1. The Start: The executor calls your source and awaits the first MoveNextAsync().
  2. The Protection: If the stream fails to start (e.g., a SocketException or a deadline), the executor treats it as a failed attempt. It applies the Classifier, checks the RetryBudget, and retries the entire stream start. A failed start throws from the first MoveNextAsync - the same exceptions a failed call throws, attempt log attached.
  3. The Handoff: The moment the first element is successfully yielded, the attempt is marked as Ok. The executor has done its job. It now hands the remaining stream over to you.

Once that first element is in your hand, the resilience loop is closed. If the stream faults on the 10th or 100th element, that exception propagates directly to you.

Why this is the only sane way

You might ask: "Why not protect the whole stream?"

Imagine a policy with 3 retries and a 30-second deadline. If the stream yields 10 elements and then fails, and the library "retried" it, you would receive: Element 1, Element 2 ... Element 10, (Failure), Element 1, Element 2...

This is a nightmare for the consumer. You'd have to implement complex deduplication logic just to handle a retry. By limiting the protection to the start of the stream, NResilience ensures that you never receive duplicate data, and the dependency isn't hammered by repeated requests for the same data.

A fault is yours, silence is not

One thing survives the handoff, and it is worth being precise about, because "the resilience loop is closed" used to mean rather more than it should have.

If the stream faults on element 100, that is yours: the exception propagates, and the library has no honest way to retry it. But if the stream goes quiet on element 100 - connection open, server still holding it, nothing arriving - there is nothing to propagate, and there used to be nothing to stop it either. No deadline covered it, because the attempt ended at element one. You got a stream that never yielded and never threw.

AttemptTimeout now bounds the gap between two elements, and it is on by default. A pull that has been outstanding for longer than that throws an AttemptStalledException at your MoveNextAsync, carrying the bound and how many elements you had received:

await foreach (var trade in policy.RunAsync(ct => feed.SubscribeAsync(ct), token))
{
    Handle(trade); // an AttemptStalledException here means the feed went quiet, not that it broke
}

It bounds the gap rather than the total, so a stream that is slow but alive runs as long as it likes. What gets measured is how long a pending pull has been waiting on the far side, so a consumer that spends a minute processing each element is never blamed for the pause. And it is a bound, not a retry: the handoff already happened, so this ends your enumeration rather than restarting it.

That is still the "First Element" rule. It just no longer means silence is indistinguishable from success.

When you can say where it picked up

Resuming mid-stream - element 101 after a failure at 100 - needs something only you can supply: a cursor, an offset, a continuation token. When you have one, RunAsync and TryRunAsync take an overload for it, and the "First Element" rule above only ever applied to the first segment.

await foreach (var trade in policy.RunAsync(
                   (offset, ct) => feed.SubscribeAsync(fromOffset: offset, ct),
                   start: lastOffset,
                   checkpoint: static trade => trade.Offset + 1))
{
    Handle(trade);
}

A connection reset after offset 500 re-invokes the source with 501, the same way a failed start re-invokes it today - it goes through the same Attempts, the same backoff, the same classifier. What's different is that it can happen after the first element too, because now there is a way to say where "again" means. Nothing already delivered is delivered twice, because you defined both what a checkpoint is and how to read one off an element - the library never has to guess.

Two guardrails come with it. Resilience.Restarts - three by default - caps how many times one call may reconnect, separately from Attempts, so a feed that delivers one trade and then drops forever can't restart forever. And a restart is charged like a retry: it draws on the same RetryBudget, is admitted through the same Breaker, and is bounded by the deadline you set for the whole call, not a fresh window every time the feed reconnects.

The same idea covers an HTTP download you read yourself: HttpResilienceOptions.ResumeDownloads resumes a stalled body with a Range request instead of ending your read, when the server's first response made that safe to do - a strong ETag and Accept-Ranges: bytes, checked back with If-Range so a changed file is never spliced onto what you already had.

Code in Action

Using a resilient stream looks almost identical to a standard call:

var policy = Resilience.Default;

// The source is cold: each attempt re-invokes it, the way the HTTP handler
// builds a fresh request per attempt.
IAsyncEnumerable<Data> stream = policy.RunAsync(ct =>
    client.GetStreamingDataAsync(url, ct), cancellationToken);

try
{
    await foreach (var item in stream)
    {
        // Once we are here, the first element has arrived.
        // The resilience loop is closed.
        Process(item);
    }
}
catch (Exception ex)
{
    // This is either a failure to start (if 0 items were yielded)
    // or a mid-flight fault.
}

Streams without the try/catch

The example above still catches, because a stream that never starts throws from the first MoveNextAsync. If you'd rather branch than catch, TryRunAsync takes a source too, and it answers the only question a stream can be asked before you consume it: did the policy get it started?

var result = await policy.TryRunAsync(ct =>
    client.GetStreamingDataAsync(url, ct), cancellationToken);

if (!result.TryGetValue(out var stream))
{
    // The policy never got it started. Reason, Exception and the attempt log
    // are all on the result - the same failure RunAsync would have thrown.
    _logger.LogWarning("The feed never opened: {Reason}", result.Reason);
    return;
}

await foreach (var item in stream)
{
    Process(item);
}

The await ends at the first element. By the time that line returns, every attempt, every backoff and every guard has already run - which is exactly the window this post is about, expressed as a result instead of an exception. A fault at item 500 still throws from MoveNextAsync, because the result was decided before item 500 existed.

There is one rule that comes with it, and it is the only one like it in the library: a successful result owns a live enumerator. Its first element has already been pulled, so enumerate it once, or dispose it if you change your mind:

if (result.IsSuccess && !stillNeeded)
{
    await ((IAsyncDisposable)result.Value!).DisposeAsync();
}

The alternatives are worse. Reporting success before the first element would report it exactly when a streaming call actually fails, and buffering the element so the result could be safely dropped would charge every stream for the rare caller who abandons one.

Constraints: The Hedge Problem

There is one important limitation: you cannot hedge a stream.

Hedging works by starting a second copy of a call and taking whichever answer arrives first. For a value, this is easy. For a stream, it's impossible without massive buffering. If a hedge "won" on the second attempt, you'd have to discard the first attempt's progress and restart the stream, or buffer the second attempt's items until the first one was cancelled.

If you pass a hedged policy to a streaming RunAsync or TryRunAsync overload, NResilience refuses it at the call with a configuration exception rather than silently doing nothing. The same policy still runs ordinary calls.

Summary

Streaming resilience is about protecting the connection, not the data. Time-to-first-element is what gets retried by default, which keeps your data linear and duplicate-free; the gap between elements is what gets bounded, which is what actually stops a "zombie" connection from hanging your app. If you can hand the policy a checkpoint, a mid-stream fault stops being yours alone to handle - it restarts from where you left off instead. One detail worth knowing for your dashboards: what the breaker samples is time to the first element, so a stream that opens fast and dies later reads as a fast success to the breaker - a stall shows up as nresilience.stalls, and a restart as nresilience.stream_resumed.

When you have multiple tenants sharing a dependency, you need per-tenant isolation. One policy per tenant introduces PolicyScope and how to avoid the noisy neighbor problem.