fix(middleware-flexible-checksums): buffer stream chunks to minimum required size

Author: kuhe

packages/middleware-flexible-checksums/src/configuration.ts

@@ -66,4 +66,9 @@ export interface PreviouslyResolved {
    * Collects streams into buffers.
    */
   streamCollector: StreamCollector;
+
+  /**
+   * Minimum bytes from a stream to buffer into a chunk before passing to chunked encoding.
+   */
+  requestStreamBufferSize: number;
 }

packages/middleware-flexible-checksums/src/flexibleChecksumsMiddleware.ts

@@ -9,6 +9,7 @@ import {
   HandlerExecutionContext,
   MetadataBearer,
 } from "@smithy/types";
+import { createBufferedReadable } from "@smithy/util-stream";
 
 import { PreviouslyResolved } from "./configuration";
 import { ChecksumAlgorithm, DEFAULT_CHECKSUM_ALGORITHM, RequestChecksumCalculation } from "./constants";
@@ -119,13 +120,18 @@ export const flexibleChecksumsMiddleware =
       const checksumAlgorithmFn = selectChecksumAlgorithmFunction(checksumAlgorithm, config);
       if (isStreaming(requestBody)) {
         const { getAwsChunkedEncodingStream, bodyLengthChecker } = config;
-        updatedBody = getAwsChunkedEncodingStream(requestBody, {
-          base64Encoder,
-          bodyLengthChecker,
-          checksumLocationName,
-          checksumAlgorithmFn,
-          streamHasher,
-        });
+        updatedBody = getAwsChunkedEncodingStream(
+          config.requestStreamBufferSize
+            ? createBufferedReadable(requestBody, config.requestStreamBufferSize, context.logger)
+            : requestBody,
+          {
+            base64Encoder,
+            bodyLengthChecker,
+            checksumLocationName,
+            checksumAlgorithmFn,
+            streamHasher,
+          }
+        );
         updatedHeaders = {
           ...headers,
           "content-encoding": headers["content-encoding"]

packages/middleware-flexible-checksums/src/middleware-flexible-checksums.e2e.spec.ts

@@ -1,8 +1,10 @@
-import { S3 } from "@aws-sdk/client-s3";
+import { S3, UploadPartCommandOutput } from "@aws-sdk/client-s3";
+import { Upload } from "@aws-sdk/lib-storage";
+import { FetchHttpHandler } from "@smithy/fetch-http-handler";
 import type { HttpRequest, HttpResponse } from "@smithy/types";
-import { headStream } from "@smithy/util-stream";
+import { ChecksumStream, headStream } from "@smithy/util-stream";
 import { Readable } from "node:stream";
-import { beforeAll, describe, expect, test as it } from "vitest";
+import { beforeAll, describe, expect, test as it, vi } from "vitest";
 
 import { getIntegTestResources } from "../../../tests/e2e/get-integ-test-resources";
 
@@ -13,6 +15,26 @@ describe("S3 checksums", () => {
   let Key: string;
   let region: string;
   const expected = new Uint8Array([97, 98, 99, 100]);
+  const logger = {
+    debug: vi.fn(),
+    info: vi.fn(),
+    warn: vi.fn(),
+    error: vi.fn(),
+  };
+
+  function stream(size: number, chunkSize: number) {
+    async function* generate() {
+      while (size > 0) {
+        const z = Math.min(size, chunkSize);
+        yield "a".repeat(z);
+        size -= z;
+      }
+    }
+    return Readable.from(generate());
+  }
+  function webStream(size: number, chunkSize: number) {
+    return Readable.toWeb(stream(size, chunkSize)) as unknown as ReadableStream;
+  }
 
   beforeAll(async () => {
     const integTestResourcesEnv = await getIntegTestResources();
@@ -21,7 +43,7 @@ describe("S3 checksums", () => {
     region = process?.env?.AWS_SMOKE_TEST_REGION as string;
     Bucket = process?.env?.AWS_SMOKE_TEST_BUCKET as string;
 
-    s3 = new S3({ region });
+    s3 = new S3({ logger, region, requestStreamBufferSize: 8 * 1024 });
     s3_noChecksum = new S3({
       region,
       requestChecksumCalculation: "WHEN_REQUIRED",
@@ -38,7 +60,7 @@ describe("S3 checksums", () => {
           expect(reqHeader).toEqual("CRC32");
         }
         if (resHeader) {
-          expect(resHeader).toEqual("7YLNEQ==");
+          expect(resHeader.length).toBeGreaterThanOrEqual(8);
         }
         return r;
       },
@@ -53,9 +75,125 @@ describe("S3 checksums", () => {
   });
 
   it("an object should have checksum by default", async () => {
-    await s3.getObject({ Bucket, Key });
+    const get = await s3.getObject({ Bucket, Key });
+    expect(get.Body).toBeInstanceOf(ChecksumStream);
   });
 
+  describe("PUT operations", () => {
+    it("should assist user input streams by buffering to the minimum 8kb required by S3", async () => {
+      await s3.putObject({
+        Bucket,
+        Key: Key + "small-chunks",
+        Body: stream(24 * 1024, 8),
+        ContentLength: 24 * 1024,
+      });
+      expect(logger.warn).toHaveBeenCalledWith(
+        `@smithy/util-stream - stream chunk size 8 is below threshold of 8192, automatically buffering.`
+      );
+      const get = await s3.getObject({
+        Bucket,
+        Key: Key + "small-chunks",
+      });
+      expect((await get.Body?.transformToByteArray())?.byteLength).toEqual(24 * 1024);
+    });
+    it("should be able to write an object with a webstream body (using fetch handler without checksum)", async () => {
+      const handler = s3_noChecksum.config.requestHandler;
+      s3_noChecksum.config.requestHandler = new FetchHttpHandler();
+      await s3_noChecksum.putObject({
+        Bucket,
+        Key: Key + "small-chunks-webstream",
+        Body: webStream(24 * 1024, 512),
+        ContentLength: 24 * 1024,
+      });
+      s3_noChecksum.config.requestHandler = handler;
+      const get = await s3.getObject({
+        Bucket,
+        Key: Key + "small-chunks-webstream",
+      });
+      expect((await get.Body?.transformToByteArray())?.byteLength).toEqual(24 * 1024);
+    });
+    it("@aws-sdk/lib-storage Upload should allow webstreams to be used", async () => {
+      await new Upload({
+        client: s3,
+        params: {
+          Bucket,
+          Key: Key + "small-chunks-webstream-mpu",
+          Body: webStream(6 * 1024 * 1024, 512),
+        },
+      }).done();
+      const get = await s3.getObject({
+        Bucket,
+        Key: Key + "small-chunks-webstream-mpu",
+      });
+      expect((await get.Body?.transformToByteArray())?.byteLength).toEqual(6 * 1024 * 1024);
+    });
+    it("should allow streams to be used in a manually orchestrated MPU", async () => {
+      const cmpu = await s3.createMultipartUpload({
+        Bucket,
+        Key: Key + "-mpu",
+      });
+
+      const MB = 1024 * 1024;
+      const up = [] as UploadPartCommandOutput[];
+
+      try {
+        up.push(
+          await s3.uploadPart({
+            Bucket,
+            Key: Key + "-mpu",
+            UploadId: cmpu.UploadId,
+            Body: stream(5 * MB, 1024),
+            PartNumber: 1,
+            ContentLength: 5 * MB,
+          }),
+          await s3.uploadPart({
+            Bucket,
+            Key: Key + "-mpu",
+            UploadId: cmpu.UploadId,
+            Body: stream(MB, 64),
+            PartNumber: 2,
+            ContentLength: MB,
+          })
+        );
+        expect(logger.warn).toHaveBeenCalledWith(
+          `@smithy/util-stream - stream chunk size 1024 is below threshold of 8192, automatically buffering.`
+        );
+        expect(logger.warn).toHaveBeenCalledWith(
+          `@smithy/util-stream - stream chunk size 64 is below threshold of 8192, automatically buffering.`
+        );
+
+        await s3.completeMultipartUpload({
+          Bucket,
+          Key: Key + "-mpu",
+          UploadId: cmpu.UploadId,
+          MultipartUpload: {
+            Parts: up.map((part, i) => {
+              return {
+                PartNumber: i + 1,
+                ETag: part.ETag,
+              };
+            }),
+          },
+        });
+
+        const go = await s3.getObject({
+          Bucket,
+          Key: Key + "-mpu",
+        });
+        expect((await go.Body?.transformToByteArray())?.byteLength).toEqual(6 * MB);
+
+        expect(go.$metadata.httpStatusCode).toEqual(200);
+      } catch (e) {
+        await s3.abortMultipartUpload({
+          UploadId: cmpu.UploadId,
+          Bucket,
+          Key: Key + "-mpu",
+        });
+        throw e;
+      }
+    });
+  }, 45_000);
+
   describe("the stream returned by S3::getObject should function interchangeably between ChecksumStream and default streams", () => {
     it("when collecting the stream", async () => {
       const defaultStream = (await s3_noChecksum.getObject({ Bucket, Key })).Body as Readable;

packages/middleware-flexible-checksums/src/resolveFlexibleChecksumsConfig.ts

@@ -18,11 +18,25 @@ export interface FlexibleChecksumsInputConfig {
    * Determines when checksum validation will be performed on response payloads.
    */
   responseChecksumValidation?: ResponseChecksumValidation | Provider<ResponseChecksumValidation>;
+
+  /**
+   * Default 65536.
+   *
+   * Minimum number of bytes to buffer into a chunk when processing input streams
+   * with chunked encoding (that is, when request checksums are enabled).
+   * A minimum of 8kb = 8 * 1024 is required, and 64kb or higher is recommended.
+   *
+   * See https://docs.aws.amazon.com/AmazonS3/latest/API/sigv4-streaming.html.
+   *
+   * To turn off this feature, configure the value to zero or false.
+   */
+  requestStreamBufferSize?: number | false;
 }
 
 export interface FlexibleChecksumsResolvedConfig {
   requestChecksumCalculation: Provider<RequestChecksumCalculation>;
   responseChecksumValidation: Provider<ResponseChecksumValidation>;
+  requestStreamBufferSize: number;
 }
 
 export const resolveFlexibleChecksumsConfig = <T>(
@@ -35,4 +49,5 @@ export const resolveFlexibleChecksumsConfig = <T>(
   responseChecksumValidation: normalizeProvider(
     input.responseChecksumValidation ?? DEFAULT_RESPONSE_CHECKSUM_VALIDATION
   ),
+  requestStreamBufferSize: Number(input.requestStreamBufferSize ?? 64 * 1024),
 });

packages/middleware-sdk-s3/src/check-content-length-header.spec.ts

@@ -130,4 +130,24 @@ describe("checkContentLengthHeaderMiddleware", () => {
 
     expect(spy).not.toHaveBeenCalled();
   });
+
+  it("does not warn if uploading a payload of known length via alternate header x-amz-decoded-content-length", async () => {
+    const handler = checkContentLengthHeader()(mockNextHandler, {});
+
+    await handler({
+      request: {
+        method: null,
+        protocol: null,
+        hostname: null,
+        path: null,
+        query: {},
+        headers: {
+          "x-amz-decoded-content-length": "5",
+        },
+      },
+      input: {},
+    });
+
+    expect(spy).not.toHaveBeenCalled();
+  });
 });

packages/middleware-sdk-s3/src/check-content-length-header.ts

@@ -12,6 +12,7 @@ import {
 } from "@smithy/types";
 
 const CONTENT_LENGTH_HEADER = "content-length";
+const DECODED_CONTENT_LENGTH_HEADER = "x-amz-decoded-content-length";
 
 /**
  * @internal
@@ -28,7 +29,7 @@ export function checkContentLengthHeader(): FinalizeRequestMiddleware<any, any>
       const { request } = args;
 
       if (HttpRequest.isInstance(request)) {
-        if (!(CONTENT_LENGTH_HEADER in request.headers)) {
+        if (!(CONTENT_LENGTH_HEADER in request.headers) && !(DECODED_CONTENT_LENGTH_HEADER in request.headers)) {
           const message = `Are you using a Stream of unknown length as the Body of a PutObject request? Consider using Upload instead from @aws-sdk/lib-storage.`;
           if (typeof context?.logger?.warn === "function" && !(context.logger instanceof NoOpLogger)) {
             context.logger.warn(message);

private/aws-client-api-test/src/client-interface-tests/client-s3/impl/initializeWithMaximalConfiguration.ts

@@ -130,6 +130,7 @@ export const initializeWithMaximalConfiguration = () => {
     requestChecksumCalculation: DEFAULT_REQUEST_CHECKSUM_CALCULATION,
     responseChecksumValidation: DEFAULT_RESPONSE_CHECKSUM_VALIDATION,
     userAgentAppId: "testApp",
+    requestStreamBufferSize: 8 * 1024,
   };
 
   const s3 = new S3Client(config);

supplemental-docs/CLIENTS.md

@@ -753,7 +753,8 @@ See also https://aws.amazon.com/blogs/developer/middleware-stack-modular-aws-sdk
 
 ### S3
 
-`followRegionRedirects`:
+#### `followRegionRedirects`:
+
 This feature was previously called the S3 Global Client. Setting this option to true enables failed requests to be retried with a corrected region when receiving a permanent redirect error with status 301. Note that this can result in additional latency owing to the retried request. This feature should only be used as a last resort if you do not know the region of your bucket(s) ahead of time.
 
 ```ts
@@ -763,6 +764,20 @@ new S3Client({
 });
 ```
 
+#### `requestChecksumCalculation` and `responseChecksumValidation`:
+
+These may be set to `WHEN_REQUIRED` or `WHEN_SUPPORTED`. See https://github.com/aws/aws-sdk-js-v3/issues/6810.
+
+#### `requestStreamBufferSize`:
+
+This has a default value of `64 * 1024` bytes, or 64kb. This only comes into play when request checksums are enabled.
+When this is enabled, user input streams that emit chunks having size less than the value configured will be buffered
+until the buffer exceeds the desired size before continuing to flow.
+
+This may be set to `false` to disable, or a numeric byte size.
+
+See https://docs.aws.amazon.com/AmazonS3/latest/API/sigv4-streaming.html.
+
 ### SQS
 
 #### Using Queue Names with SQS Client