openvibers/openvibe.live Commits

Restream: one start per stream and destination at a time (going live ran auto-start and resume together and spawned two FFmpeg on one stream key, which the platform refuses) (#173)

OP OpenVibers committed 27 days ago · 36069727d6 · 2 files, +72 −0

Changes

ddiff --git a/server/streaming/restream-manager.js b/server/streaming/restream-manager.js
iindex c00596a5..8c4aa22e 100644
---- a/server/streaming/restream-manager.js
++++ b/server/streaming/restream-manager.js
@@@ -151,6 +151,8 @@ class RestreamManager extends EventEmitter {
super();
/** @type {Map<string, RestreamSession>} key: `${streamId}:${destId}` */
this.sessions = new Map();
++ /** @type {Map<string, Promise>} key → the start running for it (startRestream is one at a time per key) */
++ this._starting = new Map();
this._sfuListenerBound = false;
/** @type {Map<number, { count: number|null, fetchedAt: number }>} destId → cached viewer count */
this._viewerCountCache = new Map();
@@@ -265,6 +267,25 @@ class RestreamManager extends EventEmitter {
*/
async startRestream(streamId, destination, streamInfo) {
const key = this._key(streamId, destination.id);
++ // One start per stream and destination at a time. Going live runs autoStartForStream and resumeForStream
++ // together (WHIP resumes too), and each awaits the database before its session exists, so both used to pass
++ // the "already active" check below and spawn two FFmpeg pushing the same stream key; the platform then
++ // refuses one and the other, and the restream looks broken. A second caller now waits for the first start.
++ const running = this._starting.get(key);
++ if (running) {
++ console.log(`[Restream] Start already in progress: ${key}`);
++ return running;
++ }
++ const start = this._startRestream(key, streamId, destination, streamInfo);
++ this._starting.set(key, start);
++ try {
++ return await start;
++ } finally {
++ if (this._starting.get(key) === start) this._starting.delete(key);
++ }
++ }
++
++ async _startRestream(key, streamId, destination, streamInfo) {
// Already running for this exact stream+dest?
if (this.sessions.has(key)) {
ddiff --git a/test/restream-single-start.test.js b/test/restream-single-start.test.js
nnew file mode 100644
iindex 00000000..50b2a11a
---- /dev/null
++++ b/test/restream-single-start.test.js
@@@ -0,0 +1,51 @@
++/**
++ * Going live starts restreams from two paths at once (autoStartForStream and resumeForStream, and WHIP resumes too).
++ * Each awaited the database before its session existed, so both passed startRestream's "already active" check and
++ * spawned two FFmpeg pushing one stream key (production 2026-10-17 02:15: 2615:1 and 2615:2 each spawned twice; the
++ * platform refuses a second publisher). startRestream now runs one start per stream and destination at a time.
++ *
++ * node test/restream-single-start.test.js
++ */
++'use strict';
++const assert = require('assert');
++process.env.NODE_ENV = 'test';
++const quiet = console.log;
++console.log = () => {};
++console.warn = () => {};
++
++const Manager = require('../server/streaming/restream-manager').constructor;
++
++(async () => {
++ const m = new Manager();
++ const dest = { id: 2, platform: 'kick', enabled: 1, auto_start: 1, server_url: 'rtmps://x.kick.com', stream_key: 'k' };
++ m._getDestinationsForStream = async () => [dest];
++ const starts = [];
++ let release;
++ m._startRestream = async (key) => {
++ starts.push(key);
++ await new Promise((r) => { release = r; }); // the database and SFU awaits of a real start
++ const session = { key, streamId: 2615, destId: 2, status: 'starting' };
++ m.sessions.set(key, session);
++ return session;
++ };
++
++ // The go-live race: both paths at once, on the same stream and destination.
++ const a = m.autoStartForStream(2615, 7, { protocol: 'webrtc' });
++ const b = m.resumeForStream(2615, 7, { protocol: 'webrtc' });
++ await new Promise((r) => setImmediate(r));
++ assert.deepStrictEqual(starts, ['2615:2'], 'one start, whichever paths ask');
++ release();
++ await Promise.all([a, b]);
++ assert.deepStrictEqual(starts, ['2615:2']);
++ assert.strictEqual(m._starting.size, 0, 'the in-flight entry is gone once the start finished');
++
++ // A later start (after a stop) goes through, and a start that fails does not block the next one.
++ m.sessions.clear();
++ m._startRestream = async () => { starts.push('fails'); throw new Error('no SFU'); };
++ await assert.rejects(m.startRestream(2615, dest, { protocol: 'webrtc' }), /no SFU/);
++ assert.strictEqual(m._starting.size, 0);
++ m._startRestream = async (key) => { starts.push(key); return { key, status: 'starting' }; };
++ await m.startRestream(2615, dest, { protocol: 'webrtc' });
++ assert.deepStrictEqual(starts, ['2615:2', 'fails', '2615:2']);
++ quiet('restream single start: all checks passed');
++})().catch((err) => { quiet(err); process.exit(1); });