Double chunk DDB batch writes to not overwhelm DDB on load

README.md

@@ -50,6 +50,12 @@ AWS link: https://d1gwt3w78t4dm3.cloudfront.net
 
 Vercel link: https://open-next.vercel.app
 
+## Configuration
+
+### Environment variables
+
+- DYNAMO_BATCH_WRITE_COMMAND_CONCURRENCY: The number of concurrent batch write commands to DynamoDB. Defaults to 4 in an effort to leave plenty of DynamoDB write request capacity for the production load.
+
 ## Contribute
 
 To run `OpenNext` locally:

packages/open-next/src/adapters/cache.ts

@@ -14,7 +14,9 @@
 } from "@aws-sdk/client-s3";
 import path from "path";
 
+import { MAX_DYNAMO_BATCH_WRITE_ITEM_COUNT } from "./constants.js";
 import { debug, error } from "./logger.js";
+import { chunk } from "./util.js";
 
 interface CachedFetchValue {
   kind: "FETCH";
@@ -160,7 +162,7 @@
     }
   }
 
   async getIncrementalCache(
     key: string,
     keys: string[],
   ): Promise<CacheHandlerValue | null> {
@@ -420,7 +422,7 @@
     try {
       if (disableDynamoDBCache) return;
       await Promise.all(
-        this.chunkArray(req, 25).map((Items) => {
+        chunk(req, MAX_DYNAMO_BATCH_WRITE_ITEM_COUNT).map((Items) => {
           return this.dynamoClient.send(
             new BatchWriteItemCommand({
               RequestItems: {
@@ -455,14 +457,6 @@
     };
   }
 
-  private chunkArray<T>(array: T[], chunkSize: number) {
-    const chunks = [];
-    for (let i = 0; i < array.length; i += chunkSize) {
-      chunks.push(array.slice(i, i + chunkSize));
-    }
-    return chunks;
-  }
-
   // S3 handling
 
   private buildS3Key(key: string, extension: Extension) {

packages/open-next/src/adapters/constants.ts

@@ -0,0 +1,25 @@
+export const MAX_DYNAMO_BATCH_WRITE_ITEM_COUNT = 25;
+
+/**
+ * Sending to dynamo X commands at a time, using about X * 25 write units per batch to not overwhelm DDB
+ * and give production plenty of room to work with. With DDB Response times, you can expect about 10 batches per second.
+ */
+const DEFAULT_DYNAMO_BATCH_WRITE_COMMAND_CONCURRENCY = 4;
+
+export const getDynamoBatchWriteCommandConcurrency = (): number => {
+  const dynamoBatchWriteCommandConcurrencyFromEnv =
+    process.env.DYNAMO_BATCH_WRITE_COMMAND_CONCURRENCY;
+  const parsedDynamoBatchWriteCommandConcurrencyFromEnv =
+    dynamoBatchWriteCommandConcurrencyFromEnv
+      ? parseInt(dynamoBatchWriteCommandConcurrencyFromEnv)
+      : undefined;
+
+  if (
+    parsedDynamoBatchWriteCommandConcurrencyFromEnv &&
+    !isNaN(parsedDynamoBatchWriteCommandConcurrencyFromEnv)
+  ) {
+    return parsedDynamoBatchWriteCommandConcurrencyFromEnv;
+  }
+
+  return DEFAULT_DYNAMO_BATCH_WRITE_COMMAND_CONCURRENCY;
+};

packages/open-next/src/adapters/dynamo-provider.ts

@@ -5,7 +5,15 @@ import {
 import { CdkCustomResourceEvent, CdkCustomResourceResponse } from "aws-lambda";
 import { readFileSync } from "fs";
 
-const client = new DynamoDBClient({});
+import {
+  getDynamoBatchWriteCommandConcurrency,
+  MAX_DYNAMO_BATCH_WRITE_ITEM_COUNT,
+} from "./constants.js";
+import { chunk } from "./util.js";
+
+const PHYSICAL_RESOURCE_ID = "dynamodb-cache";
+
+const dynamoClient = new DynamoDBClient({});
 
 type DataType = {
   tag: {
@@ -32,44 +40,49 @@ export async function handler(
 }
 
 async function insert(): Promise<CdkCustomResourceResponse> {
+  const tableName = process.env.CACHE_DYNAMO_TABLE!;
+
   const file = readFileSync(`dynamodb-cache.json`, "utf8");
 
   const data: DataType[] = JSON.parse(file);
 
-  // Chunk array into batches of 25
-  const chunked = data.reduce((acc, curr, i) => {
-    const index = Math.floor(i / 25);
-    acc[index] = [...(acc[index] || []), curr];
-    return acc;
-  }, [] as DataType[][]);
+  const dataChunks = chunk(data, MAX_DYNAMO_BATCH_WRITE_ITEM_COUNT);
 
-  const TableName = process.env.CACHE_DYNAMO_TABLE!;
-
-  const promises = chunked.map((chunk) => {
-    const params = {
+  const batchWriteParamsArray = dataChunks.map((chunk) => {
+    return {
       RequestItems: {
-        [TableName]: chunk.map((item) => ({
+        [tableName]: chunk.map((item) => ({
           PutRequest: {
             Item: item,
           },
         })),
       },
     };
-    return client.send(new BatchWriteItemCommand(params));
   });
 
-  await Promise.all(promises);
+  const paramsChunks = chunk(
+    batchWriteParamsArray,
+    getDynamoBatchWriteCommandConcurrency(),
+  );
+
+  for (const paramsChunk of paramsChunks) {
+    await Promise.all(
+      paramsChunk.map((params) =>
+        dynamoClient.send(new BatchWriteItemCommand(params)),
+      ),
+    );
+  }
 
   return {
-    PhysicalResourceId: "dynamodb-cache",
+    PhysicalResourceId: PHYSICAL_RESOURCE_ID,
     Data: {},
   };
 }
 
 async function remove(): Promise<CdkCustomResourceResponse> {
   // Do we want to actually delete anything here?
   return {
-    PhysicalResourceId: "dynamodb-cache",
+    PhysicalResourceId: PHYSICAL_RESOURCE_ID,
     Data: {},
   };
 }

packages/open-next/src/adapters/util.ts

@@ -38,3 +38,19 @@ export function parseCookies(
 
   return cookies;
 }
+
+/**
+ * Create an array of arrays of size `chunkSize` from `items`
+ * @param items Array of T
+ * @param chunkSize size of each chunk
+ * @returns T[][]
+ */
+export function chunk<T>(items: T[], chunkSize: number): T[][] {
+  const chunked = items.reduce((acc, curr, i) => {
+    const chunkIndex = Math.floor(i / chunkSize);
+    acc[chunkIndex] = [...(acc[chunkIndex] ?? []), curr];
+    return acc;
+  }, new Array<T[]>());
+
+  return chunked;
+}