From 27ae8dcb81c9e3e584c1cb3f10ca619720467825 Mon Sep 17 00:00:00 2001 From: Franz Rolfsvaag Date: Wed, 22 Jul 2026 21:36:50 +0200 Subject: [PATCH] Bound OBS source inventory bursts --- TODO.md | 5 +++-- .../backend/companion/gateway.js | 17 +++++++++++++---- .../lumi_transcription/companion_manifest.json | 2 +- plugins/lumi_transcription/tests/verify.js | 3 +++ 4 files changed, 20 insertions(+), 7 deletions(-) diff --git a/TODO.md b/TODO.md index 0f9992a..f8d80bf 100644 --- a/TODO.md +++ b/TODO.md @@ -55,8 +55,9 @@ payload deterministically, adds an executable-level package diagnostic, and show OBS maintenance failures directly instead of leaving them only in diagnostics. Experimental.9 coalesces OBS callbacks into fixed 20 ms network frames and gives -audio its own soft gateway budget, so normal or pathological capture cadence can -never rate-limit and disconnect the Companion control connection. +audio and initial OBS source inventory their own soft gateway budgets, so normal +or pathological capture cadence cannot rate-limit and disconnect the Companion +control connection. Release-blocking work remains: diff --git a/plugins/lumi_transcription/backend/companion/gateway.js b/plugins/lumi_transcription/backend/companion/gateway.js index e542cd1..c7a2808 100644 --- a/plugins/lumi_transcription/backend/companion/gateway.js +++ b/plugins/lumi_transcription/backend/companion/gateway.js @@ -4,6 +4,7 @@ const { insecureDeviceAllowed } = require("./device_store"); const MAX_CONTROL_MESSAGES_PER_SECOND = 120; const MAX_AUDIO_MESSAGES_PER_SECOND = 200; +const MAX_SOURCE_MESSAGES_PER_SECOND = 1000; class CompanionGateway { constructor(options) { @@ -32,7 +33,9 @@ class CompanionGateway { let windowStarted = Date.now(); let controlMessagesInWindow = 0; let audioMessagesInWindow = 0; + let sourceMessagesInWindow = 0; let audioMessagesDropped = 0; + let sourceMessagesDropped = 0; let messageChain = Promise.resolve(); const send = (type, payload, sessionId = session?.id || null) => { if (socket.readyState === WebSocket.OPEN) socket.send(JSON.stringify(envelope(type, payload, sessionId))); @@ -49,7 +52,8 @@ class CompanionGateway { try { if (Date.now() - windowStarted >= 1000) { if (audioMessagesDropped) this.log.append({ kind: "audio_rate_limited", device_id: device.id, session_id: session?.id, dropped_frames: audioMessagesDropped }); - windowStarted = Date.now(); controlMessagesInWindow = 0; audioMessagesInWindow = 0; audioMessagesDropped = 0; + if (sourceMessagesDropped) this.log.append({ kind: "source_rate_limited", device_id: device.id, session_id: session?.id, dropped_updates: sourceMessagesDropped }); + windowStarted = Date.now(); controlMessagesInWindow = 0; audioMessagesInWindow = 0; sourceMessagesInWindow = 0; audioMessagesDropped = 0; sourceMessagesDropped = 0; } if (isBinary) { if (!helloComplete || !session) throw coded("HELLO_REQUIRED", "Complete the handshake before sending audio."); @@ -59,9 +63,14 @@ class CompanionGateway { if (result.gap) send("metric", { kind: "sequence_gap", missing_frames: result.gap }); return; } - controlMessagesInWindow += 1; - if (controlMessagesInWindow > MAX_CONTROL_MESSAGES_PER_SECOND) throw coded("RATE_LIMIT", "Companion control message rate exceeded its limit."); const message = parseEnvelope(body); + if (helloComplete && message.type === "source_update") { + sourceMessagesInWindow += 1; + if (sourceMessagesInWindow > MAX_SOURCE_MESSAGES_PER_SECOND) { sourceMessagesDropped += 1; return; } + } else { + controlMessagesInWindow += 1; + if (controlMessagesInWindow > MAX_CONTROL_MESSAGES_PER_SECOND) throw coded("RATE_LIMIT", "Companion control message rate exceeded its limit."); + } if (!helloComplete) { const hello = validateHello(message); this.devices.updateRuntime(device.id, { companion_version: hello.companion_version, companion_plugin_version: hello.plugin_version }); @@ -131,4 +140,4 @@ function sameHostOrigin(origin, host) { try { return new URL(origin).host === ho function coded(code, message) { return Object.assign(new Error(message), { code }); } function cleanReason(value) { return ["requested", "test_complete", "benchmark_complete", "silence_timeout", "disconnect"].includes(String(value)) ? String(value) : "requested"; } -module.exports = { CompanionGateway, sameHostOrigin, MAX_CONTROL_MESSAGES_PER_SECOND, MAX_AUDIO_MESSAGES_PER_SECOND }; +module.exports = { CompanionGateway, sameHostOrigin, MAX_CONTROL_MESSAGES_PER_SECOND, MAX_AUDIO_MESSAGES_PER_SECOND, MAX_SOURCE_MESSAGES_PER_SECOND }; diff --git a/plugins/lumi_transcription/companion_manifest.json b/plugins/lumi_transcription/companion_manifest.json index 2ca57c8..cd48480 100644 --- a/plugins/lumi_transcription/companion_manifest.json +++ b/plugins/lumi_transcription/companion_manifest.json @@ -2,7 +2,7 @@ "schema_version": 1, "version": "0.1.0-experimental.9", "signed": false, - "release_notes": "Prevents OBS audio callback bursts from rate-limiting the Companion connection by sending steady 20 ms audio frames and keeping excess audio traffic separate from control messages.", + "release_notes": "Prevents OBS audio and initial source-inventory bursts from rate-limiting the Companion connection by sending steady 20 ms audio frames and bounding high-volume capture traffic separately from control messages.", "installer": { "id": "windows-x64-installer", "platform": "win32", diff --git a/plugins/lumi_transcription/tests/verify.js b/plugins/lumi_transcription/tests/verify.js index 7452803..2e3a303 100644 --- a/plugins/lumi_transcription/tests/verify.js +++ b/plugins/lumi_transcription/tests/verify.js @@ -387,6 +387,9 @@ async function verifyAuthenticatedGateway() { await new Promise((resolve) => setTimeout(resolve, 250)); assert.equal(client.readyState, WebSocket.OPEN, "Audio bursts must be bounded without closing the authenticated control connection"); assert.ok(audioMessages > 0 && audioMessages < 350, "Excess audio frames were not softly rate-limited"); + for (let index = 0; index < 1100; index += 1) client.send(JSON.stringify(protocol.envelope("source_update", { source_uuid: crypto.randomUUID(), display_name: `OBS source ${index}` }, sessionId))); + await new Promise((resolve) => setTimeout(resolve, 250)); + assert.equal(client.readyState, WebSocket.OPEN, "OBS source inventory bursts must not close the authenticated control connection"); await new Promise((resolve) => { client.once("close", resolve); client.close(); }); for (let attempt = 0; attempt < 20 && !disconnected; attempt += 1) await new Promise((resolve) => setTimeout(resolve, 5)); assert.equal(disconnected, true);