Add progress tracking

Author: GarrettBeatty

generator/.DevConfigs/984a2bde-687f-4ed1-b6eb-03f15b257416.json

@@ -0,0 +1,11 @@
+{
+  "services": [
+    {
+      "serviceName": "S3",
+      "type": "patch",
+      "changeLogMessages": [
+        "Added progress tracking events to multipart download operations"
+      ]
+    }
+  ]
+}

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

@@ -46,8 +46,9 @@ internal interface IDownloadManager : IDisposable
         /// </summary>
         /// <param name="discoveryResult">Results from the discovery phase.</param>
         /// <param name="cancellationToken">A token to cancel the download operation.</param>
+        /// <param name="progressCallback">Optional callback for progress tracking events.</param>
         /// <returns>A task that completes when all downloads finish or an error occurs.</returns>
-        Task StartDownloadsAsync(DownloadDiscoveryResult discoveryResult, CancellationToken cancellationToken);
+        Task StartDownloadsAsync(DownloadDiscoveryResult discoveryResult, CancellationToken cancellationToken, EventHandler<WriteObjectProgressArgs> progressCallback = null);
 
         /// <summary>
         /// Exception that occurred during downloads, if any.

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

@@ -36,6 +36,9 @@ internal partial class MultipartDownloadCommand : BaseCommand<TransferUtilityDow
         private readonly IAmazonS3 _s3Client;
         private readonly TransferUtilityDownloadRequest _request;
         private readonly TransferUtilityConfig _config;
+        
+        // Track last known transferred bytes from coordinator's progress events
+        private long _lastKnownTransferredBytes;
 
         private static Logger Logger
         {
@@ -111,5 +114,89 @@ private FileDownloadConfiguration CreateConfiguration()
                 _request.FilePath
             );
         }
+
+        #region Event Firing Methods
+
+        /// <summary>
+        /// Fires the DownloadInitiatedEvent to notify subscribers that the download has started.
+        /// This event is fired exactly once at the beginning of the download operation.
+        /// </summary>
+        private void FireTransferInitiatedEvent()
+        {
+            var transferInitiatedEventArgs = new DownloadInitiatedEventArgs(_request, _request.FilePath);
+            _request.OnRaiseTransferInitiatedEvent(transferInitiatedEventArgs);
+        }
+
+        /// <summary>
+        /// Fires the DownloadCompletedEvent to notify subscribers that the download completed successfully.
+        /// This event is fired exactly once when all parts have been downloaded and assembled.
+        /// Downloads are complete, so transferred bytes equals total bytes.
+        /// </summary>
+        /// <param name="response">The unified TransferUtilityDownloadResponse containing S3 metadata</param>
+        /// <param name="totalBytes">The total number of bytes in the file</param>
+        private void FireTransferCompletedEvent(TransferUtilityDownloadResponse response, long totalBytes)
+        {
+            var transferCompletedEventArgs = new DownloadCompletedEventArgs(
+                _request,
+                response,
+                _request.FilePath,
+                totalBytes,
+                totalBytes);
+            _request.OnRaiseTransferCompletedEvent(transferCompletedEventArgs);
+        }
+
+        /// <summary>
+        /// Fires the DownloadFailedEvent to notify subscribers that the download failed.
+        /// This event is fired exactly once when an error occurs during the download.
+        /// Uses the last known transferred bytes from progress tracking.
+        /// </summary>
+        /// <param name="totalBytes">Total file size if known, otherwise -1</param>
+        private void FireTransferFailedEvent(long totalBytes = -1)
+        {
+            var eventArgs = new DownloadFailedEventArgs(
+                _request,
+                _request.FilePath,
+                System.Threading.Interlocked.Read(ref _lastKnownTransferredBytes),
+                totalBytes);
+            _request.OnRaiseTransferFailedEvent(eventArgs);
+        }
+
+        #endregion
+
+        #region Progress Tracking
+
+        /// <summary>
+        /// Callback for part download progress.
+        /// Forwards the aggregated progress events from the coordinator to the user's progress callback.
+        /// The coordinator has already aggregated progress across all concurrent part downloads.
+        /// Tracks the last known transferred bytes for failure reporting.
+        /// </summary>
+        /// <param name="sender">The event sender (coordinator)</param>
+        /// <param name="e">Aggregated progress information from the coordinator</param>
+        internal void DownloadPartProgressEventCallback(object sender, WriteObjectProgressArgs e)
+        {
+            // Track last known transferred bytes using Exchange (not Add).
+            // 
+            // Why Exchange? The coordinator already aggregates increments from concurrent parts:
+            //   Coordinator receives: Part 1: +512 bytes, Part 2: +1024 bytes, Part 3: +768 bytes
+            //   Coordinator aggregates: 0 -> 512 -> 1536 -> 2304 (using Interlocked.Add)
+            //   Coordinator passes to us: e.TransferredBytes = 2304 (pre-aggregated total)
+            //
+            // We receive the TOTAL (e.TransferredBytes = 2304), not an increment (+768).
+            // Using Add here would incorrectly accumulate totals: 0 + 2304 + 2304 + ... = wrong!
+            // Using Exchange correctly stores the latest total: 2304 (overwrite previous value).
+            //
+            // Compare to other commands (SimpleUploadCommand, DownloadCommand) which receive
+            // INCREMENTS directly from SDK streams and must use Add to accumulate them.
+            System.Threading.Interlocked.Exchange(ref _lastKnownTransferredBytes, e.TransferredBytes);
+            
+            // Set the Request property to enable access to the original download request
+            e.Request = _request;
+            
+            // Forward the coordinator's aggregated progress event to the user
+            _request.OnRaiseProgressEvent(e);
+        }
+
+        #endregion
     }
 }

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

@@ -54,6 +54,17 @@ internal class MultipartDownloadManager : IDownloadManager
         private string _savedETag;
         private int _discoveredPartCount;
 
+        // Progress tracking fields for multipart download aggregation
+        private long _totalTransferredBytes = 0;
+        private long _totalObjectSize = 0;
+        private EventHandler<WriteObjectProgressArgs> _userProgressCallback;
+        
+        // Atomic flag to ensure completion event fires exactly once
+        // Without this, concurrent parts completing simultaneously can both see
+        // transferredBytes >= _totalObjectSize and fire duplicate completion events
+        // Uses int instead of bool because Interlocked.CompareExchange requires reference types
+        private int _completionEventFired = 0;  // 0 = false, 1 = true
+
         private Logger Logger
         {
             get { return Logger.GetLogger(typeof(TransferUtility)); }
@@ -133,13 +144,17 @@ public async Task<DownloadDiscoveryResult> DiscoverDownloadStrategyAsync(Cancell
         }
 
         /// <inheritdoc/>
-        public async Task StartDownloadsAsync(DownloadDiscoveryResult discoveryResult, CancellationToken cancellationToken)
+        public async Task StartDownloadsAsync(DownloadDiscoveryResult discoveryResult, CancellationToken cancellationToken, EventHandler<WriteObjectProgressArgs> progressCallback = null)
         {
             ThrowIfDisposed();
 
             if (discoveryResult == null)
                 throw new ArgumentNullException(nameof(discoveryResult));
 
+            // Store for progress aggregation
+            _userProgressCallback = progressCallback;
+            _totalObjectSize = discoveryResult.ObjectSize;
+
             Logger.DebugFormat("MultipartDownloadManager: Starting downloads - TotalParts={0}, IsSinglePart={1}",
                 discoveryResult.TotalParts, discoveryResult.IsSinglePart);
 
@@ -151,10 +166,27 @@ public async Task StartDownloadsAsync(DownloadDiscoveryResult discoveryResult, C
                 // Prepare the data handler (e.g., create temp files for file-based downloads)
                 await _dataHandler.PrepareAsync(discoveryResult, cancellationToken).ConfigureAwait(false);
 
+                // Create delegate once and reuse for all parts
+                var wrappedCallback = progressCallback != null 
+                    ? new EventHandler<WriteObjectProgressArgs>(DownloadPartProgressEventCallback)
+                    : null;
+                
+                // Attach progress callback to Part 1's response if provided
+                if (wrappedCallback != null)
+                {
+                    discoveryResult.InitialResponse.WriteObjectProgressEvent += wrappedCallback;
+                }
+                
                 // Process Part 1 from InitialResponse (applies to both single-part and multipart)
                 Logger.DebugFormat("MultipartDownloadManager: Buffering Part 1 from discovery response");
                 await _dataHandler.ProcessPartAsync(1, discoveryResult.InitialResponse, cancellationToken).ConfigureAwait(false);
 
+                // Detach the event handler after processing to prevent memory leak
+                if (wrappedCallback != null)
+                {
+                    discoveryResult.InitialResponse.WriteObjectProgressEvent -= wrappedCallback;
+                }
+
                 if (discoveryResult.IsSinglePart)
                 {
                     // Single-part: Part 1 is the entire object
@@ -169,7 +201,7 @@ public async Task StartDownloadsAsync(DownloadDiscoveryResult discoveryResult, C
 
                 for (int partNum = 2; partNum <= discoveryResult.TotalParts; partNum++)
                 {
-                    var task = CreateDownloadTaskAsync(partNum, discoveryResult.ObjectSize, internalCts.Token);
+                    var task = CreateDownloadTaskAsync(partNum, discoveryResult.ObjectSize, wrappedCallback, internalCts.Token);
                     downloadTasks.Add(task);
                 }
 
@@ -245,7 +277,7 @@ public async Task StartDownloadsAsync(DownloadDiscoveryResult discoveryResult, C
 
 
 
-        private async Task CreateDownloadTaskAsync(int partNumber, long objectSize, CancellationToken cancellationToken)
+        private async Task CreateDownloadTaskAsync(int partNumber, long objectSize, EventHandler<WriteObjectProgressArgs> progressCallback, CancellationToken cancellationToken)
         {
                 Logger.DebugFormat("MultipartDownloadManager: [Part {0}] Waiting for buffer space", partNumber);
 
@@ -301,6 +333,12 @@ private async Task CreateDownloadTaskAsync(int partNumber, long objectSize, Canc
                     }
 
                     response = await _s3Client.GetObjectAsync(getObjectRequest, cancellationToken).ConfigureAwait(false);
+                    
+                    // Attach progress callback to response if provided
+                    if (progressCallback != null)
+                    {
+                        response.WriteObjectProgressEvent += progressCallback;
+                    }
 
                     Logger.DebugFormat("MultipartDownloadManager: [Part {0}] GetObject response received - ContentLength={1}",
                         partNumber, response.ContentLength);
@@ -553,6 +591,53 @@ internal void ValidateContentRange(GetObjectResponse response, int partNumber, l
             }
         }
 
+        /// <summary>
+        /// Creates progress args with aggregated values for multipart downloads.
+        /// </summary>
+        private WriteObjectProgressArgs CreateProgressArgs(long incrementTransferred, long transferredBytes, bool completed = false)
+        {
+            string filePath = (_request as TransferUtilityDownloadRequest)?.FilePath;
+            
+            return new WriteObjectProgressArgs(
+                _request.BucketName,
+                _request.Key,
+                filePath,
+                _request.VersionId,
+                incrementTransferred,
+                transferredBytes,
+                _totalObjectSize,
+                completed
+            );
+        }
+
+        /// <summary>
+        /// Progress aggregation callback that combines progress across all concurrent part downloads.
+        /// Uses thread-safe counter increment to handle concurrent updates.
+        /// Detects completion naturally when transferred bytes reaches total size.
+        /// Uses atomic flag to ensure completion event fires exactly once.
+        /// </summary>
+        private void DownloadPartProgressEventCallback(object sender, WriteObjectProgressArgs e)
+        {
+            long transferredBytes = Interlocked.Add(ref _totalTransferredBytes, e.IncrementTransferred);
+            
+            // Use atomic CompareExchange to ensure only first thread fires completion
+            bool isComplete = false;
+            if (transferredBytes >= _totalObjectSize)
+            {
+                // CompareExchange returns the original value before the exchange
+                // If original value was 0 (false), we're the first thread and should fire completion
+                int originalValue = Interlocked.CompareExchange(ref _completionEventFired, 1, 0);
+                if (originalValue == 0)  // Was false, now set to true
+                {
+                    isComplete = true;
+                }
+            }
+            
+            // Create and fire aggregated progress event
+            var aggregatedArgs = CreateProgressArgs(e.IncrementTransferred, transferredBytes, isComplete);
+            _userProgressCallback?.Invoke(this, aggregatedArgs);
+        }
+
         private void ThrowIfDisposed()
         {
             if (_disposed)

sdk/src/Services/S3/Custom/Transfer/Internal/_async/MultipartDownloadCommand.async.cs

@@ -33,6 +33,9 @@ public override async Task<TransferUtilityDownloadResponse> ExecuteAsync(Cancell
         {
             // Validate request parameters
             ValidateRequest();
+            
+            // Fire initiated event before starting any network operations
+            FireTransferInitiatedEvent();
 
             // Create configuration from request settings
             var config = CreateConfiguration();
@@ -54,19 +57,23 @@ public override async Task<TransferUtilityDownloadResponse> ExecuteAsync(Cancell
                     dataHandler,
                     RequestEventHandler))
                 {
+                    long totalBytes = -1;
                     try
                     {
                         // Step 1: Discover download strategy (PART or RANGE) and get metadata
                         Logger.DebugFormat("MultipartDownloadCommand: Discovering download strategy");
                         var discoveryResult = await coordinator.DiscoverDownloadStrategyAsync(cancellationToken)
                             .ConfigureAwait(false);
 
+                        totalBytes = discoveryResult.ObjectSize;
+
+                        
                         Logger.DebugFormat("MultipartDownloadCommand: Discovered {0} part(s), total size: {1} bytes, IsSinglePart={2}",
                             discoveryResult.TotalParts, discoveryResult.ObjectSize, discoveryResult.IsSinglePart);
 
                         // Step 2: Start concurrent downloads for all parts
-                        Logger.DebugFormat("MultipartDownloadCommand: Starting downloads for {0} part(s)", discoveryResult.TotalParts);
-                        await coordinator.StartDownloadsAsync(discoveryResult, cancellationToken)
+                        Logger.DebugFormat("Starting downloads for {0} part(s)", discoveryResult.TotalParts);
+                        await coordinator.StartDownloadsAsync(discoveryResult, cancellationToken, DownloadPartProgressEventCallback)
                             .ConfigureAwait(false);
 
                         // Step 2b: Wait for all downloads to complete before returning
@@ -110,11 +117,18 @@ await coordinator.StartDownloadsAsync(discoveryResult, cancellationToken)
                             mappedResponse.ChecksumSHA256 = null;
                         }
 
+                        // Fire completed event
+                        FireTransferCompletedEvent(mappedResponse, totalBytes);
+                        
                         return mappedResponse;
                     }
                     catch (Exception ex)
                     {
                         Logger.Error(ex, "Exception during multipart download");
+                        
+                        // Fire failed event
+                        FireTransferFailedEvent(totalBytes);
+                        
                         throw;
                     }
                 }