@uppy/companion: switch from node-redis to ioredis

[dschmidt]

Followup for #4597

This replaces node-redis with the (apparently) more modern ioredis. I haven't found much resources on whether one should pick one or the other, but ioredis supports Redis Sentinel natively (which I need) while node-redis does not.

It's supposed to be faster for certain use cases:
https://ably.com/blog/migrating-from-node-redis-to-ioredis
https://www.reddit.com/r/node/comments/uymb7w/redis_vs_ioredis/
https://github.com/poppinlp/node_redis-vs-ioredis

https://docs.redis.com/latest/rs/references/client_references/client_ioredis/
Both libs are "community recommended" ... for whatever that means.

This is a breaking change as config options are not fully compatible.
If someone passes in redisOptions in middleware mode (in standalone mode it wasn't possible until recently, see #4597) the behavior might change/break...
If you want to have this, it's probably for the next major release

Not sure it would be worth introducing an abstraction and support both. connect-redis does some sort of basic abstraction, but they would not accept patches to abstract more api away (I've asked)

Closes #4571

[dschmidt]

@arturi Is there any roadmap for 4.0?
Any rough eta on when you want to release that? Depending on that it might actually be worth to support ioredis and node-redis at the same time for now...

[arturi]

@dschmidt we were just discussing it, and it could be that new major for Companion will happen in around 2 months.

[github-actions]

Diff output files

diff --git a/packages/@uppy/companion/lib/server/Uploader.d.ts b/packages/@uppy/companion/lib/server/Uploader.d.ts
index 5d9a222..ae38c05 100644
--- a/packages/@uppy/companion/lib/server/Uploader.d.ts
+++ b/packages/@uppy/companion/lib/server/Uploader.d.ts
@@ -23,7 +23,7 @@ declare class Uploader {
         size: any;
         companionOptions: any;
         pathPrefix: string;
-        storage: any;
+        storage: import("ioredis").default;
         s3: {
             client: any;
             options: any;
@@ -69,6 +69,8 @@ declare class Uploader {
         useFormData?: boolean;
         chunkSize?: number;
     });
+    /** @type {import('ioredis').Redis} */
+    storage: import('ioredis').Redis;
     options: {
         endpoint: string;
         uploadUrl?: string;
@@ -89,7 +91,6 @@ declare class Uploader {
     fileName: string;
     size: number;
     uploadFileName: any;
-    storage: any;
     downloadedBytes: number;
     readStream: import("stream").Readable | fs.ReadStream;
     abortReadStream(err: any): void;
diff --git a/packages/@uppy/companion/lib/server/Uploader.js b/packages/@uppy/companion/lib/server/Uploader.js
index aed14e0..27a21f1 100644
--- a/packages/@uppy/companion/lib/server/Uploader.js
+++ b/packages/@uppy/companion/lib/server/Uploader.js
@@ -134,6 +134,8 @@ class StreamableBlob {
   }
 }
 class Uploader {
+  /** @type {import('ioredis').Redis} */
+  storage;
   /**
    * Uploads file to destination based on the supplied protocol (tus, s3-multipart, multipart)
    * For tus uploads, the deferredLength option is enabled, because file size value can be unreliable
@@ -401,9 +403,7 @@ class Uploader {
     // https://github.com/transloadit/uppy/issues/3748
     const keyExpirySec = 60 * 60 * 24;
     const redisKey = `${Uploader.STORAGE_PREFIX}:${this.token}`;
-    this.storage.set(redisKey, jsonStringify(state), {
-      EX: keyExpirySec,
-    });
+    this.storage.set(redisKey, jsonStringify(state), "EX", keyExpirySec);
   }
   throttledEmitProgress = throttle(
     (dataToEmit) => {
diff --git a/packages/@uppy/companion/lib/server/emitter/redis-emitter.d.ts b/packages/@uppy/companion/lib/server/emitter/redis-emitter.d.ts
index ba7a709..0d502d9 100644
--- a/packages/@uppy/companion/lib/server/emitter/redis-emitter.d.ts
+++ b/packages/@uppy/companion/lib/server/emitter/redis-emitter.d.ts
@@ -1,4 +1,4 @@
-declare function _exports(redisClient: any, redisPubSubScope: any): {
+declare function _exports(redisClient: import('ioredis').Redis, redisPubSubScope: string): {
     on: (eventName: string, handler: any) => void | EventEmitter<[never]>;
     off: (eventName: string, handler: any) => Promise<void> | EventEmitter<[never]>;
     once: (eventName: string, handler: any) => void | EventEmitter<[never]>;
@@ -7,3 +7,4 @@ declare function _exports(redisClient: any, redisPubSubScope: any): {
     removeAllListeners: (eventName: string) => Promise<void> | EventEmitter<[never]>;
 };
 export = _exports;
+import { EventEmitter } from "events";
diff --git a/packages/@uppy/companion/lib/server/emitter/redis-emitter.js b/packages/@uppy/companion/lib/server/emitter/redis-emitter.js
index 5612124..aa4f768 100644
--- a/packages/@uppy/companion/lib/server/emitter/redis-emitter.js
+++ b/packages/@uppy/companion/lib/server/emitter/redis-emitter.js
@@ -6,16 +6,21 @@ const logger = require("../logger");
  * This module simulates the builtin events.EventEmitter but with the use of redis.
  * This is useful for when companion is running on multiple instances and events need
  * to be distributed across.
+ *
+ * @param {import('ioredis').Redis} redisClient
+ * @param {string} redisPubSubScope
+ * @returns
  */
 module.exports = (redisClient, redisPubSubScope) => {
   const prefix = redisPubSubScope ? `${redisPubSubScope}:` : "";
   const getPrefixedEventName = (eventName) => `${prefix}${eventName}`;
-  const publisher = redisClient.duplicate();
-  publisher.on("error", err => logger.error("publisher redis error", err));
+  const publisher = redisClient.duplicate({ lazyConnect: true });
+  publisher.on("error", err => logger.error("publisher redis error", err.toString()));
+  /** @type {import('ioredis').Redis} */
   let subscriber;
   const connectedPromise = publisher.connect().then(() => {
     subscriber = publisher.duplicate();
-    subscriber.on("error", err => logger.error("subscriber redis error", err));
+    subscriber.on("error", err => logger.error("subscriber redis error", err.toString()));
     return subscriber.connect();
   });
   const handlersByEvent = new Map();
@@ -53,11 +58,20 @@ module.exports = (redisClient, redisPubSubScope) => {
       if (handlersByThisEventName.size === 0) {
         handlersByEvent.delete(eventName);
       }
-      return subscriber.pUnsubscribe(getPrefixedEventName(eventName), actualHandler);
+      subscriber.off("pmessage", actualHandler);
+      return subscriber.punsubscribe(getPrefixedEventName(eventName));
     });
   }
+  /**
+   * @param {string} eventName
+   * @param {*} handler
+   * @param {*} _once
+   */
   function addListener(eventName, handler, _once = false) {
-    function actualHandler(message) {
+    function actualHandler(pattern, channel, message) {
+      if (pattern !== getPrefixedEventName(eventName)) {
+        return;
+      }
       if (_once) {
         removeListener(eventName, handler);
       }
@@ -65,9 +79,10 @@ module.exports = (redisClient, redisPubSubScope) => {
       try {
         args = JSON.parse(message);
       } catch (ex) {
-        return handleError(new Error(`Invalid JSON received! Channel: ${eventName} Message: ${message}`));
+        handleError(new Error(`Invalid JSON received! Channel: ${eventName} Message: ${message}`));
+        return;
       }
-      return handler(...args);
+      handler(...args);
     }
     let handlersByThisEventName = handlersByEvent.get(eventName);
     if (handlersByThisEventName == null) {
@@ -75,7 +90,10 @@ module.exports = (redisClient, redisPubSubScope) => {
       handlersByEvent.set(eventName, handlersByThisEventName);
     }
     handlersByThisEventName.set(handler, actualHandler);
-    runWhenConnected(() => subscriber.pSubscribe(getPrefixedEventName(eventName), actualHandler));
+    runWhenConnected(() => {
+      subscriber.on("pmessage", actualHandler);
+      return subscriber.psubscribe(getPrefixedEventName(eventName));
+    });
   }
   /**
    * Add an event listener
@@ -129,7 +147,7 @@ module.exports = (redisClient, redisPubSubScope) => {
     }
     return runWhenConnected(() => {
       handlersByEvent.delete(eventName);
-      return subscriber.pUnsubscribe(getPrefixedEventName(eventName));
+      return subscriber.punsubscribe(getPrefixedEventName(eventName));
     });
   }
   return {
diff --git a/packages/@uppy/companion/lib/server/redis.d.ts b/packages/@uppy/companion/lib/server/redis.d.ts
index ba5aa6f..b4637d1 100644
--- a/packages/@uppy/companion/lib/server/redis.d.ts
+++ b/packages/@uppy/companion/lib/server/redis.d.ts
@@ -1 +1,6 @@
-export function client(companionOptions: any): any;
+export function client({ redisUrl, redisOptions }?: {
+    redisUrl: any;
+    redisOptions: any;
+}): Redis;
+import Redis_1 = require("ioredis/built/Redis");
+import Redis = Redis_1.default;
diff --git a/packages/@uppy/companion/lib/server/redis.js b/packages/@uppy/companion/lib/server/redis.js
index 14e804f..13ebd5a 100644
--- a/packages/@uppy/companion/lib/server/redis.js
+++ b/packages/@uppy/companion/lib/server/redis.js
@@ -1,37 +1,30 @@
 "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
-const redis = require("redis");
+const Redis = require("ioredis").default;
 const logger = require("./logger");
+/** @type {import('ioredis').Redis} */
 let redisClient;
 /**
  * A Singleton module that provides a single redis client through out
  * the lifetime of the server
  *
- * @param {{ redisUrl?: string, redisOptions?: Record<string, any> }} [companionOptions] options
+ * @param {string} [redisUrl] ioredis url
+ * @param {Record<string, any>} [redisOptions] ioredis client options
  */
-function createClient(companionOptions) {
+function createClient(redisUrl, redisOptions) {
   if (!redisClient) {
-    const { redisUrl, redisOptions } = companionOptions;
-    redisClient = redis.createClient({
-      ...redisOptions,
-      ...(redisUrl && { url: redisUrl }),
-    });
-    redisClient.on("error", err => logger.error("redis error", err));
-    (async () => {
-      try {
-        // fire and forget.
-        // any requests made on the client before connection is established will be auto-queued by node-redis
-        await redisClient.connect();
-      } catch (err) {
-        logger.error(err.message, "redis.error");
-      }
-    })();
+    if (redisUrl) {
+      redisClient = new Redis(redisUrl, redisOptions);
+    } else {
+      redisClient = new Redis(redisOptions);
+    }
+    redisClient.on("error", err => logger.error("redis error", err.toString()));
   }
   return redisClient;
 }
-module.exports.client = (companionOptions) => {
-  if (!companionOptions?.redisUrl && !companionOptions?.redisOptions) {
+module.exports.client = ({ redisUrl, redisOptions } = { redisUrl: undefined, redisOptions: undefined }) => {
+  if (!redisUrl && !redisOptions) {
     return redisClient;
   }
-  return createClient(companionOptions);
+  return createClient(redisUrl, redisOptions);
 };
diff --git a/packages/@uppy/companion/lib/standalone/helper.js b/packages/@uppy/companion/lib/standalone/helper.js
index 9c21b17..7731990 100644
--- a/packages/@uppy/companion/lib/standalone/helper.js
+++ b/packages/@uppy/companion/lib/standalone/helper.js
@@ -143,9 +143,9 @@ const getConfigFromEnv = () => {
       ? parseInt(process.env.COMPANION_PERIODIC_PING_COUNT, 10)
       : undefined,
     filePath: process.env.COMPANION_DATADIR,
-    redisUrl: process.env.COMPANION_REDIS_URL,
     redisPubSubScope: process.env.COMPANION_REDIS_PUBSUB_SCOPE,
-    //  redisOptions refers to https://www.npmjs.com/package/redis#options-object-properties
+    redisUrl: process.env.COMPANION_REDIS_URL,
+    //  redisOptions refers to https://redis.github.io/ioredis/index.html#RedisOptions
     redisOptions: (() => {
       try {
         if (!process.env.COMPANION_REDIS_OPTIONS) {

[socket-security]

👍 Dependency issues cleared. Learn more about Socket for GitHub ↗︎

This PR previously contained dependency changes with security issues that have been resolved, removed, or ignored.

View full report↗︎

[mifi]

@dschmidt is this a typo?

      subscriber.off('message', actualHandler)

shouldn't it instead be

      subscriber.off('pmessage', actualHandler)

[dschmidt]

I don't recall from the top of my head, but it pretty much looks like

[Murderlon]

Are there any docs changes needed? Do we have e2e tests covering this?

[mifi]

end to end tests initially broke and I then fixed the bug, so i would say that the e2e tests cover redis if that's what you were thinking about.

COMPANION_REDIS_OPTIONS is different now, because the options are passed to a different library. but I see that it's not even documented before this PR. I will add it

[dschmidt]

🎉 🎉 🎉

Awesome! Thank you so much for bringing this over the finish line, @mifi <3

[mifi]

thank you too!