fix(http): stream response bodies

src/ModularPipelines/Context/Downloader.cs

@@ -25,14 +25,14 @@ public Downloader(IModuleLoggerProvider moduleLoggerProvider, IHttpContext http,
     public async Task<string?> 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<File> 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))
@@ -44,10 +44,26 @@ public async Task<File> 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 destinationDirectory = Path.GetDirectoryName(filePathToSave);
+            var temporaryPath = Path.Combine(
+                destinationDirectory ?? string.Empty,
+                _fileSystemProvider.GetRandomFileName());
+            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);
@@ -68,7 +84,15 @@ public async Task<HttpResponseMessage> 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;
+        }
     }
 
     /// <summary>

src/ModularPipelines/FileSystem/IFileSystemProvider.cs

@@ -8,38 +8,68 @@ public interface IFileSystemProvider
 {
     // File read operations
     Task<string> ReadAllTextAsync(string path, CancellationToken cancellationToken = default);
+
     IAsyncEnumerable<string> ReadLinesAsync(string path, CancellationToken cancellationToken = default);
+
     Task<byte[]> 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<string> contents, CancellationToken cancellationToken = default);
+
     Task AppendAllTextAsync(string path, string contents, CancellationToken cancellationToken = default);
+
     Task AppendAllLinesAsync(string path, IEnumerable<string> 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))
+        {
+            throw new NotSupportedException(
+                "This file system provider does not support atomic overwrite moves.");
+        }
+
+        MoveFile(sourcePath, destinationPath);
+    }
+
     bool FileExists(string path);
 
     // Directory operations
     void CreateDirectory(string path);
+
     void DeleteDirectory(string path, bool recursive);
+
     void MoveDirectory(string sourcePath, string destinationPath);
+
     bool DirectoryExists(string path);
+
     IEnumerable<string> EnumerateFiles(string path, string searchPattern, SearchOption searchOption);
+
     IEnumerable<string> EnumerateDirectories(string path, string searchPattern, SearchOption searchOption);
 
     // Path operations
     string GetTempPath();
+
     string GetRandomFileName();
+
     string Combine(params string[] paths);
+
     string GetRelativePath(string relativeTo, string path);
 }

src/ModularPipelines/FileSystem/SystemFileSystemProvider.cs

@@ -9,11 +9,13 @@ namespace ModularPipelines.FileSystem;
 public sealed class SystemFileSystemProvider : IFileSystemProvider
 {
     /// <summary>
-    /// Singleton instance for use when no DI is available.
+    /// Gets singleton instance for use when no DI is available.
     /// </summary>
     public static SystemFileSystemProvider Instance { get; } = new();
 
-    private SystemFileSystemProvider() { }
+    private SystemFileSystemProvider()
+    {
+    }
 
     // File read operations
     public Task<string> ReadAllTextAsync(string path, CancellationToken cancellationToken = default)
@@ -61,6 +63,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);
 

src/ModularPipelines/Http/Http.cs

@@ -28,36 +28,82 @@ public Http(
 
     public async Task<HttpResponseMessage> 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
-        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
+        // 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;
 
-        if (httpOptions.HttpClient != null)
+        try
         {
-            return await SendAndWrapLogging(httpOptions, effectiveCancellationToken).ConfigureAwait(false);
+            HttpResponseMessage WrapResponse(HttpResponseMessage response)
+            {
+                if (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(
+                        httpClient,
+                        httpOptions,
+                        WrapResponse,
+                        effectiveCancellationToken)
+                    .ConfigureAwait(false);
+            }
+
+            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, 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)
@@ -66,34 +112,60 @@ private HttpClient GetHttpClient(HttpLoggingType loggingType)
         return _httpClientFactory.CreateClient(clientName);
     }
 
-    private async Task<HttpResponseMessage> SendAndWrapLogging(HttpOptions httpOptions, CancellationToken cancellationToken)
+    private async Task<HttpResponseMessage> SendAndWrapLogging(
+        HttpClient httpClient,
+        HttpOptions httpOptions,
+        Func<HttpResponseMessage, HttpResponseMessage> wrapResponse,
+        CancellationToken cancellationToken)
     {
         var logger = _moduleLoggerProvider.GetLogger();
         var loggingOptions = GetEffectiveLoggingOptions(httpOptions);
 
         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();
 
-        var response = await httpOptions.HttpClient!.SendAsync(httpOptions.HttpRequestMessage, cancellationToken).ConfigureAwait(false);
+        var response = await httpClient.SendAsync(
+                httpOptions.HttpRequestMessage,
+                HttpCompletionOption.ResponseHeadersRead,
+                cancellationToken)
+            .ConfigureAwait(false);
 
-        LogStatusCode(response.StatusCode, httpOptions, loggingOptions);
-        LogDuration(stopWatch.Elapsed, httpOptions, loggingOptions);
-
-        if (httpOptions.LoggingType.HasFlag(HttpLoggingType.Response))
+        try
         {
-            await _httpLogger.PrintResponse(response, logger, loggingOptions).ConfigureAwait(false);
-        }
+            LogStatusCode(response.StatusCode, httpOptions, loggingOptions);
+            LogDuration(stopWatch.Elapsed, httpOptions, loggingOptions);
 
-        if (!httpOptions.ThrowOnNonSuccessStatusCode)
+            if (httpOptions.LoggingType.HasFlag(HttpLoggingType.Response))
+            {
+                await _httpLogger
+                    .PrintResponse(response, logger, loggingOptions, cancellationToken)
+                    .ConfigureAwait(false);
+            }
+
+            response = wrapResponse(response);
+
+            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)
@@ -112,6 +184,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)