diff --git a/.env.example b/.env.example index 961ecd91..f2407339 100644 --- a/.env.example +++ b/.env.example @@ -34,6 +34,8 @@ STORAGE_STRATEGY=lru # Format: redis://[username:password@]host[:port][/database_number] # Example: redis://user:password@redis.example.com:6379/0 REDIS_URL=redis://localhost:6379 +# For Redis Cluster, list the seed nodes separated by commas: +# REDIS_URL=redis://node1:6379,redis://node2:6379,redis://node3:6379 # Timeout in milliseconds for the whiteboard server's graceful shutdown process # If the server cannot close all connections within this time, it will force shutdown diff --git a/README.md b/README.md index 5cdea4b8..a345e01f 100644 --- a/README.md +++ b/README.md @@ -206,6 +206,17 @@ STORAGE_STRATEGY=redis REDIS_URL=redis://[username:password@]host[:port][/database_number] ``` +**Redis Cluster** +- List the seed nodes separated by commas; the remaining nodes are discovered automatically +- Any credentials given on the first seed are applied to every discovered node, and so is TLS if the first seed uses `rediss://` +- In a list of seeds, write a comma inside a password as `%2C`, since the list is split on commas; a single URL with a comma in its password is still read as one node +- Note that Redis Cluster only has database 0, so a database number cannot be used to separate whiteboard keys from other tenants of the same cluster + +```bash +STORAGE_STRATEGY=redis +REDIS_URL=redis://node1:6379,redis://node2:6379,redis://node3:6379 +``` + ### Scaling (Optional) For high-traffic environments with multiple websocket servers: diff --git a/eslint-suppressions.json b/eslint-suppressions.json index 9b0bd5ca..66842b1e 100644 --- a/eslint-suppressions.json +++ b/eslint-suppressions.json @@ -1245,9 +1245,6 @@ } }, "websocket_server/Services/ServerService.js": { - "@stylistic/arrow-parens": { - "count": 1 - }, "@stylistic/padded-blocks": { "count": 3 }, diff --git a/tests/integration/redisClientErrors.spec.mjs b/tests/integration/redisClientErrors.spec.mjs new file mode 100644 index 00000000..5ab87323 --- /dev/null +++ b/tests/integration/redisClientErrors.spec.mjs @@ -0,0 +1,68 @@ +/** + * SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors + * SPDX-License-Identifier: AGPL-3.0-or-later + */ + +import { EventEmitter } from 'node:events' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' + +const redis = vi.hoisted(() => ({ + createClient: vi.fn(), + createCluster: vi.fn(), +})) +vi.mock('redis', () => redis) + +const { default: RedisAdapter } = await import('../../websocket_server/Adapters/RedisAdapter.js') +const { default: Config } = await import('../../websocket_server/Utilities/ConfigUtility.js') + +describe('errors emitted by the Redis client', () => { + const configuredUrl = Config.REDIS_URL + let errorLog + + beforeEach(() => { + vi.spyOn(console, 'log').mockImplementation(() => {}) + errorLog = vi.spyOn(console, 'error').mockImplementation(() => {}) + redis.createClient.mockImplementation(() => new EventEmitter()) + redis.createCluster.mockImplementation(() => new EventEmitter()) + }) + + afterEach(() => { + Config.REDIS_URL = configuredUrl + vi.useRealTimers() + vi.restoreAllMocks() + }) + + it.each([ + ['a single node', 'redis://localhost:6379'], + ['a unix socket', 'unix:///var/run/redis.sock'], + ['a cluster', 'redis://node1:6379,redis://node2:6379'], + ])('does not end the process when %s loses its connection', (_, url) => { + Config.REDIS_URL = url + + const client = RedisAdapter.createRedisClient() + + expect(client.listenerCount('error')).toBe(1) + expect(() => client.emit('error', new Error('read ECONNRESET'))).not.toThrow() + expect(errorLog).toHaveBeenCalledWith('Redis client error:', 'read ECONNRESET') + }) + + it('logs a repeated message once a minute rather than on every reconnect attempt', () => { + vi.useFakeTimers() + Config.REDIS_URL = 'redis://node1:6379,redis://node2:6379' + const client = RedisAdapter.createRedisClient() + const refused = new Error('connect ECONNREFUSED 127.0.0.1:7105') + + for (let i = 0; i < 10; i++) { + client.emit('error', refused) + vi.advanceTimersByTime(500) + } + expect(errorLog).toHaveBeenCalledTimes(1) + + client.emit('error', new Error('read ECONNRESET')) + expect(errorLog).toHaveBeenCalledTimes(2) + + vi.advanceTimersByTime(60 * 1000) + client.emit('error', refused) + expect(errorLog).toHaveBeenCalledTimes(3) + }) +}) diff --git a/tests/integration/redisCluster.spec.mjs b/tests/integration/redisCluster.spec.mjs new file mode 100644 index 00000000..f00b53fa --- /dev/null +++ b/tests/integration/redisCluster.spec.mjs @@ -0,0 +1,226 @@ +/** + * SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors + * SPDX-License-Identifier: AGPL-3.0-or-later + */ + +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' + +const redis = vi.hoisted(() => ({ + createClient: vi.fn(() => ({ kind: 'single', on: vi.fn() })), + createCluster: vi.fn(() => ({ kind: 'cluster', on: vi.fn() })), +})) +vi.mock('redis', () => redis) + +const { default: RedisAdapter } = await import('../../websocket_server/Adapters/RedisAdapter.js') +const { default: Config } = await import('../../websocket_server/Utilities/ConfigUtility.js') +const { deleteKeys, isClusterClient, scanKeys } = await import('../../websocket_server/Utilities/RedisUtility.js') + +const SECRET = 'p@ss:w0rd' + +describe('choosing between a single node and a cluster', () => { + const configuredUrl = Config.REDIS_URL + + beforeEach(() => { + vi.spyOn(console, 'log').mockImplementation(() => {}) + }) + + afterEach(() => { + Config.REDIS_URL = configuredUrl + redis.createClient.mockClear() + redis.createCluster.mockClear() + vi.restoreAllMocks() + }) + + it('builds a single client for one URL', () => { + Config.REDIS_URL = 'redis://localhost:6379' + + RedisAdapter.createRedisClient() + + expect(redis.createCluster).not.toHaveBeenCalled() + expect(redis.createClient).toHaveBeenCalledWith({ url: 'redis://localhost:6379', RESP: 2, commandOptions: { timeout: undefined } }) + }) + + it('builds a single client for a unix socket', () => { + Config.REDIS_URL = 'unix:///var/run/redis.sock?db=3' + + RedisAdapter.createRedisClient() + + expect(redis.createCluster).not.toHaveBeenCalled() + expect(redis.createClient).toHaveBeenCalledWith({ + socket: { path: '/var/run/redis.sock' }, + database: 3, + RESP: 2, + commandOptions: { timeout: undefined }, + }) + }) + + it('keeps a single client when the password holds a comma', () => { + Config.REDIS_URL = 'redis://:one,two@localhost:6379' + + RedisAdapter.createRedisClient() + + expect(redis.createCluster).not.toHaveBeenCalled() + expect(redis.createClient).toHaveBeenCalledWith({ url: 'redis://:one,two@localhost:6379', RESP: 2, commandOptions: { timeout: undefined } }) + }) + + it('builds a cluster client from a list of seed nodes', () => { + Config.REDIS_URL = 'redis://node1:6379,redis://node2:6379,redis://node3:6379' + + RedisAdapter.createRedisClient() + + expect(redis.createClient).not.toHaveBeenCalled() + expect(redis.createCluster).toHaveBeenCalledWith({ + rootNodes: [ + { url: 'redis://node1:6379' }, + { url: 'redis://node2:6379' }, + { url: 'redis://node3:6379' }, + ], + RESP: 2, + commandOptions: { timeout: undefined }, + }) + }) + + it('repeats the credentials of the first seed in the defaults', () => { + Config.REDIS_URL = `redis://user:${encodeURIComponent(SECRET)}@node1:6379,redis://node2:6379` + + RedisAdapter.createRedisClient() + + const [options] = redis.createCluster.mock.calls[0] + expect(options.defaults).toEqual({ username: 'user', password: SECRET }) + }) + + it('reads a comma written as %2C in a seed list password', () => { + Config.REDIS_URL = 'redis://user:one%2Ctwo@node1:6379,redis://node2:6379' + + RedisAdapter.createRedisClient() + + const [options] = redis.createCluster.mock.calls[0] + expect(options.rootNodes).toHaveLength(2) + expect(options.defaults).toEqual({ username: 'user', password: 'one,two' }) + }) + + it('repeats TLS in the defaults, which discovered nodes connect with', () => { + Config.REDIS_URL = 'rediss://node1:6380,rediss://node2:6380' + + RedisAdapter.createRedisClient() + + const [options] = redis.createCluster.mock.calls[0] + expect(options.defaults).toEqual({ socket: { tls: true } }) + }) + + it('repeats both credentials and TLS when the first seed carries both', () => { + Config.REDIS_URL = `rediss://user:${SECRET}@node1:6380,rediss://node2:6380` + + RedisAdapter.createRedisClient() + + const [options] = redis.createCluster.mock.calls[0] + expect(options.defaults).toEqual({ + username: 'user', + password: SECRET, + socket: { tls: true }, + }) + }) + + it('leaves the defaults out when the seeds need nothing', () => { + Config.REDIS_URL = 'redis://node1:6379,redis://node2:6379' + + RedisAdapter.createRedisClient() + + const [options] = redis.createCluster.mock.calls[0] + expect(options).not.toHaveProperty('defaults') + }) +}) + +/** + * A stand-in for one node of a cluster. + * + * @param {string[][]} replies the keys this node returns, one array per SCAN reply + * @return {object} an object with the scanIterator() of a single client + */ +function fakeNode(replies) { + return { + async* scanIterator() { + yield* replies + }, + } +} + +describe('scanning the keyspace', () => { + it('recognises a cluster client by its nodeClient()', () => { + expect(isClusterClient(fakeNode([]))).toBe(false) + expect(isClusterClient({ nodeClient: () => {}, masters: [] })).toBe(true) + expect(isClusterClient(null)).toBe(false) + }) + + it('scans a single client directly', async () => { + const client = fakeNode([['a', 'b'], ['c']]) + const seen = [] + for await (const batch of scanKeys(client, { MATCH: 'x*' })) { + seen.push(batch) + } + expect(seen).toEqual([['a', 'b'], ['c']]) + }) + + it('scans every primary of a cluster through its own client', async () => { + const nodes = [fakeNode([['a']]), fakeNode([['b'], ['c']]), fakeNode([])] + const masters = [{ id: 0 }, { id: 1 }, { id: 2 }] + const cluster = { + masters, + nodeClient: vi.fn((master) => nodes[master.id]), + } + + const seen = [] + for await (const batch of scanKeys(cluster, { MATCH: 'x*' })) { + seen.push(batch) + } + + expect(seen).toEqual([['a'], ['b'], ['c']]) + expect(cluster.nodeClient).toHaveBeenCalledTimes(3) + }) + + it('waits for a node client that is still connecting', async () => { + const cluster = { + masters: [{ id: 0 }], + nodeClient: () => Promise.resolve(fakeNode([['a']])), + } + + const seen = [] + for await (const batch of scanKeys(cluster, {})) { + seen.push(batch) + } + + expect(seen).toEqual([['a']]) + }) +}) + +describe('deleting keys', () => { + it('deletes a whole batch in one call on a single node', async () => { + const client = { del: vi.fn(async () => 2) } + + await deleteKeys(client, ['a', 'b']) + + expect(client.del).toHaveBeenCalledTimes(1) + expect(client.del).toHaveBeenCalledWith(['a', 'b']) + }) + + it('deletes one key at a time on a cluster, which refuses a batch across slots', async () => { + const cluster = { + masters: [], + nodeClient: () => {}, + del: vi.fn(async () => 1), + } + + await deleteKeys(cluster, ['a', 'b', 'c']) + + expect(cluster.del).toHaveBeenCalledTimes(3) + expect(cluster.del.mock.calls.map(([key]) => key)).toEqual(['a', 'b', 'c']) + }) + + it('does nothing when there is nothing to delete', async () => { + const client = { del: vi.fn() } + + await deleteKeys(client, []) + + expect(client.del).not.toHaveBeenCalled() + }) +}) diff --git a/tests/integration/redisStartupOrder.spec.mjs b/tests/integration/redisStartupOrder.spec.mjs new file mode 100644 index 00000000..1bc2f242 --- /dev/null +++ b/tests/integration/redisStartupOrder.spec.mjs @@ -0,0 +1,109 @@ +/** + * SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors + * SPDX-License-Identifier: AGPL-3.0-or-later + */ + +import http from 'node:http' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import SocketService from '../../websocket_server/Services/SocketService.js' +import StorageService from '../../websocket_server/Services/StorageService.js' + +/** + * A Redis client that records the commands it is asked to run. + * + * @param {string[]} commands the list every call is appended to + * @return {object} a stand-in for a redis client + */ +function recordingClient(commands) { + const record = (name) => () => { + commands.push(name) + return Promise.resolve(null) + } + + return { + isOpen: true, + set: record('set'), + get: record('get'), + del: record('del'), + expire: record('expire'), + hGetAll: record('hGetAll'), + hSet: record('hSet'), + async* scanIterator() { + commands.push('scanIterator') + yield* [] + }, + } +} + +/** + * Build a socket service whose Redis connection is still being established. + * + * @param {Promise} redisReady the promise the connection is behind + * @param {string[]} commands the list Redis calls are recorded into + * @return {object} the server and the service, both to be closed by the caller + */ +function buildService(redisReady, commands) { + const server = http.createServer() + const service = new SocketService( + server, + StorageService.create('in-mem'), + StorageService.create('in-mem'), + recordingClient(commands), + redisReady, + ) + + return { server, service } +} + +describe('starting while Redis is still connecting', () => { + let started = null + + beforeEach(() => { + vi.spyOn(console, 'log').mockImplementation(() => {}) + vi.spyOn(console, 'error').mockImplementation(() => {}) + // the streams adapter reads from Redis in a loop of its own, which is + // not what is under test here + vi.spyOn(SocketService.prototype, 'setupAdapter').mockResolvedValue(undefined) + }) + + afterEach(async () => { + if (started) { + await started.service.clusterService.stop() + started.service.io.close() + started.server.close() + started = null + } + vi.restoreAllMocks() + }) + + it('runs no command until the connection is up', async () => { + const commands = [] + let connected + const redisReady = new Promise((resolve) => { + connected = resolve + }) + + started = buildService(redisReady, commands) + + // give init() every chance to run ahead of the connection + await new Promise((resolve) => setImmediate(resolve)) + await new Promise((resolve) => setImmediate(resolve)) + expect(commands).toEqual([]) + + connected() + await started.service.ready + + // the heartbeat is written once as soon as the cluster service starts + expect(commands).toContain('set') + }) + + it('fails to start when the connection fails', async () => { + const commands = [] + const redisReady = Promise.reject(new Error('Redis is not listening')) + + started = buildService(redisReady, commands) + + await expect(started.service.ready).rejects.toThrow('Redis is not listening') + expect(commands).toEqual([]) + }) +}) diff --git a/websocket_server/Adapters/RedisAdapter.js b/websocket_server/Adapters/RedisAdapter.js index adbe38f0..c6be0b71 100644 --- a/websocket_server/Adapters/RedisAdapter.js +++ b/websocket_server/Adapters/RedisAdapter.js @@ -6,8 +6,9 @@ */ import StorageAdapter from './StorageAdapter.js' -import { createClient } from 'redis' +import { createClient, createCluster } from 'redis' import Config from '../Utilities/ConfigUtility.js' +import { deleteKeys, scanKeys } from '../Utilities/RedisUtility.js' // node-redis 6 changed two defaults that this server relies on, so keep the // ones redis 4 had. RESP3 opens every connection with HELLO, which Redis only @@ -20,6 +21,8 @@ const CLIENT_OPTIONS = { commandOptions: { timeout: undefined }, } +const ERROR_LOG_INTERVAL_MS = 60 * 1000 + export default class RedisAdapter extends StorageAdapter { static isClientClosedError(error) { @@ -112,23 +115,91 @@ export default class RedisAdapter extends StorageAdapter { } } + /** + * Log the errors a client emits instead of letting them end the process. + * + * A Redis client emits `error` when it loses Redis: a single node client + * whenever its connection drops, a cluster client when it can reach none of + * the nodes. An `error` event without a listener throws, so either took the + * whole server down. The client reconnects on its own, and the same message + * comes back on every attempt, so each message is logged at most once a + * minute. + * + * @param {object} client a redis client, one node or a cluster + * @return {object} the same client + */ + static logClientErrors(client) { + const lastLogged = new Map() + client.on('error', (error) => { + const message = error?.message ?? String(error) + const now = Date.now() + if (lastLogged.has(message) && now - lastLogged.get(message) < ERROR_LOG_INTERVAL_MS) { + return + } + lastLogged.set(message, now) + console.error('Redis client error:', message) + }) + return client + } + static createRedisClient() { console.log(`Creating Redis client with URL: ${RedisAdapter.redactRedisUrl(Config.REDIS_URL)}`) + const nodeUrls = RedisAdapter.splitRedisUrls(Config.REDIS_URL) + + if (nodeUrls.length > 1) { + return RedisAdapter.logClientErrors(RedisAdapter.createRedisClusterClient(nodeUrls)) + } + const redisUrl = RedisAdapter.parseRedisUrl(Config.REDIS_URL) if (redisUrl.protocol === 'unix:') { const db = redisUrl.searchParams.get('db') - return createClient({ + return RedisAdapter.logClientErrors(createClient({ socket: { path: redisUrl.pathname }, database: db !== null ? Number(db) : undefined, ...CLIENT_OPTIONS, - }) + })) } else { - return createClient({ url: Config.REDIS_URL, ...CLIENT_OPTIONS }) + return RedisAdapter.logClientErrors(createClient({ url: Config.REDIS_URL, ...CLIENT_OPTIONS })) } } + /** + * Build a cluster client from a list of seed node URLs. + * + * Redis Cluster discovers the remaining nodes itself and reports them + * without credentials, and the scheme of a seed only governs the connection + * to that seed. Everything the discovered nodes need therefore has to be + * repeated in the defaults: the credentials given on the first seed, and + * TLS when the first seed asks for it, or those connections fall back to + * plain TCP and a TLS only cluster cannot be reached. + * + * @param {string[]} nodeUrls the seed node URLs + * @return {object} a connected-on-demand cluster client + */ + static createRedisClusterClient(nodeUrls) { + const rootNodes = nodeUrls.map((url) => ({ url })) + const firstNode = RedisAdapter.parseRedisUrl(nodeUrls[0]) + + const defaults = {} + if (firstNode.username) { + defaults.username = decodeURIComponent(firstNode.username) + } + if (firstNode.password) { + defaults.password = decodeURIComponent(firstNode.password) + } + if (firstNode.protocol === 'rediss:') { + defaults.socket = { tls: true } + } + + return createCluster({ + rootNodes, + ...CLIENT_OPTIONS, + ...(Object.keys(defaults).length > 0 ? { defaults } : {}), + }) + } + constructor(redisClient, options = {}) { const { prefix = 'general_', ttl = null } = options super() @@ -184,10 +255,8 @@ export default class RedisAdapter extends StorageAdapter { async clear() { try { - for await (const keys of this.client.scanIterator({ MATCH: `${this.prefix}*`, COUNT: 100 })) { - if (keys.length > 0) { - await this.client.del(keys) - } + for await (const keys of scanKeys(this.client, { MATCH: `${this.prefix}*`, COUNT: 100 })) { + await deleteKeys(this.client, keys) } } catch (error) { if (RedisAdapter.isClientClosedError(error)) { diff --git a/websocket_server/Services/ServerService.js b/websocket_server/Services/ServerService.js index bbf25981..9e07930a 100644 --- a/websocket_server/Services/ServerService.js +++ b/websocket_server/Services/ServerService.js @@ -26,12 +26,16 @@ export default class ServerService { ? RedisAdapter.createRedisClient() : null - if (this.redisClient) { - this.redisClient.connect().catch(error => { - console.error('Failed to connect to Redis:', error) - throw error - }) - } + // Services that use Redis wait on this before they run anything, so that + // the server does not accept clients while the connection is still being + // established. With a cluster client the slots are not known until then, + // and an early command fails rather than waiting. + this.redisReady = this.redisClient + ? this.redisClient.connect().catch((error) => { + console.error('Failed to connect to Redis:', error) + throw error + }) + : Promise.resolve() this.socketDataStorage = Config.STORAGE_STRATEGY === 'redis' ? StorageService.create('redis', this.redisClient, { prefix: 'socket_', ttl: Config.SESSION_TTL }) @@ -57,6 +61,7 @@ export default class ServerService { this.socketDataStorage, this.cachedTokenStorage, this.redisClient, + this.redisReady, ) // Update system monitor with socket service reference diff --git a/websocket_server/Services/SocketService.js b/websocket_server/Services/SocketService.js index db9fc816..50b7cf73 100644 --- a/websocket_server/Services/SocketService.js +++ b/websocket_server/Services/SocketService.js @@ -63,10 +63,11 @@ export default class SocketService { return sessions } - constructor(server, socketDataStorage, cachedTokenStorage, redisClient) { + constructor(server, socketDataStorage, cachedTokenStorage, redisClient, redisReady = Promise.resolve()) { this.socketDataStorage = socketDataStorage this.cachedTokenStorage = cachedTokenStorage this.redisClient = redisClient + this.redisReady = redisReady this.sessionStore = new SessionStore(this.socketDataStorage) this.recordingServices = new Map() this.pendingRecordingStops = new Map() @@ -217,6 +218,10 @@ export default class SocketService { } async init() { + // Nothing here may talk to Redis before the connection is up: with a + // cluster client the slots are still unknown and the first command fails + // rather than waiting for them. + await this.redisReady await this.setupAdapter() await this.clusterService.start() this.setupServerSideEvents() diff --git a/websocket_server/Stores/RoomStateStore.js b/websocket_server/Stores/RoomStateStore.js index be028853..f78c7dfd 100644 --- a/websocket_server/Stores/RoomStateStore.js +++ b/websocket_server/Stores/RoomStateStore.js @@ -3,6 +3,8 @@ * SPDX-License-Identifier: AGPL-3.0-or-later */ +import { scanKeys } from '../Utilities/RedisUtility.js' + export default class RoomStateStore { constructor({ redisClient = null, prefix = 'state_', defaultTtlMs = null } = {}) { @@ -60,7 +62,7 @@ export default class RoomStateStore { if (this.shouldUseRedis()) { const keys = [] const fullMatch = this.#fullKey(matchPattern) - for await (const batch of this.redisClient.scanIterator({ MATCH: fullMatch })) { + for await (const batch of scanKeys(this.redisClient, { MATCH: fullMatch })) { for (const key of batch) { keys.push(key.slice(this.prefix.length)) } @@ -77,7 +79,7 @@ export default class RoomStateStore { if (this.shouldUseRedis()) { const keys = [] const fullMatch = this.#fullKey(matchPattern) - for await (const batch of this.redisClient.scanIterator({ MATCH: fullMatch })) { + for await (const batch of scanKeys(this.redisClient, { MATCH: fullMatch })) { for (const key of batch) { keys.push(key.slice(this.prefix.length)) } diff --git a/websocket_server/Utilities/RedisUtility.js b/websocket_server/Utilities/RedisUtility.js new file mode 100644 index 00000000..a7b4cd88 --- /dev/null +++ b/websocket_server/Utilities/RedisUtility.js @@ -0,0 +1,61 @@ +/** + * SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors + * SPDX-License-Identifier: AGPL-3.0-or-later + */ + +/** + * Whether the given client talks to a Redis Cluster rather than a single node. + * + * @param {object} client a redis client + * @return {boolean} true for a cluster client + */ +export function isClusterClient(client) { + return typeof client?.nodeClient === 'function' +} + +/** + * Iterate the keys matching a pattern, one batch per SCAN reply. + * + * A scan only ever covers the keyspace of the node it runs on, which is why the + * cluster client has no scanIterator() of its own. Scan every primary through + * its own client instead, so the caller sees the whole keyspace either way. + * + * @param {object} client a redis client, one node or a cluster + * @param {object} options the options to hand to scanIterator() + * @yields {string[]} the keys of each SCAN reply + */ +export async function* scanKeys(client, options) { + if (!isClusterClient(client)) { + yield* client.scanIterator(options) + return + } + + for (const master of client.masters) { + const nodeClient = await client.nodeClient(master) + yield* nodeClient.scanIterator(options) + } +} + +/** + * Delete the given keys. + * + * Redis refuses a DEL whose keys do not all live in the same hash slot, and the + * keys returned by a scan across a cluster generally do not, so delete them one + * by one there. A single node still takes the whole batch in one call. + * + * @param {object} client a redis client, one node or a cluster + * @param {string[]} keys the keys to delete + * @return {Promise} + */ +export async function deleteKeys(client, keys) { + if (keys.length === 0) { + return + } + + if (!isClusterClient(client)) { + await client.del(keys) + return + } + + await Promise.all(keys.map((key) => client.del(key))) +}