openvibers/openvibe.live Commits

A browser stream with no video or audio for 30 minutes ends (dead air); a long passthrough stall logs every 5 minutes (#323)

OP OpenVibers committed 21 days ago · 3dc4334a3e · 6 files, +140 −26

Changes

ddiff --git a/docs/restreaming.md b/docs/restreaming.md
iindex 25b95a8d..99b8749b 100644
---- a/docs/restreaming.md
++++ b/docs/restreaming.md
@@@ -89,6 +89,7 @@ and RS's own RTCP view of us (`rs.loss_pct`, `rs.rtt_ms`, `rs.nacks`, `rs.plis`)
| OBS stops (RTMP `donePublish`) | `index.js` `rtmpServer.on('unpublish')` | restreams, RS, chat relays, AI bots |
| WHIP session ends | `whip-handler.js` `endActiveWhipStream` | restreams, RS, chat relays, AI bots, room |
| No heartbeat for 5 min | `index.js` stale sweep | RS, chat relays, AI bots, restreams, room |
++| No video or audio for 30 min, page still open (dead air, `LIVE_DEAD_AIR_MINUTES`) | `index.js` stale sweep + `streaming/dead-air.js` (the SFU room's producer bytes, read each minute); the streamer gets a notification | the same as a stale stream |
| Admin force-end | `DELETE /api/admin/streams/:id` | RS, restreams, chat relays, AI bots, room, SFU room, voice channel |
| Process shutdown | `index.js` SIGTERM handler | `rsPassthroughRelay.stopAll()` closes each RS peer so RS's viewers get `consumerClosed` immediately |
ddiff --git a/server/index.js b/server/index.js
iindex fb0e6090..153e9c35 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -1493,6 +1493,35 @@ async function start() {
// 8. Start stale stream heartbeat cleanup (every 60 seconds)
let heartbeatCleanupRunning = false;
++ // Browser and WHIP streams with no media for 30 minutes end (server/streaming/dead-air.js).
++ const deadAir = require('./streaming/dead-air').createDeadAir({ limitMs: Math.max(5, Number(process.env.LIVE_DEAD_AIR_MINUTES) || 30) * 60 * 1000 });
++ /** Everything a stream that is gone needs: ended, its recording finalized, its bridges, restreams and room closed. */
++ async function endGoneStream(stream) {
++ await db.endStream(stream.id);
++ try { await db.computeAndCacheStreamAnalytics(stream.id); } catch {}
++ // Auto-finalize any active VOD recording for this stream (via OpenVibe.Media)
++ if (!recorder.isFinalizingStream(stream.id)) {
++ recorder.finalizeStream(stream.id).catch(err => {
++ console.warn(`[VOD] Auto-finalize failed for stale stream ${stream.id}:`, err.message);
++ });
++ }
++ // Stop RS chat bridge for this stream (prevents zombie bridges)
++ robotStreamerService.stopForStream(stream.id);
++ // Stop chat relay bridges for this stream
++ chatRelayService.stopForStream(stream.id);
++ // Stop AI chatbots for this stream
++ try { require('./integrations/ai-chatbot-service').stopForStream(stream.id); } catch { /* non-critical */ }
++ // Stop any active restreams for this stream
++ restreamManager.stopAllForStream(stream.id);
++ // Close signaling room and notify viewers
++ broadcastServer.endStream(stream.id);
++ const user = await db.getUserById(stream.user_id);
++ if (stream.protocol === 'jsmpeg' && user) {
++ jsmpegRelay.destroyChannel(user.stream_key);
++ } else if (stream.protocol === 'webrtc') {
++ webrtcSFU.closeRoom(`stream-${stream.id}`);
++ }
++ }
const liveVodThumbGeneratedAt = new Map(); // streamId → last Media frame-grab time
const maintenanceInterval = setInterval(async () => {
if (heartbeatCleanupRunning) return;
@@@ -1512,30 +1541,22 @@ async function start() {
continue;
}
console.log(`[Heartbeat] Ending stale stream ${stream.id} (no heartbeat for 5+ minutes)`);
-- await db.endStream(stream.id);
-- try { await db.computeAndCacheStreamAnalytics(stream.id); } catch {}
-- // Auto-finalize any active VOD recording for this stream (via OpenVibe.Media)
-- if (!recorder.isFinalizingStream(stream.id)) {
-- recorder.finalizeStream(stream.id).catch(err => {
-- console.warn(`[VOD] Auto-finalize failed for stale stream ${stream.id}:`, err.message);
-- });
-- }
-- // Stop RS chat bridge for this stream (prevents zombie bridges)
-- robotStreamerService.stopForStream(stream.id);
-- // Stop chat relay bridges for this stream
-- chatRelayService.stopForStream(stream.id);
-- // Stop AI chatbots for this stream
-- try { require('./integrations/ai-chatbot-service').stopForStream(stream.id); } catch { /* non-critical */ }
-- // Stop any active restreams for this stream
-- restreamManager.stopAllForStream(stream.id);
-- // Close signaling room and notify viewers
-- broadcastServer.endStream(stream.id);
-- const user = await db.getUserById(stream.user_id);
-- if (stream.protocol === 'jsmpeg' && user) {
-- jsmpegRelay.destroyChannel(user.stream_key);
-- } else if (stream.protocol === 'webrtc') {
-- webrtcSFU.closeRoom(`stream-${stream.id}`);
-- }
++ await endGoneStream(stream);
++ }
++
++ // Dead air (server/streaming/dead-air.js): a browser or WHIP stream whose page heartbeats while its
++ // producers receive nothing ends like a stale stream, and the streamer is told why.
++ const browserLive = await db.all(`SELECT id, user_id, protocol FROM streams WHERE is_live = 1 AND protocol IN ('webrtc', 'browser', 'screen')`);
++ deadAir.keep(new Set(browserLive.map((x) => x.id)));
++ for (const stream of browserLive) {
++ if (!deadAir.observe(stream.id, await webrtcSFU.producerBytes(`stream-${stream.id}`))) continue;
++ console.log(`[Heartbeat] Ending stream ${stream.id}: no video or audio for ${Math.round(deadAir.limitMs / 60000)} minutes (dead air)`);
++ await endGoneStream(stream);
++ require('./utils/notify').pushNotification({
++ user_id: stream.user_id, type: 'STREAM_SUMMARY', category: 'stream', title: 'Your stream was ended',
++ message: `No video or audio reached OpenVibe for ${Math.round(deadAir.limitMs / 60000)} minutes, so your stream was ended and its recording saved. Go live again whenever you are ready.`,
++ url: 'https://openvibe.live/broadcast',
++ });
}
// Also finalize recordings whose stream already ended (finalize was never called)
ddiff --git a/server/integrations/rs-passthrough-relay.js b/server/integrations/rs-passthrough-relay.js
iindex 0aab16e0..3db93c62 100644
---- a/server/integrations/rs-passthrough-relay.js
++++ b/server/integrations/rs-passthrough-relay.js
@@@ -990,8 +990,11 @@ class RsPassthroughRelay {
stalled_ms: _videoStalledSince ? Date.now() - _videoStalledSince : 0,
};
// 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 ` +
++ // LIVE_DEBUG_MEDIA=1, else the first one, one every 5 minutes, and every one in a stall's first minute (its
++ // diagnosis). A stall that lasts goes back to one every 5 minutes: stream 2718's 2.5-hour stall on
++ // 2026-10-23 printed 2,176 lines.
++ const youngStall = _videoStalledSince && Date.now() - _videoStalledSince < 60000;
++ if (DEBUG_MEDIA || youngStall || 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} ` +
ddiff --git a/server/streaming/dead-air.js b/server/streaming/dead-air.js
nnew file mode 100644
iindex 00000000..cc29fadf
---- /dev/null
++++ b/server/streaming/dead-air.js
@@@ -0,0 +1,28 @@
++'use strict';
++/**
++ * Dead air: a browser or WHIP stream whose page still heartbeats while its producers receive nothing. Stream 2718
++ * (2026-10-23) stayed live 7.5 hours after its camera and microphone stopped: a channel marked live with nothing to
++ * watch, a recording of silence, and every deploy waiting for idle. The heartbeat sweep (server/index.js) reads the
++ * bytes its SFU room's producers have received once a minute; a count that has not changed for `limitMs` ends the
++ * stream like a stale one. A new producer starts from 0, so any change counts as media.
++ *
++ * const dead = createDeadAir({ limitMs });
++ * dead.observe(streamId, bytes) → true once the count has not changed for limitMs (then forgets the stream)
++ * dead.keep(liveIds) forget the streams that are no longer live
++ */
++function createDeadAir({ limitMs = 30 * 60 * 1000, now = Date.now } = {}) {
++ const seen = new Map(); // streamId → { bytes, at }: when the count last changed
++ return {
++ limitMs,
++ observe(id, bytes) {
++ const s = seen.get(id);
++ if (!s || s.bytes !== bytes) { seen.set(id, { bytes, at: now() }); return false; }
++ if (now() - s.at < limitMs) return false;
++ seen.delete(id);
++ return true;
++ },
++ keep(ids) { for (const id of seen.keys()) if (!ids.has(id)) seen.delete(id); },
++ };
++}
++
++module.exports = { createDeadAir };
ddiff --git a/server/streaming/webrtc-sfu.js b/server/streaming/webrtc-sfu.js
iindex fd6add5b..422d4b52 100644
---- a/server/streaming/webrtc-sfu.js
++++ b/server/streaming/webrtc-sfu.js
@@@ -606,6 +606,21 @@ class WebRTCSFU extends EventEmitter {
});
}
++ /**
++ * Bytes every open producer in a room has received (mediasoup Producer.getStats byteCount, summed). A new producer
++ * starts from 0, so a caller compares for a change, not a rise. 0 for a room that does not exist.
++ */
++ async producerBytes(roomId) {
++ const room = this.rooms.get(roomId);
++ if (!room) return 0;
++ let total = 0;
++ for (const { producer } of room.producers.values()) {
++ if (producer.closed) continue;
++ try { for (const st of await producer.getStats()) total += Number(st.byteCount) || 0; } catch { /* closed meanwhile */ }
++ }
++ return total;
++ }
++
/**
* Get viewer count for a room
*/
ddiff --git a/test/dead-air.test.js b/test/dead-air.test.js
nnew file mode 100644
iindex 00000000..2eb1a523
---- /dev/null
++++ b/test/dead-air.test.js
@@@ -0,0 +1,46 @@
++'use strict';
++// Dead air (server/streaming/dead-air.js, 2026-10-23): a stream whose SFU producers receive nothing for the limit ends;
++// any change in the byte count (more media, or a new producer starting from 0) restarts the clock; a stream that is
++// no longer live is forgotten. And the SFU sums its open producers' received bytes (webrtc-sfu.js producerBytes).
++const assert = require('assert');
++const { createDeadAir } = require('../server/streaming/dead-air');
++
++(async () => {
++ let t = 0;
++ const MIN = 60 * 1000;
++ const d = createDeadAir({ limitMs: 30 * MIN, now: () => t });
++ assert.strictEqual(d.observe(1, 0), false, 'first sight starts the clock');
++ t = 29 * MIN;
++ assert.strictEqual(d.observe(1, 0), false, 'under the limit');
++ t = 30 * MIN;
++ assert.strictEqual(d.observe(1, 0), true, 'unchanged for the limit: dead air');
++ assert.strictEqual(d.observe(1, 0), false, 'once said, the stream is forgotten (a new clock if it is still there)');
++
++ // Media keeps the stream alive; a new producer (a lower count) is media too.
++ t = 0;
++ const e = createDeadAir({ limitMs: 30 * MIN, now: () => t });
++ e.observe(2, 1000);
++ t = 20 * MIN; e.observe(2, 5000);
++ t = 45 * MIN; assert.strictEqual(e.observe(2, 5000), false, 'the clock restarted when the count changed at 20 min');
++ t = 46 * MIN; e.observe(2, 120);
++ t = 75 * MIN; assert.strictEqual(e.observe(2, 120), false, 'a new producer counts as media');
++ t = 76 * MIN; assert.strictEqual(e.observe(2, 120), true);
++
++ // A stream that ended is forgotten, so a new stream with its id starts fresh.
++ t = 0;
++ const f = createDeadAir({ limitMs: 30 * MIN, now: () => t });
++ f.observe(3, 0);
++ f.keep(new Set());
++ t = 31 * MIN;
++ assert.strictEqual(f.observe(3, 0), false, 'forgotten when it was no longer live');
++
++ // The SFU's sum over a room's open producers.
++ const sfu = require('../server/streaming/webrtc-sfu');
++ const producer = (bytes, closed = false) => ({ closed, getStats: async () => [{ byteCount: bytes }] });
++ sfu.rooms.set('stream-dead-air-test', { producers: new Map([['a', { producer: producer(700) }], ['b', { producer: producer(300) }], ['c', { producer: producer(5000, true) }]]) });
++ assert.strictEqual(await sfu.producerBytes('stream-dead-air-test'), 1000, 'open producers only');
++ assert.strictEqual(await sfu.producerBytes('stream-none'), 0, 'no room: 0');
++ sfu.rooms.delete('stream-dead-air-test');
++ console.log('dead air: all checks passed');
++ process.exit(0);
++})().catch((e) => { console.error(e); process.exit(1); });