chore: Adds name to Redis connections for debugging (#3982)

* chore: Adds name to Redis connections for debugging, minor associated refactoring

* Upgrade bull, ioredis

* Add pid to redis connection name in development

server/env.ts

@@ -22,6 +22,7 @@ import {
 import { languages } from "@shared/i18n";
 import { CannotUseWithout } from "@server/utils/validators";
 import Deprecated from "./models/decorators/Deprecated";
+import { getArg } from "./utils/args";
 
 export class Environment {
   private validationPromise;
@@ -202,9 +203,14 @@ export class Environment {
   /**
    * A comma separated list of which services should be enabled on this
    * instance – defaults to all.
+   *
+   * If a services flag is passed it takes priority over the environment variable
+   * for example: --services=web,worker
    */
   public SERVICES =
-    process.env.SERVICES ?? "collaboration,websockets,worker,web";
+    getArg("services") ??
+    process.env.SERVICES ??
+    "collaboration,websockets,worker,web";
 
   /**
    * Auto-redirect to https in production. The default is true but you may set

server/index.ts

@@ -27,16 +27,10 @@ import {
 } from "./utils/startup";
 import { checkUpdates } from "./utils/updates";
 
-// If a services flag is passed it takes priority over the environment variable
-// for example: --services=web,worker
-const normalizedServiceFlag = getArg("services");
-
 // The default is to run all services to make development and OSS installations
 // easier to deal with. Separate services are only needed at scale.
 const serviceNames = uniq(
-  (normalizedServiceFlag || env.SERVICES)
-    .split(",")
-    .map((service) => service.trim())
+  env.SERVICES.split(",").map((service) => service.trim())
 );
 
 // The number of processes to run, defaults to the number of CPU's available

server/redis.ts

@@ -3,8 +3,14 @@ import { defaults } from "lodash";
 import env from "@server/env";
 import Logger from "@server/logging/Logger";
 
-const defaultOptions = {
+type RedisAdapterOptions = Redis.RedisOptions & {
+  /** Suffix to append to the connection name that will be displayed in Redis */
+  connectionNameSuffix?: string;
+};
+
+const defaultOptions: Redis.RedisOptions = {
   maxRetriesPerRequest: 20,
+  enableReadyCheck: false,
 
   retryStrategy(times: number) {
     Logger.warn(`Retrying redis connection: attempt ${times}`);
@@ -21,9 +27,25 @@ const defaultOptions = {
 };
 
 export default class RedisAdapter extends Redis {
-  constructor(url: string | undefined) {
+  constructor(
+    url: string | undefined,
+    { connectionNameSuffix, ...options }: RedisAdapterOptions = {}
+  ) {
+    /**
+     * For debugging. The connection name is based on the services running in
+     * this process. Note that this does not need to be unique.
+     */
+    const connectionNamePrefix =
+      env.ENVIRONMENT === "development" ? process.pid : "outline";
+    const connectionName =
+      `${connectionNamePrefix}:${env.SERVICES.replace(/,/g, "-")}` +
+      (connectionNameSuffix ? `:${connectionNameSuffix}` : "");
+
     if (!url || !url.startsWith("ioredis://")) {
-      super(env.REDIS_URL, defaultOptions);
+      super(
+        env.REDIS_URL,
+        defaults(options, { connectionName }, defaultOptions)
+      );
     } else {
       let customOptions = {};
       try {
@@ -34,8 +56,9 @@ export default class RedisAdapter extends Redis {
       }
 
       try {
-        const mergedOptions = defaults(defaultOptions, customOptions);
-        super(mergedOptions);
+        super(
+          defaults(options, { connectionName }, customOptions, defaultOptions)
+        );
       } catch (error) {
         throw new Error(`Failed to initialize redis client: ${error}`);
       }
@@ -47,14 +70,25 @@ export default class RedisAdapter extends Redis {
     this.setMaxListeners(100);
   }
 
-  private static _client: RedisAdapter;
-  private static _subscriber: RedisAdapter;
+  private static client: RedisAdapter;
+  private static subscriber: RedisAdapter;
 
   public static get defaultClient(): RedisAdapter {
-    return this._client || (this._client = new this(env.REDIS_URL));
+    return (
+      this.client ||
+      (this.client = new this(env.REDIS_URL, {
+        connectionNameSuffix: "client",
+      }))
+    );
   }
 
   public static get defaultSubscriber(): RedisAdapter {
-    return this._subscriber || (this._subscriber = new this(env.REDIS_URL));
+    return (
+      this.subscriber ||
+      (this.subscriber = new this(env.REDIS_URL, {
+        maxRetriesPerRequest: null,
+        connectionNameSuffix: "subscriber",
+      }))
+    );
   }
 }

server/utils/queue.ts

@@ -9,6 +9,9 @@ export function createQueue(
   defaultJobOptions?: Partial<Queue.JobOptions>
 ) {
   const prefix = `queue.${snakeCase(name)}`;
+
+  // Notes on reusing Redis connections for Bull:
+  // https://github.com/OptimalBits/bull/blob/b6d530f72a774be0fd4936ddb4ad9df3b183f4b6/PATTERNS.md#reusing-redis-connections
   const queue = new Queue(name, {
     createClient(type) {
       switch (type) {
@@ -18,11 +21,16 @@ export function createQueue(
         case "subscriber":
           return Redis.defaultSubscriber;
 
+        case "bclient":
+          return new Redis(env.REDIS_URL, {
+            maxRetriesPerRequest: null,
+            connectionNameSuffix: "bull",
+          });
+
         default:
-          return new Redis(env.REDIS_URL);
+          throw new Error(`Unexpected connection type: ${type}`);
       }
     },
-
     defaultJobOptions: {
       removeOnComplete: true,
       removeOnFail: true,