Improve throughput of stale image requests

packages/next/server/image-optimizer.ts

@@ -1,6 +1,6 @@
 import { mediaType } from 'next/dist/compiled/@hapi/accept'
 import { createHash } from 'crypto'
-import { createReadStream, promises } from 'fs'
+import { promises } from 'fs'
 import { getOrientation, Orientation } from 'next/dist/compiled/get-orientation'
 import imageSizeOf from 'next/dist/compiled/image-size'
 import { IncomingMessage, ServerResponse } from 'http'
@@ -11,15 +11,23 @@ import { join } from 'path'
 import Stream from 'stream'
 import nodeUrl, { UrlWithParsedQuery } from 'url'
 import { NextConfig } from './config-shared'
-import { fileExists } from '../lib/file-exists'
 import { ImageConfig, imageConfigDefault } from './image-config'
 import { processBuffer, decodeBuffer, Operation } from './lib/squoosh/main'
 import { sendEtagResponse } from './send-payload'
 import { getContentType, getExtension } from './serve-static'
 import chalk from 'next/dist/compiled/chalk'
 import { NextUrlWithParsedQuery } from './request-meta'
+import isError from '../lib/is-error'
 
 type XCacheHeader = 'MISS' | 'HIT' | 'STALE'
+type Inflight = { buffer: Buffer; filename: string }
+type FileMetadata = {
+  filename: string
+  maxAge: number
+  expireAt: number
+  etag: string
+  contentType: string | null
+}
 
 const AVIF = 'image/avif'
 const WEBP = 'image/webp'
@@ -31,7 +39,7 @@ const CACHE_VERSION = 3
 const ANIMATABLE_TYPES = [WEBP, PNG, GIF]
 const VECTOR_TYPES = [SVG]
 const BLUR_IMG_SIZE = 8 // should match `next-image-loader`
-const inflightRequests = new Map<string, Promise<void>>()
+const inflightRequests = new Map<string, Promise<Inflight | null>>()
 
 let sharp:
   | ((
@@ -179,51 +187,94 @@ export async function imageOptimizer(
   const hash = getHash([CACHE_VERSION, href, width, quality, mimeType])
   const imagesDir = join(distDir, 'cache', 'images')
   const hashDir = join(imagesDir, hash)
-  const now = Date.now()
+  const previousInflight = await inflightRequests.get(hashDir)
   let xCache: XCacheHeader = 'MISS'
-
-  // If there're concurrent requests hitting the same resource and it's still
-  // being optimized, wait before accessing the cache.
-  if (inflightRequests.has(hash)) {
-    await inflightRequests.get(hash)
+  let sendDedupe = { sent: false }
+
+  // If there are concurrent requests hitting the same STALE resource,
+  // we can serve it from memory to avoid blocking below.
+  if (previousInflight) {
+    const now = Date.now()
+    const { filename, buffer } = previousInflight
+    const { maxAge, expireAt, etag, contentType } = getFileMetadata(filename)
+    xCache = now < expireAt ? 'HIT' : 'STALE'
+    await sendResponse(
+      sendDedupe,
+      req,
+      res,
+      url,
+      maxAge,
+      contentType,
+      buffer,
+      isStatic,
+      isDev,
+      xCache,
+      etag
+    )
+    return { finished: true }
   }
-  const dedupe = new Deferred<void>()
-  inflightRequests.set(hash, dedupe.promise)
+
+  const currentInflight = new Deferred<Inflight | null>()
+  inflightRequests.set(hashDir, currentInflight.promise)
 
   try {
-    if (await fileExists(hashDir, 'directory')) {
-      const files = await promises.readdir(hashDir)
-      for (let file of files) {
-        const [maxAgeStr, expireAtSt, etag, extension] = file.split('.')
-        const maxAge = Number(maxAgeStr)
-        const expireAt = Number(expireAtSt)
-        const contentType = getContentType(extension)
-        const fsPath = join(hashDir, file)
-        xCache = now < expireAt ? 'HIT' : 'STALE'
-        const result = setResponseHeaders(
-          req,
-          res,
-          url,
-          etag,
-          maxAge,
-          contentType,
-          isStatic,
-          isDev,
-          xCache
-        )
-        if (!result.finished) {
-          await new Promise<void>((resolve, reject) => {
-            createReadStream(fsPath)
-              .on('end', resolve)
-              .on('error', reject)
-              .pipe(res)
-          })
-        }
-        if (xCache === 'HIT') {
-          return { finished: true }
-        } else {
-          await promises.unlink(fsPath)
-        }
+    const now = Date.now()
+    const freshFiles = []
+    const staleFiles = []
+    let cachedFile: FileMetadata | undefined
+    let allFiles: string[] = []
+    try {
+      allFiles = await promises.readdir(hashDir)
+    } catch (error) {
+      if (isError(error) && error.code === 'ENOENT') {
+        // Directory doesn't exist, so there is no cache.
+        // We'll create it later in writeToCacheDir()
+      } else {
+        throw error
+      }
+    }
+
+    for (let filename of allFiles) {
+      const meta = getFileMetadata(filename)
+      if (now < meta.expireAt) {
+        freshFiles.push(meta)
+      } else {
+        staleFiles.push(meta)
+      }
+    }
+
+    if (freshFiles.length > 0) {
+      cachedFile = freshFiles[0]
+      xCache = 'HIT'
+    } else if (staleFiles.length > 0) {
+      cachedFile = staleFiles[0]
+      xCache = 'STALE'
+    } else {
+      xCache = 'MISS'
+    }
+
+    if (cachedFile) {
+      const { filename, maxAge, etag, contentType } = cachedFile
+      const buffer = await promises.readFile(join(hashDir, filename))
+      await sendResponse(
+        sendDedupe,
+        req,
+        res,
+        url,
+        maxAge,
+        contentType,
+        buffer,
+        isStatic,
+        isDev,
+        xCache,
+        etag
+      )
+      for (let stale of staleFiles) {
+        await promises.unlink(join(hashDir, stale.filename))
+      }
+      currentInflight.resolve({ filename, buffer })
+      if (xCache === 'HIT') {
+        return { finished: true }
       }
     }
 
@@ -317,7 +368,7 @@ export async function imageOptimizer(
       }
     }
 
-    const expireAt = Math.max(maxAge, minimumCacheTTL) * 1000 + now
+    const expireAt = Math.max(maxAge, minimumCacheTTL) * 1000 + Date.now()
 
     if (upstreamType) {
       const vector = VECTOR_TYPES.includes(upstreamType)
@@ -331,7 +382,8 @@ export async function imageOptimizer(
           expireAt,
           upstreamBuffer
         )
-        sendResponse(
+        await sendResponse(
+          sendDedupe,
           req,
           res,
           url,
@@ -485,7 +537,8 @@ export async function imageOptimizer(
           expireAt,
           optimizedBuffer
         )
-        sendResponse(
+        await sendResponse(
+          sendDedupe,
           req,
           res,
           url,
@@ -500,7 +553,8 @@ export async function imageOptimizer(
         throw new Error('Unable to optimize buffer')
       }
     } catch (error) {
-      sendResponse(
+      await sendResponse(
+        sendDedupe,
         req,
         res,
         url,
@@ -515,8 +569,8 @@ export async function imageOptimizer(
 
     return { finished: true }
   } finally {
-    dedupe.resolve()
-    inflightRequests.delete(hash)
+    currentInflight.resolve(null)
+    inflightRequests.delete(hashDir)
   }
 }
 
@@ -527,10 +581,10 @@ async function writeToCacheDir(
   expireAt: number,
   buffer: Buffer
 ) {
-  await promises.mkdir(dir, { recursive: true })
   const extension = getExtension(contentType)
   const etag = getHash([buffer])
   const filename = join(dir, `${maxAge}.${expireAt}.${etag}.${extension}`)
+  await promises.mkdir(dir, { recursive: true })
   await promises.writeFile(filename, buffer)
 }
 
@@ -590,6 +644,7 @@ function setResponseHeaders(
 }
 
 function sendResponse(
+  sendDedupe: { sent: boolean },
   req: IncomingMessage,
   res: ServerResponse,
   url: string,
@@ -598,26 +653,36 @@ function sendResponse(
   buffer: Buffer,
   isStatic: boolean,
   isDev: boolean,
-  xCache: XCacheHeader
-) {
-  if (xCache === 'STALE') {
-    return
-  }
-  const etag = getHash([buffer])
-  const result = setResponseHeaders(
-    req,
-    res,
-    url,
-    etag,
-    maxAge,
-    contentType,
-    isStatic,
-    isDev,
-    xCache
-  )
-  if (!result.finished) {
-    res.end(buffer)
-  }
+  xCache: XCacheHeader,
+  etag?: string
+): Promise<void> {
+  return new Promise((resolve, reject) => {
+    if (sendDedupe.sent) {
+      return
+    } else {
+      sendDedupe.sent = true
+    }
+    res.on('finish', () => resolve())
+    res.on('close', () => resolve())
+    res.on('end', () => resolve())
+    res.on('error', (e) => reject(e))
+    const result = setResponseHeaders(
+      req,
+      res,
+      url,
+      etag || getHash([buffer]),
+      maxAge,
+      contentType,
+      isStatic,
+      isDev,
+      xCache
+    )
+    if (result.finished) {
+      resolve()
+    } else {
+      res.end(buffer, () => resolve())
+    }
+  })
 }
 
 function getSupportedMimeType(options: string[], accept = ''): string {
@@ -783,6 +848,14 @@ export async function getImageSize(
   return { width, height }
 }
 
+function getFileMetadata(filename: string): FileMetadata {
+  const [maxAgeStr, expireAtSt, etag, extension] = filename.split('.')
+  const maxAge = Number(maxAgeStr)
+  const expireAt = Number(expireAtSt)
+  const contentType = getContentType(extension)
+  return { filename, maxAge, expireAt, etag, contentType }
+}
+
 export class Deferred<T> {
   promise: Promise<T>
   resolve!: (value: T) => void