Workers: keep the RTMP worker up on an uncaught exception once started (node-media-server's own behaviour), kill ffmpeg children when the restream worker exits, hand unstarted outputs of a draining generation back; Live patch: events must come from source openre, FLV proxy stops if the viewer left

ddiff --git a/docs/live-patch.diff b/docs/live-patch.diff
iindex a511d6d..b6988b2 100644
---- a/docs/live-patch.diff
++++ b/docs/live-patch.diff
@@@ -273,10 +273,10 @@ index 0000000..865fff1
  +module.exports = { OPENRE_PROTOCOLS, authorityOf, slotById, slotIsOpenre, refusesLiveIngest, serializeSlot, ingestFor, rotateFor, setAuthority };
  diff --git a/server/openre/mirror.js b/server/openre/mirror.js
  new file mode 100644
--index 0000000..365f61d
++index 0000000..3f739cc
  --- /dev/null
  +++ b/server/openre/mirror.js
--@@ -0,0 +1,205 @@
++@@ -0,0 +1,206 @@
  +'use strict';
  +/**
  + * OpenRe sessions → Live's `streams` table (roadmap Wave 7, ADR-009: Live observes sessions through
@@@ -331,6 +331,7 @@ index 0000000..365f61d
  +    const p = event.payload || {};
  +    const sessionId = String(p.session_id || (event.subject && event.subject.id) || '');
  +    const revision = Number(event.subject && event.subject.revision) || 0;
+++    if (event.source !== 'openre') return { outcome: 'ignored' };
  +    if (!/^openre\.session\.(started|ended|failed)$/.test(type) || !/^ses_[0-9A-HJKMNP-TV-Z]{26}$/.test(sessionId)) return { outcome: 'ignored' };
  +    const known = db.get('SELECT * FROM openre_sessions WHERE session_id = ?', [sessionId]);
  +    if (known && known.revision >= revision) return { outcome: 'stale' };
@@@ -463,7 +464,7 @@ index 0000000..365f61d
  +            db.run("UPDATE openre_sessions SET confirmed_at = CURRENT_TIMESTAMP WHERE session_id = ?", [r.session_id]);
  +            if (r.stream_id) db.run('UPDATE streams SET last_heartbeat = CURRENT_TIMESTAMP WHERE id = ? AND is_live = 1', [r.stream_id]);
  +        } else if (['ending', 'ended', 'failed'].includes(s.state)) {
--+            const event = { event_type: `openre.session.${s.state === 'failed' ? 'failed' : 'ended'}`, subject: { type: 'ingest_session', id: r.session_id, revision: Math.max(Number(s.revision) || 0, r.revision + 1) }, payload: { session_id: r.session_id, ended_at: s.ended_at } };
+++            const event = { source: 'openre', event_type: `openre.session.${s.state === 'failed' ? 'failed' : 'ended'}`, subject: { type: 'ingest_session', id: r.session_id, revision: Math.max(Number(s.revision) || 0, r.revision + 1) }, payload: { session_id: r.session_id, ended_at: s.ended_at } };
  +            const out = db.getDb().transaction(() => apply(event))();
  +            if (out.after) out.after();
  +            n++;
@@@ -790,7 +791,7 @@ index 4b5f8e4..4ac081b 100644
           const liveStreams = getLiveStreamsForDestination(dest, req.user.id);
           for (const stream of liveStreams) {
  diff --git a/server/streaming/routes.js b/server/streaming/routes.js
--index 70a0857..105be81 100644
++index 70a0857..2c5d856 100644
  --- a/server/streaming/routes.js
  +++ b/server/streaming/routes.js
  @@ -28,6 +28,8 @@ const { requireAuth, requireStreamer, optionalAuth } = require('../auth/auth');
@@@ -913,7 +914,7 @@ index 70a0857..105be81 100644
           const rtmpKey = stream.managed_stream_key || db.getUserById(stream.user_id)?.stream_key;
           const rtmpServer = require('./rtmp-server');
           const status = rtmpServer.getStatus(rtmpKey);
--@@ -2402,17 +2440,27 @@ router.get('/:id/call', optionalAuth, (req, res) => {
++@@ -2402,18 +2440,29 @@ router.get('/:id/call', optionalAuth, (req, res) => {
   // ── RTMP FLV Proxy ───────────────────────────────────────────
   // Proxies HTTP-FLV from the internal NMS server so the browser fetches
   // from the same HTTPS origin, avoiding CSP / mixed-content issues.
@@@ -945,8 +946,10 @@ index 70a0857..105be81 100644
  +            url = `http://127.0.0.1:${nmsPort}/live/${flvKey}.flv`;
  +        }
   
+++        if (req.destroyed) return undefined; // the viewer left while OpenRe was asked
           const http = require('http');
           const upstream = http.get(url, (nmsRes) => {
++             if (nmsRes.statusCode !== 200) {
  diff --git a/server/streaming/rtmp-server.js b/server/streaming/rtmp-server.js
  index fb8ba5c..709da89 100644
  --- a/server/streaming/rtmp-server.js
ddiff --git a/server/store/outputs.js b/server/store/outputs.js
iindex f945ac1..ed7a396 100644
---- a/server/store/outputs.js
++++ b/server/store/outputs.js
@@@ -33,6 +33,7 @@ function createOutputs({ db, config, events, clock, box, definitions, sessions }
              WHERE o.desired = 'run' AND o.state = 'pending' AND o.worker_id IS NULL AND s.state = 'live' ORDER BY o.created_at`),
          assign: db.prepare("UPDATE outputs SET worker_id = ?, worker_generation = ?, updated_at = ? WHERE id = ? AND worker_id IS NULL AND state = 'pending'"),
          forWorker: db.prepare(`SELECT * FROM outputs WHERE worker_id = ? AND state IN ${OPEN}`),
++        unassign: db.prepare("UPDATE outputs SET worker_id = NULL, worker_generation = NULL, updated_at = ? WHERE id = ? AND state = 'pending'"),
          release: db.prepare(`UPDATE outputs SET worker_id = NULL, worker_generation = NULL, state = 'pending', updated_at = ?
              WHERE worker_id = ? AND state IN ${OPEN} AND desired = 'run'`),
          stopOrphans: db.prepare(`UPDATE outputs SET state = 'stopped', ended_at = ?, updated_at = ?
@@@ -339,6 +340,7 @@ function createOutputs({ db, config, events, clock, box, definitions, sessions }
          assignPending,
          forWorker: (workerId) => q.forWorker.all(workerId),
          release: (workerId) => q.release.run(now(), workerId).changes,
++        unassign: (outputId) => q.unassign.run(now(), outputId).changes,
          stopOrphans: () => q.stopOrphans.run(now(), now()).changes,
          report,
          log,
ddiff --git a/workers/restream-worker.js b/workers/restream-worker.js
iindex 91204a8..a76285b 100644
---- a/workers/restream-worker.js
++++ b/workers/restream-worker.js
@@@ -27,7 +27,7 @@ function createRestreamWorker({ rt, log = console, exit = (code) => process.exit
              onDrainDeadline: () => handOver(),
              onLost: () => stopAll('worker lost'),
              activeCount: () => runners.size,
--            onExit: () => { closing = true; clearTimeout(pollTimer); },
++            onExit: () => { closing = true; clearTimeout(pollTimer); if (runtime.me) store.outputs.release(runtime.me.id); },
          },
      });
  
@@@ -52,7 +52,9 @@ function createRestreamWorker({ rt, log = console, exit = (code) => process.exit
              }
              if (runner && runner.stopped) { runners.delete(row.id); runner = null; continue; }
              if (runner) continue;
--            if (runtime.draining) continue; // a draining generation starts nothing new
++            // A draining generation starts nothing new: an output assigned to it but not started yet
++            // goes back to the coordinator for the newest generation.
++            if (runtime.draining) { store.outputs.unassign(row.id); seen.delete(row.id); continue; }
              const input = inputUrlFor(session);
              if (!input) { store.outputs.report(row.id, { state: 'failed', last_error: `restream from ${session.protocol} sessions is not supported by OpenRe yet`, ended_at: Date.now() }); continue; }
              // Let the ingest settle before pulling (Live waits 3 s for node-media-server's FLV).
@@@ -113,6 +115,11 @@ if (require.main === module) {
      const rt = openRuntime({ config: load() });
      const worker = createRestreamWorker({ rt });
      worker.start();
++    // If this process dies, its ffmpeg children must not keep pushing unsupervised: the outputs are
++    // reassigned to another worker, and two pushes to one ingest make platforms drop both.
++    process.on('exit', () => {
++        for (const r of worker.runners().values()) { try { if (r.proc) r.proc.kill('SIGKILL'); } catch { /* */ } }
++    });
      for (const sig of ['SIGTERM', 'SIGINT']) process.on(sig, () => worker.drain(sig));
  }
  
ddiff --git a/workers/rtmp-ingest.js b/workers/rtmp-ingest.js
iindex 773d9ae..b890913 100644
---- a/workers/rtmp-ingest.js
++++ b/workers/rtmp-ingest.js
@@@ -253,7 +253,15 @@ if (require.main === module) {
      const { openRuntime } = require('../server/store');
      const rt = openRuntime({ config: load() });
      const worker = createRtmpIngest({ rt });
--    worker.start().catch((err) => { console.error(`[rtmp] failed to start: ${err.stack || err}`); process.exit(1); });
++    let started = false;
++    worker.start().then(() => { started = true; }, (err) => { console.error(`[rtmp] failed to start: ${err.stack || err}`); process.exit(1); });
++    // node-media-server's own server logs an uncaught exception and keeps going (NodeMediaServer.run
++    // registers exactly that). A malformed packet from one encoder must not take every other
++    // publisher on this worker down with it, so the worker keeps that behaviour once it is up.
++    process.on('uncaughtException', (err) => {
++        console.error(`[rtmp] uncaught exception (worker keeps running): ${err && err.stack || err}`);
++        if (!started) process.exit(1);
++    });
      for (const sig of ['SIGTERM', 'SIGINT']) process.on(sig, () => worker.drain(sig));
  }