From 921d55e565e6098bfe2d8021a5a333ca2dd63e14 Mon Sep 17 00:00:00 2001
From: marcopiraccini <marco.piraccini@gmail.com>
Date: Thu, 10 Sep 2026 10:33:08 +0200
Subject: [PATCH 1/7] fix(cache): deliver a 304's start before its end with an
 async store

---
 lib/handler/cache-handler.js           | 131 ++++++++++++++++---------
 test/interceptors/cache-async-store.js | 109 +++++++++++++++++++-
 2 files changed, 191 insertions(+), 49 deletions(-)

diff --git a/lib/handler/cache-handler.js b/lib/handler/cache-handler.js
index 2ec550b431a..586fd40abba 100644
--- a/lib/handler/cache-handler.js
+++ b/lib/handler/cache-handler.js
@@ -149,6 +149,12 @@ class CacheHandler {
    */
   #handler
 
+  /**
+   * Pending 304 resolution (async store lookup or cached body replay).
+   * @type {Promise<void> | null}
+   */
+  #pending304 = null
+
   /**
    * @type {import('node:stream').Writable | undefined}
    */
@@ -345,59 +351,67 @@ class CacheHandler {
         }
 
         if (typeof cachedValue.body.values === 'function') {
-          const bodyIterator = cachedValue.body.values()
-
-          const streamCachedBody = () => {
-            for (const chunk of bodyIterator) {
-              const full = this.#writeStream.write(chunk) === false
-              this.#handler.onResponseData?.(controller, chunk)
-              // when stream is full stop writing until we get a 'drain' event
-              if (full) {
-                break
+          return new Promise((resolve) => {
+            const bodyIterator = cachedValue.body.values()
+
+            const streamCachedBody = () => {
+              for (const chunk of bodyIterator) {
+                const full = this.#writeStream?.write(chunk) === false
+                this.#handler.onResponseData?.(controller, chunk)
+                // when stream is full stop writing until we get a 'drain' event
+                if (full) {
+                  return
+                }
               }
+              resolve()
             }
-          }
 
-          this.#writeStream
-            .on('error', function () {
-              handler.#writeStream = undefined
-              handler.#store.delete(handler.#cacheKey)
-            })
-            .on('drain', () => {
-              streamCachedBody()
-            })
-            .on('close', function () {
-              if (handler.#writeStream === this) {
+            this.#writeStream
+              .on('error', function () {
                 handler.#writeStream = undefined
-              }
-            })
-
-          streamCachedBody()
+                handler.#store.delete(handler.#cacheKey)
+                resolve()
+              })
+              .on('drain', () => {
+                streamCachedBody()
+              })
+              .on('close', function () {
+                if (handler.#writeStream === this) {
+                  handler.#writeStream = undefined
+                }
+              })
+
+            streamCachedBody()
+          })
         } else if (typeof cachedValue.body.on === 'function') {
           // Readable stream body (e.g. from async/remote cache stores)
-          cachedValue.body
-            .on('data', (chunk) => {
-              this.#writeStream.write(chunk)
-              this.#handler.onResponseData?.(controller, chunk)
-            })
-            .on('end', () => {
-              this.#writeStream.end()
-            })
-            .on('error', () => {
-              this.#writeStream = undefined
-              this.#store.delete(this.#cacheKey)
-            })
-
-          this.#writeStream
-            .on('error', function () {
-              handler.#writeStream = undefined
-              handler.#store.delete(handler.#cacheKey)
-            })
-            .on('close', function () {
-              if (handler.#writeStream === this) {
+          return new Promise((resolve) => {
+            cachedValue.body
+              .on('data', (chunk) => {
+                this.#writeStream?.write(chunk)
+                this.#handler.onResponseData?.(controller, chunk)
+              })
+              .on('end', () => {
+                this.#writeStream?.end()
+                resolve()
+              })
+              .on('error', () => {
+                this.#writeStream = undefined
+                this.#store.delete(this.#cacheKey)
+                resolve()
+              })
+
+            this.#writeStream
+              .on('error', function () {
                 handler.#writeStream = undefined
-              }
-            })
+                handler.#store.delete(handler.#cacheKey)
+              })
+              .on('close', function () {
+                if (handler.#writeStream === this) {
+                  handler.#writeStream = undefined
+                }
+              })
+          })
         }
       }
 
@@ -406,9 +420,15 @@ class CacheHandler {
        */
       const result = this.#store.get(this.#cacheKey)
       if (result && typeof result.then === 'function') {
-        result.then(handle304)
+        // The 304 ends before the lookup settles; hold the end until then.
+        this.#pending304 = result
+          .then(handle304, () => handle304(undefined))
+          .then(() => { this.#pending304 = null })
       } else {
-        handle304(result)
+        const replay = handle304(result)
+        if (replay) {
+          this.#pending304 = replay.then(() => { this.#pending304 = null })
+        }
       }
     } else {
       if (typeof resHeaders.etag === 'string' && isEtagUsable(resHeaders.etag)) {
@@ -445,6 +465,11 @@ class CacheHandler {
   }
 
   onResponseData (controller, chunk) {
+    if (this.#pending304) {
+      this.#pending304 = this.#pending304.then(() => this.onResponseData(controller, chunk))
+      return
+    }
+
     if (this.#writeStream?.write(chunk) === false) {
       controller.pause()
     }
@@ -453,11 +478,21 @@ class CacheHandler {
   }
 
   onResponseEnd (controller, trailers) {
+    if (this.#pending304) {
+      this.#pending304 = this.#pending304.then(() => this.onResponseEnd(controller, trailers))
+      return
+    }
+
     this.#writeStream?.end()
     this.#handler.onResponseEnd?.(controller, trailers)
   }
 
   onResponseError (controller, err) {
+    if (this.#pending304) {
+      this.#pending304 = this.#pending304.then(() => this.onResponseError(controller, err))
+      return
+    }
+
     this.#writeStream?.destroy(err)
     this.#writeStream = undefined
     this.#handler.onResponseError?.(controller, err)
diff --git a/test/interceptors/cache-async-store.js b/test/interceptors/cache-async-store.js
index f5bab853c2d..5b168319a82 100644
--- a/test/interceptors/cache-async-store.js
+++ b/test/interceptors/cache-async-store.js
@@ -5,7 +5,7 @@ const { strictEqual, notStrictEqual } = require('node:assert')
 const { createServer } = require('node:http')
 const { once } = require('node:events')
 const { Readable } = require('node:stream')
-const { request, Client, interceptors } = require('../../index')
+const { request, Client, Dispatcher, interceptors } = require('../../index')
 const MemoryCacheStore = require('../../lib/cache/memory-cache-store')
 const FakeTimers = require('@sinonjs/fake-timers')
 const { setTimeout } = require('node:timers/promises')
@@ -48,6 +48,113 @@ class AsyncCacheStore {
 }
 
 describe('cache interceptor with async store', () => {
+  // Delivers start, data and end synchronously inside dispatch(), so an
+  // empty 304 ends before an async store lookup settles.
+  class SyncDispatcher extends Dispatcher {
+    requests = 0
+
+    dispatch (opts, handler) {
+      this.requests++
+      const controller = {
+        paused: false,
+        aborted: false,
+        reason: null,
+        pause () {},
+        resume () {},
+        abort () {}
+      }
+      handler.onRequestStart?.(controller, {})
+      if (opts.headers?.['if-none-match'] === '"abc"') {
+        // Without cache-control the 304 is passed through untouched.
+        handler.onResponseStart?.(controller, 304, { etag: '"abc"', 'cache-control': 'public, max-age=60' }, 'Not Modified')
+        handler.onResponseEnd?.(controller, {})
+        return true
+      }
+      const body = Buffer.from('cached body')
+      handler.onResponseStart?.(controller, 200, {
+        'cache-control': 'public, max-age=60',
+        etag: '"abc"',
+        'content-length': String(body.length)
+      }, 'OK')
+      handler.onResponseData?.(controller, body)
+      handler.onResponseEnd?.(controller, {})
+      return true
+    }
+  }
+
+  test('a 304 to a conditional request that missed the cache reaches the client intact', async () => {
+    const store = new AsyncCacheStore()
+    const client = new SyncDispatcher().compose(interceptors.cache({ store }))
+
+    const response = await client.request({
+      origin: 'http://localhost',
+      method: 'GET',
+      path: '/',
+      headers: { 'if-none-match': '"abc"' }
+    })
+    strictEqual(response.statusCode, 304)
+    strictEqual(await response.body.text(), '')
+  })
+
+  // Misses on the interceptor's lookup and hits on the lookup CacheHandler
+  // makes after the 304, so handle304 runs with a cached value to replay.
+  class MissThenHitStore {
+    #inner = new MemoryCacheStore()
+    #asStream
+    misses = 0
+
+    constructor ({ asStream = false } = {}) {
+      this.#asStream = asStream
+    }
+
+    async get (key) {
+      if (this.misses > 0) {
+        this.misses--
+        return undefined
+      }
+      const result = this.#inner.get(key)
+      if (!result || !this.#asStream) return result
+      const { body, ...rest } = result
+      return { ...rest, body: Readable.from(body ?? []) }
+    }
+
+    createWriteStream (key, value) {
+      return this.#inner.createWriteStream(key, value)
+    }
+
+    delete (key) {
+      return this.#inner.delete(key)
+    }
+  }
+
+  for (const [name, asStream] of [['an array of Buffers', false], ['a Readable', true]]) {
+    test(`a 304 resolved against an async store replays a cached body that is ${name} before the end`, async () => {
+      const store = new MissThenHitStore({ asStream })
+      const dispatcher = new SyncDispatcher()
+      const client = dispatcher.compose(interceptors.cache({ store }))
+
+      {
+        const response = await client.request({ origin: 'http://localhost', method: 'GET', path: '/' })
+        strictEqual(response.statusCode, 200)
+        strictEqual(await response.body.text(), 'cached body')
+      }
+
+      // The origin's 304 goes downstream, followed by the cached body, then the end.
+      store.misses = 1
+      {
+        const response = await client.request({
+          origin: 'http://localhost',
+          method: 'GET',
+          path: '/',
+          headers: { 'if-none-match': '"abc"' }
+        })
+        strictEqual(response.statusCode, 304)
+        strictEqual(await response.body.text(), 'cached body')
+      }
+      strictEqual(dispatcher.requests, 2)
+    })
+  }
+
   test('stale-while-revalidate 304 refreshes cache with async store', async () => {
     const clock = FakeTimers.install({ now: 1 })
     after(() => clock.uninstall())

From bfeed00fd7b9499a0b541dbf5af2245ce97dd20f Mon Sep 17 00:00:00 2001
From: marcopiraccini <marco.piraccini@gmail.com>
Date: Thu, 10 Sep 2026 11:08:13 +0200
Subject: [PATCH 2/7] coverage and fixup

---
 lib/handler/cache-handler.js           |   3 +-
 test/interceptors/cache-async-store.js | 106 +++++++++++++++++++++++--
 2 files changed, 101 insertions(+), 8 deletions(-)

diff --git a/lib/handler/cache-handler.js b/lib/handler/cache-handler.js
index 586fd40abba..16fb1be0b8c 100644
--- a/lib/handler/cache-handler.js
+++ b/lib/handler/cache-handler.js
@@ -370,7 +370,8 @@ class CacheHandler {
               .on('error', function () {
                 handler.#writeStream = undefined
                 handler.#store.delete(handler.#cacheKey)
-                resolve()
+                // Keep replaying downstream without the store
+                streamCachedBody()
               })
               .on('drain', () => {
                 streamCachedBody()
diff --git a/test/interceptors/cache-async-store.js b/test/interceptors/cache-async-store.js
index 5b168319a82..7ada2f248b2 100644
--- a/test/interceptors/cache-async-store.js
+++ b/test/interceptors/cache-async-store.js
@@ -1,10 +1,10 @@
 'use strict'
 
 const { test, after, describe } = require('node:test')
-const { strictEqual, notStrictEqual } = require('node:assert')
+const { strictEqual, notStrictEqual, rejects } = require('node:assert')
 const { createServer } = require('node:http')
 const { once } = require('node:events')
-const { Readable } = require('node:stream')
+const { Readable, Writable } = require('node:stream')
 const { request, Client, Dispatcher, interceptors } = require('../../index')
 const MemoryCacheStore = require('../../lib/cache/memory-cache-store')
 const FakeTimers = require('@sinonjs/fake-timers')
@@ -53,6 +53,12 @@ describe('cache interceptor with async store', () => {
   class SyncDispatcher extends Dispatcher {
     requests = 0
 
+    constructor ({ dataOn304, errorOn304 } = {}) {
+      super()
+      this.dataOn304 = dataOn304
+      this.errorOn304 = errorOn304
+    }
+
     dispatch (opts, handler) {
       this.requests++
       const controller = {
@@ -67,16 +73,23 @@ describe('cache interceptor with async store', () => {
       if (opts.headers?.['if-none-match'] === '"abc"') {
         // Without cache-control the 304 is passed through untouched.
         handler.onResponseStart?.(controller, 304, { etag: '"abc"', 'cache-control': 'public, max-age=60' }, 'Not Modified')
+        if (this.errorOn304) {
+          handler.onResponseError?.(controller, this.errorOn304)
+          return true
+        }
+        if (this.dataOn304) {
+          handler.onResponseData?.(controller, this.dataOn304)
+        }
         handler.onResponseEnd?.(controller, {})
         return true
       }
-      const body = Buffer.from('cached body')
       handler.onResponseStart?.(controller, 200, {
         'cache-control': 'public, max-age=60',
         etag: '"abc"',
-        'content-length': String(body.length)
+        'content-length': '11'
       }, 'OK')
-      handler.onResponseData?.(controller, body)
+      handler.onResponseData?.(controller, Buffer.from('cached '))
+      handler.onResponseData?.(controller, Buffer.from('body'))
       handler.onResponseEnd?.(controller, {})
       return true
     }
@@ -101,10 +114,15 @@ describe('cache interceptor with async store', () => {
   class MissThenHitStore {
     #inner = new MemoryCacheStore()
     #asStream
+    #bodyError
     misses = 0
+    deletes = 0
+    // 'slow' or 'error': the write stream handed out from now on
+    writeStream = null
 
-    constructor ({ asStream = false } = {}) {
+    constructor ({ asStream = false, bodyError = null } = {}) {
       this.#asStream = asStream
+      this.#bodyError = bodyError
     }
 
     async get (key) {
@@ -115,14 +133,29 @@ describe('cache interceptor with async store', () => {
       const result = this.#inner.get(key)
       if (!result || !this.#asStream) return result
       const { body, ...rest } = result
-      return { ...rest, body: Readable.from(body ?? []) }
+      const bodyError = this.#bodyError
+      const readable = Readable.from(body ?? [])
+      if (bodyError) {
+        readable.push = readable.push.bind(readable)
+        readable.once('data', () => readable.destroy(bodyError))
+      }
+      return { ...rest, body: readable }
     }
 
     createWriteStream (key, value) {
+      if (this.writeStream === 'slow') {
+        // Each write exceeds the high water mark: write() returns false and
+        // the replay has to wait for 'drain'.
+        return new Writable({ highWaterMark: 1, write (chunk, encoding, callback) { setImmediate(callback) } })
+      }
+      if (this.writeStream === 'error') {
+        return new Writable({ write (chunk, encoding, callback) { callback(new Error('write failed')) } })
+      }
       return this.#inner.createWriteStream(key, value)
     }
 
     delete (key) {
+      this.deletes++
       return this.#inner.delete(key)
     }
   }
@@ -155,6 +188,65 @@ describe('cache interceptor with async store', () => {
     })
   }
 
+  test('the replay of an array body waits for the write stream to drain', async () => {
+    const store = new MissThenHitStore()
+    const client = new SyncDispatcher().compose(interceptors.cache({ store }))
+    await (await client.request({ origin: 'http://localhost', method: 'GET', path: '/' })).body.text()
+
+    store.misses = 1
+    store.writeStream = 'slow'
+    const response = await client.request({ origin: 'http://localhost', method: 'GET', path: '/', headers: { 'if-none-match': '"abc"' } })
+    strictEqual(response.statusCode, 304)
+    strictEqual(await response.body.text(), 'cached body')
+  })
+
+  for (const [name, asStream] of [['an array of Buffers', false], ['a Readable', true]]) {
+    test(`a failing write stream during the replay of ${name} drops the entry and still ends the response`, async () => {
+      const store = new MissThenHitStore({ asStream })
+      const client = new SyncDispatcher().compose(interceptors.cache({ store }))
+      await (await client.request({ origin: 'http://localhost', method: 'GET', path: '/' })).body.text()
+
+      store.misses = 1
+      store.writeStream = 'error'
+      const response = await client.request({ origin: 'http://localhost', method: 'GET', path: '/', headers: { 'if-none-match': '"abc"' } })
+      strictEqual(response.statusCode, 304)
+      strictEqual(await response.body.text(), 'cached body')
+      strictEqual(store.deletes, 1)
+    })
+  }
+
+  test('a cached body stream that errors during the replay drops the entry and still ends the response', async () => {
+    const store = new MissThenHitStore({ asStream: true, bodyError: new Error('body failed') })
+    const client = new SyncDispatcher().compose(interceptors.cache({ store }))
+    await (await client.request({ origin: 'http://localhost', method: 'GET', path: '/' })).body.text()
+
+    store.misses = 1
+    const response = await client.request({ origin: 'http://localhost', method: 'GET', path: '/', headers: { 'if-none-match': '"abc"' } })
+    strictEqual(response.statusCode, 304)
+    await response.body.text()
+    strictEqual(store.deletes, 1)
+  })
+
+  test('data from the origin after a pending 304 is delivered after the cached body', async () => {
+    const store = new MissThenHitStore()
+    const client = new SyncDispatcher({ dataOn304: Buffer.from('+extra') }).compose(interceptors.cache({ store }))
+    await (await client.request({ origin: 'http://localhost', method: 'GET', path: '/' })).body.text()
+
+    store.misses = 1
+    const response = await client.request({ origin: 'http://localhost', method: 'GET', path: '/', headers: { 'if-none-match': '"abc"' } })
+    strictEqual(response.statusCode, 304)
+    strictEqual(await response.body.text(), 'cached body+extra')
+  })
+
+  test('an error from the origin after a pending 304 is delivered after the start', async () => {
+    const store = new AsyncCacheStore()
+    const client = new SyncDispatcher({ errorOn304: new Error('origin failed') }).compose(interceptors.cache({ store }))
+
+    const response = await client.request({ origin: 'http://localhost', method: 'GET', path: '/', headers: { 'if-none-match': '"abc"' } })
+    strictEqual(response.statusCode, 304)
+    await rejects(response.body.text(), { message: 'origin failed' })
+  })
+
   test('stale-while-revalidate 304 refreshes cache with async store', async () => {
     const clock = FakeTimers.install({ now: 1 })
     after(() => clock.uninstall())

From bda57a0b56909ecc65e48388f4d1b4878c90a187 Mon Sep 17 00:00:00 2001
From: marcopiraccini <marco.piraccini@gmail.com>
Date: Wed, 16 Sep 2026 15:33:18 +0200
Subject: [PATCH 3/7] fixup: removed chain of promises

Signed-off-by: marcopiraccini <marco.piraccini@gmail.com>
---
 lib/handler/cache-handler.js           | 218 +++++++++++++++----------
 test/interceptors/cache-async-store.js | 175 +++++++++++++++++---
 2 files changed, 280 insertions(+), 113 deletions(-)

diff --git a/lib/handler/cache-handler.js b/lib/handler/cache-handler.js
index 16fb1be0b8c..74d52664194 100644
--- a/lib/handler/cache-handler.js
+++ b/lib/handler/cache-handler.js
@@ -13,6 +13,50 @@ const { parseHttpDate } = require('../util/date.js')
 
 function noop () {}
 
+class CacheController {
+  #blocked = false
+  #downstreamPaused = false
+
+  constructor (target) {
+    this.target = target
+  }
+
+  block () {
+    this.#blocked = true
+  }
+
+  release () {
+    this.#blocked = false
+    if (!this.#downstreamPaused) {
+      this.target.resume()
+    }
+  }
+
+  pause () {
+    this.#downstreamPaused = true
+    this.target.pause()
+  }
+
+  resume () {
+    this.#downstreamPaused = false
+    if (!this.#blocked) {
+      this.target.resume()
+    }
+  }
+
+  abort (reason) {
+    this.target.abort(reason)
+  }
+
+  get paused () { return this.#downstreamPaused }
+  get aborted () { return this.target.aborted }
+  get reason () { return this.target.reason }
+  get rawHeaders () { return this.target.rawHeaders }
+  set rawHeaders (value) { this.target.rawHeaders = value }
+  get rawTrailers () { return this.target.rawTrailers }
+  set rawTrailers (value) { this.target.rawTrailers = value }
+}
+
 // Status codes that we can use some heuristics on to cache
 const HEURISTICALLY_CACHEABLE_STATUS_CODES = [
   200, 203, 204, 206, 300, 301, 308, 404, 405, 410, 414, 501
@@ -150,10 +194,9 @@ class CacheHandler {
   #handler
 
   /**
-   * Pending 304 resolution (async store lookup or cached body replay).
-   * @type {Promise<void> | null}
+   * @type {CacheController | undefined}
    */
-  #pending304 = null
+  #downstreamController
 
   /**
    * @type {import('node:stream').Writable | undefined}
@@ -176,7 +219,8 @@ class CacheHandler {
   onRequestStart (controller, context) {
     this.#writeStream?.destroy()
     this.#writeStream = undefined
-    this.#handler.onRequestStart?.(controller, context)
+    this.#downstreamController = new CacheController(controller)
+    this.#handler.onRequestStart?.(this.#downstreamController, context)
   }
 
   onBodySent (chunk) {
@@ -187,8 +231,8 @@ class CacheHandler {
     this.#handler.onRequestSent?.()
   }
 
-  onRequestUpgrade (controller, statusCode, headers, socket) {
-    this.#handler.onRequestUpgrade?.(controller, statusCode, headers, socket)
+  onRequestUpgrade (_controller, statusCode, headers, socket) {
+    this.#handler.onRequestUpgrade?.(this.#downstreamController, statusCode, headers, socket)
   }
 
   /**
@@ -205,7 +249,7 @@ class CacheHandler {
   ) {
     const downstreamOnHeaders = () =>
       this.#handler.onResponseStart?.(
-        controller,
+        this.#downstreamController,
         statusCode,
         resHeaders,
         statusMessage
@@ -329,10 +373,21 @@ class CacheHandler {
     // Not modified, re-use the cached value
     // https://www.rfc-editor.org/rfc/rfc9111.html#name-handling-304-not-modified
     if (statusCode === 304) {
+      let done = false
+      const finish304 = () => {
+        if (done) {
+          return
+        }
+        done = true
+        this.#downstreamController.release()
+      }
+
       const handle304 = (cachedValue) => {
         if (!cachedValue) {
           // Do not create a new cache entry, as a 304 won't have a body - so cannot be cached.
-          return downstreamOnHeaders()
+          downstreamOnHeaders()
+          finish304()
+          return
         }
 
         // Re-use the cached value: statuscode, statusmessage, headers and body
@@ -347,72 +402,76 @@ class CacheHandler {
         this.#writeStream = this.#store.createWriteStream(this.#cacheKey, value)
 
         if (!this.#writeStream || !cachedValue?.body) {
+          finish304()
           return
         }
 
         if (typeof cachedValue.body.values === 'function') {
-          return new Promise((resolve) => {
-            const bodyIterator = cachedValue.body.values()
-
-            const streamCachedBody = () => {
-              for (const chunk of bodyIterator) {
-                const full = this.#writeStream?.write(chunk) === false
-                this.#handler.onResponseData?.(controller, chunk)
-                // when stream is full stop writing until we get a 'drain' event
-                if (full) {
-                  return
-                }
+          const bodyIterator = cachedValue.body.values()
+
+          const streamCachedBody = () => {
+            if (done) {
+              return
+            }
+            for (const chunk of bodyIterator) {
+              const full = this.#writeStream?.write(chunk) === false
+              this.#handler.onResponseData?.(this.#downstreamController, chunk)
+              // when stream is full stop writing until we get a 'drain' event
+              if (full) {
+                return
               }
-              resolve()
             }
+            finish304()
+          }
 
-            this.#writeStream
-              .on('error', function () {
+          this.#writeStream
+            .on('error', function () {
+              handler.#writeStream = undefined
+              handler.#store.delete(handler.#cacheKey)
+              // Keep replaying downstream without the store
+              streamCachedBody()
+            })
+            .on('drain', () => {
+              streamCachedBody()
+            })
+            .on('close', function () {
+              if (handler.#writeStream === this) {
                 handler.#writeStream = undefined
-                handler.#store.delete(handler.#cacheKey)
                 // Keep replaying downstream without the store
                 streamCachedBody()
-              })
-              .on('drain', () => {
-                streamCachedBody()
-              })
-              .on('close', function () {
-                if (handler.#writeStream === this) {
-                  handler.#writeStream = undefined
-                }
-              })
-
-            streamCachedBody()
-          })
+              }
+            })
+
+          streamCachedBody()
         } else if (typeof cachedValue.body.on === 'function') {
           // Readable stream body (e.g. from async/remote cache stores)
-          return new Promise((resolve) => {
-            cachedValue.body
-              .on('data', (chunk) => {
-                this.#writeStream?.write(chunk)
-                this.#handler.onResponseData?.(controller, chunk)
-              })
-              .on('end', () => {
-                this.#writeStream?.end()
-                resolve()
-              })
-              .on('error', () => {
-                this.#writeStream = undefined
-                this.#store.delete(this.#cacheKey)
-                resolve()
-              })
-
-            this.#writeStream
-              .on('error', function () {
+          cachedValue.body
+            .on('data', (chunk) => {
+              this.#writeStream?.write(chunk)
+              this.#handler.onResponseData?.(this.#downstreamController, chunk)
+            })
+            .on('end', () => {
+              this.#writeStream?.end()
+              finish304()
+            })
+            .on('error', () => {
+              this.#writeStream = undefined
+              this.#store.delete(this.#cacheKey)
+              finish304()
+            })
+
+          this.#writeStream
+            .on('error', function () {
+              handler.#writeStream = undefined
+              handler.#store.delete(handler.#cacheKey)
+            })
+            .on('close', function () {
+              if (handler.#writeStream === this) {
                 handler.#writeStream = undefined
-                handler.#store.delete(handler.#cacheKey)
-              })
-              .on('close', function () {
-                if (handler.#writeStream === this) {
-                  handler.#writeStream = undefined
-                }
-              })
-          })
+              }
+            })
+        } else {
+          finish304()
         }
       }
 
@@ -420,16 +479,16 @@ class CacheHandler {
        * @type {import('../../types/cache-interceptor.d.ts').default.CacheValue}
        */
       const result = this.#store.get(this.#cacheKey)
+      // A 304 ends before the cached body is replayed. Keep the origin
+      // response paused until replay is complete.
+      this.#downstreamController.block()
+      controller.pause()
       if (result && typeof result.then === 'function') {
-        // The 304 ends before the lookup settles; hold the end until then.
-        this.#pending304 = result
-          .then(handle304, () => handle304(undefined))
-          .then(() => { this.#pending304 = null })
+        result.then(handle304, () => handle304(undefined)).catch((err) => {
+          controller.abort(err)
+        })
       } else {
-        const replay = handle304(result)
-        if (replay) {
-          this.#pending304 = replay.then(() => { this.#pending304 = null })
-        }
+        handle304(result)
       }
     } else {
       if (typeof resHeaders.etag === 'string' && isEtagUsable(resHeaders.etag)) {
@@ -466,37 +525,22 @@ class CacheHandler {
   }
 
   onResponseData (controller, chunk) {
-    if (this.#pending304) {
-      this.#pending304 = this.#pending304.then(() => this.onResponseData(controller, chunk))
-      return
-    }
-
     if (this.#writeStream?.write(chunk) === false) {
       controller.pause()
     }
 
-    this.#handler.onResponseData?.(controller, chunk)
+    this.#handler.onResponseData?.(this.#downstreamController, chunk)
   }
 
   onResponseEnd (controller, trailers) {
-    if (this.#pending304) {
-      this.#pending304 = this.#pending304.then(() => this.onResponseEnd(controller, trailers))
-      return
-    }
-
     this.#writeStream?.end()
-    this.#handler.onResponseEnd?.(controller, trailers)
+    this.#handler.onResponseEnd?.(this.#downstreamController, trailers)
   }
 
   onResponseError (controller, err) {
-    if (this.#pending304) {
-      this.#pending304 = this.#pending304.then(() => this.onResponseError(controller, err))
-      return
-    }
-
     this.#writeStream?.destroy(err)
     this.#writeStream = undefined
-    this.#handler.onResponseError?.(controller, err)
+    this.#handler.onResponseError?.(this.#downstreamController, err)
   }
 }
 
diff --git a/test/interceptors/cache-async-store.js b/test/interceptors/cache-async-store.js
index 7ada2f248b2..a12e0ebdfb4 100644
--- a/test/interceptors/cache-async-store.js
+++ b/test/interceptors/cache-async-store.js
@@ -48,8 +48,8 @@ class AsyncCacheStore {
 }
 
 describe('cache interceptor with async store', () => {
-  // Delivers start, data and end synchronously inside dispatch(), so an
-  // empty 304 ends before an async store lookup settles.
+  // Respects dispatcher backpressure, including when an async 304 lookup
+  // pauses the response before its end is delivered.
   class SyncDispatcher extends Dispatcher {
     requests = 0
 
@@ -65,22 +65,43 @@ describe('cache interceptor with async store', () => {
         paused: false,
         aborted: false,
         reason: null,
-        pause () {},
-        resume () {},
-        abort () {}
+        resumeCallback: null,
+        pause () {
+          this.paused = true
+        },
+        resume () {
+          if (!this.paused) return
+          this.paused = false
+          const callback = this.resumeCallback
+          this.resumeCallback = null
+          callback?.()
+        },
+        abort (reason) {
+          if (this.aborted) return
+          this.aborted = true
+          this.reason = reason
+          this.resumeCallback = null
+          handler.onResponseError?.(this, reason)
+        }
       }
       handler.onRequestStart?.(controller, {})
       if (opts.headers?.['if-none-match'] === '"abc"') {
-        // Without cache-control the 304 is passed through untouched.
         handler.onResponseStart?.(controller, 304, { etag: '"abc"', 'cache-control': 'public, max-age=60' }, 'Not Modified')
-        if (this.errorOn304) {
-          handler.onResponseError?.(controller, this.errorOn304)
-          return true
+        const onResponseEnd = () => {
+          if (this.errorOn304) {
+            handler.onResponseError?.(controller, this.errorOn304)
+            return
+          }
+          if (this.dataOn304) {
+            handler.onResponseData?.(controller, this.dataOn304)
+          }
+          handler.onResponseEnd?.(controller, {})
         }
-        if (this.dataOn304) {
-          handler.onResponseData?.(controller, this.dataOn304)
+        if (controller.paused) {
+          controller.resumeCallback = onResponseEnd
+        } else {
+          onResponseEnd()
         }
-        handler.onResponseEnd?.(controller, {})
         return true
       }
       handler.onResponseStart?.(controller, 200, {
@@ -113,33 +134,40 @@ describe('cache interceptor with async store', () => {
   // makes after the 304, so handle304 runs with a cached value to replay.
   class MissThenHitStore {
     #inner = new MemoryCacheStore()
+    #asyncGet
     #asStream
     #bodyError
     misses = 0
     deletes = 0
-    // 'slow' or 'error': the write stream handed out from now on
+    // 'slow', 'error', 'close' or 'throw': the write stream handed out from now on
     writeStream = null
 
-    constructor ({ asStream = false, bodyError = null } = {}) {
+    constructor ({ asyncGet = true, asStream = false, bodyError = null } = {}) {
+      this.#asyncGet = asyncGet
       this.#asStream = asStream
       this.#bodyError = bodyError
     }
 
-    async get (key) {
+    get (key) {
+      let value
       if (this.misses > 0) {
         this.misses--
-        return undefined
-      }
-      const result = this.#inner.get(key)
-      if (!result || !this.#asStream) return result
-      const { body, ...rest } = result
-      const bodyError = this.#bodyError
-      const readable = Readable.from(body ?? [])
-      if (bodyError) {
-        readable.push = readable.push.bind(readable)
-        readable.once('data', () => readable.destroy(bodyError))
+      } else {
+        const result = this.#inner.get(key)
+        if (!result || !this.#asStream) {
+          value = result
+        } else {
+          const { body, ...rest } = result
+          const bodyError = this.#bodyError
+          const readable = Readable.from(body ?? [])
+          if (bodyError) {
+            readable.push = readable.push.bind(readable)
+            readable.once('data', () => readable.destroy(bodyError))
+          }
+          value = { ...rest, body: readable }
+        }
       }
-      return { ...rest, body: readable }
+      return this.#asyncGet ? Promise.resolve(value) : value
     }
 
     createWriteStream (key, value) {
@@ -151,6 +179,12 @@ describe('cache interceptor with async store', () => {
       if (this.writeStream === 'error') {
         return new Writable({ write (chunk, encoding, callback) { callback(new Error('write failed')) } })
       }
+      if (this.writeStream === 'close') {
+        return new Writable({ highWaterMark: 1, write (chunk, encoding, callback) { this.destroy(); callback() } })
+      }
+      if (this.writeStream === 'throw') {
+        throw new Error('write failed')
+      }
       return this.#inner.createWriteStream(key, value)
     }
 
@@ -200,6 +234,95 @@ describe('cache interceptor with async store', () => {
     strictEqual(await response.body.text(), 'cached body')
   })
 
+  test('a synchronous store pauses the 304 until cached body replay completes', async () => {
+    const store = new MissThenHitStore({ asyncGet: false })
+    const client = new SyncDispatcher().compose(interceptors.cache({ store }))
+    await (await client.request({ origin: 'http://localhost', method: 'GET', path: '/' })).body.text()
+
+    store.misses = 1
+    store.writeStream = 'slow'
+    const response = await client.request({ origin: 'http://localhost', method: 'GET', path: '/', headers: { 'if-none-match': '"abc"' } })
+    strictEqual(response.statusCode, 304)
+    strictEqual(await response.body.text(), 'cached body')
+  })
+
+  test('a downstream pause during replay holds the 304 end until it resumes', async () => {
+    const store = new MissThenHitStore()
+    const client = new SyncDispatcher().compose(interceptors.cache({ store }))
+    await (await client.request({ origin: 'http://localhost', method: 'GET', path: '/' })).body.text()
+
+    store.misses = 1
+    let controller
+    let body = ''
+    let ended = false
+    let resolvePaused
+    const paused = new Promise((resolve) => {
+      resolvePaused = resolve
+    })
+    let resolveEnd
+    const end = new Promise((resolve) => {
+      resolveEnd = resolve
+    })
+
+    client.dispatch({
+      origin: 'http://localhost',
+      method: 'GET',
+      path: '/',
+      headers: { 'if-none-match': '"abc"' }
+    }, {
+      onRequestStart () {},
+      onResponseStart () {},
+      onResponseData (downstreamController, chunk) {
+        body += chunk
+        if (!controller) {
+          controller = downstreamController
+          strictEqual(controller.paused, false)
+          controller.pause()
+          strictEqual(controller.paused, true)
+          resolvePaused()
+        }
+      },
+      onResponseEnd () {
+        ended = true
+        resolveEnd()
+      },
+      onResponseError (_, err) {
+        throw err
+      }
+    })
+
+    await paused
+    strictEqual(body, 'cached body')
+    strictEqual(ended, false)
+    controller.resume()
+    strictEqual(controller.paused, false)
+    await end
+  })
+
+  test('the replay of an array body continues if the write stream closes while full', async () => {
+    const store = new MissThenHitStore()
+    const client = new SyncDispatcher().compose(interceptors.cache({ store }))
+    await (await client.request({ origin: 'http://localhost', method: 'GET', path: '/' })).body.text()
+
+    store.misses = 1
+    store.writeStream = 'close'
+    const response = await client.request({ origin: 'http://localhost', method: 'GET', path: '/', headers: { 'if-none-match': '"abc"' } })
+    strictEqual(response.statusCode, 304)
+    strictEqual(await response.body.text(), 'cached body')
+  })
+
+  test('an error while handling an async 304 aborts the request', async () => {
+    const store = new MissThenHitStore()
+    const client = new SyncDispatcher().compose(interceptors.cache({ store }))
+    await (await client.request({ origin: 'http://localhost', method: 'GET', path: '/' })).body.text()
+
+    store.misses = 1
+    store.writeStream = 'throw'
+    const response = await client.request({ origin: 'http://localhost', method: 'GET', path: '/', headers: { 'if-none-match': '"abc"' } })
+    strictEqual(response.statusCode, 304)
+    await rejects(response.body.text(), { message: 'write failed' })
+  })
+
   for (const [name, asStream] of [['an array of Buffers', false], ['a Readable', true]]) {
     test(`a failing write stream during the replay of ${name} drops the entry and still ends the response`, async () => {
       const store = new MissThenHitStore({ asStream })

From 3a7bacd8a6c05400231d587527cb0e124e67adca Mon Sep 17 00:00:00 2001
From: marcopiraccini <marco.piraccini@gmail.com>
Date: Wed, 16 Sep 2026 15:45:39 +0200
Subject: [PATCH 4/7] fixup: removed chain of promises

Signed-off-by: marcopiraccini <marco.piraccini@gmail.com>
---
 test/interceptors/cache-async-store.js | 30 ++++++++++++++++++++++++++
 1 file changed, 30 insertions(+)

diff --git a/test/interceptors/cache-async-store.js b/test/interceptors/cache-async-store.js
index a12e0ebdfb4..b96727a5a00 100644
--- a/test/interceptors/cache-async-store.js
+++ b/test/interceptors/cache-async-store.js
@@ -130,6 +130,36 @@ describe('cache interceptor with async store', () => {
     strictEqual(await response.body.text(), '')
   })
 
+  test('a downstream handler can abort a pending 304', async () => {
+    const store = new AsyncCacheStore()
+    const client = new SyncDispatcher().compose(interceptors.cache({ store }))
+    const expected = new Error('aborted by downstream')
+
+    await new Promise((resolve, reject) => {
+      client.dispatch({
+        origin: 'http://localhost',
+        method: 'GET',
+        path: '/',
+        headers: { 'if-none-match': '"abc"' }
+      }, {
+        onRequestStart () {},
+        onResponseStart (controller) {
+          controller.abort(expected)
+        },
+        onResponseData () {
+          reject(new Error('unexpected response data'))
+        },
+        onResponseEnd () {
+          reject(new Error('unexpected response end'))
+        },
+        onResponseError (_, err) {
+          strictEqual(err, expected)
+          resolve()
+        }
+      })
+    })
+  })
+
   // Misses on the interceptor's lookup and hits on the lookup CacheHandler
   // makes after the 304, so handle304 runs with a cached value to replay.
   class MissThenHitStore {

From bb2c67766845c97411f58cc0ec0850e27793ad08 Mon Sep 17 00:00:00 2001
From: marcopiraccini <marco.piraccini@gmail.com>
Date: Wed, 16 Sep 2026 16:39:32 +0200
Subject: [PATCH 5/7] fixup: removed chain of promises

Signed-off-by: marcopiraccini <marco.piraccini@gmail.com>
---
 test/interceptors/cache-async-store.js | 61 ++++++++++++++++++++++++--
 1 file changed, 57 insertions(+), 4 deletions(-)

diff --git a/test/interceptors/cache-async-store.js b/test/interceptors/cache-async-store.js
index b96727a5a00..b92c7ed38f7 100644
--- a/test/interceptors/cache-async-store.js
+++ b/test/interceptors/cache-async-store.js
@@ -3,7 +3,7 @@
 const { test, after, describe } = require('node:test')
 const { strictEqual, notStrictEqual, rejects } = require('node:assert')
 const { createServer } = require('node:http')
-const { once } = require('node:events')
+const { once, EventEmitter } = require('node:events')
 const { Readable, Writable } = require('node:stream')
 const { request, Client, Dispatcher, interceptors } = require('../../index')
 const MemoryCacheStore = require('../../lib/cache/memory-cache-store')
@@ -167,15 +167,19 @@ describe('cache interceptor with async store', () => {
     #asyncGet
     #asStream
     #bodyError
+    #bodyErrorThenEnd
+    #unsupportedBody
     misses = 0
     deletes = 0
-    // 'slow', 'error', 'close' or 'throw': the write stream handed out from now on
+    // 'slow', 'error', 'close', 'none' or 'throw': the write stream handed out from now on
     writeStream = null
 
-    constructor ({ asyncGet = true, asStream = false, bodyError = null } = {}) {
+    constructor ({ asyncGet = true, asStream = false, bodyError = null, bodyErrorThenEnd = false, unsupportedBody = false } = {}) {
       this.#asyncGet = asyncGet
       this.#asStream = asStream
       this.#bodyError = bodyError
+      this.#bodyErrorThenEnd = bodyErrorThenEnd
+      this.#unsupportedBody = unsupportedBody
     }
 
     get (key) {
@@ -184,7 +188,18 @@ describe('cache interceptor with async store', () => {
         this.misses--
       } else {
         const result = this.#inner.get(key)
-        if (!result || !this.#asStream) {
+        if (!result) {
+          value = result
+        } else if (this.#bodyErrorThenEnd) {
+          const body = new EventEmitter()
+          setImmediate(() => {
+            body.emit('error', new Error('body failed'))
+            body.emit('end')
+          })
+          value = { ...result, body }
+        } else if (this.#unsupportedBody) {
+          value = { ...result, body: {} }
+        } else if (!this.#asStream) {
           value = result
         } else {
           const { body, ...rest } = result
@@ -212,6 +227,9 @@ describe('cache interceptor with async store', () => {
       if (this.writeStream === 'close') {
         return new Writable({ highWaterMark: 1, write (chunk, encoding, callback) { this.destroy(); callback() } })
       }
+      if (this.writeStream === 'none') {
+        return undefined
+      }
       if (this.writeStream === 'throw') {
         throw new Error('write failed')
       }
@@ -341,6 +359,41 @@ describe('cache interceptor with async store', () => {
     strictEqual(await response.body.text(), 'cached body')
   })
 
+  test('a 304 without a cache write stream ends after the lookup', async () => {
+    const store = new MissThenHitStore()
+    const client = new SyncDispatcher().compose(interceptors.cache({ store }))
+    await (await client.request({ origin: 'http://localhost', method: 'GET', path: '/' })).body.text()
+
+    store.misses = 1
+    store.writeStream = 'none'
+    const response = await client.request({ origin: 'http://localhost', method: 'GET', path: '/', headers: { 'if-none-match': '"abc"' } })
+    strictEqual(response.statusCode, 304)
+    strictEqual(await response.body.text(), '')
+  })
+
+  test('a 304 with an unsupported cached body ends after the lookup', async () => {
+    const store = new MissThenHitStore({ unsupportedBody: true })
+    const client = new SyncDispatcher().compose(interceptors.cache({ store }))
+    await (await client.request({ origin: 'http://localhost', method: 'GET', path: '/' })).body.text()
+
+    store.misses = 1
+    const response = await client.request({ origin: 'http://localhost', method: 'GET', path: '/', headers: { 'if-none-match': '"abc"' } })
+    strictEqual(response.statusCode, 304)
+    strictEqual(await response.body.text(), '')
+  })
+
+  test('a cached body error followed by end releases a 304 once', async () => {
+    const store = new MissThenHitStore({ bodyErrorThenEnd: true })
+    const client = new SyncDispatcher().compose(interceptors.cache({ store }))
+    await (await client.request({ origin: 'http://localhost', method: 'GET', path: '/' })).body.text()
+
+    store.misses = 1
+    const response = await client.request({ origin: 'http://localhost', method: 'GET', path: '/', headers: { 'if-none-match': '"abc"' } })
+    strictEqual(response.statusCode, 304)
+    strictEqual(await response.body.text(), '')
+    strictEqual(store.deletes, 1)
+  })
+
   test('an error while handling an async 304 aborts the request', async () => {
     const store = new MissThenHitStore()
     const client = new SyncDispatcher().compose(interceptors.cache({ store }))

From 11eb4b0b6de6b665a789950238da14a68d9bd198 Mon Sep 17 00:00:00 2001
From: marcopiraccini <marco.piraccini@gmail.com>
Date: Wed, 16 Sep 2026 17:19:53 +0200
Subject: [PATCH 6/7] fixup: test coverage

Signed-off-by: marcopiraccini <marco.piraccini@gmail.com>
---
 test/interceptors/cache-async-store.js | 21 +++++++++++++++++++--
 1 file changed, 19 insertions(+), 2 deletions(-)

diff --git a/test/interceptors/cache-async-store.js b/test/interceptors/cache-async-store.js
index b92c7ed38f7..d5b7929cb20 100644
--- a/test/interceptors/cache-async-store.js
+++ b/test/interceptors/cache-async-store.js
@@ -53,10 +53,11 @@ describe('cache interceptor with async store', () => {
   class SyncDispatcher extends Dispatcher {
     requests = 0
 
-    constructor ({ dataOn304, errorOn304 } = {}) {
+    constructor ({ dataOn304, errorOn304, headers304 } = {}) {
       super()
       this.dataOn304 = dataOn304
       this.errorOn304 = errorOn304
+      this.headers304 = headers304
     }
 
     dispatch (opts, handler) {
@@ -86,7 +87,7 @@ describe('cache interceptor with async store', () => {
       }
       handler.onRequestStart?.(controller, {})
       if (opts.headers?.['if-none-match'] === '"abc"') {
-        handler.onResponseStart?.(controller, 304, { etag: '"abc"', 'cache-control': 'public, max-age=60' }, 'Not Modified')
+        handler.onResponseStart?.(controller, 304, this.headers304 ?? { etag: '"abc"', 'cache-control': 'public, max-age=60' }, 'Not Modified')
         const onResponseEnd = () => {
           if (this.errorOn304) {
             handler.onResponseError?.(controller, this.errorOn304)
@@ -160,6 +161,22 @@ describe('cache interceptor with async store', () => {
     })
   })
 
+  test('a rejected asynchronous delete does not interrupt a 304 response', async () => {
+    const store = new AsyncCacheStore()
+    store.delete = () => Promise.reject(new Error('delete failed'))
+    const client = new SyncDispatcher({ headers304: { etag: '"abc"', 'cache-control': 'no-store' } })
+      .compose(interceptors.cache({ store }))
+
+    const response = await client.request({
+      origin: 'http://localhost',
+      method: 'GET',
+      path: '/',
+      headers: { 'if-none-match': '"abc"' }
+    })
+    strictEqual(response.statusCode, 304)
+    strictEqual(await response.body.text(), '')
+  })
+
   // Misses on the interceptor's lookup and hits on the lookup CacheHandler
   // makes after the 304, so handle304 runs with a cached value to replay.
   class MissThenHitStore {

From e3933b4a991d4bb1adfd2b02b1860279fa63bc58 Mon Sep 17 00:00:00 2001
From: marcopiraccini <marco.piraccini@gmail.com>
Date: Wed, 16 Sep 2026 17:45:35 +0200
Subject: [PATCH 7/7] fixup: test coverage

Signed-off-by: marcopiraccini <marco.piraccini@gmail.com>
---
 test/interceptors/cache-async-store.js | 27 ++++++++++++++++++++++++++
 1 file changed, 27 insertions(+)

diff --git a/test/interceptors/cache-async-store.js b/test/interceptors/cache-async-store.js
index d5b7929cb20..cace2446104 100644
--- a/test/interceptors/cache-async-store.js
+++ b/test/interceptors/cache-async-store.js
@@ -86,6 +86,10 @@ describe('cache interceptor with async store', () => {
         }
       }
       handler.onRequestStart?.(controller, {})
+      if (opts.upgrade) {
+        handler.onRequestUpgrade?.(controller, 101, {}, {})
+        return true
+      }
       if (opts.headers?.['if-none-match'] === '"abc"') {
         handler.onResponseStart?.(controller, 304, this.headers304 ?? { etag: '"abc"', 'cache-control': 'public, max-age=60' }, 'Not Modified')
         const onResponseEnd = () => {
@@ -161,6 +165,29 @@ describe('cache interceptor with async store', () => {
     })
   })
 
+  test('an upgrade receives the downstream controller', () => {
+    const client = new SyncDispatcher().compose(interceptors.cache({ store: new AsyncCacheStore() }))
+    let requestController
+
+    client.dispatch({
+      origin: 'http://localhost',
+      method: 'GET',
+      path: '/',
+      upgrade: 'websocket'
+    }, {
+      onRequestStart (controller) {
+        requestController = controller
+      },
+      onRequestUpgrade (controller, statusCode) {
+        strictEqual(controller, requestController)
+        strictEqual(statusCode, 101)
+      },
+      onResponseError (_, err) {
+        throw err
+      }
+    })
+  })
+
   test('a rejected asynchronous delete does not interrupt a 304 response', async () => {
     const store = new AsyncCacheStore()
     store.delete = () => Promise.reject(new Error('delete failed'))