diff --git a/.changeset/short-impalas-nail.md b/.changeset/short-impalas-nail.md new file mode 100644 index 0000000000..7352754a13 --- /dev/null +++ b/.changeset/short-impalas-nail.md @@ -0,0 +1,5 @@ +--- +"livekit-client": patch +--- + +forward worker logs to main thread diff --git a/src/api/WebSocketStream.ts b/src/api/WebSocketStream.ts index b5a4e6b6c8..d869aeb7e5 100644 --- a/src/api/WebSocketStream.ts +++ b/src/api/WebSocketStream.ts @@ -2,6 +2,7 @@ import { ConnectionError } from '../room/errors'; import { sleep } from '../room/utils'; import TypedPromise from '../utils/TypedPromise'; +import { getErrorDescription } from './utils'; export interface WebSocketConnection { readable: ReadableStream; @@ -68,13 +69,7 @@ export class WebSocketStream controller.enqueue(data); ws.onerror = (e) => - controller.error( - ConnectionError.websocket( - e instanceof Error - ? `${e.name}: ${e.message}` - : `Encountered unknown websocket error: ${String(e)}`, - ), - ); + controller.error(ConnectionError.websocket(getErrorDescription(e, 'websocket'))); ws.onclose = (ev) => { if (ev.wasClean || ev.code === 1000) { controller.close(); diff --git a/src/api/utils.ts b/src/api/utils.ts index 94d6662b45..c3549a077f 100644 --- a/src/api/utils.ts +++ b/src/api/utils.ts @@ -63,3 +63,13 @@ export function getAbortReasonAsString( return 'toString' in reason ? reason.toString() : defaultMessage; } } + +export function getErrorDescription(error: unknown, errorCategory: string): string { + if (error instanceof Error) { + if (error.name && error.message) { + return `${error.name}: ${error.message}`; + } + return error.name; + } + return `Encountered unknown ${errorCategory} error: ${String(error)}`; +} diff --git a/src/e2ee/E2eeManager.ts b/src/e2ee/E2eeManager.ts index 6395baddd7..f2c30db573 100644 --- a/src/e2ee/E2eeManager.ts +++ b/src/e2ee/E2eeManager.ts @@ -3,7 +3,7 @@ import { EventEmitter } from 'events'; import type TypedEventEmitter from 'typed-emitter'; import type { FrameMetadata } from '../frameMetadata/types'; import { hasFrameMetadataPublishOptions } from '../frameMetadata/utils'; -import log, { LogLevel, workerLogger } from '../logger'; +import log, { LogLevel, onWorkerLogLevelChanged, workerLogger } from '../logger'; import type RTCEngine from '../room/RTCEngine'; import type Room from '../room/Room'; import { ConnectionState } from '../room/Room'; @@ -129,6 +129,9 @@ export class E2EEManager this.worker.onmessage = this.onWorkerMessage; this.worker.onerror = this.onWorkerError; this.worker.postMessage(msg); + onWorkerLogLevelChanged((level) => { + this.worker?.postMessage({ kind: 'setLogLevel', data: { level } }); + }); } } } @@ -156,25 +159,23 @@ export class E2EEManager const { kind, data } = ev.data; switch (kind) { case 'error': - log.error(data.error.message); - - // If error has uuid, it's from an async operation (encrypt/decrypt) - // Reject the corresponding future + // If error has uuid, it's from an async operation (encrypt/decrypt). + // Reject the corresponding future and let the caller decide how to log/handle; + // logging here would duplicate whatever the caller does. if (data.uuid) { const decryptFuture = this.decryptDataRequests.get(data.uuid); if (decryptFuture?.reject) { decryptFuture.reject(data.error); - break; // Don't emit general error if it's handled by future + break; } const encryptFuture = this.encryptDataRequests.get(data.uuid); if (encryptFuture?.reject) { encryptFuture.reject(data.error); - break; // Don't emit general error if it's handled by future + break; } } - // Emit general error event for unhandled errors this.emit(EncryptionEvent.EncryptionError, data.error, data.participantIdentity); break; case 'initAck': @@ -235,6 +236,9 @@ export class E2EEManager case 'packetTrailerMetadata': this.handleFrameMetadata(data.trackId, data.rtpTimestamp, data.ssrc, data.metadata); break; + case 'log': + workerLogger[data.level](data.msg, data.context); + break; default: break; } diff --git a/src/e2ee/types.ts b/src/e2ee/types.ts index dacef75ce0..756af8395f 100644 --- a/src/e2ee/types.ts +++ b/src/e2ee/types.ts @@ -176,6 +176,22 @@ export interface PTMetadataFromE2EEMessage extends BaseMessage { data: FrameMetadataPayload; } +export interface LogMessage extends BaseMessage { + kind: 'log'; + data: { + level: 'trace' | 'debug' | 'info' | 'warn' | 'error'; + msg: string; + context?: object; + }; +} + +export interface SetLogLevelMessage extends BaseMessage { + kind: 'setLogLevel'; + data: { + level: LogLevel; + }; +} + export type E2EEWorkerMessage = | InitMessage | SetKeyMessage @@ -193,7 +209,9 @@ export type E2EEWorkerMessage = | DecryptDataResponseMessage | EncryptDataRequestMessage | EncryptDataResponseMessage - | PTMetadataFromE2EEMessage; + | PTMetadataFromE2EEMessage + | LogMessage + | SetLogLevelMessage; export type KeySet = { material: CryptoKey; encryptionKey: CryptoKey }; diff --git a/src/e2ee/worker/DataCryptor.ts b/src/e2ee/worker/DataCryptor.ts index 6a95681a04..19d2b7068c 100644 --- a/src/e2ee/worker/DataCryptor.ts +++ b/src/e2ee/worker/DataCryptor.ts @@ -1,3 +1,4 @@ +import { getErrorDescription } from '../../api/utils'; import { workerLogger } from '../../logger'; import type { NonSharedUint8Array } from '../../type-polyfills/non-shared-typed-arrays'; import { ENCRYPTION_ALGORITHM } from '../constants'; @@ -135,7 +136,7 @@ export class DataCryptor { } } else { throw new CryptorError( - `DataCryptor: Decryption failed: ${error.message}`, + `DataCryptor: Decryption failed: ${getErrorDescription(error, 'decryption')}`, CryptorErrorReason.InvalidKey, keys.participantIdentity, ); diff --git a/src/e2ee/worker/ErrorRateLimiter.test.ts b/src/e2ee/worker/ErrorRateLimiter.test.ts new file mode 100644 index 0000000000..7d56e085e5 --- /dev/null +++ b/src/e2ee/worker/ErrorRateLimiter.test.ts @@ -0,0 +1,53 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { ErrorRateLimiter } from './ErrorRateLimiter'; + +describe('ErrorRateLimiter', () => { + beforeEach(() => { + vi.useFakeTimers(); + vi.setSystemTime(1_000_000); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it('emits first call, throttles the immediate next, then allows after throttle window', () => { + const l = new ErrorRateLimiter(1000, 60_000, 5); + expect(l.shouldEmit('k')).toBe(true); + vi.setSystemTime(1_000_500); + expect(l.shouldEmit('k')).toBe(false); + vi.setSystemTime(1_001_600); + expect(l.shouldEmit('k')).toBe(true); + }); + + it('caps at maxPerWindow and invokes onSuppress once', () => { + const l = new ErrorRateLimiter(0, 60_000, 3); + const onSuppress = vi.fn(); + // one free emit on window reset (count stays 0), then increments to 3 + expect(l.shouldEmit('k', onSuppress)).toBe(true); + expect(l.shouldEmit('k', onSuppress)).toBe(true); + expect(l.shouldEmit('k', onSuppress)).toBe(true); + expect(l.shouldEmit('k', onSuppress)).toBe(true); + // now count == 3 == max → suppressed, callback fires once + expect(l.shouldEmit('k', onSuppress)).toBe(false); + expect(l.shouldEmit('k', onSuppress)).toBe(false); + expect(onSuppress).toHaveBeenCalledTimes(1); + }); + + it('resets count after windowMs', () => { + const l = new ErrorRateLimiter(0, 1000, 2); + l.shouldEmit('k'); + l.shouldEmit('k'); + l.shouldEmit('k'); + expect(l.shouldEmit('k')).toBe(false); + vi.setSystemTime(1_003_000); + expect(l.shouldEmit('k')).toBe(true); + }); + + it('tracks keys independently', () => { + const l = new ErrorRateLimiter(1000); + expect(l.shouldEmit('a')).toBe(true); + expect(l.shouldEmit('b')).toBe(true); + expect(l.shouldEmit('a')).toBe(false); + }); +}); diff --git a/src/e2ee/worker/ErrorRateLimiter.ts b/src/e2ee/worker/ErrorRateLimiter.ts new file mode 100644 index 0000000000..36cf32b30a --- /dev/null +++ b/src/e2ee/worker/ErrorRateLimiter.ts @@ -0,0 +1,52 @@ +/** + * Per-key rate limiter for repeated errors. Prevents log/emit floods and the + * unbounded map growth that a per-event log would cause when a broken key + * keeps producing failures. + */ +export class ErrorRateLimiter { + private lastAt: Map = new Map(); + + private counts: Map = new Map(); + + constructor( + private readonly throttleMs: number = 1000, + private readonly windowMs: number = 60_000, + private readonly maxPerWindow: number = 5, + ) {} + + reset() { + this.lastAt.clear(); + this.counts.clear(); + } + + countFor(key: string): number { + return this.counts.get(key) ?? 0; + } + + /** + * Returns true if the caller should emit for this key. Invokes `onSuppress` + * exactly once per window when the per-window limit is first crossed. + */ + shouldEmit(key: string, onSuppress?: () => void): boolean { + const now = Date.now(); + const last = this.lastAt.get(key) ?? 0; + const count = this.counts.get(key) ?? 0; + + if (now - last > this.windowMs) { + this.counts.set(key, 0); + this.lastAt.set(key, now); + return true; + } + if (now - last < this.throttleMs) return false; + if (count >= this.maxPerWindow) { + if (count === this.maxPerWindow) { + onSuppress?.(); + this.counts.set(key, count + 1); + } + return false; + } + this.lastAt.set(key, now); + this.counts.set(key, count + 1); + return true; + } +} diff --git a/src/e2ee/worker/FrameCryptor.ts b/src/e2ee/worker/FrameCryptor.ts index 046e59d76f..bc0b771a61 100644 --- a/src/e2ee/worker/FrameCryptor.ts +++ b/src/e2ee/worker/FrameCryptor.ts @@ -1,6 +1,7 @@ // TODO code inspired by https://github.com/webrtc/samples/blob/gh-pages/src/content/insertable-streams/endtoend-encryption/js/worker.js import { EventEmitter } from 'events'; import type TypedEventEmitter from 'typed-emitter'; +import { getErrorDescription } from '../../api/utils'; import { appendPacketTrailerToEncodedFrame, processPacketTrailer, @@ -22,6 +23,7 @@ import type { RatchetResult, } from '../types'; import { deriveKeys, isVideoFrame, needsRbspUnescaping, parseRbsp, writeRbsp } from '../utils'; +import { ErrorRateLimiter } from './ErrorRateLimiter'; import type { ParticipantKeyHandler } from './ParticipantKeyHandler'; import { processNALUsForEncryption } from './naluUtils'; import { identifySifPayload } from './sifPayload'; @@ -111,18 +113,7 @@ export class FrameCryptor extends BaseFrameCryptor { private frameMetadataFrameId = 0; - /** - * Throttling mechanism for decryption errors to prevent memory leaks - */ - private lastErrorTimestamp: Map = new Map(); - - private errorCounts: Map = new Map(); - - private readonly ERROR_THROTTLE_MS = 1000; // Emit error at most once per second - - private readonly MAX_ERRORS_PER_MINUTE = 5; // Maximum errors to emit per minute per key - - private readonly ERROR_WINDOW_MS = 60000; // 1 minute window + private errorLimiter = new ErrorRateLimiter(); private undecryptedTrackTimeout?: ReturnType; @@ -196,8 +187,7 @@ export class FrameCryptor extends BaseFrameCryptor { clearTimeout(this.undecryptedTrackTimeout); this.undecryptedTrackTimeout = undefined; this.participantIdentity = undefined; - this.lastErrorTimestamp = new Map(); - this.errorCounts = new Map(); + this.errorLimiter.reset(); } isEnabled() { @@ -421,64 +411,24 @@ export class FrameCryptor extends BaseFrameCryptor { this.sifTrailer = trailer; } - /** - * Checks if we should emit an error based on throttling rules to prevent memory leaks - * @param errorKey - unique key identifying the error context - * @returns true if the error should be emitted, false otherwise - */ - private shouldEmitError(errorKey: string): boolean { - const now = Date.now(); - const lastErrorTime = this.lastErrorTimestamp.get(errorKey) ?? 0; - const errorCount = this.errorCounts.get(errorKey) ?? 0; - - // Reset count if we're in a new time window - if (now - lastErrorTime > this.ERROR_WINDOW_MS) { - this.errorCounts.set(errorKey, 0); - this.lastErrorTimestamp.set(errorKey, now); - return true; - } - - // Check if we've exceeded the throttle time - if (now - lastErrorTime < this.ERROR_THROTTLE_MS) { - return false; - } - - // Check if we've exceeded the max errors per window - if (errorCount >= this.MAX_ERRORS_PER_MINUTE) { - // Only log a warning once when hitting the limit - if (errorCount === this.MAX_ERRORS_PER_MINUTE) { - workerLogger.warn(`Suppressing further decryption errors for ${this.participantIdentity}`, { - ...this.logContext, - errorKey, - }); - this.errorCounts.set(errorKey, errorCount + 1); - } - return false; - } - - // Update tracking - this.lastErrorTimestamp.set(errorKey, now); - this.errorCounts.set(errorKey, errorCount + 1); - return true; - } - - /** - * Emits a throttled error to prevent memory leaks from repeated decryption failures - * @param error - the CryptorError to emit - */ private emitThrottledError(error: CryptorError) { const errorKey = `${this.participantIdentity}-${error.reason}-decrypt`; + const emit = this.errorLimiter.shouldEmit(errorKey, () => { + workerLogger.warn(`Suppressing further decryption errors for ${this.participantIdentity}`, { + ...this.logContext, + errorKey, + }); + }); + if (!emit) return; - if (this.shouldEmitError(errorKey)) { - const errorCount = this.errorCounts.get(errorKey) ?? 0; - if (errorCount > 1) { - workerLogger.debug(`Decryption error (${errorCount} occurrences in window)`, { - ...this.logContext, - reason: CryptorErrorReason[error.reason], - }); - } - this.emit(CryptorEvent.Error, error); + const count = this.errorLimiter.countFor(errorKey); + if (count > 1) { + workerLogger.debug(`Decryption error (${count} occurrences in window)`, { + ...this.logContext, + reason: CryptorErrorReason[error.reason], + }); } + this.emit(CryptorEvent.Error, error); } /** @@ -587,7 +537,7 @@ export class FrameCryptor extends BaseFrameCryptor { return controller.enqueue(encodedFrame); } catch (e: any) { // TODO: surface this to the app. - workerLogger.error(e); + workerLogger.error(`error while encrypting`, { ...this.logContext, error: e }); } } else { workerLogger.debug('failed to encrypt, emitting error', this.logContext); @@ -858,7 +808,7 @@ export class FrameCryptor extends BaseFrameCryptor { } } else { throw new CryptorError( - `Decryption failed: ${error.message}`, + `Decryption failed: ${getErrorDescription(error, 'decryption')}`, CryptorErrorReason.InvalidKey, this.participantIdentity, ); diff --git a/src/e2ee/worker/e2ee.worker.ts b/src/e2ee/worker/e2ee.worker.ts index 912f305024..a0ca1c5275 100644 --- a/src/e2ee/worker/e2ee.worker.ts +++ b/src/e2ee/worker/e2ee.worker.ts @@ -18,6 +18,7 @@ import type { ScriptTransformOptions, } from '../types'; import { DataCryptor } from './DataCryptor'; +import { ErrorRateLimiter } from './ErrorRateLimiter'; import { FrameCryptor, encryptionEnabledMap } from './FrameCryptor'; import { ParticipantKeyHandler } from './ParticipantKeyHandler'; @@ -36,8 +37,21 @@ let keyProviderOptions: KeyProviderOptions = KEY_PROVIDER_DEFAULTS; let rtpMap: Map = new Map(); +const dataDecryptErrorLimiter = new ErrorRateLimiter(); + workerLogger.setDefaultLevel('info'); +// Forward worker log calls to the main thread so they reach any +// setLogExtension consumer installed there. The main-thread workerLogger +// re-emits them, which invokes both the console and the extension. +workerLogger.methodFactory = (methodName) => (msg, context) => { + postMessage({ + kind: 'log', + data: { level: methodName, msg, context }, + }); +}; +workerLogger.setLevel(workerLogger.getLevel()); + onmessage = (ev) => { messageQueue.run(async () => { const { kind, data }: E2EEWorkerMessage = ev.data; @@ -45,7 +59,7 @@ onmessage = (ev) => { switch (kind) { case 'init': workerLogger.setLevel(data.loglevel); - workerLogger.info('worker initialized'); + workerLogger.info('e2ee worker initialized'); keyProviderOptions = data.keyProviderOptions; useSharedKey = !!data.keyProviderOptions.sharedKey; // acknowledge init successful @@ -55,6 +69,9 @@ onmessage = (ev) => { }; postMessage(ackMsg); break; + case 'setLogLevel': + workerLogger.setLevel(data.level); + break; case 'enable': setEncryptionEnabled(data.enabled, data.participantIdentity); workerLogger.info( @@ -96,11 +113,6 @@ onmessage = (ev) => { data.payload, getParticipantKeyHandler(data.participantIdentity), ); - console.log('encrypted payload', { - original: data.payload, - encrypted: encryptedPayload, - iv, - }); postMessage({ kind: 'encryptDataResponse', data: { @@ -125,12 +137,23 @@ onmessage = (ev) => { data: { payload: decryptedPayload, uuid: data.uuid }, } satisfies DecryptDataResponseMessage); } catch (error) { - // Send error back to main thread with uuid so it can reject the corresponding promise - workerLogger.error('DataCryptor decryption failed', { - error, - participantIdentity: data.participantIdentity, - uuid: data.uuid, + // Send error back to main thread with uuid so it can reject the corresponding promise. + // The error response must always be posted so the awaiting future resolves; only the + // log is throttled to avoid flooding when a broken key keeps producing failures. + const errorKey = `${data.participantIdentity}-datadecrypt`; + const shouldLog = dataDecryptErrorLimiter.shouldEmit(errorKey, () => { + workerLogger.warn( + `Suppressing further data decryption errors for ${data.participantIdentity}`, + { errorKey }, + ); }); + if (shouldLog) { + workerLogger.error('DataCryptor decryption failed', { + error, + participantIdentity: data.participantIdentity, + uuid: data.uuid, + }); + } postMessage({ kind: 'error', data: { diff --git a/src/logger.ts b/src/logger.ts index 5a3de61b84..d03cd24dc4 100644 --- a/src/logger.ts +++ b/src/logger.ts @@ -131,3 +131,20 @@ export function setLogExtension(extension: LogExtension, logger?: StructuredLogg } export const workerLogger = log.getLogger(LoggerNames.E2EE) as StructuredLogger; + +const workerLogLevelListeners = new Set<(level: LogLevel) => void>(); + +const originalWorkerSetLevel = workerLogger.setLevel.bind(workerLogger); +workerLogger.setLevel = ((level: log.LogLevelDesc, persist?: boolean) => { + originalWorkerSetLevel(level, persist); + const numeric = workerLogger.getLevel() as LogLevel; + workerLogLevelListeners.forEach((cb) => cb(numeric)); +}) as typeof workerLogger.setLevel; + +/** @internal Subscribe to workerLogger level changes (so E2EE workers can be kept in sync). */ +export function onWorkerLogLevelChanged(cb: (level: LogLevel) => void): () => void { + workerLogLevelListeners.add(cb); + return () => { + workerLogLevelListeners.delete(cb); + }; +} diff --git a/src/room/RTCEngine.ts b/src/room/RTCEngine.ts index 08b4f310e6..fca176ff17 100644 --- a/src/room/RTCEngine.ts +++ b/src/room/RTCEngine.ts @@ -930,12 +930,21 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit this.log.error('Received encrypted packet but E2EE not set up'); return; } - const decryptedData = await this.e2eeManager?.handleEncryptedData( - dp.value.value.encryptedValue as NonSharedUint8Array, - dp.value.value.iv as NonSharedUint8Array, - dp.participantIdentity, - dp.value.value.keyIndex, - ); + let decryptedData; + try { + decryptedData = await this.e2eeManager.handleEncryptedData( + dp.value.value.encryptedValue as NonSharedUint8Array, + dp.value.value.iv as NonSharedUint8Array, + dp.participantIdentity, + dp.value.value.keyIndex, + ); + } catch (err) { + this.log.debug('failed to decrypt data packet', { + error: err, + participantIdentity: dp.participantIdentity, + }); + return; + } const decryptedPacket = EncryptedPacketPayload.fromBinary(decryptedData.payload); const newDp = new DataPacket({ value: decryptedPacket.value,