fix(kv): stop multi-key commands spanning cluster slots (#2310)

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<Array<string>>;
+} {
+	const commands: Array<Array<string>> = [];
+	const client = {
+		set: async (key: string) => {
+			commands.push([key]);
+			return 'OK';
+		},
+		setex: async (key: string) => {
+			commands.push([key]);
+		},
+		pipeline: () => {
+			const keys: Array<string> = [];
+			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([]);
+	});
+});

fluxer_api/pkgs/cache/src/providers/KVCacheProvider.ts

@@ -137,37 +137,16 @@ export class KVCacheProvider extends ICacheService {
 		}>,
 	): Promise<void> {
 		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<number> {

fluxer_api/pkgs/kv_client/src/KVClient.ts

@@ -355,7 +355,10 @@ export class KVClient implements IKVProvider {
 	}
 
 	async mget(...keys: Array<string>): Promise<Array<string | null>> {
-		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<string>): Promise<void> {
@@ -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<number> {

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<string>}>,
+	store: new Map<string, string>(),
+}));
+
+vi.mock('ioredis', () => {
+	class MockRedis {
+		async get(key: string): Promise<string | null> {
+			commands.push({name: 'get', keys: [key]});
+			return store.get(key) ?? null;
+		}
+
+		async set(key: string, value: string): Promise<string> {
+			commands.push({name: 'set', keys: [key]});
+			store.set(key, value);
+			return 'OK';
+		}
+
+		async del(...keys: Array<string>): Promise<number> {
+			commands.push({name: 'del', keys});
+			return keys.filter((key) => store.delete(key)).length;
+		}
+
+		async mget(...keys: Array<string>): Promise<Array<string | null>> {
+			commands.push({name: 'mget', keys});
+			return keys.map((key) => store.get(key) ?? null);
+		}
+
+		async mset(...args: Array<string>): Promise<string> {
+			const keys: Array<string> = [];
+			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<string>}> {
+	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([]);
+	});
+});

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<void> {
+		await Promise.all(batch.map(async (entry) => this.kvClient.setex(entry.key, TTL_SECONDS, entry.value)));
+	}
+
 	async rebuildActivities(): Promise<void> {
 		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');

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([]);
+	});
+});

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<string>;
-}
-
-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<RecordedBatch> = [];
-
-	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([]);
 	});
 });

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<string>;
+}
+
+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<RecordedBatch> = [];
+
+	override pipeline(): IKVPipeline {
+		return this.recordBatch('pipeline', super.pipeline());
+	}
+
+	override multi(): IKVPipeline {
+		return this.recordBatch('multi', super.multi());
+	}
+
+	crossSlotBatches(): Array<RecordedBatch> {
+		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;
+	}
+}