fix(cache): settle Cache.add/addAll for responses with a body

lib/web/cache/cache.js

@@ -153,6 +153,10 @@ class Cache {
       // 5.6
       const responsePromise = Promise.withResolvers()
 
+      // Holds the response object that should be written into the cache. For
+      // body-bearing responses this is a clone with body.source filled in.
+      let responseForCache = null
+
       // 5.7
       fetchControllers.push(fetching({
         request: r,
@@ -163,6 +167,7 @@ class Cache {
               header: 'Cache.addAll',
               message: 'Received an invalid status code or the request failed.'
             }))
+            return
           } else if (response.headersList.contains('vary')) { // 2.
             // 2.1
             const fieldValues = getFieldValues(response.headersList.get('vary'))
@@ -184,6 +189,36 @@ class Cache {
               }
             }
           }
+
+          // undici's fetchFinale skips the TransformStream flush hook from the
+          // fetch spec and only observes body completion via stream.finished().
+          // Body streams are pull-driven, so with no consumer processResponseEndOfBody
+          // never runs and Cache.add/addAll hang forever on body-bearing responses
+          // (https://github.com/nodejs/undici/issues/5615).
+          // Mirror Cache.put: clone first, fully read the original stream (which
+          // also unlocks end-of-body), stash bytes on the clone for storage.
+          if (response.body != null) {
+            try {
+              const clonedResponse = cloneResponse(response)
+              const reader = response.body.stream.getReader()
+              readAllBytes(
+                reader,
+                (bytes) => {
+                  if (clonedResponse.body != null) {
+                    clonedResponse.body.source = bytes
+                  }
+                  responseForCache = clonedResponse
+                },
+                (error) => {
+                  responsePromise.reject(error)
+                }
+              )
+            } catch (error) {
+              responsePromise.reject(error)
+            }
+          } else {
+            responseForCache = response
+          }
         },
         processResponseEndOfBody (response) {
           // 1.
@@ -193,7 +228,7 @@ class Cache {
           }
 
           // 2.
-          responsePromise.resolve(response)
+          responsePromise.resolve(responseForCache ?? response)
         }
       }))
 

test/web/cache-add-body.js

@@ -0,0 +1,67 @@
+'use strict'
+
+/**
+ * Regression for https://github.com/nodejs/undici/issues/5615
+ * Cache.add / Cache.addAll hung forever when the response had a body
+ * because processResponseEndOfBody only ran after the pull-driven body
+ * stream finished, and nothing pulled it.
+ */
+
+const { test } = require('node:test')
+const { createServer } = require('node:http')
+const { once } = require('node:events')
+const { caches } = require('../..')
+
+test('cache.add settles for a 200 response with a body', async (t) => {
+  const server = createServer((req, res) => {
+    res.writeHead(200, { 'content-type': 'text/plain' })
+    res.end('hello')
+  })
+  t.after(() => server.close())
+  server.listen(0, '127.0.0.1')
+  await once(server, 'listening')
+
+  const base = `http://127.0.0.1:${server.address().port}`
+  const cache = await caches.open('cache-add-body')
+  t.after(async () => {
+    await caches.delete('cache-add-body')
+  })
+
+  await Promise.race([
+    cache.add(`${base}/`),
+    new Promise((_, reject) =>
+      setTimeout(() => reject(new Error('cache.add hung for body-bearing response')), 3000)
+    )
+  ])
+
+  const keys = await cache.keys()
+  t.assert.strictEqual(keys.length, 1)
+  t.assert.strictEqual(keys[0].url, `${base}/`)
+
+  const match = await cache.match(`${base}/`)
+  t.assert.ok(match)
+  t.assert.strictEqual(await match.text(), 'hello')
+})
+
+test('cache.add settles for a 204 response (no body)', async (t) => {
+  const server = createServer((req, res) => {
+    res.writeHead(204)
+    res.end()
+  })
+  t.after(() => server.close())
+  server.listen(0, '127.0.0.1')
+  await once(server, 'listening')
+
+  const base = `http://127.0.0.1:${server.address().port}`
+  const cache = await caches.open('cache-add-empty')
+  t.after(async () => {
+    await caches.delete('cache-add-empty')
+  })
+
+  await Promise.race([
+    cache.add(`${base}/`),
+    new Promise((_, reject) =>
+      setTimeout(() => reject(new Error('cache.add hung for empty response')), 3000)
+    )
+  ])
+})