openvibers/openrestream Commits

Review fixes: output reports only land for the worker that owns the output (a lost worker can no longer stop a released or reassigned output), destination keys scrubbed from ffmpeg errors, logs and events, loopback players cannot keep a drained or lost RTMP worker alive (sockets destroyed, exit hooks bounded), second publish on one connection left to node-media-server; Live patch: personal keys refused for channels with an OpenRe slot, chat bridges stopped when a mirrored session ends, switch re-checks liveness in its transaction, rows Live ended are detached from the mirror, id-less events acknowledged; patch re-made against Live f11f809

OP OpenVibers committed 23 Sept 2026 · a2ddd2b941 · 10 files, +125 −37





Changes
ddiff --git a/README.md b/README.md
iindex 0bd3567..28135fb 100644
---- a/README.md
++++ b/README.md
@@@ -161,13 +161,13 @@ sudo deploy/scripts/deploy.sh status
  
  ## Live integration (patch)
  
--[docs/live-patch.diff](docs/live-patch.diff) is the Live side, made against OpenVibe.Live `c384787` and verified there (applies with `git apply`; Live's `npm test` 62/62 on Node 22.22.1). **Switch off (the default) changes nothing**: with `OPENRE_URL` unset, or a slot on `'live'`, every changed code path returns what it returned before (`test/openre-switch.test.js` in the patch). It adds:
++[docs/live-patch.diff](docs/live-patch.diff) is the Live side, made against OpenVibe.Live `f11f809` and verified there (applies with `git apply`; Live's `npm test` 63/63 on Node 22.22.1). **Switch off (the default) changes nothing**: with `OPENRE_URL` unset, or a slot on `'live'`, every changed code path returns what it returned before (`test/openre-switch.test.js` in the patch). The only observable difference is additive: owner responses built from `managed_streams` rows carry the two new columns (`ingest_authority: 'live'`, `openre_stream_id: null`). It adds:
  
  - `server/openre/openre-client.js` — service-token client for OpenRe (streams by `external_ref`, create, rotate, sessions, playback descriptors).
  - `managed_streams.ingest_authority` (`'live'` default | `'openre'`) + `openre_stream_id`, and an `openre_sessions` projection table (additive, in `initDb`).
--- Live's RTMP `prePublish` refuses a slot key when the slot is on `'openre'` (a stream can never be ingested twice). Only RTMP moves; WHIP/browser/JSMPEG for that slot stay on Live.
++- Live's RTMP `prePublish` refuses a slot key when the slot is on `'openre'`, and a personal key (`users.stream_key`) when any of the user's slots is (a stream can never be ingested twice, and Live never pushes the same destinations as OpenRe). Only RTMP moves; WHIP/browser/JSMPEG for that slot stay on Live.
  - Go Live / stream-key UI for `'openre'` slots: `GET /api/streams/managed` returns the slot without its Live key; `GET /api/streams/managed/:id/profile` and `GET /api/streams/:id/endpoint` show OpenRe's RTMP URL and a key hint (never a key); `POST /api/streams/managed/:id/regenerate-key` rotates on OpenRe and shows the new key once; `GET /api/streams/:id/rtmp-status` reads the mirror. Small fallbacks in `broadcast.js` / `broadcast-workspace.js` display the hint.
--- `POST /internal/openre-events` — OpenVibe.Events delivery endpoint (SDK `parseDelivery` + `createInbox`, consumer `live-openre-mirror`): `openre.session.started` creates or attaches the slot's `streams` row (control config, go-live notifications, Live's own `live.stream.started`), `ended`/`failed` end it; revision-ordered, exactly once; only for switched slots whose definition has `mirror_to_live` (consent). A 30 s reconcile asks OpenRe about mirrored live sessions (heartbeat refresh; ends rows OpenRe ended). Live's stale-stream cleanup skips mirrored sessions OpenRe confirmed in the last 30 min, and Live's boot-time restream resume skips them.
++- `POST /internal/openre-events` — OpenVibe.Events delivery endpoint (SDK `parseDelivery` + `createInbox`, consumer `live-openre-mirror`): `openre.session.started` creates or attaches the slot's `streams` row (control config, go-live notifications, Live's own `live.stream.started`), `ended`/`failed` end it and stop the RobotStreamer bridge, chat relays and AI bots as Live's own unpublish does; revision-ordered, exactly once; only for switched slots whose definition has `mirror_to_live` (consent). A 30 s reconcile asks OpenRe about mirrored live sessions (heartbeat refresh; ends rows OpenRe ended). Live's stale-stream cleanup skips mirrored sessions OpenRe confirmed in the last 30 min, and Live's boot-time restream resume skips them.
  - Playback: `GET /api/streams/rtmp-proxy/:id.flv` proxies the session's `flv.internal_url` from OpenRe's playback descriptor (validated as a loopback `…/live/ses_….flv` URL), so Live's player is unchanged.
  - Restream routes refuse (409 + `manage_url`) edits/start/stop for destinations of `'openre'` slots: those restreams run and are managed on openre.stream.
  - Admin: `GET /api/admin/openre/status`, `PUT /api/admin/openre/managed/:id/ingest-authority {authority, force}` — `'openre'` needs the slot offline, an RTMP slot (or `force`), the owner's canonical subject; it finds or creates the OpenRe definition and, in one transaction, flips the slot and rotates Live's own key for it. `'live'` flips back.
ddiff --git a/docs/live-patch.diff b/docs/live-patch.diff
iindex b6988b2..addd796 100644
---- a/docs/live-patch.diff
++++ b/docs/live-patch.diff
@@@ -80,7 +80,7 @@ index e8f5903..60c20bf 100644
       // A critical (money) migration that fails throws here and stops the boot rather than serving
       // half-converted balances.
  diff --git a/server/index.js b/server/index.js
--index 4bba504..a26d028 100644
++index 63c0969..c88df08 100644
  --- a/server/index.js
  +++ b/server/index.js
  @@ -221,6 +221,8 @@ function webrtcStreamHasActiveProducer(streamId) {
@@@ -102,7 +102,7 @@ index 4bba504..a26d028 100644
   
   // Internal (server-to-server) routes — allow openvibe.network to call into this service
   // Summary numbers for the Network's navigation service (ordering sites by real use).
--@@ -650,6 +655,7 @@ app.use('/api/recap', require('./recap/routes'));            // after-show repor
++@@ -652,6 +657,7 @@ app.use('/api/recap', require('./recap/routes'));            // after-show repor
   app.use('/api/comments', commentRoutes);
   app.use('/api/controls', controlRoutes);
   app.use('/api/onvif', onvifRoutes);
@@@ -110,7 +110,7 @@ index 4bba504..a26d028 100644
   app.use('/api/admin', adminRoutes);
   app.use('/api/mod', require('./admin/mod-routes'));
   app.use('/api/channels', require('./admin/channel-mod-routes'));
--@@ -1153,6 +1159,8 @@ async function start() {
++@@ -1155,6 +1161,8 @@ async function start() {
       // 5b. Resume enabled restreams for streams that survived the restart.
       // WHIP/RTMP broadcasters have no browser session to re-start them manually.
       for (const stream of db.getLiveStreams()) {
@@@ -119,7 +119,7 @@ index 4bba504..a26d028 100644
           restreamManager.resumeForStream(stream.id, stream.user_id, {
               protocol: stream.protocol,
               streamKey: stream.managed_stream_key,
--@@ -1220,6 +1228,9 @@ async function start() {
++@@ -1222,6 +1230,9 @@ async function start() {
       // 6e. Start periodic viewer count polling for restream destinations
       restreamManager.startViewerCountPolling();
   
@@@ -131,10 +131,10 @@ index 4bba504..a26d028 100644
       // 7. Start HTTP server.
  diff --git a/server/openre/authority.js b/server/openre/authority.js
  new file mode 100644
--index 0000000..865fff1
++index 0000000..fa4b4ab
  --- /dev/null
  +++ b/server/openre/authority.js
--@@ -0,0 +1,136 @@
++@@ -0,0 +1,148 @@
  +'use strict';
  +/**
  + * The per-slot ingest switch: managed_streams.ingest_authority = 'live' (default) | 'openre'.
@@@ -169,10 +169,18 @@ index 0000000..865fff1
  +    return authorityOf(slotById(id)) === 'openre';
  +}
  +
--+/** Live's RTMP publish handler asks this before accepting a slot key. */
--+function refusesLiveIngest({ managedStream, protocol = 'rtmp' }) {
+++/**
+++ * Live's RTMP publish handler asks this before accepting a key. A slot key is refused when its slot
+++ * is on OpenRe; a personal key (users.stream_key, no slot) is refused when any of the user's slots is,
+++ * because Live would attach it to the user's live rows and push the user's destinations twice.
+++ */
+++function refusesLiveIngest({ managedStream, user = null, protocol = 'rtmp' }) {
  +    try {
--+        if (!managedStream || !OPENRE_PROTOCOLS.has(protocol)) return false;
+++        if (!OPENRE_PROTOCOLS.has(protocol)) return false;
+++        if (!managedStream) {
+++            if (!user || !client.enabled()) return false;
+++            return Boolean(db.get("SELECT 1 FROM managed_streams WHERE user_id = ? AND ingest_authority = 'openre' LIMIT 1", [user.id]));
+++        }
  +        // Unsetting OPENRE_URL is an emergency rollback to Live for every switched slot at once:
  +        // the slot then behaves as a Live slot again, with the Live key rotated at the switch
  +        // (only visible to the owner through Regenerate on the Go Live page).
@@@ -254,9 +262,13 @@ index 0000000..865fff1
  +        stream = await client.createStreamForSlot(slot, { subject, recordingMode: recordingModeFor(slot), recordingVisibility: visibility });
  +    }
  +    const newLiveKey = crypto.randomBytes(20).toString('hex');
--+    db.getDb().transaction(() => {
+++    const flipped = db.getDb().transaction(() => {
+++        // Checked again here: a Live publish may have started while OpenRe was being asked.
+++        if (db.get('SELECT id FROM streams WHERE managed_stream_id = ? AND is_live = 1 LIMIT 1', [slot.id])) return false;
  +        db.run("UPDATE managed_streams SET ingest_authority = 'openre', openre_stream_id = ?, stream_key = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?", [stream.id, newLiveKey, slot.id]);
+++        return true;
  +    })();
+++    if (!flipped) return { status: 409, error: `Slot ${slot.id} went live on Live meanwhile; switch it while it is offline` };
  +    return {
  +        status: 200,
  +        body: {
@@@ -273,10 +285,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..3f739cc
++index 0000000..5a32b1c
  --- /dev/null
  +++ b/server/openre/mirror.js
--@@ -0,0 +1,206 @@
++@@ -0,0 +1,216 @@
  +'use strict';
  +/**
  + * OpenRe sessions → Live's `streams` table (roadmap Wave 7, ADR-009: Live observes sessions through
@@@ -403,6 +415,10 @@ index 0000000..3f739cc
  +        after: () => {
  +            try { db.computeAndCacheStreamAnalytics(streamId); } catch { /* */ }
  +            try { require('../streaming/broadcast-server').endStream(streamId); } catch { /* */ }
+++            // What Live's own RTMP unpublish stops (server/index.js): the RS bridge, chat relays, AI bots.
+++            try { require('../integrations/robotstreamer-service').stopForStream(streamId); } catch { /* */ }
+++            try { require('../integrations/chat-relay-service').stopForStream(streamId); } catch { /* */ }
+++            try { require('../integrations/ai-chatbot-service').stopForStream(streamId); } catch { /* */ }
  +            console.log(`[OpenRe] session ${sessionId} ${state}: stream ${streamId} ended`);
  +        },
  +    };
@@@ -415,6 +431,8 @@ index 0000000..3f739cc
  +    const { parseDelivery } = require('openvibe-sdk/events');
  +    const delivery = parseDelivery(req.rawBody, req.headers, secret);
  +    if (!delivery) return res.status(401).json({ error: 'bad signature' });
+++    // Signed but unusable (no event_id): acknowledge so it is not redelivered forever.
+++    if (!delivery.event || !delivery.event.event_id) return res.status(204).end();
  +    let result;
  +    try {
  +        let after = null;
@@@ -455,6 +473,10 @@ index 0000000..3f739cc
  +    if (!rows.length || !client.enabled()) return 0;
  +    let n = 0;
  +    for (const r of rows) {
+++        // Live ended the row itself (End Stream, or stale cleanup while OpenRe was unreachable):
+++        // stop tracking it rather than keeping a live mirror of an offline stream.
+++        const row = r.stream_id ? db.getStreamById(r.stream_id) : null;
+++        if (row && !row.is_live) { db.run("UPDATE openre_sessions SET state = 'detached', updated_at = CURRENT_TIMESTAMP WHERE session_id = ?", [r.session_id]); continue; }
  +        let s;
  +        try { s = await client.getSession(r.session_id); } catch (err) {
  +            if (err.status === 404) s = { state: 'failed', revision: r.revision + 1, ended_at: null };
@@@ -951,7 +973,7 @@ index 70a0857..2c5d856 100644
           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
++index fb8ba5c..c857f5b 100644
  --- a/server/streaming/rtmp-server.js
  +++ b/server/streaming/rtmp-server.js
  @@ -121,6 +121,16 @@ class RTMPServer extends EventEmitter {
@@@ -961,8 +983,8 @@ index fb8ba5c..709da89 100644
  +            // OpenRe.Stream ingests this slot (managed_streams.ingest_authority = 'openre'): refuse
  +            // here so one stream can never be ingested twice. Slots on 'live' (the default) are
  +            // untouched. See server/openre/authority.js.
--+            if (require('../openre/authority').refusesLiveIngest({ managedStream, protocol: 'rtmp' })) {
--+                console.log(`[RTMP] Rejected: slot ${managedStream.id} is ingested by OpenRe`);
+++            if (require('../openre/authority').refusesLiveIngest({ managedStream, user: managedStream ? null : resolvedUser, protocol: 'rtmp' })) {
+++                console.log(`[RTMP] Rejected: ${managedStream ? `slot ${managedStream.id}` : `personal key of ${resolvedUser.username}`} is ingested by OpenRe`);
  +                const session = this.nms.getSession(id);
  +                if (session) session.reject();
  +                return;
@@@ -973,10 +995,10 @@ index fb8ba5c..709da89 100644
               const existingStreams = db.getLiveStreamsByUserId(resolvedUser.id);
  diff --git a/test/openre-switch.test.js b/test/openre-switch.test.js
  new file mode 100644
--index 0000000..d9630e8
++index 0000000..6ffdc33
  --- /dev/null
  +++ b/test/openre-switch.test.js
--@@ -0,0 +1,216 @@
++@@ -0,0 +1,231 @@
  +'use strict';
  +
  +// OpenRe ingest switch (roadmap Wave 7, ADR-009). With OPENRE_URL unset, or a slot on 'live' (the
@@@ -1077,6 +1099,7 @@ index 0000000..d9630e8
  +    const slot701 = db.getManagedStreamById(701);
  +    assert.strictEqual(authority.serializeSlot(slot701), slot701, 'the same object: responses unchanged');
  +    assert.strictEqual(authority.refusesLiveIngest({ managedStream: slot701 }), false);
+++    assert.strictEqual(authority.refusesLiveIngest({ managedStream: null, user: { id: 601 } }), false);
  +    // Even a slot marked 'openre' behaves as before while OpenRe is not configured.
  +    d.prepare("UPDATE managed_streams SET ingest_authority = 'openre' WHERE id = 701").run();
  +    const marked = db.getManagedStreamById(701);
@@@ -1127,7 +1150,8 @@ index 0000000..d9630e8
  +    assert.strictEqual(db.getManagedStreamByStreamKey('livekey701aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa'), undefined, "Live's old key is rotated away");
  +    assert.strictEqual(authority.refusesLiveIngest({ managedStream: switched }), true, "Live's RTMP refuses the slot");
  +    assert.strictEqual(authority.refusesLiveIngest({ managedStream: switched, protocol: 'whip' }), false, 'only RTMP moved');
--+    assert.strictEqual(authority.refusesLiveIngest({ managedStream: null }), false, 'personal keys stay on Live');
+++    assert.strictEqual(authority.refusesLiveIngest({ managedStream: null, user: { id: 602 } }), false, 'personal keys of users without an OpenRe slot stay on Live');
+++    assert.strictEqual(authority.refusesLiveIngest({ managedStream: null, user: { id: 601 } }), true, 'a personal key cannot double-ingest a switched channel');
  +    const shown = authority.serializeSlot(switched);
  +    assert.strictEqual(shown.stream_key, null);
  +    assert.strictEqual(shown.stream_key_managed_by, 'openre');
@@@ -1178,6 +1202,19 @@ index 0000000..d9630e8
  +    assert.strictEqual(await deliver(base, sessionEvent('failed', SES(6), 3, { failure_reason: 'worker_lost' })), 204);
  +    assert.strictEqual(db.getStreamById(s6).is_live, 0);
  +
+++    // Live ends a mirrored row itself (End Stream): reconcile stops tracking it.
+++    assert.strictEqual(await deliver(base, sessionEvent('started', SES(7), 2)), 204);
+++    const s7 = d.prepare('SELECT id FROM streams WHERE is_live = 1').get().id;
+++    openreSessions[SES(7)] = { id: SES(7), state: 'live', revision: 2 };
+++    db.endStream(s7);
+++    await mirror.reconcileOnce();
+++    assert.strictEqual(d.prepare('SELECT state FROM openre_sessions WHERE session_id = ?').get(SES(7)).state, 'detached');
+++    assert.strictEqual(mirror.hasLiveSession(s7), false);
+++    // A signed event without an id is acknowledged, not retried forever.
+++    const noId = sessionEvent('started', SES(8), 2);
+++    delete noId.event_id;
+++    assert.strictEqual(await deliver(base, noId), 204);
+++
  +    // ── Rollback ─────────────────────────────────────────────
  +    assert.strictEqual((await authority.setAuthority(701, 'live')).status, 200);
  +    const back = db.getManagedStreamById(701);
ddiff --git a/server/store/outputs.js b/server/store/outputs.js
iindex ed7a396..438df33 100644
---- a/server/store/outputs.js
++++ b/server/store/outputs.js
@@@ -273,10 +273,16 @@ function createOutputs({ db, config, events, clock, box, definitions, sessions }
      }
  
      /** State changes reported by the restream worker, with the events they imply. */
--    function report(outputId, change) {
++    /**
++     * State reported by a restream worker. With { workerId }, the report only lands while the output
++     * is still assigned to that worker: a worker the coordinator has given up (lost) or that handed
++     * the output over must not overwrite what the new owner does.
++     */
++    function report(outputId, change, { workerId } = {}) {
          return db.transaction(() => {
              const o = q.getOut.get(outputId);
              if (!o) return null;
++            if (workerId && o.worker_id !== workerId) return null;
              const set = { updated_at: now() };
              for (const k of ['state', 'last_error', 'restart_attempts', 'next_restart_at', 'started_at', 'live_at', 'ended_at']) {
                  if (change[k] !== undefined) set[k] = change[k];
ddiff --git a/test/coordinator.test.js b/test/coordinator.test.js
iindex b9a569a..24211eb 100644
---- a/test/coordinator.test.js
++++ b/test/coordinator.test.js
@@@ -74,6 +74,9 @@ t('a worker without heartbeats is lost: its sessions fail (session.failed) and i
      assert.strictEqual(rt.store.sessions.get(a.session.id).state, 'live', 'a lost restream worker never ends the source session');
      assert.strictEqual(rt.store.outputs.getOutput(out.id).state, 'pending');
      assert.strictEqual(rt.store.outputs.getOutput(out.id).worker, null);
++    // The lost worker waking up and stopping its ffmpeg must not overwrite the released output.
++    assert.strictEqual(rt.store.outputs.report(out.id, { state: 'stopped' }, { workerId: rs.id }), null);
++    assert.strictEqual(rt.store.outputs.getOutput(out.id).state, 'pending');
      // A new restream generation picks it up.
      const rs2 = worker('restream');
      assert.strictEqual(coordinator.syncTick().outputsAssigned, 1);
ddiff --git a/test/destinations-ui.test.js b/test/destinations-ui.test.js
iindex a9cee6b..d2be0e8 100644
---- a/test/destinations-ui.test.js
++++ b/test/destinations-ui.test.js
@@@ -41,6 +41,8 @@ t('ffmpeg arguments match Live: codec copy from HTTP-FLV, platform URL fixes, SR
      assert.strictEqual(ff.redactUrl('rtmps://live.twitch.tv/app/live_123456789'), 'rtmps://live.twitch.tv/app/****6789');
      assert.ok(!ff.redactUrl('srt://h:1?streamid=secret&passphrase=pw12345678').includes('secret'));
      assert.match(ff.friendlyError('Connection refused', 'twitch'), /Twitch's ingest server/);
++    const echoed = ff.friendlyError('Error opening output rtmp://ingest.example/app/sekrit_key_9999: Operation not permitted');
++    assert.ok(!echoed.includes('sekrit_key_9999') && echoed.includes('****9999'), echoed);
      assert.match(ff.friendlyError('NetStream.Publish.BadName', 'youtube'), /Youtube rejected the stream key/);
  });
  
ddiff --git a/workers/restream-worker.js b/workers/restream-worker.js
iindex a76285b..873a6cb 100644
---- a/workers/restream-worker.js
++++ b/workers/restream-worker.js
@@@ -47,7 +47,7 @@ function createRestreamWorker({ rt, log = console, exit = (code) => process.exit
              const session = store.sessions.get(row.session_id);
              const shouldRun = row.desired === 'run' && session && session.state === 'live';
              if (!shouldRun) {
--                if (runner) { runner.stop(row.desired === 'stop' ? 'stop requested' : 'session not live'); runners.delete(row.id); } else store.outputs.report(row.id, { state: 'stopped', ended_at: Date.now() });
++                if (runner) { runner.stop(row.desired === 'stop' ? 'stop requested' : 'session not live'); runners.delete(row.id); } else store.outputs.report(row.id, { state: 'stopped', ended_at: Date.now() }, { workerId: runtime.me.id });
                  continue;
              }
              if (runner && runner.stopped) { runners.delete(row.id); runner = null; continue; }
@@@ -56,10 +56,10 @@ function createRestreamWorker({ rt, log = console, exit = (code) => process.exit
              // 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; }
++            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() }, { workerId: runtime.me.id }); continue; }
              // Let the ingest settle before pulling (Live waits 3 s for node-media-server's FLV).
              if (session.live_at && Date.now() - session.live_at < config.outputs.startDelayMs) continue;
--            runner = new OutputRunner({ outputId: row.id, destinationId: row.destination_id, inputUrl: input, store, config, log, spawnImpl, lookup });
++            runner = new OutputRunner({ outputId: row.id, destinationId: row.destination_id, inputUrl: input, store, config, log, spawnImpl, lookup, workerId: runtime.me.id });
              runners.set(row.id, runner);
              runner.start().catch((err) => runner.fail(err.message, { cooldown: false }));
          }
ddiff --git a/workers/restream/ffmpeg-args.js b/workers/restream/ffmpeg-args.js
iindex d2c9072..13d169a 100644
---- a/workers/restream/ffmpeg-args.js
++++ b/workers/restream/ffmpeg-args.js
@@@ -123,7 +123,16 @@ function friendlyError(raw, platform = 'the destination') {
      if (/Unrecognized option|Invalid argument|Option .* not found/i.test(t)) return `Encoder configuration was rejected by ffmpeg (${t.slice(0, 80)})`;
      if (/No such file|not found/i.test(t) && /ffmpeg/i.test(t)) return 'ffmpeg is not installed on the server';
      if (!t.trim()) return 'Restream process stopped without a message';
--    return t.slice(0, 160);
++    return redactText(t).slice(0, 160);
++}
++
++/** Redact every rtmp(s)/srt URL inside a free text (ffmpeg stderr echoes its output URL). */
++function redactText(text) {
++    return String(text || '').replace(/(?:rtmps?|srt):\/\/[^\s'"|]+/gi, (u) => {
++        const tail = /[:;,.)\]]+$/.exec(u);
++        const url = tail ? u.slice(0, -tail[0].length) : u;
++        return redactUrl(url) + (tail ? tail[0] : '');
++    });
  }
  
  /** Mask keys in URLs before anything is logged (rtmp path key, SRT streamid/passphrase). */
@@@ -141,5 +150,5 @@ function redactUrl(value) {
  
  module.exports = {
      QUALITY_PRESETS, PLATFORM_DEFAULT_PRESET, ENCODER_PRESETS, SRT_DEFAULT_LATENCY_MS,
--    isSrtUrl, buildDestUrl, buildSrtUrl, outputArgs, rtmpCopyArgs, withProgress, resolvePreset, customOverrides, encodingArgs, friendlyError, redactUrl,
++    isSrtUrl, buildDestUrl, buildSrtUrl, outputArgs, rtmpCopyArgs, withProgress, resolvePreset, customOverrides, encodingArgs, friendlyError, redactUrl, redactText,
  };
ddiff --git a/workers/restream/output-runner.js b/workers/restream/output-runner.js
iindex 9e19936..4012cb7 100644
---- a/workers/restream/output-runner.js
++++ b/workers/restream/output-runner.js
@@@ -8,12 +8,14 @@
   */
  const { spawn } = require('child_process');
  const fs = require('fs');
--const { rtmpCopyArgs, withProgress, buildDestUrl, friendlyError, redactUrl } = require('./ffmpeg-args');
++const { rtmpCopyArgs, withProgress, buildDestUrl, friendlyError, redactUrl, redactText } = require('./ffmpeg-args');
  const { validateDestinationUrl, checkResolvedHost } = require('../../server/destination-url');
  
  class OutputRunner {
--    constructor({ outputId, destinationId, inputUrl, store, config, log = console, spawnImpl = spawn, lookup }) {
++    constructor({ outputId, destinationId, inputUrl, store, config, log = console, spawnImpl = spawn, lookup, workerId = null }) {
          this.outputId = outputId;
++        this.workerId = workerId;
++        this.secrets = [];
          this.destinationId = destinationId;
          this.inputUrl = inputUrl;
          this.store = store;
@@@ -40,8 +42,15 @@ class OutputRunner {
          return t;
      }
  
--    report(change) { try { return this.store.outputs.report(this.outputId, change); } catch (err) { this.log.error(`[restream] report ${this.outputId}: ${err.message}`); return null; } }
--    note(level, message) { try { this.store.outputs.log(this.outputId, this.destinationId, level, message); } catch { /* logs are best effort */ } }
++    report(change) { try { return this.store.outputs.report(this.outputId, change, { workerId: this.workerId }); } catch (err) { this.log.error(`[restream] report ${this.outputId}: ${err.message}`); return null; } }
++    note(level, message) { try { this.store.outputs.log(this.outputId, this.destinationId, level, this.scrub(message)); } catch { /* logs are best effort */ } }
++
++    /** ffmpeg's stderr can echo the output URL: never let a destination key reach a log or event. */
++    scrub(text) {
++        let t = redactText(text);
++        for (const s of this.secrets) if (s && s.length >= 4) t = t.split(s).join('****');
++        return t;
++    }
  
      async start() {
          if (this.stopped) return;
@@@ -51,6 +60,7 @@ class OutputRunner {
          }
          if (!dest || !dest.enabled || dest.hold_reason) return this.stop('destination disabled');
          this.platform = dest.platform || 'custom';
++        this.secrets = [dest.stream_key, dest.srt_passphrase].filter(Boolean);
          const v = validateDestinationUrl(dest.server_url, { allowPrivate: this.o.allowPrivateHosts });
          if (!v.ok) return this.fail(v.error, { cooldown: false });
          const resolved = await checkResolvedHost(v.host, { allowPrivate: this.o.allowPrivateHosts, ...(this.lookup ? { lookup: this.lookup } : {}) });
@@@ -160,9 +170,9 @@ class OutputRunner {
              this.stopped = true;
              return;
          }
--        const message = this.ackTimeout
++        const message = this.scrub(this.ackTimeout
              ? `No response from the ingest within ${this.o.liveAckTimeoutMs / 1000}s — ${friendlyError(raw, this.platform)}`
--            : friendlyError(raw || `ffmpeg exit code ${code}`, this.platform);
++            : friendlyError(raw || `ffmpeg exit code ${code}`, this.platform));
          this.ackTimeout = false;
          this.note('error', `ffmpeg exited (code ${code}) after ${(ran / 1000).toFixed(1)}s: ${message}`);
          this.status = 'error';
@@@ -193,6 +203,7 @@ class OutputRunner {
      /** The circuit breaker: this output is done; the destination may cool down. */
      fail(message, { cooldown }) {
          if (this.stopped) return;
++        message = this.scrub(message);
          this.stopped = true;
          this.clearTimers();
          let cooldownMinutes = null;
ddiff --git a/workers/rtmp-ingest.js b/workers/rtmp-ingest.js
iindex b890913..d29dea9 100644
---- a/workers/rtmp-ingest.js
++++ b/workers/rtmp-ingest.js
@@@ -105,6 +105,9 @@ function createRtmpIngest({ rt, log = console, exit = (code) => process.exit(cod
          const s = nms(id);
          if (!s) return;
          const reject = (why) => { log.log(`[rtmp] publish refused (${why}) from ${s.ip}`); s.reject(); };
++        // A second publish on a connection that already publishes: leave it to node-media-server,
++        // which answers NetStream.Publish.BadConnection without touching the first stream.
++        if (publishers.has(id)) return undefined;
          if (!isPublicSocket(s)) return reject('not the public listener');
          if (runtime.draining || !runtime.me || runtime.me.state !== 'ready') return reject('generation not taking sessions');
          if (publishers.size >= config.rtmp.maxPublishersPerWorker) return reject('worker full');
@@@ -188,7 +191,12 @@ function createRtmpIngest({ rt, log = console, exit = (code) => process.exit(cod
          closePublic();
          const closing = [];
          for (const srv of [playServer, flvServer]) {
--            if (srv) closing.push(new Promise(r => { srv.closeAllConnections?.(); srv.close(() => r()); }));
++            if (!srv) continue;
++            // Loopback players (Media's recorder, restream ffmpeg, the playback proxy) would keep
++            // close() waiting forever: node-media-server parks them as idle players.
++            for (const sock of srv.openSockets || []) sock.destroy();
++            srv.closeAllConnections?.();
++            closing.push(new Promise(r => srv.close(() => r())));
          }
          playServer = null;
          flvServer = null;
@@@ -197,10 +205,14 @@ function createRtmpIngest({ rt, log = console, exit = (code) => process.exit(cod
      }
  
      function newRtmpServer() {
--        return net.createServer((socket) => {
++        const srv = net.createServer((socket) => {
++            srv.openSockets.add(socket);
++            socket.on('close', () => srv.openSockets.delete(socket));
              const session = new NodeRtmpSession(nmsConfig, socket);
              session.run();
          });
++        srv.openSockets = new Set();
++        return srv;
      }
  
      async function start() {
ddiff --git a/workers/runtime.js b/workers/runtime.js
iindex 1386c54..b09c92c 100644
---- a/workers/runtime.js
++++ b/workers/runtime.js
@@@ -18,6 +18,14 @@
   * The heartbeat (every heartbeatMs) is also the lease renewal for every session the worker owns.
   * A worker never depends on openre-api being up; it needs only the database.
   */
++/** Run a cleanup hook, but never let it keep the process from exiting (5 s at most). */
++function settle(fn) {
++    return Promise.race([
++        Promise.resolve().then(() => fn && fn()).catch(() => {}),
++        new Promise((r) => { const t = setTimeout(r, 5000); t.unref?.(); }),
++    ]);
++}
++
  function createWorkerRuntime({ rt, kind, log = console, hooks = {}, exit = (code) => process.exit(code) }) {
      const { store, config } = rt;
      let me = null;
@@@ -48,7 +56,7 @@ function createWorkerRuntime({ rt, kind, log = console, hooks = {}, exit = (code
              log.error(`[${kind}] coordinator marked ${me.id} lost; dropping transports and exiting`);
              stopped = true;
              clearTimeout(timer);
--            Promise.resolve(hooks.onLost && hooks.onLost()).finally(() => exit(1));
++            settle(hooks.onLost).finally(() => exit(1));
              return row;
          }
          if (row.state === 'draining' && !draining) {
@@@ -99,7 +107,7 @@ function createWorkerRuntime({ rt, kind, log = console, hooks = {}, exit = (code
          clearTimeout(timer);
          store.workers.stop(me.id, draining ? 'drained' : 'exit');
          log.log(`[${kind}] generation ${me.generation} idle after drain; exiting`);
--        Promise.resolve(hooks.onExit && hooks.onExit()).finally(() => exit(0));
++        settle(hooks.onExit).finally(() => exit(0));
          return true;
      }