remove dependency on memory queue from kv cache again

Author: james-elicx

packages/cloudflare/src/api/kv-cache.ts

@@ -2,7 +2,6 @@ import type { CacheValue, IncrementalCache, WithLastModified } from "@opennextjs
 import { IgnorableError, RecoverableError } from "@opennextjs/aws/utils/error.js";
 
 import { getCloudflareContext } from "./cloudflare-context.js";
-import memoryQueue from "./memory-queue.js";
 
 export const CACHE_ASSET_DIR = "cdn-cgi/_next_cache";
 
@@ -117,8 +116,6 @@ class Cache implements IncrementalCache {
       );
     } catch {
       throw new RecoverableError(`Failed to set cache [${key}]`);
-    } finally {
-      memoryQueue.remove(key);
     }
   }
 

packages/cloudflare/src/api/memory-queue.spec.ts

@@ -16,28 +16,28 @@ const defaultOpts = {
 describe("MemoryQueue", () => {
   beforeAll(() => {
     vi.useFakeTimers();
-    globalThis.internalFetch = vi.fn();
+    globalThis.internalFetch = vi.fn().mockReturnValue(new Promise((res) => setTimeout(() => res(true), 1)));
   });
 
   it("should de-dupe revalidations", async () => {
-    await cache.send(defaultOpts);
+    const firstBatch = [cache.send(defaultOpts), cache.send(defaultOpts)];
+    vi.advanceTimersByTime(1);
+    await Promise.all(firstBatch);
     expect(globalThis.internalFetch).toHaveBeenCalledTimes(1);
-    await cache.send(defaultOpts);
-    expect(globalThis.internalFetch).toHaveBeenCalledTimes(1);
-
-    cache.remove("/test");
 
-    await cache.send(defaultOpts);
-    expect(globalThis.internalFetch).toHaveBeenCalledTimes(2);
-    await cache.send(defaultOpts);
+    const secondBatch = [cache.send(defaultOpts)];
+    vi.advanceTimersByTime(1);
+    await Promise.all(secondBatch);
     expect(globalThis.internalFetch).toHaveBeenCalledTimes(2);
 
     vi.advanceTimersByTime(10_000);
 
-    await cache.send(defaultOpts);
-    expect(globalThis.internalFetch).toHaveBeenCalledTimes(3);
-
-    await cache.send({ ...defaultOpts, MessageGroupId: generateMessageGroupId("/other") });
+    const thirdBatch = [
+      cache.send(defaultOpts),
+      cache.send({ ...defaultOpts, MessageGroupId: generateMessageGroupId("/other") }),
+    ];
+    vi.advanceTimersByTime(1);
+    await Promise.all(thirdBatch);
     expect(globalThis.internalFetch).toHaveBeenCalledTimes(4);
   });
 });

packages/cloudflare/src/api/memory-queue.ts

@@ -1,4 +1,3 @@
-import { generateMessageGroupId } from "@opennextjs/aws/core/routing/queue.js";
 import logger from "@opennextjs/aws/logger.js";
 import type { Queue, QueueMessage } from "@opennextjs/aws/types/overrides.js";
 
@@ -18,7 +17,7 @@ class MemoryQueue implements Queue {
     this.revalidatedPaths.set(
       MessageGroupId,
       // force remove to allow new revalidations incase something went wrong
-      setTimeout(() => this.removeId(MessageGroupId), 10_000)
+      setTimeout(() => this.revalidatedPaths.delete(MessageGroupId), 10_000)
     );
 
     try {
@@ -36,20 +35,11 @@ class MemoryQueue implements Queue {
       });
     } catch (e) {
       logger.error(e);
+    } finally {
+      clearTimeout(this.revalidatedPaths.get(MessageGroupId));
       this.revalidatedPaths.delete(MessageGroupId);
     }
   }
-
-  private removeId(id: string) {
-    clearTimeout(this.revalidatedPaths.get(id));
-    this.revalidatedPaths.delete(id);
-  }
-
-  public remove(path: string) {
-    if (this.revalidatedPaths.size > 0) {
-      this.removeId(generateMessageGroupId(path));
-    }
-  }
 }
 
 export default new MemoryQueue();