improve dbs.toml, improve fanout, remove .multi() call

Author: NickOvt

config/dbs.toml

@@ -26,8 +26,9 @@ sender = "zone-mta"
 
 #queued="mail"
 
-# Id to use for distributing notifications
-workerId = false # use os.hostname() instead
+# Worker identifier prefix for distributing notifications.
+# Each process appends :PID automatically. Set to false to use os.hostname().
+workerId = false
 
 [redis]
 host = "127.0.0.1"

lib/imap-notifier.js

@@ -337,12 +337,8 @@ class ImapNotifier extends EventEmitter {
             let minScore = Date.now() - USER_WORKERS_TTL_MS;
             // send message only to workers that actually service this user
             this.redis
-                .multi()
-                .zremrangebyscore(key, 0, minScore) // remove stale workers
                 .zrangebyscore(key, minScore, '+inf') // get active workers
-                .exec()
-                .then(res => {
-                    let workers = (res && res[1] && res[1][1]) || [];
+                .then(workers => {
                     if (!workers || !workers.length) {
                         return;
                     }
@@ -386,7 +382,12 @@ class ImapNotifier extends EventEmitter {
 
     _registerUserWorker(userId) {
         let key = this._getUserWorkersKey(userId);
-        this.redis.zadd(key, Date.now(), this.workerId).catch(() => false);
+        this.redis
+            .pipeline()
+            .zadd(key, Date.now(), this.workerId)
+            .expire(key, USER_WORKERS_TTL + 60)
+            .exec()
+            .catch(() => false);
     }
 
     _unregisterUserWorker(userId) {
@@ -403,9 +404,12 @@ class ImapNotifier extends EventEmitter {
         let redis = this._userRegistryState.redis || this.redis;
         let pipeline = redis.pipeline();
         let now = Date.now();
+        let minScore = now - USER_WORKERS_TTL_MS;
         for (let userId of counts.keys()) {
             let key = this._getUserWorkersKey(userId);
+            pipeline.zremrangebyscore(key, 0, minScore);
             pipeline.zadd(key, now, this.workerId);
+            pipeline.expire(key, USER_WORKERS_TTL + 60);
         }
         pipeline.exec().catch(() => false);
     }