opennextjs · conico974 · Mar 20, 2025 · Mar 20, 2025 · Mar 20, 2025 · Mar 20, 2025
diff --git a/.changeset/friendly-experts-eat.md b/.changeset/friendly-experts-eat.md
@@ -0,0 +1,5 @@
+---
+"@opennextjs/cloudflare": patch
+---
+
+fix deduplication for memory queue and add some log
diff --git a/examples/common/apps.ts b/examples/common/apps.ts
@@ -14,6 +14,7 @@ const apps = [
   "pages-router",
   // overrides
   "d1-tag-next",
+  "memory-queue",
   // bugs
   "gh-119",
   "gh-219",

diff --git a/examples/overrides/memory-queue/.gitignore b/examples/overrides/memory-queue/.gitignore
@@ -0,0 +1,47 @@
+# See https://help.github.com/articles/ignoring-files/ for more about ignoring files.
+
+# dependencies
+/node_modules
+/.pnp
+.pnp.*
+.yarn/*
+!.yarn/patches
+!.yarn/plugins
+!.yarn/releases
+!.yarn/versions
+
+# testing
+/coverage
+
+# next.js
+/.next/
+/out/
+
+# production
+/build
+
+# misc
+.DS_Store
+*.pem
+
+# debug
+npm-debug.log*
+yarn-debug.log*
+yarn-error.log*
+.pnpm-debug.log*
+
+# env files (can opt-in for committing if needed)
+.env*
+
+# vercel
+.vercel
+
+# typescript
+*.tsbuildinfo
+next-env.d.ts
+
+# playwright
+/test-results/
+/playwright-report/
+/blob-report/
+/playwright/.cache/
diff --git a/examples/overrides/memory-queue/app/favicon.ico b/examples/overrides/memory-queue/app/favicon.ico
diff --git a/examples/overrides/memory-queue/app/globals.css b/examples/overrides/memory-queue/app/globals.css
@@ -0,0 +1,14 @@
+html,
+body {
+  max-width: 100vw;
+  overflow-x: hidden;
+  height: 100vh;
+  display: flex;
+  flex-direction: column;
+}
+
+footer {
+  padding: 1rem;
+  display: flex;
+  justify-content: end;
+}
diff --git a/examples/overrides/memory-queue/app/layout.tsx b/examples/overrides/memory-queue/app/layout.tsx
@@ -0,0 +1,25 @@
+import type { Metadata } from "next";
+import "./globals.css";
+
+import { getCloudflareContext } from "@opennextjs/cloudflare";
+
+export const metadata: Metadata = {
+  title: "SSG App",
+  description: "An app in which all the routes are SSG'd",
+};
+
+export default async function RootLayout({
+  children,
+}: Readonly<{
+  children: React.ReactNode;
+}>) {
+  const cloudflareContext = await getCloudflareContext({
+    async: true,
+  });
+
+  return (
+    <html lang="en">
+      <body>{children}</body>
+    </html>
+  );
+}
diff --git a/examples/overrides/memory-queue/app/page.module.css b/examples/overrides/memory-queue/app/page.module.css
@@ -0,0 +1,17 @@
+.page {
+  display: grid;
+  grid-template-rows: 20px 1fr 20px;
+  align-items: center;
+  justify-items: center;
+  flex: 1;
+  border: 3px solid gray;
+  margin: 1rem;
+  margin-block-end: 0;
+}
+
+.main {
+  display: flex;
+  flex-direction: column;
+  gap: 32px;
+  grid-row-start: 2;
+}
diff --git a/examples/overrides/memory-queue/app/page.tsx b/examples/overrides/memory-queue/app/page.tsx
@@ -0,0 +1,16 @@
+import styles from "./page.module.css";
+
+export const revalidate = 5;
+
+export default async function Home() {
+  // We purposefully wait for 2 seconds to allow deduplication to occur
+  await new Promise((resolve) => setTimeout(resolve, 2000));
+  return (
+    <div className={styles.page}>
+      <main className={styles.main}>
+        <h1>Hello from a Statically generated page</h1>
+        <p data-testid="date-local">{Date.now()}</p>
+      </main>
+    </div>
+  );
+}
diff --git a/examples/overrides/memory-queue/e2e/base.spec.ts b/examples/overrides/memory-queue/e2e/base.spec.ts
@@ -0,0 +1,36 @@
+import { test, expect } from "@playwright/test";
+
+test.describe("memory-queue", () => {
+  test("the index page should work", async ({ page }) => {
+    await page.goto("/");
+    await expect(page.getByText("Hello from a Statically generated page")).toBeVisible();
+  });
+
+  test("the index page should revalidate", async ({ page, request }) => {
+    // We need to make sure the page is loaded and is a HIT
+    // If it is STALE, the next hit may have an updated date and thus fail the test
+    let cacheHeaders = "";
+    do {
+      const req = await request.get("/");
+      cacheHeaders = req.headers()["x-nextjs-cache"];
+      await page.waitForTimeout(500);
+    } while (cacheHeaders !== "HIT");
+
+    await page.goto("/");
+    const firstDate = await page.getByTestId("date-local").textContent();
+
+    await page.reload();
+    let newDate = await page.getByTestId("date-local").textContent();
+    expect(newDate).toBe(firstDate);
+
+    await page.waitForTimeout(5000);
+
+    do {
+      await page.reload();
+      newDate = await page.getByTestId("date-local").textContent();
+      await page.waitForTimeout(1000);
+    } while (newDate === firstDate);
+
+    expect(newDate).not.toBe(firstDate);
+  });
+});
diff --git a/examples/overrides/memory-queue/e2e/playwright.config.ts b/examples/overrides/memory-queue/e2e/playwright.config.ts
@@ -0,0 +1,8 @@
+import { configurePlaywright } from "../../../common/config-e2e";
+
+// Here we don't want to run the tests in parallel
+export default configurePlaywright("memory-queue", {
+  isCI: !!process.env.CI,
+  parallel: false,
+  multipleBrowsers: false,
+});
diff --git a/examples/overrides/memory-queue/next.config.ts b/examples/overrides/memory-queue/next.config.ts
@@ -0,0 +1,11 @@
+import type { NextConfig } from "next";
+import { initOpenNextCloudflareForDev } from "@opennextjs/cloudflare";
+
+initOpenNextCloudflareForDev();
+
+const nextConfig: NextConfig = {
+  typescript: { ignoreBuildErrors: true },
+  eslint: { ignoreDuringBuilds: true },
+};
+
+export default nextConfig;
diff --git a/examples/overrides/memory-queue/open-next.config.ts b/examples/overrides/memory-queue/open-next.config.ts
@@ -0,0 +1,8 @@
+import { defineCloudflareConfig } from "@opennextjs/cloudflare";
+import kvIncrementalCache from "@opennextjs/cloudflare/kv-cache";
+import memoryQueue from "@opennextjs/cloudflare/memory-queue";
+
+export default defineCloudflareConfig({
+  incrementalCache: kvIncrementalCache,
+  queue: memoryQueue,
+});
diff --git a/examples/overrides/memory-queue/package.json b/examples/overrides/memory-queue/package.json
@@ -0,0 +1,28 @@
+{
+  "name": "memory-queue",
+  "version": "0.1.0",
+  "private": true,
+  "scripts": {
+    "dev": "next dev",
+    "build": "next build",
+    "start": "next start",
+    "lint": "next lint",
+    "build:worker": "opennextjs-cloudflare",
+    "preview": "pnpm build:worker && pnpm wrangler dev",
+    "e2e": "playwright test -c e2e/playwright.config.ts"
+  },
+  "dependencies": {
+    "react": "catalog:e2e",
+    "react-dom": "catalog:e2e",
+    "next": "catalog:e2e"
+  },
+  "devDependencies": {
+    "@opennextjs/cloudflare": "workspace:*",
+    "@playwright/test": "catalog:",
+    "@types/node": "catalog:",
+    "@types/react": "catalog:e2e",
+    "@types/react-dom": "catalog:e2e",
+    "typescript": "catalog:",
+    "wrangler": "catalog:"
+  }
+}
diff --git a/examples/overrides/memory-queue/tsconfig.json b/examples/overrides/memory-queue/tsconfig.json
@@ -0,0 +1,27 @@
+{
+  "compilerOptions": {
+    "target": "ES2017",
+    "lib": ["dom", "dom.iterable", "esnext"],
+    "allowJs": true,
+    "skipLibCheck": true,
+    "strict": true,
+    "noEmit": true,
+    "esModuleInterop": true,
+    "module": "esnext",
+    "moduleResolution": "bundler",
+    "resolveJsonModule": true,
+    "isolatedModules": true,
+    "jsx": "preserve",
+    "incremental": true,
+    "plugins": [
+      {
+        "name": "next"
+      }
+    ],
+    "paths": {
+      "@/*": ["./*"]
+    }
+  },
+  "include": ["next-env.d.ts", "**/*.ts", "**/*.tsx", ".next/types/**/*.ts"],
+  "exclude": ["node_modules"]
+}
diff --git a/examples/overrides/memory-queue/wrangler.jsonc b/examples/overrides/memory-queue/wrangler.jsonc
@@ -0,0 +1,23 @@
+{
+  "$schema": "node_modules/wrangler/config-schema.json",
+  "main": ".open-next/worker.js",
+  "name": "memory-queue",
+  "compatibility_date": "2025-02-04",
+  "compatibility_flags": ["nodejs_compat"],
+  "assets": {
+    "directory": ".open-next/assets",
+    "binding": "ASSETS"
+  },
+  "kv_namespaces": [
+    {
+      "binding": "NEXT_CACHE_WORKERS_KV",
+      "id": "<BINDING_ID>"
+    }
+  ],
+  "services": [
+    {
+      "binding": "NEXT_CACHE_REVALIDATION_WORKER",
+      "service": "memory-queue"
+    }
+  ]
+}
diff --git a/packages/cloudflare/src/api/memory-queue.spec.ts b/packages/cloudflare/src/api/memory-queue.spec.ts
@@ -31,7 +31,7 @@ describe("MemoryQueue", () => {
     const firstRequest = cache.send({
       MessageBody: generateMessageBody({ host: "test.local", url: "/test" }),
       MessageGroupId: generateMessageGroupId("/test"),
-      MessageDeduplicationId: "",
+      MessageDeduplicationId: "/test",
     });
     vi.advanceTimersByTime(DEFAULT_REVALIDATION_TIMEOUT_MS);
     await firstRequest;
@@ -40,7 +40,7 @@ describe("MemoryQueue", () => {
     const secondRequest = cache.send({
       MessageBody: generateMessageBody({ host: "test.local", url: "/test" }),
       MessageGroupId: generateMessageGroupId("/test"),
-      MessageDeduplicationId: "",
+      MessageDeduplicationId: "/test",
     });
     vi.advanceTimersByTime(1);
     await secondRequest;
@@ -51,7 +51,7 @@ describe("MemoryQueue", () => {
     const firstRequest = cache.send({
       MessageBody: generateMessageBody({ host: "test.local", url: "/test" }),
       MessageGroupId: generateMessageGroupId("/test"),
-      MessageDeduplicationId: "",
+      MessageDeduplicationId: "/test",
     });
     vi.advanceTimersByTime(1);
     await firstRequest;
@@ -60,7 +60,7 @@ describe("MemoryQueue", () => {
     const secondRequest = cache.send({
       MessageBody: generateMessageBody({ host: "test.local", url: "/test" }),
       MessageGroupId: generateMessageGroupId("/other"),
-      MessageDeduplicationId: "",
+      MessageDeduplicationId: "/other",
     });
     vi.advanceTimersByTime(1);
     await secondRequest;
@@ -72,12 +72,12 @@ describe("MemoryQueue", () => {
       cache.send({
         MessageBody: generateMessageBody({ host: "test.local", url: "/test" }),
         MessageGroupId: generateMessageGroupId("/test"),
-        MessageDeduplicationId: "",
+        MessageDeduplicationId: "/test",
       }),
       cache.send({
         MessageBody: generateMessageBody({ host: "test.local", url: "/test" }),
         MessageGroupId: generateMessageGroupId("/test"),
-        MessageDeduplicationId: "",
+        MessageDeduplicationId: "/test",
       }),
     ];
     vi.advanceTimersByTime(1);

diff --git a/packages/cloudflare/src/api/memory-queue.ts b/packages/cloudflare/src/api/memory-queue.ts
@@ -1,4 +1,4 @@
-import logger from "@opennextjs/aws/logger.js";
+import { debug, error } from "@opennextjs/aws/adapters/logger.js";
 import type { Queue, QueueMessage } from "@opennextjs/aws/types/overrides.js";
 import { IgnorableError } from "@opennextjs/aws/utils/error.js";
 
@@ -16,40 +16,43 @@ export const DEFAULT_REVALIDATION_TIMEOUT_MS = 10_000;
 export class MemoryQueue implements Queue {
   readonly name = "memory-queue";
 
-  revalidatedPaths = new Map<string, ReturnType<typeof setTimeout>>();
+  revalidatedPaths = new Set<string>();
 
   constructor(private opts = { revalidationTimeoutMs: DEFAULT_REVALIDATION_TIMEOUT_MS }) {}
 
-  async send({ MessageBody: { host, url }, MessageGroupId }: QueueMessage): Promise<void> {
+  async send({ MessageBody: { host, url }, MessageDeduplicationId }: QueueMessage): Promise<void> {
     const service = getCloudflareContext().env.NEXT_CACHE_REVALIDATION_WORKER;
     if (!service) throw new IgnorableError("No service binding for cache revalidation worker");
 
-    if (this.revalidatedPaths.has(MessageGroupId)) return;
+    if (this.revalidatedPaths.has(MessageDeduplicationId)) return;
 
-    this.revalidatedPaths.set(
-      MessageGroupId,
-      // force remove to allow new revalidations incase something went wrong
-      setTimeout(() => this.revalidatedPaths.delete(MessageGroupId), this.opts.revalidationTimeoutMs)
-    );
+    this.revalidatedPaths.add(MessageDeduplicationId);
 
     try {
       const protocol = host.includes("localhost") ? "http" : "https";
 
       // TODO: Drop the import - https://github.com/opennextjs/opennextjs-cloudflare/issues/361
       // @ts-ignore
       const manifest = await import("./.next/prerender-manifest.json");
-      await service.fetch(`${protocol}://${host}${url}`, {
+      const response = await service.fetch(`${protocol}://${host}${url}`, {
         method: "HEAD",
         headers: {
           "x-prerender-revalidate": manifest.preview.previewModeId,
           "x-isr": "1",
         },
+        // We want to timeout the revalidation to avoid hanging the queue
+        signal: AbortSignal.timeout(this.opts.revalidationTimeoutMs),
       });
+
+      // Here we want at least to log when the revalidation was not successful
+      if (response.status !== 200 || response.headers.get("x-nextjs-cache") !== "REVALIDATED") {
+        error(`Revalidation failed for ${url} with status ${response.status}`);
+      }
+      debug(`Revalidation successful for ${url}`);
     } catch (e) {
-      logger.error(e);
+      error(e);
     } finally {
-      clearTimeout(this.revalidatedPaths.get(MessageGroupId));
-      this.revalidatedPaths.delete(MessageGroupId);
+      this.revalidatedPaths.delete(MessageDeduplicationId);
     }
   }
 }