diff --git a/CHANGELOG.md b/CHANGELOG.md index e3dfd4cd175..ce547cd0292 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,7 @@ - Fix: 画像の表示時にBlurhashが描画されない場合があるのを修正 ### Server +- Enhance: ノート作成時などに行われるロール計算処理のパフォーマンスを改善 - Fix: 既にミュートしているスレッドに対して再度スレッドミュートを作成しようとするとサーバーエラーになる問題を修正 diff --git a/packages/backend/src/GlobalModule.ts b/packages/backend/src/GlobalModule.ts index adccb4dc3ee..8c548c24ff6 100644 --- a/packages/backend/src/GlobalModule.ts +++ b/packages/backend/src/GlobalModule.ts @@ -14,6 +14,7 @@ import { createPostgresDataSource } from './postgres.js'; import { RepositoryModule } from './models/RepositoryModule.js'; import { allSettled } from './misc/promise-tracker.js'; import { GlobalEvents } from './core/GlobalEventService.js'; +import { OperationContextService } from './core/OperationContextService.js'; import type { Provider, OnApplicationShutdown } from '@nestjs/common'; const $config: Provider = { @@ -157,8 +158,8 @@ const $meta: Provider = { @Global() @Module({ imports: [RepositoryModule], - providers: [$config, $db, $meta, $meilisearch, $redis, $redisForPub, $redisForSub, $redisForTimelines, $redisForReactions], - exports: [$config, $db, $meta, $meilisearch, $redis, $redisForPub, $redisForSub, $redisForTimelines, $redisForReactions, RepositoryModule], + providers: [$config, $db, $meta, $meilisearch, $redis, $redisForPub, $redisForSub, $redisForTimelines, $redisForReactions, OperationContextService], + exports: [$config, $db, $meta, $meilisearch, $redis, $redisForPub, $redisForSub, $redisForTimelines, $redisForReactions, OperationContextService, RepositoryModule], }) export class GlobalModule implements OnApplicationShutdown { constructor( diff --git a/packages/backend/src/core/OperationContextService.ts b/packages/backend/src/core/OperationContextService.ts new file mode 100644 index 00000000000..5b2f6cf4017 --- /dev/null +++ b/packages/backend/src/core/OperationContextService.ts @@ -0,0 +1,159 @@ +/* + * SPDX-FileCopyrightText: syuilo and misskey-project + * SPDX-License-Identifier: AGPL-3.0-only + */ + +import { AsyncLocalStorage } from 'node:async_hooks'; +import { Injectable } from '@nestjs/common'; +import { bindThis } from '@/decorators.js'; + +export type OperationMemoToken = Readonly<{ + id: symbol; + getCacheKey: (key: K) => PropertyKey; + _valueType?: V; +}>; + +/** + * Operation内memoの名前空間とキーの作り方を定義する。 + * + * 同じtokenを使った呼び出しだけが値を共有するため、tokenは呼び出しごとではなく + * module scopeなどで一度だけ定義する。symbolにより、別の利用箇所との名前衝突を防ぐ。 + */ +export function defineOperationMemo( + name: string, + getCacheKey: (key: K) => PropertyKey, +): OperationMemoToken { + return { + id: Symbol(name), + getCacheKey, + }; +} + +/** + * 1つのOperationが所有するmemo。インスタンスはrunRoot()ごとに新しく作られる。 + */ +class OperationContext { + private readonly memo = new Map>>(); + + public memoize( + token: OperationMemoToken, + key: K, + loader: () => V | PromiseLike, + ): Promise { + let entries = this.memo.get(token.id); + if (entries == null) { + entries = new Map(); + this.memo.set(token.id, entries); + } + + const cacheKey = token.getCacheKey(key); + const cached = entries.get(cacheKey); + if (cached != null) return cached as Promise; + + const value = Promise.resolve().then(loader); + entries.set(cacheKey, value); + return value; + } + + public invalidate( + token: OperationMemoToken, + key: K, + ): void { + this.memo.get(token.id)?.delete(token.getCacheKey(key)); + } + + public invalidateAll( + token: OperationMemoToken, + ): void { + this.memo.delete(token.id); + } +} + +/** + * 1回の短命なAPIリクエストやQueueジョブ内だけで値を共有する。 + * Redisやプロセス内キャッシュとは異なり、Operationの境界を越えて値を共有しない。 + * + * Contextの分離: + * `runRoot()`は呼び出しごとに新しいOperationContextを作成し、AsyncLocalStorageを介して + * Node.jsが内部で管理するasync execution chainへ関連付ける。`getStore()`は現在実行中の + * chainに対応するContextを返すため、リクエストIDのようなキーを明示的に管理する必要はない。 + * 並行するrunRoot()はそれぞれ異なるContextを参照し、値を共有しない。 + * この分離は、各APIリクエストやQueueジョブの入口でrunRoot()を呼ぶことを前提とする。 + * + * Contextの伝播: + * runRoot()のcallback内で開始した通常のPromise/await、timer、I/O callbackには + * 同じContextが伝播する。runRoot()をネストした場合は新しいContextへ切り替わり、 + * 内側を抜けると親のContextへ戻る。 + * + * Operation境界: + * Contextが伝播するのは同一Node.jsプロセスのasync chain内だけであり、別プロセス、 + * Worker、Redis、Queueへserializeした処理には伝播しない。伝播しない処理では、 + * それぞれの入口で新しいrunRoot()を呼ぶ。 + * + * ライフサイクル: + * Contextを参照するasync resourceが解放されるとGC対象になる。callback内で開始した + * fire-and-forgetの処理は、ContextとmemoをOperation終了後も保持し得るため、 + * Operationに属する処理はすべてawaitし、短命な処理だけを対象とする。 + * + * 制約: + * これは認可やデータアクセスを隔離するsecurity boundaryではない。値の参照を明示的に + * 外へ渡せば共有できるため、Operation間で値を持ち出さないことは呼び出し側の責務となる。 + */ +@Injectable() +export class OperationContextService { + private readonly storage = new AsyncLocalStorage(); + + /** + * 新しいOperationを開始する。既にOperation内であっても、親とは別のContextを作成する。 + * callback内で作られたasync chainでは、明示的に引数を渡さず同じContextを参照できる。 + * + * memoにはTTLがないため、APIリクエストなどの短時間で完了する処理だけを対象とする。 + * バッチやインポートなど、長時間・大量のキーを扱う処理全体を囲んではならない。 + */ + @bindThis + public runRoot( + callback: () => T, + ): T { + return this.storage.run(new OperationContext(), callback); + } + + /** + * Operation内では同じtoken/keyのloaderを1回だけ実行する。 + * 実行中のPromise自体を保持するため、同時呼び出しも1つに束ねられる。 + * rejectionもinvalidateされるまで同じOperation内で保持される。 + * + * 境界が未導入の処理ではキャッシュせず、loaderを通常どおり実行する。 + */ + @bindThis + public memoizeIfActive( + token: OperationMemoToken, + key: K, + loader: () => V | PromiseLike, + ): Promise { + const context = this.storage.getStore(); + if (context == null) return Promise.resolve().then(loader); + + return context.memoize(token, key, loader); + } + + /** + * 同一Operation内の指定token/keyを破棄する。書き込み後のread-your-own-writeに使用する。 + */ + @bindThis + public invalidateIfActive( + token: OperationMemoToken, + key: K, + ): void { + this.storage.getStore()?.invalidate(token, key); + } + + /** + * 同一Operation内の指定tokenに紐づく全エントリを破棄する。 + */ + @bindThis + public invalidateAllIfActive( + token: OperationMemoToken, + ): void { + this.storage.getStore()?.invalidateAll(token); + } +} diff --git a/packages/backend/src/core/RoleService.ts b/packages/backend/src/core/RoleService.ts index 54884646394..74ecfc426c7 100644 --- a/packages/backend/src/core/RoleService.ts +++ b/packages/backend/src/core/RoleService.ts @@ -30,6 +30,7 @@ import { ModerationLogService } from '@/core/ModerationLogService.js'; import type { Packed } from '@/misc/json-schema.js'; import { FanoutTimelineService } from '@/core/FanoutTimelineService.js'; import { NotificationService } from '@/core/NotificationService.js'; +import { defineOperationMemo, OperationContextService } from '@/core/OperationContextService.js'; import type { OnApplicationShutdown, OnModuleInit } from '@nestjs/common'; // misskey-js の rolePolicies と同期すべし @@ -123,6 +124,28 @@ export const DEFAULT_POLICIES: RolePolicies = { watermarkAvailable: true, }; +const anonymousUserPoliciesKey = Symbol('RoleService.anonymousUserPolicies'); +const rolesMemo = defineOperationMemo( + 'RoleService.roles', + () => 'roles', +); +const userAssignsMemo = defineOperationMemo( + 'RoleService.userAssigns', + userId => userId, +); +const userRolesMemo = defineOperationMemo( + 'RoleService.userRoles', + userId => userId, +); +const userBadgeRolesMemo = defineOperationMemo( + 'RoleService.userBadgeRoles', + userId => userId, +); +const userPoliciesMemo = defineOperationMemo>( + 'RoleService.userPolicies', + userId => userId ?? anonymousUserPoliciesKey, +); + @Injectable() export class RoleService implements OnApplicationShutdown, OnModuleInit { private rolesCache: MemorySingleCache; @@ -156,6 +179,7 @@ export class RoleService implements OnApplicationShutdown, OnModuleInit { @Inject(DI.roleAssignmentsRepository) private roleAssignmentsRepository: RoleAssignmentsRepository, + private operationContextService: OperationContextService, private cacheService: CacheService, private userEntityService: UserEntityService, private globalEventService: GlobalEventService, @@ -173,6 +197,65 @@ export class RoleService implements OnApplicationShutdown, OnModuleInit { this.notificationService = this.moduleRef.get(NotificationService.name); } + @bindThis + private applyRoleChange(role: MiRole, mode: 'create' | 'update'): void { + const cached = this.rolesCache.get(); + if (cached == null) return; + + const index = cached.findIndex(candidate => candidate.id === role.id); + if (index === -1) { + if (mode === 'create') cached.push(role); + return; + } + + cached[index] = role; + for (let i = cached.length - 1; i > index; i--) { + if (cached[i].id === role.id) cached.splice(i, 1); + } + } + + @bindThis + private applyRoleDelete(roleId: MiRole['id']): void { + const cached = this.rolesCache.get(); + if (cached == null) return; + + for (let i = cached.length - 1; i >= 0; i--) { + if (cached[i].id === roleId) cached.splice(i, 1); + } + } + + @bindThis + private applyUserRoleAssignmentUpsert(assignment: MiRoleAssignment): void { + const cached = this.roleAssignmentByUserIdCache.get(assignment.userId); + if (cached == null) return; + + const normalizedAssignment = { + ...assignment, + user: assignment.user ?? null, + role: assignment.role ?? null, + }; + const index = cached.findIndex(candidate => candidate.id === assignment.id); + if (index === -1) { + cached.push(normalizedAssignment); + return; + } + + cached[index] = normalizedAssignment; + for (let i = cached.length - 1; i > index; i--) { + if (cached[i].id === assignment.id) cached.splice(i, 1); + } + } + + @bindThis + private applyUserRoleAssignmentDelete(userId: MiUser['id'], assignmentId: MiRoleAssignment['id']): void { + const cached = this.roleAssignmentByUserIdCache.get(userId); + if (cached == null) return; + + for (let i = cached.length - 1; i >= 0; i--) { + if (cached[i].id === assignmentId) cached.splice(i, 1); + } + } + @bindThis private async onMessage(_: string, data: string): Promise { const obj = JSON.parse(data); @@ -181,54 +264,36 @@ export class RoleService implements OnApplicationShutdown, OnModuleInit { const { type, body } = obj.message as GlobalEvents['internal']['payload']; switch (type) { case 'roleCreated': { - const cached = this.rolesCache.get(); - if (cached) { - cached.push({ - ...body, - updatedAt: new Date(body.updatedAt), - lastUsedAt: new Date(body.lastUsedAt), - }); - } + this.applyRoleChange({ + ...body, + updatedAt: new Date(body.updatedAt), + lastUsedAt: new Date(body.lastUsedAt), + }, 'create'); break; } case 'roleUpdated': { - const cached = this.rolesCache.get(); - if (cached) { - const i = cached.findIndex(x => x.id === body.id); - if (i > -1) { - cached[i] = { - ...body, - updatedAt: new Date(body.updatedAt), - lastUsedAt: new Date(body.lastUsedAt), - }; - } - } + this.applyRoleChange({ + ...body, + updatedAt: new Date(body.updatedAt), + lastUsedAt: new Date(body.lastUsedAt), + }, 'update'); break; } case 'roleDeleted': { - const cached = this.rolesCache.get(); - if (cached) { - this.rolesCache.set(cached.filter(x => x.id !== body.id)); - } + this.applyRoleDelete(body.id); break; } case 'userRoleAssigned': { - const cached = this.roleAssignmentByUserIdCache.get(body.userId); - if (cached) { - cached.push({ // TODO: このあたりのデシリアライズ処理は各modelファイル内に関数としてexportしたい - ...body, - expiresAt: body.expiresAt ? new Date(body.expiresAt) : null, - user: null, // joinなカラムは通常取ってこないので - role: null, // joinなカラムは通常取ってこないので - }); - } + this.applyUserRoleAssignmentUpsert({ // TODO: このあたりのデシリアライズ処理は各modelファイル内に関数としてexportしたい + ...body, + expiresAt: body.expiresAt ? new Date(body.expiresAt) : null, + user: null, // joinなカラムは通常取ってこないので + role: null, // joinなカラムは通常取ってこないので + }); break; } case 'userRoleUnassigned': { - const cached = this.roleAssignmentByUserIdCache.get(body.userId); - if (cached) { - this.roleAssignmentByUserIdCache.set(body.userId, cached.filter(x => x.id !== body.id)); - } + this.applyUserRoleAssignmentDelete(body.userId, body.id); break; } default: @@ -327,144 +392,204 @@ export class RoleService implements OnApplicationShutdown, OnModuleInit { } @bindThis - public async getRoles() { - const roles = await this.rolesCache.fetch(() => this.rolesRepository.findBy({})); - return roles; + private getRolesInternal(): Promise { + return this.operationContextService.memoizeIfActive(rolesMemo, undefined, async () => { + const roles = await this.rolesCache.fetch(() => this.rolesRepository.findBy({})); + return [...roles]; + }); } @bindThis - public async getUserAssigns(userId: MiUser['id']) { - const now = Date.now(); - let assigns = await this.roleAssignmentByUserIdCache.fetch(userId, () => this.roleAssignmentsRepository.findBy({ userId })); - // 期限切れのロールを除外 - assigns = assigns.filter(a => a.expiresAt == null || (a.expiresAt.getTime() > now)); - return assigns; + public async getRoles(): Promise { + return [...await this.getRolesInternal()]; + } + + @bindThis + private getUserAssignsInternal(userId: MiUser['id']): Promise { + return this.operationContextService.memoizeIfActive(userAssignsMemo, userId, async () => { + const now = Date.now(); + const assigns = await this.roleAssignmentByUserIdCache.fetch(userId, () => this.roleAssignmentsRepository.findBy({ userId })); + // 期限切れのロールを除外 + return assigns.filter(a => a.expiresAt == null || (a.expiresAt.getTime() > now)); + }); } @bindThis - public async getUserRoles(userId: MiUser['id']) { - const roles = await this.rolesCache.fetch(() => this.rolesRepository.findBy({})); - const assigns = await this.getUserAssigns(userId); - const assignedRoles = roles.filter(r => assigns.map(x => x.roleId).includes(r.id)); - const user = roles.some(r => r.target === 'conditional') ? await this.cacheService.findUserById(userId) : null; - const matchedCondRoles = roles.filter(r => r.target === 'conditional' && this.evalCond(user!, assignedRoles, r.condFormula)); - return [...assignedRoles, ...matchedCondRoles]; + public async getUserAssigns(userId: MiUser['id']): Promise { + return [...await this.getUserAssignsInternal(userId)]; + } + + @bindThis + private getUserRolesInternal(userId: MiUser['id']): Promise { + return this.operationContextService.memoizeIfActive(userRolesMemo, userId, async () => { + const roles = await this.getRolesInternal(); + const assigns = await this.getUserAssignsInternal(userId); + const assignedRoleIds = new Set(assigns.map(assignment => assignment.roleId)); + const assignedRoles = roles.filter(role => assignedRoleIds.has(role.id)); + const user = roles.some(role => role.target === 'conditional') ? await this.cacheService.findUserById(userId) : null; + const matchedCondRoles = roles.filter(role => role.target === 'conditional' && this.evalCond(user!, assignedRoles, role.condFormula)); + return [...assignedRoles, ...matchedCondRoles]; + }); + } + + @bindThis + public async getUserRoles(userId: MiUser['id']): Promise { + return [...await this.getUserRolesInternal(userId)]; } /** * 指定ユーザーのバッジロール一覧取得 */ @bindThis - public async getUserBadgeRoles(userId: MiUser['id']) { - const now = Date.now(); - let assigns = await this.roleAssignmentByUserIdCache.fetch(userId, () => this.roleAssignmentsRepository.findBy({ userId })); - // 期限切れのロールを除外 - assigns = assigns.filter(a => a.expiresAt == null || (a.expiresAt.getTime() > now)); - const roles = await this.rolesCache.fetch(() => this.rolesRepository.findBy({})); - const assignedRoles = roles.filter(r => assigns.map(x => x.roleId).includes(r.id)); - const assignedBadgeRoles = assignedRoles.filter(r => r.asBadge); - const badgeCondRoles = roles.filter(r => r.asBadge && (r.target === 'conditional')); - if (badgeCondRoles.length > 0) { - const user = roles.some(r => r.target === 'conditional') ? await this.cacheService.findUserById(userId) : null; - const matchedBadgeCondRoles = badgeCondRoles.filter(r => this.evalCond(user!, assignedRoles, r.condFormula)); - return [...assignedBadgeRoles, ...matchedBadgeCondRoles]; - } else { - return assignedBadgeRoles; - } + private getUserBadgeRolesInternal(userId: MiUser['id']): Promise { + return this.operationContextService.memoizeIfActive(userBadgeRolesMemo, userId, async () => { + const assigns = await this.getUserAssignsInternal(userId); + const roles = await this.getRolesInternal(); + const assignedRoleIds = new Set(assigns.map(assignment => assignment.roleId)); + const assignedRoles = roles.filter(role => assignedRoleIds.has(role.id)); + const assignedBadgeRoles = assignedRoles.filter(role => role.asBadge); + const badgeCondRoles = roles.filter(role => role.asBadge && role.target === 'conditional'); + if (badgeCondRoles.length > 0) { + const user = await this.cacheService.findUserById(userId); + const matchedBadgeCondRoles = badgeCondRoles.filter(role => this.evalCond(user, assignedRoles, role.condFormula)); + return [...assignedBadgeRoles, ...matchedBadgeCondRoles]; + } else { + return assignedBadgeRoles; + } + }); } @bindThis - public async getUserPolicies(userId: MiUser['id'] | null): Promise { - const basePolicies = { ...DEFAULT_POLICIES, ...this.meta.policies }; + public async getUserBadgeRoles(userId: MiUser['id']): Promise { + return [...await this.getUserBadgeRolesInternal(userId)]; + } - if (userId == null) return basePolicies; + @bindThis + private getUserPoliciesInternal(userId: MiUser['id'] | null): Promise> { + return this.operationContextService.memoizeIfActive(userPoliciesMemo, userId, async () => { + // meta.policiesの変更はOperation中のスナップショットには反映しない。 + // RoleService自身によるロール変更だけは、mutation側で明示的にこのmemoを破棄する。 + const mergedBasePolicies = { ...DEFAULT_POLICIES, ...this.meta.policies }; + const basePolicies = { + ...mergedBasePolicies, + uploadableFileTypes: [...mergedBasePolicies.uploadableFileTypes], + }; - const roles = await this.getUserRoles(userId); + if (userId == null) return basePolicies; - function calc(name: T, aggregate: (values: RolePolicies[T][]) => RolePolicies[T]) { - if (roles.length === 0) return aggregate([basePolicies[name]]); + const roles = await this.getUserRolesInternal(userId); - const policies = roles.map(role => role.policies[name] ?? { priority: 0, useDefault: true }); + function calc(name: T, aggregate: (values: RolePolicies[T][]) => RolePolicies[T]) { + if (roles.length === 0) return aggregate([basePolicies[name]]); - const p2 = policies.filter(policy => policy.priority === 2); - if (p2.length > 0) return aggregate(p2.map(policy => policy.useDefault ? basePolicies[name] : policy.value)); + const policies = roles.map(role => role.policies[name] ?? { priority: 0, useDefault: true }); - const p1 = policies.filter(policy => policy.priority === 1); - if (p1.length > 0) return aggregate(p1.map(policy => policy.useDefault ? basePolicies[name] : policy.value)); + const p2 = policies.filter(policy => policy.priority === 2); + if (p2.length > 0) return aggregate(p2.map(policy => policy.useDefault ? basePolicies[name] : policy.value)); - return aggregate(policies.map(policy => policy.useDefault ? basePolicies[name] : policy.value)); - } + const p1 = policies.filter(policy => policy.priority === 1); + if (p1.length > 0) return aggregate(p1.map(policy => policy.useDefault ? basePolicies[name] : policy.value)); - function aggregateChatAvailability(vs: RolePolicies['chatAvailability'][]) { - if (vs.some(v => v === 'available')) return 'available'; - if (vs.some(v => v === 'readonly')) return 'readonly'; - return 'unavailable'; - } + return aggregate(policies.map(policy => policy.useDefault ? basePolicies[name] : policy.value)); + } - const serverMaxFileSizeMb = Math.floor(this.config.maxFileSize / (1024 * 1024)); + function aggregateChatAvailability(vs: RolePolicies['chatAvailability'][]) { + if (vs.some(v => v === 'available')) return 'available'; + if (vs.some(v => v === 'readonly')) return 'readonly'; + return 'unavailable'; + } - return { - gtlAvailable: calc('gtlAvailable', vs => vs.some(v => v === true)), - ltlAvailable: calc('ltlAvailable', vs => vs.some(v => v === true)), - canPublicNote: calc('canPublicNote', vs => vs.some(v => v === true)), - mentionLimit: calc('mentionLimit', vs => Math.max(...vs)), - canInvite: calc('canInvite', vs => vs.some(v => v === true)), - inviteLimit: calc('inviteLimit', vs => Math.max(...vs)), - inviteLimitCycle: calc('inviteLimitCycle', vs => Math.max(...vs)), - inviteExpirationTime: calc('inviteExpirationTime', vs => Math.max(...vs)), - canManageCustomEmojis: calc('canManageCustomEmojis', vs => vs.some(v => v === true)), - canManageAvatarDecorations: calc('canManageAvatarDecorations', vs => vs.some(v => v === true)), - canSearchNotes: calc('canSearchNotes', vs => vs.some(v => v === true)), - canSearchUsers: calc('canSearchUsers', vs => vs.some(v => v === true)), - canUseTranslator: calc('canUseTranslator', vs => vs.some(v => v === true)), - canHideAds: calc('canHideAds', vs => vs.some(v => v === true)), - canCreateChannel: calc('canCreateChannel', vs => vs.some(v => v === true)), - driveCapacityMb: calc('driveCapacityMb', vs => Math.max(...vs)), - maxFileSizeMb: calc('maxFileSizeMb', vs => Math.min(serverMaxFileSizeMb, Math.max(...vs))), - alwaysMarkNsfw: calc('alwaysMarkNsfw', vs => vs.some(v => v === true)), - canUpdateBioMedia: calc('canUpdateBioMedia', vs => vs.some(v => v === true)), - pinLimit: calc('pinLimit', vs => Math.max(...vs)), - antennaLimit: calc('antennaLimit', vs => Math.max(...vs)), - wordMuteLimit: calc('wordMuteLimit', vs => Math.max(...vs)), - webhookLimit: calc('webhookLimit', vs => Math.max(...vs)), - clipLimit: calc('clipLimit', vs => Math.max(...vs)), - noteEachClipsLimit: calc('noteEachClipsLimit', vs => Math.max(...vs)), - userListLimit: calc('userListLimit', vs => Math.max(...vs)), - userEachUserListsLimit: calc('userEachUserListsLimit', vs => Math.max(...vs)), - rateLimitFactor: calc('rateLimitFactor', vs => Math.max(...vs)), - avatarDecorationLimit: calc('avatarDecorationLimit', vs => Math.max(...vs)), - canImportAntennas: calc('canImportAntennas', vs => vs.some(v => v === true)), - canImportBlocking: calc('canImportBlocking', vs => vs.some(v => v === true)), - canImportFollowing: calc('canImportFollowing', vs => vs.some(v => v === true)), - canImportMuting: calc('canImportMuting', vs => vs.some(v => v === true)), - canImportUserLists: calc('canImportUserLists', vs => vs.some(v => v === true)), - chatAvailability: calc('chatAvailability', aggregateChatAvailability), - uploadableFileTypes: calc('uploadableFileTypes', vs => { - const set = new Set(); - for (const v of vs) { - for (const type of v) { - if (type.trim() === '') continue; - set.add(type.trim()); + const serverMaxFileSizeMb = Math.floor(this.config.maxFileSize / (1024 * 1024)); + + return { + gtlAvailable: calc('gtlAvailable', vs => vs.some(v => v === true)), + ltlAvailable: calc('ltlAvailable', vs => vs.some(v => v === true)), + canPublicNote: calc('canPublicNote', vs => vs.some(v => v === true)), + mentionLimit: calc('mentionLimit', vs => Math.max(...vs)), + canInvite: calc('canInvite', vs => vs.some(v => v === true)), + inviteLimit: calc('inviteLimit', vs => Math.max(...vs)), + inviteLimitCycle: calc('inviteLimitCycle', vs => Math.max(...vs)), + inviteExpirationTime: calc('inviteExpirationTime', vs => Math.max(...vs)), + canManageCustomEmojis: calc('canManageCustomEmojis', vs => vs.some(v => v === true)), + canManageAvatarDecorations: calc('canManageAvatarDecorations', vs => vs.some(v => v === true)), + canSearchNotes: calc('canSearchNotes', vs => vs.some(v => v === true)), + canSearchUsers: calc('canSearchUsers', vs => vs.some(v => v === true)), + canUseTranslator: calc('canUseTranslator', vs => vs.some(v => v === true)), + canHideAds: calc('canHideAds', vs => vs.some(v => v === true)), + canCreateChannel: calc('canCreateChannel', vs => vs.some(v => v === true)), + driveCapacityMb: calc('driveCapacityMb', vs => Math.max(...vs)), + maxFileSizeMb: calc('maxFileSizeMb', vs => Math.min(serverMaxFileSizeMb, Math.max(...vs))), + alwaysMarkNsfw: calc('alwaysMarkNsfw', vs => vs.some(v => v === true)), + canUpdateBioMedia: calc('canUpdateBioMedia', vs => vs.some(v => v === true)), + pinLimit: calc('pinLimit', vs => Math.max(...vs)), + antennaLimit: calc('antennaLimit', vs => Math.max(...vs)), + wordMuteLimit: calc('wordMuteLimit', vs => Math.max(...vs)), + webhookLimit: calc('webhookLimit', vs => Math.max(...vs)), + clipLimit: calc('clipLimit', vs => Math.max(...vs)), + noteEachClipsLimit: calc('noteEachClipsLimit', vs => Math.max(...vs)), + userListLimit: calc('userListLimit', vs => Math.max(...vs)), + userEachUserListsLimit: calc('userEachUserListsLimit', vs => Math.max(...vs)), + rateLimitFactor: calc('rateLimitFactor', vs => Math.max(...vs)), + avatarDecorationLimit: calc('avatarDecorationLimit', vs => Math.max(...vs)), + canImportAntennas: calc('canImportAntennas', vs => vs.some(v => v === true)), + canImportBlocking: calc('canImportBlocking', vs => vs.some(v => v === true)), + canImportFollowing: calc('canImportFollowing', vs => vs.some(v => v === true)), + canImportMuting: calc('canImportMuting', vs => vs.some(v => v === true)), + canImportUserLists: calc('canImportUserLists', vs => vs.some(v => v === true)), + chatAvailability: calc('chatAvailability', aggregateChatAvailability), + uploadableFileTypes: calc('uploadableFileTypes', vs => { + const set = new Set(); + for (const v of vs) { + for (const type of v) { + if (type.trim() === '') continue; + set.add(type.trim()); + } } - } - return [...set]; - }), - noteDraftLimit: calc('noteDraftLimit', vs => Math.max(...vs)), - scheduledNoteLimit: calc('scheduledNoteLimit', vs => Math.max(...vs)), - watermarkAvailable: calc('watermarkAvailable', vs => vs.some(v => v === true)), + return [...set]; + }), + noteDraftLimit: calc('noteDraftLimit', vs => Math.max(...vs)), + scheduledNoteLimit: calc('scheduledNoteLimit', vs => Math.max(...vs)), + watermarkAvailable: calc('watermarkAvailable', vs => vs.some(v => v === true)), + }; + }); + } + + @bindThis + public async getUserPolicies(userId: MiUser['id'] | null): Promise { + const policies = await this.getUserPoliciesInternal(userId); + return { + ...policies, + uploadableFileTypes: [...policies.uploadableFileTypes], }; } + @bindThis + private invalidateUserRoleCalculations(userId: MiUser['id']): void { + this.operationContextService.invalidateIfActive(userAssignsMemo, userId); + this.operationContextService.invalidateIfActive(userRolesMemo, userId); + this.operationContextService.invalidateIfActive(userBadgeRolesMemo, userId); + this.operationContextService.invalidateIfActive(userPoliciesMemo, userId); + } + + @bindThis + private invalidateAllRoleCalculations(): void { + // ロール定義の変更はロールアサイン自体には影響しないため、userAssignsMemoは保持する。 + this.operationContextService.invalidateAllIfActive(rolesMemo); + this.operationContextService.invalidateAllIfActive(userRolesMemo); + this.operationContextService.invalidateAllIfActive(userBadgeRolesMemo); + this.operationContextService.invalidateAllIfActive(userPoliciesMemo); + } + @bindThis public async isModerator(user: { id: MiUser['id'] } | null): Promise { if (user == null) return false; - return (this.meta.rootUserId === user.id) || (await this.getUserRoles(user.id)).some(r => r.isModerator || r.isAdministrator); + return (this.meta.rootUserId === user.id) || (await this.getUserRolesInternal(user.id)).some(r => r.isModerator || r.isAdministrator); } @bindThis public async isAdministrator(user: { id: MiUser['id'] } | null): Promise { if (user == null) return false; - return (this.meta.rootUserId === user.id) || (await this.getUserRoles(user.id)).some(r => r.isAdministrator); + return (this.meta.rootUserId === user.id) || (await this.getUserRolesInternal(user.id)).some(r => r.isAdministrator); } @bindThis @@ -492,7 +617,7 @@ export class RoleService implements OnApplicationShutdown, OnModuleInit { const includeRoot = opts?.includeRoot ?? false; const excludeExpire = opts?.excludeExpire ?? false; - const roles = await this.rolesCache.fetch(() => this.rolesRepository.findBy({})); + const roles = await this.getRolesInternal(); const moderatorRoles = includeAdmins ? roles.filter(r => r.isModerator || r.isAdministrator) : roles.filter(r => r.isModerator); @@ -536,7 +661,7 @@ export class RoleService implements OnApplicationShutdown, OnModuleInit { @bindThis public async getAdministratorIds(): Promise { - const roles = await this.rolesCache.fetch(() => this.rolesRepository.findBy({})); + const roles = await this.getRolesInternal(); const administratorRoles = roles.filter(r => r.isAdministrator); const assigns = administratorRoles.length > 0 ? await this.roleAssignmentsRepository.findBy({ roleId: In(administratorRoles.map(r => r.id)), @@ -572,6 +697,7 @@ export class RoleService implements OnApplicationShutdown, OnModuleInit { roleId: roleId, userId: userId, }); + this.applyUserRoleAssignmentDelete(userId, existing.id); } else { throw new RoleService.AlreadyAssignedError(); } @@ -584,6 +710,9 @@ export class RoleService implements OnApplicationShutdown, OnModuleInit { userId: userId, }); + this.applyUserRoleAssignmentUpsert(created); + this.invalidateUserRoleCalculations(userId); + this.rolesRepository.update(roleId, { lastUsedAt: new Date(), }); @@ -622,10 +751,14 @@ export class RoleService implements OnApplicationShutdown, OnModuleInit { roleId: roleId, userId: userId, }); + this.applyUserRoleAssignmentDelete(userId, existing.id); + this.invalidateUserRoleCalculations(userId); throw new RoleService.NotAssignedError(); } await this.roleAssignmentsRepository.delete(existing.id); + this.applyUserRoleAssignmentDelete(userId, existing.id); + this.invalidateUserRoleCalculations(userId); this.rolesRepository.update(roleId, { lastUsedAt: now, @@ -650,7 +783,7 @@ export class RoleService implements OnApplicationShutdown, OnModuleInit { @bindThis public async addNoteToRoleTimeline(note: Packed<'Note'>): Promise { - const roles = await this.getUserRoles(note.userId); + const roles = await this.getUserRolesInternal(note.userId); const redisPipeline = this.redisForTimelines.pipeline(); @@ -686,6 +819,8 @@ export class RoleService implements OnApplicationShutdown, OnModuleInit { policies: values.policies, }); + this.applyRoleChange(created, 'create'); + this.invalidateAllRoleCalculations(); this.globalEventService.publishInternalEvent('roleCreated', created); if (moderator) { @@ -707,6 +842,8 @@ export class RoleService implements OnApplicationShutdown, OnModuleInit { }); const updated = await this.rolesRepository.findOneByOrFail({ id: role.id }); + this.applyRoleChange(updated, 'update'); + this.invalidateAllRoleCalculations(); this.globalEventService.publishInternalEvent('roleUpdated', updated); if (moderator) { @@ -721,6 +858,8 @@ export class RoleService implements OnApplicationShutdown, OnModuleInit { @bindThis public async delete(role: MiRole, moderator?: MiUser): Promise { await this.rolesRepository.delete({ id: role.id }); + this.applyRoleDelete(role.id); + this.invalidateAllRoleCalculations(); this.globalEventService.publishInternalEvent('roleDeleted', role); if (moderator) { diff --git a/packages/backend/src/queue/QueueProcessorService.ts b/packages/backend/src/queue/QueueProcessorService.ts index 925a8e3cbb0..a5ea712e207 100644 --- a/packages/backend/src/queue/QueueProcessorService.ts +++ b/packages/backend/src/queue/QueueProcessorService.ts @@ -9,6 +9,7 @@ import type { Config } from '@/config.js'; import { DI } from '@/di-symbols.js'; import type Logger from '@/logger.js'; import { bindThis } from '@/decorators.js'; +import { OperationContextService } from '@/core/OperationContextService.js'; import { TelemetryService } from '@/core/telemetry/TelemetryService.js'; import { CheckModeratorsActivityProcessorService } from '@/queue/processors/CheckModeratorsActivityProcessorService.js'; import { runQueueJob } from './queue-job-runner.js'; @@ -94,6 +95,7 @@ export class QueueProcessorService implements OnApplicationShutdown { private config: Config, private queueLoggerService: QueueLoggerService, + private operationContextService: OperationContextService, private telemetryService: TelemetryService, private userWebhookDeliverProcessorService: UserWebhookDeliverProcessorService, private systemWebhookDeliverProcessorService: SystemWebhookDeliverProcessorService, @@ -179,18 +181,19 @@ export class QueueProcessorService implements OnApplicationShutdown { const logger = this.logger.createSubLogger('system'); this.systemQueueWorker = new Bull.Worker(QUEUE.SYSTEM, (job) => { - return runQueueJob( - this.telemetryService, - 'Queue: System: ' + job.name, - () => processer(job) as Promise, - err => { + return runQueueJob({ + operationContext: { mode: 'none' }, + telemetryService: this.telemetryService, + spanName: 'Queue: System: ' + job.name, + processJob: () => processer(job) as Promise, + onError: err => { logger.error(`failed(${err.name}: ${err.message}) id=${job.id}`, { job: renderJob(job), e: renderError(err) }); this.telemetryService.captureMessage(`Queue: System: ${job.name}: ${err.name}: ${err.message}`, { level: 'error', extra: { job, err }, }); }, - ); + }); }, { ...baseWorkerOptions(this.config, QUEUE.SYSTEM), autorun: false, @@ -233,18 +236,19 @@ export class QueueProcessorService implements OnApplicationShutdown { const logger = this.logger.createSubLogger('db'); this.dbQueueWorker = new Bull.Worker(QUEUE.DB, (job) => { - return runQueueJob( - this.telemetryService, - 'Queue: DB: ' + job.name, - () => processer(job), - err => { + return runQueueJob({ + operationContext: { mode: 'none' }, + telemetryService: this.telemetryService, + spanName: 'Queue: DB: ' + job.name, + processJob: () => processer(job), + onError: err => { logger.error(`failed(${err.name}: ${err.message}) id=${job.id}`, { job: renderJob(job), e: renderError(err) }); this.telemetryService.captureMessage(`Queue: DB: ${job.name}: ${err.name}: ${err.message}`, { level: 'error', extra: { job, err }, }); }, - ); + }); }, { ...baseWorkerOptions(this.config, QUEUE.DB), autorun: false, @@ -263,18 +267,19 @@ export class QueueProcessorService implements OnApplicationShutdown { const logger = this.logger.createSubLogger('deliver'); this.deliverQueueWorker = new Bull.Worker(QUEUE.DELIVER, (job) => { - return runQueueJob( - this.telemetryService, - 'Queue: Deliver', - () => this.deliverProcessorService.process(job), - err => { + return runQueueJob({ + operationContext: { mode: 'none' }, + telemetryService: this.telemetryService, + spanName: 'Queue: Deliver', + processJob: () => this.deliverProcessorService.process(job), + onError: err => { logger.error(`failed(${err.name}: ${err.message}) ${getJobInfo(job)} to=${job.data.to}`, { e: renderError(err) }); this.telemetryService.captureMessage(`Queue: Deliver: ${err.name}: ${err.message}`, { level: 'error', extra: { job, err }, }); }, - ); + }); }, { ...baseWorkerOptions(this.config, QUEUE.DELIVER), autorun: false, @@ -301,11 +306,13 @@ export class QueueProcessorService implements OnApplicationShutdown { const logger = this.logger.createSubLogger('inbox'); this.inboxQueueWorker = new Bull.Worker(QUEUE.INBOX, (job) => { - return runQueueJob( - this.telemetryService, - 'Queue: Inbox', - () => this.inboxProcessorService.process(job), - err => { + return runQueueJob({ + // Inboxは1 Activity単位の短命処理で、NoteCreateのOperation境界として扱う。 + operationContext: { mode: 'per-job', service: this.operationContextService }, + telemetryService: this.telemetryService, + spanName: 'Queue: Inbox', + processJob: () => this.inboxProcessorService.process(job), + onError: err => { const activityId = job.data.activity ? job.data.activity.id : 'none'; logger.error(`failed(${err.name}: ${err.message}) ${getJobInfo(job)} activity=${activityId}`, { job: renderJob(job), e: renderError(err) }); this.telemetryService.captureMessage(`Queue: Inbox: ${err.name}: ${err.message}`, { @@ -313,7 +320,7 @@ export class QueueProcessorService implements OnApplicationShutdown { extra: { job, err }, }); }, - ); + }); }, { ...baseWorkerOptions(this.config, QUEUE.INBOX), autorun: false, @@ -340,18 +347,19 @@ export class QueueProcessorService implements OnApplicationShutdown { const logger = this.logger.createSubLogger('user-webhook'); this.userWebhookDeliverQueueWorker = new Bull.Worker(QUEUE.USER_WEBHOOK_DELIVER, (job) => { - return runQueueJob( - this.telemetryService, - 'Queue: UserWebhookDeliver', - () => this.userWebhookDeliverProcessorService.process(job), - err => { + return runQueueJob({ + operationContext: { mode: 'none' }, + telemetryService: this.telemetryService, + spanName: 'Queue: UserWebhookDeliver', + processJob: () => this.userWebhookDeliverProcessorService.process(job), + onError: err => { logger.error(`failed(${err.name}: ${err.message}) ${getJobInfo(job)} to=${job.data.to}`, { e: renderError(err) }); this.telemetryService.captureMessage(`Queue: UserWebhookDeliver: ${err.name}: ${err.message}`, { level: 'error', extra: { job, err }, }); }, - ); + }); }, { ...baseWorkerOptions(this.config, QUEUE.USER_WEBHOOK_DELIVER), autorun: false, @@ -378,18 +386,19 @@ export class QueueProcessorService implements OnApplicationShutdown { const logger = this.logger.createSubLogger('system-webhook'); this.systemWebhookDeliverQueueWorker = new Bull.Worker(QUEUE.SYSTEM_WEBHOOK_DELIVER, (job) => { - return runQueueJob( - this.telemetryService, - 'Queue: SystemWebhookDeliver', - () => this.systemWebhookDeliverProcessorService.process(job), - err => { + return runQueueJob({ + operationContext: { mode: 'none' }, + telemetryService: this.telemetryService, + spanName: 'Queue: SystemWebhookDeliver', + processJob: () => this.systemWebhookDeliverProcessorService.process(job), + onError: err => { logger.error(`failed(${err.name}: ${err.message}) ${getJobInfo(job)} to=${job.data.to}`, { e: renderError(err) }); this.telemetryService.captureMessage(`Queue: SystemWebhookDeliver: ${err.name}: ${err.message}`, { level: 'error', extra: { job, err }, }); }, - ); + }); }, { ...baseWorkerOptions(this.config, QUEUE.SYSTEM_WEBHOOK_DELIVER), autorun: false, @@ -425,18 +434,19 @@ export class QueueProcessorService implements OnApplicationShutdown { const logger = this.logger.createSubLogger('relationship'); this.relationshipQueueWorker = new Bull.Worker(QUEUE.RELATIONSHIP, (job) => { - return runQueueJob( - this.telemetryService, - 'Queue: Relationship: ' + job.name, - () => processer(job), - err => { + return runQueueJob({ + operationContext: { mode: 'none' }, + telemetryService: this.telemetryService, + spanName: 'Queue: Relationship: ' + job.name, + processJob: () => processer(job), + onError: err => { logger.error(`failed(${err.name}: ${err.message}) id=${job.id}`, { job: renderJob(job), e: renderError(err) }); this.telemetryService.captureMessage(`Queue: Relationship: ${job.name}: ${err.name}: ${err.message}`, { level: 'error', extra: { job, err }, }); }, - ); + }); }, { ...baseWorkerOptions(this.config, QUEUE.RELATIONSHIP), autorun: false, @@ -467,18 +477,19 @@ export class QueueProcessorService implements OnApplicationShutdown { const logger = this.logger.createSubLogger('objectStorage'); this.objectStorageQueueWorker = new Bull.Worker(QUEUE.OBJECT_STORAGE, (job) => { - return runQueueJob( - this.telemetryService, - 'Queue: ObjectStorage: ' + job.name, - () => processer(job) as Promise, - err => { + return runQueueJob({ + operationContext: { mode: 'none' }, + telemetryService: this.telemetryService, + spanName: 'Queue: ObjectStorage: ' + job.name, + processJob: () => processer(job) as Promise, + onError: err => { logger.error(`failed(${err.name}: ${err.message}) id=${job.id}`, { job: renderJob(job), e: renderError(err) }); this.telemetryService.captureMessage(`Queue: ObjectStorage: ${job.name}: ${err.name}: ${err.message}`, { level: 'error', extra: { job, err }, }); }, - ); + }); }, { ...baseWorkerOptions(this.config, QUEUE.OBJECT_STORAGE), autorun: false, @@ -498,18 +509,19 @@ export class QueueProcessorService implements OnApplicationShutdown { const logger = this.logger.createSubLogger('ended-poll-notification'); this.endedPollNotificationQueueWorker = new Bull.Worker(QUEUE.ENDED_POLL_NOTIFICATION, (job) => { - return runQueueJob( - this.telemetryService, - 'Queue: EndedPollNotification', - () => this.endedPollNotificationProcessorService.process(job), - err => { + return runQueueJob({ + operationContext: { mode: 'none' }, + telemetryService: this.telemetryService, + spanName: 'Queue: EndedPollNotification', + processJob: () => this.endedPollNotificationProcessorService.process(job), + onError: err => { logger.error(`failed(${err.name}: ${err.message}) id=${job.id}`, { job: renderJob(job), e: renderError(err) }); this.telemetryService.captureMessage(`Queue: EndedPollNotification: ${err.name}: ${err.message}`, { level: 'error', extra: { job, err }, }); }, - ); + }); }, { ...baseWorkerOptions(this.config, QUEUE.ENDED_POLL_NOTIFICATION), autorun: false, @@ -522,18 +534,20 @@ export class QueueProcessorService implements OnApplicationShutdown { const logger = this.logger.createSubLogger('post-scheduled-note'); this.postScheduledNoteQueueWorker = new Bull.Worker(QUEUE.POST_SCHEDULED_NOTE, (job) => { - return runQueueJob( - this.telemetryService, - 'Queue: PostScheduledNote', - () => this.postScheduledNoteProcessorService.process(job), - err => { + return runQueueJob({ + // 予約投稿は1 Note単位の短命処理で、NoteCreateのOperation境界として扱う。 + operationContext: { mode: 'per-job', service: this.operationContextService }, + telemetryService: this.telemetryService, + spanName: 'Queue: PostScheduledNote', + processJob: () => this.postScheduledNoteProcessorService.process(job), + onError: err => { logger.error(`failed(${err.name}: ${err.message}) id=${job.id}`, { job: renderJob(job), e: renderError(err) }); this.telemetryService.captureMessage(`Queue: PostScheduledNote: ${err.name}: ${err.message}`, { level: 'error', extra: { job, err }, }); }, - ); + }); }, { ...baseWorkerOptions(this.config, QUEUE.POST_SCHEDULED_NOTE), autorun: false, diff --git a/packages/backend/src/queue/queue-job-runner.ts b/packages/backend/src/queue/queue-job-runner.ts index 72c9957a8e3..95dd8c0a75b 100644 --- a/packages/backend/src/queue/queue-job-runner.ts +++ b/packages/backend/src/queue/queue-job-runner.ts @@ -3,29 +3,49 @@ * SPDX-License-Identifier: AGPL-3.0-only */ +import type { OperationContextService } from '@/core/OperationContextService.js'; import type { TelemetryService } from '@/core/telemetry/TelemetryService.js'; +type QueueOperationContextService = Pick; type QueueTelemetryService = Pick; +export type RunQueueJobOptions = { + operationContext: + | { mode: 'none' } + | { mode: 'per-job'; service: QueueOperationContextService }; + telemetryService: QueueTelemetryService; + spanName: string; + processJob: () => T | Promise; + onError: (error: Error) => void; +}; + /** Queueのprocessorを実行し、失敗処理をSpan内で行います。 */ -export function runQueueJob( - telemetryService: QueueTelemetryService, - spanName: string, - processJob: () => T | Promise, - onError: (error: Error) => void, -): Promise { - return telemetryService.startSpan(spanName, async (): Promise => { - try { - return await processJob(); - } catch (error) { - // 失敗イベントを待たず、processor Spanがactiveな間にログと通知を行います。 - const normalizedError = error instanceof Error ? error : new Error(String(error)); +export function runQueueJob(options: RunQueueJobOptions): Promise { + const { + operationContext, + telemetryService, + spanName, + processJob, + onError, + } = options; + const run = () => { + return telemetryService.startSpan(spanName, async (): Promise => { try { - onError(normalizedError); - } catch { - // 失敗ログの処理が例外を投げても、Queueへは元のエラーを返します。 + return await processJob(); + } catch (error) { + // 失敗イベントを待たず、processor Spanがactiveな間にログと通知を行います。 + const normalizedError = error instanceof Error ? error : new Error(String(error)); + try { + onError(normalizedError); + } catch { + // 失敗ログの処理が例外を投げても、Queueへは元のエラーを返します。 + } + throw error; } - throw error; - } - }); + }); + }; + + return operationContext.mode === 'none' + ? run() + : operationContext.service.runRoot(run); } diff --git a/packages/backend/src/server/api/ApiServerService.ts b/packages/backend/src/server/api/ApiServerService.ts index b88edaf1563..3a71b8cc1a3 100644 --- a/packages/backend/src/server/api/ApiServerService.ts +++ b/packages/backend/src/server/api/ApiServerService.ts @@ -7,10 +7,10 @@ import { Inject, Injectable } from '@nestjs/common'; import cors from '@fastify/cors'; import multipart from '@fastify/multipart'; import { ModuleRef } from '@nestjs/core'; -import type { AuthenticationResponseJSON } from '@simplewebauthn/server'; import type { Config } from '@/config.js'; import type { InstancesRepository, AccessTokensRepository } from '@/models/_.js'; import { DI } from '@/di-symbols.js'; +import { OperationContextService } from '@/core/OperationContextService.js'; import { UserEntityService } from '@/core/entities/UserEntityService.js'; import { bindThis } from '@/decorators.js'; import endpoints from './endpoints.js'; @@ -18,6 +18,7 @@ import { ApiCallService } from './ApiCallService.js'; import { SignupApiService } from './SignupApiService.js'; import { SigninApiService } from './SigninApiService.js'; import { SigninWithPasskeyApiService } from './SigninWithPasskeyApiService.js'; +import type { AuthenticationResponseJSON } from '@simplewebauthn/server'; import type { FastifyInstance, FastifyPluginOptions } from 'fastify'; @Injectable() @@ -35,6 +36,7 @@ export class ApiServerService { private accessTokensRepository: AccessTokensRepository, private userEntityService: UserEntityService, + private operationContextService: OperationContextService, private apiCallService: ApiCallService, private signupApiService: SignupApiService, private signinApiService: SigninApiService, @@ -62,6 +64,7 @@ export class ApiServerService { done(); }); + // OperationContextの初期導入範囲は、ロール計算を行う通常API endpointに限定する。 for (const endpoint of endpoints) { const ep = { name: endpoint.name, @@ -83,7 +86,9 @@ export class ApiServerService { } // Await so that any error can automatically be translated to HTTP 500 - await this.apiCallService.handleMultipartRequest(ep, request, reply); + await this.operationContextService.runRoot( + () => this.apiCallService.handleMultipartRequest(ep, request, reply), + ); return reply; }); } else { @@ -99,7 +104,9 @@ export class ApiServerService { } // Await so that any error can automatically be translated to HTTP 500 - await this.apiCallService.handleRequest(ep, request, reply); + await this.operationContextService.runRoot( + () => this.apiCallService.handleRequest(ep, request, reply), + ); return reply; }); } diff --git a/packages/backend/test/unit/OperationContextService.ts b/packages/backend/test/unit/OperationContextService.ts new file mode 100644 index 00000000000..0581efe867d --- /dev/null +++ b/packages/backend/test/unit/OperationContextService.ts @@ -0,0 +1,122 @@ +/* + * SPDX-FileCopyrightText: syuilo and misskey-project + * SPDX-License-Identifier: AGPL-3.0-only + */ + +import { setImmediate } from 'node:timers/promises'; +import { describe, expect, test, vi } from 'vitest'; +import { defineOperationMemo, OperationContextService } from '@/core/OperationContextService.js'; + +describe('OperationContextService', () => { + test('coalesces concurrent loads for the same token and key', async () => { + const service = new OperationContextService(); + const token = defineOperationMemo('test', key => key); + const loader = vi.fn(async () => { + await setImmediate(); + return {}; + }); + + const [first, second] = await service.runRoot(() => Promise.all([ + service.memoizeIfActive(token, 'key', loader), + service.memoizeIfActive(token, 'key', loader), + ])); + + expect(first).toBe(second); + expect(loader).toHaveBeenCalledOnce(); + }); + + test('does not share values between root operations', async () => { + const service = new OperationContextService(); + const token = defineOperationMemo('test', key => key); + const loader = vi.fn(async () => { + await setImmediate(); + return {}; + }); + + const [first, second] = await Promise.all([ + service.runRoot(() => service.memoizeIfActive(token, 'key', loader)), + service.runRoot(() => service.memoizeIfActive(token, 'key', loader)), + ]); + + expect(first).not.toBe(second); + expect(loader).toHaveBeenCalledTimes(2); + }); + + test('nested root operations restore the parent context', async () => { + const service = new OperationContextService(); + const token = defineOperationMemo('test', key => key); + const loader = vi.fn(() => ({})); + + await service.runRoot(async () => { + const outer = await service.memoizeIfActive(token, 'key', loader); + const inner = await service.runRoot(() => service.memoizeIfActive(token, 'key', loader)); + + expect(inner).not.toBe(outer); + await expect(service.memoizeIfActive(token, 'key', loader)).resolves.toBe(outer); + }); + + expect(loader).toHaveBeenCalledTimes(2); + }); + + test('falls back to loading every time outside an operation context', async () => { + const service = new OperationContextService(); + const token = defineOperationMemo('test', key => key); + const loader = vi.fn(() => ({})); + + const first = await service.memoizeIfActive(token, 'key', loader); + const second = await service.memoizeIfActive(token, 'key', loader); + + expect(first).not.toBe(second); + expect(loader).toHaveBeenCalledTimes(2); + }); + + test('keeps rejected loads stable for the duration of the operation', async () => { + const service = new OperationContextService(); + const token = defineOperationMemo('test', key => key); + const error = new Error('failed'); + const loader = vi.fn(() => Promise.reject(error)); + + await service.runRoot(async () => { + await expect(service.memoizeIfActive(token, 'key', loader)).rejects.toBe(error); + await expect(service.memoizeIfActive(token, 'key', loader)).rejects.toBe(error); + }); + + expect(loader).toHaveBeenCalledOnce(); + }); + + test('can invalidate one memoized key after a write', async () => { + const service = new OperationContextService(); + const token = defineOperationMemo('test', key => key); + let value = 1; + const loader = vi.fn(() => value); + + await service.runRoot(async () => { + await expect(service.memoizeIfActive(token, 'a', loader)).resolves.toBe(1); + await expect(service.memoizeIfActive(token, 'b', loader)).resolves.toBe(1); + + value = 2; + service.invalidateIfActive(token, 'a'); + + await expect(service.memoizeIfActive(token, 'a', loader)).resolves.toBe(2); + await expect(service.memoizeIfActive(token, 'b', loader)).resolves.toBe(1); + }); + }); + + test('can invalidate every key for a memo token after a write', async () => { + const service = new OperationContextService(); + const token = defineOperationMemo('test', key => key); + let value = 1; + const loader = vi.fn(() => value); + + await service.runRoot(async () => { + await service.memoizeIfActive(token, 'a', loader); + await service.memoizeIfActive(token, 'b', loader); + + value = 2; + service.invalidateAllIfActive(token); + + await expect(service.memoizeIfActive(token, 'a', loader)).resolves.toBe(2); + await expect(service.memoizeIfActive(token, 'b', loader)).resolves.toBe(2); + }); + }); +}); diff --git a/packages/backend/test/unit/RoleService.ts b/packages/backend/test/unit/RoleService.ts index ec1e7ca1348..600d1c99063 100644 --- a/packages/backend/test/unit/RoleService.ts +++ b/packages/backend/test/unit/RoleService.ts @@ -7,11 +7,9 @@ process.env.NODE_ENV = 'test'; import { setTimeout } from 'node:timers/promises'; import { describe, beforeEach, afterEach, test, expect, vi } from 'vitest'; -import type { Mocked } from 'vitest'; import { mockDeep } from 'vitest-mock-extended'; import { Test } from '@nestjs/testing'; import * as lolex from '@sinonjs/fake-timers'; -import type { TestingModule } from '@nestjs/testing'; import { GlobalModule } from '@/GlobalModule.js'; import { RoleService } from '@/core/RoleService.js'; import { @@ -29,10 +27,13 @@ import { genAidx } from '@/misc/id/aidx.js'; import { CacheService } from '@/core/CacheService.js'; import { IdService } from '@/core/IdService.js'; import { GlobalEventService } from '@/core/GlobalEventService.js'; +import { OperationContextService } from '@/core/OperationContextService.js'; import { secureRndstr } from '@/misc/secure-rndstr.js'; import { NotificationService } from '@/core/NotificationService.js'; import { RoleCondFormulaValue } from '@/models/Role.js'; import { UserEntityService } from '@/core/entities/UserEntityService.js'; +import type { TestingModule } from '@nestjs/testing'; +import type { Mocked } from 'vitest'; describe('RoleService', () => { let app: TestingModule; @@ -40,6 +41,7 @@ describe('RoleService', () => { let usersRepository: UsersRepository; let rolesRepository: RolesRepository; let roleAssignmentsRepository: RoleAssignmentsRepository; + let operationContextService: OperationContextService; let meta: Mocked; let notificationService: Mocked; let clock: lolex.Clock; @@ -146,6 +148,7 @@ describe('RoleService', () => { usersRepository = app.get(DI.usersRepository); rolesRepository = app.get(DI.rolesRepository); roleAssignmentsRepository = app.get(DI.roleAssignmentsRepository); + operationContextService = app.get(OperationContextService); meta = app.get(DI.meta) as Mocked; notificationService = app.get(NotificationService) as Mocked; @@ -206,6 +209,68 @@ describe('RoleService', () => { expect(assigns.some(a => a.roleId === roleNotExpired.id)).toBe(true); expect(assigns.some(a => a.roleId === roleExpired.id)).toBe(false); }); + + test('userRoleAssignedイベントを重複適用してもアサインが重複しない', async () => { + const user = await createUser(); + const role = await createRole({ name: 'role' }); + const assignment = await assignRole({ userId: user.id, roleId: role.id }); + await roleService.getUserAssigns(user.id); + + await roleService['onMessage']('', JSON.stringify({ + channel: 'internal', + message: { + type: 'userRoleAssigned', + body: assignment, + }, + })); + + const assigns = await roleService.getUserAssigns(user.id); + expect(assigns.filter(candidate => candidate.id === assignment.id)).toHaveLength(1); + }); + }); + + describe('getRoles', () => { + test('roleCreatedイベントを重複適用してもロールが重複しない', async () => { + const role = await createRole({ name: 'role' }); + await roleService.getRoles(); + + await roleService['onMessage']('', JSON.stringify({ + channel: 'internal', + message: { + type: 'roleCreated', + body: role, + }, + })); + + const roles = await roleService.getRoles(); + expect(roles.filter(candidate => candidate.id === role.id)).toHaveLength(1); + }); + + test('削除後に遅れて届いたroleUpdatedイベントでロールが復活しない', async () => { + const role = await createRole({ name: 'role' }); + await roleService.getRoles(); + + await roleService['onMessage']('', JSON.stringify({ + channel: 'internal', + message: { + type: 'roleDeleted', + body: role, + }, + })); + await roleService['onMessage']('', JSON.stringify({ + channel: 'internal', + message: { + type: 'roleUpdated', + body: { + ...role, + name: 'updated role', + }, + }, + })); + + const roles = await roleService.getRoles(); + expect(roles.some(candidate => candidate.id === role.id)).toBe(false); + }); }); describe('getUserRoles', () => { @@ -223,6 +288,37 @@ describe('RoleService', () => { expect(roles.some(r => r.id === manualRole.id)).toBe(true); expect(roles.some(r => r.id === conditionalRole.id)).toBe(true); }); + + test('Operation内で共有した配列を呼び出し元から変更できない', async () => { + const user = await createUser(); + const role = await createRole({ name: 'manual role' }); + await roleService.assign(user.id, role.id); + + await operationContextService.runRoot(async () => { + const first = await roleService.getUserRoles(user.id); + first.splice(0); + + const second = await roleService.getUserRoles(user.id); + expect(second.some(candidate => candidate.id === role.id)).toBe(true); + }); + }); + + test('Operation内のassign/unassign後は再計算する', async () => { + const user = await createUser(); + const role = await createRole({ name: 'manual role' }); + + await operationContextService.runRoot(async () => { + await expect(roleService.getUserRoles(user.id)).resolves.toEqual([]); + + await roleService.assign(user.id, role.id); + const assigned = await roleService.getUserRoles(user.id); + expect(assigned.some(candidate => candidate.id === role.id)).toBe(true); + + await roleService.unassign(user.id, role.id); + const unassigned = await roleService.getUserRoles(user.id); + expect(unassigned.some(candidate => candidate.id === role.id)).toBe(false); + }); + }); }); describe('getUserPolicies', () => { @@ -365,6 +461,69 @@ describe('RoleService', () => { // roleWithoutPolicy は default 値 (5) を使い、roleWithPolicy の 10 と比較して大きい方が採用される expect(result.pinLimit).toBe(10); }); + + test('Operation内では同じポリシースナップショットを返す', async () => { + const user = await createUser(); + meta.policies = { + pinLimit: 5, + uploadableFileTypes: ['image/*'], + }; + + await operationContextService.runRoot(async () => { + const first = await roleService.getUserPolicies(user.id); + meta.policies = { + pinLimit: 10, + uploadableFileTypes: ['video/*'], + }; + first.pinLimit = 100; + first.uploadableFileTypes.push('application/json'); + + const second = await roleService.getUserPolicies(user.id); + expect(second.pinLimit).toBe(5); + expect(second.uploadableFileTypes).toEqual(['image/*']); + }); + + const nextOperation = await operationContextService.runRoot( + () => roleService.getUserPolicies(user.id), + ); + expect(nextOperation.pinLimit).toBe(10); + expect(nextOperation.uploadableFileTypes).toEqual(['video/*']); + }); + + test('Operation内のrole更新後はポリシーを再計算する', async () => { + const user = await createUser(); + const role = await createRole({ + name: 'role', + policies: { + pinLimit: { + useDefault: false, + priority: 0, + value: 10, + }, + }, + }); + await roleService.assign(user.id, role.id); + + await operationContextService.runRoot(async () => { + await expect(roleService.getUserPolicies(user.id)).resolves.toMatchObject({ + pinLimit: 10, + }); + + await roleService.update(role, { + policies: { + pinLimit: { + useDefault: false, + priority: 0, + value: 20, + }, + }, + }); + + await expect(roleService.getUserPolicies(user.id)).resolves.toMatchObject({ + pinLimit: 20, + }); + }); + }); }); describe('getUserBadgeRoles', () => { diff --git a/packages/backend/test/unit/queue/queue-job-runner.ts b/packages/backend/test/unit/queue/queue-job-runner.ts index 72c4a77ef28..cd202f9f59d 100644 --- a/packages/backend/test/unit/queue/queue-job-runner.ts +++ b/packages/backend/test/unit/queue/queue-job-runner.ts @@ -4,10 +4,13 @@ */ import { describe, expect, test, vi } from 'vitest'; +import { defineOperationMemo, OperationContextService } from '@/core/OperationContextService.js'; import { runQueueJob } from '@/queue/queue-job-runner.js'; import { TelemetryService } from '@/core/telemetry/TelemetryService.js'; describe('runQueueJob', () => { + const operationContextService = new OperationContextService(); + test('returns the processor result without invoking the error handler', async () => { let spanActive = false; const startSpan = vi.fn((_name: string, fn: () => T): T => { @@ -22,7 +25,13 @@ describe('runQueueJob', () => { } as unknown as TelemetryService; const onError = vi.fn(); - await expect(runQueueJob(telemetryService, 'Queue: test', () => 'ok', onError)).resolves.toBe('ok'); + await expect(runQueueJob({ + operationContext: { mode: 'none' }, + telemetryService, + spanName: 'Queue: test', + processJob: () => 'ok', + onError, + })).resolves.toBe('ok'); expect(onError).not.toHaveBeenCalled(); expect(spanActive).toBe(false); @@ -46,11 +55,72 @@ describe('runQueueJob', () => { }); const originalError = new Error('failed'); - await expect(runQueueJob(telemetryService, 'Queue: test', async () => { - throw originalError; - }, onError)).rejects.toBe(originalError); + await expect(runQueueJob({ + operationContext: { mode: 'none' }, + telemetryService, + spanName: 'Queue: test', + processJob: async () => { + throw originalError; + }, + onError, + })).rejects.toBe(originalError); expect(onError).toHaveBeenCalledOnce(); expect(spanActive).toBe(false); }); + + test('creates an isolated operation context for each job', async () => { + const token = defineOperationMemo('queue test', key => key); + const loader = vi.fn(() => ({})); + const telemetryService = { + startSpan: (_name: string, fn: () => T): T => fn(), + } as unknown as TelemetryService; + + const first = await runQueueJob({ + operationContext: { mode: 'per-job', service: operationContextService }, + telemetryService, + spanName: 'Queue: test', + processJob: async () => { + const [a, b] = await Promise.all([ + operationContextService.memoizeIfActive(token, 'key', loader), + operationContextService.memoizeIfActive(token, 'key', loader), + ]); + expect(a).toBe(b); + return a; + }, + onError: vi.fn(), + }); + const second = await runQueueJob({ + operationContext: { mode: 'per-job', service: operationContextService }, + telemetryService, + spanName: 'Queue: test', + processJob: () => operationContextService.memoizeIfActive(token, 'key', loader), + onError: vi.fn(), + }); + + expect(first).not.toBe(second); + expect(loader).toHaveBeenCalledTimes(2); + }); + + test('does not create an operation context unless explicitly requested', async () => { + const token = defineOperationMemo('queue test', key => key); + const loader = vi.fn(() => ({})); + const telemetryService = { + startSpan: (_name: string, fn: () => T): T => fn(), + } as unknown as TelemetryService; + + await runQueueJob({ + operationContext: { mode: 'none' }, + telemetryService, + spanName: 'Queue: long running test', + processJob: async () => { + const first = await operationContextService.memoizeIfActive(token, 'key', loader); + const second = await operationContextService.memoizeIfActive(token, 'key', loader); + expect(first).not.toBe(second); + }, + onError: vi.fn(), + }); + + expect(loader).toHaveBeenCalledTimes(2); + }); }); diff --git a/packages/backend/test/unit/server/ApiServerService.ts b/packages/backend/test/unit/server/ApiServerService.ts new file mode 100644 index 00000000000..ccf6fc2a7b1 --- /dev/null +++ b/packages/backend/test/unit/server/ApiServerService.ts @@ -0,0 +1,108 @@ +/* + * SPDX-FileCopyrightText: syuilo and misskey-project + * SPDX-License-Identifier: AGPL-3.0-only + */ + +import { setImmediate } from 'node:timers/promises'; +import Fastify, { type FastifyInstance, type FastifyReply } from 'fastify'; +import { afterEach, describe, expect, test, vi } from 'vitest'; +import { defineOperationMemo, OperationContextService } from '@/core/OperationContextService.js'; +import { ApiServerService } from '@/server/api/ApiServerService.js'; + +const endpoints = vi.hoisted(() => [ + { + name: 'test/regular', + meta: {}, + params: {}, + }, + { + name: 'test/multipart', + meta: { requireFile: true }, + params: {}, + }, +]); + +vi.mock('@/server/api/endpoints.js', () => ({ + default: endpoints, +})); + +type HandlerName = 'handleRequest' | 'handleMultipartRequest'; + +async function createServer() { + const operationContextService = new OperationContextService(); + const runRoot = vi.spyOn(operationContextService, 'runRoot'); + const apiCallService = { + handleRequest: vi.fn(), + handleMultipartRequest: vi.fn(), + }; + const service = new ApiServerService( + { get: vi.fn(() => ({ exec: vi.fn() })) } as never, + { maxFileSize: 1024 * 1024 } as never, + { find: vi.fn() } as never, + { findOneBy: vi.fn(), update: vi.fn() } as never, + { pack: vi.fn() } as never, + operationContextService, + apiCallService as never, + { signup: vi.fn(), signupPending: vi.fn() } as never, + { signin: vi.fn() } as never, + { signin: vi.fn() } as never, + ); + const fastify = Fastify(); + service.createServer(fastify, {}, error => { + if (error != null) throw error; + }); + await fastify.ready(); + + return { apiCallService, fastify, operationContextService, runRoot }; +} + +describe('ApiServerService operation context boundary', () => { + const servers: FastifyInstance[] = []; + + afterEach(async () => { + await Promise.all(servers.splice(0).map(server => server.close())); + }); + + test.each([ + ['regular', '/test/regular', 'handleRequest'], + ['multipart', '/test/multipart', 'handleMultipartRequest'], + ] as const)('%s requests share one context and isolate separate requests', async (_label, url, handlerName: HandlerName) => { + const { apiCallService, fastify, operationContextService, runRoot } = await createServer(); + servers.push(fastify); + const token = defineOperationMemo('api server test', () => 'value'); + const loader = vi.fn(() => ({})); + const values: object[][] = []; + apiCallService[handlerName].mockImplementation(async (...args: unknown[]) => { + const first = await operationContextService.memoizeIfActive(token, undefined, loader); + await setImmediate(); + const second = await operationContextService.memoizeIfActive(token, undefined, loader); + values.push([first, second]); + (args[2] as FastifyReply).send({ ok: true }); + }); + + const firstResponse = await fastify.inject({ method: 'POST', url }); + const secondResponse = await fastify.inject({ method: 'POST', url }); + + expect(firstResponse.statusCode).toBe(200); + expect(secondResponse.statusCode).toBe(200); + expect(runRoot).toHaveBeenCalledTimes(2); + expect(values[0][0]).toBe(values[0][1]); + expect(values[1][0]).toBe(values[1][1]); + expect(values[0][0]).not.toBe(values[1][0]); + expect(loader).toHaveBeenCalledTimes(2); + }); + + test.each([ + ['regular', '/test/regular', 'handleRequest'], + ['multipart', '/test/multipart', 'handleMultipartRequest'], + ] as const)('%s request errors propagate to Fastify', async (_label, url, handlerName: HandlerName) => { + const { apiCallService, fastify, runRoot } = await createServer(); + servers.push(fastify); + apiCallService[handlerName].mockRejectedValue(new Error('failed')); + + const response = await fastify.inject({ method: 'POST', url }); + + expect(response.statusCode).toBe(500); + expect(runRoot).toHaveBeenCalledOnce(); + }); +});