feat: initial implementation

Author: pieh

src/run/handlers/server.ts

@@ -25,6 +25,7 @@ import { setupWaitUntil } from './wait-until.cjs'
 setFetchBeforeNextPatchedIt(globalThis.fetch)
 // configure some globals that Next.js make use of before we start importing any Next.js code
 // as some globals are consumed at import time
+// TODO: only call this if Next.js version is using CacheHandlerV2 as we don't have V1 compatible implementation
 configureUseCacheHandlers()
 setupWaitUntil()
 

src/run/handlers/tags-handler.cts

@@ -25,6 +25,27 @@ async function lastTagRevalidationTimestamp(
   return tagManifest.revalidatedAt
 }
 
+/**
+ * Get the most recent revalidation timestamp for a list of tags
+ */
+export async function getMostRecentTagRevalidationTimestamp(tags: string[]) {
+  if (tags.length === 0) {
+    return 0
+  }
+
+  const cacheStore = getMemoizedKeyValueStoreBackedByRegionalBlobStore({ consistency: 'strong' })
+
+  const timestampsOrNulls = await Promise.all(
+    tags.map((tag) => lastTagRevalidationTimestamp(tag, cacheStore)),
+  )
+
+  const timestamps = timestampsOrNulls.filter((timestamp) => timestamp !== null)
+  if (timestamps.length === 0) {
+    return 0
+  }
+  return Math.max(...timestamps)
+}
+
 /**
  * Check if any of the tags were invalidated since the given timestamp
  */

src/run/handlers/use-cache-handler.ts

@@ -1,89 +1,324 @@
-import defaultCacheHandlerImport from 'next-next/dist/server/lib/cache-handlers/default.js'
+import { Buffer } from 'node:buffer'
+
+import { LRUCache } from 'lru-cache'
 import type {
   CacheEntry,
   CacheHandlerV2 as CacheHandler,
   Timestamp,
 } from 'next-next/dist/server/lib/cache-handlers/types.js'
 
-import { getRequestContext } from './request-context.cjs'
+import { getLogger, getRequestContext } from './request-context.cjs'
+import {
+  getMostRecentTagRevalidationTimestamp,
+  isAnyTagStale,
+  markTagsAsStaleAndPurgeEdgeCache,
+} from './tags-handler.cjs'
+import { getTracer } from './tracer.cjs'
 
-const defaultCacheHandler = defaultCacheHandlerImport.default
+// copied from default implementation
+// packages/next/src/server/lib/cache-handlers/default.ts
+type PrivateCacheEntry = {
+  entry: CacheEntry
 
-export const NetlifyDefaultUseCacheHandler = {
-  get: function (cacheKey: string): Promise<undefined | CacheEntry> {
-    console.log('NetlifyDefaultUseCacheHandler::get start', { cacheKey })
-    return defaultCacheHandler.get(cacheKey).then(async (entry) => {
-      if (!entry) {
-        console.log('NetlifyDefaultUseCacheHandler::get MISS', { cacheKey })
-        return entry
-      }
+  // For the default cache we store errored cache
+  // entries and allow them to be used up to 3 times
+  // after that we want to dispose it and try for fresh
 
-      const [cloneStream, newSaved] = entry.value.tee()
-      entry.value = newSaved
+  // If an entry is errored we return no entry
+  // three times so that we retry hitting origin (MISS)
+  // and then if it still fails to set after the third we
+  // return the errored content and use expiration of
+  // Math.min(30, entry.expiration)
 
-      const [returnStream, debugStream] = cloneStream.tee()
+  isErrored: boolean // pieh: this doesn't seem to be actually used in the default implementation
+  errorRetryCount: number // pieh: this doesn't seem to be actually used in the default implementation
 
-      const text = await new Response(debugStream).text()
-      console.log('NetlifyDefaultUseCacheHandler::get HIT', { cacheKey, entry, text })
+  // compute size on set since we need to read size
+  // of the ReadableStream for LRU evicting
+  size: number
+}
 
-      return {
-        ...entry,
-        value: returnStream,
-      }
-    })
+type CacheHandleLRUCache = LRUCache<string, PrivateCacheEntry>
+type PendingSets = Map<string, Promise<void>>
+
+const LRU_CACHE_GLOBAL_KEY = Symbol.for('nf-use-cache-handler-lru-cache')
+const PENDING_SETS_GLOBAL_KEY = Symbol.for('nf-use-cache-handler-pending-sets')
+const cacheHandlersSymbol = Symbol.for('@next/cache-handlers')
+const extendedGlobalThis = globalThis as typeof globalThis & {
+  // Used by Next Runtime to ensure we have single instance of
+  //  - LRUCache
+  //  - pending sets
+  // even if this module gets copied multiple times
+  [LRU_CACHE_GLOBAL_KEY]?: CacheHandleLRUCache
+  [PENDING_SETS_GLOBAL_KEY]?: PendingSets
+
+  // Used by Next.js to provide implementation of cache handlers
+  [cacheHandlersSymbol]?: {
+    RemoteCache?: CacheHandler
+    DefaultCache?: CacheHandler
+  }
+}
+
+function getLRUCache(): CacheHandleLRUCache {
+  if (extendedGlobalThis[LRU_CACHE_GLOBAL_KEY]) {
+    return extendedGlobalThis[LRU_CACHE_GLOBAL_KEY]
+  }
+
+  const lruCache = new LRUCache<string, PrivateCacheEntry>({
+    max: 1000,
+    maxSize: 50 * 1024 * 1024, // same as hardcoded default in Next.js
+    sizeCalculation: (value) => value.size,
+  })
+
+  extendedGlobalThis[LRU_CACHE_GLOBAL_KEY] = lruCache
+
+  return lruCache
+}
+
+function getPendingSets(): PendingSets {
+  if (extendedGlobalThis[PENDING_SETS_GLOBAL_KEY]) {
+    return extendedGlobalThis[PENDING_SETS_GLOBAL_KEY]
+  }
+
+  const pendingSets = new Map()
+  extendedGlobalThis[PENDING_SETS_GLOBAL_KEY] = pendingSets
+  return pendingSets
+}
+
+const debugLog = (...args: unknown[]) => console.log(new Date().toISOString(), ...args)
+
+// eslint-disable-next-line @typescript-eslint/no-empty-function
+const tmpResolvePendingBeforeCreatingAPromise = () => {}
+
+const WIPNetlifyDefaultUseCacheHandler = {
+  get(cacheKey: string): ReturnType<CacheHandler['get']> {
+    return getTracer().withActiveSpan(
+      'DefaultUseCacheHandler.get',
+      async (span): ReturnType<CacheHandler['get']> => {
+        span.setAttributes({
+          cacheKey,
+        })
+
+        const pendingPromise = getPendingSets().get(cacheKey)
+        if (pendingPromise) {
+          await pendingPromise
+        }
+
+        const privateEntry = getLRUCache().get(cacheKey)
+        if (!privateEntry) {
+          span.setAttributes({
+            cacheStatus: 'miss',
+          })
+          return undefined
+        }
+
+        const { entry } = privateEntry
+        const ttl = (entry.timestamp + entry.revalidate * 1000 - Date.now()) / 1000
+        if (ttl < 0) {
+          // In-memory caches should expire after revalidate time because it is
+          // unlikely that a new entry will be able to be used before it is dropped
+          // from the cache.
+          span.setAttributes({
+            cacheStatus: 'expired, discarded',
+            ttl,
+          })
+          return undefined
+        }
+
+        if (await isAnyTagStale(entry.tags, entry.timestamp)) {
+          span.setAttributes({
+            cacheStatus: 'stale tag, discarded',
+            ttl,
+          })
+          return undefined
+        }
+
+        // returning entry will cause stream to be consumed
+        // so we need to clone it first, so in-memory cache can
+        // be used again
+        const [returnStream, newSaved] = entry.value.tee()
+        entry.value = newSaved
+
+        span.setAttributes({
+          cacheStatus: 'hit',
+          ttl,
+        })
+
+        return {
+          ...entry,
+          value: returnStream,
+        }
+      },
+    )
   },
-  set: async function (cacheKey: string, pendingEntry: Promise<CacheEntry>): Promise<void> {
-    console.log('NetlifyDefaultUseCacheHandler::set start', { cacheKey })
-    const entry = await pendingEntry
-    const [storeStream, debugStream] = entry.value.tee()
+  set(cacheKey: string, pendingEntry: Promise<CacheEntry>): ReturnType<CacheHandler['set']> {
+    return getTracer().withActiveSpan(
+      'DefaultUseCacheHandler.set',
+      async (span): ReturnType<CacheHandler['set']> => {
+        span.setAttributes({
+          cacheKey,
+        })
+
+        let resolvePending: () => void = tmpResolvePendingBeforeCreatingAPromise
+        const pendingPromise = new Promise<void>((resolve) => {
+          resolvePending = resolve
+        })
+
+        const pendingSets = getPendingSets()
+
+        pendingSets.set(cacheKey, pendingPromise)
 
-    const toSetEntry = Promise.resolve({
-      ...entry,
-      value: storeStream,
-    })
+        const entry = await pendingEntry
 
-    const text = await new Response(debugStream).text()
+        span.setAttributes({
+          cacheKey,
+        })
 
-    console.log('NetlifyDefaultUseCacheHandler::set awaited pending entry', {
-      cacheKey,
-      entry,
-      text,
-    })
+        let size = 0
+        try {
+          const [value, clonedValue] = entry.value.tee()
+          entry.value = value
+          const reader = clonedValue.getReader()
 
-    const setPromise = defaultCacheHandler.set(cacheKey, toSetEntry).then(() => {
-      console.log('NetlifyDefaultUseCacheHandler::set finish', { cacheKey })
-    })
+          for (let chunk; !(chunk = await reader.read()).done; ) {
+            size += Buffer.from(chunk.value).byteLength
+          }
 
-    getRequestContext()?.trackBackgroundWork(setPromise)
+          span.setAttributes({
+            tags: entry.tags,
+            timestamp: entry.timestamp,
+            revalidate: entry.revalidate,
+            expire: entry.expire,
+          })
 
-    return setPromise
+          getLRUCache().set(cacheKey, {
+            entry,
+            isErrored: false,
+            errorRetryCount: 0,
+            size,
+          })
+        } catch (error) {
+          getLogger().withError(error).error('UseCacheHandler.set error')
+        } finally {
+          resolvePending()
+          pendingSets.delete(cacheKey)
+        }
+      },
+    )
   },
-  refreshTags: function (): Promise<void> {
-    console.log('NetlifyDefaultUseCacheHandler::refreshTags')
-    return defaultCacheHandler.refreshTags()
+  async refreshTags(): Promise<void> {
+    // we check tags on demand, so we don't need to do anything here
+    // additionally this is blocking and we do need to check tags in
+    // persisted storage, so if we would maintain in-memory tags manifests
+    // we would need to check more tags than current request needs
+    // while blocking pipeline
   },
-  getExpiration: function (...tags: string[]): Promise<Timestamp> {
-    console.log('NetlifyDefaultUseCacheHandler::getExpiration start', { tags })
-    return defaultCacheHandler.getExpiration(...tags).then((expiration) => {
-      console.log('NetlifyDefaultUseCacheHandler::getExpiration finish', { tags, expiration })
-      return expiration
-    })
+  getExpiration: function (...tags: string[]): ReturnType<CacheHandler['getExpiration']> {
+    return getTracer().withActiveSpan(
+      'DefaultUseCacheHandler.getExpiration',
+      async (span): ReturnType<CacheHandler['getExpiration']> => {
+        span.setAttributes({
+          tags,
+        })
+
+        const expiration = await getMostRecentTagRevalidationTimestamp(tags)
+
+        span.setAttributes({
+          expiration,
+        })
+
+        return expiration
+      },
+    )
   },
-  expireTags: function (...tags: string[]): Promise<void> {
-    console.log('NetlifyDefaultUseCacheHandler::expireTags', { tags })
-    return defaultCacheHandler.expireTags(...tags)
+  expireTags(...tags: string[]): ReturnType<CacheHandler['expireTags']> {
+    return getTracer().withActiveSpan(
+      'DefaultUseCacheHandler.expireTags',
+      async (span): ReturnType<CacheHandler['expireTags']> => {
+        span.setAttributes({
+          tags,
+        })
+
+        await markTagsAsStaleAndPurgeEdgeCache(tags)
+      },
+    )
   },
 } satisfies CacheHandler
 
-const cacheHandlersSymbol = Symbol.for('@next/cache-handlers')
+function createDebugUseCacheHandler(cacheHandlerToDebug: CacheHandler): CacheHandler {
+  return {
+    get: function (cacheKey: string): Promise<undefined | CacheEntry> {
+      debugLog('NetlifyDefaultUseCacheHandler::get start', { cacheKey })
+      return cacheHandlerToDebug.get(cacheKey).then(async (entry) => {
+        if (!entry) {
+          debugLog('NetlifyDefaultUseCacheHandler::get MISS', { cacheKey })
+          return entry
+        }
 
-const extendedGlobalThis = globalThis as typeof globalThis & {
-  [cacheHandlersSymbol]?: {
-    RemoteCache?: CacheHandler
-    DefaultCache?: CacheHandler
+        const [cloneStream, newSaved] = entry.value.tee()
+        entry.value = newSaved
+
+        const [returnStream, debugStream] = cloneStream.tee()
+
+        const text = await new Response(debugStream).text()
+        debugLog('NetlifyDefaultUseCacheHandler::get HIT', { cacheKey, entry, text })
+
+        return {
+          ...entry,
+          value: returnStream,
+        }
+      })
+    },
+    set: async function (cacheKey: string, pendingEntry: Promise<CacheEntry>): Promise<void> {
+      debugLog('NetlifyDefaultUseCacheHandler::set start', { cacheKey })
+      const entry = await pendingEntry
+      const [storeStream, debugStream] = entry.value.tee()
+
+      const toSetEntry = Promise.resolve({
+        ...entry,
+        value: storeStream,
+      })
+
+      const text = await new Response(debugStream).text()
+
+      debugLog('NetlifyDefaultUseCacheHandler::set awaited pending entry', {
+        cacheKey,
+        entry,
+        text,
+      })
+
+      const setPromise = cacheHandlerToDebug.set(cacheKey, toSetEntry).then(() => {
+        debugLog('NetlifyDefaultUseCacheHandler::set finish', { cacheKey })
+      })
+
+      getRequestContext()?.trackBackgroundWork(setPromise)
+
+      return setPromise
+    },
+    refreshTags: function (): Promise<void> {
+      debugLog('NetlifyDefaultUseCacheHandler::refreshTags start')
+      return cacheHandlerToDebug.refreshTags().then(() => {
+        debugLog('NetlifyDefaultUseCacheHandler::refreshTags end')
+      })
+    },
+    getExpiration: function (...tags: string[]): Promise<Timestamp> {
+      debugLog('NetlifyDefaultUseCacheHandler::getExpiration start', { tags })
+      return cacheHandlerToDebug.getExpiration(...tags).then((expiration) => {
+        debugLog('NetlifyDefaultUseCacheHandler::getExpiration finish', { tags, expiration })
+        return expiration
+      })
+    },
+    expireTags: function (...tags: string[]): Promise<void> {
+      debugLog('NetlifyDefaultUseCacheHandler::expireTags', { tags })
+      return cacheHandlerToDebug.expireTags(...tags)
+    },
   }
 }
 
+export const NetlifyDefaultUseCacheHandler = createDebugUseCacheHandler(
+  WIPNetlifyDefaultUseCacheHandler,
+)
+
 export function configureUseCacheHandlers() {
   extendedGlobalThis[cacheHandlersSymbol] = {
     DefaultCache: NetlifyDefaultUseCacheHandler,