openvibers/openvibe.live Commits

No floating promises left in Live (plan T0, the shared checker reports 0): the control socket awaits handleCommand so a rejection is caught and logged; the Kick and YouTube chat-relay connects stay fire-and-forget but log their rejections; the calls into functions that catch their own errors are marked '// floating-ok: <why>'. A regression test drives both fixed paths with rejecting dependencies (logged, never unhandled). npm test 152/153, 1 skipped (verified).

OP OpenVibers committed 29 Sept 2026 · 6e7ed605cb · 8 files, +136 −14


Changes

ddiff --git a/server/chat/chat-remote.js b/server/chat/chat-remote.js
iindex 78097250..f2be6b81 100644
---- a/server/chat/chat-remote.js
++++ b/server/chat/chat-remote.js
@@@ -269,7 +269,7 @@ function create(ChatServer) {
ensureOutbox();
replayOutbox();
schedule();
-- pollPresence();
++ pollPresence(); // floating-ok: pollPresence catches and keeps the last snapshot
presenceTimer = setInterval(pollPresence, PRESENCE_MS);
if (presenceTimer.unref) presenceTimer.unref();
console.log(`[Chat] CHAT_AUTHORITY=chat — chat runs in OpenVibe.Chat (${CHAT_URL}); Live forwards to it`);
ddiff --git a/server/chat/chat-server.js b/server/chat/chat-server.js
iindex 844dd8ee..fa77769f 100644
---- a/server/chat/chat-server.js
++++ b/server/chat/chat-server.js
@@@ -943,6 +943,7 @@ class ChatServer {
// Trigger server-side TTS synthesis (async, non-blocking).
// identityKey uses the immutable login handle (or anon id) so the
// per-user voice is stable even if the display name changes.
++ // floating-ok: synthesizeAndBroadcastTTS catches and logs its own errors (never rejects)
this.synthesizeAndBroadcastTTS(
client.streamId,
username,
@@@ -955,6 +956,7 @@ class ChatServer {
);
// Check for 101soundboards links in the message (async, non-blocking)
++ // floating-ok: processSoundboard catches and logs its own errors (never rejects)
this.processSoundboard(ws, client, text);
// Let AI chat viewers react to REAL typed chat (streamer or viewers).
@@@ -984,6 +986,7 @@ class ChatServer {
// synthesize TTS so their "TTS on my channel even when offline" option has
// audio to play (client-side settings + the one-speaking-tab lock decide
// whether it's actually audible). Owner absent = skip the synth cost.
++ // floating-ok: synthesizeAndBroadcastTTS catches and logs its own errors (never rejects)
this.synthesizeAndBroadcastTTS(
null,
username,
@@@ -1095,6 +1098,7 @@ class ChatServer {
this.sendTo(ws, { type: 'system', message: 'Usage: !sb <sound-id or 101soundboards URL> [100p|-100p] [0.5-3 speed]' });
return;
}
++ // floating-ok: processSoundboard catches and logs its own errors (never rejects)
this.processSoundboard(ws, client, text);
return;
}
@@@ -1457,6 +1461,7 @@ class ChatServer {
this.broadcastToStream(client.streamId, ttsMsg);
// Also synthesize server-side TTS for site-wide mode
++ // floating-ok: synthesizeAndBroadcastTTS catches and logs its own errors (never rejects)
this.synthesizeAndBroadcastTTS(
client.streamId,
ttsMsg.username,
ddiff --git a/server/controls/control-server.js b/server/controls/control-server.js
iindex db715935..4c95fed2 100644
---- a/server/controls/control-server.js
++++ b/server/controls/control-server.js
@@@ -157,17 +157,22 @@ class ControlServer {
ws.send(JSON.stringify({ type: 'hardware_status', connected: hardwareConnected }));
}
-- ws.on('message', (data) => {
++ ws.on('message', async (data) => {
++ let msg;
++ try { msg = JSON.parse(data.toString()); } catch { return; } // malformed frame: ignore
try {
-- const msg = JSON.parse(data.toString());
if (msg.type === 'command') {
-- this.handleCommand(ws, msg);
++ // Awaited: handleCommand is async, so without this a rejection (a bad camera row,
++ // a closed socket) escaped the try as an unhandled rejection nobody could see.
++ await this.handleCommand(ws, msg);
} else if (msg.type === 'key_down' || msg.type === 'key_up') {
this.handleKeyEvent(ws, msg);
} else if (msg.type === 'video_click') {
this.handleVideoClick(ws, msg);
}
-- } catch { /* ignore */ }
++ } catch (err) {
++ console.warn('[Control] Message handling failed:', err && err.message);
++ }
});
ws.on('close', () => {
ddiff --git a/server/integrations/chat-relay-service.js b/server/integrations/chat-relay-service.js
iindex 08573961..73ffeec2 100644
---- a/server/integrations/chat-relay-service.js
++++ b/server/integrations/chat-relay-service.js
@@@ -179,10 +179,12 @@ class ChatRelayService {
this._connectTwitch(bridge);
break;
case 'kick':
-- this._connectKick(bridge);
++ // Fire-and-forget by design (the stream is already live; the relay must not block
++ // go-live), but a rejection is logged rather than left as an unhandled rejection.
++ this._connectKick(bridge).catch((err) => console.warn(`[ChatRelay] Kick bridge failed for stream ${bridge.streamId}:`, err && err.message));
break;
case 'youtube':
-- this._connectYouTube(bridge);
++ this._connectYouTube(bridge).catch((err) => console.warn(`[ChatRelay] YouTube bridge failed for stream ${bridge.streamId}:`, err && err.message));
break;
default:
console.warn(`[ChatRelay] Unsupported platform: ${parsed.platform}`);
@@@ -892,7 +894,7 @@ class ChatRelayService {
if (!bridge.stopped) bridge.pollTimer = setTimeout(poll, 6000);
}
};
-- poll();
++ poll(); // floating-ok: the poll loop catches its own errors and reschedules itself
return true;
}
@@@ -975,7 +977,7 @@ class ChatRelayService {
bridge._ytFailures = 0; // Reset on successful connection
// Start polling
-- this._pollYouTubeChat(bridge, continuationToken);
++ this._pollYouTubeChat(bridge, continuationToken); // floating-ok: _pollYouTubeChat catches its own errors and reconnects
} catch (err) {
bridge._ytFailures = (bridge._ytFailures || 0) + 1;
ddiff --git a/server/integrations/powerchat-platform.js b/server/integrations/powerchat-platform.js
iindex 6ac9d8e5..b6f36fea 100644
---- a/server/integrations/powerchat-platform.js
++++ b/server/integrations/powerchat-platform.js
@@@ -436,7 +436,7 @@ function startViewerCountSweeper() {
if (!s.user_id) continue;
liveOwners.set(s.user_id, (liveOwners.get(s.user_id) || 0) + totalViewersForStream(s));
}
-- for (const [userId, count] of liveOwners) { seenLive.add(userId); sendViewCount(userId, count); }
++ for (const [userId, count] of liveOwners) { seenLive.add(userId); sendViewCount(userId, count); } // floating-ok: sendViewCount catches and notes its own errors
// Owners that were live last tick but aren't now → send null (stream ended).
for (const userId of Array.from(seenLive)) {
if (!liveOwners.has(userId)) {
ddiff --git a/server/news/sources/base-source.js b/server/news/sources/base-source.js
iindex 193cb019..61dfe5de 100644
---- a/server/news/sources/base-source.js
++++ b/server/news/sources/base-source.js
@@@ -26,6 +26,7 @@ class BaseNewsSource {
start(onNews) {
this._onNews = onNews;
this._timer = setInterval(() => this._doPoll(), this.pollIntervalMs);
++ // floating-ok: _doPoll catches and logs its own errors (never rejects)
this._doPoll(); // initial fetch
}
ddiff --git a/server/streaming/broadcast-server.js b/server/streaming/broadcast-server.js
iindex 73e07953..431dee80 100644
---- a/server/streaming/broadcast-server.js
++++ b/server/streaming/broadcast-server.js
@@@ -448,25 +448,25 @@ class BroadcastServer extends EventEmitter {
case 'sfu-get-capabilities':
if (client.role === 'broadcaster') {
console.log(`[SFU Signaling] stream ${client.streamId}: get-capabilities`);
-- this._handleSfuGetCapabilities(ws, client);
++ this._handleSfuGetCapabilities(ws, client); // floating-ok: the handler catches and reports its own errors (never rejects)
}
break;
case 'sfu-create-transport':
if (client.role === 'broadcaster') {
console.log(`[SFU Signaling] stream ${client.streamId}: create-transport`);
-- this._handleSfuCreateTransport(ws, client);
++ this._handleSfuCreateTransport(ws, client); // floating-ok: the handler catches and reports its own errors (never rejects)
}
break;
case 'sfu-connect-transport':
if (client.role === 'broadcaster') {
console.log(`[SFU Signaling] stream ${client.streamId}: connect-transport`);
-- this._handleSfuConnectTransport(ws, client, msg);
++ this._handleSfuConnectTransport(ws, client, msg); // floating-ok: the handler catches and reports its own errors (never rejects)
}
break;
case 'sfu-produce':
if (client.role === 'broadcaster') {
console.log(`[SFU Signaling] stream ${client.streamId}: produce (${msg.kind})`);
-- this._handleSfuProduce(ws, client, msg);
++ this._handleSfuProduce(ws, client, msg); // floating-ok: the handler catches and reports its own errors (never rejects)
}
break;
case 'sfu-stop-produce':
ddiff --git a/test/floating-promise-fixes.test.js b/test/floating-promise-fixes.test.js
nnew file mode 100644
iindex 00000000..12153d5f
---- /dev/null
++++ b/test/floating-promise-fixes.test.js
@@@ -0,0 +1,109 @@
++'use strict';
++
++// Plan T0 regressions: the async calls whose result nobody awaited, fixed in the two places where
++// the call site — not the callee — had to change.
++//
++// 1. The control socket's `message` handler now AWAITS handleCommand(), so a rejection is caught
++// and logged by the handler instead of escaping as an unhandled rejection. (The other handlers
++// on that socket, key events and video clicks, are synchronous and stay as they are.)
++// 2. A chat-relay bridge that fails to connect (Kick/YouTube) is still fire-and-forget — the
++// stream is already live and the relay must not block go-live — but its rejection is now
++// logged at the call site.
++//
++// Both modules are loaded with their DB / auth / peer-service dependencies stubbed so the test
++// exercises the promise plumbing only: a rejecting dependency, never a live socket.
++
++const assert = require('assert');
++const path = require('path');
++
++function stub(modPath, exportsObj) {
++ const full = require.resolve(modPath);
++ require.cache[full] = { id: full, filename: full, loaded: true, exports: exportsObj };
++}
++
++stub(path.join(__dirname, '../server/db/database'), {});
++stub(path.join(__dirname, '../server/auth/auth'), { authenticateWs: () => null, extractWsToken: () => null });
++stub(path.join(__dirname, '../server/chat/chat-server'), { _broadcastMessage: () => {} });
++
++const controlServer = require('../server/controls/control-server');
++const chatRelay = require('../server/integrations/chat-relay-service');
++
++const tick = () => new Promise((resolve) => setImmediate(resolve));
++
++/** Run `fn` with console.warn captured and every unhandled rejection recorded. */
++async function capture(fn) {
++ const warns = [];
++ const unhandled = [];
++ const originalWarn = console.warn;
++ const onUnhandled = (err) => unhandled.push(err);
++ console.warn = (...args) => warns.push(args.map(String).join(' '));
++ process.on('unhandledRejection', onUnhandled);
++ try {
++ await fn();
++ await tick();
++ await tick();
++ } finally {
++ process.removeListener('unhandledRejection', onUnhandled);
++ console.warn = originalWarn;
++ }
++ return { warns, unhandled };
++}
++
++(async () => {
++ // ── 1. control socket: a rejecting command handler ──────────────────────
++ const listeners = [];
++ const ws = {
++ readyState: 1,
++ on: (event, fn) => { if (event === 'message') listeners.push(fn); },
++ send: () => {},
++ close: () => {},
++ };
++ controlServer.handleViewerConnection(ws, null, new URLSearchParams(''));
++ assert.strictEqual(listeners.length, 1, 'the viewer control socket registers exactly one message handler');
++
++ const realHandleCommand = controlServer.handleCommand;
++ controlServer.handleCommand = async () => { throw new Error('camera exploded'); };
++
++ let control;
++ try {
++ control = await capture(() => listeners[0](JSON.stringify({ type: 'command', control_id: 1 })));
++ } finally {
++ controlServer.handleCommand = realHandleCommand;
++ }
++ assert.deepStrictEqual(control.unhandled, [], 'an awaited command failure is not an unhandled rejection');
++ assert.ok(control.warns.some((w) => w.includes('camera exploded')),
++ `the command failure is logged, got ${JSON.stringify(control.warns)}`);
++
++ // The synchronous branches still run and a malformed frame is still ignored.
++ let keyEvents = 0;
++ controlServer.handleKeyEvent = () => { keyEvents++; };
++ listeners[0]('not json at all');
++ listeners[0](JSON.stringify({ type: 'key_down', key: 'w' }));
++ assert.strictEqual(keyEvents, 1, 'key events still reach their handler');
++ delete controlServer.handleKeyEvent;
++
++ // ── 2. chat relay: a bridge whose connect() rejects ─────────────────────
++ const realKick = chatRelay._connectKick;
++ const realYouTube = chatRelay._connectYouTube;
++ chatRelay._connectKick = async () => { throw new Error('kick ws refused'); };
++ chatRelay._connectYouTube = async () => { throw new Error('youtube api refused'); };
++
++ let relay;
++ try {
++ relay = await capture(() => {
++ chatRelay.startBridge(101, { id: 7, name: 'kick example', channel_url: 'https://kick.com/example', enabled: 1, chat_relay: 1 });
++ chatRelay.startBridge(102, { id: 8, name: 'yt example', channel_url: 'https://youtube.com/live/abcdefghijk', enabled: 1, chat_relay: 1 });
++ });
++ } finally {
++ chatRelay._connectKick = realKick;
++ chatRelay._connectYouTube = realYouTube;
++ for (const key of Array.from(chatRelay.bridges.keys())) chatRelay.bridges.delete(key);
++ }
++ assert.deepStrictEqual(relay.unhandled, [], 'a failed bridge connect is not an unhandled rejection');
++ assert.ok(relay.warns.some((w) => w.includes('kick ws refused')),
++ `the Kick connect failure is logged, got ${JSON.stringify(relay.warns)}`);
++ assert.ok(relay.warns.some((w) => w.includes('youtube api refused')),
++ `the YouTube connect failure is logged, got ${JSON.stringify(relay.warns)}`);
++
++ console.log('floating-promise-fixes: control command + chat-relay bridge rejections are logged, not unhandled');
++})().catch((err) => { console.error(err); process.exit(1); });