diff --git a/src/lotus/lotusDenoise.ts b/src/lotus/lotusDenoise.ts index e4702390..752e1271 100644 --- a/src/lotus/lotusDenoise.ts +++ b/src/lotus/lotusDenoise.ts @@ -101,8 +101,12 @@ export function startLotusDenoise(vm: CallViewModel): () => void { const roomListeners = new Map void>(); let rooms: LivekitRoom[] = []; - const sub = vm.livekitRoomItems$.subscribe((items) => { - const next = items.map((i) => i.livekitRoom); + // Drive activation off the LOCAL participant's connection(s), not + // `livekitRoomItems$` — that stream excludes the local participant and only + // surfaces rooms with ≥1 remote member, so it wouldn't denoise you while + // you're alone and is a fragile coupling to a remote-render concern. + const sub = vm.allConnections$.subscribe((data) => { + const next = data.getConnections().map((c) => c.livekitRoom); rooms = next; for (const [room, off] of roomListeners) { if (!next.includes(room)) { diff --git a/src/lotus/lotusDenoiseProcessor.ts b/src/lotus/lotusDenoiseProcessor.ts index 3600cd2b..ace2631b 100644 --- a/src/lotus/lotusDenoiseProcessor.ts +++ b/src/lotus/lotusDenoiseProcessor.ts @@ -64,18 +64,41 @@ const sampleRateFor = (model: LotusDenoiseModel): number => // Cache fetched wasm per URL so a reconnect/device-switch doesn't re-download. const wasmCache = new Map>(); -function fetchWasm(url: string): Promise { +async function fetchWasmUncached(url: string): Promise { + const r = await fetch(url); + if (!r.ok) throw new Error(`denoise wasm ${url} -> ${r.status}`); + return r.arrayBuffer(); +} +async function fetchWasm(url: string): Promise { let p = wasmCache.get(url); if (!p) { - p = fetch(url).then((r) => { - if (!r.ok) throw new Error(`denoise wasm ${url} -> ${r.status}`); - return r.arrayBuffer(); - }); + p = fetchWasmUncached(url); + // Never cache a REJECTED fetch: a transient failure (e.g. a blip during a + // reconnect) must not permanently disable denoise for the whole session. + // Evict on failure so the next restart/device-switch retries. + void p.catch(() => wasmCache.delete(url)); wasmCache.set(url, p); } return p; } +/** + * Resume an AudioContext, but never block indefinitely. A suspended context can + * only resume after a user gesture; the denoise action can arrive via host + * postMessage (no gesture), so `resume()` may stay pending forever. Since this + * runs inside LiveKit's per-track change lock, a hung resume() would deadlock + * every later mute/unmute/device-switch. Race it against a timeout and proceed + * either way — the processor's `statechange` watcher resumes it once a gesture + * lands, and a still-suspended context degrades to (temporary) silence that the + * watcher heals, not a hang. + */ +async function resumeCtx(ctx: AudioContext, timeoutMs = 3_000): Promise { + await Promise.race([ + ctx.resume().catch(() => undefined), + new Promise((resolve) => setTimeout(resolve, timeoutMs)), + ]); +} + function supportsSimd(): boolean { try { // Minimal SIMD module (v128) — validates only where SIMD is supported. @@ -123,6 +146,7 @@ export class LotusDenoiseProcessor private ctx?: AudioContext; private graph?: Graph; + private ctxStateHandler?: () => void; public constructor(private readonly config: LotusDenoiseConfig) {} @@ -152,38 +176,63 @@ export class LotusDenoiseProcessor this.disposeGraph(this.graph); this.graph = undefined; this.processedTrack = undefined; - if (this.ctx && this.ctx.state !== "closed") - await this.ctx.close().catch(() => undefined); + await this.closeContext(); + } + + /** Remove the state watcher and close the owned context, if any. */ + private async closeContext(): Promise { + const ctx = this.ctx; + if (!ctx) return; + if (this.ctxStateHandler) { + ctx.removeEventListener("statechange", this.ctxStateHandler); + this.ctxStateHandler = undefined; + } this.ctx = undefined; + if (ctx.state !== "closed") await ctx.close().catch(() => undefined); } /** Create (once) the model-rate context + register the flat worklet modules. */ private async ensureContext(): Promise { const rate = sampleRateFor(this.config.model); if (this.ctx && this.ctx.state !== "closed" && this.ctx.sampleRate === rate) { - if (this.ctx.state === "suspended") await this.ctx.resume(); + if (this.ctx.state === "suspended") await resumeCtx(this.ctx); return; } - if (this.ctx && this.ctx.state !== "closed") - await this.ctx.close().catch(() => undefined); + await this.closeContext(); const ctx = new AudioContext({ sampleRate: rate }); - if (ctx.sampleRate !== rate) { + try { + if (ctx.sampleRate !== rate) + throw new Error(`denoise: got ${ctx.sampleRate}Hz, need ${rate}Hz`); + + // Auto-resume if the OS/browser suspends the context mid-call (mobile + // backgrounding, audio interruption): the dest node otherwise emits + // silence with no recovery. Only resume while a graph is live. + const onStateChange = (): void => { + if (ctx.state === "suspended" && this.graph) + void ctx.resume().catch(() => undefined); + }; + ctx.addEventListener("statechange", onStateChange); + + // Flat models register via addModule here; DTLN/DeepFilterNet bring their + // own processor via the dynamic-imported helper (see buildMlNode). + if (this.config.model === "rnnoise" || this.config.model === "speex") + await ctx.audioWorklet.addModule( + this.config.assetBase + FLAT[this.config.model].script, + ); + if (this.config.gate) + await ctx.audioWorklet.addModule(this.config.assetBase + GATE.script); + // The action can arrive via host postMessage, not a gesture in this + // iframe, so the context can start suspended — resume without hanging. + if (ctx.state === "suspended") await resumeCtx(ctx); + + this.ctx = ctx; + this.ctxStateHandler = onStateChange; + } catch (e) { + // Don't leak a half-initialised context on any failure path. await ctx.close().catch(() => undefined); - throw new Error(`denoise: got ${ctx.sampleRate}Hz, need ${rate}Hz`); + throw e; } - // Flat models register via addModule here; DTLN/DeepFilterNet bring their - // own processor via the dynamic-imported helper (see buildMlNode). - if (this.config.model === "rnnoise" || this.config.model === "speex") - await ctx.audioWorklet.addModule( - this.config.assetBase + FLAT[this.config.model].script, - ); - if (this.config.gate) - await ctx.audioWorklet.addModule(this.config.assetBase + GATE.script); - // The action arrives via host postMessage, not a gesture in this iframe, so - // the context can start suspended. - if (ctx.state === "suspended") await ctx.resume(); - this.ctx = ctx; } private async buildGraph(track: MediaStreamTrack): Promise {