Bound OBS source inventory bursts

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
+audio and initial OBS source inventory their own soft gateway budgets, so normal
-never rate-limit and disconnect the Companion control connection.
+or pathological capture cadence cannot rate-limit and disconnect the Companion
+control connection.
 
 Release-blocking work remains:
 

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

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",

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