diff --git a/fluxer_api/src/api/voice/VoiceReconciliationWorker.ts b/fluxer_api/src/api/voice/VoiceReconciliationWorker.ts index 040e9cc0d..b647bc9cd 100644 --- a/fluxer_api/src/api/voice/VoiceReconciliationWorker.ts +++ b/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 | 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 { + 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 { @@ -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 { 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; } } diff --git a/fluxer_api/src/api/voice/tests/VoiceReconciliationWorkerStop.test.ts b/fluxer_api/src/api/voice/tests/VoiceReconciliationWorkerStop.test.ts new file mode 100644 index 000000000..75aecbede --- /dev/null +++ b/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); + }); +}); diff --git a/fluxer_api/src/api/worker/WorkerDependencies.ts b/fluxer_api/src/api/worker/WorkerDependencies.ts index f1db6af71..a6ef6961e 100644 --- a/fluxer_api/src/api/worker/WorkerDependencies.ts +++ b/fluxer_api/src/api/worker/WorkerDependencies.ts @@ -352,7 +352,7 @@ export async function initializeWorkerDependencies(snowflakeService: ISnowflakeS export async function shutdownWorkerDependencies(deps: WorkerDependencies): Promise { Logger.info('Shutting down worker dependencies...'); if (deps.voiceReconciliationWorker !== null) { - deps.voiceReconciliationWorker.stop(); + await deps.voiceReconciliationWorker.stop(); } Logger.info('Worker dependencies shut down successfully'); }