Migrate redis from v3 to v6

README.md

@@ -68,6 +68,19 @@ config = {
 }
 ```
 
+### Redis client (node-redis v6)
+
+Orka uses [node-redis](https://github.com/redis/node-redis) v6. The client returned by `getRedis()`
+is promise-based (`await getRedis().get('key')`) — the callback API of node-redis v3 is gone.
+The `config.redis` schema is unchanged: `url`, `options.tls` and the legacy retry/keepalive options
+(`timesConnected`, `totalRetryTime`, `reconnectAfterMultiplier`, `socketKeepalive`, `socketInitialDelay`)
+are mapped to the new driver's `socket` options and reconnect strategy. Any other key in
+`config.redis.options` (e.g. `RESP`, `pingInterval`, `socket`) is passed through to `createClient`.
+Orka pins `RESP: 2` by default to preserve the wire protocol reply shapes; set
+`config.redis.options.RESP = 3` to opt into RESP3.
+The initial connection is awaited during boot; if redis is down the app still starts
+and `/health` reports unhealthy.
+
 ```js
 const { orka } = require('@workablehr/orka');
 

examples/redis-example/routes.js

@@ -1,5 +1,4 @@
 const { getRedis } = require('../../build');
-const { promisify } = require('util');
 const {
   middlewares: { health }
 } = require('../../build');
@@ -9,12 +8,12 @@ module.exports = {
   get: {
     health: health,
     '/key': async (ctx, next) => {
-      ctx.body = await promisify(redis.get.bind(redis))('key');
+      ctx.body = await redis.get('key');
     }
   },
   put: {
     '/key': async (ctx, next) => {
-      ctx.body = await promisify(redis.set.bind(redis))('key', ctx.request.body.key);
+      ctx.body = await redis.set('key', ctx.request.body.key);
     }
   }
 };

package.json

@@ -71,7 +71,7 @@
     "node-cron": "^2.0.3",
     "qs": "^6.15.3",
     "rabbit-queue": "^5.9.1",
-    "redis": "^3.1.1",
+    "redis": "^6.1.0",
     "sanitize-html": "^2.17.5",
     "source-map-support": "^0.5.16",
     "tsconfig-paths": "^3.9.0",
@@ -86,7 +86,6 @@
     "@types/node": "^20.7.0",
     "@types/pg": "^7.14.11",
     "@types/qs": "^6.9.7",
-    "@types/redis": "^2.8.14",
     "@types/sinon": "^21.0.0",
     "axios": "^1.18.1",
     "bullmq": "*",

src/initializers/redis.ts

@@ -1,9 +1,11 @@
-import { createClient as createClientType, RedisClient as RedisClientType } from 'redis';
+import type { createClient as createClientType, RedisClientType } from 'redis';
 import { getLogger } from './log4js';
 import { isEmpty, cloneDeep } from 'lodash';
 
 const logger = getLogger('services.redisService');
 
+export type OrkaRedisClient = RedisClientType;
+
 function getRedisUrl(config) {
   return config && config.url;
 }
@@ -12,9 +14,15 @@ function getHost(url) {
   return url.split('@')[1] || url;
 }
 
-let firstClient: RedisClientType;
+function isConnectionRefused(cause) {
+  if (!cause) return false;
+  if (cause.code === 'ECONNREFUSED') return true;
+  return Array.isArray(cause.errors) && cause.errors.some(e => e?.code === 'ECONNREFUSED');
+}
+
+let firstClient: OrkaRedisClient;
 
-export function createRedisConnection(config) {
+export async function createRedisConnection(config) {
   const { createClient }: { createClient: typeof createClientType } = require('redis');
   config = cloneDeep(config);
   const redisUrl = getRedisUrl(config);
@@ -27,45 +35,68 @@ export function createRedisConnection(config) {
     if (isEmpty(config.options.tls)) delete config.options.tls;
   }
 
-  const options = {
-    timesConnected: 10,
-    totalRetryTime: 1000 * 60 * 60,
-    reconnectAfterMultiplier: 1000,
-    socketKeepalive: true,
-    socketInitialDelay: 60000,
-    ...config.options
-  };
+  const {
+    timesConnected = 10,
+    totalRetryTime = 1000 * 60 * 60,
+    reconnectAfterMultiplier = 1000,
+    socketKeepalive = true,
+    socketInitialDelay = 60000,
+    tls,
+    socket: socketOptions,
+    ...clientOptions
+  } = config.options ?? {};
+
+  let timesConnectedCounter = 0;
+  let firstRetryAt: number;
 
-  options.retry_strategy = function(opts) {
-    logger.error('Retrying to connect to Redis', opts);
-    if (opts.error && opts.error.code === 'ECONNREFUSED') return new Error('The server refused the connection');
-    if (opts.total_retry_time > options.totalRetryTime) return new Error('Retry time exhausted');
-    if (opts.times_connected > options.timesConnected) {
-      const msg =
-        'redis error retry_strategy options.times_connected exhausted.' +
-        'Please verify that the redis-server "timeout" config is large enough or disabled(0).' +
-        'Redis-cli:"config get timeout" ';
-      // This will be thrown globally and will stop the server
-      throw new Error(msg);
+  // Preserves the semantics of the v3 retry_strategy: give up on refused connections,
+  // on exhausted total retry time, and after too many reconnections (the health check
+  // then reports unhealthy so the orchestrator can restart the service).
+  const reconnectStrategy = (retries: number, cause: Error): number | Error => {
+    logger.error(`Retrying to connect to Redis (retries: ${retries})`, cause);
+    if (isConnectionRefused(cause)) return new Error('The server refused the connection');
+    if (firstRetryAt === undefined) firstRetryAt = Date.now();
+    if (Date.now() - firstRetryAt > totalRetryTime) return new Error('Retry time exhausted');
+    if (timesConnectedCounter > timesConnected) {
+      return new Error(
+        'redis error reconnectStrategy timesConnected exhausted.' +
+          'Please verify that the redis-server "timeout" config is large enough or disabled(0).' +
+          'Redis-cli:"config get timeout" '
+      );
     }
-    const retryInMS = Math.pow(2, opts.attempt) * options.reconnectAfterMultiplier;
+    const retryInMS = Math.pow(2, retries + 1) * reconnectAfterMultiplier;
     logger.info(`Retrying to connect to redis in ${retryInMS}ms`);
     return retryInMS;
   };
 
-  const client = createClient(redisUrl, options);
+  const client = createClient({
+    url: redisUrl,
+    RESP: 2,
+    ...clientOptions,
+    socket: {
+      keepAlive: socketKeepalive,
+      keepAliveInitialDelay: socketInitialDelay,
+      reconnectStrategy,
+      ...(tls ? { tls: true, ...tls } : {}),
+      ...socketOptions
+    }
+  });
   if (!firstClient) firstClient = client;
-  client.on('connect', () => {
-    const socket = client.stream;
-    (socket as any).setKeepAlive(options.socketKeepalive, options.socketInitialDelay);
+
+  client.on('ready', () => {
+    timesConnectedCounter++;
+    firstRetryAt = undefined;
     logger.info(`Redis connected ${getHost(redisUrl)}`);
   });
 
   client.on('error', e => {
-    if (Array.isArray(e.args)) e.args[0] = getHost(redisUrl);
-    logger.error(e, `Redis disconnected`);
+    logger.error(e, `Redis disconnected ${getHost(redisUrl)}`);
   });
 
+  // Awaited during orka boot so the client is ready before the server starts listening.
+  // A failed connection is only logged: the app still boots and /health reports unhealthy.
+  await client.connect().catch(e => logger.error(e, `Redis connection failed ${getHost(redisUrl)}`));
+
   return client;
 }
 
@@ -76,5 +107,5 @@ export function getRedis() {
 
 export const isHealthy = () => {
   if (!firstClient) return false;
-  return firstClient.connected;
+  return firstClient.isReady;
 };

test/examples/health-example.test.ts

@@ -91,7 +91,7 @@ describe('Health examples', () => {
     });
 
     it('/health returns not ok', async () => {
-      getRedis().end(true);
+      getRedis().destroy();
       await supertest('localhost:3210').get('/health').expect(503);
     });
   });