diff --git a/packages/backend/src/GlobalModule.ts b/packages/backend/src/GlobalModule.ts index 3e9d19f82598d71c9e161aff176d76e062089379..c83845b94c534ae87bce5cb1b0ec20038c15421e 100644 --- a/packages/backend/src/GlobalModule.ts +++ b/packages/backend/src/GlobalModule.ts @@ -3,7 +3,6 @@ * SPDX-License-Identifier: AGPL-3.0-only */ -import { setTimeout } from 'node:timers/promises'; import { Global, Inject, Module } from '@nestjs/common'; import * as Redis from 'ioredis'; import { DataSource } from 'typeorm'; @@ -12,6 +11,7 @@ import { DI } from './di-symbols.js'; import { Config, loadConfig } from './config.js'; import { createPostgresDataSource } from './postgres.js'; import { RepositoryModule } from './models/RepositoryModule.js'; +import { allSettled } from './misc/promise-tracker.js'; import type { Provider, OnApplicationShutdown } from '@nestjs/common'; const $config: Provider = { @@ -33,7 +33,7 @@ const $meilisearch: Provider = { useFactory: (config: Config) => { if (config.meilisearch) { return new MeiliSearch({ - host: `${config.meilisearch.ssl ? 'https' : 'http' }://${config.meilisearch.host}:${config.meilisearch.port}`, + host: `${config.meilisearch.ssl ? 'https' : 'http'}://${config.meilisearch.host}:${config.meilisearch.port}`, apiKey: config.meilisearch.apiKey, }); } else { @@ -91,17 +91,12 @@ export class GlobalModule implements OnApplicationShutdown { @Inject(DI.redisForPub) private redisForPub: Redis.Redis, @Inject(DI.redisForSub) private redisForSub: Redis.Redis, @Inject(DI.redisForTimelines) private redisForTimelines: Redis.Redis, - ) {} + ) { } public async dispose(): Promise<void> { - if (process.env.NODE_ENV === 'test') { - // XXX: - // Shutting down the existing connections causes errors on Jest as - // Misskey has asynchronous postgres/redis connections that are not - // awaited. - // Let's wait for some random time for them to finish. - await setTimeout(5000); - } + // Wait for all potential DB queries + await allSettled(); + // And then disconnect from DB await Promise.all([ this.db.destroy(), this.redisClient.disconnect(), diff --git a/packages/backend/src/core/NoteCreateService.ts b/packages/backend/src/core/NoteCreateService.ts index ed8d51df16cbbe4917ad39cd29d16228d214902b..97fb80ab394f9c1d39efbe6c37d8b490d2201b01 100644 --- a/packages/backend/src/core/NoteCreateService.ts +++ b/packages/backend/src/core/NoteCreateService.ts @@ -58,6 +58,7 @@ import { FanoutTimelineService } from '@/core/FanoutTimelineService.js'; import { UtilityService } from '@/core/UtilityService.js'; import { UserBlockingService } from '@/core/UserBlockingService.js'; import { isReply } from '@/misc/is-reply.js'; +import { trackPromise } from '@/misc/promise-tracker.js'; type NotificationType = 'reply' | 'renote' | 'quote' | 'mention'; @@ -676,7 +677,7 @@ export class NoteCreateService implements OnApplicationShutdown { this.relayService.deliverToRelays(user, noteActivity); } - dm.execute(); + trackPromise(dm.execute()); })(); } //#endregion diff --git a/packages/backend/src/core/NoteReadService.ts b/packages/backend/src/core/NoteReadService.ts index 03c1735e04f004ab7fb183fe37a1eb9344d8c962..c73cf765922bcb8a85d4c79366e669d95af08ba4 100644 --- a/packages/backend/src/core/NoteReadService.ts +++ b/packages/backend/src/core/NoteReadService.ts @@ -14,6 +14,7 @@ import { IdService } from '@/core/IdService.js'; import { GlobalEventService } from '@/core/GlobalEventService.js'; import type { NoteUnreadsRepository, MutingsRepository, NoteThreadMutingsRepository } from '@/models/_.js'; import { bindThis } from '@/decorators.js'; +import { trackPromise } from '@/misc/promise-tracker.js'; @Injectable() export class NoteReadService implements OnApplicationShutdown { @@ -107,7 +108,7 @@ export class NoteReadService implements OnApplicationShutdown { // TODO: ↓ã¾ã¨ã‚ã¦ã‚¯ã‚¨ãƒªã—ãŸã„ - this.noteUnreadsRepository.countBy({ + trackPromise(this.noteUnreadsRepository.countBy({ userId: userId, isMentioned: true, }).then(mentionsCount => { @@ -115,9 +116,9 @@ export class NoteReadService implements OnApplicationShutdown { // å…¨ã¦æ—¢èªã«ãªã£ãŸã‚¤ãƒ™ãƒ³ãƒˆã‚’発行 this.globalEventService.publishMainStream(userId, 'readAllUnreadMentions'); } - }); + })); - this.noteUnreadsRepository.countBy({ + trackPromise(this.noteUnreadsRepository.countBy({ userId: userId, isSpecified: true, }).then(specifiedCount => { @@ -125,7 +126,7 @@ export class NoteReadService implements OnApplicationShutdown { // å…¨ã¦æ—¢èªã«ãªã£ãŸã‚¤ãƒ™ãƒ³ãƒˆã‚’発行 this.globalEventService.publishMainStream(userId, 'readAllUnreadSpecifiedNotes'); } - }); + })); } } diff --git a/packages/backend/src/core/NotificationService.ts b/packages/backend/src/core/NotificationService.ts index ad7be83e5b40a7240f9916f7d4f01dfe30b89f31..765fcae0632656a83b178182f135bdc6a5110f69 100644 --- a/packages/backend/src/core/NotificationService.ts +++ b/packages/backend/src/core/NotificationService.ts @@ -20,6 +20,7 @@ import { CacheService } from '@/core/CacheService.js'; import type { Config } from '@/config.js'; import { UserListService } from '@/core/UserListService.js'; import type { FilterUnionByProperty } from '@/types.js'; +import { trackPromise } from '@/misc/promise-tracker.js'; @Injectable() export class NotificationService implements OnApplicationShutdown { @@ -74,7 +75,18 @@ export class NotificationService implements OnApplicationShutdown { } @bindThis - public async createNotification<T extends MiNotification['type']>( + public createNotification<T extends MiNotification['type']>( + notifieeId: MiUser['id'], + type: T, + data: Omit<FilterUnionByProperty<MiNotification, 'type', T>, 'type' | 'id' | 'createdAt' | 'notifierId'>, + notifierId?: MiUser['id'] | null, + ) { + trackPromise( + this.#createNotificationInternal(notifieeId, type, data, notifierId), + ); + } + + async #createNotificationInternal<T extends MiNotification['type']>( notifieeId: MiUser['id'], type: T, data: Omit<FilterUnionByProperty<MiNotification, 'type', T>, 'type' | 'id' | 'createdAt' | 'notifierId'>, diff --git a/packages/backend/src/core/QueueModule.ts b/packages/backend/src/core/QueueModule.ts index 4444dc978790f9f11763b9e68c28a510abf2ad35..20a53ff2827d487188fe1a0c5d981f44ad091564 100644 --- a/packages/backend/src/core/QueueModule.ts +++ b/packages/backend/src/core/QueueModule.ts @@ -3,12 +3,12 @@ * SPDX-License-Identifier: AGPL-3.0-only */ -import { setTimeout } from 'node:timers/promises'; import { Inject, Module, OnApplicationShutdown } from '@nestjs/common'; import * as Bull from 'bullmq'; import { DI } from '@/di-symbols.js'; import type { Config } from '@/config.js'; import { QUEUE, baseQueueOptions } from '@/queue/const.js'; +import { allSettled } from '@/misc/promise-tracker.js'; import type { Provider } from '@nestjs/common'; import type { DeliverJobData, InboxJobData, EndedPollNotificationJobData, WebhookDeliverJobData, RelationshipJobData } from '../queue/types.js'; @@ -106,14 +106,9 @@ export class QueueModule implements OnApplicationShutdown { ) {} public async dispose(): Promise<void> { - if (process.env.NODE_ENV === 'test') { - // XXX: - // Shutting down the existing connections causes errors on Jest as - // Misskey has asynchronous postgres/redis connections that are not - // awaited. - // Let's wait for some random time for them to finish. - await setTimeout(5000); - } + // Wait for all potential queue jobs + await allSettled(); + // And then close all queues await Promise.all([ this.systemQueue.close(), this.endedPollNotificationQueue.close(), diff --git a/packages/backend/src/core/ReactionService.ts b/packages/backend/src/core/ReactionService.ts index 3ca12551b1db4c52e83562bf83c9344dc1425d65..2e8f76fa8a5d646f018435b86195444f30d9d6fb 100644 --- a/packages/backend/src/core/ReactionService.ts +++ b/packages/backend/src/core/ReactionService.ts @@ -28,6 +28,7 @@ import { UserBlockingService } from '@/core/UserBlockingService.js'; import { CustomEmojiService } from '@/core/CustomEmojiService.js'; import { RoleService } from '@/core/RoleService.js'; import { FeaturedService } from '@/core/FeaturedService.js'; +import { trackPromise } from '@/misc/promise-tracker.js'; const FALLBACK = 'â¤'; const PER_NOTE_REACTION_USER_PAIR_CACHE_MAX = 16; @@ -268,7 +269,7 @@ export class ReactionService { } } - dm.execute(); + trackPromise(dm.execute()); } //#endregion } @@ -316,7 +317,7 @@ export class ReactionService { dm.addDirectRecipe(reactee as MiRemoteUser); } dm.addFollowersRecipe(); - dm.execute(); + trackPromise(dm.execute()); } //#endregion } diff --git a/packages/backend/src/core/activitypub/ApDeliverManagerService.ts b/packages/backend/src/core/activitypub/ApDeliverManagerService.ts index 81003bcf1c795f3eb1b7e7e116377052c225f0ad..d7414e9c99622a32f8e7e3d6db2441bf3a8d9bec 100644 --- a/packages/backend/src/core/activitypub/ApDeliverManagerService.ts +++ b/packages/backend/src/core/activitypub/ApDeliverManagerService.ts @@ -144,7 +144,7 @@ class DeliverManager { } // deliver - this.queueService.deliverMany(this.actor, this.activity, inboxes); + await this.queueService.deliverMany(this.actor, this.activity, inboxes); } } diff --git a/packages/backend/src/misc/promise-tracker.ts b/packages/backend/src/misc/promise-tracker.ts new file mode 100644 index 0000000000000000000000000000000000000000..c7166c6de9be522a39df7a3d7c5be2c5586af7d3 --- /dev/null +++ b/packages/backend/src/misc/promise-tracker.ts @@ -0,0 +1,23 @@ +/* + * SPDX-FileCopyrightText: syuilo and other misskey contributors + * SPDX-License-Identifier: AGPL-3.0-only + */ + +const promiseRefs: Set<WeakRef<Promise<unknown>>> = new Set(); + +/** + * This tracks promises that other modules decided not to wait for, + * and makes sure they are all settled before fully closing down the server. + */ +export function trackPromise(promise: Promise<unknown>) { + if (process.env.NODE_ENV !== 'test') { + return; + } + const ref = new WeakRef(promise); + promiseRefs.add(ref); + promise.finally(() => promiseRefs.delete(ref)); +} + +export async function allSettled(): Promise<void> { + await Promise.allSettled([...promiseRefs].map(r => r.deref())); +} diff --git a/packages/backend/src/server/api/endpoints/antennas/notes.ts b/packages/backend/src/server/api/endpoints/antennas/notes.ts index 0bf2688b4a7e2c5282e135fa7c3c2946585b7798..7293c2e39bd5321903b81dfe67aee71148b4295f 100644 --- a/packages/backend/src/server/api/endpoints/antennas/notes.ts +++ b/packages/backend/src/server/api/endpoints/antennas/notes.ts @@ -14,6 +14,7 @@ import { NoteEntityService } from '@/core/entities/NoteEntityService.js'; import { IdService } from '@/core/IdService.js'; import { FanoutTimelineService } from '@/core/FanoutTimelineService.js'; import { GlobalEventService } from '@/core/GlobalEventService.js'; +import { trackPromise } from '@/misc/promise-tracker.js'; import { ApiError } from '../../error.js'; export const meta = { @@ -92,7 +93,7 @@ export default class extends Endpoint<typeof meta, typeof paramDef> { // eslint- antenna.isActive = true; antenna.lastUsedAt = new Date(); - this.antennasRepository.update(antenna.id, antenna); + trackPromise(this.antennasRepository.update(antenna.id, antenna)); if (needPublishEvent) { this.globalEventService.publishInternalEvent('antennaUpdated', antenna);