mirror of
https://github.com/fluxerapp/fluxer
synced 2026-10-17 13:05:32 +09:00
perf(rpc): bound concurrency in user batch fanout handlers (#2124)
This commit is contained in:
Hampus authored and GitHub committed
1 parent
8aa39bc7db
commit
7d826d1602
6 files changed
+148
-12
No files matched your search
@@ -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<UserGuildSettings | null> = [];
|
||||
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<UserID>}): Promise<Record<string, Array<string>>> {
|
||||
const {userIds} = params;
|
||||
const settled = await allSettledWithConcurrency(userIds, RPC_RESPONSE_MAP_CONCURRENCY, (userId) =>
|
||||
this.userRepository.listBlockedUserIds(userId),
|
||||
);
|
||||
const result: Record<string, Array<string>> = {};
|
||||
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;
|
||||
}
|
||||
|
||||
@@ -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<string, Array<string>>;
|
||||
}
|
||||
|
||||
const UNKNOWN_USER_ID = '1';
|
||||
const BATCH_SIZE = 96;
|
||||
|
||||
async function setDmNotificationLevel(harness: ApiTestHarness, token: string, level: number): Promise<void> {
|
||||
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<RpcUserGuildSettingsResponse>(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<RpcUserBlockedIdsResponse>(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([]);
|
||||
});
|
||||
});
|
||||
@@ -7,6 +7,7 @@ import type {UserNote} from '../../models/UserNote';
|
||||
|
||||
export interface IUserRelationshipRepository {
|
||||
listRelationships(sourceUserId: UserID): Promise<Array<Relationship>>;
|
||||
listBlockedUserIds(sourceUserId: UserID): Promise<Array<UserID>>;
|
||||
listIncomingRequests(userId: UserID): Promise<Array<Relationship>>;
|
||||
hasReachedRelationshipLimit(sourceUserId: UserID, limit: number): Promise<boolean>;
|
||||
getRelationship(sourceUserId: UserID, targetUserId: UserID, type: number): Promise<Relationship | null>;
|
||||
|
||||
@@ -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<Array<UserID>> {
|
||||
const rows = await fetchMany<Pick<RelationshipRow, 'target_user_id' | 'type'>>(
|
||||
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<Array<Relationship>> {
|
||||
const relationships = await this.listRelationships(userId);
|
||||
return relationships.filter((rel) => rel.type === RelationshipTypes.INCOMING_REQUEST);
|
||||
|
||||
@@ -430,6 +430,10 @@ export class UserRepository implements IUserRepositoryAggregate {
|
||||
return this.relationshipRepo.listRelationships(sourceUserId);
|
||||
}
|
||||
|
||||
async listBlockedUserIds(sourceUserId: UserID): Promise<Array<UserID>> {
|
||||
return this.relationshipRepo.listBlockedUserIds(sourceUserId);
|
||||
}
|
||||
|
||||
async hasReachedRelationshipLimit(sourceUserId: UserID, limit: number): Promise<boolean> {
|
||||
return this.relationshipRepo.hasReachedRelationshipLimit(sourceUserId, limit);
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user