Perform deferred jobs on shutdown (#729)

packages/backend/src/core/NoteCreateService.ts

@@ -1101,17 +1101,18 @@ export class NoteCreateService implements OnApplicationShutdown {
  }
 
  @bindThis
- private performUpdateNotesCount(id: MiNote['id'], incrBy: number) {
- this.instancesRepository.increment({ id: id }, 'notesCount', incrBy);
+ private async performUpdateNotesCount(id: MiNote['id'], incrBy: number) {
+ await this.instancesRepository.increment({ id: id }, 'notesCount', incrBy);
  }
 
  @bindThis
- public dispose(): void {
+ public async dispose(): Promise<void> {
  this.#shutdownController.abort();
+ await this.updateNotesCountQueue.performAllNow();
  }
 
  @bindThis
- public onApplicationShutdown(signal?: string | undefined): void {
- this.dispose();
+ public async onApplicationShutdown(signal?: string | undefined): Promise<void> {
+ await this.dispose();
  }
 }

packages/backend/src/misc/collapsed-queue.ts

@@ -1,24 +1,43 @@
+/*
+ * SPDX-FileCopyrightText: syuilo and misskey-project
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+type Job<V> = {
+ value: V;
+ timer: NodeJS.Timeout;
+};
+
 export class CollapsedQueue<K, V> {
- private jobs: Map<K, V> = new Map();
+ private jobs: Map<K, Job<V>> = new Map();
 
  constructor(
  private timeout: number,
  private collapse: (oldValue: V, newValue: V) => V,
- private doJob: (key: K, value: V) => void,
- ) { }
+ private perform: (key: K, value: V) => Promise<void>,
+ ) {}
 
  enqueue(key: K, value: V) {
  if (this.jobs.has(key)) {
  const old = this.jobs.get(key)!;
- const merged = this.collapse(old, value);
- this.jobs.set(key, merged);
+ const merged = this.collapse(old.value, value);
+ this.jobs.set(key, { ...old, value: merged });
  } else {
- this.jobs.set(key, value);
- setTimeout(() => {
- const value = this.jobs.get(key)!;
+ const timer = setTimeout(() => {
+ const job = this.jobs.get(key)!;
  this.jobs.delete(key);
- this.doJob(key, value);
+ this.perform(key, job.value);
  }, this.timeout);
+ this.jobs.set(key, { value, timer });
+ }
+ }
+
+ async performAllNow() {
+ const entries = [...this.jobs.entries()];
+ this.jobs.clear();
+ for (const [_key, job] of entries) {
+ clearTimeout(job.timer);
  }
+ await Promise.allSettled(entries.map(([key, job]) => this.perform(key, job.value)));
  }
 }

packages/backend/src/queue/processors/InboxProcessorService.ts

@@ -4,7 +4,7 @@
  */
 
 import { URL } from 'node:url';
-import { Injectable } from '@nestjs/common';
+import { Injectable, OnApplicationShutdown } from '@nestjs/common';
 import httpSignature from '@peertube/http-signature';
 import * as Bull from 'bullmq';
 import type Logger from '@/logger.js';
@@ -26,13 +26,13 @@ import { JsonLdService } from '@/core/activitypub/JsonLdService.js';
 import { ApInboxService } from '@/core/activitypub/ApInboxService.js';
 import { bindThis } from '@/decorators.js';
 import { IdentifiableError } from '@/misc/identifiable-error.js';
-import { QueueLoggerService } from '../QueueLoggerService.js';
-import type { InboxJobData } from '../types.js';
 import { CollapsedQueue } from '@/misc/collapsed-queue.js';
 import { MiNote } from '@/models/Note.js';
+import { QueueLoggerService } from '../QueueLoggerService.js';
+import type { InboxJobData } from '../types.js';
 
 @Injectable()
-export class InboxProcessorService {
+export class InboxProcessorService implements OnApplicationShutdown {
  private logger: Logger;
  private updateInstanceQueue: CollapsedQueue<MiNote['id'], Date>;
 
@@ -219,10 +219,20 @@ export class InboxProcessorService {
  }
 
  @bindThis
- public performUpdateInstance(id: string, value: Date) {
- this.federatedInstanceService.update(id, {
+ public async performUpdateInstance(id: string, value: Date) {
+ await this.federatedInstanceService.update(id, {
  latestRequestReceivedAt: value,
  isNotResponding: false,
  });
  }
+
+ @bindThis
+ public async dispose(): Promise<void> {
+ await this.updateInstanceQueue.performAllNow();
+ }
+
+ @bindThis
+ async onApplicationShutdown(signal?: string) {
+ await this.dispose();
+ }
 }