feat(cache): compress big sqlite cache entries with zstd

docs/docs/api/CacheStore.md

@@ -163,7 +163,9 @@ added: v7.0.0
 -->
 
 Stores cached responses in a SQLite database using the [`node:sqlite`][] API.
-The constructor throws when [`node:sqlite`][] is not available.
+The constructor throws when [`node:sqlite`][] is not available. Response bodies
+larger than `compressThreshold` are stored compressed with zstd and
+decompressed when read back, saving database space.
 
 ```mjs
 import { interceptors, cacheStores, Agent, setGlobalDispatcher } from 'undici'
@@ -189,9 +191,16 @@ added: v7.0.0
   * `maxEntrySize` {number} The maximum size, in bytes, of a single response
     body. Responses whose body exceeds this value are not cached. Must not
     exceed `2000000000` (2 GB). **Default:** `2000000000` (2 GB).
+  * `compressThreshold` {number} The size, in bytes, above which a response
+    body is stored compressed with zstd. Responses with an easily compressible
+    Content-Type (e.g. json, xml, plain text) are compressed regardless of
+    size. Requires a Node version with zstd
+    support (22.19+); when unavailable, bodies are stored uncompressed.
+    **Default:** `1048576` (1 MiB).
 
-`maxCount` and `maxEntrySize` must be non-negative integers; a `TypeError` is
-thrown otherwise, or if `maxEntrySize` is greater than 2 GB.
+`maxCount`, `maxEntrySize`, and `compressThreshold` must be non-negative
+integers; a `TypeError` is thrown otherwise, or if `maxEntrySize` is greater
+than 2 GB.
 
 ### `sqliteCacheStore.close()`
 
@@ -225,7 +234,8 @@ added: v7.0.0
   See [`GetResult`](#getresult).
 
 Looks up a cached response for `key`, comparing the request method and every
-header named in the stored `vary` map.
+header named in the stored `vary` map. A body that was stored compressed is
+decompressed before being returned.
 
 ### `sqliteCacheStore.set(key, value)`
 
@@ -240,9 +250,11 @@ added: v7.0.0
 * Returns: {undefined}
 
 Writes a response into the database directly, without going through a stream. If
-the body exceeds `maxEntrySize`, nothing is stored. When an entry already exists
-for `key` it is overwritten; otherwise a new row is inserted and old entries are
-pruned to honour `maxCount`. This method is used internally by
+the body exceeds `maxEntrySize`, nothing is stored. A body larger than
+`compressThreshold`, or with an easily compressible Content-Type (e.g. json,
+xml, plain text), is compressed with zstd before being written. When an entry
+already exists for `key` it is overwritten; otherwise a new row is inserted and
+old entries are pruned to honour `maxCount`. This method is used internally by
 [`sqliteCacheStore.createWriteStream()`](#sqlitecachestorecreatewritestreamkey-value).
 
 ### `sqliteCacheStore.createWriteStream(key, value)`
@@ -260,7 +272,9 @@ added: v7.0.0
 
 Returns a {Writable} stream used to write the response body into the store. When
 the stream finishes, the buffered body is committed via
-[`sqliteCacheStore.set()`](#sqlitecachestoresetkey-value). If the body exceeds
+[`sqliteCacheStore.set()`](#sqlitecachestoresetkey-value), where bodies larger
+than `compressThreshold` or with an easily compressible Content-Type are
+compressed with zstd. If the body exceeds
 `maxEntrySize`, the stream is destroyed and nothing is stored.
 
 ### `sqliteCacheStore.delete(key)`

lib/cache/sqlite-cache-store.js

@@ -2,21 +2,30 @@
 
 const { Writable } = require('node:stream')
 const { assertCacheKey, assertCacheValue } = require('../util/cache.js')
+const { zstdCompressSync, zstdDecompressSync } = require('node:zlib')
 
 let DatabaseSync
 
-const VERSION = 3
+const VERSION = 4
 
 // 2gb
 const MAX_ENTRY_SIZE = 2 * 1000 * 1000 * 1000
 
+// 1MiB
+const COMPRESS_THRESHOLD = 1024 * 1024
+
+// zstd support landed in Node 22.19.0/23.4.0, so it may be missing even when
+// node:sqlite is available.
+const hasZstd = typeof zstdCompressSync === 'function' && typeof zstdDecompressSync === 'function'
+
 /**
  * @typedef {import('../../types/cache-interceptor.d.ts').default.CacheStore} CacheStore
  * @implements {CacheStore}
  *
  * @typedef {{
  *  id: Readonly<number>,
  *  body?: Uint8Array
+ *  compressed?: string
  *  statusCode: number
  *  statusMessage: string
  *  headers?: string
@@ -31,6 +40,7 @@ const MAX_ENTRY_SIZE = 2 * 1000 * 1000 * 1000
 module.exports = class SqliteCacheStore {
   #maxEntrySize = MAX_ENTRY_SIZE
   #maxCount = Infinity
+  #compressThreshold = COMPRESS_THRESHOLD
 
   /**
    * @type {import('node:sqlite').DatabaseSync}
@@ -107,6 +117,17 @@ module.exports = class SqliteCacheStore {
         }
         this.#maxCount = opts.maxCount
       }
+
+      if (opts.compressThreshold !== undefined) {
+        if (
+          typeof opts.compressThreshold !== 'number' ||
+          !Number.isInteger(opts.compressThreshold) ||
+          opts.compressThreshold < 0
+        ) {
+          throw new TypeError('SqliteCacheStore options.compressThreshold must be a non-negative integer')
+        }
+        this.#compressThreshold = opts.compressThreshold
+      }
     }
 
     if (!DatabaseSync) {
@@ -128,6 +149,7 @@ module.exports = class SqliteCacheStore {
 
         -- Data returned to the interceptor
         body BUF NULL,
+        compressed TEXT NULL,
         deleteAt INTEGER NOT NULL,
         statusCode INTEGER NOT NULL,
         statusMessage TEXT NOT NULL,
@@ -147,6 +169,7 @@ module.exports = class SqliteCacheStore {
       SELECT
         id,
         body,
+        compressed,
         deleteAt,
         statusCode,
         statusMessage,
@@ -167,6 +190,7 @@ module.exports = class SqliteCacheStore {
     this.#updateValueQuery = this.#db.prepare(`
       UPDATE cacheInterceptorV${VERSION} SET
         body = ?,
+        compressed = ?,
         deleteAt = ?,
         statusCode = ?,
         statusMessage = ?,
@@ -185,6 +209,7 @@ module.exports = class SqliteCacheStore {
         url,
         method,
         body,
+        compressed,
         deleteAt,
         statusCode,
         statusMessage,
@@ -194,7 +219,7 @@ module.exports = class SqliteCacheStore {
         vary,
         cachedAt,
         staleAt
-      ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+      ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
     `)
 
     this.#deleteByUrlQuery = this.#db.prepare(
@@ -237,7 +262,11 @@ module.exports = class SqliteCacheStore {
     const value = this.#findValue(key)
     return value
       ? {
-          body: value.body ? Buffer.from(value.body.buffer, value.body.byteOffset, value.body.byteLength) : undefined,
+          body: value.body
+            ? value.compressed === 'zstd'
+              ? zstdDecompressSync(Buffer.from(value.body.buffer, value.body.byteOffset, value.body.byteLength))
+              : Buffer.from(value.body.buffer, value.body.byteOffset, value.body.byteLength)
+            : undefined,
           statusCode: value.statusCode,
           statusMessage: value.statusMessage,
           headers: value.headers ? JSON.parse(value.headers) : undefined,
@@ -268,11 +297,22 @@ module.exports = class SqliteCacheStore {
       return
     }
 
+    let compressed = null
+    let storedBody = body
+    if (body && hasZstd && (size > this.#compressThreshold || isCompressibleContentType(getContentType(value.headers)))) {
+      const candidate = zstdCompressSync(body)
+      if (candidate.byteLength < body.byteLength) {
+        storedBody = candidate
+        compressed = 'zstd'
+      }
+    }
+
     const existingValue = this.#findValue(key, true)
     if (existingValue) {
       // Updating an existing response, let's overwrite it
       this.#updateValueQuery.run(
-        body,
+        storedBody,
+        compressed,
         value.deleteAt,
         value.statusCode,
         value.statusMessage,
@@ -289,7 +329,8 @@ module.exports = class SqliteCacheStore {
       this.#insertValueQuery.run(
         url,
         key.method,
-        body,
+        storedBody,
+        compressed,
         value.deleteAt,
         value.statusCode,
         value.statusMessage,
@@ -467,3 +508,40 @@ function headerValueEquals (lhs, rhs) {
 
   return lhs === rhs
 }
+
+/**
+ * @param {Record<string, string|string[]>|null|undefined} headers
+ * @returns {string|undefined}
+ */
+function getContentType (headers) {
+  if (headers == null) {
+    return undefined
+  }
+
+  for (const key in headers) {
+    if (key.toLowerCase() === 'content-type') {
+      const value = headers[key]
+      return Array.isArray(value) ? value[0] : value
+    }
+  }
+
+  return undefined
+}
+
+/**
+ * @param {string|undefined} contentType
+ * @returns {boolean}
+ */
+function isCompressibleContentType (contentType) {
+  if (contentType == null) {
+    return false
+  }
+
+  const type = contentType.toLowerCase()
+
+  // text/* and application/json, xml, javascript, ... are easily compressible
+  // even at smaller sizes, so ignore the compressThreshold for them.
+  return /^text\//.test(type) ||
+    /^application\/(json|xml|javascript|x-www-form-urlencoded|graphql)(\s|;|$)/.test(type) ||
+    /\+(json|xml)(\s|;|$)/.test(type)
+}

test/cache-interceptor/sqlite-cache-store-tests.js

@@ -370,3 +370,100 @@ test('SqliteCacheStore updates vary when overwriting an existing row', { skip: r
     }
   }), undefined)
 })
+
+let hasZstd = false
+try {
+  hasZstd = typeof require('node:zlib').zstdCompressSync === 'function'
+} catch {
+  // zstd may be unavailable on older Node versions
+}
+
+const skipCompression = runtimeFeatures.has('sqlite') === false || hasZstd === false
+
+test('SqliteCacheStore compresses big files with zstd', { skip: skipCompression }, async (t) => {
+  const SqliteCacheStore = require('../../lib/cache/sqlite-cache-store.js')
+  const sqliteLocation = 'cache-interceptor-compressed.sqlite'
+
+  const store = new SqliteCacheStore({
+    location: sqliteLocation,
+    compressThreshold: 1024
+  })
+
+  let storeClosed = false
+  t.after(async () => {
+    if (!storeClosed) {
+      store.close()
+    }
+    await rm(sqliteLocation)
+  })
+
+  /**
+   * @type {import('../../types/cache-interceptor.d.ts').default.CacheKey}
+   */
+  const key = {
+    origin: 'localhost',
+    path: '/',
+    method: 'GET',
+    headers: {}
+  }
+
+  /**
+   * @type {import('../../types/cache-interceptor.d.ts').default.CacheValue}
+   */
+  const value = {
+    statusCode: 200,
+    statusMessage: '',
+    headers: { foo: 'bar' },
+    cachedAt: Date.now(),
+    staleAt: Date.now() + 10000,
+    deleteAt: Date.now() + 20000
+  }
+
+  // A body larger than compressThreshold must be stored compressed and
+  // decompressed when read back.
+  const bigBody = Buffer.from('a'.repeat(10_000))
+  store.set({ ...key, path: '/big' }, { ...value, body: bigBody })
+
+  {
+    const result = store.get(structuredClone({ ...key, path: '/big' }))
+    notEqual(result, undefined)
+    deepStrictEqual(result.body, bigBody)
+  }
+
+  // A body smaller than compressThreshold must be stored uncompressed.
+  const smallBody = Buffer.from('small')
+  store.set({ ...key, path: '/small' }, { ...value, body: smallBody })
+
+  {
+    const result = store.get(structuredClone({ ...key, path: '/small' }))
+    notEqual(result, undefined)
+    deepStrictEqual(result.body, smallBody)
+  }
+
+  // A compressible mime type (application/json) must be compressed even when
+  // the body is below compressThreshold.
+  const jsonBody = Buffer.from('{"key":"value"}'.repeat(20))
+  store.set({ ...key, path: '/json' }, { ...value, headers: { 'content-type': 'application/json' }, body: jsonBody })
+
+  {
+    const result = store.get(structuredClone({ ...key, path: '/json' }))
+    notEqual(result, undefined)
+    deepStrictEqual(result.body, jsonBody)
+  }
+
+  // Inspect the raw rows to make sure compression actually happened.
+  store.close()
+  storeClosed = true
+  const DatabaseSync = require('node:sqlite').DatabaseSync
+  const db = new DatabaseSync(sqliteLocation)
+  const rows = db.prepare('SELECT compressed, body FROM cacheInterceptorV4 ORDER BY id').all()
+  db.close()
+
+  strictEqual(rows.length, 3)
+  strictEqual(rows[0].compressed, 'zstd')
+  strictEqual(rows[0].body.byteLength < bigBody.byteLength, true)
+  strictEqual(rows[1].compressed, null)
+  strictEqual(Buffer.from(rows[1].body).equals(smallBody), true)
+  strictEqual(rows[2].compressed, 'zstd')
+  strictEqual(rows[2].body.byteLength < jsonBody.byteLength, true)
+})

types/cache-interceptor.d.ts

@@ -160,6 +160,16 @@ declare namespace CacheHandler {
      * @default Infinity
      */
     maxEntrySize?: number
+
+    /**
+     * Responses whose body size, in bytes, is greater than this value are
+     * stored compressed with zstd. Responses with an easily compressible
+     * Content-Type (e.g. json, xml, plain text) are compressed regardless of
+     * size. Requires Node with zstd support
+     * (22.19+).
+     * @default 1048576 (1 MiB)
+     */
+    compressThreshold?: number
   }
 
   export class SqliteCacheStore implements CacheStore {