Refactor node-redis handler to use pub/sub

Author: better-salmon

apps/cache-testing/next.config.ts

@@ -20,6 +20,12 @@ const nextConfig: NextConfig = {
       default: './redis.js',
     },
   },
+  eslint: {
+    ignoreDuringBuilds: true,
+  },
+  typescript: {
+    ignoreBuildErrors: true,
+  },
 };
 
 export default nextConfig;

apps/cache-testing/package.json

@@ -17,8 +17,8 @@
   },
   "dependencies": {
     "@neshca/cache-handler": "workspace:*",
-    "axios": "1.8.3",
-    "next": "15.3.0-canary.13",
+    "axios": "1.8.4",
+    "next": "15.3.0-canary.14",
     "react": "19.0.0",
     "react-dom": "19.0.0",
     "redis": "4.7.0"

apps/cache-testing/src/utils/create-get-data.ts

@@ -15,7 +15,6 @@ export function createGetData(path: string, revalidate?: number) {
 
     cacheLife({
       revalidate,
-      expire: revalidate ? revalidate * 2 : undefined,
     });
 
     cacheTag(pathAndTag, 'whole-app-route');

docs/cache-handler-docs/package.json

@@ -12,7 +12,7 @@
     "start:docs": "serve out"
   },
   "dependencies": {
-    "next": "15.3.0-canary.13",
+    "next": "15.3.0-canary.14",
     "nextra": "4.2.16",
     "nextra-theme-docs": "4.2.16",
     "react": "19.0.0",

internal/eslint-config/package.json

@@ -8,14 +8,14 @@
   },
   "devDependencies": {
     "@eslint/js": "9.22.0",
-    "@next/eslint-plugin-next": "15.3.0-canary.13",
+    "@next/eslint-plugin-next": "15.3.0-canary.14",
     "eslint": "9.22.0",
     "eslint-config-prettier": "10.1.1",
     "eslint-plugin-react": "7.37.4",
     "eslint-plugin-react-hooks": "5.2.0",
     "eslint-plugin-turbo": "2.4.4",
     "globals": "16.0.0",
     "typescript": "5.8.2",
-    "typescript-eslint": "8.26.1"
+    "typescript-eslint": "8.27.0"
   }
 }

packages/cache-handler/package.json

@@ -44,7 +44,7 @@
   "devDependencies": {
     "@repo/typescript-config": "workspace:*",
     "@types/node": "22.13.10",
-    "next": "15.3.0-canary.13",
+    "next": "15.3.0-canary.14",
     "tsx": "4.19.3",
     "typescript": "5.8.2",
     "vitest": "3.0.9"

packages/cache-handler/src/cache-handler.ts

@@ -323,6 +323,8 @@ async function removeEntryFromHandlers(
   });
 }
 
+export type CacheHandlerType = typeof CacheHandler;
+
 export class CacheHandler implements NextCacheHandler {
   /**
    * Provides a descriptive name for the CacheHandler class.

packages/cache-handler/src/instrumentation/register-initial-cache.ts

@@ -2,10 +2,9 @@ import { promises as fsPromises } from 'node:fs';
 import path from 'node:path';
 import { PRERENDER_MANIFEST, SERVER_DIRECTORY } from 'next/constants.js';
 import type { PrerenderManifest } from 'next/dist/build/index.js';
+import type { CacheHandlerType } from '../cache-handler.js';
 import { CachedRouteKind, type Revalidate } from '../next-common-types.js';
 
-type CacheHandlerType = typeof import('../cache-handler.js').CacheHandler;
-
 type Router = 'pages' | 'app';
 
 const PRERENDER_MANIFEST_VERSION = 4;

packages/cache-handler/src/use-cache-cache.ts

@@ -1,6 +1,7 @@
 import type {
   CacheEntry,
   CacheHandlerV2,
+  Timestamp,
 } from 'next/dist/server/lib/cache-handlers/types.js';
 import {
   isStale,
@@ -15,9 +16,16 @@ export type { CacheHandlerV2, CacheEntry };
 
 const pendingSets = new Map<string, Promise<void>>();
 
+type RemoteStoreOptions = {
+  timestamp: Timestamp;
+  revalidate: number;
+  expire: number;
+  stale: number;
+};
+
 export type RemoteStore = {
   get(key: string): Promise<string | undefined>;
-  set(key: string, value: string): Promise<void>;
+  set(key: string, value: string, options: RemoteStoreOptions): Promise<void>;
   refreshTags(tagsManifest: Map<string, number>): Promise<void>;
   getExpirationTimestamps(tags: string[]): Promise<number[]>;
   expireTags(expiredTags: Map<string, number>): Promise<void>;
@@ -100,7 +108,12 @@ export async function createCacheHandler(
 
           const remoteStore = await remoteStorePromise;
 
-          await remoteStore.set(cacheKey, JSON.stringify(storageEntry));
+          await remoteStore.set(cacheKey, JSON.stringify(storageEntry), {
+            expire: entry.expire,
+            revalidate: entry.revalidate,
+            stale: entry.stale,
+            timestamp: entry.timestamp,
+          });
         } catch (_error) {
           //
         } finally {

packages/cache-handler/src/use-cache/node-redis.ts

@@ -1,111 +1,118 @@
+import { randomUUID } from 'node:crypto';
+import { tagsManifest } from 'next/dist/server/lib/incremental-cache/tags-manifest.external.js';
 import type { createClient } from 'redis';
 import { createRedisTimeoutConfig } from '../helpers/create-redis-timeout-config.js';
 import { type RemoteStore, createCacheHandler } from '../use-cache-cache.js';
 
 export type Config<T extends ReturnType<typeof createClient>> = {
   client: T;
+  pubClient: T;
+  subClientId: string;
+  channel: string;
   keyPrefix?: string;
-  sharedTagsKey?: string;
   timeoutMs?: number;
+  expireTrigger?: 'stale' | 'expire';
+};
+
+type Message = {
+  expiredTags: Record<string, number>;
+  subClientId: string;
 };
 
 function createRedisStore<T extends ReturnType<typeof createClient>>({
   client,
+  pubClient,
+  channel,
+  subClientId,
   keyPrefix,
-  sharedTagsKey,
   timeoutMs = 5000,
+  expireTrigger = 'stale',
 }: Config<T>): RemoteStore {
   const getKey = (key: string) => `${keyPrefix}${key}`;
-  const getSharedTagsKey = () => `${keyPrefix}${sharedTagsKey}`;
 
   return {
     async get(key) {
       if (!client.isReady) {
-        return Promise.resolve(undefined);
+        return;
       }
 
       const options = createRedisTimeoutConfig(timeoutMs);
 
       return (await client.get(options, getKey(key))) ?? undefined;
     },
-    async set(key, value) {
+    async set(key, value, { expire, timestamp, stale }) {
       if (!client.isReady) {
         return;
       }
 
       const options = createRedisTimeoutConfig(timeoutMs);
 
-      await client.set(options, getKey(key), value);
+      const expireAt =
+        expireTrigger === 'stale'
+          ? Math.floor(timestamp / 1000 + Math.min(expire, stale))
+          : Math.floor(timestamp / 1000 + expire);
+
+      await client.set(options, getKey(key), value, {
+        EXAT: expireAt,
+      });
     },
     async expireTags(expiredTags) {
-      if (!client.isReady) {
+      if (!pubClient.isReady) {
         return;
       }
 
-      const options = createRedisTimeoutConfig(timeoutMs);
-
-      await client.hSet(
-        options,
-        getSharedTagsKey(),
-        Object.fromEntries(expiredTags),
+      await pubClient.publish(
+        channel,
+        JSON.stringify({
+          expiredTags: Object.fromEntries(expiredTags),
+          subClientId,
+        } satisfies Message),
       );
     },
     getExpirationTimestamps() {
-      return Promise.resolve([]);
+      return Promise.resolve([0]);
     },
-    async refreshTags(tagsManifest) {
-      try {
-        if (!client?.isReady) {
-          return;
-        }
-
-        let cursor = 0;
-
-        const hScanOptions = { COUNT: 10 };
-
-        do {
-          const options = createRedisTimeoutConfig(timeoutMs);
-
-          const remoteTagsPortion = await client.hScan(
-            options,
-            getSharedTagsKey(),
-            cursor,
-            hScanOptions,
-          );
-
-          for (const {
-            field: tag,
-            value: timestampStr,
-          } of remoteTagsPortion.tuples) {
-            try {
-              const timestamp = Number.parseInt(timestampStr, 10);
-
-              if (!Number.isNaN(timestamp)) {
-                tagsManifest.set(tag, timestamp);
-              }
-            } catch (err) {
-              console.error(`Failed to parse tag timestamp for ${tag}:`, err);
-            }
-          }
-
-          cursor = remoteTagsPortion.cursor;
-        } while (cursor !== 0);
-      } catch (error) {
-        console.error('Failed to load tags from Redis:', error);
-      }
+    async refreshTags() {
+      // must be empty when using pub/sub
     },
   };
 }
 
 export function createRedisCacheHandler<
   T extends ReturnType<typeof createClient>,
->({ client, keyPrefix, sharedTagsKey, timeoutMs }: Config<T>) {
-  const remoteStore = client
-    .connect()
-    .then((redisClient) => ({
-      client: redisClient,
+>({ client, keyPrefix, timeoutMs }: Config<T>) {
+  const remoteStore = Promise.all([
+    client.connect(),
+    client.duplicate().connect(),
+    client.duplicate().connect(),
+  ])
+    .then(async ([mainClient, pubClient, subClient]) => {
+      const subClientId = randomUUID();
+      const channel = `${keyPrefix}__revalidate_channel__`;
+
+      await subClient.subscribe(channel, (message) => {
+        const { expiredTags, subClientId: messageSubClientId } = JSON.parse(
+          message,
+        ) as Message;
+
+        if (subClientId === messageSubClientId) {
+          console.info('ignoring message from self');
+          return;
+        }
+
+        for (const [tag, timestamp] of Object.entries(expiredTags)) {
+          tagsManifest.set(tag, timestamp);
+        }
+      });
+
+      return { mainClient, pubClient, subClientId, channel };
+    })
+    .then(({ mainClient, pubClient, subClientId, channel }) => ({
+      client: mainClient,
+      pubClient,
+      subClientId,
+      channel,
       keyPrefix,
-      sharedTagsKey,
       timeoutMs,
     }))
     .then(createRedisStore);