From 7d826d1602d947d7bf204b6475853fce1e5858a3 Mon Sep 17 00:00:00 2001 From: Hampus Date: Sun, 30 Aug 2026 22:15:11 +0200 Subject: [PATCH] perf(rpc): bound concurrency in user batch fanout handlers (#2124) --- fluxer_api/src/api/rpc/RpcService.ts | 27 +++-- .../api/rpc/tests/RpcUserBatchFanout.test.ts | 110 ++++++++++++++++++ .../IUserRelationshipRepository.ts | 1 + .../UserRelationshipRepository.ts | 12 ++ .../api/user/repositories/UserRepository.ts | 4 + packages/schema/src/domains/rpc/RpcSchemas.ts | 6 +- 6 files changed, 148 insertions(+), 12 deletions(-) create mode 100644 fluxer_api/src/api/rpc/tests/RpcUserBatchFanout.test.ts diff --git a/fluxer_api/src/api/rpc/RpcService.ts b/fluxer_api/src/api/rpc/RpcService.ts index 811476560..494b10bcf 100644 --- a/fluxer_api/src/api/rpc/RpcService.ts +++ b/fluxer_api/src/api/rpc/RpcService.ts @@ -1627,9 +1627,16 @@ export class RpcService { }> { const {userIds, guildId} = params; const actualGuildId = guildId === createGuildID(0n) ? null : guildId; - const userGuildSettings = await Promise.all( - userIds.map((userId) => this.userRepository.findGuildSettings(userId, actualGuildId)), + const settled = await allSettledWithConcurrency(userIds, RPC_RESPONSE_MAP_CONCURRENCY, (userId) => + this.userRepository.findGuildSettings(userId, actualGuildId), ); + const userGuildSettings: Array = []; + for (const result of settled) { + if (result.status === 'rejected') { + throw result.reason; + } + userGuildSettings.push(result.value); + } return {user_guild_settings: userGuildSettings}; } @@ -1861,15 +1868,15 @@ export class RpcService { private async getUserBlockedIds(params: {userIds: Array}): Promise>> { const {userIds} = params; + const settled = await allSettledWithConcurrency(userIds, RPC_RESPONSE_MAP_CONCURRENCY, (userId) => + this.userRepository.listBlockedUserIds(userId), + ); const result: Record> = {}; - const relationshipsPromises = userIds.map(async (userId) => { - const relationships = await this.userRepository.listRelationships(userId); - const blockedIds = relationships.filter((rel) => rel.type === 2).map((rel) => rel.targetUserId.toString()); - return {userId, blockedIds}; - }); - const results = await Promise.all(relationshipsPromises); - for (const {userId, blockedIds} of results) { - result[userId.toString()] = blockedIds; + for (const [index, settledBlockedIds] of settled.entries()) { + if (settledBlockedIds.status === 'rejected') { + throw settledBlockedIds.reason; + } + result[userIds[index]!.toString()] = settledBlockedIds.value.map((blockedUserId) => blockedUserId.toString()); } return result; } diff --git a/fluxer_api/src/api/rpc/tests/RpcUserBatchFanout.test.ts b/fluxer_api/src/api/rpc/tests/RpcUserBatchFanout.test.ts new file mode 100644 index 000000000..c46e65972 --- /dev/null +++ b/fluxer_api/src/api/rpc/tests/RpcUserBatchFanout.test.ts @@ -0,0 +1,110 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +import {UserNotificationSettings} from '@fluxer/constants/src/UserConstants'; +import {RpcRequest} from '@fluxer/schema/src/domains/rpc/RpcSchemas'; +import {afterEach, beforeEach, describe, expect, test} from 'vitest'; +import {createTestAccount} from '../../auth/tests/AuthTestUtils'; +import {type ApiTestHarness, createApiTestHarness} from '../../test/ApiTestHarness'; +import {HTTP_STATUS} from '../../test/TestConstants'; +import {createBuilder} from '../../test/TestRequestBuilder'; +import {blockUser} from '../../user/tests/RelationshipTestUtils'; + +interface RpcUserGuildSettingsResponse { + type: 'get_user_guild_settings'; + data: { + user_guild_settings: Array<{message_notifications: number} | null>; + }; +} + +interface RpcUserBlockedIdsResponse { + type: 'get_user_blocked_ids'; + data: Record>; +} + +const UNKNOWN_USER_ID = '1'; +const BATCH_SIZE = 96; + +async function setDmNotificationLevel(harness: ApiTestHarness, token: string, level: number): Promise { + await createBuilder(harness, token) + .patch('/users/@me/guilds/@me/settings') + .body({message_notifications: level}) + .expect(HTTP_STATUS.OK) + .execute(); +} + +describe('RpcService user batch fanout', () => { + let harness: ApiTestHarness; + beforeEach(async () => { + harness = await createApiTestHarness(); + }); + afterEach(async () => { + await harness?.shutdown(); + }); + + test('keeps get_user_guild_settings index aligned for batches larger than the concurrency cap', async () => { + const allMessages = await createTestAccount(harness); + const onlyMentions = await createTestAccount(harness); + const noMessages = await createTestAccount(harness); + await setDmNotificationLevel(harness, allMessages.token, UserNotificationSettings.ALL_MESSAGES); + await setDmNotificationLevel(harness, onlyMentions.token, UserNotificationSettings.ONLY_MENTIONS); + await setDmNotificationLevel(harness, noMessages.token, UserNotificationSettings.NO_MESSAGES); + const cycle = [ + {userId: allMessages.userId, expected: UserNotificationSettings.ALL_MESSAGES}, + {userId: onlyMentions.userId, expected: UserNotificationSettings.ONLY_MENTIONS}, + {userId: UNKNOWN_USER_ID, expected: null}, + {userId: noMessages.userId, expected: UserNotificationSettings.NO_MESSAGES}, + ]; + const requested = Array.from({length: BATCH_SIZE}, (_value, index) => cycle[index % cycle.length]!); + const rpcResponse = await createBuilder(harness, '') + .post('/test/rpc-session-init') + .body({ + type: 'get_user_guild_settings', + user_ids: requested.map((entry) => entry.userId), + guild_id: '0', + }) + .expect(HTTP_STATUS.OK) + .execute(); + expect(rpcResponse.data.user_guild_settings).toHaveLength(BATCH_SIZE); + expect( + rpcResponse.data.user_guild_settings.map((settings) => settings?.message_notifications ?? null), + ).toStrictEqual(requested.map((entry) => entry.expected)); + }); + + test('rejects user batch requests above the schema bound', () => { + for (const type of ['get_user_guild_settings', 'get_user_blocked_ids']) { + const withinBound = RpcRequest.safeParse({ + type, + user_ids: Array.from({length: 1000}, () => UNKNOWN_USER_ID), + guild_id: '0', + }); + expect(withinBound.success).toBe(true); + const aboveBound = RpcRequest.safeParse({ + type, + user_ids: Array.from({length: 1001}, () => UNKNOWN_USER_ID), + guild_id: '0', + }); + expect(aboveBound.success).toBe(false); + } + }); + + test('returns blocked ids per user for get_user_blocked_ids', async () => { + const blocker = await createTestAccount(harness); + const blockedFirst = await createTestAccount(harness); + const blockedSecond = await createTestAccount(harness); + const bystander = await createTestAccount(harness); + await blockUser(harness, blocker.token, blockedFirst.userId); + await blockUser(harness, blocker.token, blockedSecond.userId); + const rpcResponse = await createBuilder(harness, '') + .post('/test/rpc-session-init') + .body({ + type: 'get_user_blocked_ids', + user_ids: [blocker.userId, bystander.userId], + }) + .expect(HTTP_STATUS.OK) + .execute(); + expect(rpcResponse.data[blocker.userId]?.slice().sort()).toStrictEqual( + [blockedFirst.userId, blockedSecond.userId].sort(), + ); + expect(rpcResponse.data[bystander.userId]).toStrictEqual([]); + }); +}); diff --git a/fluxer_api/src/api/user/repositories/IUserRelationshipRepository.ts b/fluxer_api/src/api/user/repositories/IUserRelationshipRepository.ts index b8362d268..b7c5dc93e 100644 --- a/fluxer_api/src/api/user/repositories/IUserRelationshipRepository.ts +++ b/fluxer_api/src/api/user/repositories/IUserRelationshipRepository.ts @@ -7,6 +7,7 @@ import type {UserNote} from '../../models/UserNote'; export interface IUserRelationshipRepository { listRelationships(sourceUserId: UserID): Promise>; + listBlockedUserIds(sourceUserId: UserID): Promise>; listIncomingRequests(userId: UserID): Promise>; hasReachedRelationshipLimit(sourceUserId: UserID, limit: number): Promise; getRelationship(sourceUserId: UserID, targetUserId: UserID, type: number): Promise; diff --git a/fluxer_api/src/api/user/repositories/UserRelationshipRepository.ts b/fluxer_api/src/api/user/repositories/UserRelationshipRepository.ts index dc14d4778..ecab4455e 100644 --- a/fluxer_api/src/api/user/repositories/UserRelationshipRepository.ts +++ b/fluxer_api/src/api/user/repositories/UserRelationshipRepository.ts @@ -21,6 +21,10 @@ const FETCH_NOTE_CQL = Notes.selectCql({ const FETCH_RELATIONSHIPS_CQL = Relationships.selectCql({ where: Relationships.where.eq('source_user_id'), }); +const FETCH_RELATIONSHIP_TARGET_TYPES_CQL = Relationships.selectCql({ + columns: ['target_user_id', 'type'], + where: Relationships.where.eq('source_user_id'), +}); const FETCH_RELATIONSHIPS_BY_TARGET_CQL = RelationshipsByTarget.selectCql({ where: RelationshipsByTarget.where.eq('target_user_id'), }); @@ -167,6 +171,14 @@ export class UserRelationshipRepository implements IUserRelationshipRepository { return relationships.map((rel) => new Relationship(rel)); } + async listBlockedUserIds(sourceUserId: UserID): Promise> { + const rows = await fetchMany>( + FETCH_RELATIONSHIP_TARGET_TYPES_CQL, + {source_user_id: sourceUserId}, + ); + return rows.filter((row) => row.type === RelationshipTypes.BLOCKED).map((row) => row.target_user_id); + } + async listIncomingRequests(userId: UserID): Promise> { const relationships = await this.listRelationships(userId); return relationships.filter((rel) => rel.type === RelationshipTypes.INCOMING_REQUEST); diff --git a/fluxer_api/src/api/user/repositories/UserRepository.ts b/fluxer_api/src/api/user/repositories/UserRepository.ts index 512851cde..67ee137c5 100644 --- a/fluxer_api/src/api/user/repositories/UserRepository.ts +++ b/fluxer_api/src/api/user/repositories/UserRepository.ts @@ -430,6 +430,10 @@ export class UserRepository implements IUserRepositoryAggregate { return this.relationshipRepo.listRelationships(sourceUserId); } + async listBlockedUserIds(sourceUserId: UserID): Promise> { + return this.relationshipRepo.listBlockedUserIds(sourceUserId); + } + async hasReachedRelationshipLimit(sourceUserId: UserID, limit: number): Promise { return this.relationshipRepo.hasReachedRelationshipLimit(sourceUserId, limit); } diff --git a/packages/schema/src/domains/rpc/RpcSchemas.ts b/packages/schema/src/domains/rpc/RpcSchemas.ts index 23fffbf7a..74ddffb4f 100644 --- a/packages/schema/src/domains/rpc/RpcSchemas.ts +++ b/packages/schema/src/domains/rpc/RpcSchemas.ts @@ -26,6 +26,8 @@ import { } from '@fluxer/schema/src/primitives/SchemaPrimitives'; import {z} from 'zod'; +const RPC_USER_BATCH_MAX = 1000; + export const RpcGuildCollectionType = z.enum([ 'guild', 'roles', @@ -79,7 +81,7 @@ export const RpcRequest = z.discriminatedUnion('type', [ }), z.object({ type: z.literal('get_user_guild_settings').describe('Request type for fetching user guild settings'), - user_ids: z.array(SnowflakeType).describe('IDs of users to fetch settings for'), + user_ids: z.array(SnowflakeType).max(RPC_USER_BATCH_MAX).describe('IDs of users to fetch settings for'), guild_id: SnowflakeType.describe('ID of the guild'), }), z.object({ @@ -118,7 +120,7 @@ export const RpcRequest = z.discriminatedUnion('type', [ }), z.object({ type: z.literal('get_user_blocked_ids').describe('Request type for fetching blocked user IDs'), - user_ids: z.array(SnowflakeType).describe('IDs of users to fetch blocked lists for'), + user_ids: z.array(SnowflakeType).max(RPC_USER_BATCH_MAX).describe('IDs of users to fetch blocked lists for'), }), z.object({ type: z.literal('voice_get_token').describe('Request type for getting voice connection token'),