⚡ perf: Run Stream Lua Scripts via EVALSHA Instead of Per-Call EVAL

packages/api/src/cache/redisScript.ts

@@ -0,0 +1,42 @@
+import { createHash } from 'node:crypto';
+import type { Redis, Cluster } from 'ioredis';
+
+export type RedisScriptArg = string | number | Buffer;
+export type RedisScriptClient = Pick<Redis | Cluster, 'eval' | 'evalsha'>;
+
+const scriptShas = new Map<string, string>();
+
+function scriptSha(script: string): string {
+  let sha = scriptShas.get(script);
+  if (sha == null) {
+    sha = createHash('sha1').update(script).digest('hex');
+    scriptShas.set(script, sha);
+  }
+  return sha;
+}
+
+export function isNoScriptError(error: unknown): boolean {
+  return error instanceof Error && error.message.includes('NOSCRIPT');
+}
+
+/**
+ * Runs a Lua script by its SHA1 (EVALSHA) so only the 40-byte digest crosses the wire on
+ * every call, and falls back to EVAL — which also loads the script into the server cache —
+ * when the server reports NOSCRIPT (first use, restart, SCRIPT FLUSH). Same semantics,
+ * atomicity, key slotting and return value as `client.eval(script, ...)`.
+ */
+export async function evalScript<T = unknown>(
+  client: RedisScriptClient,
+  script: string,
+  numberOfKeys: number,
+  ...args: RedisScriptArg[]
+): Promise<T> {
+  try {
+    return (await client.evalsha(scriptSha(script), numberOfKeys, ...args)) as T;
+  } catch (error) {
+    if (!isNoScriptError(error)) {
+      throw error;
+    }
+    return (await client.eval(script, numberOfKeys, ...args)) as T;
+  }
+}

packages/api/src/stream/__tests__/RedisJobStore.spec.ts

@@ -7,6 +7,11 @@ jest.mock('~/cache/redisTelemetry', () => ({
   instrumentIORedisClient: (client: unknown) => client,
 }));
 
+/** Cold script cache: every EVALSHA reports NOSCRIPT so the store falls back to EVAL. */
+function evalshaNoScript(): jest.Mock {
+  return jest.fn().mockRejectedValue(new Error('NOSCRIPT No matching script. Please use EVAL.'));
+}
+
 type Deferred<T> = {
   promise: Promise<T>;
   resolve: (value: T) => void;
@@ -48,6 +53,7 @@ describe('RedisJobStore', () => {
     const evalDrain = jest.fn().mockResolvedValue(1);
     const redis = {
       isCluster: true,
+      evalsha: evalshaNoScript(),
       eval: evalDrain,
     } as unknown as Cluster;
     const store = new RedisJobStore(redis);
@@ -72,6 +78,7 @@ describe('RedisJobStore', () => {
     const evalBegin = jest.fn().mockResolvedValue(1);
     const redis = {
       isCluster: true,
+      evalsha: evalshaNoScript(),
       eval: evalBegin,
     } as unknown as Cluster;
     const store = new RedisJobStore(redis);
@@ -97,6 +104,7 @@ describe('RedisJobStore', () => {
     const evalTransition = jest.fn().mockResolvedValue(0);
     const redis = {
       isCluster: true,
+      evalsha: evalshaNoScript(),
       eval: evalTransition,
       hgetall: jest.fn().mockResolvedValue({}),
     } as unknown as Cluster;
@@ -174,6 +182,7 @@ describe('RedisJobStore', () => {
     const evalTransition = jest.fn().mockResolvedValue(0);
     const redis = {
       isCluster: true,
+      evalsha: evalshaNoScript(),
       eval: evalTransition,
       hgetall: jest.fn().mockResolvedValue({
         streamId: 'stream-terminal-barrier',
@@ -207,6 +216,7 @@ describe('RedisJobStore', () => {
     const evalTransition = jest.fn().mockResolvedValue(0);
     const redis = {
       isCluster: true,
+      evalsha: evalshaNoScript(),
       eval: evalTransition,
     } as unknown as Cluster;
     const store = new RedisJobStore(redis, { requiresActionTtl: 4321 });
@@ -231,6 +241,7 @@ describe('RedisJobStore', () => {
     const evalCommand = jest.fn().mockResolvedValue(1);
     const redis = {
       isCluster: true,
+      evalsha: evalshaNoScript(),
       eval: evalCommand,
       hgetall: jest
         .fn()
@@ -285,6 +296,7 @@ describe('RedisJobStore', () => {
     const lrange = jest.fn();
     const redis = {
       isCluster: true,
+      evalsha: evalshaNoScript(),
       eval: evalPeek,
       lrange,
     } as unknown as Cluster;
@@ -310,6 +322,7 @@ describe('RedisJobStore', () => {
       .mockImplementation((...args: unknown[]) => ['', '', args[Number(args[1]) + 3]]);
     const redis = {
       isCluster: true,
+      evalsha: evalshaNoScript(),
       eval: evalJobCreation,
       hgetall: jest.fn(() => jobHashFromCreationCall(evalJobCreation.mock.calls[0])),
       sadd: jest.fn().mockResolvedValue(1),
@@ -584,6 +597,7 @@ describe('RedisJobStore', () => {
       ]);
     const redis = {
       isCluster: true,
+      evalsha: evalshaNoScript(),
       eval: evalJobCreation,
       hgetall: jest.fn(() => jobHashFromCreationCall(evalJobCreation.mock.calls[0])),
       sadd: jest.fn().mockResolvedValue(1),
@@ -639,6 +653,7 @@ describe('RedisJobStore', () => {
     const now = jest.spyOn(Date, 'now').mockReturnValue(100);
     const redis = {
       isCluster: true,
+      evalsha: evalshaNoScript(),
       eval: jest.fn().mockResolvedValue(['user-1', '', '100']),
       hgetall: jest.fn().mockResolvedValue({
         streamId: 'stream-overlap',
@@ -669,6 +684,7 @@ describe('RedisJobStore', () => {
   test('rejects creation when its durable epoch is already terminal', async () => {
     const redis = {
       isCluster: true,
+      evalsha: evalshaNoScript(),
       eval: jest.fn().mockResolvedValue(['', '', '100']),
       hgetall: jest.fn().mockResolvedValue({
         streamId: 'stream-terminal-create',
@@ -692,6 +708,7 @@ describe('RedisJobStore', () => {
     const evalRedis = jest.fn().mockResolvedValue(false);
     const redis = {
       isCluster: true,
+      evalsha: evalshaNoScript(),
       eval: evalRedis,
     } as unknown as Cluster;
     const store = new RedisJobStore(redis);
@@ -770,6 +787,7 @@ describe('RedisJobStore', () => {
     });
     const redis = {
       isCluster: true,
+      evalsha: evalshaNoScript(),
       eval: evalJobCreation,
       sadd: jest.fn((key: string) => {
         if (key === 'stream:running') {
@@ -800,6 +818,7 @@ describe('RedisJobStore', () => {
       return job;
     });
 
+    await waitFor(() => started.length === 1);
     expect(started).toEqual(['job']);
     evalResult.resolve(1);
     await waitFor(() => started.length === 6);
@@ -879,6 +898,7 @@ describe('RedisJobStore', () => {
     });
     const redis = {
       isCluster: true,
+      evalsha: evalshaNoScript(),
       eval: jest.fn(async (_script: string, keyCount: number, ...args: string[]) => {
         if (keyCount === 10) {
           durableHash = { ...durableHash, status: 'requires_action' };
@@ -922,6 +942,7 @@ describe('RedisJobStore', () => {
     const evalCommand = jest.fn().mockResolvedValue(0);
     const redis = {
       isCluster: true,
+      evalsha: evalshaNoScript(),
       eval: evalCommand,
       hgetall: jest.fn().mockResolvedValue({
         streamId: 'stream-guarded',
@@ -957,6 +978,7 @@ describe('RedisJobStore', () => {
     const evalCommand = jest.fn().mockResolvedValue(0);
     const redis = {
       isCluster: true,
+      evalsha: evalshaNoScript(),
       eval: evalCommand,
     } as unknown as Cluster;
     const store = new RedisJobStore(redis);
@@ -995,6 +1017,7 @@ describe('RedisJobStore', () => {
     const evalCommand = jest.fn().mockResolvedValue(0);
     const redis = {
       isCluster: true,
+      evalsha: evalshaNoScript(),
       eval: evalCommand,
     } as unknown as Cluster;
     const store = new RedisJobStore(redis);
@@ -1026,6 +1049,7 @@ describe('RedisJobStore', () => {
     const sadd = jest.fn().mockResolvedValue(1);
     const redis = {
       isCluster: true,
+      evalsha: evalshaNoScript(),
       eval: evalTransition,
       sadd,
       srem: jest.fn().mockResolvedValue(1),
@@ -1083,6 +1107,7 @@ describe('RedisJobStore', () => {
     };
     const redis = {
       isCluster: true,
+      evalsha: evalshaNoScript(),
       eval: evalTransition,
       sadd,
       srem: jest.fn().mockResolvedValue(1),
@@ -1144,6 +1169,7 @@ describe('RedisJobStore', () => {
     const srem = jest.fn().mockResolvedValue(1);
     const redis = {
       isCluster: true,
+      evalsha: evalshaNoScript(),
       eval: jest.fn().mockResolvedValue(1),
       smembers: jest.fn().mockResolvedValue([member]),
       srem,
@@ -1191,6 +1217,7 @@ describe('RedisJobStore', () => {
     const srem = jest.fn().mockResolvedValue(1);
     const redis = {
       isCluster: true,
+      evalsha: evalshaNoScript(),
       eval: evalClear,
       srem,
       hgetall: jest.fn().mockResolvedValue({}),
@@ -1233,6 +1260,7 @@ describe('RedisJobStore', () => {
     const evalCommand = jest.fn().mockResolvedValue(0);
     const redis = {
       isCluster: true,
+      evalsha: evalshaNoScript(),
       eval: evalCommand,
     } as unknown as Cluster;
     const store = new RedisJobStore(redis);

packages/api/src/stream/__tests__/RedisJobStore.stream_integration.spec.ts

@@ -8,7 +8,7 @@ import {
   PAUSE_PERSISTENCE_TIMEOUT_ERROR,
   STEER_ENQUEUE_RECEIPT_FULL,
 } from '../interfaces/IJobStore';
-import { clearRedisTestPrefix } from './helpers/redis';
+import { clearRedisTestPrefix, flushScriptCache } from './helpers/redis';
 
 /** Suppress winston Console transport output (survives jest.resetModules) */
 jest.spyOn(console, 'log').mockImplementation();
@@ -492,6 +492,7 @@ describe('RedisJobStore Integration Tests', () => {
       const streamId = `terminal-epoch-${Date.now()}`;
       const userId = 'terminal-epoch-user';
       const now = jest.spyOn(Date, 'now').mockReturnValue(1000);
+      await flushScriptCache(ioredisClient);
       const originalEval = ioredisClient.eval.bind(ioredisClient) as (
         script: string | Buffer,
         numberOfKeys: number,
@@ -1632,7 +1633,7 @@ describe('RedisJobStore Integration Tests', () => {
       store.setCollectedUsage(streamId, [{ input_tokens: 1, output_tokens: 2 }]);
 
       const evalSpy = jest
-        .spyOn(ioredisClient, 'eval')
+        .spyOn(ioredisClient, 'evalsha')
         .mockRejectedValueOnce(new Error('replacement write failed'));
       try {
         await expect(store.createJob(streamId, 'user-1', streamId)).rejects.toThrow(
@@ -4084,6 +4085,7 @@ describe('RedisJobStore Integration Tests', () => {
         String(Date.now() - 10_000),
       );
 
+      await flushScriptCache(ioredisClient);
       const originalEval = ioredisClient.eval.bind(ioredisClient) as (
         script: string | Buffer,
         numberOfKeys: number,

packages/api/src/stream/__tests__/helpers/publisher.ts

@@ -6,6 +6,7 @@ export interface MockPublisher {
   get: jest.Mock;
   set: jest.Mock;
   del: jest.Mock;
+  evalsha: jest.Mock;
   eval: jest.Mock;
 }
 
@@ -54,6 +55,9 @@ export function createMockPublisher(): MockPublisher {
       }
       return Promise.resolve(keys.length);
     }),
+    evalsha: jest
+      .fn()
+      .mockRejectedValue(new Error('NOSCRIPT No matching script. Please use EVAL.')),
     eval: jest.fn(),
   };
 

packages/api/src/stream/__tests__/helpers/redis.ts

@@ -34,6 +34,17 @@ export function createRedisTestClient(keyPrefix: string): RedisTestClient {
   return new IoRedis(primary.href, { ...redisOptions, lazyConnect: true });
 }
 
+/**
+ * Empties every node's Lua script cache so the next run of each script misses EVALSHA and
+ * falls back to EVAL — the only path where a test spying on `eval` can observe the script body.
+ */
+export async function flushScriptCache(redis: RedisTestClient): Promise<void> {
+  const nodes = (redis as Cluster).isCluster
+    ? (redis as Cluster).nodes('master')
+    : [redis as Redis];
+  await Promise.all(nodes.map((node) => node.script('FLUSH')));
+}
+
 /** Delete only this suite's keys, including keys spread across cluster masters. */
 export async function clearRedisTestPrefix(
   redis: RedisTestClient,

packages/api/src/stream/__tests__/protocolRollout.stream_integration.spec.ts

@@ -1,5 +1,10 @@
 import type { SteerQueueItem } from '../interfaces/IJobStore';
-import { clearRedisTestPrefix, createRedisTestClient, type RedisTestClient } from './helpers/redis';
+import {
+  clearRedisTestPrefix,
+  createRedisTestClient,
+  flushScriptCache,
+  type RedisTestClient,
+} from './helpers/redis';
 import { InMemoryEventTransport } from '../implementations/InMemoryEventTransport';
 import { RedisEventTransport } from '../implementations/RedisEventTransport';
 import { GenerationJobManagerClass } from '../GenerationJobManager';
@@ -377,6 +382,7 @@ describe('Redis generation protocol rollout bridge', () => {
         racedPredecessor = await owner.createJob(streamId, userId, streamId, {
           initialMetadata: { generationProtocolVersion: 2 },
         });
+        await flushScriptCache(redis);
         injectLostCreateReply = true;
       }
       return observed;
@@ -530,6 +536,7 @@ describe('Redis generation protocol rollout bridge', () => {
         initialMetadata: { generationProtocolVersion: 2 },
       });
       expect(await ownerStore.getJob(streamId)).toMatchObject({ providerAbortReady: true });
+      await flushScriptCache(redis);
       injectLostCreateReply = true;
 
       await expect(

packages/api/src/stream/implementations/RedisEventTransport.ts

@@ -15,6 +15,7 @@ import {
 import { registerChunkPublicationCapability } from '~/stream/internal/chunkPublication';
 import { GenerationPublicationFencedError } from '~/stream/interfaces/IJobStore';
 import { instrumentIORedisClient, RedisUseCases } from '~/cache/redisTelemetry';
+import { evalScript } from '~/cache/redisScript';
 
 /**
  * Redis key prefixes for pub/sub channels
@@ -352,7 +353,8 @@ export class RedisEventTransport implements IEventTransport {
        * commits but its promise never settles, the marker timeout still releases attachment
        * admission instead of leaving every surviving local subscriber deferred forever. */
       const operation = Promise.all([
-        this.publisher.eval(
+        evalScript(
+          this.publisher,
           CAPTURE_SUBSCRIPTION_FRONTIER_LUA,
           1,
           KEYS.sequence(streamId),
@@ -471,7 +473,8 @@ export class RedisEventTransport implements IEventTransport {
     allowRetainedEpoch = false,
     requireActiveJob = false,
   ): Promise<number> {
-    const seq = await this.publisher.eval(
+    const seq = await evalScript(
+      this.publisher,
       PUBLISH_SEQ_LUA,
       3,
       KEYS.sequence(streamId),
@@ -1372,7 +1375,8 @@ export class RedisEventTransport implements IEventTransport {
       data: event,
       generationId: replacedGenerationId,
     });
-    const result = await this.publisher.eval(
+    const result = await evalScript(
+      this.publisher,
       PUBLISH_REPLACED_DONE_LUA,
       2,
       KEYS.sequence(streamId),