openvibers/openvibe.live Commits
Live exits 0 when the database drops during shutdown, prints its effective URLs only at boot or on change, and logs RS passthrough flow, keyframe requests and repeated PowerChat relay failures far less often (#297)
Changes
@@@ -65,6 +65,8 @@ function normalizeRtmpHost(host) {return cleaned;}++let _lastUrlLines = ''; // the Effective URLs block last printed++function buildConfig(registryValues) {registryValues = registryValues || {};const getRegistryEntry = (key) => {@@@ -135,12 +137,18 @@ function buildConfig(registryValues) {? mediaSoupEntry.source: 'default';-- console.log('[Config] Effective URLs:');-- console.log('[Config] BASE_URL=', baseUrl, `(${baseUrlSource})`);-- console.log('[Config] WEBRTC_PUBLIC_URL=', webRtcPublicUrl, `(${webRtcSource})`);-- console.log('[Config] WHIP_PUBLIC_URL=', whipPublicUrl, `(${whipSource})`);-- console.log('[Config] MEDIASOUP_ANNOUNCED_IP=', mediasoupAnnouncedIp, `(${mediaSoupSource})`);-- console.log('[Config] OV_NETWORK_URL=', openvibeToolsUrl || '(not set — SSO base will fall back to https://openvibe.network)');++ // The registry refresh rebuilds the config every few minutes; print the block at boot and when it changes only.++ const urlLines = [++ `[Config] BASE_URL= ${baseUrl} (${baseUrlSource})`,++ `[Config] WEBRTC_PUBLIC_URL= ${webRtcPublicUrl} (${webRtcSource})`,++ `[Config] WHIP_PUBLIC_URL= ${whipPublicUrl} (${whipSource})`,++ `[Config] MEDIASOUP_ANNOUNCED_IP= ${mediasoupAnnouncedIp} (${mediaSoupSource})`,++ `[Config] OV_NETWORK_URL= ${openvibeToolsUrl || '(not set — SSO base will fall back to https://openvibe.network)'}`,++ ].join('\n');++ if (urlLines !== _lastUrlLines) {++ console.log(`[Config] Effective URLs${_lastUrlLines ? ' (changed)' : ''}:\n${urlLines}`);++ _lastUrlLines = urlLines;++ }if (process.env.NODE_ENV === 'production' && baseUrl.includes('localhost')) {console.error('[Config] CRITICAL: BASE_URL is localhost in production! CORS will reject all browser requests.');@@@ -14,9 +14,17 @@*/// Prevent sub-service port conflicts from crashing the main HTTP server++// While stopping, a database connection that drops (PgBouncer or PostgreSQL restarting in the same++// systemd wave, as on 2026-10-22) is expected: a client checked out of the pool for a transaction has++// no 'error' listener (pg-pool removes its idle one on checkout), so pg's "Connection terminated++// unexpectedly" arrives here. Log it and let the graceful stop finish and exit 0.++let _shuttingDown = false;++const { isPgConnectionError } = require('./utils/pg-errors');process.on('uncaughtException', (err) => {if (err.code === 'EADDRINUSE') {console.warn(`[Server] Port ${err.port || '?'} already in use — sub-service skipped`);++ } else if (_shuttingDown && isPgConnectionError(err)) {++ console.warn(`[Server] Database connection dropped during shutdown (expected): ${err.message}`);} else {console.error('[Server] Uncaught exception:', err);process.exit(1);@@@ -158,6 +166,7 @@ const stopper = gracefulStop({deadlineExitCode: 1,stop: [() => {++ _shuttingDown = true;console.log('\n[Server] Shutting down...');// Stop advertising readiness immediately: from here on this process is draining, and anything// gating on /api/ready should see that before the socket actually closes.@@@ -1675,6 +1684,7 @@ async function startDrill() {// ── Graceful Shutdown ────────────────────────────────────────function shutdown(signal) {++ _shuttingDown = true;if (drill.enabled) {// Nothing was started but the HTTP server and the two databases.console.log('[Drill] Shutting down');@@@ -306,8 +306,11 @@ async function chatRelayStats(userId) {return st;}--// Relay health: quiet on success, but surface failures at a sane rate.--const _relayState = new Map(); // userId → { failAt, okAfterFail }++// Relay health: quiet on success, but surface failures at a sane rate. The same failure repeats++// for as long as PowerChat refuses the call, so it is logged once an hour per person (with how many++// were held back); a different failure is logged at once.++const RELAY_FAIL_LOG_MS = 60 * 60 * 1000;++const _relayState = new Map(); // userId → { failAt, reason, held }function _relayOk(userId) {const st = _relayState.get(userId);if (st && st.failAt) {@@@ -315,14 +318,17 @@ function _relayOk(userId) {_relayState.delete(userId);}}--function _relayFail(userId, reason) {-- const now = Date.now();-- const st = _relayState.get(userId) || { failAt: 0 };-- if (now - st.failAt > 60000) {-- console.warn(`[PowerChat] chat relay failed for user ${userId}: ${reason || 'unknown'}`);-- st.failAt = now;-- _relayState.set(userId, st);++function _relayFail(userId, reason, now = Date.now()) {++ const why = reason || 'unknown';++ const st = _relayState.get(userId) || { failAt: 0, reason: null, held: 0 };++ if (why !== st.reason || now - st.failAt >= RELAY_FAIL_LOG_MS) {++ const held = st.reason === why && st.held ? ` (${st.held} more since the last report)` : '';++ console.warn(`[PowerChat] chat relay failed for user ${userId}: ${why}${held}`);++ _relayState.set(userId, { failAt: now, reason: why, held: 0 });++ return true;}++ st.held++;++ return false;}// ── follows:write ────────────────────────────────────────────@@@ -597,5 +603,5 @@ module.exports = {sendViewCount, startViewerCountSweeper,sendCurrencyRedemption, queueCurrencyEarn, startCurrencyEarnFlusher, sendCustomAlert,startChatVerifier, chatRelayStats,-- _test: { verifyTick: _verifyTick, setNow: (fn) => { _now = fn || (() => Date.now()); }, reset: () => _verify.clear() },++ _test: { relayFail: _relayFail, relayReset: () => _relayState.clear(), verifyTick: _verifyTick, setNow: (fn) => { _now = fn || (() => Date.now()); }, reset: () => _verify.clear() },};@@@ -48,6 +48,9 @@ const MAX_RESTARTS = 12; // consecutive restarts before giving up (aconst STABLE_MS = 60000; // a run that stayed live this long resets the restart counterconst SOURCE_CHECK_MS = 5000; // how often a running session re-checks that its source is live++const DEBUG_MEDIA = process.env.LIVE_DEBUG_MEDIA === '1';++const FLOW_LOG_EVERY = 30; // stats ticks (10 s each) between flow lines when not debugging: 5 minutes++function log(streamId, ...a) { console.log(`[RS Passthrough ${streamId}]`, ...a); }// ── RS HTTP API (robot_page_load → rtc_sfu host/port) ────────────────────@@@ -966,7 +969,9 @@ class RsPassthroughRelay {// where video freezes originate: IN = streamer→relay (our seq-gap detection), RS = relay→RS// (werift's RTCP receiver reports from RobotStreamer). pliRx/firRx = keyframes RS asked for.let lastVIn = 0, lastVOut = 0, lastAIn = 0, lastAOut = 0, lastLostIn = 0, lastKf = 0, lastLateFill = 0;++ let flowTicks = 0;session.statsTimer = setInterval(() => {++ flowTicks++;const s = vSender || {};const rsLostPct = typeof s.remoteFractionLost === 'number' ? ((s.remoteFractionLost / 256) * 100).toFixed(1) : '?';const kbps = (n) => Math.round(n);@@@ -984,7 +989,9 @@ class RsPassthroughRelay {nacks: s.nackCount ?? null, plis: s.pliCount ?? null,stalled_ms: _videoStalledSince ? Date.now() - _videoStalledSince : 0,};-- log(sid, `flow: vIn ${Math.round((stats.vIn - lastVIn) / 10)}/s vOut ${Math.round((stats.vOut - lastVOut) / 10)}/s ` +++ // session.rs (for /status) updates every 10 s; the flow line prints every 10 s only under++ // LIVE_DEBUG_MEDIA=1, else the first one, one every 5 minutes, and every one while video is stalled.++ if (DEBUG_MEDIA || _videoStalledSince || flowTicks % FLOW_LOG_EVERY === 1) log(sid, `flow: vIn ${Math.round((stats.vIn - lastVIn) / 10)}/s vOut ${Math.round((stats.vOut - lastVOut) / 10)}/s ` +`aIn ${Math.round((stats.aIn - lastAIn) / 10)}/s aOut ${Math.round((stats.aOut - lastAOut) / 10)}/s${stalled} ` +`| writeErr v=${stats.vErr} a=${stats.aErr} alienSsrc v=${stats.vAlien} a=${stats.aAlien} | stalls=${stats.ingestStalls} (${Math.round(stats.ingestStallMs / 1000)}s total) ` +`| kf ${stats.kf - lastKf}/10s maxKfGap ${stats.maxKfGap}ms injSPS/PPS=${stats.inj}${INJECT_PARAMS ? '' : '(off)'} gopDmg=${stats.gopDamaged} ` +@@@ -10,6 +10,8 @@const EventEmitter = require('events');const config = require('../config');++const DEBUG_MEDIA = process.env.LIVE_DEBUG_MEDIA === '1'; // per-request keyframe lines (off: one line per failure only)++let mediasoup;try {mediasoup = require('mediasoup');@@@ -363,7 +365,7 @@ class WebRTCSFU extends EventEmitter {const timer = setTimeout(async () => {try {await consumer.requestKeyFrame();-- console.log(`[WebRTC] Recording keyframe request #${attempt} sent for consumer ${consumer.id} in room ${roomId}`);++ if (DEBUG_MEDIA) console.log(`[WebRTC] Recording keyframe request #${attempt} sent for consumer ${consumer.id} in room ${roomId}`);} catch (err) {console.warn(`[WebRTC] Recording keyframe request #${attempt} failed for consumer ${consumer.id} in room ${roomId}:`, err.message);}nnew file mode 100644@@@ -0,0 +1,30 @@++'use strict';++/**++ * Is this error a PostgreSQL / PgBouncer connection going away (not a query failing)?++ * Used by the process-level handler to tell an expected drop during shutdown from a real crash.++ */++++// SQLSTATE class 08 (connection exception) and the 57P0x shutdown codes.++const PG_CONNECTION_CODES = new Set(['08000', '08001', '08003', '08004', '08006', '57P01', '57P02', '57P03']);++const SOCKET_CODES = new Set(['ECONNRESET', 'EPIPE', 'ECONNREFUSED', 'ETIMEDOUT']);++const MESSAGES = [++ /Connection terminated unexpectedly/i,++ /Connection terminated/i,++ /terminating connection/i,++ /Client has encountered a connection error and is not queryable/i,++ /server closed the connection unexpectedly/i,++ /server conn crashed/i, // PgBouncer++ /server shutting down/i, // PgBouncer++];++++function isPgConnectionError(err) {++ if (!err || typeof err !== 'object') return false;++ if (PG_CONNECTION_CODES.has(err.code)) return true;++ const msg = String(err.message || '');++ if (MESSAGES.some((re) => re.test(msg))) return true;++ // A bare socket error counts only when it came out of the pg driver.++ const stack = String(err.stack || '');++ return SOCKET_CODES.has(err.code) && /node_modules[\\/](pg|pg-pool|pg-protocol)[\\/]/.test(stack);++}++++module.exports = { isPgConnectionError };nnew file mode 100644@@@ -0,0 +1,82 @@++// Log hygiene and shutdown: the Effective URLs block prints once per change, the PowerChat relay++// failure is logged once an hour per person while it stays the same, and a pg connection drop is++// recognised (the process handler lets it pass while shutting down).++const assert = require('assert');++++const lines = [];++const origLog = console.log;++const origWarn = console.warn;++console.log = (...a) => { lines.push(a.join(' ')); };++console.warn = (...a) => { lines.push(a.join(' ')); };++const count = (re) => lines.filter((l) => re.test(l)).length;++++(async () => {++ try {++ // ── Effective URLs: printed at load, not again by a refresh that changes nothing ──++ process.env.OV_OAUTH_CLIENT_SECRET = 'live-secret';++ const config = require('../server/config');++ assert.strictEqual(count(/\[Config\] Effective URLs/), 1, 'printed once at load');++ const { tokenBody } = require('./helpers/network-token-stub');++ let base = config.baseUrl;++ const originalFetch = global.fetch;++ global.fetch = async (url) => String(url).endsWith('/oauth/token')++ ? new Response(JSON.stringify(tokenBody()), { status: 200, headers: { 'content-type': 'application/json' } })++ : { ok: true, json: async () => ({ registry: { BASE_URL: { value: base, source: 'bootstrap' } } }) };++ config.openvibeToolsInternalUrl = 'http://127.0.0.1:3100';++ await config.refreshRegistry();++ const afterFirst = count(/\[Config\] Effective URLs/);++ await config.refreshRegistry();++ await config.refreshRegistry();++ assert.strictEqual(count(/\[Config\] Effective URLs/), afterFirst, 'unchanged refreshes print nothing');++ base = 'https://changed.example';++ await config.refreshRegistry();++ assert.strictEqual(count(/\[Config\] Effective URLs/), afterFirst + 1, 'a change prints the block again');++ assert.ok(lines.some((l) => l.includes('BASE_URL= https://changed.example')));++ global.fetch = originalFetch;++++ // ── PowerChat relay failure: once an hour per person while the reason is the same ──++ const pc = require('../server/integrations/powerchat-platform');++ const { relayFail, relayReset } = pc._test;++ relayReset();++ lines.length = 0;++ const t0 = Date.parse('2026-10-22T00:00:00Z');++ assert.strictEqual(relayFail(1, 'HTTP 404', t0), true);++ for (let i = 1; i <= 4; i++) assert.strictEqual(relayFail(1, 'HTTP 404', t0 + i * 15 * 60 * 1000 - 1), false);++ assert.strictEqual(count(/chat relay failed for user 1/), 1);++ assert.strictEqual(relayFail(2, 'HTTP 404', t0), true, 'another person is logged on their own');++ assert.strictEqual(relayFail(1, 'HTTP 500', t0 + 1000), true, 'a different reason is logged at once');++ assert.strictEqual(relayFail(1, 'HTTP 500', t0 + 2000), false);++ assert.strictEqual(relayFail(1, 'HTTP 500', t0 + 1000 + 60 * 60 * 1000), true, 'an hour later it is logged again');++ assert.ok(lines[lines.length - 1].includes('(1 more since the last report)'), lines[lines.length - 1]);++ relayReset();++++ // ── pg connection errors ──++ const { isPgConnectionError } = require('../server/utils/pg-errors');++ assert.strictEqual(isPgConnectionError(new Error('Connection terminated unexpectedly')), true);++ assert.strictEqual(isPgConnectionError(Object.assign(new Error('terminating connection due to administrator command'), { code: '57P01' })), true);++ assert.strictEqual(isPgConnectionError(Object.assign(new Error('x'), { code: '08006' })), true);++ const reset = Object.assign(new Error('read ECONNRESET'), { code: 'ECONNRESET' });++ reset.stack = 'Error: read ECONNRESET\n at Socket.<anonymous> (/opt/x/node_modules/pg/lib/connection.js:1:1)';++ assert.strictEqual(isPgConnectionError(reset), true, 'a socket error from the pg driver');++ const otherReset = Object.assign(new Error('read ECONNRESET'), { code: 'ECONNRESET', stack: 'Error: read ECONNRESET\n at ws.js:1:1' });++ assert.strictEqual(isPgConnectionError(otherReset), false, 'a socket error elsewhere is not');++ assert.strictEqual(isPgConnectionError(new TypeError('x is not a function')), false);++ assert.strictEqual(isPgConnectionError(Object.assign(new Error('duplicate key'), { code: '23505' })), false);++ assert.strictEqual(isPgConnectionError(null), false);++++ // ── the process handler: pg drop while stopping passes, anything else still exits 1 ──++ const src = require('fs').readFileSync(require('path').join(__dirname, '../server/index.js'), 'utf8');++ assert.ok(/else if \(_shuttingDown && isPgConnectionError\(err\)\)/.test(src), 'uncaughtException lets a pg drop pass while shutting down');++ assert.ok(/function shutdown\(signal\) \{\s*_shuttingDown = true;/.test(src), 'shutdown() marks the process as stopping first');++++ console.log = origLog;++ console.warn = origWarn;++ console.log('✅ ops log hygiene test passed');++ process.exit(0);++ } catch (err) {++ console.log = origLog;++ console.warn = origWarn;++ console.error(err);++ process.exit(1);++ }++})();
@@ -65,6 +65,8 @@ function normalizeRtmpHost(host) {
@@ -65,6 +65,8 @@ function normalizeRtmpHost(host) {
return cleaned;
return cleaned;
}
}
+let _lastUrlLines = ''; // the Effective URLs block last printed
+
function buildConfig(registryValues) {
function buildConfig(registryValues) {
registryValues = registryValues || {};
registryValues = registryValues || {};
const getRegistryEntry = (key) => {
const getRegistryEntry = (key) => {
@@ -135,12 +137,18 @@ function buildConfig(registryValues) {
@@ -135,12 +137,18 @@ function buildConfig(registryValues) {
? mediaSoupEntry.source
? mediaSoupEntry.source
: 'default';
: 'default';
- console.log('[Config] Effective URLs:');
- console.log('[Config] BASE_URL=', baseUrl, `(${baseUrlSource})`);
- console.log('[Config] WEBRTC_PUBLIC_URL=', webRtcPublicUrl, `(${webRtcSource})`);
- console.log('[Config] WHIP_PUBLIC_URL=', whipPublicUrl, `(${whipSource})`);
- console.log('[Config] MEDIASOUP_ANNOUNCED_IP=', mediasoupAnnouncedIp, `(${mediaSoupSource})`);
- console.log('[Config] OV_NETWORK_URL=', openvibeToolsUrl || '(not set — SSO base will fall back to https://openvibe.network)');
+ // The registry refresh rebuilds the config every few minutes; print the block at boot and when it changes only.
+ const urlLines = [
+ `[Config] BASE_URL= ${baseUrl} (${baseUrlSource})`,
+ `[Config] WEBRTC_PUBLIC_URL= ${webRtcPublicUrl} (${webRtcSource})`,
+ `[Config] WHIP_PUBLIC_URL= ${whipPublicUrl} (${whipSource})`,
+ `[Config] MEDIASOUP_ANNOUNCED_IP= ${mediasoupAnnouncedIp} (${mediaSoupSource})`,
+ `[Config] OV_NETWORK_URL= ${openvibeToolsUrl || '(not set — SSO base will fall back to https://openvibe.network)'}`,
+ ].join('\n');
+ if (urlLines !== _lastUrlLines) {
+ console.log(`[Config] Effective URLs${_lastUrlLines ? ' (changed)' : ''}:\n${urlLines}`);
+ _lastUrlLines = urlLines;
+ }
if (process.env.NODE_ENV === 'production' && baseUrl.includes('localhost')) {
if (process.env.NODE_ENV === 'production' && baseUrl.includes('localhost')) {
console.error('[Config] CRITICAL: BASE_URL is localhost in production! CORS will reject all browser requests.');
console.error('[Config] CRITICAL: BASE_URL is localhost in production! CORS will reject all browser requests.');
@@ -14,9 +14,17 @@
@@ -14,9 +14,17 @@
*/
*/
// Prevent sub-service port conflicts from crashing the main HTTP server
// Prevent sub-service port conflicts from crashing the main HTTP server
+// While stopping, a database connection that drops (PgBouncer or PostgreSQL restarting in the same
+// systemd wave, as on 2026-10-22) is expected: a client checked out of the pool for a transaction has
+// no 'error' listener (pg-pool removes its idle one on checkout), so pg's "Connection terminated
+// unexpectedly" arrives here. Log it and let the graceful stop finish and exit 0.
+let _shuttingDown = false;
+const { isPgConnectionError } = require('./utils/pg-errors');
process.on('uncaughtException', (err) => {
process.on('uncaughtException', (err) => {
if (err.code === 'EADDRINUSE') {
if (err.code === 'EADDRINUSE') {
console.warn(`[Server] Port ${err.port || '?'} already in use — sub-service skipped`);
console.warn(`[Server] Port ${err.port || '?'} already in use — sub-service skipped`);
+ } else if (_shuttingDown && isPgConnectionError(err)) {
+ console.warn(`[Server] Database connection dropped during shutdown (expected): ${err.message}`);
} else {
} else {
console.error('[Server] Uncaught exception:', err);
console.error('[Server] Uncaught exception:', err);
process.exit(1);
process.exit(1);
@@ -158,6 +166,7 @@ const stopper = gracefulStop({
@@ -158,6 +166,7 @@ const stopper = gracefulStop({
deadlineExitCode: 1,
deadlineExitCode: 1,
stop: [
stop: [
() => {
() => {
+ _shuttingDown = true;
console.log('\n[Server] Shutting down...');
console.log('\n[Server] Shutting down...');
// Stop advertising readiness immediately: from here on this process is draining, and anything
// Stop advertising readiness immediately: from here on this process is draining, and anything
// gating on /api/ready should see that before the socket actually closes.
// gating on /api/ready should see that before the socket actually closes.
@@ -1675,6 +1684,7 @@ async function startDrill() {
@@ -1675,6 +1684,7 @@ async function startDrill() {
// ── Graceful Shutdown ────────────────────────────────────────
// ── Graceful Shutdown ────────────────────────────────────────
function shutdown(signal) {
function shutdown(signal) {
+ _shuttingDown = true;
if (drill.enabled) {
if (drill.enabled) {
// Nothing was started but the HTTP server and the two databases.
// Nothing was started but the HTTP server and the two databases.
console.log('[Drill] Shutting down');
console.log('[Drill] Shutting down');
@@ -306,8 +306,11 @@ async function chatRelayStats(userId) {
@@ -306,8 +306,11 @@ async function chatRelayStats(userId) {
return st;
return st;
}
}
-// Relay health: quiet on success, but surface failures at a sane rate.
-const _relayState = new Map(); // userId → { failAt, okAfterFail }
+// Relay health: quiet on success, but surface failures at a sane rate. The same failure repeats
+// for as long as PowerChat refuses the call, so it is logged once an hour per person (with how many
+// were held back); a different failure is logged at once.
+const RELAY_FAIL_LOG_MS = 60 * 60 * 1000;
+const _relayState = new Map(); // userId → { failAt, reason, held }
function _relayOk(userId) {
function _relayOk(userId) {
const st = _relayState.get(userId);
const st = _relayState.get(userId);
if (st && st.failAt) {
if (st && st.failAt) {
@@ -315,14 +318,17 @@ function _relayOk(userId) {
@@ -315,14 +318,17 @@ function _relayOk(userId) {
_relayState.delete(userId);
_relayState.delete(userId);
}
}
}
}
-function _relayFail(userId, reason) {
- const now = Date.now();
- const st = _relayState.get(userId) || { failAt: 0 };
- if (now - st.failAt > 60000) {
- console.warn(`[PowerChat] chat relay failed for user ${userId}: ${reason || 'unknown'}`);
- st.failAt = now;
- _relayState.set(userId, st);
+function _relayFail(userId, reason, now = Date.now()) {
+ const why = reason || 'unknown';
+ const st = _relayState.get(userId) || { failAt: 0, reason: null, held: 0 };
+ if (why !== st.reason || now - st.failAt >= RELAY_FAIL_LOG_MS) {
+ const held = st.reason === why && st.held ? ` (${st.held} more since the last report)` : '';
+ console.warn(`[PowerChat] chat relay failed for user ${userId}: ${why}${held}`);
+ _relayState.set(userId, { failAt: now, reason: why, held: 0 });
+ return true;
}
}
+ st.held++;
+ return false;
}
}
// ── follows:write ────────────────────────────────────────────
// ── follows:write ────────────────────────────────────────────
@@ -597,5 +603,5 @@ module.exports = {
@@ -597,5 +603,5 @@ module.exports = {
sendViewCount, startViewerCountSweeper,
sendViewCount, startViewerCountSweeper,
sendCurrencyRedemption, queueCurrencyEarn, startCurrencyEarnFlusher, sendCustomAlert,
sendCurrencyRedemption, queueCurrencyEarn, startCurrencyEarnFlusher, sendCustomAlert,
startChatVerifier, chatRelayStats,
startChatVerifier, chatRelayStats,
- _test: { verifyTick: _verifyTick, setNow: (fn) => { _now = fn || (() => Date.now()); }, reset: () => _verify.clear() },
+ _test: { relayFail: _relayFail, relayReset: () => _relayState.clear(), verifyTick: _verifyTick, setNow: (fn) => { _now = fn || (() => Date.now()); }, reset: () => _verify.clear() },
};
};
@@ -48,6 +48,9 @@ const MAX_RESTARTS = 12; // consecutive restarts before giving up (a
@@ -48,6 +48,9 @@ const MAX_RESTARTS = 12; // consecutive restarts before giving up (a
const STABLE_MS = 60000; // a run that stayed live this long resets the restart counter
const STABLE_MS = 60000; // a run that stayed live this long resets the restart counter
const SOURCE_CHECK_MS = 5000; // how often a running session re-checks that its source is live
const SOURCE_CHECK_MS = 5000; // how often a running session re-checks that its source is live
+const DEBUG_MEDIA = process.env.LIVE_DEBUG_MEDIA === '1';
+const FLOW_LOG_EVERY = 30; // stats ticks (10 s each) between flow lines when not debugging: 5 minutes
+
function log(streamId, ...a) { console.log(`[RS Passthrough ${streamId}]`, ...a); }
function log(streamId, ...a) { console.log(`[RS Passthrough ${streamId}]`, ...a); }
// ── RS HTTP API (robot_page_load → rtc_sfu host/port) ────────────────────
// ── RS HTTP API (robot_page_load → rtc_sfu host/port) ────────────────────
@@ -966,7 +969,9 @@ class RsPassthroughRelay {
@@ -966,7 +969,9 @@ class RsPassthroughRelay {
// where video freezes originate: IN = streamer→relay (our seq-gap detection), RS = relay→RS
// where video freezes originate: IN = streamer→relay (our seq-gap detection), RS = relay→RS
// (werift's RTCP receiver reports from RobotStreamer). pliRx/firRx = keyframes RS asked for.
// (werift's RTCP receiver reports from RobotStreamer). pliRx/firRx = keyframes RS asked for.
let lastVIn = 0, lastVOut = 0, lastAIn = 0, lastAOut = 0, lastLostIn = 0, lastKf = 0, lastLateFill = 0;
let lastVIn = 0, lastVOut = 0, lastAIn = 0, lastAOut = 0, lastLostIn = 0, lastKf = 0, lastLateFill = 0;
+ let flowTicks = 0;
session.statsTimer = setInterval(() => {
session.statsTimer = setInterval(() => {
+ flowTicks++;
const s = vSender || {};
const s = vSender || {};
const rsLostPct = typeof s.remoteFractionLost === 'number' ? ((s.remoteFractionLost / 256) * 100).toFixed(1) : '?';
const rsLostPct = typeof s.remoteFractionLost === 'number' ? ((s.remoteFractionLost / 256) * 100).toFixed(1) : '?';
const kbps = (n) => Math.round(n);
const kbps = (n) => Math.round(n);
@@ -984,7 +989,9 @@ class RsPassthroughRelay {
@@ -984,7 +989,9 @@ class RsPassthroughRelay {
nacks: s.nackCount ?? null, plis: s.pliCount ?? null,
nacks: s.nackCount ?? null, plis: s.pliCount ?? null,
stalled_ms: _videoStalledSince ? Date.now() - _videoStalledSince : 0,
stalled_ms: _videoStalledSince ? Date.now() - _videoStalledSince : 0,
};
};
- log(sid, `flow: vIn ${Math.round((stats.vIn - lastVIn) / 10)}/s vOut ${Math.round((stats.vOut - lastVOut) / 10)}/s ` +
+ // session.rs (for /status) updates every 10 s; the flow line prints every 10 s only under
+ // LIVE_DEBUG_MEDIA=1, else the first one, one every 5 minutes, and every one while video is stalled.
+ if (DEBUG_MEDIA || _videoStalledSince || flowTicks % FLOW_LOG_EVERY === 1) log(sid, `flow: vIn ${Math.round((stats.vIn - lastVIn) / 10)}/s vOut ${Math.round((stats.vOut - lastVOut) / 10)}/s ` +
`aIn ${Math.round((stats.aIn - lastAIn) / 10)}/s aOut ${Math.round((stats.aOut - lastAOut) / 10)}/s${stalled} ` +
`aIn ${Math.round((stats.aIn - lastAIn) / 10)}/s aOut ${Math.round((stats.aOut - lastAOut) / 10)}/s${stalled} ` +
`| writeErr v=${stats.vErr} a=${stats.aErr} alienSsrc v=${stats.vAlien} a=${stats.aAlien} | stalls=${stats.ingestStalls} (${Math.round(stats.ingestStallMs / 1000)}s total) ` +
`| writeErr v=${stats.vErr} a=${stats.aErr} alienSsrc v=${stats.vAlien} a=${stats.aAlien} | stalls=${stats.ingestStalls} (${Math.round(stats.ingestStallMs / 1000)}s total) ` +
`| kf ${stats.kf - lastKf}/10s maxKfGap ${stats.maxKfGap}ms injSPS/PPS=${stats.inj}${INJECT_PARAMS ? '' : '(off)'} gopDmg=${stats.gopDamaged} ` +
`| kf ${stats.kf - lastKf}/10s maxKfGap ${stats.maxKfGap}ms injSPS/PPS=${stats.inj}${INJECT_PARAMS ? '' : '(off)'} gopDmg=${stats.gopDamaged} ` +
@@ -10,6 +10,8 @@
@@ -10,6 +10,8 @@
const EventEmitter = require('events');
const EventEmitter = require('events');
const config = require('../config');
const config = require('../config');
+const DEBUG_MEDIA = process.env.LIVE_DEBUG_MEDIA === '1'; // per-request keyframe lines (off: one line per failure only)
+
let mediasoup;
let mediasoup;
try {
try {
mediasoup = require('mediasoup');
mediasoup = require('mediasoup');
@@ -363,7 +365,7 @@ class WebRTCSFU extends EventEmitter {
@@ -363,7 +365,7 @@ class WebRTCSFU extends EventEmitter {
const timer = setTimeout(async () => {
const timer = setTimeout(async () => {
try {
try {
await consumer.requestKeyFrame();
await consumer.requestKeyFrame();
- console.log(`[WebRTC] Recording keyframe request #${attempt} sent for consumer ${consumer.id} in room ${roomId}`);
+ if (DEBUG_MEDIA) console.log(`[WebRTC] Recording keyframe request #${attempt} sent for consumer ${consumer.id} in room ${roomId}`);
} catch (err) {
} catch (err) {
console.warn(`[WebRTC] Recording keyframe request #${attempt} failed for consumer ${consumer.id} in room ${roomId}:`, err.message);
console.warn(`[WebRTC] Recording keyframe request #${attempt} failed for consumer ${consumer.id} in room ${roomId}:`, err.message);
}
}
new file mode 100644
new file mode 100644
@@ -0,0 +1,30 @@
@@ -0,0 +1,30 @@
+'use strict';
+/**
+ * Is this error a PostgreSQL / PgBouncer connection going away (not a query failing)?
+ * Used by the process-level handler to tell an expected drop during shutdown from a real crash.
+ */
+
+// SQLSTATE class 08 (connection exception) and the 57P0x shutdown codes.
+const PG_CONNECTION_CODES = new Set(['08000', '08001', '08003', '08004', '08006', '57P01', '57P02', '57P03']);
+const SOCKET_CODES = new Set(['ECONNRESET', 'EPIPE', 'ECONNREFUSED', 'ETIMEDOUT']);
+const MESSAGES = [
+ /Connection terminated unexpectedly/i,
+ /Connection terminated/i,
+ /terminating connection/i,
+ /Client has encountered a connection error and is not queryable/i,
+ /server closed the connection unexpectedly/i,
+ /server conn crashed/i, // PgBouncer
+ /server shutting down/i, // PgBouncer
+];
+
+function isPgConnectionError(err) {
+ if (!err || typeof err !== 'object') return false;
+ if (PG_CONNECTION_CODES.has(err.code)) return true;
+ const msg = String(err.message || '');
+ if (MESSAGES.some((re) => re.test(msg))) return true;
+ // A bare socket error counts only when it came out of the pg driver.
+ const stack = String(err.stack || '');
+ return SOCKET_CODES.has(err.code) && /node_modules[\\/](pg|pg-pool|pg-protocol)[\\/]/.test(stack);
+}
+
+module.exports = { isPgConnectionError };
new file mode 100644
new file mode 100644
@@ -0,0 +1,82 @@
@@ -0,0 +1,82 @@
+// Log hygiene and shutdown: the Effective URLs block prints once per change, the PowerChat relay
+// failure is logged once an hour per person while it stays the same, and a pg connection drop is
+// recognised (the process handler lets it pass while shutting down).
+const assert = require('assert');
+
+const lines = [];
+const origLog = console.log;
+const origWarn = console.warn;
+console.log = (...a) => { lines.push(a.join(' ')); };
+console.warn = (...a) => { lines.push(a.join(' ')); };
+const count = (re) => lines.filter((l) => re.test(l)).length;
+
+(async () => {
+ try {
+ // ── Effective URLs: printed at load, not again by a refresh that changes nothing ──
+ process.env.OV_OAUTH_CLIENT_SECRET = 'live-secret';
+ const config = require('../server/config');
+ assert.strictEqual(count(/\[Config\] Effective URLs/), 1, 'printed once at load');
+ const { tokenBody } = require('./helpers/network-token-stub');
+ let base = config.baseUrl;
+ const originalFetch = global.fetch;
+ global.fetch = async (url) => String(url).endsWith('/oauth/token')
+ ? new Response(JSON.stringify(tokenBody()), { status: 200, headers: { 'content-type': 'application/json' } })
+ : { ok: true, json: async () => ({ registry: { BASE_URL: { value: base, source: 'bootstrap' } } }) };
+ config.openvibeToolsInternalUrl = 'http://127.0.0.1:3100';
+ await config.refreshRegistry();
+ const afterFirst = count(/\[Config\] Effective URLs/);
+ await config.refreshRegistry();
+ await config.refreshRegistry();
+ assert.strictEqual(count(/\[Config\] Effective URLs/), afterFirst, 'unchanged refreshes print nothing');
+ base = 'https://changed.example';
+ await config.refreshRegistry();
+ assert.strictEqual(count(/\[Config\] Effective URLs/), afterFirst + 1, 'a change prints the block again');
+ assert.ok(lines.some((l) => l.includes('BASE_URL= https://changed.example')));
+ global.fetch = originalFetch;
+
+ // ── PowerChat relay failure: once an hour per person while the reason is the same ──
+ const pc = require('../server/integrations/powerchat-platform');
+ const { relayFail, relayReset } = pc._test;
+ relayReset();
+ lines.length = 0;
+ const t0 = Date.parse('2026-10-22T00:00:00Z');
+ assert.strictEqual(relayFail(1, 'HTTP 404', t0), true);
+ for (let i = 1; i <= 4; i++) assert.strictEqual(relayFail(1, 'HTTP 404', t0 + i * 15 * 60 * 1000 - 1), false);
+ assert.strictEqual(count(/chat relay failed for user 1/), 1);
+ assert.strictEqual(relayFail(2, 'HTTP 404', t0), true, 'another person is logged on their own');
+ assert.strictEqual(relayFail(1, 'HTTP 500', t0 + 1000), true, 'a different reason is logged at once');
+ assert.strictEqual(relayFail(1, 'HTTP 500', t0 + 2000), false);
+ assert.strictEqual(relayFail(1, 'HTTP 500', t0 + 1000 + 60 * 60 * 1000), true, 'an hour later it is logged again');
+ assert.ok(lines[lines.length - 1].includes('(1 more since the last report)'), lines[lines.length - 1]);
+ relayReset();
+
+ // ── pg connection errors ──
+ const { isPgConnectionError } = require('../server/utils/pg-errors');
+ assert.strictEqual(isPgConnectionError(new Error('Connection terminated unexpectedly')), true);
+ assert.strictEqual(isPgConnectionError(Object.assign(new Error('terminating connection due to administrator command'), { code: '57P01' })), true);
+ assert.strictEqual(isPgConnectionError(Object.assign(new Error('x'), { code: '08006' })), true);
+ const reset = Object.assign(new Error('read ECONNRESET'), { code: 'ECONNRESET' });
+ reset.stack = 'Error: read ECONNRESET\n at Socket.<anonymous> (/opt/x/node_modules/pg/lib/connection.js:1:1)';
+ assert.strictEqual(isPgConnectionError(reset), true, 'a socket error from the pg driver');
+ const otherReset = Object.assign(new Error('read ECONNRESET'), { code: 'ECONNRESET', stack: 'Error: read ECONNRESET\n at ws.js:1:1' });
+ assert.strictEqual(isPgConnectionError(otherReset), false, 'a socket error elsewhere is not');
+ assert.strictEqual(isPgConnectionError(new TypeError('x is not a function')), false);
+ assert.strictEqual(isPgConnectionError(Object.assign(new Error('duplicate key'), { code: '23505' })), false);
+ assert.strictEqual(isPgConnectionError(null), false);
+
+ // ── the process handler: pg drop while stopping passes, anything else still exits 1 ──
+ const src = require('fs').readFileSync(require('path').join(__dirname, '../server/index.js'), 'utf8');
+ assert.ok(/else if \(_shuttingDown && isPgConnectionError\(err\)\)/.test(src), 'uncaughtException lets a pg drop pass while shutting down');
+ assert.ok(/function shutdown\(signal\) \{\s*_shuttingDown = true;/.test(src), 'shutdown() marks the process as stopping first');
+
+ console.log = origLog;
+ console.warn = origWarn;
+ console.log('✅ ops log hygiene test passed');
+ process.exit(0);
+ } catch (err) {
+ console.log = origLog;
+ console.warn = origWarn;
+ console.error(err);
+ process.exit(1);
+ }
+})();