[Storage][DataMovement] Fix progress reports appearing out of order (Azure#47539)

jalauzon-msft · web-flow · commit 75cabf713386 · 2024-12-12T14:13:10.000-08:00
diff --git a/sdk/storage/Azure.Storage.DataMovement.Blobs/tests/ProgressHandlerTests.cs b/sdk/storage/Azure.Storage.DataMovement.Blobs/tests/ProgressHandlerTests.cs
@@ -8,6 +8,7 @@
 using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
+using Azure.Core.TestFramework;
 using Azure.Storage.DataMovement.Tests;
 using BaseBlobs::Azure.Storage.Blobs;
 using DMBlobs::Azure.Storage.DataMovement.Blobs;
@@ -298,7 +299,7 @@ await TransferAndAssertProgress(
                 waitTime: 30);
         }
 
-        [Ignore("https://github.com/Azure/azure-sdk-for-net/issues/35558")]
+        [LiveOnly]
         [Test]
         [TestCase(0)]
         [TestCase(150)]
@@ -310,22 +311,27 @@ public async Task ProgressHandler_PauseResume(int delayInMs)
 
             await PopulateTestContainer(source.Container);
 
-            StorageResourceContainer sourceResource =
-                new BlobStorageResourceContainer(source.Container);
-            StorageResourceContainer destinationResource =
-                new LocalDirectoryStorageResourceContainer(destination.DirectoryPath);
+            BlobsStorageResourceProvider blobProvider = new(TestEnvironment.Credential);
+            LocalFilesStorageResourceProvider localProvider = new();
 
-            TransferManager transferManager = new TransferManager();
+            TransferManagerOptions transferManagerOptions = new()
+            {
+                ResumeProviders = [blobProvider, localProvider]
+            };
+            TransferManager transferManager = new(transferManagerOptions);
 
-            TestProgressHandler progressHandler = new TestProgressHandler();
-            DataTransferOptions transferOptions = new DataTransferOptions()
+            TestProgressHandler progressHandler = new();
+            DataTransferOptions transferOptions = new()
             {
                 ProgressHandlerOptions = new ProgressHandlerOptions(progressHandler, true)
             };
-            TestEventsRaised testEventsRaised = new TestEventsRaised(transferOptions);
+            TestEventsRaised testEventsRaised = new(transferOptions);
 
             // Act - Start transfer
-            DataTransfer transfer = await transferManager.StartTransferAsync(sourceResource, destinationResource, transferOptions);
+            DataTransfer transfer = await transferManager.StartTransferAsync(
+                blobProvider.FromContainer(source.Container.Uri),
+                localProvider.FromDirectory(destination.DirectoryPath),
+                transferOptions);
 
             // TODO: This can likely be replaced with something better once mocking is in place
             // Wait for the transfer to start happening
@@ -334,7 +340,7 @@ public async Task ProgressHandler_PauseResume(int delayInMs)
             // Pause transfer
             CancellationTokenSource tokenSource = new CancellationTokenSource(TimeSpan.FromSeconds(30));
             await transferManager.PauseTransferIfRunningAsync(transfer.Id, tokenSource.Token);
-            Assert.AreEqual(DataTransferState.Paused, transfer.TransferStatus);
+            Assert.AreEqual(DataTransferState.Paused, transfer.TransferStatus.State);
 
             // Record the current number of progress updates to use during assertions
             int pause = progressHandler.Updates.Count;
@@ -348,7 +354,7 @@ public async Task ProgressHandler_PauseResume(int delayInMs)
             await resumeTransfer.WaitForCompletionAsync(tokenSource.Token);
 
             // Assert
-            Assert.AreEqual(DataTransferState.Completed, resumeTransfer.TransferStatus);
+            Assert.AreEqual(DataTransferState.Completed, resumeTransfer.TransferStatus.State);
             ProgressHandlerAsserts.AssertFileProgress(progressHandler.Updates, 5, pauseIndexes: pause);
             ProgressHandlerAsserts.AssertBytesTransferred(progressHandler.Updates, _expectedBytesTransferred);
         }
diff --git a/sdk/storage/Azure.Storage.DataMovement/api/Azure.Storage.DataMovement.net6.0.cs b/sdk/storage/Azure.Storage.DataMovement/api/Azure.Storage.DataMovement.net6.0.cs
@@ -51,7 +51,7 @@ public enum DataTransferOrder
     }
     public partial class DataTransferProgress
     {
-        protected internal DataTransferProgress() { }
+        internal DataTransferProgress() { }
         public long? BytesTransferred { get { throw null; } }
         public long CompletedCount { get { throw null; } }
         public long FailedCount { get { throw null; } }
diff --git a/sdk/storage/Azure.Storage.DataMovement/api/Azure.Storage.DataMovement.net8.0.cs b/sdk/storage/Azure.Storage.DataMovement/api/Azure.Storage.DataMovement.net8.0.cs
@@ -51,7 +51,7 @@ public enum DataTransferOrder
     }
     public partial class DataTransferProgress
     {
-        protected internal DataTransferProgress() { }
+        internal DataTransferProgress() { }
         public long? BytesTransferred { get { throw null; } }
         public long CompletedCount { get { throw null; } }
         public long FailedCount { get { throw null; } }
diff --git a/sdk/storage/Azure.Storage.DataMovement/api/Azure.Storage.DataMovement.netstandard2.0.cs b/sdk/storage/Azure.Storage.DataMovement/api/Azure.Storage.DataMovement.netstandard2.0.cs
@@ -51,7 +51,7 @@ public enum DataTransferOrder
     }
     public partial class DataTransferProgress
     {
-        protected internal DataTransferProgress() { }
+        internal DataTransferProgress() { }
         public long? BytesTransferred { get { throw null; } }
         public long CompletedCount { get { throw null; } }
         public long FailedCount { get { throw null; } }
diff --git a/sdk/storage/Azure.Storage.DataMovement/src/CommitChunkHandler.cs b/sdk/storage/Azure.Storage.DataMovement/src/CommitChunkHandler.cs
@@ -13,7 +13,7 @@ internal class CommitChunkHandler : IAsyncDisposable
         #region Delegate Definitions
         public delegate Task QueuePutBlockTaskInternal(long offset, long blockSize, long expectedLength, StorageResourceItemProperties properties);
         public delegate Task QueueCommitBlockTaskInternal(StorageResourceItemProperties sourceProperties);
-        public delegate void ReportProgressInBytes(long bytesWritten);
+        public delegate ValueTask ReportProgressInBytes(long bytesWritten);
         public delegate Task InvokeFailedEventHandlerInternal(Exception ex);
         #endregion Delegate Definitions
 
@@ -99,7 +99,7 @@ private async Task ProcessCommitRange(QueueStageChunkArgs args, CancellationToke
             try
             {
                 _bytesTransferred += args.BytesTransferred;
-                _reportProgressInBytes(args.BytesTransferred);
+                await _reportProgressInBytes(args.BytesTransferred).ConfigureAwait(false);
 
                 if (_bytesTransferred == _expectedLength)
                 {
diff --git a/sdk/storage/Azure.Storage.DataMovement/src/DataTransferProgress.cs b/sdk/storage/Azure.Storage.DataMovement/src/DataTransferProgress.cs
@@ -8,10 +8,7 @@ namespace Azure.Storage.DataMovement
     /// </summary>
     public class DataTransferProgress
     {
-        /// <summary>
-        /// Constructor for mocking.
-        /// </summary>
-        protected internal DataTransferProgress() { }
+        internal DataTransferProgress() { }
 
         /// <summary>
         /// Number of files that were transferred successfully.
diff --git a/sdk/storage/Azure.Storage.DataMovement/src/DownloadChunkHandler.cs b/sdk/storage/Azure.Storage.DataMovement/src/DownloadChunkHandler.cs
@@ -13,7 +13,7 @@ internal class DownloadChunkHandler : IAsyncDisposable
     {
         #region Delegate Definitions
         public delegate Task CopyToDestinationFileInternal(long offset, long length, Stream stream, long expectedLength, bool initial);
-        public delegate void ReportProgressInBytes(long bytesWritten);
+        public delegate ValueTask ReportProgressInBytes(long bytesWritten);
         public delegate Task QueueCompleteFileDownloadInternal();
         public delegate Task InvokeFailedEventHandlerInternal(Exception ex);
         #endregion Delegate Definitions
@@ -119,7 +119,8 @@ await _copyToDestinationFile(
                         _expectedLength,
                         initial: _bytesTransferred == 0).ConfigureAwait(false);
                 }
-                UpdateBytesAndRange(args.Length);
+                _bytesTransferred += args.Length;
+                await _reportProgressInBytes(args.Length).ConfigureAwait(false);
 
                 // Check if we finished downloading the blob
                 if (_bytesTransferred == _expectedLength)
@@ -139,11 +140,5 @@ await _copyToDestinationFile(
                 }
             }
         }
-
-        private void UpdateBytesAndRange(long bytesDownloaded)
-        {
-            _bytesTransferred += bytesDownloaded;
-            _reportProgressInBytes(bytesDownloaded);
-        }
     }
 }
diff --git a/sdk/storage/Azure.Storage.DataMovement/src/JobPartInternal.cs b/sdk/storage/Azure.Storage.DataMovement/src/JobPartInternal.cs
@@ -284,11 +284,11 @@ internal async Task OnTransferStateChangedAsync(DataTransferState transferState)
                 // Progress tracking, do before invoking the event below
                 if (transferState == DataTransferState.InProgress)
                 {
-                    _progressTracker.IncrementInProgressFiles();
+                    await _progressTracker.IncrementInProgressFilesAsync(_cancellationToken).ConfigureAwait(false);
                 }
                 else if (JobPartStatus.HasCompletedSuccessfully)
                 {
-                    _progressTracker.IncrementCompletedFiles();
+                    await _progressTracker.IncrementCompletedFilesAsync(_cancellationToken).ConfigureAwait(false);
                     await InvokeSingleCompletedArgAsync().ConfigureAwait(false);
                 }
 
@@ -309,13 +309,9 @@ await PartTransferStatusEventHandler.RaiseAsync(
             }
         }
 
-        /// <summary>
-        /// To change all transfer statues at the same time
-        /// </summary>
-        /// <param name="bytesTransferred"></param>
-        internal void ReportBytesWritten(long bytesTransferred)
+        protected async ValueTask ReportBytesWrittenAsync(long bytesTransferred)
         {
-            _progressTracker.IncrementBytesTransferred(bytesTransferred);
+            await _progressTracker.IncrementBytesTransferred(bytesTransferred, _cancellationToken).ConfigureAwait(false);
         }
 
         public async virtual Task InvokeSingleCompletedArgAsync()
@@ -356,7 +352,7 @@ await TransferSkippedEventHandler.RaiseAsync(
                     ClientDiagnostics)
                     .ConfigureAwait(false);
             }
-            _progressTracker.IncrementSkippedFiles();
+            await _progressTracker.IncrementSkippedFilesAsync(_cancellationToken).ConfigureAwait(false);
 
             // Update the JobPartStatus. If was already updated (e.g. there was a failed item before)
             // then don't raise the PartTransferStatusEventHandler
@@ -413,7 +409,7 @@ await TransferFailedEventHandler.RaiseAsync(
                         ClientDiagnostics)
                         .ConfigureAwait(false);
                 }
-                _progressTracker.IncrementFailedFiles();
+                await _progressTracker.IncrementFailedFilesAsync(_cancellationToken).ConfigureAwait(false);
 
                 // Update the JobPartStatus. If was already updated (e.g. there was a failed item before)
                 // then don't raise the PartTransferStatusEventHandler
diff --git a/sdk/storage/Azure.Storage.DataMovement/src/ServiceToServiceJobPart.cs b/sdk/storage/Azure.Storage.DataMovement/src/ServiceToServiceJobPart.cs
@@ -275,7 +275,7 @@ await _destinationResource.CopyFromUriAsync(
                     options: options,
                     cancellationToken: _cancellationToken).ConfigureAwait(false);
 
-                ReportBytesWritten(completeLength);
+                await ReportBytesWrittenAsync(completeLength).ConfigureAwait(false);
                 await OnTransferStateChangedAsync(DataTransferState.Completed).ConfigureAwait(false);
             }
             catch (RequestFailedException exception)
@@ -316,7 +316,7 @@ await _destinationResource.CopyBlockFromUriAsync(
                     cancellationToken: _cancellationToken).ConfigureAwait(false);
 
                 // Report first chunk written to progress tracker
-                ReportBytesWritten(blockSize);
+                await ReportBytesWrittenAsync(blockSize).ConfigureAwait(false);
 
                 if (blockSize == length)
                 {
@@ -346,7 +346,7 @@ internal static CommitChunkHandler.Behaviors GetBlockListCommitHandlerBehaviors(
             {
                 QueuePutBlockTask = jobPart.QueueStageBlockRequest,
                 QueueCommitBlockTask = jobPart.QueueCompleteTransferAsync,
-                ReportProgressInBytes = jobPart.ReportBytesWritten,
+                ReportProgressInBytes = jobPart.ReportBytesWrittenAsync,
                 InvokeFailedHandler = jobPart.InvokeFailedArgAsync,
             };
         }
diff --git a/sdk/storage/Azure.Storage.DataMovement/src/StreamToUriJobPart.cs b/sdk/storage/Azure.Storage.DataMovement/src/StreamToUriJobPart.cs
@@ -328,7 +328,7 @@ await _destinationResource.CopyFromStreamAsync(
                 }
 
                 // Report bytes written before completion
-                ReportBytesWritten(blockSize);
+                await ReportBytesWrittenAsync(blockSize).ConfigureAwait(false);
 
                 // Set completion status to completed
                 await OnTransferStateChangedAsync(DataTransferState.Completed).ConfigureAwait(false);
@@ -360,7 +360,7 @@ await _destinationResource.CopyFromStreamAsync(
                         cancellationToken: _cancellationToken).ConfigureAwait(false);
                 }
 
-                ReportBytesWritten(blockSize);
+                await ReportBytesWrittenAsync(blockSize).ConfigureAwait(false);
             }
         }
 
@@ -372,7 +372,7 @@ internal static CommitChunkHandler.Behaviors GetBlockListCommitHandlerBehaviors(
             {
                 QueuePutBlockTask = jobPart.QueueStageBlockRequest,
                 QueueCommitBlockTask = jobPart.QueueCompleteTransferAsync,
-                ReportProgressInBytes = jobPart.ReportBytesWritten,
+                ReportProgressInBytes = jobPart.ReportBytesWrittenAsync,
                 InvokeFailedHandler = jobPart.InvokeFailedArgAsync,
             };
         }
diff --git a/sdk/storage/Azure.Storage.DataMovement/src/TransferJobInternal.cs b/sdk/storage/Azure.Storage.DataMovement/src/TransferJobInternal.cs
@@ -12,7 +12,7 @@
 
 namespace Azure.Storage.DataMovement
 {
-    internal class TransferJobInternal : IDisposable
+    internal class TransferJobInternal : IAsyncDisposable
     {
         internal delegate Task<JobPartInternal> CreateJobPartSingleAsync(
             TransferJobInternal job,
@@ -261,17 +261,14 @@ internal TransferJobInternal(
             _progressTracker = new TransferProgressTracker(transferOptions?.ProgressHandlerOptions);
         }
 
-        public void Dispose()
-        {
-            DisposeHandlers();
-        }
-
-        public void DisposeHandlers()
+        public async ValueTask DisposeAsync()
         {
             if (JobPartStatusEvents != default)
             {
                 JobPartStatusEvents -= JobPartStatusEventAsync;
             }
+            // This will block until all pending progress reports have gone out
+            await _progressTracker.DisposeAsync().ConfigureAwait(false);
         }
 
         /// <summary>
@@ -609,7 +606,7 @@ public async Task OnJobStateChangedAsync(DataTransferState state)
                 if (state == DataTransferState.Completed ||
                     state == DataTransferState.Paused)
                 {
-                    DisposeHandlers();
+                    await DisposeAsync().ConfigureAwait(false);
                 }
 
                 await OnJobPartStatusChangedAsync().ConfigureAwait(false);
@@ -720,9 +717,9 @@ internal HashSet<Uri> GetJobPartSourceResourcePaths()
             return new HashSet<Uri>(_jobParts.Select(x => x._sourceResource.Uri));
         }
 
-        internal void IncrementJobParts()
+        internal async ValueTask IncrementJobParts()
         {
-            _progressTracker.IncrementQueuedFiles();
+            await _progressTracker.IncrementQueuedFilesAsync(_cancellationToken).ConfigureAwait(false);
         }
     }
 }
diff --git a/sdk/storage/Azure.Storage.DataMovement/src/TransferManager.cs b/sdk/storage/Azure.Storage.DataMovement/src/TransferManager.cs
@@ -109,7 +109,7 @@ private async Task ProcessJobAsync(TransferJobInternal job, CancellationToken ca
         {
             await foreach (JobPartInternal partItem in job.ProcessJobToJobPartAsync().ConfigureAwait(false))
             {
-                job.IncrementJobParts();
+                await job.IncrementJobParts().ConfigureAwait(false);
                 await _partsProcessor.QueueAsync(partItem, cancellationToken).ConfigureAwait(false);
             }
         }
diff --git a/sdk/storage/Azure.Storage.DataMovement/src/TransferProgressTracker.cs b/sdk/storage/Azure.Storage.DataMovement/src/TransferProgressTracker.cs
diff --git a/sdk/storage/Azure.Storage.DataMovement/src/UriToStreamJobPart.cs b/sdk/storage/Azure.Storage.DataMovement/src/UriToStreamJobPart.cs
diff --git a/sdk/storage/Azure.Storage.DataMovement/tests/CommitChunkHandlerTests.cs b/sdk/storage/Azure.Storage.DataMovement/tests/CommitChunkHandlerTests.cs
diff --git a/sdk/storage/Azure.Storage.DataMovement/tests/DownloadChunkHandlerTests.cs b/sdk/storage/Azure.Storage.DataMovement/tests/DownloadChunkHandlerTests.cs

Original file line number	Diff line number	Diff line change
`@@ -51,7 +51,7 @@ public enum DataTransferOrder`
`51`	`51`	`}`
`52`	`52`	`public partial class DataTransferProgress`
`53`	`53`	`{`
`54`		`- protected internal DataTransferProgress() { }`
	`54`	`+ internal DataTransferProgress() { }`
`55`	`55`	`public long? BytesTransferred { get { throw null; } }`
`56`	`56`	`public long CompletedCount { get { throw null; } }`
`57`	`57`	`public long FailedCount { get { throw null; } }`
Original file line number	Diff line number	Diff line change
`@@ -13,7 +13,7 @@ internal class DownloadChunkHandler : IAsyncDisposable`
`13`	`13`	`{`
`14`	`14`	`#region Delegate Definitions`
`15`	`15`	`public delegate Task CopyToDestinationFileInternal(long offset, long length, Stream stream, long expectedLength, bool initial);`
`16`		`- public delegate void ReportProgressInBytes(long bytesWritten);`
	`16`	`+ public delegate ValueTask ReportProgressInBytes(long bytesWritten);`
`17`	`17`	`public delegate Task QueueCompleteFileDownloadInternal();`
`18`	`18`	`public delegate Task InvokeFailedEventHandlerInternal(Exception ex);`
`19`	`19`	`#endregion Delegate Definitions`
`@@ -119,7 +119,8 @@ await _copyToDestinationFile(`
`119`	`119`	`_expectedLength,`
`120`	`120`	`initial: _bytesTransferred == 0).ConfigureAwait(false);`
`121`	`121`	`}`
`122`		`- UpdateBytesAndRange(args.Length);`
	`122`	`+ _bytesTransferred += args.Length;`
	`123`	`+ await _reportProgressInBytes(args.Length).ConfigureAwait(false);`
`123`	`124`
`124`	`125`	`// Check if we finished downloading the blob`
`125`	`126`	`if (_bytesTransferred == _expectedLength)`
`@@ -139,11 +140,5 @@ await _copyToDestinationFile(`
`139`	`140`	`}`
`140`	`141`	`}`
`141`	`142`	`}`
`142`		`-`
`143`		`- private void UpdateBytesAndRange(long bytesDownloaded)`
`144`		`- {`
`145`		`- _bytesTransferred += bytesDownloaded;`
`146`		`- _reportProgressInBytes(bytesDownloaded);`
`147`		`- }`
`148`	`143`	`}`
`149`	`144`	`}`
Original file line number	Diff line number	Diff line change
`@@ -284,11 +284,11 @@ internal async Task OnTransferStateChangedAsync(DataTransferState transferState)`
`284`	`284`	`// Progress tracking, do before invoking the event below`
`285`	`285`	`if (transferState == DataTransferState.InProgress)`
`286`	`286`	`{`
`287`		`- _progressTracker.IncrementInProgressFiles();`
	`287`	`+ await _progressTracker.IncrementInProgressFilesAsync(_cancellationToken).ConfigureAwait(false);`
`288`	`288`	`}`
`289`	`289`	`else if (JobPartStatus.HasCompletedSuccessfully)`
`290`	`290`	`{`
`291`		`- _progressTracker.IncrementCompletedFiles();`
	`291`	`+ await _progressTracker.IncrementCompletedFilesAsync(_cancellationToken).ConfigureAwait(false);`
`292`	`292`	`await InvokeSingleCompletedArgAsync().ConfigureAwait(false);`
`293`	`293`	`}`
`294`	`294`
`@@ -309,13 +309,9 @@ await PartTransferStatusEventHandler.RaiseAsync(`
`309`	`309`	`}`
`310`	`310`	`}`
`311`	`311`
`312`		`- /// <summary>`
`313`		`- /// To change all transfer statues at the same time`
`314`		`- /// </summary>`
`315`		`- /// <param name="bytesTransferred"></param>`
`316`		`- internal void ReportBytesWritten(long bytesTransferred)`
	`312`	`+ protected async ValueTask ReportBytesWrittenAsync(long bytesTransferred)`
`317`	`313`	`{`
`318`		`- _progressTracker.IncrementBytesTransferred(bytesTransferred);`
	`314`	`+ await _progressTracker.IncrementBytesTransferred(bytesTransferred, _cancellationToken).ConfigureAwait(false);`
`319`	`315`	`}`
`320`	`316`
`321`	`317`	`public async virtual Task InvokeSingleCompletedArgAsync()`
`@@ -356,7 +352,7 @@ await TransferSkippedEventHandler.RaiseAsync(`
`356`	`352`	`ClientDiagnostics)`
`357`	`353`	`.ConfigureAwait(false);`
`358`	`354`	`}`
`359`		`- _progressTracker.IncrementSkippedFiles();`
	`355`	`+ await _progressTracker.IncrementSkippedFilesAsync(_cancellationToken).ConfigureAwait(false);`
`360`	`356`
`361`	`357`	`// Update the JobPartStatus. If was already updated (e.g. there was a failed item before)`
`362`	`358`	`// then don't raise the PartTransferStatusEventHandler`
`@@ -413,7 +409,7 @@ await TransferFailedEventHandler.RaiseAsync(`
`413`	`409`	`ClientDiagnostics)`
`414`	`410`	`.ConfigureAwait(false);`
`415`	`411`	`}`
`416`		`- _progressTracker.IncrementFailedFiles();`
	`412`	`+ await _progressTracker.IncrementFailedFilesAsync(_cancellationToken).ConfigureAwait(false);`
`417`	`413`
`418`	`414`	`// Update the JobPartStatus. If was already updated (e.g. there was a failed item before)`
`419`	`415`	`// then don't raise the PartTransferStatusEventHandler`
Original file line number	Diff line number	Diff line change
`@@ -275,7 +275,7 @@ await _destinationResource.CopyFromUriAsync(`
`275`	`275`	`options: options,`
`276`	`276`	`cancellationToken: _cancellationToken).ConfigureAwait(false);`
`277`	`277`
`278`		`- ReportBytesWritten(completeLength);`
	`278`	`+ await ReportBytesWrittenAsync(completeLength).ConfigureAwait(false);`
`279`	`279`	`await OnTransferStateChangedAsync(DataTransferState.Completed).ConfigureAwait(false);`
`280`	`280`	`}`
`281`	`281`	`catch (RequestFailedException exception)`
`@@ -316,7 +316,7 @@ await _destinationResource.CopyBlockFromUriAsync(`
`316`	`316`	`cancellationToken: _cancellationToken).ConfigureAwait(false);`
`317`	`317`
`318`	`318`	`// Report first chunk written to progress tracker`
`319`		`- ReportBytesWritten(blockSize);`
	`319`	`+ await ReportBytesWrittenAsync(blockSize).ConfigureAwait(false);`
`320`	`320`
`321`	`321`	`if (blockSize == length)`
`322`	`322`	`{`
`@@ -346,7 +346,7 @@ internal static CommitChunkHandler.Behaviors GetBlockListCommitHandlerBehaviors(`
`346`	`346`	`{`
`347`	`347`	`QueuePutBlockTask = jobPart.QueueStageBlockRequest,`
`348`	`348`	`QueueCommitBlockTask = jobPart.QueueCompleteTransferAsync,`
`349`		`- ReportProgressInBytes = jobPart.ReportBytesWritten,`
	`349`	`+ ReportProgressInBytes = jobPart.ReportBytesWrittenAsync,`
`350`	`350`	`InvokeFailedHandler = jobPart.InvokeFailedArgAsync,`
`351`	`351`	`};`
`352`	`352`	`}`
Original file line number	Diff line number	Diff line change
`@@ -328,7 +328,7 @@ await _destinationResource.CopyFromStreamAsync(`
`328`	`328`	`}`
`329`	`329`
`330`	`330`	`// Report bytes written before completion`
`331`		`- ReportBytesWritten(blockSize);`
	`331`	`+ await ReportBytesWrittenAsync(blockSize).ConfigureAwait(false);`
`332`	`332`
`333`	`333`	`// Set completion status to completed`
`334`	`334`	`await OnTransferStateChangedAsync(DataTransferState.Completed).ConfigureAwait(false);`
`@@ -360,7 +360,7 @@ await _destinationResource.CopyFromStreamAsync(`
`360`	`360`	`cancellationToken: _cancellationToken).ConfigureAwait(false);`
`361`	`361`	`}`
`362`	`362`
`363`		`- ReportBytesWritten(blockSize);`
	`363`	`+ await ReportBytesWrittenAsync(blockSize).ConfigureAwait(false);`
`364`	`364`	`}`
`365`	`365`	`}`
`366`	`366`
`@@ -372,7 +372,7 @@ internal static CommitChunkHandler.Behaviors GetBlockListCommitHandlerBehaviors(`
`372`	`372`	`{`
`373`	`373`	`QueuePutBlockTask = jobPart.QueueStageBlockRequest,`
`374`	`374`	`QueueCommitBlockTask = jobPart.QueueCompleteTransferAsync,`
`375`		`- ReportProgressInBytes = jobPart.ReportBytesWritten,`
	`375`	`+ ReportProgressInBytes = jobPart.ReportBytesWrittenAsync,`
`376`	`376`	`InvokeFailedHandler = jobPart.InvokeFailedArgAsync,`
`377`	`377`	`};`
`378`	`378`	`}`