openvibers/openvibe.live Commits
When Live ends a stream itself (dead air, a lost heartbeat), it closes the stream's WHIP sessions too (#335)
Changes
@@@ -1501,6 +1501,9 @@ async function start() {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, reason) {++ // A WHIP client can stay connected while sending nothing: close its session first, so the stream ends with++ // this reason and nothing keeps it alive.++ try { await require('./streaming/whip-handler').endSessionsForStream(stream.id, reason); } catch (e) { console.warn(`[WHIP] closing stream ${stream.id}'s sessions failed:`, e.message); }await db.endStream(stream.id, reason);try { await db.computeAndCacheStreamAnalytics(stream.id); } catch {}// Auto-finalize any active VOD recording for this stream (via OpenVibe.Media)@@@ -148,6 +148,19 @@ function findSessionsByStreamId(streamId) {const _WHIP_KEEPALIVE_INTERVAL_MS = 30000;++/**++ * Close every WHIP session of a stream that Live is ending for its own reasons (dead air, a lost heartbeat): the++ * producers and transport go, the keepalive stops touching the stream's heartbeat, and the client learns it is over++ * when its connection fails. The stream itself is the caller's to end, with its reason (no endActiveWhipStream here).++ * 2026-10-23: stream 2718's client stayed connected and silent for 9 hours; ending only the row would have left its++ * session in place.++ */++async function endSessionsForStream(streamId, reason = 'stream_ended') {++ const found = findSessionsByStreamId(streamId);++ for (const { resourceId } of found) await cleanupSession(resourceId, { endStreamIfNoActiveSessions: false, reason });++ return found.length;++}++async function cleanupExistingSessionsForStream(streamId) {for (const { resourceId } of findSessionsByStreamId(streamId)) {console.log(`[WHIP] Cleaning existing WHIP session ${resourceId} for stream ${streamId}`);@@@ -1201,6 +1214,7 @@ module.exports = {hasActiveSessionsForStream,sessions,cleanupSession,++ endSessionsForStream,available: !!sdpTransform,_extractRtpParameters: extractRtpParameters,_extractDtlsParameters: extractDtlsParameters,@@@ -11,7 +11,8 @@ const path = require('path');// Stub that single lookup instead of building a schema. Answering "no such stream" ends// endActiveWhipStream on its first line, which is the path this test wants anyway — what// it asserts is that the producer, transport and session are released.--require('../server/db/database').getStreamById = () => null;++let streamLookups = 0;++require('../server/db/database').getStreamById = () => { streamLookups++; return null; };const webrtcSFU = require('../server/streaming/webrtc-sfu');const whipHandler = require('../server/streaming/whip-handler');@@@ -50,4 +51,25 @@ assert.strictEqual(room.transports.has('whip-test-transport-1'), false, 'transpoassert.strictEqual(producerClosed.value, true, 'producer should be closed');assert.strictEqual(transportClosed.value, true, 'transport should be closed');--console.log('✅ WHIP cleanup session regression test passed');++// Live ending a stream itself (dead air, a lost heartbeat) closes that stream's sessions, and only those, without++// ending the stream again from here: the caller ends it with its own reason.++(async () => {++ const mk = (sid, id) => {++ const r = webrtcSFU.rooms.get(`stream-${sid}`) || { producers: new Map(), transports: new Map() };++ webrtcSFU.rooms.set(`stream-${sid}`, r);++ r.producers.set(`p-${id}`, { producer: { close() {} }, peerId: `whip-${id}`, transportId: `t-${id}` });++ r.transports.set(`whip-${id}-t-${id}`, { close() {} });++ whipHandler.sessions.set(id, { streamId: sid, roomId: `stream-${sid}`, peerId: `whip-${id}`, transportId: `t-${id}`, producerIds: [`p-${id}`], heartbeatTimer: setInterval(() => {}, 60000) });++ };++ mk(7, 'r7a'); mk(7, 'r7b'); mk(8, 'r8');++ await new Promise((r) => setImmediate(r));++ const before = streamLookups;++ assert.strictEqual(await whipHandler.endSessionsForStream(7, 'dead_air'), 2);++ assert.ok(!whipHandler.sessions.has('r7a') && !whipHandler.sessions.has('r7b'), 'stream 7\'s sessions are closed');++ assert.ok(whipHandler.sessions.has('r8'), 'another stream\'s session stays');++ assert.strictEqual(webrtcSFU.rooms.get('stream-7').transports.size, 0, 'their transports are closed');++ assert.strictEqual(streamLookups, before, 'the stream is the caller\'s to end');++ assert.strictEqual(await whipHandler.endSessionsForStream(9, 'dead_air'), 0, 'a stream with no WHIP session: nothing to do');++ await whipHandler.cleanupSession('r8');++ console.log('✅ WHIP cleanup session regression test passed');++})().catch((err) => { console.error(err); process.exit(1); });
@@ -1501,6 +1501,9 @@ async function start() {
@@ -1501,6 +1501,9 @@ async function start() {
const deadAir = require('./streaming/dead-air').createDeadAir({ limitMs: Math.max(5, Number(process.env.LIVE_DEAD_AIR_MINUTES) || 30) * 60 * 1000 });
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. */
/** Everything a stream that is gone needs: ended, its recording finalized, its bridges, restreams and room closed. */
async function endGoneStream(stream, reason) {
async function endGoneStream(stream, reason) {
+ // A WHIP client can stay connected while sending nothing: close its session first, so the stream ends with
+ // this reason and nothing keeps it alive.
+ try { await require('./streaming/whip-handler').endSessionsForStream(stream.id, reason); } catch (e) { console.warn(`[WHIP] closing stream ${stream.id}'s sessions failed:`, e.message); }
await db.endStream(stream.id, reason);
await db.endStream(stream.id, reason);
try { await db.computeAndCacheStreamAnalytics(stream.id); } catch {}
try { await db.computeAndCacheStreamAnalytics(stream.id); } catch {}
// Auto-finalize any active VOD recording for this stream (via OpenVibe.Media)
// Auto-finalize any active VOD recording for this stream (via OpenVibe.Media)
@@ -148,6 +148,19 @@ function findSessionsByStreamId(streamId) {
@@ -148,6 +148,19 @@ function findSessionsByStreamId(streamId) {
const _WHIP_KEEPALIVE_INTERVAL_MS = 30000;
const _WHIP_KEEPALIVE_INTERVAL_MS = 30000;
+/**
+ * Close every WHIP session of a stream that Live is ending for its own reasons (dead air, a lost heartbeat): the
+ * producers and transport go, the keepalive stops touching the stream's heartbeat, and the client learns it is over
+ * when its connection fails. The stream itself is the caller's to end, with its reason (no endActiveWhipStream here).
+ * 2026-10-23: stream 2718's client stayed connected and silent for 9 hours; ending only the row would have left its
+ * session in place.
+ */
+async function endSessionsForStream(streamId, reason = 'stream_ended') {
+ const found = findSessionsByStreamId(streamId);
+ for (const { resourceId } of found) await cleanupSession(resourceId, { endStreamIfNoActiveSessions: false, reason });
+ return found.length;
+}
+
async function cleanupExistingSessionsForStream(streamId) {
async function cleanupExistingSessionsForStream(streamId) {
for (const { resourceId } of findSessionsByStreamId(streamId)) {
for (const { resourceId } of findSessionsByStreamId(streamId)) {
console.log(`[WHIP] Cleaning existing WHIP session ${resourceId} for stream ${streamId}`);
console.log(`[WHIP] Cleaning existing WHIP session ${resourceId} for stream ${streamId}`);
@@ -1201,6 +1214,7 @@ module.exports = {
@@ -1201,6 +1214,7 @@ module.exports = {
hasActiveSessionsForStream,
hasActiveSessionsForStream,
sessions,
sessions,
cleanupSession,
cleanupSession,
+ endSessionsForStream,
available: !!sdpTransform,
available: !!sdpTransform,
_extractRtpParameters: extractRtpParameters,
_extractRtpParameters: extractRtpParameters,
_extractDtlsParameters: extractDtlsParameters,
_extractDtlsParameters: extractDtlsParameters,
@@ -11,7 +11,8 @@ const path = require('path');
@@ -11,7 +11,8 @@ const path = require('path');
// Stub that single lookup instead of building a schema. Answering "no such stream" ends
// Stub that single lookup instead of building a schema. Answering "no such stream" ends
// endActiveWhipStream on its first line, which is the path this test wants anyway — what
// endActiveWhipStream on its first line, which is the path this test wants anyway — what
// it asserts is that the producer, transport and session are released.
// it asserts is that the producer, transport and session are released.
-require('../server/db/database').getStreamById = () => null;
+let streamLookups = 0;
+require('../server/db/database').getStreamById = () => { streamLookups++; return null; };
const webrtcSFU = require('../server/streaming/webrtc-sfu');
const webrtcSFU = require('../server/streaming/webrtc-sfu');
const whipHandler = require('../server/streaming/whip-handler');
const whipHandler = require('../server/streaming/whip-handler');
@@ -50,4 +51,25 @@ assert.strictEqual(room.transports.has('whip-test-transport-1'), false, 'transpo
@@ -50,4 +51,25 @@ assert.strictEqual(room.transports.has('whip-test-transport-1'), false, 'transpo
assert.strictEqual(producerClosed.value, true, 'producer should be closed');
assert.strictEqual(producerClosed.value, true, 'producer should be closed');
assert.strictEqual(transportClosed.value, true, 'transport should be closed');
assert.strictEqual(transportClosed.value, true, 'transport should be closed');
-console.log('✅ WHIP cleanup session regression test passed');
+// Live ending a stream itself (dead air, a lost heartbeat) closes that stream's sessions, and only those, without
+// ending the stream again from here: the caller ends it with its own reason.
+(async () => {
+ const mk = (sid, id) => {
+ const r = webrtcSFU.rooms.get(`stream-${sid}`) || { producers: new Map(), transports: new Map() };
+ webrtcSFU.rooms.set(`stream-${sid}`, r);
+ r.producers.set(`p-${id}`, { producer: { close() {} }, peerId: `whip-${id}`, transportId: `t-${id}` });
+ r.transports.set(`whip-${id}-t-${id}`, { close() {} });
+ whipHandler.sessions.set(id, { streamId: sid, roomId: `stream-${sid}`, peerId: `whip-${id}`, transportId: `t-${id}`, producerIds: [`p-${id}`], heartbeatTimer: setInterval(() => {}, 60000) });
+ };
+ mk(7, 'r7a'); mk(7, 'r7b'); mk(8, 'r8');
+ await new Promise((r) => setImmediate(r));
+ const before = streamLookups;
+ assert.strictEqual(await whipHandler.endSessionsForStream(7, 'dead_air'), 2);
+ assert.ok(!whipHandler.sessions.has('r7a') && !whipHandler.sessions.has('r7b'), 'stream 7\'s sessions are closed');
+ assert.ok(whipHandler.sessions.has('r8'), 'another stream\'s session stays');
+ assert.strictEqual(webrtcSFU.rooms.get('stream-7').transports.size, 0, 'their transports are closed');
+ assert.strictEqual(streamLookups, before, 'the stream is the caller\'s to end');
+ assert.strictEqual(await whipHandler.endSessionsForStream(9, 'dead_air'), 0, 'a stream with no WHIP session: nothing to do');
+ await whipHandler.cleanupSession('r8');
+ console.log('✅ WHIP cleanup session regression test passed');
+})().catch((err) => { console.error(err); process.exit(1); });