fix(app): terminate the voice e2ee worker with the room that owns it (#1972)

fluxer_app/src/features/voice/engine/VoiceE2EEKeyProvider.ts

@@ -1,6 +1,8 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
-import {ExternalE2EEKeyProvider} from 'livekit-client';
+import {ExternalE2EEKeyProvider, type Room} from 'livekit-client';
+
+const roomE2EEWorkers = new WeakMap<Room, Worker>();
 
 export function createE2EEWorker(): Worker {
 	return new Worker(
@@ -15,3 +17,19 @@ export function createE2EEWorker(): Worker {
 export function createE2EEKeyProvider(): ExternalE2EEKeyProvider {
 	return new ExternalE2EEKeyProvider();
 }
+
+export function ownE2EEWorker(room: Room, worker: Worker | null): void {
+	if (worker === null) return;
+	const previous = roomE2EEWorkers.get(room) ?? null;
+	if (previous !== null && previous !== worker) previous.terminate();
+	roomE2EEWorkers.set(room, worker);
+}
+
+export function releaseE2EEWorker(room: Room | null): boolean {
+	if (room === null) return false;
+	const worker = roomE2EEWorkers.get(room) ?? null;
+	if (worker === null) return false;
+	roomE2EEWorkers.delete(room);
+	worker.terminate();
+	return true;
+}

fluxer_app/src/features/voice/engine/v2/VoiceEngineV2AppConnectionHostAdapter.ts

@@ -20,7 +20,12 @@ import {
 	type VoiceConnectionSnapshot,
 } from '@app/features/voice/engine/VoiceConnectionStateMachine';
 import {VoiceConnectionThrottle} from '@app/features/voice/engine/VoiceConnectionThrottle';
-import {createE2EEKeyProvider, createE2EEWorker} from '@app/features/voice/engine/VoiceE2EEKeyProvider';
+import {
+	createE2EEKeyProvider,
+	createE2EEWorker,
+	ownE2EEWorker,
+	releaseE2EEWorker,
+} from '@app/features/voice/engine/VoiceE2EEKeyProvider';
 import {getSharedVoiceAudioContext} from '@app/features/voice/engine/VoiceSharedAudioContext';
 import {selectLocalMediaPublicationsForConnectionRepublish} from '@app/features/voice/engine/VoiceTrackPublicationUtils';
 import {
@@ -156,6 +161,7 @@ function createRoomOptions(
 ): {
 	roomOptions: RoomOptions;
 	e2eeKeyProvider: ExternalE2EEKeyProvider | null;
+	e2eeWorker: Worker | null;
 } {
 	const roomOptions: RoomOptions = {
 		adaptiveStream: false,
@@ -164,17 +170,20 @@ function createRoomOptions(
 		subscriberVideoCodecExclusions,
 	};
 	let e2eeKeyProvider: ExternalE2EEKeyProvider | null = null;
+	let e2eeWorker: Worker | null = null;
 	if (e2eeKey) {
 		try {
 			e2eeKeyProvider = createE2EEKeyProvider();
-			const worker = createE2EEWorker();
+			e2eeWorker = createE2EEWorker();
-			roomOptions.e2ee = {keyProvider: e2eeKeyProvider, worker};
+			roomOptions.e2ee = {keyProvider: e2eeKeyProvider, worker: e2eeWorker};
 		} catch (error) {
 			logger.error('Failed to construct E2EE key provider/worker', error);
+			e2eeWorker?.terminate();
 			e2eeKeyProvider = null;
+			e2eeWorker = null;
 		}
 	}
-	return {roomOptions, e2eeKeyProvider};
+	return {roomOptions, e2eeKeyProvider, e2eeWorker};
 }
 
 function createRoomConnectOptions(): RoomConnectOptions {
@@ -610,12 +619,14 @@ export class VoiceEngineV2AppConnectionHostAdapter extends Store {
 			logger.warn('Aborting LiveKit room creation after codec probing because attempt is stale', {attemptId});
 			return;
 		}
-		const {roomOptions, e2eeKeyProvider} = createRoomOptions(e2eeKey, subscriberVideoCodecExclusions);
+		const {roomOptions, e2eeKeyProvider, e2eeWorker} = createRoomOptions(e2eeKey, subscriberVideoCodecExclusions);
 		const room = new LiveKitRoom(roomOptions);
+		ownE2EEWorker(room, e2eeWorker);
 		let roomClosed = false;
 		const closeRoom = () => {
 			if (roomClosed) return;
 			roomClosed = true;
+			releaseE2EEWorker(room);
 			onRoomClosed?.(room, attemptId);
 		};
 		const failConnectBeforeRoomConnect = (message: string, error?: unknown) => {
@@ -831,6 +842,7 @@ export class VoiceEngineV2AppConnectionHostAdapter extends Store {
 				} catch (error) {
 					logger.warn('Region hot-swap: failed to disconnect old room', {error});
 				}
+				releaseE2EEWorker(previousRoom);
 				logger.info('Region hot-swap: complete', {
 					previousEndpoint: this.connectionState.voiceServerEndpoint,
 					newEndpoint: endpoint,
@@ -994,6 +1006,7 @@ export class VoiceEngineV2AppConnectionHostAdapter extends Store {
 		if (room) {
 			room.removeAllListeners();
 			room.disconnect();
+			releaseE2EEWorker(room);
 		}
 		this.update(() => {
 			this.transitionConnection({type: 'connection.disconnected', reason});
@@ -1015,6 +1028,7 @@ export class VoiceEngineV2AppConnectionHostAdapter extends Store {
 		if (room) {
 			room.removeAllListeners();
 			room.disconnect();
+			releaseE2EEWorker(room);
 		}
 		this.update(() => {
 			this.transitionConnection({type: 'connection.disconnectForChannelMove'});
@@ -1033,6 +1047,7 @@ export class VoiceEngineV2AppConnectionHostAdapter extends Store {
 		} catch (error) {
 			logger.warn('Terminal unload LiveKit room disconnect failed', {label, error});
 		}
+		releaseE2EEWorker(room);
 	}
 
 	hasTerminalUnloadTransports(): boolean {
@@ -1239,6 +1254,7 @@ export class VoiceEngineV2AppConnectionHostAdapter extends Store {
 		} catch (error) {
 			logger.warn('Failed to disconnect previous room', error);
 		}
+		releaseE2EEWorker(previousRoom);
 	}
 
 	private getPreviousRoomNonScreenShareTracks(previousRoom: Room): Array<LocalTrack> {
@@ -1283,6 +1299,7 @@ export class VoiceEngineV2AppConnectionHostAdapter extends Store {
 		if (room) {
 			room.removeAllListeners();
 			room.disconnect();
+			releaseE2EEWorker(room);
 		}
 		this.update(() => {
 			this.isLocalDisconnecting = false;