forward worker logs to main thread

.changeset/short-impalas-nail.md

@@ -0,0 +1,5 @@
+---
+"livekit-client": patch
+---
+
+forward worker logs to main thread

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<T extends ArrayBuffer | string = ArrayBuffer | string> {
   readable: ReadableStream<T>;
@@ -68,13 +69,7 @@ export class WebSocketStream<T extends ArrayBuffer | string = ArrayBuffer | stri
             start(controller) {
               ws.onmessage = ({ data }) => 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) {
                   controller.close();

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)}`;
+}

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;
     }

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 };
 

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,
         );

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);
+  });
+});

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<string, number> = new Map();
+
+  private counts: Map<string, number> = 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;
+  }
+}