openvibers/openvibe.live Commits
A stream records why it ended (streamer, staff, broadcaster_left, ingest_ended, heartbeat_lost, dead_air) (#327)
Changes
nnew file mode 100644@@@ -0,0 +1,6 @@++-- phase: expand++-- Why a stream ended (plan 2026-10-22, track C2: where first-time streamers drop). A streamer whose first stream ran++-- under a minute either stopped it or lost it; the stream row could not say which. db.endStream(id, reason) records the++-- first reason given: streamer, staff, broadcaster_left, ingest_ended, heartbeat_lost, dead_air. NULL for streams that++-- ended before this column, and for an end path that names none.++ALTER TABLE streams ADD COLUMN IF NOT EXISTS end_reason text;@@@ -333,7 +333,7 @@ router.delete('/streams/:id', async (req, res) => {return res.status(403).json({ error: "You cannot end the owner's stream" });}-- await db.endStream(req.params.id);++ await db.endStream(req.params.id, 'staff');// A force-end must actually take the stream down everywhere, not just flip is_live:// outbound forwarders (RobotStreamer passthrough, ffmpeg restreams, chat relays), the// signaling room and its viewers, the SFU room and any voice channel.@@@ -331,15 +331,21 @@ async function createStream({ user_id, channel_id, managed_stream_id, title, des});}--async function endStream(streamId) {++/** The reasons a stream ends (streams.end_reason, migration 0014). */++const END_REASONS = new Set(['streamer', 'staff', 'broadcaster_left', 'ingest_ended', 'heartbeat_lost', 'dead_air']);++++async function endStream(streamId, reason = null) {return await getDb().tx(async () => {const stream = await get('SELECT started_at, is_live FROM streams WHERE id = ?', [streamId]);if (!stream) return null;++ // The first reason given stays: a stream another path already ended keeps why it ended.++ const why = END_REASONS.has(reason) ? reason : null;const result = await run(`UPDATE streams SET is_live = 0, ended_at = ov_now(),-- duration_seconds = CAST((julianday(ov_now()) - julianday(started_at)) * 86400 AS INTEGER)++ duration_seconds = CAST((julianday(ov_now()) - julianday(started_at)) * 86400 AS INTEGER),++ end_reason = COALESCE(end_reason, ?)WHERE id = ?`,-- [streamId]++ [why, streamId]);if (stream.is_live) await fireStreamLifecycle('ended', streamId);return result;@@@ -1498,8 +1498,8 @@ async function start() {// 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);++ async function endGoneStream(stream, reason) {++ 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)if (!recorder.isFinalizingStream(stream.id)) {@@@ -1543,7 +1543,7 @@ async function start() {continue;}console.log(`[Heartbeat] Ending stale stream ${stream.id} (no heartbeat for 5+ minutes)`);-- await endGoneStream(stream);++ await endGoneStream(stream, 'heartbeat_lost');}// Dead air (server/streaming/dead-air.js): a browser or WHIP stream whose page heartbeats while its@@@ -1553,7 +1553,7 @@ async function start() {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);++ await endGoneStream(stream, 'dead_air');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.`,@@@ -111,7 +111,7 @@ async function apply(event) {[sessionId, known ? known.managed_stream_id : slotRef(p), streamId, state, revision, toSqlTime(p.ended_at)]);if (!streamId) return { outcome: 'recorded' };const row = await db.getStreamById(streamId);-- if (row && row.is_live) await db.endStream(streamId);++ if (row && row.is_live) await db.endStream(streamId, 'ingest_ended');return {outcome: 'ended',after: async () => {@@@ -510,7 +510,7 @@ class BroadcastServer extends EventEmitter {if (currentRoom && !currentRoom.broadcaster) {console.log(`[Broadcast] Broadcaster did not reconnect, ending stream ${client.streamId}`);try {-- await db.endStream(client.streamId);++ await db.endStream(client.streamId, 'broadcaster_left');require('./recorder').finalizeStream(client.streamId).catch((err) => {console.warn(`[Broadcast] Failed to finalize VOD for stale stream ${client.streamId}:`, err.message);});@@@ -2040,7 +2040,7 @@ router.delete('/:id', requireAuth, async (req, res) => {return res.status(403).json({ error: 'Not your stream' });}-- await db.endStream(stream.id);++ await db.endStream(stream.id, stream.user_id === req.user.id ? 'streamer' : 'staff');// Stop server-side recording (RTMP is handled in rtmp-server.js, but JSMPEG needs it here)if (stream.protocol === 'jsmpeg') {@@@ -247,7 +247,7 @@ class RTMPServer extends EventEmitter {info.heartbeatTimer = null;}-- await db.endStream(info.streamId);++ await db.endStream(info.streamId, 'ingest_ended');try { await db.computeAndCacheStreamAnalytics(info.streamId); } catch {}this.activeStreams.delete(streamKey);console.log(`[RTMP] Stream ended: ${maskKey(streamKey)} (stream ${info.streamId})`);@@@ -187,7 +187,7 @@ async function endActiveWhipStream(streamId, reason = 'whip_cleanup') {console.log(`[WHIP] Ending live stream ${streamId} due to WHIP session termination (${reason})`);try {-- await db.endStream(streamId);++ await db.endStream(streamId, 'ingest_ended');} catch (err) {console.error(`[WHIP] Failed to end stream ${streamId}:`, err.message);}nnew file mode 100644@@@ -0,0 +1,32 @@++'use strict';++// Why a stream ended (streams.end_reason, migration 0014; plan 2026-10-22 track C2): db.endStream(id, reason) records the++// first known reason, a later end keeps it, and an unknown reason records nothing.++const assert = require('assert');++const db = require('../server/db/database');++++(async () => {++ await db.initDb();++ const user = (await db.createUser({ username: 'endreason', email: 'endreason@x', password_hash: 'x', display_name: 'endreason', stream_key: 'endreason'.padEnd(32, '0') })).lastInsertRowid++ || (await db.get('SELECT id FROM users WHERE username = ?', ['endreason'])).id;++ const mk = async () => Number((await db.run(`INSERT INTO streams (user_id, title, protocol, is_live, started_at) VALUES (?, 't', 'webrtc', 1, ov_now()) RETURNING id`, [user])).lastInsertRowid);++ const reasonOf = async (id) => (await db.get('SELECT end_reason, is_live FROM streams WHERE id = ?', [id]));++++ const a = await mk();++ await db.endStream(a, 'broadcaster_left');++ assert.deepStrictEqual(await reasonOf(a), { end_reason: 'broadcaster_left', is_live: 0 });++ await db.endStream(a, 'heartbeat_lost');++ assert.strictEqual((await reasonOf(a)).end_reason, 'broadcaster_left', 'the first reason stays');++++ const b = await mk();++ await db.endStream(b, 'something made up');++ assert.strictEqual((await reasonOf(b)).end_reason, null, 'an unknown reason records nothing');++ await db.endStream(b, 'staff');++ assert.strictEqual((await reasonOf(b)).end_reason, 'staff', 'a known reason fills an empty one');++++ const c = await mk();++ await db.endStream(c);++ assert.strictEqual((await reasonOf(c)).end_reason, null, 'an end path that names none records none');++++ console.log('stream end reason: first known reason kept');++ process.exit(0);++})().catch((e) => { console.error(e); process.exit(1); });
new file mode 100644
new file mode 100644
@@ -0,0 +1,6 @@
@@ -0,0 +1,6 @@
+-- phase: expand
+-- Why a stream ended (plan 2026-10-22, track C2: where first-time streamers drop). A streamer whose first stream ran
+-- under a minute either stopped it or lost it; the stream row could not say which. db.endStream(id, reason) records the
+-- first reason given: streamer, staff, broadcaster_left, ingest_ended, heartbeat_lost, dead_air. NULL for streams that
+-- ended before this column, and for an end path that names none.
+ALTER TABLE streams ADD COLUMN IF NOT EXISTS end_reason text;
@@ -333,7 +333,7 @@ router.delete('/streams/:id', async (req, res) => {
@@ -333,7 +333,7 @@ router.delete('/streams/:id', async (req, res) => {
return res.status(403).json({ error: "You cannot end the owner's stream" });
return res.status(403).json({ error: "You cannot end the owner's stream" });
}
}
- await db.endStream(req.params.id);
+ await db.endStream(req.params.id, 'staff');
// A force-end must actually take the stream down everywhere, not just flip is_live:
// A force-end must actually take the stream down everywhere, not just flip is_live:
// outbound forwarders (RobotStreamer passthrough, ffmpeg restreams, chat relays), the
// outbound forwarders (RobotStreamer passthrough, ffmpeg restreams, chat relays), the
// signaling room and its viewers, the SFU room and any voice channel.
// signaling room and its viewers, the SFU room and any voice channel.
@@ -331,15 +331,21 @@ async function createStream({ user_id, channel_id, managed_stream_id, title, des
@@ -331,15 +331,21 @@ async function createStream({ user_id, channel_id, managed_stream_id, title, des
});
});
}
}
-async function endStream(streamId) {
+/** The reasons a stream ends (streams.end_reason, migration 0014). */
+const END_REASONS = new Set(['streamer', 'staff', 'broadcaster_left', 'ingest_ended', 'heartbeat_lost', 'dead_air']);
+
+async function endStream(streamId, reason = null) {
return await getDb().tx(async () => {
return await getDb().tx(async () => {
const stream = await get('SELECT started_at, is_live FROM streams WHERE id = ?', [streamId]);
const stream = await get('SELECT started_at, is_live FROM streams WHERE id = ?', [streamId]);
if (!stream) return null;
if (!stream) return null;
+ // The first reason given stays: a stream another path already ended keeps why it ended.
+ const why = END_REASONS.has(reason) ? reason : null;
const result = await run(
const result = await run(
`UPDATE streams SET is_live = 0, ended_at = ov_now(),
`UPDATE streams SET is_live = 0, ended_at = ov_now(),
- duration_seconds = CAST((julianday(ov_now()) - julianday(started_at)) * 86400 AS INTEGER)
+ duration_seconds = CAST((julianday(ov_now()) - julianday(started_at)) * 86400 AS INTEGER),
+ end_reason = COALESCE(end_reason, ?)
WHERE id = ?`,
WHERE id = ?`,
- [streamId]
+ [why, streamId]
);
);
if (stream.is_live) await fireStreamLifecycle('ended', streamId);
if (stream.is_live) await fireStreamLifecycle('ended', streamId);
return result;
return result;
@@ -1498,8 +1498,8 @@ async function start() {
@@ -1498,8 +1498,8 @@ async function start() {
// Browser and WHIP streams with no media for 30 minutes end (server/streaming/dead-air.js).
// 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 });
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) {
- await db.endStream(stream.id);
+ async function endGoneStream(stream, 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)
if (!recorder.isFinalizingStream(stream.id)) {
if (!recorder.isFinalizingStream(stream.id)) {
@@ -1543,7 +1543,7 @@ async function start() {
@@ -1543,7 +1543,7 @@ async function start() {
continue;
continue;
}
}
console.log(`[Heartbeat] Ending stale stream ${stream.id} (no heartbeat for 5+ minutes)`);
console.log(`[Heartbeat] Ending stale stream ${stream.id} (no heartbeat for 5+ minutes)`);
- await endGoneStream(stream);
+ await endGoneStream(stream, 'heartbeat_lost');
}
}
// Dead air (server/streaming/dead-air.js): a browser or WHIP stream whose page heartbeats while its
// Dead air (server/streaming/dead-air.js): a browser or WHIP stream whose page heartbeats while its
@@ -1553,7 +1553,7 @@ async function start() {
@@ -1553,7 +1553,7 @@ async function start() {
for (const stream of browserLive) {
for (const stream of browserLive) {
if (!deadAir.observe(stream.id, await webrtcSFU.producerBytes(`stream-${stream.id}`))) continue;
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)`);
console.log(`[Heartbeat] Ending stream ${stream.id}: no video or audio for ${Math.round(deadAir.limitMs / 60000)} minutes (dead air)`);
- await endGoneStream(stream);
+ await endGoneStream(stream, 'dead_air');
require('./utils/notify').pushNotification({
require('./utils/notify').pushNotification({
user_id: stream.user_id, type: 'STREAM_SUMMARY', category: 'stream', title: 'Your stream was ended',
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.`,
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.`,
@@ -111,7 +111,7 @@ async function apply(event) {
@@ -111,7 +111,7 @@ async function apply(event) {
[sessionId, known ? known.managed_stream_id : slotRef(p), streamId, state, revision, toSqlTime(p.ended_at)]);
[sessionId, known ? known.managed_stream_id : slotRef(p), streamId, state, revision, toSqlTime(p.ended_at)]);
if (!streamId) return { outcome: 'recorded' };
if (!streamId) return { outcome: 'recorded' };
const row = await db.getStreamById(streamId);
const row = await db.getStreamById(streamId);
- if (row && row.is_live) await db.endStream(streamId);
+ if (row && row.is_live) await db.endStream(streamId, 'ingest_ended');
return {
return {
outcome: 'ended',
outcome: 'ended',
after: async () => {
after: async () => {
@@ -510,7 +510,7 @@ class BroadcastServer extends EventEmitter {
@@ -510,7 +510,7 @@ class BroadcastServer extends EventEmitter {
if (currentRoom && !currentRoom.broadcaster) {
if (currentRoom && !currentRoom.broadcaster) {
console.log(`[Broadcast] Broadcaster did not reconnect, ending stream ${client.streamId}`);
console.log(`[Broadcast] Broadcaster did not reconnect, ending stream ${client.streamId}`);
try {
try {
- await db.endStream(client.streamId);
+ await db.endStream(client.streamId, 'broadcaster_left');
require('./recorder').finalizeStream(client.streamId).catch((err) => {
require('./recorder').finalizeStream(client.streamId).catch((err) => {
console.warn(`[Broadcast] Failed to finalize VOD for stale stream ${client.streamId}:`, err.message);
console.warn(`[Broadcast] Failed to finalize VOD for stale stream ${client.streamId}:`, err.message);
});
});
@@ -2040,7 +2040,7 @@ router.delete('/:id', requireAuth, async (req, res) => {
@@ -2040,7 +2040,7 @@ router.delete('/:id', requireAuth, async (req, res) => {
return res.status(403).json({ error: 'Not your stream' });
return res.status(403).json({ error: 'Not your stream' });
}
}
- await db.endStream(stream.id);
+ await db.endStream(stream.id, stream.user_id === req.user.id ? 'streamer' : 'staff');
// Stop server-side recording (RTMP is handled in rtmp-server.js, but JSMPEG needs it here)
// Stop server-side recording (RTMP is handled in rtmp-server.js, but JSMPEG needs it here)
if (stream.protocol === 'jsmpeg') {
if (stream.protocol === 'jsmpeg') {
@@ -247,7 +247,7 @@ class RTMPServer extends EventEmitter {
@@ -247,7 +247,7 @@ class RTMPServer extends EventEmitter {
info.heartbeatTimer = null;
info.heartbeatTimer = null;
}
}
- await db.endStream(info.streamId);
+ await db.endStream(info.streamId, 'ingest_ended');
try { await db.computeAndCacheStreamAnalytics(info.streamId); } catch {}
try { await db.computeAndCacheStreamAnalytics(info.streamId); } catch {}
this.activeStreams.delete(streamKey);
this.activeStreams.delete(streamKey);
console.log(`[RTMP] Stream ended: ${maskKey(streamKey)} (stream ${info.streamId})`);
console.log(`[RTMP] Stream ended: ${maskKey(streamKey)} (stream ${info.streamId})`);
@@ -187,7 +187,7 @@ async function endActiveWhipStream(streamId, reason = 'whip_cleanup') {
@@ -187,7 +187,7 @@ async function endActiveWhipStream(streamId, reason = 'whip_cleanup') {
console.log(`[WHIP] Ending live stream ${streamId} due to WHIP session termination (${reason})`);
console.log(`[WHIP] Ending live stream ${streamId} due to WHIP session termination (${reason})`);
try {
try {
- await db.endStream(streamId);
+ await db.endStream(streamId, 'ingest_ended');
} catch (err) {
} catch (err) {
console.error(`[WHIP] Failed to end stream ${streamId}:`, err.message);
console.error(`[WHIP] Failed to end stream ${streamId}:`, err.message);
}
}
new file mode 100644
new file mode 100644
@@ -0,0 +1,32 @@
@@ -0,0 +1,32 @@
+'use strict';
+// Why a stream ended (streams.end_reason, migration 0014; plan 2026-10-22 track C2): db.endStream(id, reason) records the
+// first known reason, a later end keeps it, and an unknown reason records nothing.
+const assert = require('assert');
+const db = require('../server/db/database');
+
+(async () => {
+ await db.initDb();
+ const user = (await db.createUser({ username: 'endreason', email: 'endreason@x', password_hash: 'x', display_name: 'endreason', stream_key: 'endreason'.padEnd(32, '0') })).lastInsertRowid
+ || (await db.get('SELECT id FROM users WHERE username = ?', ['endreason'])).id;
+ const mk = async () => Number((await db.run(`INSERT INTO streams (user_id, title, protocol, is_live, started_at) VALUES (?, 't', 'webrtc', 1, ov_now()) RETURNING id`, [user])).lastInsertRowid);
+ const reasonOf = async (id) => (await db.get('SELECT end_reason, is_live FROM streams WHERE id = ?', [id]));
+
+ const a = await mk();
+ await db.endStream(a, 'broadcaster_left');
+ assert.deepStrictEqual(await reasonOf(a), { end_reason: 'broadcaster_left', is_live: 0 });
+ await db.endStream(a, 'heartbeat_lost');
+ assert.strictEqual((await reasonOf(a)).end_reason, 'broadcaster_left', 'the first reason stays');
+
+ const b = await mk();
+ await db.endStream(b, 'something made up');
+ assert.strictEqual((await reasonOf(b)).end_reason, null, 'an unknown reason records nothing');
+ await db.endStream(b, 'staff');
+ assert.strictEqual((await reasonOf(b)).end_reason, 'staff', 'a known reason fills an empty one');
+
+ const c = await mk();
+ await db.endStream(c);
+ assert.strictEqual((await reasonOf(c)).end_reason, null, 'an end path that names none records none');
+
+ console.log('stream end reason: first known reason kept');
+ process.exit(0);
+})().catch((e) => { console.error(e); process.exit(1); });