Norm PR comments

Author: GarrettBeatty

generator/.DevConfigs/9d07dc1e-d82d-4f94-8700-c7b57f872042.json

@@ -4,7 +4,7 @@
       "serviceName": "S3",
       "type": "minor",
       "changeLogMessages": [
-        "Create OpenStreamWithResponseAsync API and implement multi part download"
+        "Created new OpenStreamWithResponseAsync method on the Amazon.S3.Transfer.TransferUtility class. The new operation supports downloading in parallel parts of the S3 object in the background while reading from the stream for improved performance."
       ]
     }
   ]

sdk/src/Services/S3/Custom/Transfer/Internal/BufferedMultipartStream.cs

@@ -42,6 +42,7 @@ internal class BufferedMultipartStream : Stream
         private bool _initialized = false;
         private bool _disposed = false;
         private DownloadDiscoveryResult _discoveryResult;
+        private long _totalBytesRead = 0;
 
         /// <summary>
         /// Gets the discovery result containing metadata from the initial GetObject response.
@@ -143,6 +144,12 @@ public override async Task<int> ReadAsync(byte[] buffer, int offset, int count,
             var bytesRead = await _partBufferManager.ReadAsync(buffer, offset, count, cancellationToken)
                 .ConfigureAwait(false);
 
+            // Track total bytes read for Position property
+            if (bytesRead > 0)
+            {
+                Interlocked.Add(ref _totalBytesRead, bytesRead);
+            }
+            
             return bytesRead;
         }
 
@@ -153,11 +160,25 @@ public override async Task<int> ReadAsync(byte[] buffer, int offset, int count,
         public override bool CanSeek => false;
         public override bool CanWrite => false;
 
-        public override long Length => throw new NotSupportedException("Length not supported for multipart download streams");
+        public override long Length 
+        { 
+            get 
+            {
+                if (!_initialized)
+                    throw new InvalidOperationException("Stream must be initialized before accessing Length");
+                return _discoveryResult.ObjectSize;
+            }
+        }
+
         public override long Position 
         { 
-            get => throw new NotSupportedException("Position not supported for multipart download streams"); 
-            set => throw new NotSupportedException("Position not supported for multipart download streams"); 
+            get 
+            {
+                if (!_initialized)
+                    throw new InvalidOperationException("Stream must be initialized before accessing Position");
+                return Interlocked.Read(ref _totalBytesRead);
+            }
+            set => throw new NotSupportedException("Position setter not supported for read-only streams"); 
         }
 
         public override void Flush() { }

sdk/src/Services/S3/Custom/Transfer/Internal/BufferedPartDataHandler.cs

@@ -85,15 +85,17 @@ private async Task<StreamPartBuffer> BufferPartFromResponseAsync(
             CancellationToken cancellationToken)
         {
             StreamPartBuffer downloadedPart = null;
-            byte[] partBuffer = null;
 
             try
             {
                 // Use ContentLength to determine exact bytes to read and allocate
                 long expectedBytes = response.ContentLength;
                 int initialBufferSize = (int)expectedBytes;
 
-                partBuffer = ArrayPool<byte>.Shared.Rent(initialBufferSize);
+                downloadedPart = StreamPartBuffer.Create(partNumber, initialBufferSize);
+                
+                // Get reference to the buffer for writing
+                var partBuffer = downloadedPart.ArrayPoolBuffer;
 
                 int totalRead = 0;
 
@@ -104,13 +106,6 @@ private async Task<StreamPartBuffer> BufferPartFromResponseAsync(
                     int remainingBytes = (int)(expectedBytes - totalRead);
                     int bufferSpace = partBuffer.Length - totalRead;
 
-                    // Expand buffer if needed BEFORE reading
-                    if (bufferSpace == 0)
-                    {
-                        partBuffer = ExpandPartBuffer(partBuffer, totalRead);
-                        bufferSpace = partBuffer.Length - totalRead;
-                    }
-                    
                     // Read size is minimum of: remaining bytes needed, buffer space available, and configured buffer size
                     int readSize = Math.Min(Math.Min(remainingBytes, bufferSpace), _config.BufferSize);
 
@@ -130,36 +125,17 @@ private async Task<StreamPartBuffer> BufferPartFromResponseAsync(
                     totalRead += bytesRead;
                 }
 
-                // Create ArrayPool-based StreamPartBuffer (no copying!)
-                downloadedPart = StreamPartBuffer.FromArrayPoolBuffer(
-                    partNumber,
-                    partBuffer,     // Transfer ownership to StreamPartBuffer
-                    totalRead       // Actual data length
-                );
-
-                partBuffer = null; // Clear reference to prevent return in finally
+                // Set the length to reflect actual bytes read
+                downloadedPart.SetLength(totalRead);
+                
                 return downloadedPart;
             }
-            finally
+            catch
             {
-                // Only return partBuffer if ownership wasn't transferred
-                if (partBuffer != null)
-                    ArrayPool<byte>.Shared.Return(partBuffer);
+                // If something goes wrong, StreamPartBuffer.Dispose() will handle cleanup
+                downloadedPart?.Dispose();
+                throw;
             }
         }
-
-        private byte[] ExpandPartBuffer(byte[] currentBuffer, int validDataLength)
-        {
-            var newSize = Math.Max(currentBuffer.Length * 2, validDataLength + (int)_config.TargetPartSizeBytes);
-            var expandedBuffer = ArrayPool<byte>.Shared.Rent(newSize);
-
-            // Single copy of valid data to expanded buffer
-            Buffer.BlockCopy(currentBuffer, 0, expandedBuffer, 0, validDataLength);
-
-            // Return old buffer to pool
-            ArrayPool<byte>.Shared.Return(currentBuffer);
-
-            return expandedBuffer;
-        }
     }
 }

sdk/src/Services/S3/Custom/Transfer/Internal/MultipartDownloadCoordinator.cs

@@ -47,7 +47,6 @@ internal class MultipartDownloadCoordinator : IDownloadCoordinator
         private Exception _downloadException;
         private bool _disposed = false;
         private bool _discoveryCompleted = false;
-        private readonly object _lockObject = new object();
 
         private string _savedETag;
         private int _discoveredPartCount;
@@ -67,22 +66,16 @@ public Exception DownloadException
         { 
             get 
             { 
-                lock (_lockObject)
-                {
-                    return _downloadException;
-                }
+                return _downloadException;
             }
         }
 
         public async Task<DownloadDiscoveryResult> DiscoverDownloadStrategyAsync(CancellationToken cancellationToken)
         {
             ThrowIfDisposed();
 
-            lock (_lockObject)
-            {
-                if (_discoveryCompleted)
-                    throw new InvalidOperationException("Discovery has already been performed");
-            }
+            if (_discoveryCompleted)
+                throw new InvalidOperationException("Discovery has already been performed");
 
             try
             {
@@ -91,19 +84,13 @@ public async Task<DownloadDiscoveryResult> DiscoverDownloadStrategyAsync(Cancell
                     ? await DiscoverUsingPartStrategyAsync(cancellationToken).ConfigureAwait(false)
                     : await DiscoverUsingRangeStrategyAsync(cancellationToken).ConfigureAwait(false);
 
-                lock (_lockObject)
-                {
-                    _discoveryCompleted = true;
-                }
+                _discoveryCompleted = true;
 
                 return result;
             }
             catch (Exception ex)
             {
-                lock (_lockObject)
-                {
-                    _downloadException = ex;
-                }
+                _downloadException = ex;
                 throw;
             }
         }
@@ -161,10 +148,7 @@ public async Task StartDownloadsAsync(DownloadDiscoveryResult discoveryResult, C
             }
             catch (Exception ex)
             {
-                lock (_lockObject)
-                {
-                    _downloadException = ex;
-                }
+                _downloadException = ex;
 
                 _dataHandler.OnDownloadComplete(ex);
                 throw;
@@ -304,8 +288,8 @@ private async Task<DownloadDiscoveryResult> DiscoverUsingPartStrategyAsync(Cance
 
         private async Task<DownloadDiscoveryResult> DiscoverUsingRangeStrategyAsync(CancellationToken cancellationToken)
         {
-            // Get target part size for RANGE strategy
-            var targetPartSize = _request.IsSetPartSize() ? _request.PartSize : _config.TargetPartSizeBytes;
+            // Get target part size for RANGE strategy (already set in config from request or default)
+            var targetPartSize = _config.TargetPartSizeBytes;
 
             // SEP Ranged GET Step 1: "create a new GetObject request copying all fields in the original request. 
             // Set range value to bytes=0-{targetPartSizeBytes-1} to request the first part."
@@ -393,62 +377,62 @@ private GetObjectRequest CreateGetObjectRequest()
             return request;
         }
 
-        private (long startByte, long endByte) CalculatePartRange(int partNumber, long objectSize)
+        internal (long startByte, long endByte) CalculatePartRange(int partNumber, long objectSize)
         {
-            var targetPartSize = _request.IsSetPartSize() ? _request.PartSize : _config.TargetPartSizeBytes;
+            var targetPartSize = _config.TargetPartSizeBytes;
 
             var startByte = (partNumber - 1) * targetPartSize;
             var endByte = Math.Min(startByte + targetPartSize - 1, objectSize - 1);
             return (startByte, endByte);
         }
 
-        private long ExtractTotalSizeFromContentRange(string contentRange)
+        internal (long startByte, long endByte, long totalSize) ParseContentRange(string contentRange)
         {
             if (string.IsNullOrEmpty(contentRange))
-                throw new InvalidOperationException("Content-Range header is missing from range request response");
-
-            // Format: "bytes 0-{end}/{total-size}" or "bytes 0-{end}/*"
-            var parts = contentRange.Split('/');
-            if (parts.Length == 2 && parts[1] != "*")
-            {
-                if (long.TryParse(parts[1], out var totalSize))
-                {
-                    return totalSize;
-                }
-            }
+                throw new InvalidOperationException("Content-Range header is missing");
+
+            // Format: "bytes {start}-{end}/{total-size}"
+            var parts = contentRange.Replace("bytes ", "").Split('/');
+            if (parts.Length != 2)
+                throw new InvalidOperationException($"Invalid ContentRange format: {contentRange}");
+
+            // Parse byte range
+            var rangeParts = parts[0].Split('-');
+            if (rangeParts.Length != 2 ||
+                !long.TryParse(rangeParts[0], out var startByte) ||
+                !long.TryParse(rangeParts[1], out var endByte))
+                throw new InvalidOperationException($"Unable to parse ContentRange byte range: {contentRange}");
+
+            // Parse total size - S3 always returns exact sizes, never wildcards
+            if (parts[1] == "*")
+                throw new InvalidOperationException($"Unexpected wildcard in ContentRange total size: {contentRange}. S3 always returns exact object sizes.");
+            if (!long.TryParse(parts[1], out var totalSize))
+                throw new InvalidOperationException($"Unable to parse ContentRange total size: {contentRange}");
+
+            return (startByte, endByte, totalSize);
+        }
 
-            throw new InvalidOperationException($"Unable to parse Content-Range header: {contentRange}");
+        internal long ExtractTotalSizeFromContentRange(string contentRange)
+        {
+            var (_, _, totalSize) = ParseContentRange(contentRange);
+            return totalSize;
         }
 
-        private void ValidateContentRange(GetObjectResponse response, int partNumber, long objectSize)
+        internal void ValidateContentRange(GetObjectResponse response, int partNumber, long objectSize)
         {
             // Ranged GET Step 7: 
             // "validate that ContentRange matches with the requested range"            
             if (_request.MultipartDownloadType == MultipartDownloadType.RANGE)
             {
                 var (expectedStartByte, expectedEndByte) = CalculatePartRange(partNumber, objectSize);
 
-                // Parse actual ContentRange from response
-                // Format: "bytes {start}-{end}/{total}"
-                var contentRange = response.ContentRange;
-                if (string.IsNullOrEmpty(contentRange))
+                // Parse actual ContentRange from response using unified helper
+                if (string.IsNullOrEmpty(response.ContentRange))
                 {
                     throw new InvalidOperationException($"ContentRange header missing from part {partNumber} response");
                 }
 
-                var parts = contentRange.Replace("bytes ", "").Split('/');
-                if (parts.Length != 2)
-                {
-                    throw new InvalidOperationException($"Invalid ContentRange format: {contentRange}");
-                }
-
-                var rangeParts = parts[0].Split('-');
-                if (rangeParts.Length != 2 ||
-                    !long.TryParse(rangeParts[0], out var actualStartByte) ||
-                    !long.TryParse(rangeParts[1], out var actualEndByte))
-                {
-                    throw new InvalidOperationException($"Unable to parse ContentRange: {contentRange}");
-                }
+                var (actualStartByte, actualEndByte, _) = ParseContentRange(response.ContentRange);
 
                 // Validate range matches what we requested
                 if (actualStartByte != expectedStartByte || actualEndByte != expectedEndByte)