feat: invoke a remove method on the queue after cache set

.changeset/stupid-gifts-cough.md

@@ -0,0 +1,5 @@
+---
+"@opennextjs/aws": minor
+---
+
+Invoke a remove method on the queue after an incremental cache set for a route takes place.

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

@@ -393,6 +393,19 @@ export default class Cache {
     } finally {
       // We need to resolve the promise even if there was an error
       detachedPromise?.resolve();
+
+      switch (data?.kind) {
+        case "ROUTE":
+        case "APP_ROUTE":
+        case "PAGE":
+        case "PAGES":
+        case "APP_PAGE":
+        case "REDIRECT":
+          globalThis.queue.remove?.(key);
+          break;
+        default:
+        // Fetch cache entries and images do not use the queue
+      }
     }
   }
 

packages/open-next/src/core/routing/cacheInterceptor.ts

@@ -5,9 +5,9 @@ import type { InternalEvent, InternalResult } from "types/open-next";
 import type { CacheValue } from "types/overrides";
 import { emptyReadableStream, toReadableStream } from "utils/stream";
 
-import { debug } from "../../adapters/logger";
-import { localizePath } from "./i18n";
-import { generateMessageGroupId } from "./util";
+import { debug } from "../../adapters/logger.js";
+import { localizePath } from "./i18n/index.js";
+import { generateMessageGroupId } from "./queue.js";
 
 const CACHE_ONE_YEAR = 60 * 60 * 24 * 365;
 const CACHE_ONE_MONTH = 60 * 60 * 24 * 30;

packages/open-next/src/core/routing/queue.ts

@@ -0,0 +1,44 @@
+// Since we're using a FIFO queue, every messageGroupId is treated sequentially
+// This could cause a backlog of messages in the queue if there is too much page to
+// revalidate at once. To avoid this, we generate a random messageGroupId for each
+// revalidation request.
+// We can't just use a random string because we need to ensure that the same rawPath
+// will always have the same messageGroupId.
+// https://stackoverflow.com/questions/521295/seeding-the-random-number-generator-in-javascript#answer-47593316
+export function generateMessageGroupId(rawPath: string) {
+  let a = cyrb128(rawPath);
+  // We use mulberry32 to generate a random int between 0 and MAX_REVALIDATE_CONCURRENCY
+  let t = (a += 0x6d2b79f5);
+  t = Math.imul(t ^ (t >>> 15), t | 1);
+  t ^= t + Math.imul(t ^ (t >>> 7), t | 61);
+  const randomFloat = ((t ^ (t >>> 14)) >>> 0) / 4294967296;
+  // This will generate a random int between 0 and MAX_REVALIDATE_CONCURRENCY
+  // This means that we could have 1000 revalidate request at the same time
+  const maxConcurrency = Number.parseInt(
+    process.env.MAX_REVALIDATE_CONCURRENCY ?? "10",
+  );
+  const randomInt = Math.floor(randomFloat * maxConcurrency);
+  return `revalidate-${randomInt}`;
+}
+
+// Used to generate a hash int from a string
+function cyrb128(str: string) {
+  let h1 = 1779033703;
+  let h2 = 3144134277;
+  let h3 = 1013904242;
+  let h4 = 2773480762;
+  for (let i = 0, k: number; i < str.length; i++) {
+    k = str.charCodeAt(i);
+    h1 = h2 ^ Math.imul(h1 ^ k, 597399067);
+    h2 = h3 ^ Math.imul(h2 ^ k, 2869860233);
+    h3 = h4 ^ Math.imul(h3 ^ k, 951274213);
+    h4 = h1 ^ Math.imul(h4 ^ k, 2716044179);
+  }
+  h1 = Math.imul(h3 ^ (h1 >>> 18), 597399067);
+  h2 = Math.imul(h4 ^ (h2 >>> 22), 2869860233);
+  h3 = Math.imul(h1 ^ (h3 >>> 17), 951274213);
+  h4 = Math.imul(h2 ^ (h4 >>> 19), 2716044179);
+  // biome-ignore lint/style/noCommaOperator:
+  (h1 ^= h2 ^ h3 ^ h4), (h2 ^= h1), (h3 ^= h1), (h4 ^= h1);
+  return h1 >>> 0;
+}

packages/open-next/src/core/routing/util.ts

@@ -17,6 +17,7 @@ import type {
 import { debug, error } from "../../adapters/logger.js";
 import { isBinaryContentType } from "../../utils/binary.js";
 import { localizePath } from "./i18n/index.js";
+import { generateMessageGroupId } from "./queue.js";
 
 /**
  *
@@ -310,51 +311,6 @@ export async function revalidateIfRequired(
   }
 }
 
-// Since we're using a FIFO queue, every messageGroupId is treated sequentially
-// This could cause a backlog of messages in the queue if there is too much page to
-// revalidate at once. To avoid this, we generate a random messageGroupId for each
-// revalidation request.
-// We can't just use a random string because we need to ensure that the same rawPath
-// will always have the same messageGroupId.
-// https://stackoverflow.com/questions/521295/seeding-the-random-number-generator-in-javascript#answer-47593316
-export function generateMessageGroupId(rawPath: string) {
-  let a = cyrb128(rawPath);
-  // We use mulberry32 to generate a random int between 0 and MAX_REVALIDATE_CONCURRENCY
-  let t = (a += 0x6d2b79f5);
-  t = Math.imul(t ^ (t >>> 15), t | 1);
-  t ^= t + Math.imul(t ^ (t >>> 7), t | 61);
-  const randomFloat = ((t ^ (t >>> 14)) >>> 0) / 4294967296;
-  // This will generate a random int between 0 and MAX_REVALIDATE_CONCURRENCY
-  // This means that we could have 1000 revalidate request at the same time
-  const maxConcurrency = Number.parseInt(
-    process.env.MAX_REVALIDATE_CONCURRENCY ?? "10",
-  );
-  const randomInt = Math.floor(randomFloat * maxConcurrency);
-  return `revalidate-${randomInt}`;
-}
-
-// Used to generate a hash int from a string
-function cyrb128(str: string) {
-  let h1 = 1779033703;
-  let h2 = 3144134277;
-  let h3 = 1013904242;
-  let h4 = 2773480762;
-  for (let i = 0, k: number; i < str.length; i++) {
-    k = str.charCodeAt(i);
-    h1 = h2 ^ Math.imul(h1 ^ k, 597399067);
-    h2 = h3 ^ Math.imul(h2 ^ k, 2869860233);
-    h3 = h4 ^ Math.imul(h3 ^ k, 951274213);
-    h4 = h1 ^ Math.imul(h4 ^ k, 2716044179);
-  }
-  h1 = Math.imul(h3 ^ (h1 >>> 18), 597399067);
-  h2 = Math.imul(h4 ^ (h2 >>> 22), 2869860233);
-  h3 = Math.imul(h1 ^ (h3 >>> 17), 951274213);
-  h4 = Math.imul(h2 ^ (h4 >>> 19), 2716044179);
-  // biome-ignore lint/style/noCommaOperator:
-  (h1 ^= h2 ^ h3 ^ h4), (h2 ^= h1), (h3 ^= h1), (h4 ^= h1);
-  return h1 >>> 0;
-}
-
 /**
  *
  * @__PURE__

packages/open-next/src/types/overrides.ts

@@ -24,8 +24,9 @@ export interface QueueMessage {
 }
 
 export interface Queue {
-  send(message: QueueMessage): Promise<void>;
   name: string;
+  send(message: QueueMessage): Promise<void>;
+  remove?(path: string): void;
 }
 
 // Incremental cache

packages/tests-unit/tests/adapters/cache.test.ts

@@ -41,6 +41,13 @@ describe("CacheHandler", () => {
   };
   globalThis.tagCache = tagCache;
 
+  const queue = {
+    name: "mock",
+    send: vi.fn(),
+    remove: vi.fn(),
+  };
+  globalThis.queue = queue;
+
   const invalidateCdnHandler = {
     name: "mock",
     invalidatePaths: vi.fn(),
@@ -335,6 +342,7 @@ describe("CacheHandler", () => {
         { type: "route", body: "{}", meta: { status: 200, headers: {} } },
         false,
       );
+      expect(globalThis.queue.remove).toHaveBeenCalledWith("key");
     });
 
     it("Should set cache when for APP_ROUTE", async () => {
@@ -356,6 +364,7 @@ describe("CacheHandler", () => {
         },
         false,
       );
+      expect(globalThis.queue.remove).toHaveBeenCalledWith("key");
     });
 
     it("Should set cache when for PAGE", async () => {
@@ -376,6 +385,7 @@ describe("CacheHandler", () => {
         },
         false,
       );
+      expect(globalThis.queue.remove).toHaveBeenCalledWith("key");
     });
 
     it("Should set cache when for PAGES", async () => {
@@ -397,6 +407,7 @@ describe("CacheHandler", () => {
         },
         false,
       );
+      expect(globalThis.queue.remove).toHaveBeenCalledWith("key");
     });
 
     it("Should set cache when for APP_PAGE", async () => {
@@ -418,6 +429,7 @@ describe("CacheHandler", () => {
         },
         false,
       );
+      expect(globalThis.queue.remove).toHaveBeenCalledWith("key");
     });
 
     it("Should set cache when for FETCH", async () => {
@@ -448,6 +460,7 @@ describe("CacheHandler", () => {
         },
         true,
       );
+      expect(globalThis.queue.remove).not.toHaveBeenCalled();
     });
 
     it("Should set cache when for REDIRECT", async () => {
@@ -461,6 +474,7 @@ describe("CacheHandler", () => {
         },
         false,
       );
+      expect(globalThis.queue.remove).toHaveBeenCalledWith("key");
     });
 
     it("Should not set cache when for IMAGE (not implemented)", async () => {
@@ -472,6 +486,7 @@ describe("CacheHandler", () => {
       });
 
       expect(incrementalCache.set).not.toHaveBeenCalled();
+      expect(globalThis.queue.remove).not.toHaveBeenCalled();
     });
 
     it("Should not throw when set cache throws", async () => {