merge: Reorder relay activities to avoid delivery race condition (resolves #989) (!986)

View MR for information: https://activitypub.software/TransFem-org/Sharkey/-/merge_requests/986

Closes #989

Approved-by: dakkar <dakkar@thenautilus.net>
Approved-by: Marie <github@yuugi.dev>

packages/backend/src/core/AccountMoveService.ts

@@ -95,11 +95,12 @@ export class AccountMoveService {
 		const srcPerson = await this.apRendererService.renderPerson(src);
 		const updateAct = this.apRendererService.addContext(this.apRendererService.renderUpdate(srcPerson, src));
 		await this.apDeliverManagerService.deliverToFollowers(src, updateAct);
-		this.relayService.deliverToRelays(src, updateAct);
+		await this.relayService.deliverToRelays(src, updateAct);
 
 		// Deliver Move activity to the followers of the old account
 		const moveAct = this.apRendererService.addContext(this.apRendererService.renderMove(src, dst));
 		await this.apDeliverManagerService.deliverToFollowers(src, moveAct);
+		await this.relayService.deliverToRelays(src, moveAct);
 
 		// Publish meUpdated event
 		const iObj = await this.userEntityService.pack(src.id, src, { schema: 'MeDetailed', includeSecrets: true });

packages/backend/src/core/AccountUpdateService.ts

@@ -27,15 +27,12 @@ export class AccountUpdateService {
 	}
 
 	@bindThis
-	public async publishToFollowers(userId: MiUser['id']) {
+	public async publishToFollowers(user: MiUser) {
-		const user = await this.usersRepository.findOneBy({ id: userId });
-		if (user == null) throw new Error('user not found');
-
 		// フォロワーがリモートユーザーかつ投稿者がローカルユーザーならUpdateを配信
 		if (this.userEntityService.isLocalUser(user)) {
 			const content = this.apRendererService.addContext(this.apRendererService.renderUpdate(await this.apRendererService.renderPerson(user), user));
-			this.apDeliverManagerService.deliverToFollowers(user, content);
+			await this.apDeliverManagerService.deliverToFollowers(user, content);
-			this.relayService.deliverToRelays(user, content);
+			await this.relayService.deliverToRelays(user, content);
 		}
 	}
 }

packages/backend/src/core/NoteCreateService.ts

@@ -51,7 +51,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';
+import { trackTask } from '@/misc/promise-tracker.js';
 import { isUserRelated } from '@/misc/is-user-related.js';
 import { IdentifiableError } from '@/misc/identifiable-error.js';
 import { LatestNoteService } from '@/core/LatestNoteService.js';
@@ -729,7 +729,7 @@ export class NoteCreateService implements OnApplicationShutdown {
 
 			//#region AP deliver
 			if (!data.localOnly && this.userEntityService.isLocalUser(user)) {
-				(async () => {
+				trackTask(async () => {
 					const noteActivity = await this.renderNoteOrRenoteActivity(data, note, user);
 					const dm = this.apDeliverManagerService.createDeliverManager(user, noteActivity);
 
@@ -755,12 +755,12 @@ export class NoteCreateService implements OnApplicationShutdown {
 						dm.addFollowersRecipe();
 					}
 
-					if (['public'].includes(note.visibility)) {
+					await dm.execute();
-						this.relayService.deliverToRelays(user, noteActivity);
-					}
 
-					trackPromise(dm.execute());
+					if (['public'].includes(note.visibility)) {
-				})();
+						await this.relayService.deliverToRelays(user, noteActivity);
+					}
+				});
 			}
 			//#endregion
 		}

packages/backend/src/core/NoteDeleteService.ts

@@ -247,11 +247,11 @@ export class NoteDeleteService {
 
 	@bindThis
 	private async deliverToConcerned(user: { id: MiLocalUser['id']; host: null; }, note: MiNote, content: any) {
-		this.apDeliverManagerService.deliverToFollowers(user, content);
+		await this.apDeliverManagerService.deliverToFollowers(user, content);
-		this.relayService.deliverToRelays(user, content);
+		await this.apDeliverManagerService.deliverToUsers(user, content, [
-		this.apDeliverManagerService.deliverToUsers(user, content, [
 			...await this.getMentionedRemoteUsers(note),
 			...await this.getRenotedOrRepliedRemoteUsers(note),
 		]);
+		await this.relayService.deliverToRelays(user, content);
 	}
 }

packages/backend/src/core/NoteEditService.ts

@@ -46,7 +46,7 @@ import { UtilityService } from '@/core/UtilityService.js';
 import { UserBlockingService } from '@/core/UserBlockingService.js';
 import { CacheService } from '@/core/CacheService.js';
 import { isReply } from '@/misc/is-reply.js';
-import { trackPromise } from '@/misc/promise-tracker.js';
+import { trackTask } from '@/misc/promise-tracker.js';
 import { isUserRelated } from '@/misc/is-user-related.js';
 import { IdentifiableError } from '@/misc/identifiable-error.js';
 import { LatestNoteService } from '@/core/LatestNoteService.js';
@@ -669,7 +669,7 @@ export class NoteEditService implements OnApplicationShutdown {
 
 			//#region AP deliver
 			if (!data.localOnly && this.userEntityService.isLocalUser(user)) {
-				(async () => {
+				trackTask(async () => {
 					const noteActivity = await this.renderNoteOrRenoteActivity(data, note, user);
 					const dm = this.apDeliverManagerService.createDeliverManager(user, noteActivity);
 
@@ -713,12 +713,12 @@ export class NoteEditService implements OnApplicationShutdown {
 						}
 					}
 
-					if (['public'].includes(note.visibility)) {
+					await dm.execute();
-						this.relayService.deliverToRelays(user, noteActivity);
-					}
 
-					trackPromise(dm.execute());
+					if (['public'].includes(note.visibility)) {
-				})();
+						await this.relayService.deliverToRelays(user, noteActivity);
+					}
+				});
 			}
 			//#endregion
 		}

packages/backend/src/core/NotePiningService.ts

@@ -49,7 +49,7 @@ export class NotePiningService {
 	 * @param noteId
 	 */
 	@bindThis
-	public async addPinned(user: { id: MiUser['id']; host: MiUser['host']; }, noteId: MiNote['id']) {
+	public async addPinned(user: MiUser, noteId: MiNote['id']) {
 	// Fetch pinee
 		const note = await this.notesRepository.findOneBy({
 			id: noteId,
@@ -78,7 +78,7 @@ export class NotePiningService {
 
 		// Deliver to remote followers
 		if (this.userEntityService.isLocalUser(user) && !note.localOnly && ['public', 'home'].includes(note.visibility)) {
-			this.deliverPinnedChange(user.id, note.id, true);
+			this.deliverPinnedChange(user, note.id, true);
 		}
 	}
 
@@ -88,7 +88,7 @@ export class NotePiningService {
 	 * @param noteId
 	 */
 	@bindThis
-	public async removePinned(user: { id: MiUser['id']; host: MiUser['host']; }, noteId: MiNote['id']) {
+	public async removePinned(user: MiUser, noteId: MiNote['id']) {
 	// Fetch unpinee
 		const note = await this.notesRepository.findOneBy({
 			id: noteId,
@@ -106,22 +106,19 @@ export class NotePiningService {
 
 		// Deliver to remote followers
 		if (this.userEntityService.isLocalUser(user) && !note.localOnly && ['public', 'home'].includes(note.visibility)) {
-			this.deliverPinnedChange(user.id, noteId, false);
+			this.deliverPinnedChange(user, noteId, false);
 		}
 	}
 
 	@bindThis
-	public async deliverPinnedChange(userId: MiUser['id'], noteId: MiNote['id'], isAddition: boolean) {
+	public async deliverPinnedChange(user: MiUser, noteId: MiNote['id'], isAddition: boolean) {
-		const user = await this.usersRepository.findOneBy({ id: userId });
-		if (user == null) throw new Error('user not found');
-
 		if (!this.userEntityService.isLocalUser(user)) return;
 
 		const target = `${this.config.url}/users/${user.id}/collections/featured`;
 		const item = `${this.config.url}/notes/${noteId}`;
 		const content = this.apRendererService.addContext(isAddition ? this.apRendererService.renderAdd(user, target, item) : this.apRendererService.renderRemove(user, target, item));
 
-		this.apDeliverManagerService.deliverToFollowers(user, content);
+		await this.apDeliverManagerService.deliverToFollowers(user, content);
-		this.relayService.deliverToRelays(user, content);
+		await this.relayService.deliverToRelays(user, content);
 	}
 }

packages/backend/src/core/PollService.ts

@@ -90,10 +90,7 @@ export class PollService {
 	}
 
 	@bindThis
-	public async deliverQuestionUpdate(noteId: MiNote['id']) {
+	public async deliverQuestionUpdate(note: MiNote) {
-		const note = await this.notesRepository.findOneBy({ id: noteId });
-		if (note == null) throw new Error('note not found');
-
 		if (note.localOnly) return;
 
 		const user = await this.usersRepository.findOneBy({ id: note.userId });
@@ -101,8 +98,8 @@ export class PollService {
 
 		if (this.userEntityService.isLocalUser(user)) {
 			const content = this.apRendererService.addContext(this.apRendererService.renderUpdate(await this.apRendererService.renderNote(note, user, false), user));
-			this.apDeliverManagerService.deliverToFollowers(user, content);
+			await this.apDeliverManagerService.deliverToFollowers(user, content);
-			this.relayService.deliverToRelays(user, content);
+			await this.relayService.deliverToRelays(user, content);
 		}
 	}
 }

packages/backend/src/core/activitypub/models/ApNoteService.ts

@@ -296,7 +296,7 @@ export class ApNoteService {
 					await this.pollService.vote(actor, reply, index);
 
 					// リモートフォロワーにUpdate配信
-					this.pollService.deliverQuestionUpdate(reply.id);
+					this.pollService.deliverQuestionUpdate(reply);
 				}
 				return null;
 			};
@@ -493,7 +493,7 @@ export class ApNoteService {
 					await this.pollService.vote(actor, reply, index);
 
 					// リモートフォロワーにUpdate配信
-					this.pollService.deliverQuestionUpdate(reply.id);
+					this.pollService.deliverQuestionUpdate(reply);
 				}
 				return null;
 			};

packages/backend/src/misc/promise-tracker.ts

@@ -5,6 +5,10 @@
 
 const promiseRefs: Set<WeakRef<Promise<unknown>>> = new Set();
 
+export function trackTask(task: () => Promise<unknown>): void {
+	trackPromise(task());
+}
+
 /**
  * This tracks promises that other modules decided not to wait for,
  * and makes sure they are all settled before fully closing down the server.

packages/backend/src/server/api/endpoints/i/update.ts

@@ -614,7 +614,7 @@ export default class extends Endpoint<typeof meta, typeof paramDef> { // eslint-
 
 			// フォロワーにUpdateを配信
 			if (this.userNeedsPublishing(user, updates) || this.profileNeedsPublishing(profile, updatedProfile)) {
-				this.accountUpdateService.publishToFollowers(user.id);
+				this.accountUpdateService.publishToFollowers(user);
 			}
 
 			return iObj;

packages/backend/src/server/api/endpoints/notes/polls/vote.ts

@@ -174,7 +174,7 @@ export default class extends Endpoint<typeof meta, typeof paramDef> { // eslint-
 			}
 
 			// リモートフォロワーにUpdate配信
-			this.pollService.deliverQuestionUpdate(note.id);
+			this.pollService.deliverQuestionUpdate(note);
 		});
 	}
 }