fix(storage): Ensure progress from Amplify.Storage.uploadFile completes

Author: jcjimenez

Amplify/Core/Support/ChildTask.swift

@@ -11,11 +11,11 @@ import Foundation
 /// Child Task is cancelled it will also cancel the parent.
 actor ChildTask<InProcess, Success, Failure: Error>: BufferingSequence {
     typealias Element = InProcess
-    let parent: Cancellable
-    var inProcessChannel: AmplifyAsyncSequence<InProcess>? = nil
-    var valueContinuations: [CheckedContinuation<Success, Error>] = []
-    var storedResult: Result<Success, Failure>? = nil
-    var isCancelled = false
+    private let parent: Cancellable
+    private var inProcessChannel: AmplifyAsyncSequence<InProcess>? = nil
+    private var valueContinuations: [CheckedContinuation<Success, Error>] = []
+    private var storedResult: Result<Success, Failure>? = nil
+    private var isCancelled = false
 
     var inProcess: AmplifyAsyncSequence<InProcess> {
         let channel: AmplifyAsyncSequence<InProcess>
@@ -75,9 +75,11 @@ actor ChildTask<InProcess, Success, Failure: Error>: BufferingSequence {
     func finish(_ result: Result<Success, Failure>) {
         if !valueContinuations.isEmpty {
             send(result)
-        } else {
-            // store result for when the value property is used
-            self.storedResult = result
+        }
+        // store result for when the value property is used
+        self.storedResult = result
+        if let channel = inProcessChannel {
+            channel.finish()
         }
     }
 

AmplifyPlugins/Storage/Tests/StorageHostApp/AWSS3StoragePluginIntegrationTests/AWSS3StoragePluginProgressTests.swift

@@ -24,6 +24,9 @@ import Amplify
 /// of ~0-1MB. After ~1 MB, I start getting notified more frequently.
 class AWSS3StoragePluginProgressTests: AWSS3StoragePluginTestBase {
 
+    /// - Given: An upload task
+    /// - When: A subscription to its progress `resultPublisher` is established **during** upload
+    /// - Then: Its publisher emits progress values and completes alongside the upload
     func testUploadProgressViaPublisher() async throws {
         var cancellables = Set<AnyCancellable>()
 
@@ -56,6 +59,9 @@ class AWSS3StoragePluginProgressTests: AWSS3StoragePluginTestBase {
         await remove(key: key)
     }
 
+    /// - Given: An upload task
+    /// - When: A subscription to its progress `resultPublisher` is established **after** completion
+    /// - Then: Its publisher completes immediately
     func testPublisherDeliveryAfterUploadCompletes() async throws {
         var cancellables = Set<AnyCancellable>()
 
@@ -94,6 +100,44 @@ class AWSS3StoragePluginProgressTests: AWSS3StoragePluginTestBase {
         // Remove the key
         await remove(key: key)
     }
+
+    /// - Given: An upload task
+    /// - When: Its progress AsyncSequence is iterated while the upload is **in progress**
+    /// - Then: Its iteration completes when the upload is done
+    func testUploadAndMonitorProgressUntilCompletion() async throws {
+        let key = UUID().uuidString
+        let task = Amplify.Storage.uploadData(
+            key: key,
+            data: .testDataOfSize(.bytes(64*1024))
+        )
+        let progress = await task.progress
+        var progressReports: [Progress] = []
+        for await current in progress {
+            XCTAssertGreaterThan(current.fractionCompleted, 0)
+            XCTAssertLessThanOrEqual(current.fractionCompleted, 1)
+            progressReports.append(current)
+        }
+        XCTAssertGreaterThan(progressReports.count, 0)
+        _ = try await task.value
+        await remove(key: key)
+    }
+
+    /// - Given: An upload task
+    /// - When: Its progress AsyncSequence is iterated **after** completion
+    /// - Then: Its iteration completes without providing a value
+    func testUploadAndMonitorProgressAfterCompletion() async throws {
+        let key = UUID().uuidString
+        let task = Amplify.Storage.uploadData(
+            key: key,
+            data: .testDataOfSize(.bytes(256))
+        )
+        _ = try await task.value
+        let progress = await task.progress
+        for await current in progress {
+            XCTFail("Not expecting a current progress value but got: \(current)")
+        }
+        await remove(key: key)
+    }
 }
 
 private extension Data {

AmplifyTests/CoreTests/AmplifyTaskTests.swift

@@ -66,39 +66,33 @@ class AmplifyTaskTests: XCTestCase {
     }
 #endif
 
+    /// Given: A operation with discrete steps (10)
+    /// When: Its progress sequence is iterated
+    /// Then: It reports each step up to a `fractionCompleted` value that represents 100% completion
     func testLongOperation() async throws {
-        var success = false
-        var output: String? = nil
-        var thrown: Error? = nil
-
         let request = LongOperationRequest(steps: 10, delay: 0.01)
         let longTask = await runLongOperation(request: request)
 
-        Task {
-            var progressCount = 0
-            var lastProgress: Double = 0
+        var progressCount = 0
+        var lastProgress: Double = 0
+        await longTask.progress.forEach { progress in
+            lastProgress = progress.fractionCompleted
+            progressCount += 1
+        }
 
-            await longTask.progress.forEach { progress in
-                lastProgress = progress.fractionCompleted
-                progressCount += 1
-            }
+        // The first progress report happens on `fractionCompleted` 0.0
+        XCTAssertEqual(progressCount, 11)
 
-            XCTAssertEqual(progressCount, 11)
-            XCTAssertEqual(lastProgress, 100)
-        }
+        // Note that `fractionComleted` is calculated by dividing
+        // `completedUnitCount` by `totalUnitCount`. See:
+        //https://developer.apple.com/documentation/foundation/progress/1408579-fractioncompleted
+        XCTAssertEqual(lastProgress, 1.0)
 
-        do {
-            let value = try await longTask.value
-            output = value.id
-            success = true
-        } catch {
-            thrown = error
-        }
+        let value = try await longTask.value
+        let output = value.id
 
-        XCTAssertTrue(success)
         XCTAssertNotNil(output)
         XCTAssertFalse(output.isEmpty)
-        XCTAssertNil(thrown)
     }
 
 #if canImport(Combine)

AmplifyTests/CoreTests/ChildTaskTests.swift

@@ -18,12 +18,16 @@ class ChildTaskTests: XCTestCase {
         }
     }
 
+    /// Given: A ChildTask instance associated to a fast operation
+    /// When: Multiple `Task` instances are created to wait for its results
+    /// Then: All `Task` instances receive the exact-same `value`
     func testFastOperationWithMultipleAwaits() async throws {
         let input = [1, 2, 3]
         let request = FastOperationRequest(numbers: input)
         let queue = OperationQueue()
         let operation = FastOperation(request: request)
         let childTask: ChildTask<Void, FastOperation.Success, FastOperation.Failure> = ChildTask(parent: operation)
+        let progressSequence = await childTask.inProcess
         let token = operation.subscribe { result in
             Task {
                 await childTask.finish(result)
@@ -54,11 +58,17 @@ class ChildTaskTests: XCTestCase {
         XCTAssertEqual(output1, expectedOutput)
         XCTAssertEqual(output2, expectedOutput)
         XCTAssertEqual(output3, expectedOutput)
+
+        // Ensure the channel's AsyncSequence does not block after completion
+        for await _ in progressSequence {
+            XCTFail("Unexpected channel iteration since task has completed.")
+        }
     }
 
     func testChildTaskResultCancelled() async throws {
         let worker = Worker()
         let childTask: ChildTask<Void, String, Never> = ChildTask(parent: worker)
+        let progressSequence = await childTask.inProcess
         let cancelExp = expectation(description: "cancel")
         cancelExp.isInverted = true
 
@@ -77,11 +87,17 @@ class ChildTaskTests: XCTestCase {
 
         await waitForExpectations(timeout: 0.01)
         task.cancel()
+
+        // Ensure the channel's AsyncSequence does not block after completion
+        for await _ in progressSequence {
+            XCTFail("Unexpected channel iteration since task was cancelled.")
+        }
     }
 
     func testChildTaskResultAlreadyCancelled() async throws {
         let worker = Worker()
         let childTask: ChildTask<Void, String, Never> = ChildTask(parent: worker)
+        let progressSequence = await childTask.inProcess
         let cancelExp = expectation(description: "cancel")
         cancelExp.isInverted = true
 
@@ -100,6 +116,11 @@ class ChildTaskTests: XCTestCase {
         }
 
         await waitForExpectations(timeout: 0.01)
+
+        // Ensure the channel's AsyncSequence does not block after completion
+        for await _ in progressSequence {
+            XCTFail("Unexpected channel iteration since task was cancelled.")
+        }
     }
 
 }