Redis client factory (error handling and good defaults) (#1761)

* @internal/redis package for creating Redis clients with sensible defaults and error handling

* Added @internal/redis to webapp

Author: matt-aitken

apps/webapp/package.json

@@ -52,6 +52,7 @@
     "@heroicons/react": "^2.0.12",
     "@internal/run-engine": "workspace:*",
     "@internal/zod-worker": "workspace:*",
+    "@internal/redis": "workspace:*",
     "@internal/redis-worker": "workspace:*",
     "@internationalized/date": "^3.5.1",
     "@lezer/highlight": "^1.1.6",
@@ -258,4 +259,4 @@
   "engines": {
     "node": ">=16.0.0"
   }
-}
+}

apps/webapp/tsconfig.json

@@ -38,7 +38,9 @@
       "@internal/run-engine": ["../../internal-packages/run-engine/src/index"],
       "@internal/run-engine/*": ["../../internal-packages/run-engine/src/*"],
       "@internal/redis-worker": ["../../internal-packages/redis-worker/src/index"],
-      "@internal/redis-worker/*": ["../../internal-packages/redis-worker/src/*"]
+      "@internal/redis-worker/*": ["../../internal-packages/redis-worker/src/*"],
+      "@internal/redis": ["../../internal-packages/redis/src/index"],
+      "@internal/redis/*": ["../../internal-packages/redis/src/*"]
     },
     "noEmit": true
   }

internal-packages/redis-worker/package.json

@@ -7,6 +7,7 @@
   "type": "module",
   "dependencies": {
     "@opentelemetry/api": "^1.9.0",
+    "@internal/redis": "workspace:*",
     "@trigger.dev/core": "workspace:*",
     "ioredis": "^5.3.2",
     "lodash.omit": "^4.5.0",
@@ -23,4 +24,4 @@
     "typecheck": "tsc --noEmit",
     "test": "vitest --no-file-parallelism"
   }
-}
+}

internal-packages/redis-worker/src/queue.ts

@@ -1,3 +1,4 @@
+import { createRedisClient } from "@internal/redis";
 import { Logger } from "@trigger.dev/core/logger";
 import Redis, { type Callback, type RedisOptions, type Result } from "ioredis";
 import { nanoid } from "nanoid";
@@ -50,35 +51,29 @@ export class SimpleQueue<TMessageCatalog extends MessageCatalogSchema> {
     logger?: Logger;
   }) {
     this.name = name;
-    this.redis = new Redis({
-      ...redisOptions,
-      keyPrefix: `${redisOptions.keyPrefix ?? ""}{queue:${name}:}`,
-      retryStrategy(times) {
-        const delay = Math.min(times * 50, 1000);
-        return delay;
+    this.logger = logger ?? new Logger("SimpleQueue", "debug");
+
+    this.redis = createRedisClient(
+      {
+        ...redisOptions,
+        keyPrefix: `${redisOptions.keyPrefix ?? ""}{queue:${name}:}`,
+        retryStrategy(times) {
+          const delay = Math.min(times * 50, 1000);
+          return delay;
+        },
+        maxRetriesPerRequest: 20,
       },
-      maxRetriesPerRequest: 20,
-    });
+      {
+        onError: (error) => {
+          this.logger.error(`RedisWorker queue redis client error:`, {
+            error,
+            keyPrefix: redisOptions.keyPrefix,
+          });
+        },
+      }
+    );
     this.#registerCommands();
     this.schema = schema;
-
-    this.logger = logger ?? new Logger("SimpleQueue", "debug");
-
-    this.redis.on("error", (error) => {
-      this.logger.error(`Redis Error for queue ${this.name}:`, { queue: this.name, error });
-    });
-
-    this.redis.on("connect", () => {
-      this.logger.log(`Redis connected for queue ${this.name}`);
-    });
-
-    this.redis.on("reconnecting", () => {
-      this.logger.warn(`Redis reconnecting for queue ${this.name}`);
-    });
-
-    this.redis.on("close", () => {
-      this.logger.warn(`Redis connection closed for queue ${this.name}`);
-    });
   }
 
   async enqueue({

internal-packages/redis-worker/src/worker.test.ts

@@ -5,6 +5,7 @@ import { expect } from "vitest";
 import { z } from "zod";
 import { Worker } from "./worker.js";
 import Redis from "ioredis";
+import { createRedisClient } from "@internal/redis";
 
 describe("Worker", () => {
   redisTest("Process items that don't throw", { timeout: 30_000 }, async ({ redisContainer }) => {
@@ -241,7 +242,7 @@ describe("Worker", () => {
         expect(dlqSize).toBe(1);
 
         // Create a Redis client to publish the redrive message
-        const redisClient = new Redis({
+        const redisClient = createRedisClient({
           host: redisContainer.getHost(),
           port: redisContainer.getPort(),
           password: redisContainer.getPassword(),

internal-packages/redis-worker/src/worker.ts

@@ -9,6 +9,7 @@ import Redis from "ioredis";
 import { nanoid } from "nanoid";
 import { startSpan } from "./telemetry.js";
 import pLimit from "p-limit";
+import { createRedisClient } from "@internal/redis";
 
 export type WorkerCatalog = {
   [key: string]: {
@@ -108,7 +109,14 @@ class Worker<TCatalog extends WorkerCatalog> {
 
     this.setupShutdownHandlers();
 
-    this.subscriber = new Redis(this.options.redisOptions);
+    this.subscriber = createRedisClient(this.options.redisOptions, {
+      onError: (error) => {
+        this.logger.error(`RedisWorker subscriber redis client error:`, {
+          error,
+          keyPrefix: this.options.redisOptions.keyPrefix,
+        });
+      },
+    });
     this.setupSubscriber();
 
     return this;

internal-packages/redis-worker/tsconfig.json

@@ -18,7 +18,9 @@
       "@internal/testcontainers": ["../../internal-packages/testcontainers/src/index"],
       "@internal/testcontainers/*": ["../../internal-packages/testcontainers/src/*"],
       "@trigger.dev/core": ["../../packages/core/src/index"],
-      "@trigger.dev/core/*": ["../../packages/core/src/*"]
+      "@trigger.dev/core/*": ["../../packages/core/src/*"],
+      "@internal/redis": ["../../internal-packages/redis/src/index"],
+      "@internal/redis/*": ["../../internal-packages/redis/src/*"]
     }
   },
   "exclude": ["node_modules"]

internal-packages/redis/README.md

@@ -0,0 +1,3 @@
+# Redis
+
+This is a simple package that is used to return a valid Redis client and provides an error callback. It will log and swallow errors if they're not handled.

internal-packages/redis/package.json

@@ -0,0 +1,18 @@
+{
+  "name": "@internal/redis",
+  "private": true,
+  "version": "0.0.1",
+  "main": "./src/index.ts",
+  "types": "./src/index.ts",
+  "type": "module",
+  "dependencies": {
+    "ioredis": "^5.3.2",
+    "@trigger.dev/core": "workspace:*"
+  },
+  "devDependencies": {
+    "vitest": "^1.4.0"
+  },
+  "scripts": {
+    "typecheck": "tsc --noEmit"
+  }
+}

internal-packages/redis/src/index.ts

@@ -0,0 +1,40 @@
+import { Redis, RedisOptions } from "ioredis";
+import { Logger } from "@trigger.dev/core/logger";
+
+const defaultOptions: Partial<RedisOptions> = {
+  retryStrategy: (times: number) => {
+    const delay = Math.min(times * 50, 1000);
+    return delay;
+  },
+  maxRetriesPerRequest: 20,
+};
+
+const logger = new Logger("Redis", "debug");
+
+export function createRedisClient(
+  options: RedisOptions,
+  handlers?: { onError?: (err: Error) => void }
+): Redis {
+  const client = new Redis({
+    ...defaultOptions,
+    ...options,
+  });
+
+  // Skip error handling setup if running in Vitest
+  if (process.env.VITEST) {
+    client.on("error", (error) => {
+      // swallow errors
+    });
+    return client;
+  }
+
+  client.on("error", (error) => {
+    if (handlers?.onError) {
+      handlers.onError(error);
+    } else {
+      logger.error(`Redis client error:`, { error, keyPrefix: options.keyPrefix });
+    }
+  });
+
+  return client;
+}