From 24138b70f1dd39afa95bf61edaa2b5e4ffa1e88d Mon Sep 17 00:00:00 2001 From: Hampus Date: Tue, 1 Sep 2026 01:39:16 +0200 Subject: [PATCH] fix(kv): stop multi-key commands spanning cluster slots (#2310) --- .../src/__tests__/CacheClusterSlots.test.ts | 74 ++++++++++++ .../cache/src/providers/KVCacheProvider.ts | 41 ++----- fluxer_api/pkgs/kv_client/src/KVClient.ts | 13 +- .../__tests__/KVClientClusterSlots.test.ts | 110 +++++++++++++++++ .../api/infrastructure/KVActivityTracker.ts | 23 ++-- .../KVActivityTrackerClusterSlots.test.ts | 33 ++++++ .../tests/VoiceRoomStoreClusterSlots.test.ts | 110 +---------------- .../test/mocks/BatchRecordingKVProvider.ts | 112 ++++++++++++++++++ 8 files changed, 361 insertions(+), 155 deletions(-) create mode 100644 fluxer_api/pkgs/cache/src/__tests__/CacheClusterSlots.test.ts create mode 100644 fluxer_api/pkgs/kv_client/src/__tests__/KVClientClusterSlots.test.ts create mode 100644 fluxer_api/src/api/infrastructure/tests/KVActivityTrackerClusterSlots.test.ts create mode 100644 fluxer_api/src/api/test/mocks/BatchRecordingKVProvider.ts diff --git a/fluxer_api/pkgs/cache/src/__tests__/CacheClusterSlots.test.ts b/fluxer_api/pkgs/cache/src/__tests__/CacheClusterSlots.test.ts new file mode 100644 index 000000000..62042abd5 --- /dev/null +++ b/fluxer_api/pkgs/cache/src/__tests__/CacheClusterSlots.test.ts @@ -0,0 +1,74 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +import {KVCacheProvider} from '@pkgs/cache/src/providers/KVCacheProvider'; +import type {IKVPipeline, IKVProvider} from '@pkgs/kv_client/src/IKVProvider'; +import {describe, expect, it} from 'vitest'; + +function hashSlot(key: string): number { + let hashed = key; + const start = key.indexOf('{'); + if (start !== -1) { + const end = key.indexOf('}', start + 1); + if (end > start + 1) { + hashed = key.slice(start + 1, end); + } + } + let crc = 0; + for (let index = 0; index < hashed.length; index += 1) { + crc ^= (hashed.charCodeAt(index) & 0xff) << 8; + for (let bit = 0; bit < 8; bit += 1) { + crc = (crc & 0x8000) === 0 ? (crc << 1) & 0xffff : ((crc << 1) ^ 0x1021) & 0xffff; + } + } + return crc % 16384; +} + +function createRecordingProvider(): { + client: IKVProvider; + commands: Array>; +} { + const commands: Array> = []; + const client = { + set: async (key: string) => { + commands.push([key]); + return 'OK'; + }, + setex: async (key: string) => { + commands.push([key]); + }, + pipeline: () => { + const keys: Array = []; + commands.push(keys); + const batch = { + set: (key: string) => { + keys.push(key); + return batch; + }, + setex: (key: string) => { + keys.push(key); + return batch; + }, + exec: async () => [], + } as unknown as IKVPipeline; + return batch; + }, + } as unknown as IKVProvider; + return {client, commands}; +} + +describe('KVCacheProvider cluster hash slots', () => { + it('keeps a multi entry write off batched commands that span hash slots', async () => { + const {client, commands} = createRecordingProvider(); + const provider = new KVCacheProvider({client}); + + expect(hashSlot('cache:alpha')).not.toBe(hashSlot('cache:beta')); + + await provider.mset([ + {key: 'cache:alpha', value: 1, ttlSeconds: 60}, + {key: 'cache:beta', value: 2}, + ]); + + expect(commands.flat().sort()).toEqual(['cache:alpha', 'cache:beta']); + expect(commands.filter((keys) => new Set(keys.map(hashSlot)).size > 1)).toEqual([]); + }); +}); diff --git a/fluxer_api/pkgs/cache/src/providers/KVCacheProvider.ts b/fluxer_api/pkgs/cache/src/providers/KVCacheProvider.ts index 8b37a9918..444c5e880 100644 --- a/fluxer_api/pkgs/cache/src/providers/KVCacheProvider.ts +++ b/fluxer_api/pkgs/cache/src/providers/KVCacheProvider.ts @@ -137,37 +137,16 @@ export class KVCacheProvider extends ICacheService { }>, ): Promise { if (entries.length === 0) return; - const withoutTtl: Array<{ - key: string; - value: T; - }> = []; - const withTtl: Array<{ - key: string; - value: T; - ttlSeconds: number; - }> = []; - for (const entry of entries) { - if (entry.ttlSeconds) { - withTtl.push({ - key: entry.key, - value: entry.value, - ttlSeconds: entry.ttlSeconds, - }); - } else { - withoutTtl.push({ - key: entry.key, - value: entry.value, - }); - } - } - const pipeline = this.client.pipeline(); - for (const entry of withoutTtl) { - pipeline.set(entry.key, serializeValue(entry.value)); - } - for (const entry of withTtl) { - pipeline.setex(entry.key, entry.ttlSeconds, serializeValue(entry.value)); - } - await pipeline.exec(); + await Promise.all( + entries.map(async (entry) => { + const serialized = serializeValue(entry.value); + if (entry.ttlSeconds) { + await this.client.setex(entry.key, entry.ttlSeconds, serialized); + return; + } + await this.client.set(entry.key, serialized); + }), + ); } async deletePattern(pattern: string): Promise { diff --git a/fluxer_api/pkgs/kv_client/src/KVClient.ts b/fluxer_api/pkgs/kv_client/src/KVClient.ts index 46ba39be0..1fca0be6e 100644 --- a/fluxer_api/pkgs/kv_client/src/KVClient.ts +++ b/fluxer_api/pkgs/kv_client/src/KVClient.ts @@ -355,7 +355,10 @@ export class KVClient implements IKVProvider { } async mget(...keys: Array): Promise> { - return await this.execute('mget', async () => this.client.mget(...keys)); + if (keys.length === 0) { + return []; + } + return await this.execute('mget', async () => await Promise.all(keys.map(async (key) => this.client.get(key)))); } async mset(...args: Array): Promise { @@ -363,9 +366,8 @@ export class KVClient implements IKVProvider { if (entries.length === 0) { return; } - const pairs = entries.flatMap((entry) => [entry.key, entry.value]); await this.execute('mset', async () => { - await this.client.mset(...pairs); + await Promise.all(entries.map(async (entry) => this.client.set(entry.key, entry.value))); }); } @@ -373,7 +375,10 @@ export class KVClient implements IKVProvider { if (keys.length === 0) { return 0; } - return await this.execute('del', async () => this.client.del(...keys)); + return await this.execute('del', async () => { + const deleted = await Promise.all(keys.map(async (key) => this.client.del(key))); + return deleted.reduce((total, count) => total + count, 0); + }); } async exists(key: string): Promise { diff --git a/fluxer_api/pkgs/kv_client/src/__tests__/KVClientClusterSlots.test.ts b/fluxer_api/pkgs/kv_client/src/__tests__/KVClientClusterSlots.test.ts new file mode 100644 index 000000000..cb5093ebc --- /dev/null +++ b/fluxer_api/pkgs/kv_client/src/__tests__/KVClientClusterSlots.test.ts @@ -0,0 +1,110 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +import {KVClient} from '@pkgs/kv_client/src/KVClient'; +import {beforeEach, describe, expect, it, vi} from 'vitest'; + +const {commands, store} = vi.hoisted(() => ({ + commands: [] as Array<{name: string; keys: Array}>, + store: new Map(), +})); + +vi.mock('ioredis', () => { + class MockRedis { + async get(key: string): Promise { + commands.push({name: 'get', keys: [key]}); + return store.get(key) ?? null; + } + + async set(key: string, value: string): Promise { + commands.push({name: 'set', keys: [key]}); + store.set(key, value); + return 'OK'; + } + + async del(...keys: Array): Promise { + commands.push({name: 'del', keys}); + return keys.filter((key) => store.delete(key)).length; + } + + async mget(...keys: Array): Promise> { + commands.push({name: 'mget', keys}); + return keys.map((key) => store.get(key) ?? null); + } + + async mset(...args: Array): Promise { + const keys: Array = []; + for (let index = 0; index + 1 < args.length; index += 2) { + keys.push(args[index]); + store.set(args[index], args[index + 1]); + } + commands.push({name: 'mset', keys}); + return 'OK'; + } + } + return {default: MockRedis, Cluster: MockRedis}; +}); + +function hashSlot(key: string): number { + let hashed = key; + const start = key.indexOf('{'); + if (start !== -1) { + const end = key.indexOf('}', start + 1); + if (end > start + 1) { + hashed = key.slice(start + 1, end); + } + } + let crc = 0; + for (let index = 0; index < hashed.length; index += 1) { + crc ^= (hashed.charCodeAt(index) & 0xff) << 8; + for (let bit = 0; bit < 8; bit += 1) { + crc = (crc & 0x8000) === 0 ? (crc << 1) & 0xffff : ((crc << 1) ^ 0x1021) & 0xffff; + } + } + return crc % 16384; +} + +function crossSlotCommands(): Array<{name: string; keys: Array}> { + return commands.filter((command) => new Set(command.keys.map(hashSlot)).size > 1); +} + +function createClient(): KVClient { + return new KVClient('redis://127.0.0.1:6379'); +} + +describe('KVClient cluster hash slots', () => { + beforeEach(() => { + commands.length = 0; + store.clear(); + }); + + it('reads several keys without a command spanning hash slots', async () => { + expect(hashSlot('slot:alpha')).not.toBe(hashSlot('slot:beta')); + const client = createClient(); + await client.set('slot:alpha', 'one'); + + await expect(client.mget('slot:alpha', 'slot:beta')).resolves.toEqual(['one', null]); + expect(crossSlotCommands()).toEqual([]); + }); + + it('writes several keys without a command spanning hash slots', async () => { + expect(hashSlot('slot:alpha')).not.toBe(hashSlot('slot:beta')); + const client = createClient(); + + await client.mset('slot:alpha', 'one', 'slot:beta', 'two'); + + expect(store.get('slot:alpha')).toBe('one'); + expect(store.get('slot:beta')).toBe('two'); + expect(crossSlotCommands()).toEqual([]); + }); + + it('deletes several keys without a command spanning hash slots', async () => { + expect(hashSlot('slot:alpha')).not.toBe(hashSlot('slot:beta')); + const client = createClient(); + await client.set('slot:alpha', 'one'); + await client.set('slot:beta', 'two'); + + await expect(client.del('slot:alpha', 'slot:beta', 'slot:gamma')).resolves.toBe(2); + expect(store.size).toBe(0); + expect(crossSlotCommands()).toEqual([]); + }); +}); diff --git a/fluxer_api/src/api/infrastructure/KVActivityTracker.ts b/fluxer_api/src/api/infrastructure/KVActivityTracker.ts index d3b2caaa5..36209b205 100644 --- a/fluxer_api/src/api/infrastructure/KVActivityTracker.ts +++ b/fluxer_api/src/api/infrastructure/KVActivityTracker.ts @@ -74,6 +74,10 @@ export class KVActivityTracker { return age > STATE_VERSION_TTL_SECONDS; } + private async writeActivityBatch(batch: ReadonlyArray<{key: string; value: string}>): Promise { + await Promise.all(batch.map(async (entry) => this.kvClient.setex(entry.key, TTL_SECONDS, entry.value))); + } + async rebuildActivities(): Promise { Logger.info('Starting activity tracker rebuild from Cassandra'); const userRepository = new UserRepository(); @@ -81,8 +85,7 @@ export class KVActivityTracker { const kvBatchSize = 1000; let processedCount = 0; let usersWithActivity = 0; - let pipeline = this.kvClient.pipeline(); - let pipelineCount = 0; + let batch: Array<{key: string; value: string}> = []; let pageState: string | null = null; let iterationCount = 0; while (!this.isShuttingDown) { @@ -93,15 +96,11 @@ export class KVActivityTracker { } for (const user of users) { if (user.lastActiveAt) { - const key = this.getActivityKey(user.id); - const value = user.lastActiveAt.getTime().toString(); - pipeline.setex(key, TTL_SECONDS, value); - pipelineCount++; + batch.push({key: this.getActivityKey(user.id), value: user.lastActiveAt.getTime().toString()}); usersWithActivity++; - if (pipelineCount >= kvBatchSize) { - await pipeline.exec(); - pipeline = this.kvClient.pipeline(); - pipelineCount = 0; + if (batch.length >= kvBatchSize) { + await this.writeActivityBatch(batch); + batch = []; } } processedCount++; @@ -122,8 +121,8 @@ export class KVActivityTracker { Logger.warn({processedCount, usersWithActivity}, 'Activity tracker rebuild interrupted by shutdown'); return; } - if (pipelineCount > 0) { - await pipeline.exec(); + if (batch.length > 0) { + await this.writeActivityBatch(batch); } await this.kvClient.setex(STATE_VERSION_KEY, STATE_VERSION_TTL_SECONDS, Date.now().toString()); Logger.info({processedCount, usersWithActivity}, 'Activity tracker rebuild completed'); diff --git a/fluxer_api/src/api/infrastructure/tests/KVActivityTrackerClusterSlots.test.ts b/fluxer_api/src/api/infrastructure/tests/KVActivityTrackerClusterSlots.test.ts new file mode 100644 index 000000000..aca0b4343 --- /dev/null +++ b/fluxer_api/src/api/infrastructure/tests/KVActivityTrackerClusterSlots.test.ts @@ -0,0 +1,33 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +import {afterEach, describe, expect, it, vi} from 'vitest'; +import {createUserID} from '../../BrandedTypes'; +import type {User} from '../../models/User'; +import {BatchRecordingKVProvider, hashSlot} from '../../test/mocks/BatchRecordingKVProvider'; +import {UserRepository} from '../../user/repositories/UserRepository'; +import {KVActivityTracker} from '../KVActivityTracker'; + +function createUser(id: bigint, lastActiveAt: Date): User { + return {id: createUserID(id), lastActiveAt} as unknown as User; +} + +describe('KVActivityTracker cluster hash slots', () => { + afterEach(() => { + vi.restoreAllMocks(); + }); + + it('keeps the rebuild writes off batched commands that span hash slots', async () => { + const kvClient = new BatchRecordingKVProvider(); + const lastActiveAt = new Date('2026-06-01T00:00:00.000Z'); + const users = [createUser(1234n, lastActiveAt), createUser(5678n, lastActiveAt)]; + vi.spyOn(UserRepository.prototype, 'scanAllUsersPage').mockResolvedValue({users, pageState: null}); + + expect(hashSlot('user_activity:1234')).not.toBe(hashSlot('user_activity:5678')); + + await new KVActivityTracker(kvClient).rebuildActivities(); + + expect(await kvClient.get('user_activity:1234')).toBe(lastActiveAt.getTime().toString()); + expect(await kvClient.get('user_activity:5678')).toBe(lastActiveAt.getTime().toString()); + expect(kvClient.crossSlotBatches()).toEqual([]); + }); +}); diff --git a/fluxer_api/src/api/infrastructure/tests/VoiceRoomStoreClusterSlots.test.ts b/fluxer_api/src/api/infrastructure/tests/VoiceRoomStoreClusterSlots.test.ts index ace168179..b5debb4b3 100644 --- a/fluxer_api/src/api/infrastructure/tests/VoiceRoomStoreClusterSlots.test.ts +++ b/fluxer_api/src/api/infrastructure/tests/VoiceRoomStoreClusterSlots.test.ts @@ -1,116 +1,11 @@ // SPDX-License-Identifier: AGPL-3.0-or-later -import type {IKVPipeline} from '@pkgs/kv_client/src/IKVProvider'; import {describe, expect, it} from 'vitest'; import {createChannelID, createGuildID} from '../../BrandedTypes'; -import {MockKVProvider} from '../../test/mocks/MockKVProvider'; +import {BatchRecordingKVProvider, hashSlot} from '../../test/mocks/BatchRecordingKVProvider'; import {VOICE_OCCUPANCY_REGION_KEY_PREFIX, VOICE_OCCUPANCY_SERVER_KEY_PREFIX} from '../../voice/VoiceConstants'; import {VoiceRoomStore} from '../VoiceRoomStore'; -type BatchMode = 'multi' | 'pipeline'; - -interface RecordedBatch { - mode: BatchMode; - keys: Array; -} - -function hashSlot(key: string): number { - let hashed = key; - const start = key.indexOf('{'); - if (start !== -1) { - const end = key.indexOf('}', start + 1); - if (end > start + 1) { - hashed = key.slice(start + 1, end); - } - } - let crc = 0; - for (let index = 0; index < hashed.length; index += 1) { - crc ^= (hashed.charCodeAt(index) & 0xff) << 8; - for (let bit = 0; bit < 8; bit += 1) { - crc = (crc & 0x8000) === 0 ? (crc << 1) & 0xffff : ((crc << 1) ^ 0x1021) & 0xffff; - } - } - return crc % 16384; -} - -class BatchRecordingKVProvider extends MockKVProvider { - readonly batches: Array = []; - - override pipeline(): IKVPipeline { - return this.recordBatch('pipeline', super.pipeline()); - } - - override multi(): IKVPipeline { - return this.recordBatch('multi', super.multi()); - } - - private recordBatch(mode: BatchMode, inner: IKVPipeline): IKVPipeline { - const batch: RecordedBatch = {mode, keys: []}; - this.batches.push(batch); - const recorded: IKVPipeline = { - get: (key) => { - batch.keys.push(key); - inner.get(key); - return recorded; - }, - set: (key, value) => { - batch.keys.push(key); - inner.set(key, value); - return recorded; - }, - setex: (key, ttlSeconds, value) => { - batch.keys.push(key); - inner.setex(key, ttlSeconds, value); - return recorded; - }, - del: (key) => { - batch.keys.push(key); - inner.del(key); - return recorded; - }, - expire: (key, ttlSeconds) => { - batch.keys.push(key); - inner.expire(key, ttlSeconds); - return recorded; - }, - sadd: (key, ...members) => { - batch.keys.push(key); - inner.sadd(key, ...members); - return recorded; - }, - srem: (key, ...members) => { - batch.keys.push(key); - inner.srem(key, ...members); - return recorded; - }, - zadd: (key, score, value) => { - batch.keys.push(key); - inner.zadd(key, score, value); - return recorded; - }, - zrem: (key, ...members) => { - batch.keys.push(key); - inner.zrem(key, ...members); - return recorded; - }, - hgetall: (key) => { - batch.keys.push(key); - inner.hgetall(key); - return recorded; - }, - mset: (...args) => { - for (let index = 0; index + 1 < args.length; index += 2) { - batch.keys.push(args[index]); - } - inner.mset(...args); - return recorded; - }, - exec: async () => await inner.exec(), - }; - return recorded; - } -} - describe('VoiceRoomStore cluster hash slots', () => { it('keeps occupancy writes off batched commands that span hash slots', async () => { const kvClient = new BatchRecordingKVProvider(); @@ -131,7 +26,6 @@ describe('VoiceRoomStore cluster hash slots', () => { expect(await kvClient.smembers(regionKey)).toEqual([]); expect(await kvClient.smembers(serverKey)).toEqual([]); - const crossSlotBatches = kvClient.batches.filter((batch) => new Set(batch.keys.map(hashSlot)).size > 1); - expect(crossSlotBatches).toEqual([]); + expect(kvClient.crossSlotBatches()).toEqual([]); }); }); diff --git a/fluxer_api/src/api/test/mocks/BatchRecordingKVProvider.ts b/fluxer_api/src/api/test/mocks/BatchRecordingKVProvider.ts new file mode 100644 index 000000000..477b1322b --- /dev/null +++ b/fluxer_api/src/api/test/mocks/BatchRecordingKVProvider.ts @@ -0,0 +1,112 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +import type {IKVPipeline} from '@pkgs/kv_client/src/IKVProvider'; +import {MockKVProvider} from './MockKVProvider'; + +type BatchMode = 'multi' | 'pipeline'; + +interface RecordedBatch { + mode: BatchMode; + keys: Array; +} + +export function hashSlot(key: string): number { + let hashed = key; + const start = key.indexOf('{'); + if (start !== -1) { + const end = key.indexOf('}', start + 1); + if (end > start + 1) { + hashed = key.slice(start + 1, end); + } + } + let crc = 0; + for (let index = 0; index < hashed.length; index += 1) { + crc ^= (hashed.charCodeAt(index) & 0xff) << 8; + for (let bit = 0; bit < 8; bit += 1) { + crc = (crc & 0x8000) === 0 ? (crc << 1) & 0xffff : ((crc << 1) ^ 0x1021) & 0xffff; + } + } + return crc % 16384; +} + +export class BatchRecordingKVProvider extends MockKVProvider { + readonly batches: Array = []; + + override pipeline(): IKVPipeline { + return this.recordBatch('pipeline', super.pipeline()); + } + + override multi(): IKVPipeline { + return this.recordBatch('multi', super.multi()); + } + + crossSlotBatches(): Array { + return this.batches.filter((batch) => new Set(batch.keys.map(hashSlot)).size > 1); + } + + private recordBatch(mode: BatchMode, inner: IKVPipeline): IKVPipeline { + const batch: RecordedBatch = {mode, keys: []}; + this.batches.push(batch); + const recorded: IKVPipeline = { + get: (key) => { + batch.keys.push(key); + inner.get(key); + return recorded; + }, + set: (key, value) => { + batch.keys.push(key); + inner.set(key, value); + return recorded; + }, + setex: (key, ttlSeconds, value) => { + batch.keys.push(key); + inner.setex(key, ttlSeconds, value); + return recorded; + }, + del: (key) => { + batch.keys.push(key); + inner.del(key); + return recorded; + }, + expire: (key, ttlSeconds) => { + batch.keys.push(key); + inner.expire(key, ttlSeconds); + return recorded; + }, + sadd: (key, ...members) => { + batch.keys.push(key); + inner.sadd(key, ...members); + return recorded; + }, + srem: (key, ...members) => { + batch.keys.push(key); + inner.srem(key, ...members); + return recorded; + }, + zadd: (key, score, value) => { + batch.keys.push(key); + inner.zadd(key, score, value); + return recorded; + }, + zrem: (key, ...members) => { + batch.keys.push(key); + inner.zrem(key, ...members); + return recorded; + }, + hgetall: (key) => { + batch.keys.push(key); + inner.hgetall(key); + return recorded; + }, + mset: (...args) => { + for (let index = 0; index + 1 < args.length; index += 2) { + batch.keys.push(args[index]); + } + inner.mset(...args); + return recorded; + }, + exec: async () => await inner.exec(), + }; + return recorded; + } +}