fix(storage-s3): ensure s3 sockets are cleaned up (#11626)

Ensures all s3 sockets are cleaned up. Now passes through default
request handler options that `@smithy/node-http-handler` now handles
properly.

Fixes #6382 

```ts
const defaultRequestHandlerOpts: NodeHttpHandlerOptions = {
  httpAgent: {
    keepAlive: true,
    maxSockets: 100,
  },
  httpsAgent: {
    keepAlive: true,
    maxSockets: 100,
  },
}
```

If you continue to have socket issues, you can customize any of the
options by setting `requestHandler` property on your s3 config. This
will take precedence if set.

```ts
requestHandler: {
  httpAgent: {
    maxSockets: 300,
    keepAlive: true,
  },
  httpsAgent: {
    maxSockets: 300,
    keepAlive: true,
  },

  // Optional, only set these if you continue to see issues. Be wary of timeouts if you're dealing with large files.

  // time limit (ms) for receiving response.
  requestTimeout: 5_000,

  // time limit (ms) for establishing connection.
  connectionTimeout: 5_000,
}),
```

Author: denolfe

packages/storage-s3/package.json

@@ -52,6 +52,7 @@
     "@payloadcms/plugin-cloud-storage": "workspace:*"
   },
   "devDependencies": {
+    "@smithy/node-http-handler": "4.0.3",
     "payload": "workspace:*"
   },
   "peerDependencies": {

packages/storage-s3/src/index.ts

@@ -5,6 +5,7 @@ import type {
   CollectionOptions,
   GeneratedAdapter,
 } from '@payloadcms/plugin-cloud-storage/types'
+import type { NodeHttpHandlerOptions } from '@smithy/node-http-handler'
 import type { Config, Plugin, UploadCollectionSlug } from 'payload'
 
 import * as AWS from '@aws-sdk/client-s3'
@@ -64,16 +65,31 @@ export type S3StorageOptions = {
 
 type S3StoragePlugin = (storageS3Args: S3StorageOptions) => Plugin
 
+let storageClient: AWS.S3 | null = null
+
+const defaultRequestHandlerOpts: NodeHttpHandlerOptions = {
+  httpAgent: {
+    keepAlive: true,
+    maxSockets: 100,
+  },
+  httpsAgent: {
+    keepAlive: true,
+    maxSockets: 100,
+  },
+}
+
 export const s3Storage: S3StoragePlugin =
   (s3StorageOptions: S3StorageOptions) =>
   (incomingConfig: Config): Config => {
-    let storageClient: AWS.S3 | null = null
-
     const getStorageClient: () => AWS.S3 = () => {
       if (storageClient) {
         return storageClient
       }
-      storageClient = new AWS.S3(s3StorageOptions.config ?? {})
+
+      storageClient = new AWS.S3({
+        requestHandler: defaultRequestHandlerOpts,
+        ...(s3StorageOptions.config ?? {}),
+      })
       return storageClient
     }
 

packages/storage-s3/src/staticHandler.ts

@@ -24,6 +24,12 @@ const isNodeReadableStream = (body: unknown): body is Readable => {
   )
 }
 
+const destroyStream = (object: AWS.GetObjectOutput | undefined) => {
+  if (object?.Body && isNodeReadableStream(object.Body)) {
+    object.Body.destroy()
+  }
+}
+
 // Convert a stream into a promise that resolves with a Buffer
 // eslint-disable-next-line @typescript-eslint/no-explicit-any
 const streamToBuffer = async (readableStream: any) => {
@@ -36,12 +42,13 @@ const streamToBuffer = async (readableStream: any) => {
 
 export const getHandler = ({ bucket, collection, getStorageClient }: Args): StaticHandler => {
   return async (req, { params: { clientUploadContext, filename } }) => {
+    let object: AWS.GetObjectOutput | undefined = undefined
     try {
       const prefix = await getFilePrefix({ clientUploadContext, collection, filename, req })
 
       const key = path.posix.join(prefix, filename)
 
-      const object = await getStorageClient().getObject({
+      object = await getStorageClient().getObject({
         Bucket: bucket,
         Key: key,
       })
@@ -54,7 +61,7 @@ export const getHandler = ({ bucket, collection, getStorageClient }: Args): Stat
       const objectEtag = object.ETag
 
       if (etagFromHeaders && etagFromHeaders === objectEtag) {
-        const response = new Response(null, {
+        return new Response(null, {
           headers: new Headers({
             'Accept-Ranges': String(object.AcceptRanges),
             'Content-Length': String(object.ContentLength),
@@ -63,13 +70,6 @@ export const getHandler = ({ bucket, collection, getStorageClient }: Args): Stat
           }),
           status: 304,
         })
-
-        // Manually destroy stream before returning cached results to close socket
-        if (object.Body && isNodeReadableStream(object.Body)) {
-          object.Body.destroy()
-        }
-
-        return response
       }
 
       // On error, manually destroy stream to close socket
@@ -99,6 +99,8 @@ export const getHandler = ({ bucket, collection, getStorageClient }: Args): Stat
     } catch (err) {
       req.payload.logger.error(err)
       return new Response('Internal Server Error', { status: 500 })
+    } finally {
+      destroyStream(object)
     }
   }
 }