fix(redis): Tolerate Redis reconnections better

postalsys · Jan 26, 2025 · 5d3627c · 5d3627c
1 parent 647651c
commit 5d3627c
Show file tree

Hide file tree

Showing 3 changed files with 53 additions and 49 deletions.
diff --git a/lib/db.js b/lib/db.js
@@ -9,8 +9,7 @@ const fs = require('fs');
 const config = require('wild-config');
 const pathlib = require('path');
 const redisUrl = require('./redis-url');
-//const packageData = require('../package.json');
-//const { threadId } = require('worker_threads');
+const { threadId } = require('worker_threads');
 const logger = require('./logger');
 const { REDIS_PREFIX } = require('./consts');
 const Path = require('path');
@@ -50,16 +49,15 @@ const REDIS_CONF = Object.assign(
         showFriendlyErrorStack: true,
         retryStrategy(times) {
             const delay = !times ? 1000 : Math.min(2 ** times * 500, 15 * 1000);
-            logger.trace({ msg: 'Connection retry', times, delay });
+            logger.trace({ msg: 'Connection retry', isMainThread, threadId, times, delay });
             return delay;
         },
         reconnectOnError(err) {
-            logger.fatal({ msg: 'Redis connection error', err });
+            logger.fatal({ msg: 'Redis connection error', isMainThread, threadId, err });
             // always try to reconnect
             return true;
-        }
-        // Setting connection name triggers CLIENT.SETNAME command which is not supported by many managed hosts
-        //connectionName: `${packageData.name}@${packageData.version}[${process.pid}${threadId ? `:${threadId}` : ''}]`
+        },
+        offlineQueue: true
     },
     typeof redisConf === 'string' ? redisUrl(redisConf) : redisConf || {}
 );
@@ -104,13 +102,10 @@ const getRedisURL = (masked = true) => {
 
 const redis = new Redis(REDIS_CONF);
 
+const reqisQueue = new Redis(REDIS_CONF);
+
 module.exports.queueConf = {
-    connection: Object.assign(
-        {
-            //connectionName: `${REDIS_CONF.connectionName}[notify]`
-        },
-        REDIS_CONF
-    ),
+    connection: reqisQueue,
     prefix: `${REDIS_PREFIX}bull`
 };
 
@@ -247,41 +242,43 @@ redis.on('connect', () => {
     redisConnected = true;
 });
 
-redis.on('error', err => {
-    if (/NOAUTH/.test(err.message)) {
-        if (REDIS_CONF.password) {
-            return showRedisError('Redis requires a valid password', err, true);
-        } else {
-            return showRedisError('Redis password is required but not provided', err, true);
+for (let redisClient of [redis, reqisQueue]) {
+    redisClient.on('error', err => {
+        if (/NOAUTH/.test(err.message)) {
+            if (REDIS_CONF.password) {
+                return showRedisError('Redis requires a valid password', err, true);
+            } else {
+                return showRedisError('Redis password is required but not provided', err, true);
+            }
         }
-    }
-
-    if (/WRONGPASS/.test(err.message)) {
-        return showRedisError('Provided Redis password was not accepted', err, true);
-    }
 
-    switch (err.code) {
-        case 'ECONNREFUSED':
-            return showRedisError(
-                'Can not connect to the database. Redis might not be running. Are you using correct hostname and port values?',
-                err,
-                !redisConnected
-            );
-
-        case 'ETIMEDOUT':
-            return showRedisError(
-                'Connection to the database timed out. Seems like you are firewalled. Are you using correct hostname and port values?',
-                err,
-                !redisConnected
-            );
-
-        case 'ReplyError':
-            if (/MISCONF/.test(err.message)) {
-                return showRedisError(false, err, true);
-            }
-            return showRedisError(false, err);
+        if (/WRONGPASS/.test(err.message)) {
+            return showRedisError('Provided Redis password was not accepted', err, true);
+        }
 
-        default:
-            return showRedisError(false, err);
-    }
-});
+        switch (err.code) {
+            case 'ECONNREFUSED':
+                return showRedisError(
+                    'Can not connect to the database. Redis might not be running. Are you using correct hostname and port values?',
+                    err,
+                    !redisConnected
+                );
+
+            case 'ETIMEDOUT':
+                return showRedisError(
+                    'Connection to the database timed out. Seems like you are firewalled. Are you using correct hostname and port values?',
+                    err,
+                    !redisConnected
+                );
+
+            case 'ReplyError':
+                if (/MISCONF/.test(err.message)) {
+                    return showRedisError(false, err, true);
+                }
+                return showRedisError(false, err);
+
+            default:
+                return showRedisError(false, err);
+        }
+    });
+}
diff --git a/server.js b/server.js
@@ -551,7 +551,9 @@ const countUnassignment = async account => {
         }
     }
 
-    await redis.hSetExists(`${REDIS_PREFIX}iad:${account}`, 'state', 'disconnected');
+    redis.hSetExists(`${REDIS_PREFIX}iad:${account}`, 'state', 'disconnected').catch(err => {
+        logger.error({ msg: 'Failed to post update account state', account, state: 'disconnected', err });
+    });
 
     for (let worker of workers.get('api')) {
         let callPayload = {
@@ -754,6 +756,7 @@ let spawnWorker = async type => {
                             })
                             .finally(() => {
                                 unassigned.add(account);
+
                                 if (shouldReassign) {
                                     assignAccounts().catch(err => logger.error({ msg: 'Failed to assign accounts', n: 1, err }));
                                 }
@@ -1063,6 +1066,7 @@ async function assignAccounts() {
     if (assigning) {
         return false;
     }
+
     assigning = true;
     try {
         if (!unassigned) {

diff --git a/workers/imap.js b/workers/imap.js
@@ -823,6 +823,9 @@ class ConnectionHandler {
 let connectionHandler = new ConnectionHandler();
 
 async function main() {
+    // Try to run a redis command to be sure that Redis is connected
+    await redis.ping();
+
     logger.info({ msg: 'Started IMAP worker thread', version: packageData.version });
     await connectionHandler.init();
 }