Rolfsvaag_Datateknikk/Lumi
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 22 Wiki Activity

Prevent audio rate limits from closing Companion

Browse Source
This commit is contained in:
Franz Rolfsvaag 2026-07-22 21:29:27 +02:00
parent b2c9d4b8ac
commit 2e09754b9e
10 changed files with 78 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
TODO.md
View File

@ -54,6 +54,10 @@ Experimental.8 makes the installed single-file build discover and verify its OBS
payload deterministically, adds an executable-level package diagnostic, and shows
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.
Release-blocking work remains:
- Package the native whisper.cpp worker as the Windows CUDA runtime, then benchmark

2
companion/installer/Lumi.Companion.iss
View File

@ -1,5 +1,5 @@
#ifndef AppVersion
#define AppVersion "0.1.0-experimental.8"
#define AppVersion "0.1.0-experimental.9"
#endif
#ifndef SourceRoot
#error SourceRoot must point at the self-contained Companion publish directory.

2
companion/scripts/publish-companion.ps1
View File

@ -1,5 +1,5 @@
param(
[string]$Version = "0.1.0-experimental.8",
[string]$Version = "0.1.0-experimental.9",
[string]$BridgeVersion = "0.1.0-experimental.5"
)

3
companion/src/Lumi.Companion.App/CompanionRuntime.cs
View File

@ -600,7 +600,8 @@ public sealed class CompanionRuntime : IAsyncDisposable
{
if (State.BenchmarkRunning) _benchmarkLastVoiceAt = DateTimeOffset.UtcNow;
}
if (_socket is not null && State.Connected && !_socket.QueueBridgeAudio(frame))
var audioNeeded = State.TestRunning || State.BenchmarkRunning || State.ObsStreaming;
if (audioNeeded && _socket is not null && State.Connected && !_socket.QueueBridgeAudio(frame))
_ = WriteLogAsync("audio_dropped", "An obsolete audio frame was dropped before network delivery.");
return Task.CompletedTask;
}

2
companion/src/Lumi.Companion.App/Lumi.Companion.App.csproj
View File

@ -5,7 +5,7 @@
<Nullable>enable</Nullable>
<ImplicitUsings>enable</ImplicitUsings>
<ApplicationManifest>app.manifest</ApplicationManifest>
<Version>0.1.0-experimental.8</Version>
<Version>0.1.0-experimental.9</Version>
<AssemblyVersion>0.1.0.0</AssemblyVersion>
</PropertyGroup>
<ItemGroup>

41
companion/src/Lumi.Companion.Core/CompanionSocket.cs
View File

@ -7,7 +7,13 @@ namespace Lumi.Companion.Core;
public sealed class CompanionSocket : IAsyncDisposable
{
private const int NetworkAudioPayloadBytes = 640; // 20 ms of 16 kHz mono PCM16.
private readonly Channel<ReadOnlyMemory<byte>> _audio = Channel.CreateBounded<ReadOnlyMemory<byte>>(new BoundedChannelOptions(250) { FullMode = BoundedChannelFullMode.Wait, SingleReader = true });
private readonly object _bridgeAudioGate = new();
private readonly byte[] _bridgeAudioBuffer = new byte[ProtocolV1.MaxAudioPayloadBytes + NetworkAudioPayloadBytes];
private readonly byte[] _bridgeAudioHeader = new byte[ProtocolV1.AudioHeaderBytes];
private int _bridgeAudioBuffered;
private uint _bridgeAudioSequence;
private ClientWebSocket? _socket;
private CancellationTokenSource? _lifetime;
private readonly SemaphoreSlim _sendLock = new(1, 1);
@ -56,11 +62,36 @@ public sealed class CompanionSocket : IAsyncDisposable
}
public bool QueueBridgeAudio(ReadOnlyMemory<byte> encoded)
{
if (encoded.Length < ProtocolV1.AudioHeaderBytes || !encoded.Span[..4].SequenceEqual("LACP"u8)) return false;
var normalized = encoded.ToArray();
SessionId.TryWriteBytes(normalized.AsSpan(20, 16), bigEndian: true, out _);
BinaryPrimitives.WriteUInt64LittleEndian(normalized.AsSpan(12, 8), (ulong)DateTimeOffset.UtcNow.ToUnixTimeMilliseconds() * 1000UL);
return QueueEncodedAudio(normalized);
var input = encoded.Span;
if (input.Length < ProtocolV1.AudioHeaderBytes || input.Length > ProtocolV1.AudioHeaderBytes + ProtocolV1.MaxAudioPayloadBytes ||
!input[..4].SequenceEqual("LACP"u8) || BinaryPrimitives.ReadUInt16LittleEndian(input[6..8]) != ProtocolV1.AudioHeaderBytes) return false;
var payloadBytes = (int)BinaryPrimitives.ReadUInt32LittleEndian(input[60..64]);
if (payloadBytes < 0 || payloadBytes % 2 != 0 || input.Length != ProtocolV1.AudioHeaderBytes + payloadBytes) return false;
lock (_bridgeAudioGate)
{
var sourceChanged = !_bridgeAudioHeader.AsSpan(36, 16).SequenceEqual(input[36..52]);
var stateChanged = _bridgeAudioHeader[5] != input[5];
if (_bridgeAudioBuffered > 0 && (sourceChanged || stateChanged)) _bridgeAudioBuffered = 0;
input[..ProtocolV1.AudioHeaderBytes].CopyTo(_bridgeAudioHeader);
input[ProtocolV1.AudioHeaderBytes..].CopyTo(_bridgeAudioBuffer.AsSpan(_bridgeAudioBuffered));
_bridgeAudioBuffered += payloadBytes;
var accepted = true;
while (_bridgeAudioBuffered >= NetworkAudioPayloadBytes)
{
var normalized = new byte[ProtocolV1.AudioHeaderBytes + NetworkAudioPayloadBytes];
_bridgeAudioHeader.CopyTo(normalized, 0);
BinaryPrimitives.WriteUInt32LittleEndian(normalized.AsSpan(8, 4), _bridgeAudioSequence++);
BinaryPrimitives.WriteUInt64LittleEndian(normalized.AsSpan(12, 8), (ulong)DateTimeOffset.UtcNow.ToUnixTimeMilliseconds() * 1000UL);
SessionId.TryWriteBytes(normalized.AsSpan(20, 16), bigEndian: true, out _);
BinaryPrimitives.WriteUInt32LittleEndian(normalized.AsSpan(60, 4), NetworkAudioPayloadBytes);
_bridgeAudioBuffer.AsSpan(0, NetworkAudioPayloadBytes).CopyTo(normalized.AsSpan(ProtocolV1.AudioHeaderBytes));
_bridgeAudioBuffered -= NetworkAudioPayloadBytes;
if (_bridgeAudioBuffered > 0)
_bridgeAudioBuffer.AsSpan(NetworkAudioPayloadBytes, _bridgeAudioBuffered).CopyTo(_bridgeAudioBuffer);
accepted &= QueueEncodedAudio(normalized);
}
return accepted;
}
}
public async Task SendAsync(string type, object payload, Guid? sessionId, CancellationToken cancellationToken)
{

20
plugins/lumi_transcription/backend/companion/gateway.js
View File

@ -2,6 +2,9 @@ const { WebSocketServer, WebSocket } = require("ws");
const { MAX_JSON_BYTES, MAX_AUDIO_PAYLOAD_BYTES, AUDIO_HEADER_BYTES, parseEnvelope, validateHello, parseAudioFrame, envelope } = require("./protocol");
const { insecureDeviceAllowed } = require("./device_store");
const MAX_CONTROL_MESSAGES_PER_SECOND = 120;
const MAX_AUDIO_MESSAGES_PER_SECOND = 200;
class CompanionGateway {
constructor(options) {
this.devices = options.devices;
@ -27,7 +30,9 @@ class CompanionGateway {
let helloComplete = false;
let lastPong = Date.now();
let windowStarted = Date.now();
let messagesInWindow = 0;
let controlMessagesInWindow = 0;
let audioMessagesInWindow = 0;
let audioMessagesDropped = 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)));
@ -42,15 +47,20 @@ class CompanionGateway {
const body = Buffer.from(data);
messageChain = messageChain.then(async () => {
try {
if (Date.now() - windowStarted >= 1000) { windowStarted = Date.now(); messagesInWindow = 0; }
messagesInWindow += 1;
if (messagesInWindow > 300) throw coded("RATE_LIMIT", "Companion message rate exceeded its limit.");
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 (isBinary) {
if (!helloComplete || !session) throw coded("HELLO_REQUIRED", "Complete the handshake before sending audio.");
audioMessagesInWindow += 1;
if (audioMessagesInWindow > MAX_AUDIO_MESSAGES_PER_SECOND) { audioMessagesDropped += 1; return; }
const result = await this.sessions.audio(session.id, parseAudioFrame(body));
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) {
const hello = validateHello(message);
@ -121,4 +131,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 };
module.exports = { CompanionGateway, sameHostOrigin, MAX_CONTROL_MESSAGES_PER_SECOND, MAX_AUDIO_MESSAGES_PER_SECOND };

16
plugins/lumi_transcription/companion_manifest.json
View File

@ -1,17 +1,17 @@
{
"schema_version": 1,
"version": "0.1.0-experimental.8",
"version": "0.1.0-experimental.9",
"signed": false,
"release_notes": "Fixes OBS integration package discovery in the installed single-file app, validates the packaged payload directly, and displays actionable installation errors.",
"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.",
"installer": {
"id": "windows-x64-installer",
"platform": "win32",
"architecture": "x64",
"label": "Windows x64 per-user installer",
"filename": "Lumi.Companion-Setup.exe",
"url": "https://git.rolfsvaag.no/Rolfsvaag_Datateknikk/Lumi/releases/download/companion-v0.1.0-experimental.8/Lumi.Companion-Setup.exe",
"sha256": "075fb4dd9a84ab10655d75899a6b4944269c9fe6371573778740d80b16ca19d0",
"bytes": 32041200
"url": "https://git.rolfsvaag.no/Rolfsvaag_Datateknikk/Lumi/releases/download/companion-v0.1.0-experimental.9/Lumi.Companion-Setup.exe",
"sha256": "22bdb0f351e5b4960e2c1ae40e3bdbd8de8a8d6326b30714defb5eb202c0efcf",
"bytes": 32039974
},
"artifacts": [
{
@ -20,9 +20,9 @@
"architecture": "x64",
"label": "Windows x64 self-contained",
"filename": "Lumi.Companion-win-x64.zip",
"url": "https://git.rolfsvaag.no/Rolfsvaag_Datateknikk/Lumi/releases/download/companion-v0.1.0-experimental.8/Lumi.Companion-win-x64.zip",
"sha256": "46c58b044a213e328fa96fe86e48b9ea0b67e327439fed14fc485c711f03eb3e",
"bytes": 41757425,
"url": "https://git.rolfsvaag.no/Rolfsvaag_Datateknikk/Lumi/releases/download/companion-v0.1.0-experimental.9/Lumi.Companion-win-x64.zip",
"sha256": "70d9a780859824aa7b8a12e9f25ae8871778b83458889237f171484b2a200dbc",
"bytes": 41757969,
"entrypoint": "Lumi.Companion.App.exe"
}
]

2
plugins/lumi_transcription/plugin.json
View File

@ -1,7 +1,7 @@
{
"id": "lumi_transcription",
"name": "Lumi Transcription",
"version": "0.1.0-experimental.8",
"version": "0.1.0-experimental.9",
"description": "Server-hosted whisper.cpp transcription for Lumi Companion and OBS closed captions.",
"main": "index.js",
"channel": "experimental",

10
plugins/lumi_transcription/tests/verify.js
View File

@ -117,6 +117,7 @@ function verifyLocalhostTransportPolicy() {
}
function verifyCompanionVersionOrdering() {
assert.equal(plugin.compareVersions("0.1.0-experimental.9", "0.1.0-experimental.8"), 1);
assert.equal(plugin.compareVersions("0.1.0-experimental.8", "0.1.0-experimental.7"), 1);
assert.equal(plugin.compareVersions("0.1.0-experimental.7", "0.1.0-experimental.6"), 1);
assert.equal(plugin.compareVersions("0.1.0-experimental.6", "0.1.0-experimental.5"), 1);
@ -357,10 +358,11 @@ async function verifyAuthenticatedGateway() {
const devices = new DeviceStore(db);
const sessionId = crypto.randomUUID();
let disconnected = false;
let audioMessages = 0;
const sessions = {
create: (_device, send) => ({ session: { id: sessionId, state: "idle", send }, resumed: false }),
disconnect: () => { disconnected = true; },
audio: async () => ({ accepted: true }), updateSource: () => ({}), updateObsState: async () => ({}), start: async () => ({}), stop: async () => ({})
audio: async () => { audioMessages += 1; return { accepted: true }; }, updateSource: () => ({}), updateObsState: async () => ({}), start: async () => ({}), stop: async () => ({})
};
const gateway = new CompanionGateway({ devices, sessions });
const registry = createWebUpgradeRegistry();
@ -379,6 +381,12 @@ async function verifyAuthenticatedGateway() {
assert.equal(response.type, "hello_ack");
assert.equal(response.session_id, sessionId);
assert.equal(devices.list().find((device) => device.id === credential.device_id).metadata.companion_version, "0.1.0-experimental.6");
const sourceUuid = crypto.randomUUID();
const audioFrame = protocol.encodeAudioFrame({ session_id: sessionId, source_uuid: sourceUuid, sequence: 1, capture_timestamp_us: Date.now() * 1000, pcm: Buffer.alloc(640) });
for (let index = 0; index < 350; index += 1) client.send(audioFrame);
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");
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);