feat(redis): cluster-safe keys for the api, worker and file-server paths

ElastiCache Serverless (Valkey) is always cluster mode, so every multi-key
script, MULTI/EXEC and MGET must stay in one hash slot. Follows the key shapes
of upstream PR LibreChat-AI#17 so this can be dropped when that lands:

- redis-connection.ts: isClusterMode (USE_REDIS_CLUSTER), bullmqPrefix
  ('{codeapi}' in cluster mode, BullMQ's default otherwise), hashTag.
- runtime-session registry: rtsx:{sess,lock,gen,ckptseq}:{<id>} so the
  record+lock scripts are single-slot (not covered by PR LibreChat-AI#17).
- replay state: exec_state/exec_lock/exec_result/tool_history:{<id>};
  the stale-execution sweep reads with pipelined GETs instead of a
  cross-slot MGET.
- BullMQ prefix on every Queue/Worker/QueueEvents, including hosted apps.

Bridge, egress ledger and tool-call server are not covered (not used by
this deployment).

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Wzy53QgK2dCkSKvEFTySD1

Author: busla

service/src/hosted-app/queue.ts

@@ -6,6 +6,7 @@ import { hostedAppOperationTimeoutMs } from '../config';
 import logger from '../logger';
 import { withHostedAppQueueDeadline } from './queue-deadline';
 import type { HostedAppJobData, HostedAppJobName, HostedAppJobResult } from './jobs';
+import { bullmqPrefix } from '../redis-connection';
 
 /** Hosted apps are stateful-only. A fixed isolated queue prevents an ordinary
  * stateless worker from ever receiving an AWS lifecycle job. */
@@ -24,9 +25,9 @@ function resources(): { queue: HostedAppQueue; events: QueueEvents } {
   if (!hostedAppQueue || !hostedAppQueueEvents) {
     hostedAppQueue = new Queue<HostedAppJobData, HostedAppJobResult, HostedAppJobName>(
       HOSTED_APP_QUEUE_NAME,
-      { connection },
+      { prefix: bullmqPrefix(), connection },
     );
-    hostedAppQueueEvents = new QueueEvents(HOSTED_APP_QUEUE_NAME, { connection });
+    hostedAppQueueEvents = new QueueEvents(HOSTED_APP_QUEUE_NAME, { prefix: bullmqPrefix(), connection });
     setMaxListeners(0, hostedAppQueue, hostedAppQueueEvents);
     /* These resources are created after lifecycle startup, on first use, so the
      * ordinary queue listener registration never sees them. An unhandled

service/src/hosted-app/worker.ts

@@ -8,6 +8,7 @@ import type { HostedAppJob, HostedAppJobData, HostedAppJobName, HostedAppJobResu
 import { HOSTED_APP_QUEUE_NAME } from './queue';
 import logger from '../logger';
 import { workerRunning } from '../metrics';
+import { bullmqPrefix } from '../redis-connection';
 
 export function serializedHostedAppFailure(error: unknown): Error {
   if (error instanceof HostedAppControlPlaneError) {
@@ -101,7 +102,7 @@ export const hostedAppWorker: Worker<
   HostedAppJobName
 > | undefined = env.HOSTED_APPS_ENABLED
   ? new Worker(HOSTED_APP_QUEUE_NAME, processHostedAppJob, {
-    connection,
+    prefix: bullmqPrefix(), connection,
     /* Lifecycle transitions are serialized again by their per-app Redis lock.
      * This modest concurrency allows unrelated apps to launch in parallel while
      * the fleet-wide AWS throttle remains authoritative. */

service/src/queue.ts

@@ -19,6 +19,7 @@ import type {
 import logger from './logger';
 import { redisKeepAliveOptions } from './redis-options';
 import { bullmqQueueJobs, registerBullmqQueueMetricsCollector } from './metrics';
+import { bullmqPrefix } from './redis-connection';
 
 const MAX_RECONNECT_ATTEMPTS = 5;
 const RECONNECT_DELAY = 2000;
@@ -87,8 +88,8 @@ function getQueueResources(
   const existing = queueResources.get(name);
   if (existing != null) return existing;
 
-  const queue = new Queue<t.JobData, t.JobResult, Jobs.execute>(name, { connection });
-  const events = new QueueEvents(name, { connection });
+  const queue = new Queue<t.JobData, t.JobResult, Jobs.execute>(name, { prefix: bullmqPrefix(), connection });
+  const events = new QueueEvents(name, { prefix: bullmqPrefix(), connection });
   setMaxListeners(0, queue, events);
   const resources = { queue, events };
   queueResources.set(name, resources);

service/src/redis-connection.ts

@@ -0,0 +1,31 @@
+/** Minimal Redis Cluster key helpers. Same names and key shapes as upstream
+ * PR #17 (LibreChat-AI/code-interpreter, "Add Redis Cluster Support") so this
+ * file can be dropped when that lands. ElastiCache Serverless is always
+ * cluster mode: every multi-key script, MULTI/EXEC or MGET must stay in one
+ * hash slot.
+ *
+ * ponytail: covers what the api, worker and file-server use (session
+ * registry, replay state, BullMQ). bridge/*, egress-ledger and
+ * tool-call-server still do cross-slot work; tag them before running those
+ * components against a cluster. */
+
+export function isClusterMode(): boolean {
+  return process.env.USE_REDIS_CLUSTER === 'true' || (process.env.REDIS_HOST ?? '').includes(',');
+}
+
+/** One hash tag keeps every BullMQ key in one slot on a cluster. Standalone
+ * keeps BullMQ's default so existing queues are untouched. */
+export function bullmqPrefix(): string {
+  return isClusterMode() ? '{codeapi}' : 'bull';
+}
+
+/** Wrap an id in a hash tag so every key built from it lands in one slot.
+ * Redis Cluster hashes only the substring between the first `{` and `}`. */
+export function hashTag(id: string): string {
+  return `{${id}}`;
+}
+
+/** Reverse of `hashTag`. */
+export function stripHashTag(raw: string): string {
+  return raw.startsWith('{') && raw.endsWith('}') ? raw.slice(1, -1) : raw;
+}

service/src/runtime-session/registry.test.ts

@@ -181,10 +181,10 @@ describe('runtime session lock', () => {
     releaseDelayedSet();
     await lateSetComplete;
     for (let attempt = 0; attempt < 20; attempt += 1) {
-      if (await mock.get('rtsx:lock:rt_late_lock') == null) break;
+      if (await mock.get('rtsx:lock:{rt_late_lock}') == null) break;
       await new Promise(resolve => setTimeout(resolve, 10));
     }
-    expect(await mock.get('rtsx:lock:rt_late_lock')).toBeNull();
+    expect(await mock.get('rtsx:lock:{rt_late_lock}')).toBeNull();
   });
 
   test('an applied SET whose replay resolves null cannot leave a ghost lock', async () => {
@@ -228,10 +228,10 @@ describe('runtime session lock', () => {
     await expect(acquire).rejects.toThrow('job deadline');
     finishReplay();
     for (let attempt = 0; attempt < 20; attempt += 1) {
-      if (await mock.get('rtsx:lock:rt_ambiguous_lock') == null) break;
+      if (await mock.get('rtsx:lock:{rt_ambiguous_lock}') == null) break;
       await new Promise(resolve => setTimeout(resolve, 10));
     }
-    expect(await mock.get('rtsx:lock:rt_ambiguous_lock')).toBeNull();
+    expect(await mock.get('rtsx:lock:{rt_ambiguous_lock}')).toBeNull();
   });
 
   test('an applied SET whose replay quickly resolves null is still released', async () => {
@@ -254,10 +254,10 @@ describe('runtime session lock', () => {
 
     expect(await acquireRuntimeSessionLock('rt_fast_null_lock', 60_000)).toBeNull();
     for (let attempt = 0; attempt < 20; attempt += 1) {
-      if (await mock.get('rtsx:lock:rt_fast_null_lock') == null) break;
+      if (await mock.get('rtsx:lock:{rt_fast_null_lock}') == null) break;
       await new Promise(resolve => setTimeout(resolve, 10));
     }
-    expect(await mock.get('rtsx:lock:rt_fast_null_lock')).toBeNull();
+    expect(await mock.get('rtsx:lock:{rt_fast_null_lock}')).toBeNull();
   });
 
   test('an applied SET whose response quickly rejects is still released', async () => {
@@ -282,10 +282,10 @@ describe('runtime session lock', () => {
       acquireRuntimeSessionLock('rt_fast_reject_lock', 60_000),
     ).rejects.toThrow('connection closed after apply');
     for (let attempt = 0; attempt < 20; attempt += 1) {
-      if (await mock.get('rtsx:lock:rt_fast_reject_lock') == null) break;
+      if (await mock.get('rtsx:lock:{rt_fast_reject_lock}') == null) break;
       await new Promise(resolve => setTimeout(resolve, 10));
     }
-    expect(await mock.get('rtsx:lock:rt_fast_reject_lock')).toBeNull();
+    expect(await mock.get('rtsx:lock:{rt_fast_reject_lock}')).toBeNull();
   });
 
   test('lock polling catches abort between its precheck and listener registration', async () => {
@@ -400,7 +400,7 @@ describe('fenced record writes', () => {
   });
 
   test('reads a corrupt record as missing instead of throwing', async () => {
-    await mock.set('rtsx:sess:rt_bad', '{not valid json');
+    await mock.set('rtsx:sess:{rt_bad}', '{not valid json');
     expect(await readRuntimeSessionRecord('rt_bad')).toBeNull();
   });
 

service/src/runtime-session/registry.ts

@@ -7,6 +7,7 @@ import {
 } from '../config';
 import logger from '../logger';
 import type { HostedAppRecordDetails } from '../hosted-app/record';
+import { hashTag } from '../redis-connection';
 
 export { RUNTIME_SESSION_REDIS_COMMAND_TIMEOUT_MS } from '../config';
 
@@ -294,7 +295,7 @@ export async function acquireRuntimeSessionLock(
   try {
     result = await runRegistryCommand(
       'Runtime session lock acquire',
-      () => redis.set(`${LOCK_PREFIX}${runtimeSessionId}`, token, 'PX', ttlMs, 'NX'),
+      () => redis.set(`${LOCK_PREFIX}${hashTag(runtimeSessionId)}`, token, 'PX', ttlMs, 'NX'),
       options,
       /* A caller-side deadline cannot cancel ioredis. The original SET may
        * have succeeded even if an automatic replay eventually resolves null,
@@ -404,7 +405,7 @@ export async function releaseRuntimeSessionLock(
     try {
       await runRegistryCommand(
         'Runtime session lock release',
-        () => redis.releaseRuntimeSessionLockScript(`${LOCK_PREFIX}${runtimeSessionId}`, token),
+        () => redis.releaseRuntimeSessionLockScript(`${LOCK_PREFIX}${hashTag(runtimeSessionId)}`, token),
         { signal: options.signal, timeoutMs: remainingMs },
       );
       return;
@@ -453,7 +454,7 @@ export async function renewRuntimeSessionLock(
     const result = await runRegistryCommand(
       'Runtime session lock renewal',
       () => redis.renewRuntimeSessionLockScript(
-        `${LOCK_PREFIX}${runtimeSessionId}`,
+        `${LOCK_PREFIX}${hashTag(runtimeSessionId)}`,
         token,
         String(ttlMs),
       ),
@@ -475,7 +476,7 @@ export async function readRuntimeSessionRecord(
 ): Promise<RuntimeSessionRecord | null> {
   const data = await runRegistryCommand(
     'Runtime session record read',
-    () => redis.get(`${SESS_PREFIX}${runtimeSessionId}`),
+    () => redis.get(`${SESS_PREFIX}${hashTag(runtimeSessionId)}`),
     options,
   );
   if (data == null) return null;
@@ -502,8 +503,8 @@ export async function writeRuntimeSessionRecord(
     const result = await runRegistryCommand(
       'Runtime session record write',
       () => redis.writeRuntimeSessionRecordScript(
-        `${SESS_PREFIX}${record.runtime_session_id}`,
-        `${LOCK_PREFIX}${record.runtime_session_id}`,
+        `${SESS_PREFIX}${hashTag(record.runtime_session_id)}`,
+        `${LOCK_PREFIX}${hashTag(record.runtime_session_id)}`,
         lockToken,
         JSON.stringify(record),
         String(ttlSeconds),
@@ -533,7 +534,7 @@ export async function allocateRuntimeSessionGeneration(
   if (!Number.isSafeInteger(initialGeneration) || initialGeneration < 1) {
     throw new Error('Runtime session generation must be a positive safe integer');
   }
-  const key = `${GEN_PREFIX}${runtimeSessionId}`;
+  const key = `${GEN_PREFIX}${hashTag(runtimeSessionId)}`;
   const rawGeneration = await runRegistryCommand(
     'Runtime session generation allocation',
     () => redis.allocateRuntimeSessionGenerationScript(
@@ -561,7 +562,7 @@ export async function allocateCheckpointSequence(
   retainedMax = 0,
   options: RuntimeSessionRedisCommandOptions = {},
 ): Promise<number> {
-  const key = `${CKPT_SEQ_PREFIX}${runtimeSessionId}`;
+  const key = `${CKPT_SEQ_PREFIX}${hashTag(runtimeSessionId)}`;
   return runRegistryCommand(
     'Runtime session checkpoint sequence allocation',
     () => redis.allocateCheckpointSequenceScript(
@@ -583,8 +584,8 @@ export async function removeRuntimeSession(
   const result = await runRegistryCommand(
     'Runtime session record removal',
     () => redis.removeRuntimeSessionScript(
-      `${SESS_PREFIX}${runtimeSessionId}`,
-      `${LOCK_PREFIX}${runtimeSessionId}`,
+      `${SESS_PREFIX}${hashTag(runtimeSessionId)}`,
+      `${LOCK_PREFIX}${hashTag(runtimeSessionId)}`,
       lockToken,
     ),
     options,

service/src/sandbox-backend/lambda-microvm.test.ts

@@ -129,7 +129,7 @@ beforeAll(() => {
           await new Promise((resolve) => setTimeout(resolve, executeDelayMs));
         }
         if (stealSessionLockOnExecute) {
-          await mock.set('rtsx:lock:rt_session_1', 'stolen');
+          await mock.set('rtsx:lock:{rt_session_1}', 'stolen');
         }
         return new Response(JSON.stringify(executeResponseBody), {
           status: executeStatus,
@@ -728,7 +728,7 @@ describe('LambdaMicrovmSandboxBackend session execution', () => {
     });
     /* Model a worker that persisted its intent and whose RunMicrovm reached AWS,
      * but died before it could record the returned MicroVM id. */
-    await mock.set('rtsx:gen:rt_session_1', '7');
+    await mock.set('rtsx:gen:{rt_session_1}', '7');
     const lock = await acquireRuntimeSessionLock('rt_session_1', 60_000);
     expect(lock).not.toBeNull();
     await writeRuntimeSessionRecord({
@@ -861,7 +861,7 @@ describe('LambdaMicrovmSandboxBackend session execution', () => {
      * volatile registry (including its generation counter) is lost, while the
      * provider retains its client-token idempotency history. */
     await fake.terminateMicrovm(firstVmId);
-    await mock.del('rtsx:sess:rt_session_1', 'rtsx:gen:rt_session_1');
+    await mock.del('rtsx:sess:{rt_session_1}', 'rtsx:gen:{rt_session_1}');
 
     await expect(
       makeBackend(fake, { imageVersion: '4' }).execute(request(), sessionContext()),
@@ -1439,7 +1439,7 @@ describe('LambdaMicrovmSandboxBackend session execution', () => {
     const fake = fakeClient();
     const backend = makeBackend(fake);
     onExecute = async () => {
-      await mock.del('rtsx:sess:rt_session_1');
+      await mock.del('rtsx:sess:{rt_session_1}');
     };
 
     try {
@@ -1680,7 +1680,7 @@ describe('LambdaMicrovmSandboxBackend auto-checkpoint', () => {
     const originalGet = redisWithGet.get.bind(mock);
     let failSessionRead = false;
     redisWithGet.get = async (key: string): Promise<string | null> => {
-      if (failSessionRead && key === 'rtsx:sess:rt_ckpt_1') {
+      if (failSessionRead && key === 'rtsx:sess:{rt_ckpt_1}') {
         throw new Error('registry read unavailable');
       }
       return originalGet(key);