diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index a80df812d..dfbe5dee5 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -58,3 +58,74 @@ jobs: env: APPCONF_dbs_redis: redis://127.0.0.1:6379/1 APPCONF_dbs_mongodb: 'mongodb://127.0.0.1:27017/wildduck-test?authSource=admin' + + test-redis-cluster: + runs-on: ubuntu-latest + services: + redis-cluster: + image: grokzen/redis-cluster:7.0.10 + env: + IP: 127.0.0.1 + INITIAL_PORT: 7000 + MASTERS: 3 + SLAVES_PER_MASTER: 0 + options: >- + --health-cmd "redis-cli -c -p 7000 ping" + --health-interval 10s + --health-timeout 5s + --health-retries 20 + ports: + - 7000:7000 + - 7001:7001 + - 7002:7002 + + steps: + - uses: actions/checkout@v6 + + - name: Start MongoDB + uses: supercharge/mongodb-github-action@1.12.1 + with: + mongodb-version: 6.0 + mongodb-port: 27017 + mongodb-db: wildduck-test + + - name: Use Node.js 24.x + uses: actions/setup-node@v6 + with: + node-version: 24.x + + - name: Install mongosh, mpop and redis-cli + run: | + sudo apt-get update + sudo apt-get install -y wget gnupg mpop redis-tools + wget -qO - https://www.mongodb.org/static/pgp/server-6.0.asc | sudo apt-key add - + echo "deb [ arch=amd64,arm64 ] https://repo.mongodb.org/apt/ubuntu jammy/mongodb-org/6.0 multiverse" | sudo tee /etc/apt/sources.list.d/mongodb-org-6.0.list + sudo apt-get update + sudo apt-get install -y mongodb-mongosh + + - run: which mongosh + - run: which mpop + - run: which redis-cli + - run: npm install + + - name: Wait for Redis Cluster + run: | + for port in 7000 7001 7002; do + for i in $(seq 1 60); do + redis-cli -c -h 127.0.0.1 -p "$port" ping && break + sleep 1 + done + done + redis-cli -c -h 127.0.0.1 -p 7000 cluster info | grep 'cluster_state:ok' + + - name: Reset DB and Redis cluster + run: | + mongosh --eval 'db.dropDatabase()' wildduck-test + for port in 7000 7001 7002; do + redis-cli -c -h 127.0.0.1 -p "$port" flushdb + done + + - name: Run tests against Redis cluster + run: NODE_ENV=test-cluster npx grunt + env: + APPCONF_dbs_mongodb: 'mongodb://127.0.0.1:27017/wildduck-test?authSource=admin' diff --git a/Gruntfile.js b/Gruntfile.js index 2318e2170..fe0710ded 100644 --- a/Gruntfile.js +++ b/Gruntfile.js @@ -1,7 +1,5 @@ 'use strict'; -process.env.NODE_ENV = 'test'; - module.exports = function (grunt) { // Project configuration. grunt.initConfig({ diff --git a/api.js b/api.js index faaec158e..805aa9a96 100644 --- a/api.js +++ b/api.js @@ -593,7 +593,7 @@ module.exports = done => { settingsRoutes(db, server, settingsHandler); healthRoutes(db, server, loggelf); - if (process.env.NODE_ENV === 'test') { + if (process.env.NODE_ENV === 'test' || process.env.NODE_ENV === 'test-cluster') { server.get( { name: 'api-methods', path: '/api-methods' }, tools.responseWrapper(async (req, res) => { diff --git a/bin/access-tokens b/bin/access-tokens index 179a84921..e883a65b1 100755 --- a/bin/access-tokens +++ b/bin/access-tokens @@ -40,10 +40,7 @@ let argv = yargs await dbconnect(); let accessToken = crypto.randomBytes(20).toString('hex'); - let tokenHash = crypto - .createHash('sha256') - .update(accessToken) - .digest('hex'); + let tokenHash = crypto.createHash('sha256').update(accessToken).digest('hex'); let key = 'tn:token:' + tokenHash; let tokenData = { @@ -63,11 +60,15 @@ let argv = yargs .digest('hex') }; - await db.redis - .multi() - .hmset(key, tokenData) - .sadd('tn:user:' + argv.user, tokenHash) - .exec(); + if (db.redis.isCluster) { + await Promise.all([db.redis.hmset(key, tokenData), db.redis.sadd('tn:user:' + argv.user, tokenHash)]); + } else { + await db.redis + .multi() + .hmset(key, tokenData) + .sadd('tn:user:' + argv.user, tokenHash) + .exec(); + } console.error('Generated access token for %s[%s]:', tokenData.user, tokenData.role); @@ -92,10 +93,7 @@ let argv = yargs let accessToken = argv.token; - let tokenHash = crypto - .createHash('sha256') - .update(accessToken) - .digest('hex'); + let tokenHash = crypto.createHash('sha256').update(accessToken).digest('hex'); let key = 'tn:token:' + tokenHash; @@ -132,13 +130,20 @@ let argv = yargs process.exit(); } - let query = await db.redis.multi().del('tn:user:' + user); + if (db.redis.isCluster) { + await db.redis.del('tn:user:' + user); - tokens.forEach(tokenHash => { - query = query.del('tn:token:' + tokenHash); - }); + for (let tokenHash of tokens) { + await db.redis.del('tn:token:' + tokenHash); + } + } else { + let query = db.redis.multi().del('tn:user:' + user); + tokens.forEach(tokenHash => { + query = query.del('tn:token:' + tokenHash); + }); - await query.exec(); + await query.exec(); + } console.error('Deleted %s tokens for %s', tokens.length, user); diff --git a/config/dbs.toml b/config/dbs.toml index c67187236..0ebf23aef 100644 --- a/config/dbs.toml +++ b/config/dbs.toml @@ -49,3 +49,25 @@ db = 3 # [[redis.sentinels]] # host="54.36.85.115" # port=26379 + +## Connect to Redis Cluster instead of a single master +# [redis] +# cluster = true +# password = "" # global/fallback password +# [[redis.nodes]] +# host="54.36.85.113" +# port=26379 +# password="" # per node password +# [[redis.nodes]] +# host = "127.0.0.1" +# port = 7000 +# [[redis.nodes]] +# host = "127.0.0.1" +# port = 7001 +# [[redis.nodes]] +# host = "127.0.0.1" +# port = 7002 +# [[redis.nodes]] +# host="54.36.85.115" +# port=26379 +# password="" diff --git a/config/test-cluster.toml b/config/test-cluster.toml new file mode 100644 index 000000000..acb1cdbdb --- /dev/null +++ b/config/test-cluster.toml @@ -0,0 +1,41 @@ +[api] +port = 8080 +host = "127.0.0.1" +secure = false + +[log] +level = "silly" + +[dbs] +# mongodb connection string for the main database +mongo = "mongodb://127.0.0.1:27017/wildduck-test" + +dbname = "wildduck-test" + +[dbs.redis] +cluster = true +password = "" + +[[dbs.redis.nodes]] +host = "127.0.0.1" +port = 7000 + +[[dbs.redis.nodes]] +host = "127.0.0.1" +port = 7001 + +[[dbs.redis.nodes]] +host = "127.0.0.1" +port = 7002 + +[imap] +port = 9993 +host = "127.0.0.1" + +[lmtp] +enabled = true +port = 2424 + +[pwned] +enabled = true +type = "softfail" diff --git a/indexer.js b/indexer.js index 33aef603e..b44a92b86 100644 --- a/indexer.js +++ b/indexer.js @@ -12,6 +12,7 @@ const counters = require('./lib/counters'); const { ObjectId } = require('mongodb'); const libmime = require('libmime'); const punycode = require('punycode.js'); +const tools = require('./lib/tools'); const { getClient } = require('./lib/elasticsearch'); const { normalizeLoggelfMessage } = require('./lib/loggelf-message'); @@ -284,14 +285,15 @@ function indexingJob(esclient) { const dateKeyTdy = new Date().toISOString().substring(0, 10).replace(/-/g, ''); const dateKeyYdy = new Date(Date.now() - 24 * 3600 * 1000).toISOString().substring(0, 10).replace(/-/g, ''); - const tombstoneTdy = `indexer:tomb:${dateKeyTdy}`; - const tombstoneYdy = `indexer:tomb:${dateKeyYdy}`; + const tombstoneTag = tools.redisHashTag(db.redis, 'indexer:tomb'); + const tombstoneTdy = `${tombstoneTag}:${dateKeyTdy}`; + const tombstoneYdy = `${tombstoneTag}:${dateKeyYdy}`; switch (data.action) { case 'new': { // check tombstone for race conditions (might be already deleted) - let [[err1, isDeleted1], [err2, isDeleted2]] = await db.redis + const [[err1, isDeleted1], [err2, isDeleted2]] = await db.redis .multi() .sismember(tombstoneTdy, data.message) .sismember(tombstoneYdy, data.message) diff --git a/lib/api/addresses.js b/lib/api/addresses.js index 7dc09667b..ae928648a 100644 --- a/lib/api/addresses.js +++ b/lib/api/addresses.js @@ -2397,11 +2397,12 @@ module.exports = (db, server, userHandler, settingsHandler) => { let response; try { + const addressKey = tools.redisHashTag(db.redis, addressData._id.toString()); response = await db.redis .multi() // sending counters are stored in Redis - .get('wdf:' + addressData._id.toString()) - .ttl('wdf:' + addressData._id.toString()) + .get(`wdf:${addressKey}`) + .ttl(`wdf:${addressKey}`) .exec(); } catch (err) { // ignore @@ -2565,11 +2566,12 @@ module.exports = (db, server, userHandler, settingsHandler) => { let response; try { + const addressKey = tools.redisHashTag(db.redis, addressData._id.toString()); response = await db.redis .multi() // sending counters are stored in Redis - .get('wdf:' + addressData._id.toString()) - .ttl('wdf:' + addressData._id.toString()) + .get(`wdf:${addressKey}`) + .ttl(`wdf:${addressKey}`) .exec(); } catch (err) { // ignore diff --git a/lib/api/mailboxes.js b/lib/api/mailboxes.js index 9f4662e4f..22cb47687 100644 --- a/lib/api/mailboxes.js +++ b/lib/api/mailboxes.js @@ -257,7 +257,7 @@ module.exports = (db, server, mailboxHandler) => { counterOps.push( (async () => { try { - total = await getMailboxCounter(db, mailboxData._id); + total = await getMailboxCounter(db, mailboxData._id, false, mailboxData.user); } catch (err) { // ignore } @@ -268,7 +268,7 @@ module.exports = (db, server, mailboxHandler) => { counterOps.push( (async () => { try { - unseen = await getMailboxCounter(db, mailboxData._id, 'unseen'); + unseen = await getMailboxCounter(db, mailboxData._id, 'unseen', mailboxData.user); } catch (err) { // ignore } @@ -533,13 +533,13 @@ module.exports = (db, server, mailboxHandler) => { let total, unseen; try { - total = await getMailboxCounter(db, mailboxData._id); + total = await getMailboxCounter(db, mailboxData._id, false, mailboxData.user); } catch (err) { // ignore } try { - unseen = await getMailboxCounter(db, mailboxData._id, 'unseen'); + unseen = await getMailboxCounter(db, mailboxData._id, 'unseen', mailboxData.user); } catch (err) { // ignore } diff --git a/lib/api/messages.js b/lib/api/messages.js index 064d6f0df..6b7a22e76 100644 --- a/lib/api/messages.js +++ b/lib/api/messages.js @@ -524,7 +524,7 @@ module.exports = (db, server, messageHandler, userHandler, storageHandler, setti filter.unseen = filterUnseen; } - let total = await getFilteredMessageCount(filter); + let total = await getFilteredMessageCount(filter, user); let opts = { limit, @@ -869,7 +869,7 @@ module.exports = (db, server, messageHandler, userHandler, storageHandler, setti query = prepared.query; } - let total = await getFilteredMessageCount(filter); + let total = await getFilteredMessageCount(filter, user); log.verbose('API', 'Searching %s', JSON.stringify(filter)); let opts = { @@ -2211,8 +2211,9 @@ module.exports = (db, server, messageHandler, userHandler, storageHandler, setti } try { + const mailboxKey = `${tools.redisHashTag(db.redis, user.toString())}:${mailbox}`; // clear counters - await db.redis.multi().del(`total:${mailbox}`).del(`unseen:${mailbox}`).exec(); + await db.redis.multi().del(`total:${mailboxKey}`).del(`unseen:${mailboxKey}`).exec(); } catch (err) { // ignore } @@ -3140,7 +3141,12 @@ module.exports = (db, server, messageHandler, userHandler, storageHandler, setti let limitCheck; try { - limitCheck = await messageHandler.counters.asyncTTLCounter('wdr:' + userData._id.toString(), 0, maxRecipients, false); + limitCheck = await messageHandler.counters.asyncTTLCounter( + `wdr:${tools.redisHashTag(messageHandler.redis, userData._id.toString())}`, + 0, + maxRecipients, + false + ); } catch (err) { log.error('API', 'Failed to check draft submit rate limit for user=%s message=%s error=%s', userData._id, messageData._id, err.message); res.status(500); @@ -3913,10 +3919,10 @@ module.exports = (db, server, messageHandler, userHandler, storageHandler, setti }) ); - async function getFilteredMessageCount(filter) { + async function getFilteredMessageCount(filter, user) { if (Object.keys(filter).length === 1 && filter.mailbox) { // Try to use cached value to get the count - return await getMailboxCounter(db, filter.mailbox); + return await getMailboxCounter(db, filter.mailbox, false, user); } return await db.database.collection('messages').countDocuments(filter); @@ -4191,13 +4197,19 @@ module.exports = (db, server, messageHandler, userHandler, storageHandler, setti } // Update counters only after message has been succesfully sent and not rejected - messageHandler.counters.ttlcounter('wdr:' + userData._id.toString(), envelope.to.length, maxRecipients, false, err => { - if (err) { - err.responseCode = 500; - err.code = 'InternalDatabaseError'; - return reject(err); + messageHandler.counters.ttlcounter( + `wdr:${tools.redisHashTag(messageHandler.redis, userData._id.toString())}`, + envelope.to.length, + maxRecipients, + false, + err => { + if (err) { + err.responseCode = 500; + err.code = 'InternalDatabaseError'; + return reject(err); + } } - }); + ); // Update addressregister - output can be ignored as it is not an important operation messageHandler.updateAddressRegister( diff --git a/lib/api/submit.js b/lib/api/submit.js index a527c872a..310814fff 100644 --- a/lib/api/submit.js +++ b/lib/api/submit.js @@ -432,92 +432,98 @@ module.exports = (db, server, messageHandler, userHandler, settingsHandler) => { const recipientCount = compiledEnvelope.to.length; - messageHandler.counters.ttlcounter('wdr:' + userData._id.toString(), 0, maxRecipients, false, (err, result) => { - if (err) { - err.responseCode = 500; - err.code = 'InternalDatabaseError'; - return callback(err); - } + messageHandler.counters.ttlcounter( + `wdr:${tools.redisHashTag(messageHandler.redis, userData._id.toString())}`, + 0, + maxRecipients, + false, + (err, result) => { + if (err) { + err.responseCode = 500; + err.code = 'InternalDatabaseError'; + return callback(err); + } - let success = result.success; - let sent = result.value; - let ttl = result.ttl; + let success = result.success; + let sent = result.value; + let ttl = result.ttl; + + let ttlHuman = false; + if (ttl && ttl > 0) { + if (ttl < 60) { + ttlHuman = ttl + ' seconds'; + } else if (ttl < 3600) { + ttlHuman = Math.round(ttl / 60) + ' minutes'; + } else { + ttlHuman = Math.round(ttl / 3600) + ' hours'; + } + } - let ttlHuman = false; - if (ttl && ttl > 0) { - if (ttl < 60) { - ttlHuman = ttl + ' seconds'; - } else if (ttl < 3600) { - ttlHuman = Math.round(ttl / 60) + ' minutes'; - } else { - ttlHuman = Math.round(ttl / 3600) + ' hours'; + if (!success || sent + recipientCount > maxRecipients) { + log.info('API', 'RCPTDENY denied sent=%s allowed=%s expires=%ss.', sent, maxRecipients, ttl); + let err = new Error( + 'You reached a daily sending limit for your account' + + (ttl && ttl > 0 ? '. Limit expires in ' + ttlHuman : '') + ); + err.responseCode = 403; + err.code = 'RateLimitedError'; + return setImmediate(() => callback(err)); } - } - if (!success || sent + recipientCount > maxRecipients) { - log.info('API', 'RCPTDENY denied sent=%s allowed=%s expires=%ss.', sent, maxRecipients, ttl); - let err = new Error( - 'You reached a daily sending limit for your account' + - (ttl && ttl > 0 ? '. Limit expires in ' + ttlHuman : '') - ); - err.responseCode = 403; - err.code = 'RateLimitedError'; - return setImmediate(() => callback(err)); - } + // push message to outbound queue + let message = maildrop.push( + { + user: userData._id, + userEmail: userData.address, + parentId: messageId, + reason: 'submit', + from: compiledEnvelope.from, + to: compiledEnvelope.to, + sendTime, + origin: options.ip, + passwordType: 'master', + runPlugins: true, + mtaRelay: userData.mtaRelay || false + }, + (err, ...args) => { + if (err || !args[0]) { + if (err) { + if (!err.code && err.name === 'SMTPReject') { + err.code = 'MessageRejected'; + } - // push message to outbound queue - let message = maildrop.push( - { - user: userData._id, - userEmail: userData.address, - parentId: messageId, - reason: 'submit', - from: compiledEnvelope.from, - to: compiledEnvelope.to, - sendTime, - origin: options.ip, - passwordType: 'master', - runPlugins: true, - mtaRelay: userData.mtaRelay || false - }, - (err, ...args) => { - if (err || !args[0]) { - if (err) { - if (!err.code && err.name === 'SMTPReject') { - err.code = 'MessageRejected'; + err.code = err.code || 'ERRCOMPOSE'; } - - err.code = err.code || 'ERRCOMPOSE'; + err.responseCode = 500; + return callback(err, ...args); } - err.responseCode = 500; - return callback(err, ...args); - } - messageHandler.counters.ttlcounter( - 'wdr:' + userData._id.toString(), - recipientCount, - maxRecipients, - false, - err => { - if (err) { - err.responseCode = 500; - err.code = 'InternalDatabaseError'; - return setImmediate(() => callback(err)); + messageHandler.counters.ttlcounter( + `wdr:${tools.redisHashTag(messageHandler.redis, userData._id.toString())}`, + recipientCount, + maxRecipients, + false, + err => { + if (err) { + err.responseCode = 500; + err.code = 'InternalDatabaseError'; + return setImmediate(() => callback(err)); + } + + let outbound = args[0].id; + return next(null, outbound); } + ); + } + ); - let outbound = args[0].id; - return next(null, outbound); - } - ); + if (message) { + let stream = compiled.createReadStream(); + stream.once('error', err => message.emit('error', err)); + stream.pipe(collector).pipe(message); } - ); - - if (message) { - let stream = compiled.createReadStream(); - stream.once('error', err => message.emit('error', err)); - stream.pipe(collector).pipe(message); } - }); + ); }; addToDeliveryQueue((err, outbound) => { diff --git a/lib/api/updates.js b/lib/api/updates.js index 0bad4c944..4cd9b568e 100644 --- a/lib/api/updates.js +++ b/lib/api/updates.js @@ -11,9 +11,9 @@ const base32 = require('base32.js'); const { sessSchema, sessIPSchema } = require('../schemas'); const { userId } = require('../schemas/request/general-schemas'); -const getMailboxCounterCb = (db, mailbox, type, callback) => { +const getMailboxCounterCb = (db, mailbox, type, user, callback) => { tools - .getMailboxCounter(db, mailbox, type) + .getMailboxCounter(db, mailbox, type, user) .then(sum => callback(null, sum)) .catch(err => callback(err)); }; @@ -456,11 +456,11 @@ function loadJournalStream(db, res, user, lastEventId, done, onEntry) { }); } let mailbox = new ObjectId(mailboxes[mailboxPos++]); - getMailboxCounterCb(db, mailbox, false, (err, total) => { + getMailboxCounterCb(db, mailbox, false, user, (err, total) => { if (err) { // ignore } - getMailboxCounterCb(db, mailbox, 'unseen', (err, unseen) => { + getMailboxCounterCb(db, mailbox, 'unseen', user, (err, unseen) => { if (err) { // ignore } diff --git a/lib/api/users.js b/lib/api/users.js index 285f20e36..2faefe623 100644 --- a/lib/api/users.js +++ b/lib/api/users.js @@ -1029,35 +1029,36 @@ module.exports = (db, server, userHandler, settingsHandler) => { let response; try { + const userKey = tools.redisHashTag(db.redis, userData._id.toString()); response = await db.redis .multi() // sending counters are stored in Redis // sent messages - .get('wdr:' + userData._id.toString()) - .ttl('wdr:' + userData._id.toString()) + .get(`wdr:${userKey}`) + .ttl(`wdr:${userKey}`) // forwarded messages - .get('wdf:' + userData._id.toString()) - .ttl('wdf:' + userData._id.toString()) + .get(`wdf:${userKey}`) + .ttl(`wdf:${userKey}`) // rate limited recipient - .get('rl:rcpt:' + userData._id.toString()) - .ttl('rl:rcpt:' + userData._id.toString()) + .get(`rl:rcpt:${userKey}`) + .ttl(`rl:rcpt:${userKey}`) // rate limited imap uploads - .get('iup:' + userData._id.toString()) - .ttl('iup:' + userData._id.toString()) + .get(`iup:${userKey}`) + .ttl(`iup:${userKey}`) // rate limited imap downloads - .get('idw:' + userData._id.toString()) - .ttl('idw:' + userData._id.toString()) + .get(`idw:${userKey}`) + .ttl(`idw:${userKey}`) // rate limited pop3 downloads - .get('pdw:' + userData._id.toString()) - .ttl('pdw:' + userData._id.toString()) + .get(`pdw:${userKey}`) + .ttl(`pdw:${userKey}`) - .hget('lim:imap', userData._id.toString()) + .hget(`lim:imap:${userKey}`, userData._id.toString()) .exec(); } catch (err) { diff --git a/lib/db.js b/lib/db.js index b19bda9eb..029ac2978 100644 --- a/lib/db.js +++ b/lib/db.js @@ -4,6 +4,7 @@ const config = require('@zone-eu/wild-config'); const mongodb = require('mongodb'); const Redis = require('ioredis'); const redisUrl = require('./redis-url'); +const tools = require('./tools'); const log = require('npmlog'); const errors = require('./errors'); const packageData = require('../package.json'); @@ -42,33 +43,71 @@ let getDBConnection = (main, config, callback) => { ); }; -module.exports.connect = callback => { - const REDIS_CONF = Object.assign( - { - // some defaults - maxRetriesPerRequest: null, - showFriendlyErrorStack: true, - retryStrategy(times) { - const delay = !times ? 1000 : Math.min(2 ** times * 500, 15 * 1000); - log.info('Redis', 'Connection retry times=%s delay=%s', times, delay); - return delay; - }, - connectionName: `${packageData.name}@${packageData.version}[${process.pid}]` +let getRedisConf = defaultConfig => { + const redisDefaults = { + // some defaults + maxRetriesPerRequest: null, + showFriendlyErrorStack: true, + retryStrategy(times) { + const delay = !times ? 1000 : Math.min(2 ** times * 500, 15 * 1000); + log.info('Redis', 'Connection retry times=%s delay=%s', times, delay); + return delay; }, - typeof config.dbs.redis === 'string' ? redisUrl(config.dbs.redis) : config.dbs.redis || {} - ); + connectionName: `${packageData.name}@${packageData.version}[${process.pid}]` + }; + + const redisConfig = typeof defaultConfig === 'string' ? redisUrl(defaultConfig) : defaultConfig || {}; + + if (redisConfig && redisConfig.cluster) { + let clusterConfig = Object.assign({}, redisConfig); + let nodeDefaults = {}; + + for (let key of ['password', 'username', 'tls']) { + if (typeof clusterConfig[key] !== 'undefined') { + nodeDefaults[key] = clusterConfig[key]; + } + } + + let nodes = [] + .concat(clusterConfig.nodes || []) + .map(node => Object.assign({}, nodeDefaults, node || {})) + .filter(node => node && node.host); + + delete clusterConfig.cluster; + delete clusterConfig.nodes; + delete clusterConfig.password; + delete clusterConfig.db; + delete clusterConfig.username; + delete clusterConfig.tls; + + let redisOptions = Object.assign({}, redisDefaults, nodeDefaults, clusterConfig.redisOptions || {}); + + return { + cluster: true, + connectionName: redisOptions.connectionName, + nodes, + options: Object.assign({}, clusterConfig, { redisOptions }) + }; + } + + return Object.assign({}, redisDefaults, redisConfig); +}; + +module.exports.connect = callback => { + const REDIS_CONF = getRedisConf(config.dbs.redis); module.exports.redisConfig = REDIS_CONF; + module.exports.redis = REDIS_CONF.cluster ? new Redis.Cluster(REDIS_CONF.nodes, REDIS_CONF.options) : new Redis(REDIS_CONF); module.exports.queueConf = { - connection: Object.assign({ connectionName: `${REDIS_CONF.connectionName}[notify]` }, REDIS_CONF), - prefix: `wd:bull` + connection: REDIS_CONF.cluster ? module.exports.redis : Object.assign({ connectionName: `${REDIS_CONF.connectionName}[notify]` }, REDIS_CONF), + prefix: tools.redisHashTag(REDIS_CONF, 'wd:bull') }; - module.exports.redis = new Redis(REDIS_CONF); errors.registerRedisErrorLogger(module.exports.redis, { role: 'primary', connectionName: REDIS_CONF.connectionName, - mode: Array.isArray(REDIS_CONF.sentinels) ? 'sentinel' : 'direct', - sentinelCount: Array.isArray(REDIS_CONF.sentinels) ? REDIS_CONF.sentinels.length : undefined + mode: REDIS_CONF.cluster ? 'cluster' : Array.isArray(REDIS_CONF.sentinels) ? 'sentinel' : 'direct', + sentinelCount: Array.isArray(REDIS_CONF.sentinels) ? REDIS_CONF.sentinels.length : undefined, + clusterNodeCount: REDIS_CONF.cluster && Array.isArray(REDIS_CONF.nodes) ? REDIS_CONF.nodes.length : undefined }); getDBConnection(false, config.dbs.mongo, (err, db) => { diff --git a/lib/events.js b/lib/events.js index 0b881e933..f81794b92 100644 --- a/lib/events.js +++ b/lib/events.js @@ -4,6 +4,7 @@ const { Queue } = require('bullmq'); const log = require('npmlog'); +const tools = require('./tools'); let webhooksQueue; @@ -55,7 +56,7 @@ module.exports = { if (!webhooksQueue) { webhooksQueue = new Queue('webhooks', { connection: redisClient, - prefix: `wd:bull` + prefix: tools.redisHashTag(redisClient, 'wd:bull') }); } diff --git a/lib/filter-handler.js b/lib/filter-handler.js index c55facc77..85b3b0ef7 100644 --- a/lib/filter-handler.js +++ b/lib/filter-handler.js @@ -460,15 +460,11 @@ class FilterHandler { } // check limiting counters + const userKey = tools.redisHashTag(this.messageHandler.redis, userData._id.toString()); try { - let counterResult = await this.ttlcounter( - 'wdf:' + userData._id.toString(), - forwardTargets.size, - userData.forwards || consts.MAX_FORWARDS, - false - ); + const counterResult = await this.ttlcounter(`wdf:${userKey}`, forwardTargets.size, userData.forwards || consts.MAX_FORWARDS, false); if (!counterResult.success) { - log.silly('Filter', 'FRWRDFAIL key=%s error=%s', 'wdf:' + userData._id.toString(), 'Precondition failed'); + log.silly('Filter', 'FRWRDFAIL key=%s error=%s', `wdf:${userKey}`, 'Precondition failed'); logdata.short_message = '[FRWRDFAIL] Skipped forwarding due to rate limiting'; logdata._error = 'Skipped forwarding due to rate limiting'; @@ -479,7 +475,7 @@ class FilterHandler { } } catch (err) { // failed checks, ignore - log.info('Filter', 'FRWRDFAIL key=%s error=%s', 'wdf:' + userData._id.toString(), err.message); + log.info('Filter', 'FRWRDFAIL key=%s error=%s', `wdf:${userKey}`, err.message); logdata.short_message = '[FRWRDFAIL] Skipped forwarding due to database error'; logdata._error = err.message; diff --git a/lib/handlers/on-append.js b/lib/handlers/on-append.js index 2065d5efc..784a864d0 100644 --- a/lib/handlers/on-append.js +++ b/lib/handlers/on-append.js @@ -45,7 +45,9 @@ module.exports = (server, messageHandler, userCache) => (path, flags, date, raw, return callback(err); } - messageHandler.counters.ttlcounter('iup:' + session.user.id, 0, limit, false, (err, res) => { + const counterKey = `iup:${tools.redisHashTag(db.redis, session.user.id.toString())}`; + + messageHandler.counters.ttlcounter(counterKey, 0, limit, false, (err, res) => { if (err) { return callback(err); } @@ -58,12 +60,11 @@ module.exports = (server, messageHandler, userCache) => (path, flags, date, raw, return callback(err); } - messageHandler.counters.ttlcounter('iup:' + session.user.id, raw.length, limit, false, () => { + messageHandler.counters.ttlcounter(counterKey, raw.length, limit, false, () => { flags = Array.isArray(flags) ? flags : [].concat(flags || []); (async () => { - let encryptionKey = - userData.encryptMessages && !flags.includes('\\Draft') ? tools.getUserEncryptionKey(userData) : false; + const encryptionKey = userData.encryptMessages && !flags.includes('\\Draft') ? tools.getUserEncryptionKey(userData) : false; if (encryptionKey) { try { let encryptResult = await messageHandler.encryptMessageAsync(encryptionKey, raw); diff --git a/lib/handlers/on-fetch.js b/lib/handlers/on-fetch.js index 84c8754c7..b3ba4342a 100644 --- a/lib/handlers/on-fetch.js +++ b/lib/handlers/on-fetch.js @@ -45,7 +45,9 @@ module.exports = (server, messageHandler, userCache) => (mailbox, options, sessi return callback(err); } - messageHandler.counters.ttlcounter('idw:' + session.user.id, 0, limit, false, (err, res) => { + const counterKey = `idw:${tools.redisHashTag(db.redis, session.user.id.toString())}`; + + messageHandler.counters.ttlcounter(counterKey, 0, limit, false, (err, res) => { if (err) { return callback(err); } @@ -364,7 +366,7 @@ module.exports = (server, messageHandler, userCache) => (mailbox, options, sessi }); let limiter = new LimitedFetch({ - key: 'idw:' + session.user.id, + key: counterKey, ttlcounter: messageHandler.counters.ttlcounter, maxBytes: limit }); diff --git a/lib/imap-notifier.js b/lib/imap-notifier.js index d6ffe8358..c969edbe0 100644 --- a/lib/imap-notifier.js +++ b/lib/imap-notifier.js @@ -21,26 +21,32 @@ const USER_WORKERS_TTL_MS = USER_WORKERS_TTL * 1000; const USE_WORKER_WD_EVENTS_SETTING = 'const:imap:use_wd_worker_channels'; const USE_WORKER_WD_EVENTS_CACHE_TTL = 5000; // ms const WORKER_ID = `${config?.dbs?.workerId || os.hostname()}:${process.pid}`; +const USER_REGISTRY_STATE = { + counts: new Map(), + workerId: WORKER_ID, + workerChannel: `${WORKER_CHANNEL_PREFIX}${WORKER_ID}`, + useWorkerWdEvents: false, + useWorkerWdEventsUpdated: 0, + useWorkerWdEventsPending: null, + timer: null, + redis: null +}; class ImapNotifier extends EventEmitter { constructor(options) { super(); this.database = options.database; - this.redis = options.redis || new Redis(tools.redisConfig(config.dbs.redis)); + + const redisConfig = options.redis ? false : tools.redisConfig(config.dbs.redis); + this.redis = options.redis || (redisConfig.cluster ? new Redis.Cluster(redisConfig.nodes, redisConfig.options) : new Redis(redisConfig)); errors.registerRedisErrorLogger(this.redis, { role: 'imap-notifier' }); // if separate redis will add the appropriate role this.counters = counters(this.redis); this.settingsHandler = options.settingsHandler || new SettingsHandler({ db: this.database }); - this._userRegistryState = { - counts: new Map(), - workerId: WORKER_ID, - workerChannel: `${WORKER_CHANNEL_PREFIX}${WORKER_ID}`, - useWorkerWdEvents: false, - useWorkerWdEventsUpdated: 0, - useWorkerWdEventsPending: null, - timer: null, - redis: null - }; + this._userRegistryState = USER_REGISTRY_STATE; + this.subscribeFnName = tools.isRedisCluster(this.redis) ? 'ssubscribe' : 'subscribe'; + this.publishFnName = tools.isRedisCluster(this.redis) ? 'spublish' : 'publish'; + this.messageEventName = tools.isRedisCluster(this.redis) ? 'smessage' : 'message'; this.logger = options.logger || { info: log.silly.bind(log, 'IMAP'), @@ -56,7 +62,8 @@ class ImapNotifier extends EventEmitter { this.connectionSessions = new WeakMap(); // Subscriber needs its own client connection. This is relevant only in the context of IMAP - this.subscriber = new Redis(tools.redisConfig(config.dbs.redis)); + + this.subscriber = this.redis.duplicate([], tools.isRedisCluster(this.redis) ? { shardedSubscribers: true } : {}); // connection agnostic (standalone or Cluster) errors.registerRedisErrorLogger(this.subscriber, { role: 'imap-notifier-subscriber' }); this._listeners = new EventEmitter(); @@ -101,7 +108,7 @@ class ImapNotifier extends EventEmitter { } }; - this.subscriber.on('message', (channel, message) => { + this.subscriber.on(this.messageEventName, (channel, message) => { if (channel !== this._userRegistryState.workerChannel && channel !== LEGACY_CHANNEL) { return; } @@ -133,8 +140,8 @@ class ImapNotifier extends EventEmitter { } }); - this.subscriber.subscribe(this._userRegistryState.workerChannel); - this.subscriber.subscribe(LEGACY_CHANNEL); + this.subscriber[this.subscribeFnName](this._userRegistryState.workerChannel); + this.subscriber[this.subscribeFnName](LEGACY_CHANNEL); if (!this._userRegistryState.timer) { this._userRegistryState.redis = this.redis; @@ -335,7 +342,7 @@ class ImapNotifier extends EventEmitter { this._usesWorkerWdEvents() .then(useWorkerWdEvents => { if (!useWorkerWdEvents) { - return this.redis.publish(LEGACY_CHANNEL, data); + return this.redis[this.publishFnName](LEGACY_CHANNEL, data); } const key = this._getUserWorkersKey(userId); @@ -346,9 +353,13 @@ class ImapNotifier extends EventEmitter { return; } - const pipeline = this.redis.pipeline(); - workers.forEach(workerId => pipeline.publish(this._getWorkerChannel(workerId), data)); - return pipeline.exec(); + if (!tools.isRedisCluster(this.redis)) { + const pipeline = this.redis.pipeline(); + workers.forEach(workerId => pipeline[this.publishFnName](this._getWorkerChannel(workerId), data)); + return pipeline.exec(); + } + + return Promise.all(workers.map(workerId => this.redis[this.publishFnName](this._getWorkerChannel(workerId), data))); }); }) .catch(() => false); @@ -438,16 +449,43 @@ 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; + let commands = []; for (let userId of counts.keys()) { let key = this._getUserWorkersKey(userId); - pipeline.zremrangebyscore(key, 0, minScore); - pipeline.zadd(key, now, this._userRegistryState.workerId); - pipeline.expire(key, USER_WORKERS_TTL + 60); + + if (!tools.isRedisCluster(redis)) { + commands.push( + ['zremrangebyscore', key, 0, minScore], + ['zadd', key, now, this._userRegistryState.workerId], + ['expire', key, USER_WORKERS_TTL + 60] + ); + continue; + } + + commands.push( + redis + .multi() + .zremrangebyscore(key, 0, minScore) + .zadd(key, now, this._userRegistryState.workerId) + .expire(key, USER_WORKERS_TTL + 60) + .exec() + ); } - pipeline.exec().catch(() => false); + + if (!commands.length) { + return; + } + + if (!tools.isRedisCluster(redis)) { + return redis + .pipeline(commands) + .exec() + .catch(() => false); + } + + Promise.all(commands).catch(() => false); } /** @@ -479,7 +517,7 @@ class ImapNotifier extends EventEmitter { (Array.isArray(entries) ? entries : [].concat(entries || [])).forEach(entry => { let m = entry.mailbox.toString(); if (!counters.has(m)) { - counters.set(m, { total: 0, unseen: 0, unseenChange: false }); + counters.set(m, { user: entry.user && entry.user.toString(), total: 0, unseen: 0, unseenChange: false }); } switch (entry && entry.command) { @@ -514,15 +552,19 @@ class ImapNotifier extends EventEmitter { let mailbox = row[0]; let delta = row[1]; + if (!delta.user) { + continue; + } + let mailboxKey = `${tools.redisHashTag(this.redis, delta.user)}:${mailbox}`; - await this.redis.cachedcounter(`total:${mailbox}`, delta.total, consts.MAILBOX_COUNTER_TTL); + await this.redis.cachedcounter(`total:${mailboxKey}`, delta.total, consts.MAILBOX_COUNTER_TTL); if (delta.unseenChange) { // Message info changed in mailbox, so just te be sure, clear the unseen counter as well // Unseen counter is more volatile and also easier to count (usually only a small number on indexed messages) - await this.redis.del('unseen:' + mailbox); + await this.redis.del(`unseen:${mailboxKey}`); } else if (delta.unseen) { - await this.redis.cachedcounter(`unseen:${mailbox}`, delta.unseen, consts.MAILBOX_COUNTER_TTL); + await this.redis.cachedcounter(`unseen:${mailboxKey}`, delta.unseen, consts.MAILBOX_COUNTER_TTL); } } } @@ -532,7 +574,7 @@ class ImapNotifier extends EventEmitter { return callback(null, true); } - let rlkey = 'lim:' + data.service; + let rlkey = `lim:${data.service}:${tools.redisHashTag(this.redis, data.user.toString())}`; const { closed: socketClosed, connecting, destroyed } = data.session?.socket || {}; if (!socketClosed && !connecting && !destroyed) { @@ -567,7 +609,7 @@ class ImapNotifier extends EventEmitter { let entry = this.connectionSessions.get(data.session); this.connectionSessions.delete(data.session); - let rlkey = 'lim:' + entry.service; + let rlkey = `lim:${entry.service}:${tools.redisHashTag(this.redis, entry.user.toString())}`; this.counters.limitedcounter(rlkey, entry.user, -1, 0, err => { if (err) { this.logger.debug('[%s] Failed to release connection for user %s. %s', data.session.id, entry.user, err.message); diff --git a/lib/tasks/clear-folder.js b/lib/tasks/clear-folder.js index 7e62673a7..48e26f99a 100644 --- a/lib/tasks/clear-folder.js +++ b/lib/tasks/clear-folder.js @@ -2,6 +2,7 @@ const log = require('npmlog'); const db = require('../db'); +const tools = require('../tools'); let run = async (task, data, options) => { const messageHandler = options.messageHandler; @@ -48,8 +49,9 @@ let run = async (task, data, options) => { await cursor.close(); try { + const mailboxKey = `${tools.redisHashTag(db.redis, user.toString())}:${mailbox}`; // clear counters - await db.redis.multi().del(`total:${mailbox}`).del(`unseen:${mailbox}`).exec(); + await db.redis.multi().del(`total:${mailboxKey}`).del(`unseen:${mailboxKey}`).exec(); } catch (err) { // ignore } diff --git a/lib/tools.js b/lib/tools.js index 6fc43fa59..ef47718a5 100644 --- a/lib/tools.js +++ b/lib/tools.js @@ -134,7 +134,7 @@ function getWildcardAddresses(username, domain) { // returns a redis config object with a retry strategy function redisConfig(defaultConfig) { - return { + const redisDefaults = { // some defaults showFriendlyErrorStack: true, retryStrategy(times) { @@ -142,11 +142,58 @@ function redisConfig(defaultConfig) { log.info('Redis', 'Connection retry times=%s delay=%s', times, delay); return delay; }, - connectionName: `${packageData.name}@${packageData.version}[${process.pid}]`, - ...(typeof defaultConfig === 'string' ? redisUrl(defaultConfig) : defaultConfig || {}) // unwrap other config + connectionName: `${packageData.name}@${packageData.version}[${process.pid}]` + }; + + let parsedConfig = typeof defaultConfig === 'string' ? redisUrl(defaultConfig) : defaultConfig || {}; + + if (parsedConfig && parsedConfig.cluster) { + let clusterConfig = Object.assign({}, parsedConfig); + let nodeDefaults = {}; + + for (let key of ['password', 'username', 'tls']) { + if (typeof clusterConfig[key] !== 'undefined') { + nodeDefaults[key] = clusterConfig[key]; + } + } + + let nodes = [] + .concat(clusterConfig.nodes || []) + .map(node => Object.assign({}, nodeDefaults, node || {})) + .filter(node => node && node.host); + + delete clusterConfig.cluster; + delete clusterConfig.nodes; + delete clusterConfig.password; + delete clusterConfig.db; + delete clusterConfig.username; + delete clusterConfig.tls; + + let redisOptions = Object.assign({}, redisDefaults, nodeDefaults, clusterConfig.redisOptions || {}); + + return { + cluster: true, + connectionName: redisOptions.connectionName, + nodes, + options: Object.assign({}, clusterConfig, { redisOptions }) + }; + } + + return { + ...redisDefaults, + ...parsedConfig // unwrap other config }; } +function isRedisCluster(redis) { + return !!(redis && (redis.isCluster || redis.cluster)); +} + +function redisHashTag(redis, value) { + value = typeof value === 'undefined' || value === null ? '' : value.toString(); + return isRedisCluster(redis) ? `{${value}}` : value; +} + function decodeAddresses(addresses) { addresses.forEach(address => { address.name = (address.name || '').toString(); @@ -181,27 +228,49 @@ function flatAddresses(addresses) { return list; } -async function getMailboxCounter(db, mailbox, type) { - const cacheKey = `${type || 'total'}:${mailbox.toString()}`; +async function getMailboxCounter(db, mailbox, type, user) { + // fallback (should not happen) + if (!user) { + let mailboxObjectId = mailbox; + + let mailboxData = await db.database.collection('mailboxes').findOne( + { + _id: mailboxObjectId + }, + { + projection: { + user: true + } + } + ); + user = mailboxData && mailboxData.user; + } + + const mailboxKey = user ? `${redisHashTag(db.redis, user)}:${mailbox}` : false; + const cacheKey = mailboxKey ? `${type || 'total'}:${mailboxKey}` : false; try { // Check cache for pre-calculated counter value - let sum = await db.redis.get(cacheKey); - if (sum !== null && !isNaN(sum) && Number(sum) >= 0) { - return Number(sum); + if (cacheKey) { + const sum = await db.redis.get(cacheKey); + if (sum !== null && !isNaN(sum) && Number(sum) >= 0) { + return Number(sum); + } } // calculate sum - let query = { mailbox }; + const query = { mailbox }; if (type) { query[type] = true; } - sum = await db.database.collection('messages').countDocuments(query); + const sum = await db.database.collection('messages').countDocuments(query); // Cache calculated sum in redis - await db.redis.multi().set(cacheKey, sum).expire(cacheKey, consts.MAILBOX_COUNTER_TTL).exec(); + if (cacheKey) { + await db.redis.multi().set(cacheKey, sum).expire(cacheKey, consts.MAILBOX_COUNTER_TTL).exec(); + } return sum; } catch (err) { @@ -834,6 +903,8 @@ module.exports = { getHostname, getWildcardAddresses, redisConfig, + isRedisCluster, + redisHashTag, checkRangeQuery, decodeAddresses, flatAddresses, diff --git a/lib/user-handler.js b/lib/user-handler.js index b12bd58c1..4b3569ce2 100644 --- a/lib/user-handler.js +++ b/lib/user-handler.js @@ -1835,14 +1835,14 @@ class UserHandler { } if (data.featureFlags && Object.keys(data.featureFlags).length) { - let req = this.redis.multi(); + let req = []; for (let featureFlag of Object.keys(data.featureFlags)) { if (data.featureFlags[featureFlag]) { - req = req.sadd(`feature:${featureFlag}`, user.toString()); + req.push(this.redis.sadd(`feature:${featureFlag}`, user.toString())); } } try { - await req.exec(); + await Promise.all(req); } catch (err) { log.error('Redis', 'FEATUREFAIL failed to set feature flags id=%s error=%s', user, err.message); } @@ -3254,10 +3254,11 @@ class UserHandler { } if (resetKeys.has(key)) { - flushKeys.push(resetKeys.get(key) + ':' + user); + let prefix = resetKeys.get(key); + flushKeys.push(`${prefix}:${tools.redisHashTag(this.redis, user.toString())}`); } if (key === 'imapMaxConnections') { - flushHKeys.push({ key: 'lim:imap', value: user.toString() }); + flushHKeys.push({ key: `lim:imap:${tools.redisHashTag(this.redis, user.toString())}`, value: user.toString() }); } if (key === 'suspended' && data.suspended) { @@ -3494,18 +3495,15 @@ class UserHandler { // check if we need to reset any ttl counters if (flushKeys.length || flushHKeys.length) { - let flushreq = this.redis.multi(); - - flushKeys.forEach(key => { - flushreq = flushreq.del(key); - }); - - flushHKeys.forEach(entry => { - flushreq = flushreq.hdel(entry.key, entry.value); - }); - // just call the operations and hope for the best, no problems if fails try { + let flushreq = this.redis.multi(); + flushKeys.forEach(key => { + flushreq = flushreq.del(key); + }); + flushHKeys.forEach(entry => { + flushreq = flushreq.hdel(entry.key, entry.value); + }); await flushreq.exec(); } catch (err) { // ignore @@ -3716,14 +3714,14 @@ class UserHandler { // remove feature flag entries if (existingAccount.featureFlags && Object.keys(existingAccount.featureFlags).length) { - let req = this.redis.multi(); + let req = []; for (let featureFlag of Object.keys(existingAccount.featureFlags)) { if (existingAccount.featureFlags[featureFlag]) { - req = req.srem(`feature:${featureFlag}`, user.toString()); + req.push(this.redis.srem(`feature:${featureFlag}`, user.toString())); } } try { - await req.exec(); + await Promise.all(req); } catch (err) { log.error('Redis', 'FEATUREFAIL failed to update feature flags id=%s error=%s', user, err.message); } diff --git a/migrations/database/addressregister_add_disabled.js b/migrations/database/addressregister_add_disabled.js index fd3e8dca9..a0ac96d4d 100644 --- a/migrations/database/addressregister_add_disabled.js +++ b/migrations/database/addressregister_add_disabled.js @@ -3,7 +3,7 @@ // MongoDB Migration Script: addressregister add disabled field to all current addressregister entries in DB const config = require('@zone-eu/wild-config'); -const ENABLED = process.env.NODE_ENV === 'test' ? false : !!config?.migrations?.database?.addressregisterAddDisabled?.enabled; +const ENABLED = /^test/.test(process.env.NODE_ENV || '') ? false : !!config?.migrations?.database?.addressregisterAddDisabled?.enabled; const BATCH_SIZE = 1000; async function addDisabledToAddressregister() { diff --git a/migrations/users/addresses_to_domaincache.js b/migrations/users/addresses_to_domaincache.js index 700891391..d4803d988 100644 --- a/migrations/users/addresses_to_domaincache.js +++ b/migrations/users/addresses_to_domaincache.js @@ -3,7 +3,7 @@ // MongoDB Migration Script: addresses to domaincache const config = require('@zone-eu/wild-config'); -const ENABLED = process.env.NODE_ENV === 'test' ? false : !!config?.migrations?.users?.addressesToDomaincache?.enabled; +const ENABLED = /^test/.test(process.env.NODE_ENV || '') ? false : !!config?.migrations?.users?.addressesToDomaincache?.enabled; // Extract domain from email address function extractDomain(email) { diff --git a/migrations/users/domainaliases_to_domaincache.js b/migrations/users/domainaliases_to_domaincache.js index 3ee0a9994..be5753fbc 100644 --- a/migrations/users/domainaliases_to_domaincache.js +++ b/migrations/users/domainaliases_to_domaincache.js @@ -2,7 +2,7 @@ /* global db, log */ const config = require('@zone-eu/wild-config'); -const ENABLED = process.env.NODE_ENV === 'test' ? false : !!config?.migrations?.users?.domainaliasesToDomaincache?.enabled; +const ENABLED = /^test/.test(process.env.NODE_ENV || '') ? false : !!config?.migrations?.users?.domainaliasesToDomaincache?.enabled; async function migrateDomainAliasesToCache() { const sourceCollection = db.collection('domainaliases'); diff --git a/package.json b/package.json index 14ba1d589..0fe2b11bb 100644 --- a/package.json +++ b/package.json @@ -5,6 +5,7 @@ "main": "server.js", "scripts": { "test": "mongosh --eval 'db.dropDatabase()' wildduck-test && redis-cli -n 13 flushdb && npm run runtest", + "test:cluster": "mongosh --eval 'db.dropDatabase()' wildduck-test && redis-cli --cluster call 127.0.0.1:7000 flushdb && NODE_ENV=test-cluster grunt", "test:proto": "NODE_ENV=test grunt proto", "printconf": "NODE_CONFIG_ONLY=true npm start", "runtest": "NODE_ENV=test grunt", diff --git a/pop3.js b/pop3.js index a2cbad755..5b51dbb5a 100644 --- a/pop3.js +++ b/pop3.js @@ -132,20 +132,17 @@ const serverOptions = { } session.user.mailbox = mailbox._id; + let messageIndexKey = `pxm:${tools.redisHashTag(db.redis, session.user.id.toString())}`; - db.redis - .multi() - // "new" limit store - .hget(`pxm:${session.user.id}`, mailbox._id.toString()) - // fallback store - .hget(`pop3uid`, mailbox._id.toString()) - .exec((err, res) => { - let lastIndex = res && ((res[0] && res[0][1]) || (res[1] && res[1][1])); + db.redis.hget(messageIndexKey, mailbox._id.toString(), (err, messageIndex) => { + let lastIndex = !err && messageIndex; - let query = { - mailbox: mailbox._id - }; - if (!err && lastIndex && !isNaN(lastIndex)) { + let query = { + mailbox: mailbox._id + }; + + let processMessages = () => { + if (lastIndex && !isNaN(lastIndex)) { query.uid = { $gte: Number(lastIndex) }; } @@ -178,14 +175,11 @@ const serverOptions = { if (!oldestMessageData || !oldestMessageData.uid) { return done(); } - // try to update index, ignore result - db.redis - .multi() - // update limit store - .hset(`pxm:${session.user.id}`, mailbox._id.toString(), oldestMessageData.uid) + + db.redis.hset(messageIndexKey, mailbox._id.toString(), oldestMessageData.uid, () => { // delete fallback store as it is no longer needed - .hdel(`pop3uid`, mailbox._id.toString()) - .exec(done); + db.redis.hdel(`pop3uid`, mailbox._id.toString(), done); + }); }; updateUIDIndex(() => @@ -208,7 +202,20 @@ const serverOptions = { ); }); }); + }; + + if (lastIndex) { + return processMessages(); + } + + // fallback store + db.redis.hget(`pop3uid`, mailbox._id.toString(), (fallbackErr, fallbackIndex) => { + if (!fallbackErr && fallbackIndex) { + lastIndex = fallbackIndex; + } + processMessages(); }); + }); } ); }, @@ -219,7 +226,9 @@ const serverOptions = { return callback(err); } - messageHandler.counters.ttlcounter('pdw:' + session.user.id, 0, limit, false, (err, res) => { + let counterKey = `pdw:${tools.redisHashTag(db.redis, session.user.id.toString())}`; + + messageHandler.counters.ttlcounter(counterKey, 0, limit, false, (err, res) => { if (err) { return callback(err); } @@ -257,7 +266,7 @@ const serverOptions = { } let limiter = new LimitedFetch({ - key: 'pdw:' + session.user.id, + key: counterKey, ttlcounter: messageHandler.counters.ttlcounter, maxBytes: limit, skipCounter: true