Fix locked readable stream in composable cache (#939)

* fix(composable-cache): update pending write handling to avoid locked readable stream

* changeset

* add unit test for the composable cache

* review

---------

Co-authored-by: Nicolas Dorseuil <nicolas@gitbook.io>

Author: conico974

.changeset/wet-emus-brake.md

@@ -0,0 +1,5 @@
+---
+"@opennextjs/aws": patch
+---
+
+fix locked readable stream in the composable cache

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

@@ -1,17 +1,27 @@
 import type { ComposableCacheEntry, ComposableCacheHandler } from "types/cache";
+import type { CacheValue } from "types/overrides";
 import { writeTags } from "utils/cache";
 import { fromReadableStream, toReadableStream } from "utils/stream";
 import { debug } from "./logger";
 
-const pendingWritePromiseMap = new Map<string, Promise<ComposableCacheEntry>>();
+const pendingWritePromiseMap = new Map<
+  string,
+  Promise<CacheValue<"composable">>
+>();
 
 export default {
   async get(cacheKey: string) {
     try {
       // We first check if we have a pending write for this cache key
       // If we do, we return the pending promise instead of fetching the cache
       if (pendingWritePromiseMap.has(cacheKey)) {
-        return pendingWritePromiseMap.get(cacheKey);
+        const stored = pendingWritePromiseMap.get(cacheKey);
+        if (stored) {
+          return stored.then((entry) => ({
+            ...entry,
+            value: toReadableStream(entry.value),
+          }));
+        }
       }
       const result = await globalThis.incrementalCache.get(
         cacheKey,
@@ -59,16 +69,20 @@ export default {
   },
 
   async set(cacheKey: string, pendingEntry: Promise<ComposableCacheEntry>) {
-    pendingWritePromiseMap.set(cacheKey, pendingEntry);
-    const entry = await pendingEntry.finally(() => {
+    const promiseEntry = pendingEntry.then(async (entry) => ({
+      ...entry,
+      value: await fromReadableStream(entry.value),
+    }));
+    pendingWritePromiseMap.set(cacheKey, promiseEntry);
+
+    const entry = await promiseEntry.finally(() => {
       pendingWritePromiseMap.delete(cacheKey);
     });
-    const valueToStore = await fromReadableStream(entry.value);
     await globalThis.incrementalCache.set(
       cacheKey,
       {
         ...entry,
-        value: valueToStore,
+        value: entry.value,
       },
       "composable",
     );