fix(voice): finish the reconciliation sweep before stopping (#2289)

fluxer_api/src/api/voice/VoiceReconciliationWorker.ts

@@ -131,6 +131,8 @@ export class VoiceReconciliationWorker {
 	private intervalHandle: NodeJS.Timeout | null = null;
 	private reconciling = false;
 	private reconciliationLockLost = false;
+	private activeReconciliation: Promise<void> | null = null;
+	private stopping = false;
 
 	constructor(options: VoiceReconciliationWorkerOptions) {
 		this.gatewayService = options.gatewayService;
@@ -161,18 +163,24 @@ export class VoiceReconciliationWorker {
 			},
 			'Starting VoiceReconciliationWorker',
 		);
+		this.stopping = false;
 		void this.runReconciliation();
 		this.intervalHandle = setInterval(() => {
 			void this.runReconciliation();
 		}, this.intervalMs);
 	}
 
-	stop(): void {
+	async stop(): Promise<void> {
+		this.stopping = true;
 		if (this.intervalHandle) {
 			clearInterval(this.intervalHandle);
 			this.intervalHandle = null;
-			this.logger.info('Stopped VoiceReconciliationWorker');
 		}
+		const activeReconciliation = this.activeReconciliation;
+		if (activeReconciliation !== null) {
+			await activeReconciliation;
+		}
+		this.logger.info('Stopped VoiceReconciliationWorker');
 	}
 
 	async reconcile(): Promise<void> {
@@ -203,6 +211,10 @@ export class VoiceReconciliationWorker {
 		let totalErrors = 0;
 		let totalTransientSkips = 0;
 		for (const room of discovery.rooms) {
+			if (this.stopping) {
+				this.logger.info('Stopping reconciliation sweep because the worker is shutting down');
+				break;
+			}
 			if (this.reconciliationLockLost) {
 				this.logger.warn('Stopping reconciliation sweep because the cluster lock was lost');
 				break;
@@ -261,6 +273,17 @@ export class VoiceReconciliationWorker {
 			return;
 		}
 		this.reconciling = true;
+		const sweep = this.runReconciliationSweep();
+		this.activeReconciliation = sweep;
+		try {
+			await sweep;
+		} finally {
+			this.activeReconciliation = null;
+			this.reconciling = false;
+		}
+	}
+
+	private async runReconciliationSweep(): Promise<void> {
 		let lockToken: string | null = null;
 		let lockRenewalHandle: NodeJS.Timeout | null = null;
 		try {
@@ -287,7 +310,6 @@ export class VoiceReconciliationWorker {
 				await this.releaseReconciliationLock(lockToken);
 			}
 			this.reconciliationLockLost = false;
-			this.reconciling = false;
 		}
 	}
 

fluxer_api/src/api/voice/tests/VoiceReconciliationWorkerStop.test.ts

@@ -0,0 +1,64 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+import type {IKVProvider} from '@pkgs/kv_client/src/IKVProvider';
+import {describe, expect, it, vi} from 'vitest';
+import type {ILogger} from '../../ILogger';
+import type {IGatewayService} from '../../infrastructure/IGatewayService';
+import type {ILiveKitService} from '../../infrastructure/ILiveKitService';
+import type {IVoiceRoomStore} from '../../infrastructure/IVoiceRoomStore';
+import {VoiceReconciliationWorker} from '../VoiceReconciliationWorker';
+
+function createLogger(): ILogger {
+	const logger = {
+		trace: vi.fn(),
+		debug: vi.fn(),
+		info: vi.fn(),
+		warn: vi.fn(),
+		error: vi.fn(),
+		child: () => logger,
+	};
+	return logger as unknown as ILogger;
+}
+
+function createHarness() {
+	const releaseLock = vi.fn().mockResolvedValue(true);
+	const kvClient = {
+		acquireLock: vi.fn().mockResolvedValue(true),
+		extendLock: vi.fn().mockResolvedValue(true),
+		releaseLock,
+		setnx: vi.fn().mockResolvedValue(true),
+		get: vi.fn().mockResolvedValue(null),
+		setex: vi.fn().mockResolvedValue(undefined),
+	} as unknown as IKVProvider;
+	let finishDiscovery: () => void = () => {};
+	const discovery = new Promise<{rooms: []}>((resolve) => {
+		finishDiscovery = () => resolve({rooms: []});
+	});
+	const getActiveVoiceRooms = vi.fn().mockReturnValue(discovery);
+	const worker = new VoiceReconciliationWorker({
+		gatewayService: {getActiveVoiceRooms} as unknown as IGatewayService,
+		liveKitService: {
+			listActiveRooms: async () => ({rooms: [], errors: [], completed: true, searchedServers: 0}),
+		} as unknown as ILiveKitService,
+		voiceRoomStore: {listPinnedRooms: async () => []} as unknown as IVoiceRoomStore,
+		kvClient,
+		logger: createLogger(),
+		intervalMs: 60000,
+		staggerDelayMs: 0,
+	});
+	return {worker, releaseLock, getActiveVoiceRooms, finishDiscovery: () => finishDiscovery()};
+}
+
+describe('VoiceReconciliationWorker stop', () => {
+	it('releases the reconciliation lock before stop resolves', async () => {
+		const {worker, releaseLock, getActiveVoiceRooms, finishDiscovery} = createHarness();
+
+		worker.start();
+		await vi.waitFor(() => expect(getActiveVoiceRooms).toHaveBeenCalled());
+
+		setTimeout(finishDiscovery, 0);
+		await worker.stop();
+
+		expect(releaseLock).toHaveBeenCalledTimes(1);
+	});
+});

fluxer_api/src/api/worker/WorkerDependencies.ts

@@ -352,7 +352,7 @@ export async function initializeWorkerDependencies(snowflakeService: ISnowflakeS
 export async function shutdownWorkerDependencies(deps: WorkerDependencies): Promise<void> {
 	Logger.info('Shutting down worker dependencies...');
 	if (deps.voiceReconciliationWorker !== null) {
-		deps.voiceReconciliationWorker.stop();
+		await deps.voiceReconciliationWorker.stop();
 	}
 	Logger.info('Worker dependencies shut down successfully');
 }