'Stream was already consumed' error using Polly to retry requests in ASP.NET Core

Question

I have an ASP.NET Core 3.1 Web API service that is receiving a web request, doing some manipulations to it, and then passing it on to a backend service and returning the response synchronously. It was working fine but I wanted to introduce some retry logic to those backend requests in case some blips occurred.

I'm using a typed HttpClient and attempting to use Polly to implement the retry logic: https://github.com/App-vNext/Polly/wiki/Polly-and-HttpClientFactory#using-polly-with-ihttpclientfactory

When the backend service works, everything seems to be fine but unfortunately, whenever my backend returns an error like a 500 Internal Server Error, I get the following exception:

Microsoft.AspNetCore.Diagnostics.DeveloperExceptionPageMiddleware: Error: An unhandled exception has occurred while executing the request.

System.Net.Http.HttpRequestException: An error occurred while sending the request.
 ---> System.InvalidOperationException: The stream was already consumed. It cannot be read again.
   at System.Net.Http.StreamContent.PrepareContent()
   at System.Net.Http.StreamContent.SerializeToStreamAsync(Stream stream, TransportContext context, CancellationToken cancellationToken)
   at System.Net.Http.HttpContent.CopyToAsync(Stream stream, TransportContext context, CancellationToken cancellationToken)
   at System.Net.Http.HttpConnection.SendRequestContentAsync(HttpRequestMessage request, HttpContentWriteStream stream, CancellationToken cancellationToken)
   at System.Net.Http.HttpConnection.SendAsyncCore(HttpRequestMessage request, CancellationToken cancellationToken)
   --- End of inner exception stack trace ---
   at System.Net.Http.HttpConnection.SendAsyncCore(HttpRequestMessage request, CancellationToken cancellationToken)
   at System.Net.Http.HttpConnectionPool.SendWithNtConnectionAuthAsync(HttpConnection connection, HttpRequestMessage request, Boolean doRequestAuth, CancellationToken cancellationToken)
   at System.Net.Http.HttpConnectionPool.SendWithRetryAsync(HttpRequestMessage request, Boolean doRequestAuth, CancellationToken cancellationToken)
   at System.Net.Http.RedirectHandler.SendAsync(HttpRequestMessage request, CancellationToken cancellationToken)
   at System.Net.Http.DiagnosticsHandler.SendAsync(HttpRequestMessage request, CancellationToken cancellationToken)
   at Microsoft.Extensions.Http.Logging.LoggingHttpMessageHandler.SendAsync(HttpRequestMessage request, CancellationToken cancellationToken)
   at Polly.Retry.AsyncRetryEngine.ImplementationAsync[TResult](Func`3 action, Context context, CancellationToken cancellationToken, ExceptionPredicates shouldRetryExceptionPredicates, ResultPredicates`1 shouldRetryResultPredicates, Func`5 onRetryAsync, Int32 permittedRetryCount, IEnumerable`1 sleepDurationsEnumerable, Func`4 sleepDurationProvider, Boolean continueOnCapturedContext)
   at Polly.AsyncPolicy`1.ExecuteAsync(Func`3 action, Context context, CancellationToken cancellationToken, Boolean continueOnCapturedContext)
   at Microsoft.Extensions.Http.PolicyHttpMessageHandler.SendAsync(HttpRequestMessage request, CancellationToken cancellationToken)
   at Microsoft.Extensions.Http.Logging.LoggingScopeHttpMessageHandler.SendAsync(HttpRequestMessage request, CancellationToken cancellationToken)
   at System.Net.Http.HttpClient.FinishSendAsyncBuffered(Task`1 sendTask, HttpRequestMessage request, CancellationTokenSource cts, Boolean disposeCts)
   at MyProject.MyController.Routing.HttpMessageRouter.SendRequestAndTrackTiming(Func`1 action, String destinationID) in /mnt/c/Users/myCode/source/repos/MyProject-GitLab/MyController/src/MyController/Routing/HttpMessageRouter.cs:line 59
   at MyProject.MyController.Routing.HttpMessageRouter.SendNewRequest(IMessageWrapper`1 message) in /mnt/c/Users/myCode/source/repos/MyProject-GitLab/MyController/src/MyController/Routing/HttpMessageRouter.cs:line 33
   at MyProject.MyController.Controllers.MyControllerController.Resource(String destinationId) in /mnt/c/Users/myCode/source/repos/MyProject-GitLab/MyController/src/MyController/Controllers/MyControllerController.cs:line 151
   at Microsoft.AspNetCore.Mvc.Infrastructure.ActionMethodExecutor.TaskOfIActionResultExecutor.Execute(IActionResultTypeMapper mapper, ObjectMethodExecutor executor, Object controller, Object[] arguments)
   at Microsoft.AspNetCore.Mvc.Infrastructure.ControllerActionInvoker.<InvokeActionMethodAsync>g__Awaited|12_0(ControllerActionInvoker invoker, ValueTask`1 actionResultValueTask)
   at Microsoft.AspNetCore.Mvc.Infrastructure.ControllerActionInvoker.<InvokeNextActionFilterAsync>g__Awaited|10_0(ControllerActionInvoker invoker, Task lastTask, State next, Scope scope, Object state, Boolean isCompleted)
   at Microsoft.AspNetCore.Mvc.Infrastructure.ControllerActionInvoker.Rethrow(ActionExecutedContextSealed context)
   at Microsoft.AspNetCore.Mvc.Infrastructure.ControllerActionInvoker.Next(State& next, Scope& scope, Object& state, Boolean& isCompleted)
   at Microsoft.AspNetCore.Mvc.Infrastructure.ControllerActionInvoker.<InvokeInnerFilterAsync>g__Awaited|13_0(ControllerActionInvoker invoker, Task lastTask, State next, Scope scope, Object state, Boolean isCompleted)
   at Microsoft.AspNetCore.Mvc.Infrastructure.ResourceInvoker.<InvokeFilterPipelineAsync>g__Awaited|19_0(ResourceInvoker invoker, Task lastTask, State next, Scope scope, Object state, Boolean isCompleted)
   at Microsoft.AspNetCore.Mvc.Infrastructure.ResourceInvoker.<InvokeAsync>g__Logged|17_1(ResourceInvoker invoker)
   at Microsoft.AspNetCore.Routing.EndpointMiddleware.<Invoke>g__AwaitRequestTask|6_0(Endpoint endpoint, Task requestTask, ILogger logger)
   at Microsoft.AspNetCore.Authorization.AuthorizationMiddleware.Invoke(HttpContext context)
   at Microsoft.AspNetCore.Authentication.AuthenticationMiddleware.Invoke(HttpContext context)
   at Prometheus.HttpMetrics.HttpRequestDurationMiddleware.Invoke(HttpContext context)
   at Prometheus.HttpMetrics.HttpRequestCountMiddleware.Invoke(HttpContext context)
   at Prometheus.HttpMetrics.HttpInProgressMiddleware.Invoke(HttpContext context)
   at Microsoft.AspNetCore.Diagnostics.DeveloperExceptionPageMiddleware.Invoke(HttpContext context)

The code I have to send the request is as follows. This first method actually just utilizes a delegate so I can transparently add some metrics gathering around the typed HttpClient call. The typed HttpClient is called RoutingClient in this code:

public async Task<IActionResult> SendNewRequest(IMessageWrapper<HttpRequestMessage> message)
{
    HttpResponseMessage destinationResponse = await SendRequestAndTrackTiming(() => _client.SendAsync(message.Message), message.DestinationID);
    return CreateSerializeableResponseMessage(destinationResponse);
}

private ResponseMessageResult CreateSerializeableResponseMessage(HttpResponseMessage httpResponse)
{
    ResponseMessageResult responseMessage = new ResponseMessageResult(httpResponse);
    IOutputFormatter[] formattersList = { new HttpResponseMessageOutputFormatter() };
    FormatterCollection<IOutputFormatter> formattersCollection = new FormatterCollection<IOutputFormatter>(formattersList);
    responseMessage.Formatters = formattersCollection;
    return responseMessage;
}

private async Task<HttpResponseMessage> SendRequestAndTrackTiming(Func<Task<HttpResponseMessage>> action, string destinationID)
{
    HttpResponseMessage response = null;

    Stopwatch stopwatch = new Stopwatch();
    try
    {
        stopwatch.Start();
        response = await action();
        return response;
    }
    finally
    {
        stopwatch.Stop();
        HttpStatusCode statusCode = (response != null) ? response.StatusCode : HttpStatusCode.InternalServerError;
        _routedMessageMetricTracker.Histogram.Observe(destinationID, statusCode, stopwatch.Elapsed.TotalSeconds);
    }
}

This is my code in Startup.ConfigureServices() (actually this is in an extension method I defined):

public static IServiceCollection AddRoutingClient(this IServiceCollection services, RoutingSettings routingSettings)
{
    List<TimeSpan> retryTimeSpans = new List<TimeSpan>();

    // routingSettings.RetrySeconds is just an array of double values.
    foreach (double retrySeconds in routingSettings.RetrySeconds)
    {
        if (retrySeconds >= 0) retryTimeSpans.Add(TimeSpan.FromSeconds(retrySeconds));
    }

    services.AddHttpClient<IRoutingClient, RoutingClient>()
        .AddPolicyHandler((services, request) => HttpPolicyExtensions.HandleTransientHttpError()
            .WaitAndRetryAsync(retryTimeSpans, onRetry: (outcome, timespan, retryAttempt, context) =>
                {
                    services.GetService<ILogger<RoutingClient>>()?
                        .LogWarning($"Delaying for {timespan.TotalMilliseconds}ms, then making retry {retryAttempt}.");
                }
            ));

    return services;
}

I'd really like to use the Polly approach because it seems clean but I am not sure what I am doing wrong here. I must be doing something wrong because I would expect this to be a very common use-case for Polly that should be handled.

I may be your added retry method is creating a recursive loop so code is trying to open a stream that is already opened. If a request fails, 99.999% of the time the server is offline and trying again without waiting until server comes back online will just continue to give failures. — jdweng, Jul 06 '20 at 17:52
@js80 can you show us what's happening here: at Sigil.Hardhead.Controllers.HardheadController.Resource(String destinationId) ? — MikeJ, Jul 06 '20 at 18:18
@jdweng The whole purpose of Polly is to delay in some fashion so you can retry hoping the service is back up again. — MikeJ, Jul 06 '20 at 18:20
if a server is down for two hours and you retry 5 times at intervals of 10 seconds will the retries ever succeed? Retries won't work if the connection is already opened. A retry makes sense if it is going to eliminate a high number of failure, but is worthless if the success rate is tiny. Before you just randomly make a fix you have to understand the statistic and determine if the fix is worth making. — jdweng, Jul 06 '20 at 19:19

score 4 · Accepted Answer · answered Jul 06 '20 at 20:42

It turns out my problem was not caused by the code dealing with the response above. Instead, it was actually caused by my code that was manipulating the request.

I was reading the incoming request as an HttpRequestMessage object via the HttpContext.GetHttpRequestMessage() method and I attempted to re-use that same object to pass on to the backend service via a call to my typed HttpClient. However, the content stream for that request is read-once so I had to make a copy of that HttpRequestMessage as described in this answer to another post: https://stackoverflow.com/a/34049029/1221718

Here is a slightly more verbose version of that answer's code:

private static async Task<HttpRequestMessage> CloneHttpRequestMessageAsync(HttpRequestMessage request)
{
    HttpRequestMessage copyOfRequest = new HttpRequestMessage(request.Method, request.RequestUri);

    // Copy the request's content (via a MemoryStream) into the cloned object
    var ms = new MemoryStream();
    if (request.Content != null)
    {
        await request.Content.CopyToAsync(ms).ConfigureAwait(false);
        ms.Position = 0;
        copyOfRequest.Content = new StreamContent(ms);

        // Copy the content headers
        if (request.Content.Headers != null)
        {
            foreach (var h in request.Content.Headers)
            {
                copyOfRequest.Content.Headers.Add(h.Key, h.Value);
            }
        }
    }

    copyOfRequest.Version = request.Version;

    foreach (KeyValuePair<string, object> prop in request.Properties)
    {
        copyOfRequest.Properties.Add(prop);
    }

    foreach (KeyValuePair<string, IEnumerable<string>> header in request.Headers)
    {
        copyOfRequest.Headers.TryAddWithoutValidation(header.Key, header.Value);
    }

    return copyOfRequest;
}

score 1 · Answer 2 · answered Apr 08 '22 at 14:27

Another solution to this problem is to enable buffering on the request stream. This is done by calling the extension method EnableBuffering() for the HttpRequest, e.g.

public async Task InvokeAsync(HttpContext context, RequestDelegate next)
{
    context.Request.EnableBuffering();
  ...

see https://devblogs.microsoft.com/dotnet/re-reading-asp-net-core-request-bodies-with-enablebuffering/ for more detail.

'Stream was already consumed' error using Polly to retry requests in ASP.NET Core

2 Answers2

Linked