[Storage] [DataMovement] Fix Pause and Resume on Cleaning up unfinished transfers during a Pause (Azure#47521)

* WIP

* Fixed bug where delete was using an already invoked cancellation token, failing the cancellation

* Cleanup

* Update tests

* Change finally block to catch block

Author: amnguye

sdk/storage/Azure.Storage.DataMovement/src/JobPartInternal.cs

@@ -441,7 +441,6 @@ await PartTransferStatusEventHandler.RaiseAsync(
                 {
                     await TriggerCancellationAsync().ConfigureAwait(false);
                 }
-                await CheckAndUpdateCancellationStateAsync().ConfigureAwait(false);
             }
             catch (Exception cancellationException)
             {
@@ -460,6 +459,31 @@ await TransferFailedEventHandler.RaiseAsync(
                     ClientDiagnostics)
                     .ConfigureAwait(false);
             }
+
+            // Whether or not we were able to trigger the cancellation and successfully clean up
+            // we should always call CheckAndUpdateCancellationStateAsync to correctly make
+            // sure the job part is in the correct state.
+            try
+            {
+                await CheckAndUpdateCancellationStateAsync().ConfigureAwait(false);
+            }
+            catch (Exception checkUpdateException)
+            {
+                // If an exception is thrown while trying to clean up,
+                // raise the failed event and prevent the transfer from hanging
+                await TransferFailedEventHandler.RaiseAsync(
+                    new TransferItemFailedEventArgs(
+                        _dataTransfer.Id,
+                        _sourceResource,
+                        _destinationResource,
+                        checkUpdateException,
+                        false,
+                        _cancellationToken),
+                    nameof(JobPartInternal),
+                    nameof(TransferFailedEventHandler),
+                    ClientDiagnostics)
+                    .ConfigureAwait(false);
+            }
         }
 
         /// <summary>
@@ -472,12 +496,25 @@ public async virtual Task CleanupAbortedJobPartAsync()
             // If the failure occurred due to the file already existing or authentication,
             // and overwrite wasn't enabled, don't delete the existing file. We can remove
             // the unfinished file for other error types.
-            if (JobPartFailureType.Other == _failureType)
+            // If a Pause was called, we can remove the unfinished file.
+            if (JobPartFailureType.Other == _failureType || DataTransferState.Pausing == JobPartStatus.State)
             {
                 // If the job part is paused or ended with failures
                 // delete the destination resource because it could be unfinished or corrupted
                 // If we resume we would have to start from the beginning anyways.
-                await _destinationResource.DeleteIfExistsAsync(_cancellationToken).ConfigureAwait(false);
+                try
+                {
+                    // We can't pass the cancellation token
+                    // here due to the fact that the job's cancellation token has already been called.
+                    // We are cleaning up / deleting optimistically, which means that if
+                    // the clean / delete attempt fails, then we continue on. The failure may be due
+                    // to the overall reason why the job was cancelled in the first place.
+                    await _destinationResource.DeleteIfExistsAsync().ConfigureAwait(false);
+                }
+                catch
+                {
+                    // We are cleaning up / deleting optimistically, move on if it fails.
+                }
             }
         }
 

sdk/storage/Azure.Storage.DataMovement/tests/TransferManagerTests.cs

@@ -532,11 +532,11 @@ public async Task TransferFailAtPartProcess(
         Assert.That(transfer.TransferStatus.HasFailedItems);
         Assert.That(failures, Is.Not.Empty);
 
-        Assert.That(capturedTransferStatuses.Count, Is.EqualTo(3)); // TODO should be 4
+        Assert.That(capturedTransferStatuses.Count, Is.EqualTo(4));
         Assert.That(capturedTransferStatuses[0].State, Is.EqualTo(DataTransferState.InProgress));
         Assert.That(capturedTransferStatuses[1].State, Is.EqualTo(DataTransferState.InProgress));
         Assert.That(capturedTransferStatuses[2].State, Is.EqualTo(DataTransferState.Stopping));
-        //Assert.That(capturedTransferStatuses[3].IsCompletedWithFailedItems); // TODO this should exist
+        Assert.That(capturedTransferStatuses[3].IsCompletedWithFailedItems);
     }
 
     [Test]