Restream: one restart and one FFmpeg per session (an audio producer appearing queued two restarts, the first FFmpeg was orphaned holding the stream key, and Twitch refused goosely's restream as a second publisher for 13 minutes) (#267)

ddiff --git a/server/streaming/restream-manager.js b/server/streaming/restream-manager.js
iindex 8c4aa22e..271fe2d0 100644
---- a/server/streaming/restream-manager.js
++++ b/server/streaming/restream-manager.js
@@@ -14,7 +14,7 @@
   *   
   *   For RTMP input:   ffmpeg -i http://localhost:9935/live/{key}.flv -c copy -f flv rtmp://dest
   *   For JSMPEG input:  ffmpeg -f mpegts -i pipe:0 -c:v libx264 -preset ultrafast ... -f flv rtmp://dest
-- *   For WebRTC input:  ffmpeg -protocol_whitelist file,rtp,udp -i /tmp/openvibe-restream-{key}.sdp ... -f flv rtmp://dest
++ *   For WebRTC input:  ffmpeg -protocol_whitelist file,rtp,udp -i /tmp/openvibe-restream-{key}-{run}.sdp ... -f flv rtmp://dest
   */
  const { spawn } = require('child_process');
  const EventEmitter = require('events');
@@@ -622,7 +622,9 @@ class RestreamManager extends EventEmitter {
          }
  
          // Step 4: Build SDP file
--        const sdpPath = path.join(os.tmpdir(), `openvibe-restream-${session.key.replace(':', '-')}.sdp`);
++        // A file per run: a previous run's cleanup must never delete the file this run is about to read.
++        this._sdpSeq = (this._sdpSeq || 0) + 1;
++        const sdpPath = path.join(os.tmpdir(), `openvibe-restream-${session.key.replace(':', '-')}-${this._sdpSeq}.sdp`);
          const sdpContent = this._buildSdp(videoConsumer, audioConsumer, videoRtpPort, audioRtpPort);
          fs.writeFileSync(sdpPath, sdpContent, 'utf8');
          console.log(`[Restream] WebRTC: SDP written to ${sdpPath}`);
@@@ -667,15 +669,15 @@ class RestreamManager extends EventEmitter {
  
          this._spawnFFmpeg(session, args);
  
--        // On FFmpeg close, clean up SFU consumers + SDP
++        // On FFmpeg close, clean up this run's SFU consumers + SDP (never a newer run's).
++        const runState = session.webrtcState;
          session.dataTapCleanup = () => {
--            if (session.webrtcState) {
--                const { roomId: rid, videoTransportId, audioTransportId, sdpPath: sdp } = session.webrtcState;
--                if (videoTransportId) webrtcSFU.closePlainConsumer(rid, videoTransportId);
--                if (audioTransportId) webrtcSFU.closePlainConsumer(rid, audioTransportId);
--                try { fs.unlinkSync(sdp); } catch {}
--                session.webrtcState = null;
--            }
++            if (!runState) return;
++            const { roomId: rid, videoTransportId, audioTransportId, sdpPath: sdp } = runState;
++            if (videoTransportId) webrtcSFU.closePlainConsumer(rid, videoTransportId);
++            if (audioTransportId) webrtcSFU.closePlainConsumer(rid, audioTransportId);
++            try { fs.unlinkSync(sdp); } catch {}
++            if (session.webrtcState === runState) session.webrtcState = null;
          };
      }
  
@@@ -883,6 +885,15 @@ class RestreamManager extends EventEmitter {
       * to avoid GnuTLS TLS session invalidation issues.
       */
      _spawnFFmpeg(session, args) {
++        // One FFmpeg per session. A second on the same stream key is refused by the platform as a second publisher,
++        // and the first, orphaned, kept the key (10-19: Twitch refused goosely's restream for 13 minutes).
++        if (session.process) {
++            console.warn(`[Restream] ${session.key}: an FFmpeg was still attached at spawn — stopping it first`);
++            const old = session.process;
++            session.process = null;
++            try { old.kill('SIGTERM'); } catch { /* */ }
++            setTimeout(() => { try { old.kill('SIGKILL'); } catch { /* */ } }, 5000).unref?.();
++        }
          this._disposeTransientSessionState(session);
  
          // Use OpenSSL FFmpeg binary for RTMPS destinations (GnuTLS has TLS rekeying issues)
@@@ -987,8 +998,10 @@ class RestreamManager extends EventEmitter {
          });
  
          proc.on('error', (err) => {
++            const current = session.process === proc;
              this._disposeTransientSessionState(session, proc);
              console.error(`[Restream] FFmpeg spawn error for ${session.key}:`, err.message);
++            if (!current) return; // killed on purpose or replaced: whoever did that owns the restart
              session.lastError = err.message;
              session.status = 'error';
              this.emit('status-change', {
@@@ -999,8 +1012,16 @@ class RestreamManager extends EventEmitter {
          });
  
          proc.on('close', (code) => {
++            // A run killed on purpose (_killProcess clears session.process first) or replaced by a newer one is not
++            // this session's failure: whoever stopped it scheduled what comes next. Restarting here too queued a second
++            // restart, and two FFmpeg ended up on one stream key.
++            const current = session.process === proc;
              this._disposeTransientSessionState(session, proc);
              if (session.status === 'stopped') return; // intentional stop
++            if (!current) {
++                console.log(`[Restream] FFmpeg exited for ${session.key} (stopped for a restart): code=${code}`);
++                return;
++            }
  
              const duration = Date.now() - (session.startedAt || Date.now());
              // ffmpeg names the output URL in its errors ("Error opening output rtmps://…/app/<key>"): keys never reach a log.
@@@ -1044,6 +1065,8 @@ class RestreamManager extends EventEmitter {
       */
      async _scheduleRestart(session) {
          if (session.status === 'stopped') return;
++        // One pending restart per session: a second timer would start a second FFmpeg alongside the first.
++        if (session.restartTimer) return;
  
          // Stop this session. Only persist a destination-level cooldown when the destination is
          // genuinely dead (never once accepted our stream this session) — otherwise a healthy but
ddiff --git a/test/restream-one-restart.test.js b/test/restream-one-restart.test.js
nnew file mode 100644
iindex 00000000..bec50098
---- /dev/null
++++ b/test/restream-one-restart.test.js
@@@ -0,0 +1,86 @@
++/**
++ * One FFmpeg per restream session (production, 10-19). An audio producer appearing restarted goosely's Twitch
++ * restream: the restart and the killed run's exit handler each queued a restart, both fired, the first FFmpeg was
++ * orphaned holding the stream key, and Twitch refused every later attempt as a second publisher for 13 minutes.
++ *
++ *   node test/restream-one-restart.test.js
++ */
++'use strict';
++const assert = require('assert');
++const { EventEmitter } = require('events');
++const childProcess = require('child_process');
++
++// A fake ffmpeg: killing it closes it on the next tick, like a real one.
++const spawned = [];
++childProcess.spawn = () => {
++    const proc = new EventEmitter();
++    proc.stdout = new EventEmitter(); proc.stdout.destroy = () => {};
++    proc.stderr = new EventEmitter(); proc.stderr.destroy = () => {};
++    proc.stdin = { end() {}, destroy() {} };
++    proc.killed = false;
++    proc.kill = () => { if (!proc.killed) { proc.killed = true; setImmediate(() => proc.emit('close', null)); } return true; };
++    spawned.push(proc);
++    return proc;
++};
++
++process.env.NODE_ENV = 'test';
++const logs = [];
++console.log = (...a) => logs.push(a.join(' '));
++console.warn = (...a) => logs.push(a.join(' '));
++console.error = () => {};
++
++const manager = require('../server/streaming/restream-manager');
++const scheduled = () => logs.filter((l) => l.includes('Scheduling restart for 7:1')).length;
++const tick = () => new Promise((r) => setImmediate(r));
++
++function session() {
++    return {
++        key: '7:1', streamId: 7, destId: 1, destination: { platform: 'twitch' },
++        streamInfo: { protocol: 'webrtc' }, webrtcState: { hasAudio: false },
++        status: 'live', restartAttempts: 0, restartDelay: 10_000, restartTimer: null, stableTimer: null,
++        process: null, dataTapCleanup: null,
++    };
++}
++
++(async () => {
++    // An audio producer appears: one restart queued, and the killed run's exit adds none.
++    const s = session();
++    manager.sessions.set(s.key, s);
++    manager._spawnFFmpeg(s, ['-i', 'in.sdp', 'rtmp://ingest.example/app/key']);
++    const first = s.process;
++    assert.ok(first, 'a run is attached');
++    s.status = 'live';
++    manager._handleAudioProducerAdded('stream-7');
++    await tick(); await tick();
++    assert.ok(first.killed, 'the run without audio is stopped');
++    assert.strictEqual(scheduled(), 1, 'exactly one restart is queued (the stopped run\'s exit queues none)');
++    assert.ok(s.restartTimer, 'the restart is pending');
++    assert.ok(logs.some((l) => /exited for 7:1 \(stopped for a restart\)/.test(l)), 'the stopped run\'s exit is logged as such');
++
++    // A second request while one is pending changes nothing.
++    await manager._scheduleRestart(s);
++    assert.strictEqual(scheduled(), 1, 'one pending restart per session');
++
++    // Spawning while a run is somehow still attached stops that run instead of orphaning it.
++    clearTimeout(s.restartTimer); s.restartTimer = null;
++    manager._spawnFFmpeg(s, ['-i', 'in.sdp', 'rtmp://ingest.example/app/key']);
++    const second = s.process;
++    manager._spawnFFmpeg(s, ['-i', 'in.sdp', 'rtmp://ingest.example/app/key']);
++    assert.ok(second.killed, 'the attached run was stopped before the new one started');
++    assert.notStrictEqual(s.process, second);
++    await tick(); await tick();
++    assert.strictEqual(scheduled(), 1, 'the replaced run\'s exit queues no restart either');
++
++    // A run that dies on its own still restarts.
++    const third = s.process;
++    s.status = 'live';
++    third.emit('close', 1);
++    await tick();
++    assert.strictEqual(scheduled(), 2, 'a failure of the attached run is restarted');
++
++    s.status = 'stopped';
++    manager._killProcess(s);
++    manager.sessions.delete(s.key);
++    process.stdout.write('✅ restream: one restart and one FFmpeg per session\n');
++    process.exit(0);
++})().catch((err) => { process.stderr.write(`${err.stack}\n`); process.exit(1); });