From 7bf1cf520589b85571e206658a9045ca2f148cd9 Mon Sep 17 00:00:00 2001 From: Tom Longhurst <30480171+thomhurst@users.noreply.github.com> Date: Thu, 30 Jul 2026 13:34:08 +0100 Subject: [PATCH 01/17] fix(http): stream response bodies Bound logging previews so response logging no longer defeats streaming. Refs #3467. --- src/ModularPipelines/Http/Http.cs | 12 +- .../Http/HttpContentPreviewReader.cs | 170 ++++++++++++++++++ .../Http/HttpRequestFormatter.cs | 38 ++-- .../Http/HttpResponseFormatter.cs | 38 ++-- .../Http/ResponseLoggingHttpHandler.cs | 5 - .../Options/DownloadOptions.cs | 5 +- .../Options/HttpLoggingOptions.cs | 2 +- .../Context/HttpFormatterTests.cs | 98 ++++++++++ .../Context/HttpTests.cs | 70 ++++++++ .../Helpers/DownloaderTests.cs | 8 + 10 files changed, 412 insertions(+), 34 deletions(-) create mode 100644 src/ModularPipelines/Http/HttpContentPreviewReader.cs create mode 100644 test/ModularPipelines.UnitTests/Context/HttpFormatterTests.cs diff --git a/src/ModularPipelines/Http/Http.cs b/src/ModularPipelines/Http/Http.cs index 170d14a90d..d2cc807115 100644 --- a/src/ModularPipelines/Http/Http.cs +++ b/src/ModularPipelines/Http/Http.cs @@ -50,7 +50,11 @@ public async Task SendAsync(HttpOptions httpOptions, Cancel var httpClient = GetHttpClient(httpOptions.LoggingType); - var response = await httpClient.SendAsync(httpOptions.HttpRequestMessage, effectiveCancellationToken).ConfigureAwait(false); + var response = await httpClient.SendAsync( + httpOptions.HttpRequestMessage, + HttpCompletionOption.ResponseHeadersRead, + effectiveCancellationToken) + .ConfigureAwait(false); if (!httpOptions.ThrowOnNonSuccessStatusCode) { @@ -78,7 +82,11 @@ private async Task SendAndWrapLogging(HttpOptions httpOptio var stopWatch = Stopwatch.StartNew(); - var response = await httpOptions.HttpClient!.SendAsync(httpOptions.HttpRequestMessage, cancellationToken).ConfigureAwait(false); + var response = await httpOptions.HttpClient!.SendAsync( + httpOptions.HttpRequestMessage, + HttpCompletionOption.ResponseHeadersRead, + cancellationToken) + .ConfigureAwait(false); LogStatusCode(response.StatusCode, httpOptions, loggingOptions); LogDuration(stopWatch.Elapsed, httpOptions, loggingOptions); diff --git a/src/ModularPipelines/Http/HttpContentPreviewReader.cs b/src/ModularPipelines/Http/HttpContentPreviewReader.cs new file mode 100644 index 0000000000..a468576a44 --- /dev/null +++ b/src/ModularPipelines/Http/HttpContentPreviewReader.cs @@ -0,0 +1,170 @@ +using System.Text; + +namespace ModularPipelines.Http; + +/// +/// Reads bounded text previews while preserving the complete content stream for its consumer. +/// +internal static class HttpContentPreviewReader +{ + /// + /// Reads at most one byte beyond the configured preview limit. + /// + /// The HTTP content to preview. + /// The maximum number of bytes to include in the preview. + /// The cancellation token. + /// The preview, truncation state, replayable content, and declared total length. + public static async Task<(string Preview, bool IsTruncated, HttpContent ReplayContent, long? TotalLength)> + ReadAsync( + HttpContent content, + int maxBytes, + CancellationToken cancellationToken) + { + if (maxBytes <= 0 || maxBytes == int.MaxValue) + { + var unboundedBody = await content.ReadAsStringAsync(cancellationToken).ConfigureAwait(false); + return (unboundedBody, false, content, content.Headers.ContentLength); + } + + var stream = await content.ReadAsStreamAsync(cancellationToken).ConfigureAwait(false); + var buffer = new byte[maxBytes + 1]; + var bytesRead = 0; + + while (bytesRead < buffer.Length) + { + var read = await stream.ReadAsync( + buffer.AsMemory(bytesRead, buffer.Length - bytesRead), + cancellationToken) + .ConfigureAwait(false); + if (read == 0) + { + break; + } + + bytesRead += read; + } + + var isTruncated = bytesRead > maxBytes; + var previewLength = Math.Min(bytesRead, maxBytes); + var encoding = GetEncoding(content); + var preview = encoding.GetString(buffer, 0, previewLength); + var replayContent = CreateReplayContent(content, stream, buffer.AsSpan(0, bytesRead).ToArray()); + + return (preview, isTruncated, replayContent, content.Headers.ContentLength); + } + + private static Encoding GetEncoding(HttpContent content) + { + var charset = content.Headers.ContentType?.CharSet?.Trim('"'); + if (string.IsNullOrWhiteSpace(charset)) + { + return Encoding.UTF8; + } + + try + { + return Encoding.GetEncoding(charset); + } + catch (ArgumentException) + { + return Encoding.UTF8; + } + } + + private static StreamContent CreateReplayContent( + HttpContent originalContent, + Stream remainingStream, + byte[] prefix) + { + var replayContent = new StreamContent( + new PrefixReplayStream(prefix, remainingStream, originalContent)); + foreach (var header in originalContent.Headers) + { + replayContent.Headers.TryAddWithoutValidation(header.Key, header.Value); + } + + return replayContent; + } + + private sealed class PrefixReplayStream( + byte[] prefix, + Stream remainingStream, + HttpContent ownedContent) : Stream + { + private int _prefixPosition; + + public override bool CanRead => true; + + public override bool CanSeek => false; + + public override bool CanWrite => false; + + public override long Length => throw new NotSupportedException(); + + public override long Position + { + get => throw new NotSupportedException(); + set => throw new NotSupportedException(); + } + + public override void Flush() + { + } + + public override int Read(byte[] buffer, int offset, int count) + { + var prefixBytesRead = ReadPrefix(buffer.AsSpan(offset, count)); + return prefixBytesRead > 0 + ? prefixBytesRead + : remainingStream.Read(buffer, offset, count); + } + + public override Task ReadAsync( + byte[] buffer, + int offset, + int count, + CancellationToken cancellationToken) + { + return ReadAsync(buffer.AsMemory(offset, count), cancellationToken).AsTask(); + } + + public override async ValueTask ReadAsync( + Memory buffer, + CancellationToken cancellationToken = default) + { + var prefixBytesRead = ReadPrefix(buffer.Span); + return prefixBytesRead > 0 + ? prefixBytesRead + : await remainingStream.ReadAsync(buffer, cancellationToken).ConfigureAwait(false); + } + + public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); + + public override void SetLength(long value) => throw new NotSupportedException(); + + public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException(); + + protected override void Dispose(bool disposing) + { + if (disposing) + { + ownedContent.Dispose(); + } + + base.Dispose(disposing); + } + + private int ReadPrefix(Span destination) + { + var bytesToCopy = Math.Min(destination.Length, prefix.Length - _prefixPosition); + if (bytesToCopy <= 0) + { + return 0; + } + + prefix.AsSpan(_prefixPosition, bytesToCopy).CopyTo(destination); + _prefixPosition += bytesToCopy; + return bytesToCopy; + } + } +} diff --git a/src/ModularPipelines/Http/HttpRequestFormatter.cs b/src/ModularPipelines/Http/HttpRequestFormatter.cs index 6907bb2d97..766e74ccbb 100644 --- a/src/ModularPipelines/Http/HttpRequestFormatter.cs +++ b/src/ModularPipelines/Http/HttpRequestFormatter.cs @@ -70,7 +70,7 @@ public async Task FormatAsync(HttpRequestMessage request, HttpLoggingOpt if (options.LogRequestBody) { - await AppendBodyAsync(sb, request.Content, options.MaxBodySizeToLog, cancellationToken).ConfigureAwait(false); + await AppendBodyAsync(sb, request, options.MaxBodySizeToLog, cancellationToken).ConfigureAwait(false); } return sb.ToString(); @@ -127,10 +127,15 @@ private static bool IsSensitiveHeader(string headerName, IReadOnlyList s return false; } - private async Task AppendBodyAsync(StringBuilder sb, HttpContent? content, int maxBodySize, CancellationToken cancellationToken) + private async Task AppendBodyAsync( + StringBuilder sb, + HttpRequestMessage request, + int maxBodySize, + CancellationToken cancellationToken) { sb.AppendLine("Body"); + var content = request.Content; if (content == null) { sb.AppendLine("\t(null)"); @@ -146,9 +151,12 @@ private async Task AppendBodyAsync(StringBuilder sb, HttpContent? content, int m return; } - var body = await content.ReadAsStringAsync(cancellationToken).ConfigureAwait(false); + var (body, isTruncated, replayContent, totalLength) = await HttpContentPreviewReader + .ReadAsync(content, maxBodySize, cancellationToken) + .ConfigureAwait(false); + request.Content = replayContent; - if (string.IsNullOrWhiteSpace(body)) + if (string.IsNullOrWhiteSpace(body) && !isTruncated) { sb.AppendLine("\t(empty)"); return; @@ -157,18 +165,24 @@ private async Task AppendBodyAsync(StringBuilder sb, HttpContent? content, int m // Obfuscate sensitive values in the body var obfuscatedBody = _secretObfuscator.Obfuscate(body, null); - // Truncate if body is too large - if (maxBodySize > 0 && obfuscatedBody.Length > maxBodySize) + sb.AppendLine($"\t{obfuscatedBody}"); + if (isTruncated) { - sb.AppendLine($"\t{obfuscatedBody[..maxBodySize]}"); - sb.AppendLine($"\t... [truncated, total size: {obfuscatedBody.Length:N0} characters]"); - } - else - { - sb.AppendLine($"\t{obfuscatedBody}"); + AppendTruncationMessage(sb, maxBodySize, totalLength); } } + private static void AppendTruncationMessage( + StringBuilder sb, + int maxBodySize, + long? totalLength) + { + var size = totalLength.HasValue + ? $"total size: {totalLength.Value:N0} bytes" + : $"after {maxBodySize:N0} bytes"; + sb.AppendLine($"\t... [truncated, {size}]"); + } + private static bool IsBinaryContent(HttpContent content) { var contentType = content.Headers.ContentType?.MediaType; diff --git a/src/ModularPipelines/Http/HttpResponseFormatter.cs b/src/ModularPipelines/Http/HttpResponseFormatter.cs index 886a17d963..8a7daa34ce 100644 --- a/src/ModularPipelines/Http/HttpResponseFormatter.cs +++ b/src/ModularPipelines/Http/HttpResponseFormatter.cs @@ -70,7 +70,7 @@ public async Task FormatAsync(HttpResponseMessage response, HttpLoggingO if (options.LogResponseBody) { - await AppendBodyAsync(sb, response.Content, options.MaxBodySizeToLog, cancellationToken).ConfigureAwait(false); + await AppendBodyAsync(sb, response, options.MaxBodySizeToLog, cancellationToken).ConfigureAwait(false); } return sb.ToString(); @@ -127,10 +127,15 @@ private static bool IsSensitiveHeader(string headerName, IReadOnlyList s return false; } - private async Task AppendBodyAsync(StringBuilder sb, HttpContent? content, int maxBodySize, CancellationToken cancellationToken) + private async Task AppendBodyAsync( + StringBuilder sb, + HttpResponseMessage response, + int maxBodySize, + CancellationToken cancellationToken) { sb.AppendLine("Body"); + var content = response.Content; if (content == null) { sb.AppendLine("\t(null)"); @@ -146,9 +151,12 @@ private async Task AppendBodyAsync(StringBuilder sb, HttpContent? content, int m return; } - var body = await content.ReadAsStringAsync(cancellationToken).ConfigureAwait(false); + var (body, isTruncated, replayContent, totalLength) = await HttpContentPreviewReader + .ReadAsync(content, maxBodySize, cancellationToken) + .ConfigureAwait(false); + response.Content = replayContent; - if (string.IsNullOrWhiteSpace(body)) + if (string.IsNullOrWhiteSpace(body) && !isTruncated) { sb.AppendLine("\t(empty)"); return; @@ -157,18 +165,24 @@ private async Task AppendBodyAsync(StringBuilder sb, HttpContent? content, int m // Obfuscate sensitive values in the body var obfuscatedBody = _secretObfuscator.Obfuscate(body, null); - // Truncate if body is too large - if (maxBodySize > 0 && obfuscatedBody.Length > maxBodySize) + sb.AppendLine($"\t{obfuscatedBody}"); + if (isTruncated) { - sb.AppendLine($"\t{obfuscatedBody[..maxBodySize]}"); - sb.AppendLine($"\t... [truncated, total size: {obfuscatedBody.Length:N0} characters]"); - } - else - { - sb.AppendLine($"\t{obfuscatedBody}"); + AppendTruncationMessage(sb, maxBodySize, totalLength); } } + private static void AppendTruncationMessage( + StringBuilder sb, + int maxBodySize, + long? totalLength) + { + var size = totalLength.HasValue + ? $"total size: {totalLength.Value:N0} bytes" + : $"after {maxBodySize:N0} bytes"; + sb.AppendLine($"\t... [truncated, {size}]"); + } + private static bool IsBinaryContent(HttpContent content) { var contentType = content.Headers.ContentType?.MediaType; diff --git a/src/ModularPipelines/Http/ResponseLoggingHttpHandler.cs b/src/ModularPipelines/Http/ResponseLoggingHttpHandler.cs index d1d9fce4eb..c59f7db4af 100644 --- a/src/ModularPipelines/Http/ResponseLoggingHttpHandler.cs +++ b/src/ModularPipelines/Http/ResponseLoggingHttpHandler.cs @@ -18,11 +18,6 @@ protected override async Task SendAsync(HttpRequestMessage { var response = await base.SendAsync(request, cancellationToken).ConfigureAwait(false); - // Buffer the response content so it can be read multiple times. - // Without this, reading the body for logging consumes the stream, - // making it unreadable by subsequent code. See issue #1610. - await response.Content.LoadIntoBufferAsync().ConfigureAwait(false); - var logger = _loggerProvider.GetLogger(); await _httpLogger.PrintResponse(response, logger).ConfigureAwait(false); diff --git a/src/ModularPipelines/Options/DownloadOptions.cs b/src/ModularPipelines/Options/DownloadOptions.cs index 2b16950971..601eda114d 100644 --- a/src/ModularPipelines/Options/DownloadOptions.cs +++ b/src/ModularPipelines/Options/DownloadOptions.cs @@ -21,5 +21,6 @@ public record DownloadOptions(Uri DownloadUri) /// /// Gets the type of HTTP logging to perform during the download. /// - public HttpLoggingType LoggingType { get; init; } = HttpLoggingType.Request | HttpLoggingType.Response | HttpLoggingType.StatusCode | HttpLoggingType.Duration; -} \ No newline at end of file + public HttpLoggingType LoggingType { get; init; } = + HttpLoggingType.Request | HttpLoggingType.StatusCode | HttpLoggingType.Duration; +} diff --git a/src/ModularPipelines/Options/HttpLoggingOptions.cs b/src/ModularPipelines/Options/HttpLoggingOptions.cs index 7f364cef51..916a4e55cc 100644 --- a/src/ModularPipelines/Options/HttpLoggingOptions.cs +++ b/src/ModularPipelines/Options/HttpLoggingOptions.cs @@ -67,7 +67,7 @@ public record HttpLoggingOptions public bool LogDuration { get; init; } = true; /// - /// Gets the maximum body size in characters to log. Default is 4096. + /// Gets the maximum body size in bytes to read and log. Default is 4096. /// Bodies larger than this will be truncated with a message indicating the full size. /// Set to 0 or negative to disable truncation. /// diff --git a/test/ModularPipelines.UnitTests/Context/HttpFormatterTests.cs b/test/ModularPipelines.UnitTests/Context/HttpFormatterTests.cs new file mode 100644 index 0000000000..86d88baac3 --- /dev/null +++ b/test/ModularPipelines.UnitTests/Context/HttpFormatterTests.cs @@ -0,0 +1,98 @@ +using System.Net.Http.Headers; +using System.Text; +using ModularPipelines.Engine; +using ModularPipelines.Http; +using ModularPipelines.Options; +using Moq; + +namespace ModularPipelines.UnitTests.Context; + +public class HttpFormatterTests +{ + [Test] + public async Task ResponseFormatter_BoundsPreviewReadAndPreservesBody() + { + const string body = "abcdefghijklmnopqrstuvwxyz"; + var stream = new CountingReadStream(Encoding.UTF8.GetBytes(body)); + using var response = new HttpResponseMessage + { + Content = CreateTextContent(stream), + }; + var formatter = new HttpResponseFormatter(CreateObfuscator()); + + var formatted = await formatter.FormatAsync(response, new HttpLoggingOptions + { + MaxBodySizeToLog = 5, + }); + var bytesReadForPreview = stream.BytesRead; + var replayedBody = await response.Content.ReadAsStringAsync(); + + using (Assert.Multiple()) + { + await Assert.That(bytesReadForPreview).IsLessThanOrEqualTo(6); + await Assert.That(formatted).Contains("\tabcde"); + await Assert.That(formatted).DoesNotContain("abcdef"); + await Assert.That(replayedBody).IsEqualTo(body); + } + } + + [Test] + public async Task RequestFormatter_BoundsPreviewReadAndPreservesBody() + { + const string body = "abcdefghijklmnopqrstuvwxyz"; + var stream = new CountingReadStream(Encoding.UTF8.GetBytes(body)); + using var request = new HttpRequestMessage(HttpMethod.Post, "https://example.test") + { + Content = CreateTextContent(stream), + }; + var formatter = new HttpRequestFormatter(CreateObfuscator()); + + var formatted = await formatter.FormatAsync(request, new HttpLoggingOptions + { + MaxBodySizeToLog = 5, + }); + var bytesReadForPreview = stream.BytesRead; + var replayedBody = await request.Content.ReadAsStringAsync(); + + using (Assert.Multiple()) + { + await Assert.That(bytesReadForPreview).IsLessThanOrEqualTo(6); + await Assert.That(formatted).Contains("\tabcde"); + await Assert.That(formatted).DoesNotContain("abcdef"); + await Assert.That(replayedBody).IsEqualTo(body); + } + } + + private static StreamContent CreateTextContent(Stream stream) + { + var content = new StreamContent(stream); + content.Headers.ContentType = new MediaTypeHeaderValue("text/plain") + { + CharSet = "utf-8", + }; + return content; + } + + private static ISecretObfuscator CreateObfuscator() + { + var obfuscator = new Mock(); + obfuscator + .Setup(x => x.Obfuscate(It.IsAny(), It.IsAny())) + .Returns((string? input, object? _) => input ?? string.Empty); + return obfuscator.Object; + } + + private sealed class CountingReadStream(byte[] contents) : MemoryStream(contents) + { + public int BytesRead { get; private set; } + + public override async ValueTask ReadAsync( + Memory buffer, + CancellationToken cancellationToken = default) + { + var read = await base.ReadAsync(buffer, cancellationToken); + BytesRead += read; + return read; + } + } +} diff --git a/test/ModularPipelines.UnitTests/Context/HttpTests.cs b/test/ModularPipelines.UnitTests/Context/HttpTests.cs index 2873eeb409..7790b2bc24 100644 --- a/test/ModularPipelines.UnitTests/Context/HttpTests.cs +++ b/test/ModularPipelines.UnitTests/Context/HttpTests.cs @@ -1,3 +1,4 @@ +using System.Net; using System.Reflection; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; @@ -14,6 +15,35 @@ namespace ModularPipelines.UnitTests.Context; public class HttpTests : TestBase { + [Test] + public async Task SendAsync_ReturnsAfterHeadersWithoutBufferingResponseBody() + { + var content = new BlockingHttpContent(); + var handler = new ImmediateResponseHandler(content); + using var httpClient = new HttpClient(handler); + var result = await GetService((_, _) => { }); + var sendTask = result.T.SendAsync(new HttpOptions( + new HttpRequestMessage(HttpMethod.Get, "https://example.test/large-file")) + { + HttpClient = httpClient, + LoggingType = HttpLoggingType.None, + }); + + await handler.RequestReceived.Task.WaitAsync(TimeSpan.FromSeconds(5)); + + HttpResponseMessage? response = null; + try + { + response = await sendTask.WaitAsync(TimeSpan.FromSeconds(1)); + } + finally + { + content.Release(); + response ??= await sendTask; + response.Dispose(); + } + } + [Test] public async Task PublicApi_DoesNotExposeRawHttpClients() { @@ -155,4 +185,44 @@ await http.SendAsync(new HttpOptions(new HttpRequestMessage(HttpMethod.Get, serv await Assert.That(indexOfDuration).IsLessThan(indexOfResponse); } } + + private sealed class ImmediateResponseHandler(HttpContent content) : HttpMessageHandler + { + public TaskCompletionSource RequestReceived { get; } = + new(TaskCreationOptions.RunContinuationsAsynchronously); + + protected override Task SendAsync( + HttpRequestMessage request, + CancellationToken cancellationToken) + { + RequestReceived.TrySetResult(); + return Task.FromResult(new HttpResponseMessage(HttpStatusCode.OK) + { + Content = content, + RequestMessage = request, + }); + } + } + + private sealed class BlockingHttpContent : HttpContent + { + private readonly TaskCompletionSource _release = + new(TaskCreationOptions.RunContinuationsAsynchronously); + + public void Release() => _release.TrySetResult(); + + protected override async Task SerializeToStreamAsync( + Stream stream, + TransportContext? context) + { + await _release.Task; + await stream.WriteAsync("response body"u8.ToArray()); + } + + protected override bool TryComputeLength(out long length) + { + length = 0; + return false; + } + } } diff --git a/test/ModularPipelines.UnitTests/Helpers/DownloaderTests.cs b/test/ModularPipelines.UnitTests/Helpers/DownloaderTests.cs index 8cf8da07ac..fb68a482bd 100644 --- a/test/ModularPipelines.UnitTests/Helpers/DownloaderTests.cs +++ b/test/ModularPipelines.UnitTests/Helpers/DownloaderTests.cs @@ -8,6 +8,14 @@ namespace ModularPipelines.UnitTests.Helpers; public class DownloaderTests : TestBase { + [Test] + public async Task DownloadOptions_DisableResponseBodyLoggingByDefault() + { + var options = new DownloadOptions(new Uri("https://example.test/file")); + + await Assert.That(options.LoggingType.HasFlag(ModularPipelines.Http.HttpLoggingType.Response)).IsFalse(); + } + [Test] public async Task Can_Download() { From d1e741d1fdc25cc6a34d67e8a590f8b5cd719dad Mon Sep 17 00:00:00 2001 From: Tom Longhurst <30480171+thomhurst@users.noreply.github.com> Date: Thu, 30 Jul 2026 15:52:21 +0100 Subject: [PATCH 02/17] fix(http): secure streamed response handling Keep configured cancellation alive until response content disposal and mask partial secrets at bounded preview edges. Refs #3467 --- src/ModularPipelines/Http/Http.cs | 104 +++++++--- .../Http/HttpBodySecretRedactor.cs | 62 ++++++ .../Http/HttpRequestFormatter.cs | 17 +- .../Http/HttpResponseFormatter.cs | 17 +- .../Http/TimeoutHttpContent.cs | 195 ++++++++++++++++++ .../Context/HttpFormatterTests.cs | 83 +++++++- .../Context/HttpTests.cs | 51 +++++ 7 files changed, 493 insertions(+), 36 deletions(-) create mode 100644 src/ModularPipelines/Http/HttpBodySecretRedactor.cs create mode 100644 src/ModularPipelines/Http/TimeoutHttpContent.cs diff --git a/src/ModularPipelines/Http/Http.cs b/src/ModularPipelines/Http/Http.cs index d2cc807115..a64e157c6b 100644 --- a/src/ModularPipelines/Http/Http.cs +++ b/src/ModularPipelines/Http/Http.cs @@ -32,36 +32,72 @@ public async Task SendAsync(HttpOptions httpOptions, Cancel var effectiveTimeout = httpOptions.Timeout ?? _pipelineOptions.Value.DefaultHttpTimeout; // Create timeout token if timeout is specified - using var timeoutCts = effectiveTimeout.HasValue + var timeoutCts = effectiveTimeout.HasValue ? new CancellationTokenSource(effectiveTimeout.Value) : null; // Link the timeout token with the passed cancellation token, or just use the original if no timeout - using var linkedCts = timeoutCts != null + var linkedCts = timeoutCts != null ? CancellationTokenSource.CreateLinkedTokenSource(timeoutCts.Token, cancellationToken) : null; var effectiveCancellationToken = linkedCts?.Token ?? cancellationToken; - if (httpOptions.HttpClient != null) + try { - return await SendAndWrapLogging(httpOptions, effectiveCancellationToken).ConfigureAwait(false); + HttpResponseMessage WrapResponse(HttpResponseMessage response) + { + if (timeoutCts is null || linkedCts is null) + { + return response; + } + + response.Content = new TimeoutHttpContent(response.Content, timeoutCts, linkedCts); + timeoutCts = null; + linkedCts = null; + return response; + } + + if (httpOptions.HttpClient != null) + { + return await SendAndWrapLogging( + httpOptions, + WrapResponse, + effectiveCancellationToken) + .ConfigureAwait(false); + } + + var httpClient = GetHttpClient(httpOptions.LoggingType); + + var response = await httpClient.SendAsync( + httpOptions.HttpRequestMessage, + HttpCompletionOption.ResponseHeadersRead, + effectiveCancellationToken) + .ConfigureAwait(false); + + response = WrapResponse(response); + try + { + if (!httpOptions.ThrowOnNonSuccessStatusCode) + { + return response; + } + + return await response + .EnsureSuccessStatusCodeWithContentAsync(effectiveCancellationToken) + .ConfigureAwait(false); + } + catch + { + response.Dispose(); + throw; + } } - - var httpClient = GetHttpClient(httpOptions.LoggingType); - - var response = await httpClient.SendAsync( - httpOptions.HttpRequestMessage, - HttpCompletionOption.ResponseHeadersRead, - effectiveCancellationToken) - .ConfigureAwait(false); - - if (!httpOptions.ThrowOnNonSuccessStatusCode) + finally { - return response; + linkedCts?.Dispose(); + timeoutCts?.Dispose(); } - - return await response.EnsureSuccessStatusCodeWithContentAsync(effectiveCancellationToken).ConfigureAwait(false); } private HttpClient GetHttpClient(HttpLoggingType loggingType) @@ -70,7 +106,10 @@ private HttpClient GetHttpClient(HttpLoggingType loggingType) return _httpClientFactory.CreateClient(clientName); } - private async Task SendAndWrapLogging(HttpOptions httpOptions, CancellationToken cancellationToken) + private async Task SendAndWrapLogging( + HttpOptions httpOptions, + Func wrapResponse, + CancellationToken cancellationToken) { var logger = _moduleLoggerProvider.GetLogger(); var loggingOptions = GetEffectiveLoggingOptions(httpOptions); @@ -88,20 +127,29 @@ private async Task SendAndWrapLogging(HttpOptions httpOptio cancellationToken) .ConfigureAwait(false); - LogStatusCode(response.StatusCode, httpOptions, loggingOptions); - LogDuration(stopWatch.Elapsed, httpOptions, loggingOptions); - - if (httpOptions.LoggingType.HasFlag(HttpLoggingType.Response)) + response = wrapResponse(response); + try { - await _httpLogger.PrintResponse(response, logger, loggingOptions).ConfigureAwait(false); - } + LogStatusCode(response.StatusCode, httpOptions, loggingOptions); + LogDuration(stopWatch.Elapsed, httpOptions, loggingOptions); + + if (httpOptions.LoggingType.HasFlag(HttpLoggingType.Response)) + { + await _httpLogger.PrintResponse(response, logger, loggingOptions).ConfigureAwait(false); + } - if (!httpOptions.ThrowOnNonSuccessStatusCode) + if (!httpOptions.ThrowOnNonSuccessStatusCode) + { + return response; + } + + return await response.EnsureSuccessStatusCodeWithContentAsync(cancellationToken).ConfigureAwait(false); + } + catch { - return response; + response.Dispose(); + throw; } - - return await response.EnsureSuccessStatusCodeWithContentAsync(cancellationToken).ConfigureAwait(false); } private HttpLoggingOptions GetEffectiveLoggingOptions(HttpOptions httpOptions) diff --git a/src/ModularPipelines/Http/HttpBodySecretRedactor.cs b/src/ModularPipelines/Http/HttpBodySecretRedactor.cs new file mode 100644 index 0000000000..403bdc7776 --- /dev/null +++ b/src/ModularPipelines/Http/HttpBodySecretRedactor.cs @@ -0,0 +1,62 @@ +using Microsoft.Extensions.Options; +using ModularPipelines.Constants; +using ModularPipelines.Engine; +using ModularPipelines.Options; + +namespace ModularPipelines.Http; + +internal static class HttpBodySecretRedactor +{ + public static string Redact( + string body, + bool isTruncated, + ISecretObfuscator secretObfuscator, + ISecretProvider secretProvider, + IOptions maskingOptions) + { + var boundarySafeBody = isTruncated + ? RedactPartialSecretAtBoundary(body, secretProvider.GetSnapshot().Secrets, maskingOptions.Value) + : body; + return secretObfuscator.Obfuscate(boundarySafeBody, null); + } + + private static string RedactPartialSecretAtBoundary( + string body, + IReadOnlyList secrets, + SecretMaskingOptions options) + { + if (body.Length == 0) + { + return body; + } + + var comparison = options.CaseInsensitive + ? StringComparison.OrdinalIgnoreCase + : StringComparison.Ordinal; + var longestPartialMatch = 0; + + foreach (var secret in secrets) + { + var maximumLength = Math.Min(body.Length, secret.Length - 1); + for (var length = maximumLength; length > longestPartialMatch; length--) + { + if (body.AsSpan(body.Length - length, length) + .Equals(secret.AsSpan(0, length), comparison)) + { + longestPartialMatch = length; + break; + } + } + } + + if (longestPartialMatch == 0) + { + return body; + } + + var mask = string.IsNullOrWhiteSpace(options.MaskValue) + ? LoggingConstants.SecretMask + : options.MaskValue; + return string.Concat(body.AsSpan(0, body.Length - longestPartialMatch), mask); + } +} diff --git a/src/ModularPipelines/Http/HttpRequestFormatter.cs b/src/ModularPipelines/Http/HttpRequestFormatter.cs index 766e74ccbb..4ea3774f7f 100644 --- a/src/ModularPipelines/Http/HttpRequestFormatter.cs +++ b/src/ModularPipelines/Http/HttpRequestFormatter.cs @@ -1,5 +1,6 @@ using System.Net.Http.Headers; using System.Text; +using Microsoft.Extensions.Options; using ModularPipelines.Constants; using ModularPipelines.Engine; using ModularPipelines.Options; @@ -31,10 +32,17 @@ namespace ModularPipelines.Http; internal class HttpRequestFormatter : IHttpRequestFormatter { private readonly ISecretObfuscator _secretObfuscator; + private readonly ISecretProvider _secretProvider; + private readonly IOptions _secretMaskingOptions; - public HttpRequestFormatter(ISecretObfuscator secretObfuscator) + public HttpRequestFormatter( + ISecretObfuscator secretObfuscator, + ISecretProvider secretProvider, + IOptions secretMaskingOptions) { _secretObfuscator = secretObfuscator; + _secretProvider = secretProvider; + _secretMaskingOptions = secretMaskingOptions; } private static readonly HashSet TextMediaTypes = new(StringComparer.OrdinalIgnoreCase) @@ -163,7 +171,12 @@ private async Task AppendBodyAsync( } // Obfuscate sensitive values in the body - var obfuscatedBody = _secretObfuscator.Obfuscate(body, null); + var obfuscatedBody = HttpBodySecretRedactor.Redact( + body, + isTruncated, + _secretObfuscator, + _secretProvider, + _secretMaskingOptions); sb.AppendLine($"\t{obfuscatedBody}"); if (isTruncated) diff --git a/src/ModularPipelines/Http/HttpResponseFormatter.cs b/src/ModularPipelines/Http/HttpResponseFormatter.cs index 8a7daa34ce..3c49af94aa 100644 --- a/src/ModularPipelines/Http/HttpResponseFormatter.cs +++ b/src/ModularPipelines/Http/HttpResponseFormatter.cs @@ -1,5 +1,6 @@ using System.Net.Http.Headers; using System.Text; +using Microsoft.Extensions.Options; using ModularPipelines.Constants; using ModularPipelines.Engine; using ModularPipelines.Options; @@ -31,10 +32,17 @@ namespace ModularPipelines.Http; internal class HttpResponseFormatter : IHttpResponseFormatter { private readonly ISecretObfuscator _secretObfuscator; + private readonly ISecretProvider _secretProvider; + private readonly IOptions _secretMaskingOptions; - public HttpResponseFormatter(ISecretObfuscator secretObfuscator) + public HttpResponseFormatter( + ISecretObfuscator secretObfuscator, + ISecretProvider secretProvider, + IOptions secretMaskingOptions) { _secretObfuscator = secretObfuscator; + _secretProvider = secretProvider; + _secretMaskingOptions = secretMaskingOptions; } private static readonly HashSet TextMediaTypes = new(StringComparer.OrdinalIgnoreCase) @@ -163,7 +171,12 @@ private async Task AppendBodyAsync( } // Obfuscate sensitive values in the body - var obfuscatedBody = _secretObfuscator.Obfuscate(body, null); + var obfuscatedBody = HttpBodySecretRedactor.Redact( + body, + isTruncated, + _secretObfuscator, + _secretProvider, + _secretMaskingOptions); sb.AppendLine($"\t{obfuscatedBody}"); if (isTruncated) diff --git a/src/ModularPipelines/Http/TimeoutHttpContent.cs b/src/ModularPipelines/Http/TimeoutHttpContent.cs new file mode 100644 index 0000000000..137ef289f8 --- /dev/null +++ b/src/ModularPipelines/Http/TimeoutHttpContent.cs @@ -0,0 +1,195 @@ +using System.Net; + +namespace ModularPipelines.Http; + +internal sealed class TimeoutHttpContent : HttpContent +{ + private readonly HttpContent _innerContent; + private readonly CancellationTokenSource _timeoutCancellationTokenSource; + private readonly CancellationTokenSource _linkedCancellationTokenSource; + + public TimeoutHttpContent( + HttpContent innerContent, + CancellationTokenSource timeoutCancellationTokenSource, + CancellationTokenSource linkedCancellationTokenSource) + { + _innerContent = innerContent; + _timeoutCancellationTokenSource = timeoutCancellationTokenSource; + _linkedCancellationTokenSource = linkedCancellationTokenSource; + + foreach (var header in innerContent.Headers) + { + Headers.TryAddWithoutValidation(header.Key, header.Value); + } + } + + protected override Task SerializeToStreamAsync(Stream stream, TransportContext? context) + { + return _innerContent.CopyToAsync(stream, _linkedCancellationTokenSource.Token); + } + + protected override async Task SerializeToStreamAsync( + Stream stream, + TransportContext? context, + CancellationToken cancellationToken) + { + using var readCancellationTokenSource = CreateReadCancellationTokenSource(cancellationToken); + await _innerContent.CopyToAsync( + stream, + readCancellationTokenSource?.Token ?? _linkedCancellationTokenSource.Token) + .ConfigureAwait(false); + } + + protected override async Task CreateContentReadStreamAsync() + { + var stream = await _innerContent + .ReadAsStreamAsync(_linkedCancellationTokenSource.Token) + .ConfigureAwait(false); + return new TimeoutReadStream(stream, _linkedCancellationTokenSource.Token); + } + + protected override async Task CreateContentReadStreamAsync( + CancellationToken cancellationToken) + { + using var readCancellationTokenSource = CreateReadCancellationTokenSource(cancellationToken); + var effectiveCancellationToken = + readCancellationTokenSource?.Token ?? _linkedCancellationTokenSource.Token; + var stream = await _innerContent + .ReadAsStreamAsync(effectiveCancellationToken) + .ConfigureAwait(false); + return new TimeoutReadStream(stream, _linkedCancellationTokenSource.Token); + } + + protected override bool TryComputeLength(out long length) + { + if (_innerContent.Headers.ContentLength is { } contentLength) + { + length = contentLength; + return true; + } + + length = 0; + return false; + } + + protected override void Dispose(bool disposing) + { + if (disposing) + { + _innerContent.Dispose(); + _linkedCancellationTokenSource.Dispose(); + _timeoutCancellationTokenSource.Dispose(); + } + + base.Dispose(disposing); + } + + private CancellationTokenSource? CreateReadCancellationTokenSource( + CancellationToken cancellationToken) + { + return cancellationToken.CanBeCanceled && + cancellationToken != _linkedCancellationTokenSource.Token + ? CancellationTokenSource.CreateLinkedTokenSource( + cancellationToken, + _linkedCancellationTokenSource.Token) + : null; + } + + private sealed class TimeoutReadStream( + Stream innerStream, + CancellationToken timeoutCancellationToken) : Stream + { + public override bool CanRead => innerStream.CanRead; + + public override bool CanSeek => innerStream.CanSeek; + + public override bool CanWrite => innerStream.CanWrite; + + public override long Length => innerStream.Length; + + public override long Position + { + get => innerStream.Position; + set => innerStream.Position = value; + } + + public override void Flush() => innerStream.Flush(); + + public override Task FlushAsync(CancellationToken cancellationToken) + { + return innerStream.FlushAsync(cancellationToken); + } + + public override int Read(byte[] buffer, int offset, int count) + { + timeoutCancellationToken.ThrowIfCancellationRequested(); + return innerStream.Read(buffer, offset, count); + } + + public override int Read(Span buffer) + { + timeoutCancellationToken.ThrowIfCancellationRequested(); + return innerStream.Read(buffer); + } + + public override Task ReadAsync( + byte[] buffer, + int offset, + int count, + CancellationToken cancellationToken) + { + return ReadAsync(buffer.AsMemory(offset, count), cancellationToken).AsTask(); + } + + public override async ValueTask ReadAsync( + Memory buffer, + CancellationToken cancellationToken = default) + { + using var readCancellationTokenSource = + CreateReadCancellationTokenSource(cancellationToken); + return await innerStream.ReadAsync( + buffer, + readCancellationTokenSource?.Token ?? timeoutCancellationToken) + .ConfigureAwait(false); + } + + public override long Seek(long offset, SeekOrigin origin) + { + return innerStream.Seek(offset, origin); + } + + public override void SetLength(long value) => innerStream.SetLength(value); + + public override void Write(byte[] buffer, int offset, int count) + { + innerStream.Write(buffer, offset, count); + } + + public override async ValueTask DisposeAsync() + { + await innerStream.DisposeAsync().ConfigureAwait(false); + GC.SuppressFinalize(this); + } + + protected override void Dispose(bool disposing) + { + if (disposing) + { + innerStream.Dispose(); + } + + base.Dispose(disposing); + } + + private CancellationTokenSource? CreateReadCancellationTokenSource( + CancellationToken cancellationToken) + { + return cancellationToken.CanBeCanceled && + cancellationToken != timeoutCancellationToken + ? CancellationTokenSource.CreateLinkedTokenSource( + cancellationToken, + timeoutCancellationToken) + : null; + } + } +} diff --git a/test/ModularPipelines.UnitTests/Context/HttpFormatterTests.cs b/test/ModularPipelines.UnitTests/Context/HttpFormatterTests.cs index 86d88baac3..54d907c759 100644 --- a/test/ModularPipelines.UnitTests/Context/HttpFormatterTests.cs +++ b/test/ModularPipelines.UnitTests/Context/HttpFormatterTests.cs @@ -1,5 +1,6 @@ using System.Net.Http.Headers; using System.Text; +using ModularPipelines.Constants; using ModularPipelines.Engine; using ModularPipelines.Http; using ModularPipelines.Options; @@ -18,7 +19,7 @@ public async Task ResponseFormatter_BoundsPreviewReadAndPreservesBody() { Content = CreateTextContent(stream), }; - var formatter = new HttpResponseFormatter(CreateObfuscator()); + var formatter = CreateResponseFormatter(); var formatted = await formatter.FormatAsync(response, new HttpLoggingOptions { @@ -45,7 +46,7 @@ public async Task RequestFormatter_BoundsPreviewReadAndPreservesBody() { Content = CreateTextContent(stream), }; - var formatter = new HttpRequestFormatter(CreateObfuscator()); + var formatter = CreateRequestFormatter(); var formatted = await formatter.FormatAsync(request, new HttpLoggingOptions { @@ -63,6 +64,50 @@ public async Task RequestFormatter_BoundsPreviewReadAndPreservesBody() } } + [Test] + public async Task ResponseFormatter_RedactsSecretCrossingPreviewBoundary() + { + const string secret = "secret-value"; + using var response = new HttpResponseMessage + { + Content = CreateTextContent(new MemoryStream(Encoding.UTF8.GetBytes($"prefix-{secret}-suffix"))), + }; + var formatter = CreateResponseFormatter(secret); + + var formatted = await formatter.FormatAsync(response, new HttpLoggingOptions + { + MaxBodySizeToLog = 13, + }); + + using (Assert.Multiple()) + { + await Assert.That(formatted).Contains($"prefix-{LoggingConstants.SecretMask}"); + await Assert.That(formatted).DoesNotContain("prefix-secret"); + } + } + + [Test] + public async Task RequestFormatter_RedactsSecretCrossingPreviewBoundary() + { + const string secret = "secret-value"; + using var request = new HttpRequestMessage(HttpMethod.Post, "https://example.test") + { + Content = CreateTextContent(new MemoryStream(Encoding.UTF8.GetBytes($"prefix-{secret}-suffix"))), + }; + var formatter = CreateRequestFormatter(secret); + + var formatted = await formatter.FormatAsync(request, new HttpLoggingOptions + { + MaxBodySizeToLog = 13, + }); + + using (Assert.Multiple()) + { + await Assert.That(formatted).Contains($"prefix-{LoggingConstants.SecretMask}"); + await Assert.That(formatted).DoesNotContain("prefix-secret"); + } + } + private static StreamContent CreateTextContent(Stream stream) { var content = new StreamContent(stream); @@ -73,15 +118,45 @@ private static StreamContent CreateTextContent(Stream stream) return content; } - private static ISecretObfuscator CreateObfuscator() + private static ISecretObfuscator CreateObfuscator(params string[] secrets) { var obfuscator = new Mock(); obfuscator .Setup(x => x.Obfuscate(It.IsAny(), It.IsAny())) - .Returns((string? input, object? _) => input ?? string.Empty); + .Returns((string? input, object? _) => secrets.Aggregate( + input ?? string.Empty, + (value, secret) => value.Replace( + secret, + LoggingConstants.SecretMask, + StringComparison.Ordinal))); return obfuscator.Object; } + private static HttpRequestFormatter CreateRequestFormatter(params string[] secrets) + { + return new HttpRequestFormatter( + CreateObfuscator(secrets), + CreateSecretProvider(secrets), + Microsoft.Extensions.Options.Options.Create(new SecretMaskingOptions())); + } + + private static HttpResponseFormatter CreateResponseFormatter(params string[] secrets) + { + return new HttpResponseFormatter( + CreateObfuscator(secrets), + CreateSecretProvider(secrets), + Microsoft.Extensions.Options.Options.Create(new SecretMaskingOptions())); + } + + private static ISecretProvider CreateSecretProvider(string[] secrets) + { + var secretProvider = new Mock(); + secretProvider + .Setup(x => x.GetSnapshot()) + .Returns(new SecretSnapshot(0, secrets)); + return secretProvider.Object; + } + private sealed class CountingReadStream(byte[] contents) : MemoryStream(contents) { public int BytesRead { get; private set; } diff --git a/test/ModularPipelines.UnitTests/Context/HttpTests.cs b/test/ModularPipelines.UnitTests/Context/HttpTests.cs index 7790b2bc24..343d08af93 100644 --- a/test/ModularPipelines.UnitTests/Context/HttpTests.cs +++ b/test/ModularPipelines.UnitTests/Context/HttpTests.cs @@ -3,6 +3,7 @@ using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; using ModularPipelines.Context.Domains.Network; +using ModularPipelines.Extensions; using ModularPipelines.Http; using ModularPipelines.Options; using ModularPipelines.TestHelpers; @@ -44,6 +45,40 @@ public async Task SendAsync_ReturnsAfterHeadersWithoutBufferingResponseBody() } } + [Test] + [Arguments(true)] + [Arguments(false)] + public async Task SendAsync_ConfiguredTimeoutCancelsStreamedBody(bool useRequestTimeout) + { + var timeout = TimeSpan.FromMilliseconds(100); + var contentStream = new BlockingReadStream(); + using var httpClient = new HttpClient(new ImmediateResponseHandler(new StreamContent(contentStream))); + var result = await GetService((builder, _) => + builder.ConfigurePipelineOptions(options => options with + { + DefaultHttpTimeout = useRequestTimeout ? null : timeout, + })); + using var response = await result.T.SendAsync(new HttpOptions( + new HttpRequestMessage(HttpMethod.Get, "https://example.test/stalled-body")) + { + HttpClient = httpClient, + LoggingType = HttpLoggingType.None, + Timeout = useRequestTimeout ? timeout : null, + }); + var stream = await response.Content.ReadAsStreamAsync(); + var readTask = stream.ReadAsync(new byte[1]).AsTask(); + + try + { + await Assert.ThrowsAsync( + async () => await readTask.WaitAsync(TimeSpan.FromSeconds(1))); + } + finally + { + contentStream.Release(); + } + } + [Test] public async Task PublicApi_DoesNotExposeRawHttpClients() { @@ -225,4 +260,20 @@ protected override bool TryComputeLength(out long length) return false; } } + + private sealed class BlockingReadStream : MemoryStream + { + private readonly TaskCompletionSource _release = + new(TaskCreationOptions.RunContinuationsAsynchronously); + + public void Release() => _release.TrySetResult(); + + public override async ValueTask ReadAsync( + Memory buffer, + CancellationToken cancellationToken = default) + { + await _release.Task.WaitAsync(cancellationToken); + return 0; + } + } } From 880e431488b4baa329e98bce3e4a96ce19bbb52d Mon Sep 17 00:00:00 2001 From: Tom Longhurst <30480171+thomhurst@users.noreply.github.com> Date: Thu, 30 Jul 2026 15:55:07 +0100 Subject: [PATCH 03/17] test(http): prove unbounded content replay Document and verify that unbounded ReadAsStringAsync buffering keeps request and response content readable. Refs #3467 --- .../Http/HttpContentPreviewReader.cs | 1 + .../Context/HttpFormatterTests.cs | 46 +++++++++++++++++++ 2 files changed, 47 insertions(+) diff --git a/src/ModularPipelines/Http/HttpContentPreviewReader.cs b/src/ModularPipelines/Http/HttpContentPreviewReader.cs index a468576a44..8ff790647c 100644 --- a/src/ModularPipelines/Http/HttpContentPreviewReader.cs +++ b/src/ModularPipelines/Http/HttpContentPreviewReader.cs @@ -22,6 +22,7 @@ internal static class HttpContentPreviewReader { if (maxBytes <= 0 || maxBytes == int.MaxValue) { + // ReadAsStringAsync buffers HttpContent, so returning the same instance preserves replay. var unboundedBody = await content.ReadAsStringAsync(cancellationToken).ConfigureAwait(false); return (unboundedBody, false, content, content.Headers.ContentLength); } diff --git a/test/ModularPipelines.UnitTests/Context/HttpFormatterTests.cs b/test/ModularPipelines.UnitTests/Context/HttpFormatterTests.cs index 54d907c759..1fc2f7319d 100644 --- a/test/ModularPipelines.UnitTests/Context/HttpFormatterTests.cs +++ b/test/ModularPipelines.UnitTests/Context/HttpFormatterTests.cs @@ -64,6 +64,52 @@ public async Task RequestFormatter_BoundsPreviewReadAndPreservesBody() } } + [Test] + public async Task ResponseFormatter_UnboundedPreviewPreservesBody() + { + const string body = "abcdefghijklmnopqrstuvwxyz"; + using var response = new HttpResponseMessage + { + Content = CreateTextContent(new MemoryStream(Encoding.UTF8.GetBytes(body))), + }; + var formatter = CreateResponseFormatter(); + + var formatted = await formatter.FormatAsync(response, new HttpLoggingOptions + { + MaxBodySizeToLog = 0, + }); + var replayedBody = await response.Content.ReadAsStringAsync(); + + using (Assert.Multiple()) + { + await Assert.That(formatted).Contains(body); + await Assert.That(replayedBody).IsEqualTo(body); + } + } + + [Test] + public async Task RequestFormatter_UnboundedPreviewPreservesBody() + { + const string body = "abcdefghijklmnopqrstuvwxyz"; + using var request = new HttpRequestMessage(HttpMethod.Post, "https://example.test") + { + Content = CreateTextContent(new MemoryStream(Encoding.UTF8.GetBytes(body))), + }; + var formatter = CreateRequestFormatter(); + + var formatted = await formatter.FormatAsync(request, new HttpLoggingOptions + { + MaxBodySizeToLog = 0, + }); + var replayedBody = await request.Content.ReadAsStringAsync(); + + using (Assert.Multiple()) + { + await Assert.That(formatted).Contains(body); + await Assert.That(replayedBody).IsEqualTo(body); + } + } + [Test] public async Task ResponseFormatter_RedactsSecretCrossingPreviewBoundary() { From 554e2c0ac204800c95eac2f773fd295bc46c3aea Mon Sep 17 00:00:00 2001 From: Tom Longhurst <30480171+thomhurst@users.noreply.github.com> Date: Thu, 30 Jul 2026 16:21:07 +0100 Subject: [PATCH 04/17] fix(http): harden response streaming Propagate cancellation through response logging, dispose retry responses, and decode complete preview characters before secret-boundary redaction. Refs #3467 --- src/ModularPipelines/Http/Http.cs | 12 +- .../Http/HttpContentPreviewReader.cs | 16 ++- src/ModularPipelines/Http/HttpLogger.cs | 64 ++++++++-- src/ModularPipelines/Http/IHttpLogger.cs | 64 ++++++++++ .../Http/RequestLoggingHttpHandler.cs | 6 +- .../Http/ResilienceHttpHandler.cs | 17 ++- .../Http/ResponseLoggingHttpHandler.cs | 17 ++- .../Context/HttpFormatterTests.cs | 22 ++++ .../Context/HttpTests.cs | 119 ++++++++++++++++++ 9 files changed, 316 insertions(+), 21 deletions(-) diff --git a/src/ModularPipelines/Http/Http.cs b/src/ModularPipelines/Http/Http.cs index a64e157c6b..6591b97191 100644 --- a/src/ModularPipelines/Http/Http.cs +++ b/src/ModularPipelines/Http/Http.cs @@ -116,7 +116,13 @@ private async Task SendAndWrapLogging( if (httpOptions.LoggingType.HasFlag(HttpLoggingType.Request)) { - await _httpLogger.PrintRequest(httpOptions.HttpRequestMessage, logger, loggingOptions).ConfigureAwait(false); + await _httpLogger + .PrintRequest( + httpOptions.HttpRequestMessage, + logger, + loggingOptions, + cancellationToken) + .ConfigureAwait(false); } var stopWatch = Stopwatch.StartNew(); @@ -135,7 +141,9 @@ private async Task SendAndWrapLogging( if (httpOptions.LoggingType.HasFlag(HttpLoggingType.Response)) { - await _httpLogger.PrintResponse(response, logger, loggingOptions).ConfigureAwait(false); + await _httpLogger + .PrintResponse(response, logger, loggingOptions, cancellationToken) + .ConfigureAwait(false); } if (!httpOptions.ThrowOnNonSuccessStatusCode) diff --git a/src/ModularPipelines/Http/HttpContentPreviewReader.cs b/src/ModularPipelines/Http/HttpContentPreviewReader.cs index 8ff790647c..7953cdf4a2 100644 --- a/src/ModularPipelines/Http/HttpContentPreviewReader.cs +++ b/src/ModularPipelines/Http/HttpContentPreviewReader.cs @@ -48,7 +48,7 @@ internal static class HttpContentPreviewReader var isTruncated = bytesRead > maxBytes; var previewLength = Math.Min(bytesRead, maxBytes); var encoding = GetEncoding(content); - var preview = encoding.GetString(buffer, 0, previewLength); + var preview = DecodeCompletePrefix(encoding, buffer, previewLength); var replayContent = CreateReplayContent(content, stream, buffer.AsSpan(0, bytesRead).ToArray()); return (preview, isTruncated, replayContent, content.Headers.ContentLength); @@ -72,6 +72,20 @@ private static Encoding GetEncoding(HttpContent content) } } + private static string DecodeCompletePrefix(Encoding encoding, byte[] buffer, int length) + { + var decoder = encoding.GetDecoder(); + var characters = new char[encoding.GetMaxCharCount(length)]; + decoder.Convert( + buffer.AsSpan(0, length), + characters, + flush: false, + out _, + out var charactersUsed, + out _); + return new string(characters, 0, charactersUsed); + } + private static StreamContent CreateReplayContent( HttpContent originalContent, Stream remainingStream, diff --git a/src/ModularPipelines/Http/HttpLogger.cs b/src/ModularPipelines/Http/HttpLogger.cs index 4f08d1c917..5ffe4d0467 100644 --- a/src/ModularPipelines/Http/HttpLogger.cs +++ b/src/ModularPipelines/Http/HttpLogger.cs @@ -28,7 +28,20 @@ public HttpLogger(IHttpRequestFormatter requestFormatter, IHttpResponseFormatter /// A representing the asynchronous operation. public Task PrintRequest(HttpRequestMessage request, IModuleLogger logger) { - return PrintRequest(request, logger, HttpLoggingOptions.Default); + return PrintRequest( + request, + logger, + HttpLoggingOptions.Default, + CancellationToken.None); + } + + /// + public Task PrintRequest( + HttpRequestMessage request, + IModuleLogger logger, + CancellationToken cancellationToken) + { + return PrintRequest(request, logger, HttpLoggingOptions.Default, cancellationToken); } /// @@ -38,14 +51,26 @@ public Task PrintRequest(HttpRequestMessage request, IModuleLogger logger) /// The current module logger. /// Options controlling what parts of the request to log. /// A representing the asynchronous operation. - public async Task PrintRequest(HttpRequestMessage request, IModuleLogger logger, HttpLoggingOptions options) + public Task PrintRequest(HttpRequestMessage request, IModuleLogger logger, HttpLoggingOptions options) + { + return PrintRequest(request, logger, options, CancellationToken.None); + } + + /// + public async Task PrintRequest( + HttpRequestMessage request, + IModuleLogger logger, + HttpLoggingOptions options, + CancellationToken cancellationToken) { if (!options.LogRequest) { return; } - var formattedRequest = await _requestFormatter.FormatAsync(request, options).ConfigureAwait(false); + var formattedRequest = await _requestFormatter + .FormatAsync(request, options, cancellationToken) + .ConfigureAwait(false); logger.LogInformation("HTTP Request:\n{Request}", formattedRequest); } @@ -57,7 +82,20 @@ public async Task PrintRequest(HttpRequestMessage request, IModuleLogger logger, /// A representing the asynchronous operation. public Task PrintResponse(HttpResponseMessage response, IModuleLogger logger) { - return PrintResponse(response, logger, HttpLoggingOptions.Default); + return PrintResponse( + response, + logger, + HttpLoggingOptions.Default, + CancellationToken.None); + } + + /// + public Task PrintResponse( + HttpResponseMessage response, + IModuleLogger logger, + CancellationToken cancellationToken) + { + return PrintResponse(response, logger, HttpLoggingOptions.Default, cancellationToken); } /// @@ -67,20 +105,32 @@ public Task PrintResponse(HttpResponseMessage response, IModuleLogger logger) /// The current module logger. /// Options controlling what parts of the response to log. /// A representing the asynchronous operation. - public async Task PrintResponse(HttpResponseMessage response, IModuleLogger logger, HttpLoggingOptions options) + public Task PrintResponse(HttpResponseMessage response, IModuleLogger logger, HttpLoggingOptions options) + { + return PrintResponse(response, logger, options, CancellationToken.None); + } + + /// + public async Task PrintResponse( + HttpResponseMessage response, + IModuleLogger logger, + HttpLoggingOptions options, + CancellationToken cancellationToken) { if (!options.LogResponse) { return; } - var formattedResponse = await _responseFormatter.FormatAsync(response, options).ConfigureAwait(false); + var formattedResponse = await _responseFormatter + .FormatAsync(response, options, cancellationToken) + .ConfigureAwait(false); logger.LogInformation("HTTP Response:\n{Response}", formattedResponse); } public void PrintStatusCode(HttpStatusCode? httpStatusCode, IModuleLogger logger) { - var statusCode = httpStatusCode == null ? null as int? : (int)httpStatusCode; + var statusCode = httpStatusCode == null ? null as int? : (int) httpStatusCode; var icon = statusCode is >= 200 and < 300 ? "+" : "x"; logger.LogInformation("{Icon} HTTP Status: {StatusCode} {HttpStatusCode}", icon, statusCode, httpStatusCode); diff --git a/src/ModularPipelines/Http/IHttpLogger.cs b/src/ModularPipelines/Http/IHttpLogger.cs index 25699c97a1..20e3f530a0 100644 --- a/src/ModularPipelines/Http/IHttpLogger.cs +++ b/src/ModularPipelines/Http/IHttpLogger.cs @@ -17,6 +17,21 @@ public interface IHttpLogger /// A representing the asynchronous operation. Task PrintRequest(HttpRequestMessage request, IModuleLogger logger); + /// + /// Prints the HTTP request. + /// + /// The HTTP request to print. + /// The current module logger. + /// A token to cancel request formatting. + /// A representing the asynchronous operation. + Task PrintRequest( + HttpRequestMessage request, + IModuleLogger logger, + CancellationToken cancellationToken) + { + return PrintRequest(request, logger); + } + /// /// Prints the HTTP request with logging options. /// @@ -26,6 +41,23 @@ public interface IHttpLogger /// A representing the asynchronous operation. Task PrintRequest(HttpRequestMessage request, IModuleLogger logger, HttpLoggingOptions options); + /// + /// Prints the HTTP request with logging options. + /// + /// The HTTP request to print. + /// The current module logger. + /// Options controlling what parts of the request to log. + /// A token to cancel request formatting. + /// A representing the asynchronous operation. + Task PrintRequest( + HttpRequestMessage request, + IModuleLogger logger, + HttpLoggingOptions options, + CancellationToken cancellationToken) + { + return PrintRequest(request, logger, options); + } + /// /// Prints the HTTP response. /// @@ -34,6 +66,21 @@ public interface IHttpLogger /// A representing the asynchronous operation. Task PrintResponse(HttpResponseMessage response, IModuleLogger logger); + /// + /// Prints the HTTP response. + /// + /// The HTTP response to print. + /// The current module logger. + /// A token to cancel response formatting. + /// A representing the asynchronous operation. + Task PrintResponse( + HttpResponseMessage response, + IModuleLogger logger, + CancellationToken cancellationToken) + { + return PrintResponse(response, logger); + } + /// /// Prints the HTTP response with logging options. /// @@ -43,6 +90,23 @@ public interface IHttpLogger /// A representing the asynchronous operation. Task PrintResponse(HttpResponseMessage response, IModuleLogger logger, HttpLoggingOptions options); + /// + /// Prints the HTTP response with logging options. + /// + /// The HTTP response to print. + /// The current module logger. + /// Options controlling what parts of the response to log. + /// A token to cancel response formatting. + /// A representing the asynchronous operation. + Task PrintResponse( + HttpResponseMessage response, + IModuleLogger logger, + HttpLoggingOptions options, + CancellationToken cancellationToken) + { + return PrintResponse(response, logger, options); + } + /// /// Prints the HTTP status code. /// diff --git a/src/ModularPipelines/Http/RequestLoggingHttpHandler.cs b/src/ModularPipelines/Http/RequestLoggingHttpHandler.cs index 21aa4271df..8c3603e90f 100644 --- a/src/ModularPipelines/Http/RequestLoggingHttpHandler.cs +++ b/src/ModularPipelines/Http/RequestLoggingHttpHandler.cs @@ -17,10 +17,12 @@ public RequestLoggingHttpHandler(IModuleLoggerProvider loggerProvider, IHttpLogg protected override async Task SendAsync(HttpRequestMessage request, CancellationToken cancellationToken) { var logger = _loggerProvider.GetLogger(); - await _httpLogger.PrintRequest(request, logger).ConfigureAwait(false); + await _httpLogger + .PrintRequest(request, logger, cancellationToken) + .ConfigureAwait(false); var response = await base.SendAsync(request, cancellationToken).ConfigureAwait(false); return response; } -} \ No newline at end of file +} diff --git a/src/ModularPipelines/Http/ResilienceHttpHandler.cs b/src/ModularPipelines/Http/ResilienceHttpHandler.cs index 9774c8c6c4..49561e2c2f 100644 --- a/src/ModularPipelines/Http/ResilienceHttpHandler.cs +++ b/src/ModularPipelines/Http/ResilienceHttpHandler.cs @@ -105,14 +105,21 @@ private Task OnRetry(DelegateResult outcome, TimeSpan delay outcome.Exception.GetType().Name, outcome.Exception.Message, retryAttempt, - (int)delay.TotalMilliseconds); + (int) delay.TotalMilliseconds); } else if (outcome.Result != null) { - logger.LogWarning("HTTP request returned {StatusCode}. Retry attempt {RetryAttempt} after {Delay}ms", - (int)outcome.Result.StatusCode, - retryAttempt, - (int)delay.TotalMilliseconds); + try + { + logger.LogWarning("HTTP request returned {StatusCode}. Retry attempt {RetryAttempt} after {Delay}ms", + (int) outcome.Result.StatusCode, + retryAttempt, + (int) delay.TotalMilliseconds); + } + finally + { + outcome.Result.Dispose(); + } } return Task.CompletedTask; diff --git a/src/ModularPipelines/Http/ResponseLoggingHttpHandler.cs b/src/ModularPipelines/Http/ResponseLoggingHttpHandler.cs index c59f7db4af..df3a9c3479 100644 --- a/src/ModularPipelines/Http/ResponseLoggingHttpHandler.cs +++ b/src/ModularPipelines/Http/ResponseLoggingHttpHandler.cs @@ -18,9 +18,18 @@ protected override async Task SendAsync(HttpRequestMessage { var response = await base.SendAsync(request, cancellationToken).ConfigureAwait(false); - var logger = _loggerProvider.GetLogger(); - await _httpLogger.PrintResponse(response, logger).ConfigureAwait(false); - - return response; + try + { + var logger = _loggerProvider.GetLogger(); + await _httpLogger + .PrintResponse(response, logger, cancellationToken) + .ConfigureAwait(false); + return response; + } + catch + { + response.Dispose(); + throw; + } } } diff --git a/test/ModularPipelines.UnitTests/Context/HttpFormatterTests.cs b/test/ModularPipelines.UnitTests/Context/HttpFormatterTests.cs index 1fc2f7319d..85d6f5f20b 100644 --- a/test/ModularPipelines.UnitTests/Context/HttpFormatterTests.cs +++ b/test/ModularPipelines.UnitTests/Context/HttpFormatterTests.cs @@ -154,6 +154,28 @@ public async Task RequestFormatter_RedactsSecretCrossingPreviewBoundary() } } + [Test] + public async Task ResponseFormatter_RedactsSecretWhenPreviewSplitsMultibyteCharacter() + { + const string secret = "abcédef"; + using var response = new HttpResponseMessage + { + Content = CreateTextContent(new MemoryStream(Encoding.UTF8.GetBytes($"prefix-{secret}-suffix"))), + }; + var formatter = CreateResponseFormatter(secret); + + var formatted = await formatter.FormatAsync(response, new HttpLoggingOptions + { + MaxBodySizeToLog = 11, + }); + + using (Assert.Multiple()) + { + await Assert.That(formatted).Contains($"prefix-{LoggingConstants.SecretMask}"); + await Assert.That(formatted).DoesNotContain("prefix-abc"); + } + } + private static StreamContent CreateTextContent(Stream stream) { var content = new StreamContent(stream); diff --git a/test/ModularPipelines.UnitTests/Context/HttpTests.cs b/test/ModularPipelines.UnitTests/Context/HttpTests.cs index 343d08af93..a1ced185f8 100644 --- a/test/ModularPipelines.UnitTests/Context/HttpTests.cs +++ b/test/ModularPipelines.UnitTests/Context/HttpTests.cs @@ -1,13 +1,17 @@ using System.Net; +using System.Net.Http.Headers; using System.Reflection; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; using ModularPipelines.Context.Domains.Network; +using ModularPipelines.Engine; using ModularPipelines.Extensions; using ModularPipelines.Http; +using ModularPipelines.Logging; using ModularPipelines.Options; using ModularPipelines.TestHelpers; using ModularPipelines.UnitTests.Helpers; +using Moq; using NReco.Logging.File; using File = System.IO.File; using LogLevel = Microsoft.Extensions.Logging.LogLevel; @@ -79,6 +83,98 @@ await Assert.ThrowsAsync( } } + [Test] + public async Task SendAsync_ConfiguredTimeoutCancelsFactoryResponseLogging() + { + var timeout = TimeSpan.FromMilliseconds(100); + var contentStream = new BlockingReadStream(); + var content = new StreamContent(contentStream); + content.Headers.ContentType = new MediaTypeHeaderValue("text/plain"); + + var moduleLoggerProvider = new Mock(); + moduleLoggerProvider + .Setup(x => x.GetLogger()) + .Returns(Mock.Of()); + var httpLogger = new HttpLogger( + Mock.Of(), + new HttpResponseFormatter( + Mock.Of(), + Mock.Of(), + Microsoft.Extensions.Options.Options.Create(new SecretMaskingOptions()))); + var responseLoggingHandler = new ResponseLoggingHttpHandler( + moduleLoggerProvider.Object, + httpLogger) + { + InnerHandler = new ImmediateResponseHandler(content), + }; + using var httpClient = new HttpClient(responseLoggingHandler); + var httpClientFactory = new Mock(); + httpClientFactory + .Setup(x => x.CreateClient(It.IsAny())) + .Returns(httpClient); + var http = new ModularPipelines.Http.Http( + httpClientFactory.Object, + moduleLoggerProvider.Object, + httpLogger, + Microsoft.Extensions.Options.Options.Create(new PipelineOptions())); + + try + { + await Assert.ThrowsAsync(async () => + await http.SendAsync(new HttpOptions( + new HttpRequestMessage(HttpMethod.Get, "https://example.test/stalled-log-body")) + { + LoggingType = HttpLoggingType.Response, + Timeout = timeout, + }) + .WaitAsync(TimeSpan.FromSeconds(1))); + } + finally + { + contentStream.Release(); + } + } + + [Test] + public async Task ResilienceHandler_DisposesRetryableResponseBeforeNextAttempt() + { + var retryContent = new TrackingStringContent("retry"); + var innerHandler = new SequenceResponseHandler( + new HttpResponseMessage(HttpStatusCode.ServiceUnavailable) + { + Content = retryContent, + }, + new HttpResponseMessage(HttpStatusCode.OK)); + var moduleLoggerProvider = new Mock(); + moduleLoggerProvider + .Setup(x => x.GetLogger()) + .Returns(Mock.Of()); + using var handler = new ResilienceHttpHandler( + moduleLoggerProvider.Object, + Microsoft.Extensions.Options.Options.Create(new PipelineOptions + { + DefaultHttpResilienceOptions = new HttpResilienceOptions + { + MaxRetryAttempts = 1, + InitialDelay = TimeSpan.Zero, + JitterFactor = 0, + }, + })) + { + InnerHandler = innerHandler, + }; + using var invoker = new HttpMessageInvoker(handler); + using var request = new HttpRequestMessage(HttpMethod.Get, "https://example.test/retry"); + using var response = await invoker.SendAsync(request, CancellationToken.None); + + using (Assert.Multiple()) + { + await Assert.That(response.StatusCode).IsEqualTo(HttpStatusCode.OK); + await Assert.That(innerHandler.CallCount).IsEqualTo(2); + await Assert.That(retryContent.IsDisposed).IsTrue(); + } + } + [Test] public async Task PublicApi_DoesNotExposeRawHttpClients() { @@ -276,4 +372,27 @@ public override async ValueTask ReadAsync( return 0; } } + + private sealed class SequenceResponseHandler(params HttpResponseMessage[] responses) : HttpMessageHandler + { + public int CallCount { get; private set; } + + protected override Task SendAsync( + HttpRequestMessage request, + CancellationToken cancellationToken) + { + return Task.FromResult(responses[CallCount++]); + } + } + + private sealed class TrackingStringContent(string content) : StringContent(content) + { + public bool IsDisposed { get; private set; } + + protected override void Dispose(bool disposing) + { + IsDisposed = true; + base.Dispose(disposing); + } + } } From f995dd6f5958effa520f7a6c8f405ea1e30cc0e8 Mon Sep 17 00:00:00 2001 From: Tom Longhurst <30480171+thomhurst@users.noreply.github.com> Date: Thu, 30 Jul 2026 16:37:17 +0100 Subject: [PATCH 05/17] fix(http): preserve client body timeouts Carry finite HttpClient.Timeout values into streamed content and dispose the underlying stream when timeout cancellation must interrupt synchronous reads. Refs #3467 --- src/ModularPipelines/Http/Http.cs | 21 ++- .../Http/TimeoutHttpContent.cs | 44 ++++++- .../Context/HttpTests.cs | 122 +++++++++++++++++- 3 files changed, 179 insertions(+), 8 deletions(-) diff --git a/src/ModularPipelines/Http/Http.cs b/src/ModularPipelines/Http/Http.cs index 6591b97191..dd9a07f43e 100644 --- a/src/ModularPipelines/Http/Http.cs +++ b/src/ModularPipelines/Http/Http.cs @@ -28,8 +28,12 @@ public Http( public async Task SendAsync(HttpOptions httpOptions, CancellationToken cancellationToken = default) { - // Priority: options property > pipeline default > no timeout - var effectiveTimeout = httpOptions.Timeout ?? _pipelineOptions.Value.DefaultHttpTimeout; + var httpClient = httpOptions.HttpClient ?? GetHttpClient(httpOptions.LoggingType); + + // Priority: options property > pipeline default > HttpClient timeout + var effectiveTimeout = httpOptions.Timeout + ?? _pipelineOptions.Value.DefaultHttpTimeout + ?? GetFiniteTimeout(httpClient); // Create timeout token if timeout is specified var timeoutCts = effectiveTimeout.HasValue @@ -61,14 +65,13 @@ HttpResponseMessage WrapResponse(HttpResponseMessage response) if (httpOptions.HttpClient != null) { return await SendAndWrapLogging( + httpClient, httpOptions, WrapResponse, effectiveCancellationToken) .ConfigureAwait(false); } - var httpClient = GetHttpClient(httpOptions.LoggingType); - var response = await httpClient.SendAsync( httpOptions.HttpRequestMessage, HttpCompletionOption.ResponseHeadersRead, @@ -107,6 +110,7 @@ private HttpClient GetHttpClient(HttpLoggingType loggingType) } private async Task SendAndWrapLogging( + HttpClient httpClient, HttpOptions httpOptions, Func wrapResponse, CancellationToken cancellationToken) @@ -127,7 +131,7 @@ await _httpLogger var stopWatch = Stopwatch.StartNew(); - var response = await httpOptions.HttpClient!.SendAsync( + var response = await httpClient.SendAsync( httpOptions.HttpRequestMessage, HttpCompletionOption.ResponseHeadersRead, cancellationToken) @@ -176,6 +180,13 @@ private HttpLoggingOptions GetEffectiveLoggingOptions(HttpOptions httpOptions) return HttpLoggingOptions.Default; } + private static TimeSpan? GetFiniteTimeout(HttpClient httpClient) + { + return httpClient.Timeout == System.Threading.Timeout.InfiniteTimeSpan + ? null + : httpClient.Timeout; + } + private void LogDuration(TimeSpan duration, HttpOptions httpOptions, HttpLoggingOptions loggingOptions) { if (httpOptions.LoggingType.HasFlag(HttpLoggingType.Duration) && loggingOptions.LogDuration) diff --git a/src/ModularPipelines/Http/TimeoutHttpContent.cs b/src/ModularPipelines/Http/TimeoutHttpContent.cs index 137ef289f8..2795fd533b 100644 --- a/src/ModularPipelines/Http/TimeoutHttpContent.cs +++ b/src/ModularPipelines/Http/TimeoutHttpContent.cs @@ -48,6 +48,15 @@ protected override async Task CreateContentReadStreamAsync() return new TimeoutReadStream(stream, _linkedCancellationTokenSource.Token); } + protected override Stream CreateContentReadStream(CancellationToken cancellationToken) + { + using var readCancellationTokenSource = CreateReadCancellationTokenSource(cancellationToken); + var effectiveCancellationToken = + readCancellationTokenSource?.Token ?? _linkedCancellationTokenSource.Token; + var stream = _innerContent.ReadAsStream(effectiveCancellationToken); + return new TimeoutReadStream(stream, _linkedCancellationTokenSource.Token); + } + protected override async Task CreateContentReadStreamAsync( CancellationToken cancellationToken) { @@ -99,6 +108,11 @@ private sealed class TimeoutReadStream( Stream innerStream, CancellationToken timeoutCancellationToken) : Stream { + private readonly CancellationTokenRegistration _timeoutRegistration = + timeoutCancellationToken.Register( + static state => ((Stream) state!).Dispose(), + innerStream); + public override bool CanRead => innerStream.CanRead; public override bool CanSeek => innerStream.CanSeek; @@ -123,13 +137,29 @@ public override Task FlushAsync(CancellationToken cancellationToken) public override int Read(byte[] buffer, int offset, int count) { timeoutCancellationToken.ThrowIfCancellationRequested(); - return innerStream.Read(buffer, offset, count); + try + { + return innerStream.Read(buffer, offset, count); + } + catch (ObjectDisposedException exception) + when (timeoutCancellationToken.IsCancellationRequested) + { + throw CreateTimeoutException(exception); + } } public override int Read(Span buffer) { timeoutCancellationToken.ThrowIfCancellationRequested(); - return innerStream.Read(buffer); + try + { + return innerStream.Read(buffer); + } + catch (ObjectDisposedException exception) + when (timeoutCancellationToken.IsCancellationRequested) + { + throw CreateTimeoutException(exception); + } } public override Task ReadAsync( @@ -167,6 +197,7 @@ public override void Write(byte[] buffer, int offset, int count) public override async ValueTask DisposeAsync() { + _timeoutRegistration.Dispose(); await innerStream.DisposeAsync().ConfigureAwait(false); GC.SuppressFinalize(this); } @@ -175,6 +206,7 @@ protected override void Dispose(bool disposing) { if (disposing) { + _timeoutRegistration.Dispose(); innerStream.Dispose(); } @@ -191,5 +223,13 @@ protected override void Dispose(bool disposing) timeoutCancellationToken) : null; } + + private OperationCanceledException CreateTimeoutException(Exception exception) + { + return new OperationCanceledException( + "The HTTP response body read timed out.", + exception, + timeoutCancellationToken); + } } } diff --git a/test/ModularPipelines.UnitTests/Context/HttpTests.cs b/test/ModularPipelines.UnitTests/Context/HttpTests.cs index a1ced185f8..385a2deec0 100644 --- a/test/ModularPipelines.UnitTests/Context/HttpTests.cs +++ b/test/ModularPipelines.UnitTests/Context/HttpTests.cs @@ -83,6 +83,88 @@ await Assert.ThrowsAsync( } } + [Test] + [Arguments(true)] + [Arguments(false)] + public async Task SendAsync_HttpClientTimeoutCancelsStreamedBody(bool useCustomClient) + { + var timeout = TimeSpan.FromMilliseconds(100); + var contentStream = new BlockingReadStream(); + using var httpClient = new HttpClient( + new ImmediateResponseHandler(new StreamContent(contentStream))) + { + Timeout = timeout, + }; + var httpClientFactory = new Mock(); + httpClientFactory + .Setup(x => x.CreateClient(It.IsAny())) + .Returns(httpClient); + var http = new ModularPipelines.Http.Http( + httpClientFactory.Object, + Mock.Of(), + Mock.Of(), + Microsoft.Extensions.Options.Options.Create(new PipelineOptions())); + using var response = await http.SendAsync(new HttpOptions( + new HttpRequestMessage(HttpMethod.Get, "https://example.test/client-timeout")) + { + HttpClient = useCustomClient ? httpClient : null, + LoggingType = HttpLoggingType.None, + }); + var stream = response.Content.ReadAsStream(); + var readTask = stream.ReadAsync(new byte[1]).AsTask(); + + try + { + await Assert.ThrowsAsync( + async () => await readTask.WaitAsync(TimeSpan.FromSeconds(1))); + } + finally + { + contentStream.Release(); + } + } + + [Test] + [Arguments(true)] + [Arguments(false)] + public async Task SendAsync_ConfiguredTimeoutInterruptsSynchronousBodyRead(bool useCopyTo) + { + var timeout = TimeSpan.FromMilliseconds(100); + var contentStream = new BlockingReadStream(); + using var httpClient = new HttpClient( + new ImmediateResponseHandler(new StreamContent(contentStream))); + var result = await GetService((_, _) => { }); + using var response = await result.T.SendAsync(new HttpOptions( + new HttpRequestMessage(HttpMethod.Get, "https://example.test/synchronous-read")) + { + HttpClient = httpClient, + LoggingType = HttpLoggingType.None, + Timeout = timeout, + }); + var stream = await response.Content.ReadAsStreamAsync(); + var readTask = Task.Run(() => + { + if (useCopyTo) + { + stream.CopyTo(Stream.Null); + } + else + { + stream.ReadExactly(new byte[1]); + } + }); + + try + { + await Assert.ThrowsAsync( + async () => await readTask.WaitAsync(TimeSpan.FromSeconds(1))); + } + finally + { + contentStream.Release(); + } + } + [Test] public async Task SendAsync_ConfiguredTimeoutCancelsFactoryResponseLogging() { @@ -361,8 +443,24 @@ private sealed class BlockingReadStream : MemoryStream { private readonly TaskCompletionSource _release = new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly ManualResetEventSlim _synchronousRelease = new(); + private bool _isDisposed; - public void Release() => _release.TrySetResult(); + public void Release() + { + _release.TrySetResult(); + _synchronousRelease.Set(); + } + + public override int Read(byte[] buffer, int offset, int count) + { + return WaitForSynchronousRead(); + } + + public override int Read(Span buffer) + { + return WaitForSynchronousRead(); + } public override async ValueTask ReadAsync( Memory buffer, @@ -371,6 +469,28 @@ public override async ValueTask ReadAsync( await _release.Task.WaitAsync(cancellationToken); return 0; } + + protected override void Dispose(bool disposing) + { + if (disposing) + { + _isDisposed = true; + Release(); + } + + base.Dispose(disposing); + } + + private int WaitForSynchronousRead() + { + _synchronousRelease.Wait(); + if (_isDisposed) + { + throw new ObjectDisposedException(nameof(BlockingReadStream)); + } + + return 0; + } } private sealed class SequenceResponseHandler(params HttpResponseMessage[] responses) : HttpMessageHandler From 65c785bc44fb7a3afce22862949b4841c1bc8284 Mon Sep 17 00:00:00 2001 From: Tom Longhurst <30480171+thomhurst@users.noreply.github.com> Date: Thu, 30 Jul 2026 17:37:44 +0100 Subject: [PATCH 06/17] fix(http): preserve replay and timeouts --- src/ModularPipelines/Http/Http.cs | 3 +- .../Http/HttpContentPreviewReader.cs | 42 +++++++- .../Http/HttpRequestFormatter.cs | 3 +- .../Http/TimeoutHttpContent.cs | 24 +++-- .../Context/HttpFormatterTests.cs | 21 +++- .../Context/HttpTests.cs | 97 +++++++++++++++++-- 6 files changed, 167 insertions(+), 23 deletions(-) diff --git a/src/ModularPipelines/Http/Http.cs b/src/ModularPipelines/Http/Http.cs index dd9a07f43e..0bab5979f5 100644 --- a/src/ModularPipelines/Http/Http.cs +++ b/src/ModularPipelines/Http/Http.cs @@ -137,7 +137,6 @@ await _httpLogger cancellationToken) .ConfigureAwait(false); - response = wrapResponse(response); try { LogStatusCode(response.StatusCode, httpOptions, loggingOptions); @@ -150,6 +149,8 @@ await _httpLogger .ConfigureAwait(false); } + response = wrapResponse(response); + if (!httpOptions.ThrowOnNonSuccessStatusCode) { return response; diff --git a/src/ModularPipelines/Http/HttpContentPreviewReader.cs b/src/ModularPipelines/Http/HttpContentPreviewReader.cs index 7953cdf4a2..1e2b27604b 100644 --- a/src/ModularPipelines/Http/HttpContentPreviewReader.cs +++ b/src/ModularPipelines/Http/HttpContentPreviewReader.cs @@ -54,6 +54,35 @@ internal static class HttpContentPreviewReader return (preview, isTruncated, replayContent, content.Headers.ContentLength); } + /// + /// Reads request content into replayable storage so redirects and authentication retries can resend it. + /// + /// The HTTP request content to preview. + /// The maximum number of bytes to include in the preview. + /// The cancellation token. + /// The preview, truncation state, replayable content, and total length. + public static async Task<(string Preview, bool IsTruncated, HttpContent ReplayContent, long? TotalLength)> + ReadReplayableAsync( + HttpContent content, + int maxBytes, + CancellationToken cancellationToken) + { + var bytes = await content.ReadAsByteArrayAsync(cancellationToken).ConfigureAwait(false); + var previewLength = maxBytes <= 0 || maxBytes == int.MaxValue + ? bytes.Length + : Math.Min(bytes.Length, maxBytes); + var isTruncated = previewLength < bytes.Length; + var preview = DecodeCompletePrefix(GetEncoding(content), bytes, previewLength); + var replayContent = new ByteArrayContent(bytes); + CopyHeaders(content, replayContent); + + return ( + preview, + isTruncated, + replayContent, + content.Headers.ContentLength ?? bytes.LongLength); + } + private static Encoding GetEncoding(HttpContent content) { var charset = content.Headers.ContentType?.CharSet?.Trim('"'); @@ -93,14 +122,19 @@ private static StreamContent CreateReplayContent( { var replayContent = new StreamContent( new PrefixReplayStream(prefix, remainingStream, originalContent)); - foreach (var header in originalContent.Headers) - { - replayContent.Headers.TryAddWithoutValidation(header.Key, header.Value); - } + CopyHeaders(originalContent, replayContent); return replayContent; } + private static void CopyHeaders(HttpContent source, HttpContent destination) + { + foreach (var header in source.Headers) + { + destination.Headers.TryAddWithoutValidation(header.Key, header.Value); + } + } + private sealed class PrefixReplayStream( byte[] prefix, Stream remainingStream, diff --git a/src/ModularPipelines/Http/HttpRequestFormatter.cs b/src/ModularPipelines/Http/HttpRequestFormatter.cs index 4ea3774f7f..289c296d4f 100644 --- a/src/ModularPipelines/Http/HttpRequestFormatter.cs +++ b/src/ModularPipelines/Http/HttpRequestFormatter.cs @@ -160,9 +160,10 @@ private async Task AppendBodyAsync( } var (body, isTruncated, replayContent, totalLength) = await HttpContentPreviewReader - .ReadAsync(content, maxBodySize, cancellationToken) + .ReadReplayableAsync(content, maxBodySize, cancellationToken) .ConfigureAwait(false); request.Content = replayContent; + content.Dispose(); if (string.IsNullOrWhiteSpace(body) && !isTruncated) { diff --git a/src/ModularPipelines/Http/TimeoutHttpContent.cs b/src/ModularPipelines/Http/TimeoutHttpContent.cs index 2795fd533b..827464040c 100644 --- a/src/ModularPipelines/Http/TimeoutHttpContent.cs +++ b/src/ModularPipelines/Http/TimeoutHttpContent.cs @@ -25,6 +25,7 @@ public TimeoutHttpContent( protected override Task SerializeToStreamAsync(Stream stream, TransportContext? context) { + _linkedCancellationTokenSource.Token.ThrowIfCancellationRequested(); return _innerContent.CopyToAsync(stream, _linkedCancellationTokenSource.Token); } @@ -34,14 +35,18 @@ protected override async Task SerializeToStreamAsync( CancellationToken cancellationToken) { using var readCancellationTokenSource = CreateReadCancellationTokenSource(cancellationToken); + var effectiveCancellationToken = + readCancellationTokenSource?.Token ?? _linkedCancellationTokenSource.Token; + effectiveCancellationToken.ThrowIfCancellationRequested(); await _innerContent.CopyToAsync( stream, - readCancellationTokenSource?.Token ?? _linkedCancellationTokenSource.Token) + effectiveCancellationToken) .ConfigureAwait(false); } protected override async Task CreateContentReadStreamAsync() { + _linkedCancellationTokenSource.Token.ThrowIfCancellationRequested(); var stream = await _innerContent .ReadAsStreamAsync(_linkedCancellationTokenSource.Token) .ConfigureAwait(false); @@ -53,6 +58,7 @@ protected override Stream CreateContentReadStream(CancellationToken cancellation using var readCancellationTokenSource = CreateReadCancellationTokenSource(cancellationToken); var effectiveCancellationToken = readCancellationTokenSource?.Token ?? _linkedCancellationTokenSource.Token; + effectiveCancellationToken.ThrowIfCancellationRequested(); var stream = _innerContent.ReadAsStream(effectiveCancellationToken); return new TimeoutReadStream(stream, _linkedCancellationTokenSource.Token); } @@ -63,6 +69,7 @@ protected override async Task CreateContentReadStreamAsync( using var readCancellationTokenSource = CreateReadCancellationTokenSource(cancellationToken); var effectiveCancellationToken = readCancellationTokenSource?.Token ?? _linkedCancellationTokenSource.Token; + effectiveCancellationToken.ThrowIfCancellationRequested(); var stream = await _innerContent .ReadAsStreamAsync(effectiveCancellationToken) .ConfigureAwait(false); @@ -141,8 +148,9 @@ public override int Read(byte[] buffer, int offset, int count) { return innerStream.Read(buffer, offset, count); } - catch (ObjectDisposedException exception) - when (timeoutCancellationToken.IsCancellationRequested) + catch (Exception exception) + when (timeoutCancellationToken.IsCancellationRequested && + exception is ObjectDisposedException or IOException) { throw CreateTimeoutException(exception); } @@ -155,8 +163,9 @@ public override int Read(Span buffer) { return innerStream.Read(buffer); } - catch (ObjectDisposedException exception) - when (timeoutCancellationToken.IsCancellationRequested) + catch (Exception exception) + when (timeoutCancellationToken.IsCancellationRequested && + exception is ObjectDisposedException or IOException) { throw CreateTimeoutException(exception); } @@ -177,9 +186,12 @@ public override async ValueTask ReadAsync( { using var readCancellationTokenSource = CreateReadCancellationTokenSource(cancellationToken); + var effectiveCancellationToken = + readCancellationTokenSource?.Token ?? timeoutCancellationToken; + effectiveCancellationToken.ThrowIfCancellationRequested(); return await innerStream.ReadAsync( buffer, - readCancellationTokenSource?.Token ?? timeoutCancellationToken) + effectiveCancellationToken) .ConfigureAwait(false); } diff --git a/test/ModularPipelines.UnitTests/Context/HttpFormatterTests.cs b/test/ModularPipelines.UnitTests/Context/HttpFormatterTests.cs index 85d6f5f20b..e033003c0a 100644 --- a/test/ModularPipelines.UnitTests/Context/HttpFormatterTests.cs +++ b/test/ModularPipelines.UnitTests/Context/HttpFormatterTests.cs @@ -38,7 +38,7 @@ public async Task ResponseFormatter_BoundsPreviewReadAndPreservesBody() } [Test] - public async Task RequestFormatter_BoundsPreviewReadAndPreservesBody() + public async Task RequestFormatter_PreservesBodyForRepeatedSends() { const string body = "abcdefghijklmnopqrstuvwxyz"; var stream = new CountingReadStream(Encoding.UTF8.GetBytes(body)); @@ -52,15 +52,18 @@ public async Task RequestFormatter_BoundsPreviewReadAndPreservesBody() { MaxBodySizeToLog = 5, }); - var bytesReadForPreview = stream.BytesRead; - var replayedBody = await request.Content.ReadAsStringAsync(); + using var firstSend = new MemoryStream(); + using var secondSend = new MemoryStream(); + await request.Content.CopyToAsync(firstSend); + await request.Content.CopyToAsync(secondSend); using (Assert.Multiple()) { - await Assert.That(bytesReadForPreview).IsLessThanOrEqualTo(6); await Assert.That(formatted).Contains("\tabcde"); await Assert.That(formatted).DoesNotContain("abcdef"); - await Assert.That(replayedBody).IsEqualTo(body); + await Assert.That(Encoding.UTF8.GetString(firstSend.ToArray())).IsEqualTo(body); + await Assert.That(Encoding.UTF8.GetString(secondSend.ToArray())).IsEqualTo(body); + await Assert.That(stream.IsDisposed).IsTrue(); } } @@ -229,6 +232,8 @@ private sealed class CountingReadStream(byte[] contents) : MemoryStream(contents { public int BytesRead { get; private set; } + public bool IsDisposed { get; private set; } + public override async ValueTask ReadAsync( Memory buffer, CancellationToken cancellationToken = default) @@ -237,5 +242,11 @@ public override async ValueTask ReadAsync( BytesRead += read; return read; } + + protected override void Dispose(bool disposing) + { + IsDisposed = true; + base.Dispose(disposing); + } } } diff --git a/test/ModularPipelines.UnitTests/Context/HttpTests.cs b/test/ModularPipelines.UnitTests/Context/HttpTests.cs index 385a2deec0..dda010d2ce 100644 --- a/test/ModularPipelines.UnitTests/Context/HttpTests.cs +++ b/test/ModularPipelines.UnitTests/Context/HttpTests.cs @@ -125,12 +125,16 @@ await Assert.ThrowsAsync( } [Test] - [Arguments(true)] - [Arguments(false)] - public async Task SendAsync_ConfiguredTimeoutInterruptsSynchronousBodyRead(bool useCopyTo) + [Arguments(true, true)] + [Arguments(true, false)] + [Arguments(false, true)] + [Arguments(false, false)] + public async Task SendAsync_ConfiguredTimeoutInterruptsSynchronousBodyRead( + bool useCopyTo, + bool throwIOExceptionWhenDisposed) { var timeout = TimeSpan.FromMilliseconds(100); - var contentStream = new BlockingReadStream(); + var contentStream = new BlockingReadStream(throwIOExceptionWhenDisposed); using var httpClient = new HttpClient( new ImmediateResponseHandler(new StreamContent(contentStream))); var result = await GetService((_, _) => { }); @@ -165,6 +169,85 @@ await Assert.ThrowsAsync( } } + [Test] + public async Task SendAsync_CustomClientLogsRawContentBeforeApplyingBodyTimeout() + { + using var httpClient = new HttpClient( + new ImmediateResponseHandler(new StreamContent(new MemoryStream([1, 2, 3])))) + { + Timeout = TimeSpan.FromSeconds(5), + }; + Type? loggedContentType = null; + var httpLogger = new Mock(); + httpLogger + .Setup(x => x.PrintResponse( + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny())) + .Callback( + (response, _, _, _) => loggedContentType = response.Content.GetType()) + .Returns(Task.CompletedTask); + var http = new ModularPipelines.Http.Http( + Mock.Of(), + Mock.Of(), + httpLogger.Object, + Microsoft.Extensions.Options.Options.Create(new PipelineOptions())); + + using var response = await http.SendAsync(new HttpOptions( + new HttpRequestMessage(HttpMethod.Get, "https://example.test/binary")) + { + HttpClient = httpClient, + LoggingType = HttpLoggingType.Response, + }); + + using (Assert.Multiple()) + { + await Assert.That(loggedContentType).IsEqualTo(typeof(StreamContent)); + await Assert.That(response.Content).IsTypeOf(); + } + } + + [Test] + public async Task SendAsync_CustomClientKeepsTimeoutOutsideLoggedReplayContent() + { + var timeout = TimeSpan.FromMilliseconds(100); + var content = new StreamContent( + new MemoryStream("response body"u8.ToArray())); + content.Headers.ContentType = new MediaTypeHeaderValue("text/plain"); + using var httpClient = new HttpClient(new ImmediateResponseHandler(content)); + var moduleLoggerProvider = new Mock(); + moduleLoggerProvider + .Setup(x => x.GetLogger()) + .Returns(Mock.Of()); + var httpLogger = new HttpLogger( + Mock.Of(), + new HttpResponseFormatter( + Mock.Of(), + Mock.Of(), + Microsoft.Extensions.Options.Options.Create(new SecretMaskingOptions()))); + var http = new ModularPipelines.Http.Http( + Mock.Of(), + moduleLoggerProvider.Object, + httpLogger, + Microsoft.Extensions.Options.Options.Create(new PipelineOptions())); + using var response = await http.SendAsync(new HttpOptions( + new HttpRequestMessage(HttpMethod.Get, "https://example.test/logged-timeout")) + { + HttpClient = httpClient, + LoggingType = HttpLoggingType.Response, + Timeout = timeout, + }); + + await Task.Delay(timeout + TimeSpan.FromMilliseconds(50)); + + await Assert.ThrowsAsync(async () => + { + var stream = await response.Content.ReadAsStreamAsync(); + await stream.ReadExactlyAsync(new byte[1]); + }); + } + [Test] public async Task SendAsync_ConfiguredTimeoutCancelsFactoryResponseLogging() { @@ -439,7 +522,7 @@ protected override bool TryComputeLength(out long length) } } - private sealed class BlockingReadStream : MemoryStream + private sealed class BlockingReadStream(bool throwIOExceptionWhenDisposed = false) : MemoryStream { private readonly TaskCompletionSource _release = new(TaskCreationOptions.RunContinuationsAsynchronously); @@ -486,7 +569,9 @@ private int WaitForSynchronousRead() _synchronousRelease.Wait(); if (_isDisposed) { - throw new ObjectDisposedException(nameof(BlockingReadStream)); + throw throwIOExceptionWhenDisposed + ? new IOException("The stream was interrupted.") + : new ObjectDisposedException(nameof(BlockingReadStream)); } return 0; From 88630f1209c05f548692f2ad8114839146eaf56b Mon Sep 17 00:00:00 2001 From: Tom Longhurst <30480171+thomhurst@users.noreply.github.com> Date: Thu, 30 Jul 2026 21:05:11 +0100 Subject: [PATCH 07/17] fix(http): close streamed response lifecycle --- src/ModularPipelines/Context/Downloader.cs | 4 +- src/ModularPipelines/Http/Http.cs | 9 +- .../Http/TimeoutHttpContent.cs | 23 +++-- .../Context/HttpTests.cs | 98 ++++++++++++++++++- .../Helpers/DownloaderTests.cs | 73 ++++++++++++++ 5 files changed, 191 insertions(+), 16 deletions(-) diff --git a/src/ModularPipelines/Context/Downloader.cs b/src/ModularPipelines/Context/Downloader.cs index 332d020933..5f9ca372f8 100644 --- a/src/ModularPipelines/Context/Downloader.cs +++ b/src/ModularPipelines/Context/Downloader.cs @@ -25,14 +25,14 @@ public Downloader(IModuleLoggerProvider moduleLoggerProvider, IHttpContext http, public async Task DownloadStringAsync(DownloadOptions options, CancellationToken cancellationToken = default) { - var response = await DownloadResponseAsync(options, cancellationToken).ConfigureAwait(false); + using var response = await DownloadResponseAsync(options, cancellationToken).ConfigureAwait(false); return await response.Content.ReadAsStringAsync(cancellationToken).ConfigureAwait(false); } public async Task DownloadFileAsync(DownloadFileOptions options, CancellationToken cancellationToken = default) { - var response = await DownloadResponseAsync(options, cancellationToken).ConfigureAwait(false); + using var response = await DownloadResponseAsync(options, cancellationToken).ConfigureAwait(false); var stream = await response.Content.ReadAsStreamAsync(cancellationToken).ConfigureAwait(false); await using (stream.ConfigureAwait(false)) diff --git a/src/ModularPipelines/Http/Http.cs b/src/ModularPipelines/Http/Http.cs index 0bab5979f5..7517ce645b 100644 --- a/src/ModularPipelines/Http/Http.cs +++ b/src/ModularPipelines/Http/Http.cs @@ -40,10 +40,13 @@ public async Task SendAsync(HttpOptions httpOptions, Cancel ? new CancellationTokenSource(effectiveTimeout.Value) : null; - // Link the timeout token with the passed cancellation token, or just use the original if no timeout + // Keep caller cancellation alive through streamed body consumption even when no + // finite timeout applies. var linkedCts = timeoutCts != null ? CancellationTokenSource.CreateLinkedTokenSource(timeoutCts.Token, cancellationToken) - : null; + : cancellationToken.CanBeCanceled + ? CancellationTokenSource.CreateLinkedTokenSource(cancellationToken) + : null; var effectiveCancellationToken = linkedCts?.Token ?? cancellationToken; @@ -51,7 +54,7 @@ public async Task SendAsync(HttpOptions httpOptions, Cancel { HttpResponseMessage WrapResponse(HttpResponseMessage response) { - if (timeoutCts is null || linkedCts is null) + if (linkedCts is null) { return response; } diff --git a/src/ModularPipelines/Http/TimeoutHttpContent.cs b/src/ModularPipelines/Http/TimeoutHttpContent.cs index 827464040c..6b42c91716 100644 --- a/src/ModularPipelines/Http/TimeoutHttpContent.cs +++ b/src/ModularPipelines/Http/TimeoutHttpContent.cs @@ -5,12 +5,12 @@ namespace ModularPipelines.Http; internal sealed class TimeoutHttpContent : HttpContent { private readonly HttpContent _innerContent; - private readonly CancellationTokenSource _timeoutCancellationTokenSource; + private readonly CancellationTokenSource? _timeoutCancellationTokenSource; private readonly CancellationTokenSource _linkedCancellationTokenSource; public TimeoutHttpContent( HttpContent innerContent, - CancellationTokenSource timeoutCancellationTokenSource, + CancellationTokenSource? timeoutCancellationTokenSource, CancellationTokenSource linkedCancellationTokenSource) { _innerContent = innerContent; @@ -94,7 +94,7 @@ protected override void Dispose(bool disposing) { _innerContent.Dispose(); _linkedCancellationTokenSource.Dispose(); - _timeoutCancellationTokenSource.Dispose(); + _timeoutCancellationTokenSource?.Dispose(); } base.Dispose(disposing); @@ -189,10 +189,19 @@ public override async ValueTask ReadAsync( var effectiveCancellationToken = readCancellationTokenSource?.Token ?? timeoutCancellationToken; effectiveCancellationToken.ThrowIfCancellationRequested(); - return await innerStream.ReadAsync( - buffer, - effectiveCancellationToken) - .ConfigureAwait(false); + try + { + return await innerStream.ReadAsync( + buffer, + effectiveCancellationToken) + .ConfigureAwait(false); + } + catch (Exception exception) + when (timeoutCancellationToken.IsCancellationRequested && + exception is ObjectDisposedException or IOException) + { + throw CreateTimeoutException(exception); + } } public override long Seek(long offset, SeekOrigin origin) diff --git a/test/ModularPipelines.UnitTests/Context/HttpTests.cs b/test/ModularPipelines.UnitTests/Context/HttpTests.cs index dda010d2ce..fc8ff66c27 100644 --- a/test/ModularPipelines.UnitTests/Context/HttpTests.cs +++ b/test/ModularPipelines.UnitTests/Context/HttpTests.cs @@ -124,6 +124,81 @@ await Assert.ThrowsAsync( } } + [Test] + public async Task SendAsync_InfiniteTimeoutRetainsCallerCancellationForStreamedBody() + { + var contentStream = new BlockingReadStream(); + using var httpClient = new HttpClient( + new ImmediateResponseHandler(new StreamContent(contentStream))) + { + Timeout = Timeout.InfiniteTimeSpan, + }; + var http = new ModularPipelines.Http.Http( + Mock.Of(), + Mock.Of(), + Mock.Of(), + Microsoft.Extensions.Options.Options.Create(new PipelineOptions())); + using var cancellationTokenSource = new CancellationTokenSource(); + using var response = await http.SendAsync(new HttpOptions( + new HttpRequestMessage(HttpMethod.Get, "https://example.test/caller-cancellation")) + { + HttpClient = httpClient, + LoggingType = HttpLoggingType.None, + }, cancellationTokenSource.Token); + var stream = response.Content.ReadAsStream(); + var readTask = stream.ReadAsync(new byte[1]).AsTask(); + + cancellationTokenSource.Cancel(); + + try + { + await Assert.ThrowsAsync( + async () => await readTask.WaitAsync(TimeSpan.FromSeconds(1))); + } + finally + { + contentStream.Release(); + } + } + + [Test] + [Arguments(true)] + [Arguments(false)] + public async Task SendAsync_NormalizesAsyncStreamFailureAfterTimeout( + bool throwIOExceptionWhenDisposed) + { + var timeout = TimeSpan.FromMilliseconds(100); + var contentStream = new BlockingReadStream( + throwIOExceptionWhenDisposed, + ignoreAsyncCancellation: true); + using var httpClient = new HttpClient( + new ImmediateResponseHandler(new StreamContent(contentStream))); + var http = new ModularPipelines.Http.Http( + Mock.Of(), + Mock.Of(), + Mock.Of(), + Microsoft.Extensions.Options.Options.Create(new PipelineOptions())); + using var response = await http.SendAsync(new HttpOptions( + new HttpRequestMessage(HttpMethod.Get, "https://example.test/async-timeout")) + { + HttpClient = httpClient, + LoggingType = HttpLoggingType.None, + Timeout = timeout, + }); + var stream = response.Content.ReadAsStream(); + var readTask = stream.ReadAsync(new byte[1]).AsTask(); + + try + { + await Assert.ThrowsAsync( + async () => await readTask.WaitAsync(TimeSpan.FromSeconds(1))); + } + finally + { + contentStream.Release(); + } + } + [Test] [Arguments(true, true)] [Arguments(true, false)] @@ -522,7 +597,9 @@ protected override bool TryComputeLength(out long length) } } - private sealed class BlockingReadStream(bool throwIOExceptionWhenDisposed = false) : MemoryStream + private sealed class BlockingReadStream( + bool throwIOExceptionWhenDisposed = false, + bool ignoreAsyncCancellation = false) : MemoryStream { private readonly TaskCompletionSource _release = new(TaskCreationOptions.RunContinuationsAsynchronously); @@ -549,7 +626,16 @@ public override async ValueTask ReadAsync( Memory buffer, CancellationToken cancellationToken = default) { - await _release.Task.WaitAsync(cancellationToken); + if (ignoreAsyncCancellation) + { + await _release.Task; + ThrowIfDisposed(); + } + else + { + await _release.Task.WaitAsync(cancellationToken); + } + return 0; } @@ -567,14 +653,18 @@ protected override void Dispose(bool disposing) private int WaitForSynchronousRead() { _synchronousRelease.Wait(); + ThrowIfDisposed(); + return 0; + } + + private void ThrowIfDisposed() + { if (_isDisposed) { throw throwIOExceptionWhenDisposed ? new IOException("The stream was interrupted.") : new ObjectDisposedException(nameof(BlockingReadStream)); } - - return 0; } } diff --git a/test/ModularPipelines.UnitTests/Helpers/DownloaderTests.cs b/test/ModularPipelines.UnitTests/Helpers/DownloaderTests.cs index fb68a482bd..7137fb8a7d 100644 --- a/test/ModularPipelines.UnitTests/Helpers/DownloaderTests.cs +++ b/test/ModularPipelines.UnitTests/Helpers/DownloaderTests.cs @@ -1,8 +1,12 @@ +using System.Net; using ModularPipelines.Context; using ModularPipelines.Context.Domains.Files; using ModularPipelines.Context.Domains.Network; +using ModularPipelines.FileSystem; +using ModularPipelines.Logging; using ModularPipelines.Options; using ModularPipelines.TestHelpers; +using Moq; namespace ModularPipelines.UnitTests.Helpers; @@ -27,4 +31,73 @@ public async Task Can_Download() await Assert.That(checksum.Md5(file)).IsEqualTo("AEDF5D7C23744269F358814E602AFE89"); } + + [Test] + public async Task DownloadStringAsync_DisposesResponse() + { + var content = new TrackingStringContent("download"); + var downloader = CreateDownloader(content); + + var result = await downloader.DownloadStringAsync( + new DownloadOptions(new Uri("https://example.test/download"))); + + using (Assert.Multiple()) + { + await Assert.That(result).IsEqualTo("download"); + await Assert.That(content.IsDisposed).IsTrue(); + } + } + + [Test] + public async Task DownloadFileAsync_DisposesResponse() + { + var content = new TrackingStringContent("download"); + var fileSystemProvider = new Mock(); + fileSystemProvider + .Setup(x => x.Create(It.IsAny())) + .Returns(() => new MemoryStream()); + var downloader = CreateDownloader(content, fileSystemProvider.Object); + + await downloader.DownloadFileAsync(new DownloadFileOptions( + new Uri("https://example.test/download")) + { + SavePath = "download.bin", + }); + + await Assert.That(content.IsDisposed).IsTrue(); + } + + private static Downloader CreateDownloader( + HttpContent content, + IFileSystemProvider? fileSystemProvider = null) + { + var http = new Mock(); + http.Setup(x => x.SendAsync( + It.IsAny(), + It.IsAny())) + .ReturnsAsync(new HttpResponseMessage(HttpStatusCode.OK) + { + Content = content, + }); + var moduleLoggerProvider = new Mock(); + moduleLoggerProvider + .Setup(x => x.GetLogger()) + .Returns(Mock.Of()); + + return new Downloader( + moduleLoggerProvider.Object, + http.Object, + fileSystemProvider ?? SystemFileSystemProvider.Instance); + } + + private sealed class TrackingStringContent(string content) : StringContent(content) + { + public bool IsDisposed { get; private set; } + + protected override void Dispose(bool disposing) + { + IsDisposed = true; + base.Dispose(disposing); + } + } } From e0e6eb34e196fc67703d9d1e7ac273ea0b500718 Mon Sep 17 00:00:00 2001 From: Tom Longhurst <30480171+thomhurst@users.noreply.github.com> Date: Thu, 30 Jul 2026 21:19:36 +0100 Subject: [PATCH 08/17] fix(http): close preview failure paths --- src/ModularPipelines/Context/Downloader.cs | 10 +++++- .../Http/HttpContentPreviewReader.cs | 33 ++++++++++++----- .../Context/HttpTests.cs | 35 +++++++++++++++++++ .../Helpers/DownloaderTests.cs | 19 ++++++++-- 4 files changed, 85 insertions(+), 12 deletions(-) diff --git a/src/ModularPipelines/Context/Downloader.cs b/src/ModularPipelines/Context/Downloader.cs index 5f9ca372f8..0ff2745e23 100644 --- a/src/ModularPipelines/Context/Downloader.cs +++ b/src/ModularPipelines/Context/Downloader.cs @@ -68,7 +68,15 @@ public async Task DownloadResponseAsync(DownloadOptions opt HttpClient = options.HttpClient, }, cancellationToken).ConfigureAwait(false); - return await response.EnsureSuccessStatusCodeWithContentAsync(cancellationToken).ConfigureAwait(false); + try + { + return await response.EnsureSuccessStatusCodeWithContentAsync(cancellationToken).ConfigureAwait(false); + } + catch + { + response.Dispose(); + throw; + } } /// diff --git a/src/ModularPipelines/Http/HttpContentPreviewReader.cs b/src/ModularPipelines/Http/HttpContentPreviewReader.cs index 1e2b27604b..135e121fa2 100644 --- a/src/ModularPipelines/Http/HttpContentPreviewReader.cs +++ b/src/ModularPipelines/Http/HttpContentPreviewReader.cs @@ -28,21 +28,36 @@ internal static class HttpContentPreviewReader } var stream = await content.ReadAsStreamAsync(cancellationToken).ConfigureAwait(false); + using var cancellationRegistration = cancellationToken.Register( + static state => ((Stream) state!).Dispose(), + stream); var buffer = new byte[maxBytes + 1]; var bytesRead = 0; - while (bytesRead < buffer.Length) + try { - var read = await stream.ReadAsync( - buffer.AsMemory(bytesRead, buffer.Length - bytesRead), - cancellationToken) - .ConfigureAwait(false); - if (read == 0) + while (bytesRead < buffer.Length) { - break; + var read = await stream.ReadAsync( + buffer.AsMemory(bytesRead, buffer.Length - bytesRead), + cancellationToken) + .ConfigureAwait(false); + if (read == 0) + { + break; + } + + bytesRead += read; } - - bytesRead += read; + } + catch (Exception exception) + when (cancellationToken.IsCancellationRequested + && exception is ObjectDisposedException or IOException) + { + throw new OperationCanceledException( + "The HTTP content preview read was cancelled.", + exception, + cancellationToken); } var isTruncated = bytesRead > maxBytes; diff --git a/test/ModularPipelines.UnitTests/Context/HttpTests.cs b/test/ModularPipelines.UnitTests/Context/HttpTests.cs index fc8ff66c27..2e05d21407 100644 --- a/test/ModularPipelines.UnitTests/Context/HttpTests.cs +++ b/test/ModularPipelines.UnitTests/Context/HttpTests.cs @@ -283,6 +283,41 @@ public async Task SendAsync_CustomClientLogsRawContentBeforeApplyingBodyTimeout( } } + [Test] + public async Task SendAsync_CustomClientTimeoutInterruptsNonCooperativeResponseLogging() + { + var timeout = TimeSpan.FromMilliseconds(100); + var contentStream = new BlockingReadStream(ignoreAsyncCancellation: true); + var content = new StreamContent(contentStream); + content.Headers.ContentType = new MediaTypeHeaderValue("text/plain"); + using var httpClient = new HttpClient(new ImmediateResponseHandler(content)); + var moduleLoggerProvider = new Mock(); + moduleLoggerProvider + .Setup(x => x.GetLogger()) + .Returns(Mock.Of()); + var httpLogger = new HttpLogger( + Mock.Of(), + new HttpResponseFormatter( + Mock.Of(), + Mock.Of(), + Microsoft.Extensions.Options.Options.Create(new SecretMaskingOptions()))); + var http = new ModularPipelines.Http.Http( + Mock.Of(), + moduleLoggerProvider.Object, + httpLogger, + Microsoft.Extensions.Options.Options.Create(new PipelineOptions())); + + await Assert.ThrowsAsync(async () => + await http.SendAsync(new HttpOptions( + new HttpRequestMessage(HttpMethod.Get, "https://example.test/stalled-custom-log-body")) + { + HttpClient = httpClient, + LoggingType = HttpLoggingType.Response, + Timeout = timeout, + }) + .WaitAsync(TimeSpan.FromSeconds(1))); + } + [Test] public async Task SendAsync_CustomClientKeepsTimeoutOutsideLoggedReplayContent() { diff --git a/test/ModularPipelines.UnitTests/Helpers/DownloaderTests.cs b/test/ModularPipelines.UnitTests/Helpers/DownloaderTests.cs index 7137fb8a7d..05e8edf1ba 100644 --- a/test/ModularPipelines.UnitTests/Helpers/DownloaderTests.cs +++ b/test/ModularPipelines.UnitTests/Helpers/DownloaderTests.cs @@ -2,6 +2,7 @@ using ModularPipelines.Context; using ModularPipelines.Context.Domains.Files; using ModularPipelines.Context.Domains.Network; +using ModularPipelines.Exceptions; using ModularPipelines.FileSystem; using ModularPipelines.Logging; using ModularPipelines.Options; @@ -67,15 +68,29 @@ await downloader.DownloadFileAsync(new DownloadFileOptions( await Assert.That(content.IsDisposed).IsTrue(); } + [Test] + public async Task DownloadResponseAsync_DisposesResponseWhenStatusValidationFails() + { + var content = new TrackingStringContent("failure"); + var downloader = CreateDownloader(content, statusCode: HttpStatusCode.BadRequest); + + await Assert.ThrowsAsync(() => + downloader.DownloadResponseAsync( + new DownloadOptions(new Uri("https://example.test/download")))); + + await Assert.That(content.IsDisposed).IsTrue(); + } + private static Downloader CreateDownloader( HttpContent content, - IFileSystemProvider? fileSystemProvider = null) + IFileSystemProvider? fileSystemProvider = null, + HttpStatusCode statusCode = HttpStatusCode.OK) { var http = new Mock(); http.Setup(x => x.SendAsync( It.IsAny(), It.IsAny())) - .ReturnsAsync(new HttpResponseMessage(HttpStatusCode.OK) + .ReturnsAsync(new HttpResponseMessage(statusCode) { Content = content, }); From ec099dad4cd3dd12ba88321709c3a269ad47ee69 Mon Sep 17 00:00:00 2001 From: Tom Longhurst <30480171+thomhurst@users.noreply.github.com> Date: Thu, 30 Jul 2026 21:40:50 +0100 Subject: [PATCH 09/17] fix(http): interrupt buffered body reads --- .../Http/HttpContentPreviewReader.cs | 44 +++++++++++++++--- .../Http/TimeoutHttpContent.cs | 45 ++++++++++++++----- .../Context/HttpTests.cs | 33 +++++++++++++- 3 files changed, 106 insertions(+), 16 deletions(-) diff --git a/src/ModularPipelines/Http/HttpContentPreviewReader.cs b/src/ModularPipelines/Http/HttpContentPreviewReader.cs index 135e121fa2..0bef864af4 100644 --- a/src/ModularPipelines/Http/HttpContentPreviewReader.cs +++ b/src/ModularPipelines/Http/HttpContentPreviewReader.cs @@ -22,9 +22,15 @@ internal static class HttpContentPreviewReader { if (maxBytes <= 0 || maxBytes == int.MaxValue) { - // ReadAsStringAsync buffers HttpContent, so returning the same instance preserves replay. - var unboundedBody = await content.ReadAsStringAsync(cancellationToken).ConfigureAwait(false); - return (unboundedBody, false, content, content.Headers.ContentLength); + var totalLength = content.Headers.ContentLength; + var bytes = await ReadAllBytesAsync(content, cancellationToken).ConfigureAwait(false); + var unboundedReplayContent = new ByteArrayContent(bytes); + CopyHeaders(content, unboundedReplayContent); + content.Dispose(); + return (DecodeCompletePrefix(GetEncoding(unboundedReplayContent), bytes, bytes.Length), + false, + unboundedReplayContent, + totalLength ?? bytes.LongLength); } var stream = await content.ReadAsStreamAsync(cancellationToken).ConfigureAwait(false); @@ -82,7 +88,8 @@ internal static class HttpContentPreviewReader int maxBytes, CancellationToken cancellationToken) { - var bytes = await content.ReadAsByteArrayAsync(cancellationToken).ConfigureAwait(false); + var totalLength = content.Headers.ContentLength; + var bytes = await ReadAllBytesAsync(content, cancellationToken).ConfigureAwait(false); var previewLength = maxBytes <= 0 || maxBytes == int.MaxValue ? bytes.Length : Math.Min(bytes.Length, maxBytes); @@ -90,12 +97,39 @@ internal static class HttpContentPreviewReader var preview = DecodeCompletePrefix(GetEncoding(content), bytes, previewLength); var replayContent = new ByteArrayContent(bytes); CopyHeaders(content, replayContent); + content.Dispose(); return ( preview, isTruncated, replayContent, - content.Headers.ContentLength ?? bytes.LongLength); + totalLength ?? bytes.LongLength); + } + + private static async Task ReadAllBytesAsync( + HttpContent content, + CancellationToken cancellationToken) + { + var stream = await content.ReadAsStreamAsync(cancellationToken).ConfigureAwait(false); + using var cancellationRegistration = cancellationToken.Register( + static state => ((Stream) state!).Dispose(), + stream); + using var buffer = new MemoryStream(); + + try + { + await stream.CopyToAsync(buffer, cancellationToken).ConfigureAwait(false); + return buffer.ToArray(); + } + catch (Exception exception) + when (cancellationToken.IsCancellationRequested + && exception is ObjectDisposedException or IOException) + { + throw new OperationCanceledException( + "The HTTP content read was cancelled.", + exception, + cancellationToken); + } } private static Encoding GetEncoding(HttpContent content) diff --git a/src/ModularPipelines/Http/TimeoutHttpContent.cs b/src/ModularPipelines/Http/TimeoutHttpContent.cs index 6b42c91716..75e0a50333 100644 --- a/src/ModularPipelines/Http/TimeoutHttpContent.cs +++ b/src/ModularPipelines/Http/TimeoutHttpContent.cs @@ -25,8 +25,7 @@ public TimeoutHttpContent( protected override Task SerializeToStreamAsync(Stream stream, TransportContext? context) { - _linkedCancellationTokenSource.Token.ThrowIfCancellationRequested(); - return _innerContent.CopyToAsync(stream, _linkedCancellationTokenSource.Token); + return CopyInnerContentToAsync(stream, CancellationToken.None); } protected override async Task SerializeToStreamAsync( @@ -34,14 +33,7 @@ protected override async Task SerializeToStreamAsync( TransportContext? context, CancellationToken cancellationToken) { - using var readCancellationTokenSource = CreateReadCancellationTokenSource(cancellationToken); - var effectiveCancellationToken = - readCancellationTokenSource?.Token ?? _linkedCancellationTokenSource.Token; - effectiveCancellationToken.ThrowIfCancellationRequested(); - await _innerContent.CopyToAsync( - stream, - effectiveCancellationToken) - .ConfigureAwait(false); + await CopyInnerContentToAsync(stream, cancellationToken).ConfigureAwait(false); } protected override async Task CreateContentReadStreamAsync() @@ -100,6 +92,36 @@ protected override void Dispose(bool disposing) base.Dispose(disposing); } + private async Task CopyInnerContentToAsync( + Stream destination, + CancellationToken cancellationToken) + { + using var readCancellationTokenSource = CreateReadCancellationTokenSource(cancellationToken); + var effectiveCancellationToken = + readCancellationTokenSource?.Token ?? _linkedCancellationTokenSource.Token; + effectiveCancellationToken.ThrowIfCancellationRequested(); + var source = await _innerContent + .ReadAsStreamAsync(effectiveCancellationToken) + .ConfigureAwait(false); + using var cancellationRegistration = effectiveCancellationToken.Register( + static state => ((Stream) state!).Dispose(), + source); + + try + { + await source.CopyToAsync(destination, effectiveCancellationToken).ConfigureAwait(false); + } + catch (Exception exception) + when (effectiveCancellationToken.IsCancellationRequested + && exception is ObjectDisposedException or IOException) + { + throw new OperationCanceledException( + "The HTTP response body copy was cancelled.", + exception, + effectiveCancellationToken); + } + } + private CancellationTokenSource? CreateReadCancellationTokenSource( CancellationToken cancellationToken) { @@ -189,6 +211,9 @@ public override async ValueTask ReadAsync( var effectiveCancellationToken = readCancellationTokenSource?.Token ?? timeoutCancellationToken; effectiveCancellationToken.ThrowIfCancellationRequested(); + using var cancellationRegistration = readCancellationTokenSource?.Token.Register( + static state => ((Stream) state!).Dispose(), + innerStream); try { return await innerStream.ReadAsync( diff --git a/test/ModularPipelines.UnitTests/Context/HttpTests.cs b/test/ModularPipelines.UnitTests/Context/HttpTests.cs index 2e05d21407..9675f71016 100644 --- a/test/ModularPipelines.UnitTests/Context/HttpTests.cs +++ b/test/ModularPipelines.UnitTests/Context/HttpTests.cs @@ -284,7 +284,10 @@ public async Task SendAsync_CustomClientLogsRawContentBeforeApplyingBodyTimeout( } [Test] - public async Task SendAsync_CustomClientTimeoutInterruptsNonCooperativeResponseLogging() + [Arguments(4096)] + [Arguments(0)] + public async Task SendAsync_CustomClientTimeoutInterruptsNonCooperativeResponseLogging( + int maxBodySizeToLog) { var timeout = TimeSpan.FromMilliseconds(100); var contentStream = new BlockingReadStream(ignoreAsyncCancellation: true); @@ -313,11 +316,39 @@ await http.SendAsync(new HttpOptions( { HttpClient = httpClient, LoggingType = HttpLoggingType.Response, + LogSettings = new HttpLoggingOptions + { + MaxBodySizeToLog = maxBodySizeToLog, + }, Timeout = timeout, }) .WaitAsync(TimeSpan.FromSeconds(1))); } + [Test] + public async Task SendAsync_ConfiguredTimeoutInterruptsBufferedBodyCopy() + { + var timeout = TimeSpan.FromMilliseconds(100); + var contentStream = new BlockingReadStream(ignoreAsyncCancellation: true); + using var httpClient = new HttpClient( + new ImmediateResponseHandler(new StreamContent(contentStream))); + var http = new ModularPipelines.Http.Http( + Mock.Of(), + Mock.Of(), + Mock.Of(), + Microsoft.Extensions.Options.Options.Create(new PipelineOptions())); + using var response = await http.SendAsync(new HttpOptions( + new HttpRequestMessage(HttpMethod.Get, "https://example.test/buffered-timeout")) + { + HttpClient = httpClient, + LoggingType = HttpLoggingType.None, + Timeout = timeout, + }); + + await Assert.ThrowsAsync(async () => + await response.Content.ReadAsStringAsync().WaitAsync(TimeSpan.FromSeconds(1))); + } + [Test] public async Task SendAsync_CustomClientKeepsTimeoutOutsideLoggedReplayContent() { From 6e3255e38b3ccebbb80fa527e3ece30ba81188eb Mon Sep 17 00:00:00 2001 From: Tom Longhurst <30480171+thomhurst@users.noreply.github.com> Date: Thu, 30 Jul 2026 22:22:57 +0100 Subject: [PATCH 10/17] fix(http): preserve error-body cancellation --- .../Http/HttpResponseExtensions.cs | 20 ++--- .../Context/HttpTests.cs | 90 ++++++++++++++++++- 2 files changed, 93 insertions(+), 17 deletions(-) diff --git a/src/ModularPipelines/Http/HttpResponseExtensions.cs b/src/ModularPipelines/Http/HttpResponseExtensions.cs index d4cfff64de..c4123c4543 100644 --- a/src/ModularPipelines/Http/HttpResponseExtensions.cs +++ b/src/ModularPipelines/Http/HttpResponseExtensions.cs @@ -30,28 +30,20 @@ public static async Task EnsureSuccessStatusCodeWithContent { responseContent = await response.Content.ReadAsStringAsync(cancellationToken).ConfigureAwait(false); } - catch (ObjectDisposedException) - { - // Response content stream was already disposed - continue with null content - } - catch (InvalidOperationException) - { - // Content stream is in an invalid state - continue with null content - } - catch (HttpRequestException) - { - // Network error while reading content - continue with null content - } catch (OperationCanceledException) { - // Reading was cancelled - continue with null content + throw; } catch (Exception ex) when (ex is not (OutOfMemoryException or StackOverflowException)) { - // Other unexpected errors reading content - continue with null content + cancellationToken.ThrowIfCancellationRequested(); + + // Errors reading content do not hide the primary HTTP status failure. // The primary error information is in the status code and reason phrase } + cancellationToken.ThrowIfCancellationRequested(); + throw new HttpResponseException( response.StatusCode, response.ReasonPhrase, diff --git a/test/ModularPipelines.UnitTests/Context/HttpTests.cs b/test/ModularPipelines.UnitTests/Context/HttpTests.cs index 9675f71016..ffaa46d4b7 100644 --- a/test/ModularPipelines.UnitTests/Context/HttpTests.cs +++ b/test/ModularPipelines.UnitTests/Context/HttpTests.cs @@ -349,6 +349,55 @@ await Assert.ThrowsAsync(async () => await response.Content.ReadAsStringAsync().WaitAsync(TimeSpan.FromSeconds(1))); } + [Test] + [Arguments(true, true)] + [Arguments(true, false)] + [Arguments(false, true)] + [Arguments(false, false)] + public async Task SendAsync_PreservesCancellationWhileReadingErrorContent( + bool useCustomClient, + bool useConfiguredTimeout) + { + var contentStream = new BlockingReadStream(); + using var httpClient = new HttpClient( + new ImmediateResponseHandler( + new StreamContent(contentStream), + HttpStatusCode.InternalServerError)) + { + Timeout = Timeout.InfiniteTimeSpan, + }; + var httpClientFactory = new Mock(); + httpClientFactory + .Setup(x => x.CreateClient(It.IsAny())) + .Returns(httpClient); + var http = new ModularPipelines.Http.Http( + httpClientFactory.Object, + Mock.Of(), + Mock.Of(), + Microsoft.Extensions.Options.Options.Create(new PipelineOptions())); + using var cancellationTokenSource = new CancellationTokenSource(); + cancellationTokenSource.CancelAfter(TimeSpan.FromMilliseconds(100)); + + try + { + await Assert.ThrowsAsync(async () => + await http.SendAsync(new HttpOptions( + new HttpRequestMessage(HttpMethod.Get, "https://example.test/stalled-error-body")) + { + HttpClient = useCustomClient ? httpClient : null, + LoggingType = HttpLoggingType.None, + ThrowOnNonSuccessStatusCode = true, + Timeout = useConfiguredTimeout ? TimeSpan.FromMilliseconds(100) : null, + }, + useConfiguredTimeout ? CancellationToken.None : cancellationTokenSource.Token) + .WaitAsync(TimeSpan.FromSeconds(1))); + } + finally + { + contentStream.Release(); + } + } + [Test] public async Task SendAsync_CustomClientKeepsTimeoutOutsideLoggedReplayContent() { @@ -623,7 +672,9 @@ await http.SendAsync(new HttpOptions(new HttpRequestMessage(HttpMethod.Get, serv } } - private sealed class ImmediateResponseHandler(HttpContent content) : HttpMessageHandler + private sealed class ImmediateResponseHandler( + HttpContent content, + HttpStatusCode statusCode = HttpStatusCode.OK) : HttpMessageHandler { public TaskCompletionSource RequestReceived { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); @@ -633,7 +684,7 @@ protected override Task SendAsync( CancellationToken cancellationToken) { RequestReceived.TrySetResult(); - return Task.FromResult(new HttpResponseMessage(HttpStatusCode.OK) + return Task.FromResult(new HttpResponseMessage(statusCode) { Content = content, RequestMessage = request, @@ -665,19 +716,37 @@ protected override bool TryComputeLength(out long length) private sealed class BlockingReadStream( bool throwIOExceptionWhenDisposed = false, - bool ignoreAsyncCancellation = false) : MemoryStream + bool ignoreAsyncCancellation = false) : Stream { private readonly TaskCompletionSource _release = new(TaskCreationOptions.RunContinuationsAsynchronously); private readonly ManualResetEventSlim _synchronousRelease = new(); private bool _isDisposed; + public override bool CanRead => true; + + public override bool CanSeek => false; + + public override bool CanWrite => false; + + public override long Length => throw new NotSupportedException(); + + public override long Position + { + get => throw new NotSupportedException(); + set => throw new NotSupportedException(); + } + public void Release() { _release.TrySetResult(); _synchronousRelease.Set(); } + public override void Flush() + { + } + public override int Read(byte[] buffer, int offset, int count) { return WaitForSynchronousRead(); @@ -723,6 +792,21 @@ private int WaitForSynchronousRead() return 0; } + public override long Seek(long offset, SeekOrigin origin) + { + throw new NotSupportedException(); + } + + public override void SetLength(long value) + { + throw new NotSupportedException(); + } + + public override void Write(byte[] buffer, int offset, int count) + { + throw new NotSupportedException(); + } + private void ThrowIfDisposed() { if (_isDisposed) From 000fc185ff2f448b7274d3c481d9e34173889e1b Mon Sep 17 00:00:00 2001 From: Tom Longhurst <30480171+thomhurst@users.noreply.github.com> Date: Thu, 30 Jul 2026 22:51:33 +0100 Subject: [PATCH 11/17] fix(http): normalize per-read cancellation --- .../Http/TimeoutHttpContent.cs | 16 +++++- .../Context/HttpTests.cs | 52 +++++++++++++++++++ 2 files changed, 66 insertions(+), 2 deletions(-) diff --git a/src/ModularPipelines/Http/TimeoutHttpContent.cs b/src/ModularPipelines/Http/TimeoutHttpContent.cs index 75e0a50333..4eb7403e26 100644 --- a/src/ModularPipelines/Http/TimeoutHttpContent.cs +++ b/src/ModularPipelines/Http/TimeoutHttpContent.cs @@ -222,10 +222,12 @@ public override async ValueTask ReadAsync( .ConfigureAwait(false); } catch (Exception exception) - when (timeoutCancellationToken.IsCancellationRequested && + when (effectiveCancellationToken.IsCancellationRequested && exception is ObjectDisposedException or IOException) { - throw CreateTimeoutException(exception); + throw CreateReadCancellationException( + exception, + effectiveCancellationToken); } } @@ -277,5 +279,15 @@ private OperationCanceledException CreateTimeoutException(Exception exception) exception, timeoutCancellationToken); } + + private static OperationCanceledException CreateReadCancellationException( + Exception exception, + CancellationToken cancellationToken) + { + return new OperationCanceledException( + "The HTTP response body read was cancelled.", + exception, + cancellationToken); + } } } diff --git a/test/ModularPipelines.UnitTests/Context/HttpTests.cs b/test/ModularPipelines.UnitTests/Context/HttpTests.cs index ffaa46d4b7..05f30c294e 100644 --- a/test/ModularPipelines.UnitTests/Context/HttpTests.cs +++ b/test/ModularPipelines.UnitTests/Context/HttpTests.cs @@ -199,6 +199,53 @@ await Assert.ThrowsAsync( } } + [Test] + [Arguments(true)] + [Arguments(false)] + public async Task SendAsync_NormalizesAsyncStreamFailureAfterPerReadCancellation( + bool throwIOExceptionWhenDisposed) + { + var contentStream = new BlockingReadStream( + throwIOExceptionWhenDisposed, + ignoreAsyncCancellation: true); + using var httpClient = new HttpClient( + new ImmediateResponseHandler(new StreamContent(contentStream))) + { + Timeout = Timeout.InfiniteTimeSpan, + }; + var http = new ModularPipelines.Http.Http( + Mock.Of(), + Mock.Of(), + Mock.Of(), + Microsoft.Extensions.Options.Options.Create(new PipelineOptions())); + using var response = await http.SendAsync(new HttpOptions( + new HttpRequestMessage(HttpMethod.Get, "https://example.test/read-cancellation")) + { + HttpClient = httpClient, + LoggingType = HttpLoggingType.None, + Timeout = TimeSpan.FromMinutes(1), + }); + var stream = response.Content.ReadAsStream(); + using var cancellationTokenSource = new CancellationTokenSource(); + var readTask = stream + .ReadAsync(new byte[1], cancellationTokenSource.Token) + .AsTask(); + + await contentStream.ReadStarted.WaitAsync(TimeSpan.FromSeconds(1)); + cancellationTokenSource.Cancel(); + + try + { + var exception = await Assert.ThrowsAsync( + async () => await readTask.WaitAsync(TimeSpan.FromSeconds(1))); + await Assert.That(exception!.CancellationToken.IsCancellationRequested).IsTrue(); + } + finally + { + contentStream.Release(); + } + } + [Test] [Arguments(true, true)] [Arguments(true, false)] @@ -720,6 +767,8 @@ private sealed class BlockingReadStream( { private readonly TaskCompletionSource _release = new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly TaskCompletionSource _readStarted = + new(TaskCreationOptions.RunContinuationsAsynchronously); private readonly ManualResetEventSlim _synchronousRelease = new(); private bool _isDisposed; @@ -737,6 +786,8 @@ public override long Position set => throw new NotSupportedException(); } + public Task ReadStarted => _readStarted.Task; + public void Release() { _release.TrySetResult(); @@ -761,6 +812,7 @@ public override async ValueTask ReadAsync( Memory buffer, CancellationToken cancellationToken = default) { + _readStarted.TrySetResult(); if (ignoreAsyncCancellation) { await _release.Task; From de4665b9b0a8d53b8e93b12ce9cad95c82564234 Mon Sep 17 00:00:00 2001 From: Tom Longhurst <30480171+thomhurst@users.noreply.github.com> Date: Thu, 30 Jul 2026 23:08:09 +0100 Subject: [PATCH 12/17] fix(download): commit files atomically --- src/ModularPipelines/Context/Downloader.cs | 19 ++++- .../FileSystem/IFileSystemProvider.cs | 10 +++ .../FileSystem/SystemFileSystemProvider.cs | 3 + .../Helpers/DownloaderTests.cs | 77 +++++++++++++++++++ 4 files changed, 106 insertions(+), 3 deletions(-) diff --git a/src/ModularPipelines/Context/Downloader.cs b/src/ModularPipelines/Context/Downloader.cs index 0ff2745e23..af4ba89500 100644 --- a/src/ModularPipelines/Context/Downloader.cs +++ b/src/ModularPipelines/Context/Downloader.cs @@ -44,10 +44,23 @@ public async Task DownloadFileAsync(DownloadFileOptions options, Cancellat throw new IOException($"{filePathToSave} already exists and overwrite is false"); } - var newFile = _fileSystemProvider.Create(filePathToSave); - await using (newFile.ConfigureAwait(false)) + var temporaryPath = filePathToSave + "." + _fileSystemProvider.GetRandomFileName() + ".download"; + try { - await stream.CopyToAsync(newFile, cancellationToken).ConfigureAwait(false); + var newFile = _fileSystemProvider.Create(temporaryPath); + await using (newFile.ConfigureAwait(false)) + { + await stream.CopyToAsync(newFile, cancellationToken).ConfigureAwait(false); + } + + _fileSystemProvider.MoveFile(temporaryPath, filePathToSave, options.Overwrite); + } + finally + { + if (_fileSystemProvider.FileExists(temporaryPath)) + { + _fileSystemProvider.DeleteFile(temporaryPath); + } } _moduleLoggerProvider.GetLogger().LogInformation("File {Uri} downloaded to {SaveLocation}", options.DownloadUri, filePathToSave); diff --git a/src/ModularPipelines/FileSystem/IFileSystemProvider.cs b/src/ModularPipelines/FileSystem/IFileSystemProvider.cs index 9c544b37df..fa97c279c0 100644 --- a/src/ModularPipelines/FileSystem/IFileSystemProvider.cs +++ b/src/ModularPipelines/FileSystem/IFileSystemProvider.cs @@ -27,6 +27,16 @@ public interface IFileSystemProvider void DeleteFile(string path); void CopyFile(string sourcePath, string destinationPath, bool overwrite); void MoveFile(string sourcePath, string destinationPath); + void MoveFile(string sourcePath, string destinationPath, bool overwrite) + { + if (overwrite && FileExists(destinationPath)) + { + DeleteFile(destinationPath); + } + + MoveFile(sourcePath, destinationPath); + } + bool FileExists(string path); // Directory operations diff --git a/src/ModularPipelines/FileSystem/SystemFileSystemProvider.cs b/src/ModularPipelines/FileSystem/SystemFileSystemProvider.cs index a245065efe..8c86c74ceb 100644 --- a/src/ModularPipelines/FileSystem/SystemFileSystemProvider.cs +++ b/src/ModularPipelines/FileSystem/SystemFileSystemProvider.cs @@ -61,6 +61,9 @@ public void CopyFile(string sourcePath, string destinationPath, bool overwrite) public void MoveFile(string sourcePath, string destinationPath) => System.IO.File.Move(sourcePath, destinationPath); + public void MoveFile(string sourcePath, string destinationPath, bool overwrite) + => System.IO.File.Move(sourcePath, destinationPath, overwrite); + public bool FileExists(string path) => System.IO.File.Exists(path); diff --git a/test/ModularPipelines.UnitTests/Helpers/DownloaderTests.cs b/test/ModularPipelines.UnitTests/Helpers/DownloaderTests.cs index 05e8edf1ba..75fef13a6c 100644 --- a/test/ModularPipelines.UnitTests/Helpers/DownloaderTests.cs +++ b/test/ModularPipelines.UnitTests/Helpers/DownloaderTests.cs @@ -68,6 +68,67 @@ await downloader.DownloadFileAsync(new DownloadFileOptions( await Assert.That(content.IsDisposed).IsTrue(); } + [Test] + public async Task DownloadFileAsync_PreservesExistingFileWhenStreamingFails() + { + var directory = Path.Combine(Path.GetTempPath(), $"modular-pipelines-download-{Guid.NewGuid():N}"); + var destination = Path.Combine(directory, "download.bin"); + Directory.CreateDirectory(directory); + await System.IO.File.WriteAllTextAsync(destination, "original"); + + try + { + var content = new StreamContent(new FailingReadStream("partial download"u8.ToArray())); + var downloader = CreateDownloader(content); + + await Assert.ThrowsAsync(() => downloader.DownloadFileAsync(new DownloadFileOptions( + new Uri("https://example.test/download")) + { + SavePath = destination, + })); + + using (Assert.Multiple()) + { + await Assert.That(await System.IO.File.ReadAllTextAsync(destination)).IsEqualTo("original"); + await Assert.That(Directory.GetFiles(directory).Length).IsEqualTo(1); + } + } + finally + { + Directory.Delete(directory, recursive: true); + } + } + + [Test] + public async Task DownloadFileAsync_ReplacesExistingFileAfterStreamingSucceeds() + { + var directory = Path.Combine(Path.GetTempPath(), $"modular-pipelines-download-{Guid.NewGuid():N}"); + var destination = Path.Combine(directory, "download.bin"); + Directory.CreateDirectory(directory); + await System.IO.File.WriteAllTextAsync(destination, "original"); + + try + { + var downloader = CreateDownloader(new StringContent("replacement")); + + await downloader.DownloadFileAsync(new DownloadFileOptions( + new Uri("https://example.test/download")) + { + SavePath = destination, + }); + + using (Assert.Multiple()) + { + await Assert.That(await System.IO.File.ReadAllTextAsync(destination)).IsEqualTo("replacement"); + await Assert.That(Directory.GetFiles(directory).Length).IsEqualTo(1); + } + } + finally + { + Directory.Delete(directory, recursive: true); + } + } + [Test] public async Task DownloadResponseAsync_DisposesResponseWhenStatusValidationFails() { @@ -115,4 +176,20 @@ protected override void Dispose(bool disposing) base.Dispose(disposing); } } + + private sealed class FailingReadStream(byte[] content) : MemoryStream(content) + { + private bool _hasRead; + + public override ValueTask ReadAsync(Memory buffer, CancellationToken cancellationToken = default) + { + if (_hasRead) + { + throw new IOException("Simulated streaming failure"); + } + + _hasRead = true; + return base.ReadAsync(buffer, cancellationToken); + } + } } From 466ea2d0ccda4146f4e08b2b8fb68fcee412c2ab Mon Sep 17 00:00:00 2001 From: Tom Longhurst <30480171+thomhurst@users.noreply.github.com> Date: Thu, 30 Jul 2026 23:20:30 +0100 Subject: [PATCH 13/17] fix(files): preserve legacy destinations --- .../FileSystem/IFileSystemProvider.cs | 3 ++- .../Helpers/DownloaderTests.cs | 22 +++++++++++++++++++ 2 files changed, 24 insertions(+), 1 deletion(-) diff --git a/src/ModularPipelines/FileSystem/IFileSystemProvider.cs b/src/ModularPipelines/FileSystem/IFileSystemProvider.cs index fa97c279c0..ffecca4770 100644 --- a/src/ModularPipelines/FileSystem/IFileSystemProvider.cs +++ b/src/ModularPipelines/FileSystem/IFileSystemProvider.cs @@ -31,7 +31,8 @@ void MoveFile(string sourcePath, string destinationPath, bool overwrite) { if (overwrite && FileExists(destinationPath)) { - DeleteFile(destinationPath); + throw new NotSupportedException( + "This file system provider does not support atomic overwrite moves."); } MoveFile(sourcePath, destinationPath); diff --git a/test/ModularPipelines.UnitTests/Helpers/DownloaderTests.cs b/test/ModularPipelines.UnitTests/Helpers/DownloaderTests.cs index 75fef13a6c..bb67c86496 100644 --- a/test/ModularPipelines.UnitTests/Helpers/DownloaderTests.cs +++ b/test/ModularPipelines.UnitTests/Helpers/DownloaderTests.cs @@ -68,6 +68,28 @@ await downloader.DownloadFileAsync(new DownloadFileOptions( await Assert.That(content.IsDisposed).IsTrue(); } + [Test] + public async Task LegacyProviderFallbackDoesNotDeleteExistingDestination() + { + const string destination = "download.bin"; + var fileSystemProvider = new Mock + { + CallBase = true, + }; + fileSystemProvider + .Setup(x => x.FileExists(destination)) + .Returns(true); + + var exception = Assert.Throws(() => + fileSystemProvider.Object.MoveFile("temporary.download", destination, overwrite: true)); + + await Assert.That(exception.Message).Contains("does not support atomic overwrite moves"); + fileSystemProvider.Verify(x => x.DeleteFile(destination), Times.Never()); + fileSystemProvider.Verify( + x => x.MoveFile(It.IsAny(), It.IsAny()), + Times.Never()); + } + [Test] public async Task DownloadFileAsync_PreservesExistingFileWhenStreamingFails() { From 56b58f99e652666723fe485699da03354ad80b44 Mon Sep 17 00:00:00 2001 From: Tom Longhurst <30480171+thomhurst@users.noreply.github.com> Date: Thu, 30 Jul 2026 23:53:03 +0100 Subject: [PATCH 14/17] style: satisfy analyzer checks --- .../FileSystem/IFileSystemProvider.cs | 19 +++++++++++++++++++ .../FileSystem/SystemFileSystemProvider.cs | 6 ++++-- 2 files changed, 23 insertions(+), 2 deletions(-) diff --git a/src/ModularPipelines/FileSystem/IFileSystemProvider.cs b/src/ModularPipelines/FileSystem/IFileSystemProvider.cs index ffecca4770..7313398da8 100644 --- a/src/ModularPipelines/FileSystem/IFileSystemProvider.cs +++ b/src/ModularPipelines/FileSystem/IFileSystemProvider.cs @@ -8,25 +8,36 @@ public interface IFileSystemProvider { // File read operations Task ReadAllTextAsync(string path, CancellationToken cancellationToken = default); + IAsyncEnumerable ReadLinesAsync(string path, CancellationToken cancellationToken = default); + Task ReadAllBytesAsync(string path, CancellationToken cancellationToken = default); // File write operations Task WriteAllTextAsync(string path, string contents, CancellationToken cancellationToken = default); + Task WriteAllBytesAsync(string path, byte[] contents, CancellationToken cancellationToken = default); + Task WriteAllLinesAsync(string path, IEnumerable contents, CancellationToken cancellationToken = default); + Task AppendAllTextAsync(string path, string contents, CancellationToken cancellationToken = default); + Task AppendAllLinesAsync(string path, IEnumerable contents, CancellationToken cancellationToken = default); // File stream operations Stream OpenRead(string path); + Stream Create(string path); + Stream Open(string path, FileMode mode, FileAccess access); // File management operations void DeleteFile(string path); + void CopyFile(string sourcePath, string destinationPath, bool overwrite); + void MoveFile(string sourcePath, string destinationPath); + void MoveFile(string sourcePath, string destinationPath, bool overwrite) { if (overwrite && FileExists(destinationPath)) @@ -42,15 +53,23 @@ void MoveFile(string sourcePath, string destinationPath, bool overwrite) // Directory operations void CreateDirectory(string path); + void DeleteDirectory(string path, bool recursive); + void MoveDirectory(string sourcePath, string destinationPath); + bool DirectoryExists(string path); + IEnumerable EnumerateFiles(string path, string searchPattern, SearchOption searchOption); + IEnumerable EnumerateDirectories(string path, string searchPattern, SearchOption searchOption); // Path operations string GetTempPath(); + string GetRandomFileName(); + string Combine(params string[] paths); + string GetRelativePath(string relativeTo, string path); } diff --git a/src/ModularPipelines/FileSystem/SystemFileSystemProvider.cs b/src/ModularPipelines/FileSystem/SystemFileSystemProvider.cs index 8c86c74ceb..c7bd3a9ec7 100644 --- a/src/ModularPipelines/FileSystem/SystemFileSystemProvider.cs +++ b/src/ModularPipelines/FileSystem/SystemFileSystemProvider.cs @@ -9,11 +9,13 @@ namespace ModularPipelines.FileSystem; public sealed class SystemFileSystemProvider : IFileSystemProvider { /// - /// Singleton instance for use when no DI is available. + /// Gets singleton instance for use when no DI is available. /// public static SystemFileSystemProvider Instance { get; } = new(); - private SystemFileSystemProvider() { } + private SystemFileSystemProvider() + { + } // File read operations public Task ReadAllTextAsync(string path, CancellationToken cancellationToken = default) From e45a24a45f9834320f252050f3a38444b7d91512 Mon Sep 17 00:00:00 2001 From: Tom Longhurst <30480171+thomhurst@users.noreply.github.com> Date: Fri, 31 Jul 2026 02:12:56 +0100 Subject: [PATCH 15/17] fix(http): dispose request content once --- .../Http/HttpContentPreviewReader.cs | 1 + .../Http/HttpRequestFormatter.cs | 1 - .../Context/HttpFormatterTests.cs | 19 ++++++++++++++++--- 3 files changed, 17 insertions(+), 4 deletions(-) diff --git a/src/ModularPipelines/Http/HttpContentPreviewReader.cs b/src/ModularPipelines/Http/HttpContentPreviewReader.cs index 0bef864af4..b8abb7c0e1 100644 --- a/src/ModularPipelines/Http/HttpContentPreviewReader.cs +++ b/src/ModularPipelines/Http/HttpContentPreviewReader.cs @@ -82,6 +82,7 @@ internal static class HttpContentPreviewReader /// The maximum number of bytes to include in the preview. /// The cancellation token. /// The preview, truncation state, replayable content, and total length. + /// This method takes ownership of and disposes . public static async Task<(string Preview, bool IsTruncated, HttpContent ReplayContent, long? TotalLength)> ReadReplayableAsync( HttpContent content, diff --git a/src/ModularPipelines/Http/HttpRequestFormatter.cs b/src/ModularPipelines/Http/HttpRequestFormatter.cs index 289c296d4f..7512223da8 100644 --- a/src/ModularPipelines/Http/HttpRequestFormatter.cs +++ b/src/ModularPipelines/Http/HttpRequestFormatter.cs @@ -163,7 +163,6 @@ private async Task AppendBodyAsync( .ReadReplayableAsync(content, maxBodySize, cancellationToken) .ConfigureAwait(false); request.Content = replayContent; - content.Dispose(); if (string.IsNullOrWhiteSpace(body) && !isTruncated) { diff --git a/test/ModularPipelines.UnitTests/Context/HttpFormatterTests.cs b/test/ModularPipelines.UnitTests/Context/HttpFormatterTests.cs index e033003c0a..a14c30b6a6 100644 --- a/test/ModularPipelines.UnitTests/Context/HttpFormatterTests.cs +++ b/test/ModularPipelines.UnitTests/Context/HttpFormatterTests.cs @@ -42,9 +42,10 @@ public async Task RequestFormatter_PreservesBodyForRepeatedSends() { const string body = "abcdefghijklmnopqrstuvwxyz"; var stream = new CountingReadStream(Encoding.UTF8.GetBytes(body)); + var content = CreateTextContent(stream); using var request = new HttpRequestMessage(HttpMethod.Post, "https://example.test") { - Content = CreateTextContent(stream), + Content = content, }; var formatter = CreateRequestFormatter(); @@ -64,6 +65,7 @@ public async Task RequestFormatter_PreservesBodyForRepeatedSends() await Assert.That(Encoding.UTF8.GetString(firstSend.ToArray())).IsEqualTo(body); await Assert.That(Encoding.UTF8.GetString(secondSend.ToArray())).IsEqualTo(body); await Assert.That(stream.IsDisposed).IsTrue(); + await Assert.That(content.DisposeCount).IsEqualTo(1); } } @@ -179,9 +181,9 @@ public async Task ResponseFormatter_RedactsSecretWhenPreviewSplitsMultibyteChara } } - private static StreamContent CreateTextContent(Stream stream) + private static CountingStreamContent CreateTextContent(Stream stream) { - var content = new StreamContent(stream); + var content = new CountingStreamContent(stream); content.Headers.ContentType = new MediaTypeHeaderValue("text/plain") { CharSet = "utf-8", @@ -189,6 +191,17 @@ private static StreamContent CreateTextContent(Stream stream) return content; } + private sealed class CountingStreamContent(Stream stream) : StreamContent(stream) + { + public int DisposeCount { get; private set; } + + protected override void Dispose(bool disposing) + { + DisposeCount++; + base.Dispose(disposing); + } + } + private static ISecretObfuscator CreateObfuscator(params string[] secrets) { var obfuscator = new Mock(); From 2bab96cccbad3f5ff1ce776f0a91fcb8d15e96e6 Mon Sep 17 00:00:00 2001 From: Tom Longhurst <30480171+thomhurst@users.noreply.github.com> Date: Fri, 31 Jul 2026 04:04:57 +0100 Subject: [PATCH 16/17] fix(http): harden streamed body handling --- src/ModularPipelines/Context/Downloader.cs | 5 +- .../Http/HttpContentPreviewReader.cs | 37 ++++-- src/ModularPipelines/Http/IHttpLogger.cs | 4 +- .../Http/TimeoutHttpContent.cs | 5 +- .../Context/HttpFormatterTests.cs | 21 ++++ .../Context/HttpTests.cs | 109 +++++++++++++++++- .../Helpers/DownloaderTests.cs | 33 ++++++ 7 files changed, 201 insertions(+), 13 deletions(-) diff --git a/src/ModularPipelines/Context/Downloader.cs b/src/ModularPipelines/Context/Downloader.cs index af4ba89500..8195178bcb 100644 --- a/src/ModularPipelines/Context/Downloader.cs +++ b/src/ModularPipelines/Context/Downloader.cs @@ -44,7 +44,10 @@ public async Task DownloadFileAsync(DownloadFileOptions options, Cancellat throw new IOException($"{filePathToSave} already exists and overwrite is false"); } - var temporaryPath = filePathToSave + "." + _fileSystemProvider.GetRandomFileName() + ".download"; + var destinationDirectory = Path.GetDirectoryName(filePathToSave); + var temporaryPath = Path.Combine( + destinationDirectory ?? string.Empty, + _fileSystemProvider.GetRandomFileName()); try { var newFile = _fileSystemProvider.Create(temporaryPath); diff --git a/src/ModularPipelines/Http/HttpContentPreviewReader.cs b/src/ModularPipelines/Http/HttpContentPreviewReader.cs index b8abb7c0e1..433dde99bb 100644 --- a/src/ModularPipelines/Http/HttpContentPreviewReader.cs +++ b/src/ModularPipelines/Http/HttpContentPreviewReader.cs @@ -1,3 +1,4 @@ +using System.Buffers; using System.Text; namespace ModularPipelines.Http; @@ -7,6 +8,8 @@ namespace ModularPipelines.Http; /// internal static class HttpContentPreviewReader { + private const int MaximumPreviewBufferSize = 81920; + /// /// Reads at most one byte beyond the configured preview limit. /// @@ -37,23 +40,29 @@ internal static class HttpContentPreviewReader using var cancellationRegistration = cancellationToken.Register( static state => ((Stream) state!).Dispose(), stream); - var buffer = new byte[maxBytes + 1]; - var bytesRead = 0; + var probeLength = maxBytes + 1; + var readBuffer = ArrayPool.Shared.Rent( + Math.Min(probeLength, MaximumPreviewBufferSize)); + using var buffer = new MemoryStream(GetInitialCapacity(content, probeLength)); try { - while (bytesRead < buffer.Length) + while (buffer.Length < probeLength) { + var bytesRemaining = probeLength - (int) buffer.Length; var read = await stream.ReadAsync( - buffer.AsMemory(bytesRead, buffer.Length - bytesRead), + readBuffer.AsMemory( + 0, + Math.Min(MaximumPreviewBufferSize, bytesRemaining)), cancellationToken) .ConfigureAwait(false); + cancellationToken.ThrowIfCancellationRequested(); if (read == 0) { break; } - bytesRead += read; + buffer.Write(readBuffer, 0, read); } } catch (Exception exception) @@ -65,12 +74,18 @@ internal static class HttpContentPreviewReader exception, cancellationToken); } + finally + { + ArrayPool.Shared.Return(readBuffer); + } + var bufferedBytes = buffer.ToArray(); + var bytesRead = bufferedBytes.Length; var isTruncated = bytesRead > maxBytes; var previewLength = Math.Min(bytesRead, maxBytes); var encoding = GetEncoding(content); - var preview = DecodeCompletePrefix(encoding, buffer, previewLength); - var replayContent = CreateReplayContent(content, stream, buffer.AsSpan(0, bytesRead).ToArray()); + var preview = DecodeCompletePrefix(encoding, bufferedBytes, previewLength); + var replayContent = CreateReplayContent(content, stream, bufferedBytes); return (preview, isTruncated, replayContent, content.Headers.ContentLength); } @@ -133,6 +148,14 @@ private static async Task ReadAllBytesAsync( } } + private static int GetInitialCapacity(HttpContent content, int probeLength) + { + var declaredLength = content.Headers.ContentLength; + return declaredLength is >= 0 + ? (int) Math.Min(Math.Min(declaredLength.Value, probeLength), MaximumPreviewBufferSize) + : Math.Min(probeLength, MaximumPreviewBufferSize); + } + private static Encoding GetEncoding(HttpContent content) { var charset = content.Headers.ContentType?.CharSet?.Trim('"'); diff --git a/src/ModularPipelines/Http/IHttpLogger.cs b/src/ModularPipelines/Http/IHttpLogger.cs index 20e3f530a0..295ecd6b15 100644 --- a/src/ModularPipelines/Http/IHttpLogger.cs +++ b/src/ModularPipelines/Http/IHttpLogger.cs @@ -78,7 +78,7 @@ Task PrintResponse( IModuleLogger logger, CancellationToken cancellationToken) { - return PrintResponse(response, logger); + return PrintResponse(response, logger).WaitAsync(cancellationToken); } /// @@ -104,7 +104,7 @@ Task PrintResponse( HttpLoggingOptions options, CancellationToken cancellationToken) { - return PrintResponse(response, logger, options); + return PrintResponse(response, logger, options).WaitAsync(cancellationToken); } /// diff --git a/src/ModularPipelines/Http/TimeoutHttpContent.cs b/src/ModularPipelines/Http/TimeoutHttpContent.cs index 4eb7403e26..600ed13606 100644 --- a/src/ModularPipelines/Http/TimeoutHttpContent.cs +++ b/src/ModularPipelines/Http/TimeoutHttpContent.cs @@ -110,6 +110,7 @@ private async Task CopyInnerContentToAsync( try { await source.CopyToAsync(destination, effectiveCancellationToken).ConfigureAwait(false); + effectiveCancellationToken.ThrowIfCancellationRequested(); } catch (Exception exception) when (effectiveCancellationToken.IsCancellationRequested @@ -216,10 +217,12 @@ public override async ValueTask ReadAsync( innerStream); try { - return await innerStream.ReadAsync( + var bytesRead = await innerStream.ReadAsync( buffer, effectiveCancellationToken) .ConfigureAwait(false); + effectiveCancellationToken.ThrowIfCancellationRequested(); + return bytesRead; } catch (Exception exception) when (effectiveCancellationToken.IsCancellationRequested && diff --git a/test/ModularPipelines.UnitTests/Context/HttpFormatterTests.cs b/test/ModularPipelines.UnitTests/Context/HttpFormatterTests.cs index a14c30b6a6..3e0aed2146 100644 --- a/test/ModularPipelines.UnitTests/Context/HttpFormatterTests.cs +++ b/test/ModularPipelines.UnitTests/Context/HttpFormatterTests.cs @@ -37,6 +37,24 @@ public async Task ResponseFormatter_BoundsPreviewReadAndPreservesBody() } } + [Test] + public async Task ResponseFormatter_DoesNotAllocateEntireLargePreviewLimit() + { + var stream = new CountingReadStream([]); + using var response = new HttpResponseMessage + { + Content = CreateTextContent(stream), + }; + var formatter = CreateResponseFormatter(); + + await formatter.FormatAsync(response, new HttpLoggingOptions + { + MaxBodySizeToLog = 64 * 1024 * 1024, + }); + + await Assert.That(stream.MaximumRequestedReadSize).IsLessThanOrEqualTo(81920); + } + [Test] public async Task RequestFormatter_PreservesBodyForRepeatedSends() { @@ -245,12 +263,15 @@ private sealed class CountingReadStream(byte[] contents) : MemoryStream(contents { public int BytesRead { get; private set; } + public int MaximumRequestedReadSize { get; private set; } + public bool IsDisposed { get; private set; } public override async ValueTask ReadAsync( Memory buffer, CancellationToken cancellationToken = default) { + MaximumRequestedReadSize = Math.Max(MaximumRequestedReadSize, buffer.Length); var read = await base.ReadAsync(buffer, cancellationToken); BytesRead += read; return read; diff --git a/test/ModularPipelines.UnitTests/Context/HttpTests.cs b/test/ModularPipelines.UnitTests/Context/HttpTests.cs index 05f30c294e..664697ad71 100644 --- a/test/ModularPipelines.UnitTests/Context/HttpTests.cs +++ b/test/ModularPipelines.UnitTests/Context/HttpTests.cs @@ -246,6 +246,70 @@ public async Task SendAsync_NormalizesAsyncStreamFailureAfterPerReadCancellation } } + [Test] + [Arguments(false)] + [Arguments(true)] + public async Task SendAsync_DoesNotTreatCancellationAsSuccessfulEndOfStream( + bool useBufferedCopy) + { + var timeout = TimeSpan.FromMilliseconds(100); + var contentStream = new BlockingReadStream( + ignoreAsyncCancellation: true, + returnEofWhenDisposed: true); + using var httpClient = new HttpClient( + new ImmediateResponseHandler(new StreamContent(contentStream))); + var http = new ModularPipelines.Http.Http( + Mock.Of(), + Mock.Of(), + Mock.Of(), + Microsoft.Extensions.Options.Options.Create(new PipelineOptions())); + using var response = await http.SendAsync(new HttpOptions( + new HttpRequestMessage(HttpMethod.Get, "https://example.test/cancelled-eof")) + { + HttpClient = httpClient, + LoggingType = HttpLoggingType.None, + Timeout = timeout, + }); + + Task readTask; + if (useBufferedCopy) + { + readTask = response.Content.ReadAsStringAsync(); + } + else + { + var stream = await response.Content.ReadAsStreamAsync(); + readTask = stream.ReadAsync(new byte[1]).AsTask(); + } + + await Assert.ThrowsAsync( + async () => await readTask.WaitAsync(TimeSpan.FromSeconds(1))); + } + + [Test] + public async Task SendAsync_CancelsLegacyResponseLogger() + { + var timeout = TimeSpan.FromMilliseconds(100); + var contentStream = new BlockingReadStream(ignoreAsyncCancellation: true); + using var httpClient = new HttpClient( + new ImmediateResponseHandler(new StreamContent(contentStream))); + var http = new ModularPipelines.Http.Http( + Mock.Of(), + Mock.Of(), + new LegacyResponseBodyLogger(), + Microsoft.Extensions.Options.Options.Create(new PipelineOptions())); + + await Assert.ThrowsAsync(async () => + await http.SendAsync(new HttpOptions( + new HttpRequestMessage(HttpMethod.Get, "https://example.test/legacy-logger")) + { + HttpClient = httpClient, + LoggingType = HttpLoggingType.Response, + Timeout = timeout, + }) + .WaitAsync(TimeSpan.FromSeconds(1))); + } + [Test] [Arguments(true, true)] [Arguments(true, false)] @@ -763,7 +827,8 @@ protected override bool TryComputeLength(out long length) private sealed class BlockingReadStream( bool throwIOExceptionWhenDisposed = false, - bool ignoreAsyncCancellation = false) : Stream + bool ignoreAsyncCancellation = false, + bool returnEofWhenDisposed = false) : Stream { private readonly TaskCompletionSource _release = new(TaskCreationOptions.RunContinuationsAsynchronously); @@ -816,7 +881,10 @@ public override async ValueTask ReadAsync( if (ignoreAsyncCancellation) { await _release.Task; - ThrowIfDisposed(); + if (!returnEofWhenDisposed) + { + ThrowIfDisposed(); + } } else { @@ -870,6 +938,43 @@ private void ThrowIfDisposed() } } + private sealed class LegacyResponseBodyLogger : IHttpLogger + { + public Task PrintRequest(HttpRequestMessage request, IModuleLogger logger) + { + return Task.CompletedTask; + } + + public Task PrintRequest( + HttpRequestMessage request, + IModuleLogger logger, + HttpLoggingOptions options) + { + return Task.CompletedTask; + } + + public Task PrintResponse(HttpResponseMessage response, IModuleLogger logger) + { + return response.Content.ReadAsStringAsync(); + } + + public Task PrintResponse( + HttpResponseMessage response, + IModuleLogger logger, + HttpLoggingOptions options) + { + return response.Content.ReadAsStringAsync(); + } + + public void PrintStatusCode(HttpStatusCode? httpStatusCode, IModuleLogger logger) + { + } + + public void PrintDuration(TimeSpan duration, IModuleLogger logger) + { + } + } + private sealed class SequenceResponseHandler(params HttpResponseMessage[] responses) : HttpMessageHandler { public int CallCount { get; private set; } diff --git a/test/ModularPipelines.UnitTests/Helpers/DownloaderTests.cs b/test/ModularPipelines.UnitTests/Helpers/DownloaderTests.cs index bb67c86496..a2ef3832e4 100644 --- a/test/ModularPipelines.UnitTests/Helpers/DownloaderTests.cs +++ b/test/ModularPipelines.UnitTests/Helpers/DownloaderTests.cs @@ -57,6 +57,9 @@ public async Task DownloadFileAsync_DisposesResponse() fileSystemProvider .Setup(x => x.Create(It.IsAny())) .Returns(() => new MemoryStream()); + fileSystemProvider + .Setup(x => x.GetRandomFileName()) + .Returns("random.tmp"); var downloader = CreateDownloader(content, fileSystemProvider.Object); await downloader.DownloadFileAsync(new DownloadFileOptions( @@ -68,6 +71,36 @@ await downloader.DownloadFileAsync(new DownloadFileOptions( await Assert.That(content.IsDisposed).IsTrue(); } + [Test] + public async Task DownloadFileAsync_UsesBoundedSiblingTemporaryName() + { + var destination = Path.Combine( + "downloads", + new string('x', 240) + ".bin"); + var expectedTemporaryPath = Path.Combine("downloads", "random.tmp"); + var fileSystemProvider = new Mock(); + fileSystemProvider + .Setup(x => x.GetRandomFileName()) + .Returns("random.tmp"); + fileSystemProvider + .Setup(x => x.Create(expectedTemporaryPath)) + .Returns(() => new MemoryStream()); + var downloader = CreateDownloader( + new StringContent("download"), + fileSystemProvider.Object); + + await downloader.DownloadFileAsync(new DownloadFileOptions( + new Uri("https://example.test/download")) + { + SavePath = destination, + }); + + fileSystemProvider.Verify(x => x.Create(expectedTemporaryPath), Times.Once()); + fileSystemProvider.Verify( + x => x.MoveFile(expectedTemporaryPath, destination, true), + Times.Once()); + } + [Test] public async Task LegacyProviderFallbackDoesNotDeleteExistingDestination() { From 7cf8f8b7d1edb2bd220044f6a6b122f3c3e9a3ff Mon Sep 17 00:00:00 2001 From: Tom Longhurst <30480171+thomhurst@users.noreply.github.com> Date: Fri, 31 Jul 2026 05:24:20 +0100 Subject: [PATCH 17/17] fix(http): close cancellation gaps --- .../Http/HttpContentPreviewReader.cs | 1 + src/ModularPipelines/Http/IHttpLogger.cs | 4 +- .../Http/TimeoutHttpContent.cs | 8 +- .../Context/HttpTests.cs | 78 +++++++++++++++---- 4 files changed, 71 insertions(+), 20 deletions(-) diff --git a/src/ModularPipelines/Http/HttpContentPreviewReader.cs b/src/ModularPipelines/Http/HttpContentPreviewReader.cs index 433dde99bb..2ab2ef7a1c 100644 --- a/src/ModularPipelines/Http/HttpContentPreviewReader.cs +++ b/src/ModularPipelines/Http/HttpContentPreviewReader.cs @@ -135,6 +135,7 @@ private static async Task ReadAllBytesAsync( try { await stream.CopyToAsync(buffer, cancellationToken).ConfigureAwait(false); + cancellationToken.ThrowIfCancellationRequested(); return buffer.ToArray(); } catch (Exception exception) diff --git a/src/ModularPipelines/Http/IHttpLogger.cs b/src/ModularPipelines/Http/IHttpLogger.cs index 295ecd6b15..bd075570f9 100644 --- a/src/ModularPipelines/Http/IHttpLogger.cs +++ b/src/ModularPipelines/Http/IHttpLogger.cs @@ -29,7 +29,7 @@ Task PrintRequest( IModuleLogger logger, CancellationToken cancellationToken) { - return PrintRequest(request, logger); + return PrintRequest(request, logger).WaitAsync(cancellationToken); } /// @@ -55,7 +55,7 @@ Task PrintRequest( HttpLoggingOptions options, CancellationToken cancellationToken) { - return PrintRequest(request, logger, options); + return PrintRequest(request, logger, options).WaitAsync(cancellationToken); } /// diff --git a/src/ModularPipelines/Http/TimeoutHttpContent.cs b/src/ModularPipelines/Http/TimeoutHttpContent.cs index 600ed13606..1c7dbcf5ed 100644 --- a/src/ModularPipelines/Http/TimeoutHttpContent.cs +++ b/src/ModularPipelines/Http/TimeoutHttpContent.cs @@ -169,7 +169,9 @@ public override int Read(byte[] buffer, int offset, int count) timeoutCancellationToken.ThrowIfCancellationRequested(); try { - return innerStream.Read(buffer, offset, count); + var bytesRead = innerStream.Read(buffer, offset, count); + timeoutCancellationToken.ThrowIfCancellationRequested(); + return bytesRead; } catch (Exception exception) when (timeoutCancellationToken.IsCancellationRequested && @@ -184,7 +186,9 @@ public override int Read(Span buffer) timeoutCancellationToken.ThrowIfCancellationRequested(); try { - return innerStream.Read(buffer); + var bytesRead = innerStream.Read(buffer); + timeoutCancellationToken.ThrowIfCancellationRequested(); + return bytesRead; } catch (Exception exception) when (timeoutCancellationToken.IsCancellationRequested && diff --git a/test/ModularPipelines.UnitTests/Context/HttpTests.cs b/test/ModularPipelines.UnitTests/Context/HttpTests.cs index 664697ad71..8bbbfeafb6 100644 --- a/test/ModularPipelines.UnitTests/Context/HttpTests.cs +++ b/test/ModularPipelines.UnitTests/Context/HttpTests.cs @@ -296,7 +296,7 @@ public async Task SendAsync_CancelsLegacyResponseLogger() var http = new ModularPipelines.Http.Http( Mock.Of(), Mock.Of(), - new LegacyResponseBodyLogger(), + new LegacyBodyLogger(logRequest: false), Microsoft.Extensions.Options.Options.Create(new PipelineOptions())); await Assert.ThrowsAsync(async () => @@ -311,16 +311,50 @@ await http.SendAsync(new HttpOptions( } [Test] - [Arguments(true, true)] - [Arguments(true, false)] - [Arguments(false, true)] - [Arguments(false, false)] + public async Task SendAsync_CancelsLegacyRequestLogger() + { + var timeout = TimeSpan.FromMilliseconds(100); + var contentStream = new BlockingReadStream(ignoreAsyncCancellation: true); + using var httpClient = new HttpClient( + new ImmediateResponseHandler(new StringContent("response"))); + var http = new ModularPipelines.Http.Http( + Mock.Of(), + Mock.Of(), + new LegacyBodyLogger(logRequest: true), + Microsoft.Extensions.Options.Options.Create(new PipelineOptions())); + using var request = new HttpRequestMessage( + HttpMethod.Post, + "https://example.test/legacy-request-logger") + { + Content = new StreamContent(contentStream), + }; + + await Assert.ThrowsAsync(async () => + await http.SendAsync(new HttpOptions(request) + { + HttpClient = httpClient, + LoggingType = HttpLoggingType.Request, + Timeout = timeout, + }) + .WaitAsync(TimeSpan.FromSeconds(1))); + } + + [Test] + [Arguments(true, true, false)] + [Arguments(true, false, false)] + [Arguments(false, true, false)] + [Arguments(false, false, false)] + [Arguments(true, false, true)] + [Arguments(false, false, true)] public async Task SendAsync_ConfiguredTimeoutInterruptsSynchronousBodyRead( bool useCopyTo, - bool throwIOExceptionWhenDisposed) + bool throwIOExceptionWhenDisposed, + bool returnEofWhenDisposed) { var timeout = TimeSpan.FromMilliseconds(100); - var contentStream = new BlockingReadStream(throwIOExceptionWhenDisposed); + var contentStream = new BlockingReadStream( + throwIOExceptionWhenDisposed, + returnEofWhenDisposed: returnEofWhenDisposed); using var httpClient = new HttpClient( new ImmediateResponseHandler(new StreamContent(contentStream))); var result = await GetService((_, _) => { }); @@ -395,13 +429,17 @@ public async Task SendAsync_CustomClientLogsRawContentBeforeApplyingBodyTimeout( } [Test] - [Arguments(4096)] - [Arguments(0)] + [Arguments(4096, false)] + [Arguments(0, false)] + [Arguments(0, true)] public async Task SendAsync_CustomClientTimeoutInterruptsNonCooperativeResponseLogging( - int maxBodySizeToLog) + int maxBodySizeToLog, + bool returnEofWhenDisposed) { var timeout = TimeSpan.FromMilliseconds(100); - var contentStream = new BlockingReadStream(ignoreAsyncCancellation: true); + var contentStream = new BlockingReadStream( + ignoreAsyncCancellation: true, + returnEofWhenDisposed: returnEofWhenDisposed); var content = new StreamContent(contentStream); content.Headers.ContentType = new MediaTypeHeaderValue("text/plain"); using var httpClient = new HttpClient(new ImmediateResponseHandler(content)); @@ -938,11 +976,13 @@ private void ThrowIfDisposed() } } - private sealed class LegacyResponseBodyLogger : IHttpLogger + private sealed class LegacyBodyLogger(bool logRequest) : IHttpLogger { public Task PrintRequest(HttpRequestMessage request, IModuleLogger logger) { - return Task.CompletedTask; + return logRequest + ? request.Content!.ReadAsStringAsync() + : Task.CompletedTask; } public Task PrintRequest( @@ -950,12 +990,16 @@ public Task PrintRequest( IModuleLogger logger, HttpLoggingOptions options) { - return Task.CompletedTask; + return logRequest + ? request.Content!.ReadAsStringAsync() + : Task.CompletedTask; } public Task PrintResponse(HttpResponseMessage response, IModuleLogger logger) { - return response.Content.ReadAsStringAsync(); + return logRequest + ? Task.CompletedTask + : response.Content.ReadAsStringAsync(); } public Task PrintResponse( @@ -963,7 +1007,9 @@ public Task PrintResponse( IModuleLogger logger, HttpLoggingOptions options) { - return response.Content.ReadAsStringAsync(); + return logRequest + ? Task.CompletedTask + : response.Content.ReadAsStringAsync(); } public void PrintStatusCode(HttpStatusCode? httpStatusCode, IModuleLogger logger)