Merge branch 'transfer-manager' into add-documentation

Author: lukachad

lib/lib-storage/src/s3-transfer-manager/S3TransferManager.e2e.spec.ts

@@ -61,7 +61,7 @@ describe(S3TransferManager.name, () => {
         it(`should download an object of size ${size} with mode ${mode}`, async () => {
           const totalSizeMB = size * 1024 * 1024;
           const Body = data(totalSizeMB);
-          const Key = `${mode}-size`;
+          const Key = `${mode}-${size}`;
 
           await new Upload({
             client,
@@ -199,7 +199,7 @@ describe(S3TransferManager.name, () => {
             internalEventHandler.afterInitialGetObject = async () => {};
           };
 
-          await tm.download(
+          const downloadResponse = await tm.download(
             { Bucket, Key },
             {
               eventListeners: {
@@ -213,6 +213,7 @@ describe(S3TransferManager.name, () => {
               },
             }
           );
+          await downloadResponse.Body?.transformToByteArray();
           expect.fail("Download should have failed due to ETag mismatch");
         } catch (error) {
           expect(transferFailed).toBe(true);

lib/lib-storage/src/s3-transfer-manager/S3TransferManager.ts

@@ -431,9 +431,6 @@ export class S3TransferManager implements IS3TransferManager {
 
         let partCount = 1;
         if (initialPart.PartsCount! > 1) {
-          const concurrentRequests = [];
-          const concurrentRequestInputs = [];
-
           for (let part = 2; part <= initialPart.PartsCount!; part++) {
             this.checkAborted(transferOptions);
             const getObjectRequest = {
@@ -453,25 +450,16 @@ export class S3TransferManager implements IS3TransferManager {
                   };
                 }
                 return response.Body!;
+              })
+              .catch((error) => {
+                this.dispatchTransferFailedEvent(getObjectRequest, totalSize, error as Error);
+                throw error;
               });
-
-            concurrentRequests.push(getObject);
-            concurrentRequestInputs.push(getObjectRequest);
+            streams.push(getObject);
+            requests.push(getObjectRequest);
             partCount++;
           }
 
-          try {
-            // Add promise streams to streams array ONLY if all are resolved
-            const responses = await Promise.all(concurrentRequests);
-            for (let i = 0; i < responses.length; i++) {
-              streams.push(Promise.resolve(responses[i]));
-              requests.push(concurrentRequestInputs[i]);
-            }
-          } catch (error) {
-            this.dispatchTransferFailedEvent(request, totalSize, error as Error);
-            throw error;
-          }
-
           if (partCount !== initialPart.PartsCount) {
             throw new Error(
               `The number of parts downloaded (${partCount}) does not match the expected number (${initialPart.PartsCount})`
@@ -600,9 +588,6 @@ export class S3TransferManager implements IS3TransferManager {
     remainingLength = totalSize ? Math.min(right - left + 1, Math.max(0, totalSize - left)) : 0;
     let actualRequestCount = 1;
 
-    const concurrentRequests = [];
-    const concurrentRequestInputs = [];
-
     while (remainingLength > 0) {
       this.checkAborted(transferOptions);
 
@@ -624,31 +609,21 @@ export class S3TransferManager implements IS3TransferManager {
             };
           }
           return response.Body!;
+        })
+        .catch((error) => {
+          this.dispatchTransferFailedEvent(getObjectRequest, totalSize, error);
+          throw error;
         });
 
-      concurrentRequests.push(getObject);
-      concurrentRequestInputs.push(getObjectRequest);
+      streams.push(getObject);
+      requests.push(getObjectRequest);
       actualRequestCount++;
 
       left = right + 1;
       right = Math.min(left + S3TransferManager.MIN_PART_SIZE - 1, maxRange);
       remainingLength = totalSize ? Math.min(right - left + 1, Math.max(0, totalSize - left)) : 0;
     }
 
-    if (concurrentRequests.length > 0) {
-      try {
-        // Add promise streams to streams array ONLY if all are resolved
-        const responses = await Promise.all(concurrentRequests);
-        for (let i = 0; i < responses.length; i++) {
-          streams.push(Promise.resolve(responses[i]));
-          requests.push(concurrentRequestInputs[i]);
-        }
-      } catch (error) {
-        this.dispatchTransferFailedEvent(request, totalSize, error as Error);
-        throw error;
-      }
-    }
-
     if (expectedRequestCount !== actualRequestCount) {
       throw new Error(
         `The number of ranged GET requests sent (${actualRequestCount}) does not match the expected number (${expectedRequestCount})`

lib/lib-storage/src/s3-transfer-manager/join-streams.browser.ts

@@ -34,15 +34,22 @@ export async function joinStreams(
  * @internal
  */
 export async function* iterateStreams(
-  streams: Promise<StreamingBlobPayloadOutputTypes>[],
+  promises: Promise<StreamingBlobPayloadOutputTypes>[],
   eventListeners?: JoinStreamIterationEvents
 ): AsyncIterable<StreamingBlobPayloadOutputTypes, void, void> {
   let bytesTransferred = 0;
   let index = 0;
-  for (const streamPromise of streams) {
-    const stream = await streamPromise;
+  for (const streamPromise of promises) {
+    let stream: Awaited<(typeof promises)[0]>;
+    try {
+      stream = await streamPromise;
+    } catch (e) {
+      await destroy(promises);
+      eventListeners?.onFailure?.(e, index);
+      throw e;
+    }
+
     if (isReadableStream(stream)) {
-      // TODO: May need to acquire reader before reaching the stream
       const reader = stream.getReader();
       try {
         while (true) {
@@ -66,3 +73,20 @@ export async function* iterateStreams(
   }
   eventListeners?.onCompletion?.(bytesTransferred, index - 1);
 }
+
+/**
+ * @internal
+ */
+async function destroy(promises: Promise<StreamingBlobPayloadOutputTypes>[]): Promise<void> {
+  await Promise.all(
+    promises.map(async (streamPromise) => {
+      return streamPromise
+        .then((stream) => {
+          if (isReadableStream(stream)) {
+            return stream.cancel();
+          }
+        })
+        .catch((e: unknown) => {});
+    })
+  );
+}

lib/lib-storage/src/s3-transfer-manager/join-streams.ts

@@ -25,6 +25,9 @@ export async function joinStreams(
     });
     return sdkStreamMixin(newReadableStream);
   } else {
+    // TODO: The following line is a temp fix to handle error thrown in async iterable.
+    // We should find a better solution to improve performance.
+    await Promise.all(streams);
     return sdkStreamMixin(Readable.from(iterateStreams(streams, eventListeners)));
   }
 }
@@ -35,15 +38,22 @@ export async function joinStreams(
  * @internal
  */
 export async function* iterateStreams(
-  streams: Promise<StreamingBlobPayloadOutputTypes>[],
+  promises: Promise<StreamingBlobPayloadOutputTypes>[],
   eventListeners?: JoinStreamIterationEvents
 ): AsyncIterable<StreamingBlobPayloadOutputTypes, void, void> {
   let bytesTransferred = 0;
   let index = 0;
-  for (const streamPromise of streams) {
-    const stream = await streamPromise;
+  for (const streamPromise of promises) {
+    let stream: Awaited<(typeof promises)[0]>;
+    try {
+      stream = await streamPromise;
+    } catch (e) {
+      await destroy(promises);
+      eventListeners?.onFailure?.(e, index);
+      throw e;
+    }
+
     if (isReadableStream(stream)) {
-      // TODO: May need to acquire reader before reaching the stream
       const reader = stream.getReader();
       try {
         while (true) {
@@ -74,3 +84,23 @@ export async function* iterateStreams(
   }
   eventListeners?.onCompletion?.(bytesTransferred, index - 1);
 }
+
+/**
+ * @internal
+ */
+async function destroy(promises: Promise<StreamingBlobPayloadOutputTypes>[]): Promise<void> {
+  await Promise.all(
+    promises.map(async (streamPromise) => {
+      return streamPromise
+        .then((stream) => {
+          if (stream instanceof Readable) {
+            stream.destroy();
+            return;
+          } else if (isReadableStream(stream)) {
+            return stream.cancel();
+          }
+        })
+        .catch((e: unknown) => {});
+    })
+  );
+}

yarn.lock

@@ -31118,17 +31118,6 @@ __metadata:
   languageName: node
   linkType: hard
 
-"aws-sdk-client-mock@npm:^4.1.0":
-  version: 4.1.0
-  resolution: "aws-sdk-client-mock@npm:4.1.0"
-  dependencies:
-    "@types/sinon": "npm:^17.0.3"
-    sinon: "npm:^18.0.1"
-    tslib: "npm:^2.1.0"
-  checksum: 10c0/045caad0cff0ffeb08e69849dcae51aac8999163c58d71220bf47a82c237aabab2abf92bf6bf3bd7666e6e8984513c628e01a89eafa46fb230201d6587bc01e9
-  languageName: node
-  linkType: hard
-
 "aws-sdk-js-v3@workspace:.":
   version: 0.0.0-use.local
   resolution: "aws-sdk-js-v3@workspace:."
@@ -31150,7 +31139,6 @@ __metadata:
     "@typescript-eslint/eslint-plugin": "npm:5.55.0"
     "@typescript-eslint/parser": "npm:5.55.0"
     async: "npm:3.2.4"
-    aws-sdk-client-mock: "npm:^4.1.0"
     concurrently: "npm:7.0.0"
     decomment: "npm:0.9.5"
     downlevel-dts: "npm:0.10.1"