[Storage] Directory Pause and Resume Tests with directory resume fixes (Azure#35640)

* WIP

* Fix to incorrect download failed message tests; Update to iterate through paused job parts for directories

* WIP - tests pass for directory pause, except pause then resume

* Pause then resume directory upload/download/copy

* Correct test name BlockBlobClientTests/SyncUploadFromUriAsync_OverwriteSourceBlobProperties

* Disable pause and resume tests to prevent flaking

* Playback pause and resume tests

* Smaller size to make smaller test recordings

* Change uploaded blob to forward slash in the tests and enable childstorageresource tests

* Disable pause and resume tests

Author: amnguye

sdk/storage/Azure.Storage.Blobs/tests/BlockBlobClientTests.cs

@@ -2788,7 +2788,7 @@ await TestHelper.AssertExpectedExceptionAsync<RequestFailedException>(
 
         [RecordedTest]
         [ServiceVersion(Min = BlobClientOptions.ServiceVersion.V2020_04_08)]
-        public async Task SyncUploadFromUriAsync_OverwiteSourceBlobProperties()
+        public async Task SyncUploadFromUriAsync_OverwriteSourceBlobProperties()
         {
             // Arrange
             var constants = TestConstants.Create(this);

sdk/storage/Azure.Storage.Blobs/tests/SessionRecords/BlockBlobClientTests/SyncUploadFromUriAsync_OverwiteSourceBlobProperties.json renamed to sdk/storage/Azure.Storage.Blobs/tests/SessionRecords/BlockBlobClientTests/SyncUploadFromUriAsync_OverwriteSourceBlobProperties.json

@@ -83,19 +83,19 @@ public override StorageResource GetChildStorageResource(string path)
             if (_options?.BlobType == BlobType.Append)
             {
                 return new AppendBlobStorageResource(
-                    _blobContainerClient.GetAppendBlobClient(System.IO.Path.Combine(_directoryPrefix, path)),
+                    _blobContainerClient.GetAppendBlobClient(string.Join("/", _directoryPrefix, path)),
                     _options?.ToAppendBlobStorageResourceOptions());
             }
             else if (_options?.BlobType == BlobType.Page)
             {
                 return new PageBlobStorageResource(
-                    _blobContainerClient.GetPageBlobClient(System.IO.Path.Combine(_directoryPrefix, path)),
+                    _blobContainerClient.GetPageBlobClient(string.Join("/", _directoryPrefix, path)),
                     _options?.ToPageBlobStorageResourceOptions());
             }
             else // BlobType.Block or null
             {
                 return new BlockBlobStorageResource(
-                    _blobContainerClient.GetBlockBlobClient(System.IO.Path.Combine(_directoryPrefix, path)),
+                    _blobContainerClient.GetBlockBlobClient(string.Join("/", _directoryPrefix, path)),
                     _options?.ToBlockBlobStorageResourceOptions());
             }
         }

sdk/storage/Azure.Storage.Blobs/tests/SessionRecords/BlockBlobClientTests/SyncUploadFromUriAsync_OverwiteSourceBlobPropertiesAsync.json renamed to sdk/storage/Azure.Storage.Blobs/tests/SessionRecords/BlockBlobClientTests/SyncUploadFromUriAsync_OverwriteSourceBlobPropertiesAsync.json

@@ -103,6 +103,12 @@ internal abstract class JobPartInternal
         /// </summary>
         internal long? Length;
 
+        /// <summary>
+        /// Defines whether or not this was the final part in the list call. This would determine
+        /// whether or not we needed to keep listing in the job.
+        /// </summary>
+        public bool IsFinalPart { get; internal set; }
+
         /// <summary>
         /// If the transfer status of the job changes then the event will get added to this handler.
         /// </summary>
@@ -151,6 +157,7 @@ internal JobPartInternal(
             StorageResourceCreateMode createMode,
             TransferCheckpointer checkpointer,
             ArrayPool<byte> arrayPool,
+            bool isFinalPart,
             SyncAsyncEventHandler<TransferStatusEventArgs> jobPartEventHandler,
             SyncAsyncEventHandler<TransferStatusEventArgs> statusEventHandler,
             SyncAsyncEventHandler<TransferFailedEventArgs> failedEventHandler,
@@ -171,6 +178,7 @@ internal JobPartInternal(
             _checkpointer = checkpointer;
             _cancellationToken = cancellationToken;
             _arrayPool = arrayPool;
+            IsFinalPart = isFinalPart;
             PartTransferStatusEventHandler = jobPartEventHandler;
             TransferStatusEventHandler = statusEventHandler;
             TransferFailedEventHandler = failedEventHandler;
@@ -208,7 +216,7 @@ public void SetQueueChunkDelegate(QueueChunkDelegate chunkDelegate)
         /// when we're looking to stop/pause the job part.
         /// </summary>
         /// <returns></returns>
-        public async Task QueueChunkToChannelAsync(Task chunkTask)
+        public async Task QueueChunkToChannelAsync(Func<Task> chunkTask)
         {
             // Attach TaskCompletionSource
             TaskCompletionSource<bool> chunkCompleted = new TaskCompletionSource<bool>(
@@ -220,7 +228,7 @@ public async Task QueueChunkToChannelAsync(Task chunkTask)
             await QueueChunk(
                 async () =>
                 {
-                    await chunkTask.ConfigureAwait(false);
+                    await Task.Run(chunkTask).ConfigureAwait(false);
                     chunkCompleted.SetResult(true);
                     await CheckAndUpdateCancellationStatusAsync().ConfigureAwait(false);
                 }).ConfigureAwait(false);
@@ -243,7 +251,7 @@ await QueueChunk(
         /// and any chunks is still processing to be cancelled is will be set to <see cref="StorageTransferStatus.CancellationInProgress"/>
         /// until the chunks finish then it will be set to <see cref="StorageTransferStatus.CompletedWithFailedTransfers"/>.
         /// </summary>
-        /// <returns></returns>
+        /// <returns>The task to wait until the cancellation has been triggered.</returns>
         internal async Task TriggerCancellationAsync()
         {
             if (!_cancellationToken.IsCancellationRequested)
@@ -389,9 +397,17 @@ public async virtual Task CleanupAbortedJobPartAsync()
             }
         }
 
-        public async virtual Task AddJobPartToCheckpointer(int chunksTotal)
+        /// <summary>
+        /// Serializes the respective job part and adds it to the checkpointer.
+        /// </summary>
+        /// <param name="chunksTotal">Number of chunks in the job part.</param>
+        /// <param name="isFinalPart">Defines if this part is the last job part of the job.</param>
+        /// <returns></returns>
+        public async virtual Task AddJobPartToCheckpointerAsync(int chunksTotal, bool isFinalPart)
         {
-            JobPartPlanHeader header = this.ToJobPartPlanHeader(StorageTransferStatus.InProgress);
+            JobPartPlanHeader header = this.ToJobPartPlanHeader(
+                jobStatus: StorageTransferStatus.InProgress,
+                isFinalPart: isFinalPart);
             using (Stream stream = new MemoryStream())
             {
                 header.Serialize(stream);

sdk/storage/Azure.Storage.DataMovement.Blobs/src/BlobDirectoryStorageResourceContainer.cs

@@ -28,7 +28,8 @@ public static async Task<StreamToUriJobPart> ToJobPartAsync(
                 jobPartStatus: jobPartStatus,
                 sourceResource: sourceResource,
                 destinationResource: destinationResource,
-                partPlanFileExists: true).ConfigureAwait(false);
+                partPlanFileExists: true,
+                isFinalPart: header.IsFinalPart).ConfigureAwait(false);
 
             // TODO: When enabling resume chunked upload Add each transfer to the CommitChunkHandler
             return jobPart;
@@ -44,14 +45,15 @@ public static async Task<ServiceToServiceJobPart> ToJobPartAsync(
             JobPartPlanHeader header = JobPartPlanHeader.Deserialize(planFileStream);;
 
             // Apply credentials to the saved transfer job path
-            StorageTransferStatus jobPartStatus = (StorageTransferStatus) header.AtomicJobStatus;
+            StorageTransferStatus jobPartStatus = header.AtomicJobStatus;
             ServiceToServiceJobPart jobPart = await ServiceToServiceJobPart.CreateJobPartAsync(
                 job: baseJob,
                 partNumber: Convert.ToInt32(header.PartNumber),
                 jobPartStatus: jobPartStatus,
                 sourceResource: sourceResource,
                 destinationResource: destinationResource,
-                partPlanFileExists: true).ConfigureAwait(false);
+                partPlanFileExists: true,
+                isFinalPart: header.IsFinalPart).ConfigureAwait(false);
 
             // TODO: When enabling resume chunked upload Add each transfer to the CommitChunkHandler
             return jobPart;
@@ -67,14 +69,15 @@ public static async Task<UriToStreamJobPart> ToJobPartAsync(
             JobPartPlanHeader header = JobPartPlanHeader.Deserialize(planFileStream);;
 
             // Apply credentials to the saved transfer job path
-            StorageTransferStatus jobPartStatus = (StorageTransferStatus)header.AtomicJobStatus;
+            StorageTransferStatus jobPartStatus = header.AtomicJobStatus;
             UriToStreamJobPart jobPart = await UriToStreamJobPart.CreateJobPartAsync(
                 job: baseJob,
                 partNumber: Convert.ToInt32(header.PartNumber),
                 jobPartStatus: jobPartStatus,
                 sourceResource: sourceResource,
                 destinationResource: destinationResource,
-                partPlanFileExists: true).ConfigureAwait(false);
+                partPlanFileExists: true,
+                isFinalPart: header.IsFinalPart).ConfigureAwait(false);
 
             // TODO: When enabling resume chunked upload Add each transfer to the CommitChunkHandler
             return jobPart;
@@ -87,19 +90,22 @@ public static async Task<StreamToUriJobPart> ToJobPartAsync(
             StorageResourceContainer destinationResource)
         {
             // Convert stream to job plan header
-            JobPartPlanHeader header = JobPartPlanHeader.Deserialize(planFileStream);;
+            JobPartPlanHeader header = JobPartPlanHeader.Deserialize(planFileStream);
 
             // Apply credentials to the saved transfer job path
             string childSourcePath = header.SourcePath;
-            string childDestinationPath = header.SourcePath;
+            string childSourceName = childSourcePath.Substring(sourceResource.Path.Length + 1);
+            string childDestinationPath = header.DestinationPath;
+            string childDestinationName = childDestinationPath.Substring(destinationResource.Uri.AbsoluteUri.Length + 1);
             StorageTransferStatus jobPartStatus = header.AtomicJobStatus;
             StreamToUriJobPart jobPart = await StreamToUriJobPart.CreateJobPartAsync(
                 job: baseJob,
                 partNumber: Convert.ToInt32(header.PartNumber),
                 jobPartStatus: jobPartStatus,
-                sourceResource: sourceResource.GetChildStorageResource(childSourcePath.Substring(sourceResource.Path.Length)),
-                destinationResource: destinationResource.GetChildStorageResource(childDestinationPath.Substring(destinationResource.Path.Length)),
-                partPlanFileExists: true).ConfigureAwait(false);
+                sourceResource: sourceResource.GetChildStorageResource(childSourceName),
+                destinationResource: destinationResource.GetChildStorageResource(childDestinationName),
+                partPlanFileExists: true,
+                isFinalPart: header.IsFinalPart).ConfigureAwait(false);
 
             // TODO: When enabling resume chunked upload Add each transfer to the CommitChunkHandler
             return jobPart;
@@ -116,15 +122,16 @@ public static async Task<ServiceToServiceJobPart> ToJobPartAsync(
 
             // Apply credentials to the saved transfer job path
             string childSourcePath = header.SourcePath;
-            string childDestinationPath = header.SourcePath;
+            string childDestinationPath = header.DestinationPath;
             StorageTransferStatus jobPartStatus = header.AtomicJobStatus;
             ServiceToServiceJobPart jobPart = await ServiceToServiceJobPart.CreateJobPartAsync(
                 job: baseJob,
                 partNumber: Convert.ToInt32(header.PartNumber),
                 jobPartStatus: jobPartStatus,
-                sourceResource: sourceResource.GetChildStorageResource(childSourcePath.Substring(sourceResource.Path.Length)),
-                destinationResource: destinationResource.GetChildStorageResource(childDestinationPath.Substring(destinationResource.Path.Length)),
-                partPlanFileExists: true).ConfigureAwait(false);
+                sourceResource: sourceResource.GetChildStorageResource(childSourcePath.Substring(sourceResource.Uri.AbsoluteUri.Length + 1)),
+                destinationResource: destinationResource.GetChildStorageResource(childDestinationPath.Substring(destinationResource.Uri.AbsoluteUri.Length + 1)),
+                partPlanFileExists: true,
+                isFinalPart: header.IsFinalPart).ConfigureAwait(false);
 
             // TODO: When enabling resume chunked upload Add each transfer to the CommitChunkHandler
             return jobPart;
@@ -141,15 +148,18 @@ public static async Task<UriToStreamJobPart> ToJobPartAsync(
 
             // Apply credentials to the saved transfer job path
             string childSourcePath = header.SourcePath;
-            string childDestinationPath = header.SourcePath;
+            string childSourceName = childSourcePath.Substring(sourceResource.Uri.AbsoluteUri.Length + 1);
+            string childDestinationPath = header.DestinationPath;
+            string childDestinationName = childDestinationPath.Substring(destinationResource.Path.Length + 1);
             StorageTransferStatus jobPartStatus = header.AtomicJobStatus;
             UriToStreamJobPart jobPart = await UriToStreamJobPart.CreateJobPartAsync(
                 job: baseJob,
                 partNumber: Convert.ToInt32(header.PartNumber),
                 jobPartStatus: jobPartStatus,
-                sourceResource: sourceResource.GetChildStorageResource(childSourcePath.Substring(sourceResource.Path.Length)),
-                destinationResource: destinationResource.GetChildStorageResource(childDestinationPath.Substring(destinationResource.Path.Length)),
-                partPlanFileExists: true).ConfigureAwait(false);
+                sourceResource: sourceResource.GetChildStorageResource(childSourceName),
+                destinationResource: destinationResource.GetChildStorageResource(childDestinationName),
+                partPlanFileExists: true,
+                isFinalPart: header.IsFinalPart).ConfigureAwait(false);
 
             // TODO: When enabling resume chunked upload Add each transfer to the CommitChunkHandler
             return jobPart;
@@ -158,7 +168,9 @@ public static async Task<UriToStreamJobPart> ToJobPartAsync(
         /// <summary>
         /// Translate the initial job part header to a job plan format file
         /// </summary>
-        internal static JobPartPlanHeader ToJobPartPlanHeader(this JobPartInternal jobPart, StorageTransferStatus jobStatus)
+        internal static JobPartPlanHeader ToJobPartPlanHeader(this JobPartInternal jobPart,
+            StorageTransferStatus jobStatus,
+            bool isFinalPart)
         {
             JobPartPlanDestinationBlob dstBlobData = new JobPartPlanDestinationBlob(
                 blobType: JobPlanBlobType.Detect, // TODO: update when supported
@@ -186,11 +198,15 @@ internal static JobPartPlanHeader ToJobPartPlanHeader(this JobPartInternal jobPa
                 startTime: DateTimeOffset.UtcNow, // TODO: update to job start time
                 transferId: jobPart._dataTransfer.Id,
                 partNumber: (uint)jobPart.PartNumber,
-                sourcePath: jobPart._sourceResource.Path,
+                sourcePath: jobPart._sourceResource.CanProduceUri == ProduceUriType.ProducesUri ?
+                            jobPart._sourceResource.Uri.AbsoluteUri :
+                            jobPart._sourceResource.Path,
                 sourceExtraQuery: "", // TODO: convert options to string
-                destinationPath: jobPart._destinationResource.Path,
+                destinationPath: jobPart._destinationResource.CanProduceUri == ProduceUriType.ProducesUri ?
+                            jobPart._destinationResource.Uri.AbsoluteUri :
+                            jobPart._destinationResource.Path,
                 destinationExtraQuery: "", // TODO: convert options to string
-                isFinalPart: false, // TODO: change but we might remove this param
+                isFinalPart: isFinalPart,
                 forceWrite: jobPart._createMode == StorageResourceCreateMode.Overwrite, // TODO: change to enum value
                 forceIfReadOnly: false, // TODO: revisit for Azure Files
                 autoDecompress: false, // TODO: revisit if we want to support this feature