From 8ffd85bcb15edbfceeddb83a5fde197b9af97cc4 Mon Sep 17 00:00:00 2001 From: Mikael Finstad Date: Sat, 28 Sep 2024 11:57:02 +0200 Subject: [PATCH 1/2] fix redis emitter it had multiple issues: - memory leak in removeAllListeners - eventName === 'error' wasn't consistently handled --- .../src/server/emitter/redis-emitter.js | 129 ++++++++++-------- 1 file changed, 75 insertions(+), 54 deletions(-) diff --git a/packages/@uppy/companion/src/server/emitter/redis-emitter.js b/packages/@uppy/companion/src/server/emitter/redis-emitter.js index aa62f61baf..baf750944a 100644 --- a/packages/@uppy/companion/src/server/emitter/redis-emitter.js +++ b/packages/@uppy/companion/src/server/emitter/redis-emitter.js @@ -20,54 +20,68 @@ function replacer(key, value) { module.exports = (redisClient, redisPubSubScope) => { const prefix = redisPubSubScope ? `${redisPubSubScope}:` : '' const getPrefixedEventName = (eventName) => `${prefix}${eventName}` - 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.toString())) - return subscriber.connect() - }) - - const handlersByEvent = new Map() const errorEmitter = new EventEmitter() const handleError = (err) => errorEmitter.emit('error', err) - connectedPromise.catch((err) => handleError(err)) + async function makeRedis() { + const publisher = redisClient.duplicate({ lazyConnect: true }) + publisher.on('error', err => logger.error('publisher redis error', err.toString())) + const subscriber = publisher.duplicate() + subscriber.on('error', err => logger.error('subscriber redis error', err.toString())) + await publisher.connect() + await subscriber.connect() + return { subscriber, publisher } + } + + const redisPromise = makeRedis() + redisPromise.catch((err) => handleError(err)) + /** + * + * @param {(a: Awaited) => void} fn + */ async function runWhenConnected (fn) { try { - await connectedPromise - await fn() + await fn(await redisPromise) } catch (err) { handleError(err) } } + /** @type {Map unknown, () => unknown>>} */ + const handlersByEvent = new Map() + /** * Remove an event listener * * @param {string} eventName name of the event * @param {any} handler the handler of the event to remove */ - function removeListener (eventName, handler) { - if (eventName === 'error') return errorEmitter.removeListener('error', handler) + async function removeListener (eventName, handler) { + if (eventName === 'error') { + errorEmitter.removeListener('error', handler) + return + } - return runWhenConnected(() => { - const handlersByThisEventName = handlersByEvent.get(eventName) - if (handlersByThisEventName == null) return undefined + const thisEventNameActualHandlerByHandler = handlersByEvent.get(eventName) + if (thisEventNameActualHandlerByHandler == null) return - const actualHandler = handlersByThisEventName.get(handler) - if (actualHandler == null) return undefined + const actualHandler = thisEventNameActualHandlerByHandler.get(handler) + if (actualHandler == null) return - handlersByThisEventName.delete(handler) - if (handlersByThisEventName.size === 0) handlersByEvent.delete(eventName) + thisEventNameActualHandlerByHandler.delete(handler) + const didRemoveLastListener = thisEventNameActualHandlerByHandler.size === 0 + if (didRemoveLastListener) { + handlersByEvent.delete(eventName) + } + + await runWhenConnected(async ({ subscriber }) => { subscriber.off('pmessage', actualHandler) - return subscriber.punsubscribe(getPrefixedEventName(eventName)) + if (didRemoveLastListener) { + await subscriber.punsubscribe(getPrefixedEventName(eventName)) + } }) } @@ -77,7 +91,13 @@ module.exports = (redisClient, redisPubSubScope) => { * @param {*} handler * @param {*} _once */ - function addListener (eventName, handler, _once = false) { + async function addListener (eventName, handler, _once = false) { + if (eventName === 'error') { + if (_once) errorEmitter.once('error', handler) + else errorEmitter.addListener('error', handler) + return + } + function actualHandler (pattern, channel, message) { if (pattern !== getPrefixedEventName(eventName)) { return @@ -91,19 +111,20 @@ module.exports = (redisClient, redisPubSubScope) => { handleError(new Error(`Invalid JSON received! Channel: ${eventName} Message: ${message}`)) return } + handler(...args) } - let handlersByThisEventName = handlersByEvent.get(eventName) - if (handlersByThisEventName == null) { - handlersByThisEventName = new WeakMap() - handlersByEvent.set(eventName, handlersByThisEventName) + let thisEventNameActualHandlerByHandler = handlersByEvent.get(eventName) + if (thisEventNameActualHandlerByHandler == null) { + thisEventNameActualHandlerByHandler = new Map() + handlersByEvent.set(eventName, thisEventNameActualHandlerByHandler) } - handlersByThisEventName.set(handler, actualHandler) + thisEventNameActualHandlerByHandler.set(handler, actualHandler) - runWhenConnected(() => { + await runWhenConnected(async ({ subscriber }) => { subscriber.on('pmessage', actualHandler) - return subscriber.psubscribe(getPrefixedEventName(eventName)) + await subscriber.psubscribe(getPrefixedEventName(eventName)) }) } @@ -113,10 +134,8 @@ module.exports = (redisClient, redisPubSubScope) => { * @param {string} eventName name of the event * @param {any} handler the handler of the event */ - function on (eventName, handler) { - if (eventName === 'error') return errorEmitter.on('error', handler) - - return addListener(eventName, handler) + async function on (eventName, handler) { + await addListener(eventName, handler) } /** @@ -125,8 +144,8 @@ module.exports = (redisClient, redisPubSubScope) => { * @param {string} eventName name of the event * @param {any} handler the handler of the event */ - function off (eventName, handler) { - return removeListener(eventName, handler) + async function off (eventName, handler) { + await removeListener(eventName, handler) } /** @@ -135,10 +154,8 @@ module.exports = (redisClient, redisPubSubScope) => { * @param {string} eventName name of the event * @param {any} handler the handler of the event */ - function once (eventName, handler) { - if (eventName === 'error') return errorEmitter.once('error', handler) - - return addListener(eventName, handler, true) + async function once (eventName, handler) { + await addListener(eventName, handler, true) } /** @@ -146,11 +163,10 @@ module.exports = (redisClient, redisPubSubScope) => { * * @param {string} eventName name of the event */ - function emit (eventName, ...args) { - runWhenConnected( - () => publisher.publish(getPrefixedEventName(eventName), - safeStringify(args, replacer)), - ) + async function emit (eventName, ...args) { + await runWhenConnected(async ({ publisher }) => ( + publisher.publish(getPrefixedEventName(eventName), safeStringify(args, replacer)) + )) } /** @@ -158,13 +174,18 @@ module.exports = (redisClient, redisPubSubScope) => { * * @param {string} eventName name of the event */ - function removeAllListeners (eventName) { - if (eventName === 'error') return errorEmitter.removeAllListeners(eventName) + async function removeAllListeners (eventName) { + if (eventName === 'error') { + errorEmitter.removeAllListeners(eventName) + return + } - return runWhenConnected(() => { - handlersByEvent.delete(eventName) - return subscriber.punsubscribe(getPrefixedEventName(eventName)) - }) + const thisEventNameActualHandlerByHandler = handlersByEvent.get(eventName) + if (thisEventNameActualHandlerByHandler != null) { + for (const handler of thisEventNameActualHandlerByHandler.keys()) { + await removeListener(eventName, handler) + } + } } return { From e1ae5bafbf2a0794119bce8e53965bfec3a9f29b Mon Sep 17 00:00:00 2001 From: Mikael Finstad Date: Sat, 5 Oct 2024 15:35:51 +0200 Subject: [PATCH 2/2] rename and add comment --- .../src/server/emitter/redis-emitter.js | 31 ++++++++++--------- 1 file changed, 16 insertions(+), 15 deletions(-) diff --git a/packages/@uppy/companion/src/server/emitter/redis-emitter.js b/packages/@uppy/companion/src/server/emitter/redis-emitter.js index baf750944a..0aa145432b 100644 --- a/packages/@uppy/companion/src/server/emitter/redis-emitter.js +++ b/packages/@uppy/companion/src/server/emitter/redis-emitter.js @@ -49,8 +49,9 @@ module.exports = (redisClient, redisPubSubScope) => { } } + // because each event can have multiple listeners, we need to keep track of them /** @type {Map unknown, () => unknown>>} */ - const handlersByEvent = new Map() + const handlersByEventName = new Map() /** * Remove an event listener @@ -64,17 +65,17 @@ module.exports = (redisClient, redisPubSubScope) => { return } - const thisEventNameActualHandlerByHandler = handlersByEvent.get(eventName) - if (thisEventNameActualHandlerByHandler == null) return + const actualHandlerByHandler = handlersByEventName.get(eventName) + if (actualHandlerByHandler == null) return - const actualHandler = thisEventNameActualHandlerByHandler.get(handler) + const actualHandler = actualHandlerByHandler.get(handler) if (actualHandler == null) return - thisEventNameActualHandlerByHandler.delete(handler) + actualHandlerByHandler.delete(handler) - const didRemoveLastListener = thisEventNameActualHandlerByHandler.size === 0 + const didRemoveLastListener = actualHandlerByHandler.size === 0 if (didRemoveLastListener) { - handlersByEvent.delete(eventName) + handlersByEventName.delete(eventName) } await runWhenConnected(async ({ subscriber }) => { @@ -115,12 +116,12 @@ module.exports = (redisClient, redisPubSubScope) => { handler(...args) } - let thisEventNameActualHandlerByHandler = handlersByEvent.get(eventName) - if (thisEventNameActualHandlerByHandler == null) { - thisEventNameActualHandlerByHandler = new Map() - handlersByEvent.set(eventName, thisEventNameActualHandlerByHandler) + let actualHandlerByHandler = handlersByEventName.get(eventName) + if (actualHandlerByHandler == null) { + actualHandlerByHandler = new Map() + handlersByEventName.set(eventName, actualHandlerByHandler) } - thisEventNameActualHandlerByHandler.set(handler, actualHandler) + actualHandlerByHandler.set(handler, actualHandler) await runWhenConnected(async ({ subscriber }) => { subscriber.on('pmessage', actualHandler) @@ -180,9 +181,9 @@ module.exports = (redisClient, redisPubSubScope) => { return } - const thisEventNameActualHandlerByHandler = handlersByEvent.get(eventName) - if (thisEventNameActualHandlerByHandler != null) { - for (const handler of thisEventNameActualHandlerByHandler.keys()) { + const actualHandlerByHandler = handlersByEventName.get(eventName) + if (actualHandlerByHandler != null) { + for (const handler of actualHandlerByHandler.keys()) { await removeListener(eventName, handler) } }