openvibers/openrestream Commits
OpenRe's workers no longer drop async failures on the floor (Node ends a process on an unhandled rejection, which would drop every publisher on a worker): the coordinator's outbox prune catches its rejection; the restream worker awaits its start inside the boot's catch; an RTMP admission whose success handler fails (sessions.finish before the publisher went live) is logged instead of unhandled; SIGTERM/SIGINT drains catch their rejection; the runtime's stop() records the stopped row with a handled rejection; log lines survive a rejection with no Error. The HTTP GET handlers were already covered by openvibe-shared's routeAsyncErrors (installed by instrument()); test/http-async.test.js pins that (a rejecting store answers 500 openre.internal, no unhandled rejection), and test/worker-async.test.js pins the worker paths. Found by the estate-wide floating-promise sweep; reviewed independently (its items fixed or shown covered). npm test 17/20 (3 need PostgreSQL), test:pg 20/20. npm test 17/20 test files passed, 3 skipped (api-restart.test.js: api-restart: skipped (OpenRe proc.
Changes
nnew file mode 100644@@@ -0,0 +1,16 @@++'use strict';++/**++ * NODE_OPTIONS=--require for test/worker-async.test.js: the restream worker's boot IIFE must await++ * worker.start(), so a failing register is reported by its own `.catch` (exit 1) instead of becoming++ * an unhandled rejection. openRuntime is replaced with a runtime whose workers.register rejects, so++ * no database is opened and the child runs the same way under PGlite and PostgreSQL.++ */++const store = require('../../server/store');++const { load } = require('../../server/config');++++store.openRuntime = async () => ({++ config: load(),++ store: { workers: { register: async () => { throw new Error('register boom'); } } },++ events: {},++ db: { close() {} },++});nnew file mode 100644@@@ -0,0 +1,43 @@++'use strict';++// Express 4 ignores what an `async (req, res)` handler returns, so a rejection becomes an unhandled++// rejection and Node ends the API process. server/app.js calls openvibe-shared/metrics instrument(),++// which installs openvibe-shared/express-async's routeAsyncErrors on the app (one prototype patch, so++// it covers app-level handlers and every Router alike). This guards that wiring: a store call that++// rejects answers 500 problem+json and nothing floats.++const assert = require('assert');++const { bootApi, request, userToken, suite, OWNER } = require('./helpers');++++const t = suite('http-async');++++const cookie = `ov_token=${userToken({ subjectId: OWNER })}`;++++t('a rejecting store call in an async GET answers 500, with no unhandled rejection', async () => {++ const api = await bootApi();++ const seen = [];++ const onUnhandled = (err) => seen.push(err);++ process.on('unhandledRejection', onUnhandled);++ const list = api.rt.store.definitions.list;++ const get = api.rt.store.sessions.get;++ try {++ // server/ui/routes.js `/streams`: a bare async GET on a Router, reached as the signed-in owner.++ api.rt.store.definitions.list = async () => { throw new Error('db down (definitions.list)'); };++ const ui = await request(api.base, 'GET', '/streams', { cookie });++ assert.strictEqual(ui.status, 500, ui.text);++ assert.match(ui.headers.get('content-type'), /application\/problem\+json/);++ assert.strictEqual(ui.body.code, 'openre.internal');++++ // server/app.js `/play/:file`: a bare async GET on the app itself.++ api.rt.store.sessions.get = async () => { throw new Error('db down (sessions.get)'); };++ const play = await request(api.base, 'GET', `/play/${'ses_' + '0'.repeat(26)}.flv`);++ assert.strictEqual(play.status, 500, play.text);++ assert.strictEqual(play.body.code, 'openre.internal');++++ await new Promise((r) => setTimeout(r, 50));++ assert.deepStrictEqual(seen.map((e) => String(e && e.message)), [], 'a handler rejection floated unhandled');++ } finally {++ api.rt.store.definitions.list = list;++ api.rt.store.sessions.get = get;++ process.removeListener('unhandledRejection', onUnhandled);++ await api.close();++ }++});nnew file mode 100644@@@ -0,0 +1,84 @@++'use strict';++// Async hardening in the workers (the PostgreSQL sweep): a promise the worker runtime or the RTMP++// admission path used to drop now has a handler. Node 22 ends the process on an unhandled rejection,++// and on an RTMP worker that would drop every publisher it carries, so each of these is a regression++// test for one dropped promise — remove the `.catch` and the test sees the rejection.++const assert = require('assert');++const path = require('path');++const { runtime, testEnv, tmpDir, child, suite, ROOT } = require('./helpers');++const { createWorkerRuntime } = require('../workers/runtime');++const { admissionChain } = require('../workers/rtmp-ingest');++const { pruneOutbox } = require('../workers/coordinator');++++const t = suite('worker-async');++++/** Run fn while collecting the process's unhandled rejections (adding a listener suppresses Node's exit). */++async function withUnhandled(fn) {++ const unhandled = [];++ const onUnhandled = (err) => unhandled.push(err);++ process.on('unhandledRejection', onUnhandled);++ try { await fn(); await new Promise((r) => setTimeout(r, 30)); } finally { process.removeListener('unhandledRejection', onUnhandled); }++ return unhandled;++}++++t('the worker runtime handles a failing stop(): it is logged, not an unhandled rejection', async () => {++ const rt = await runtime();++ const errors = [];++ const wr = createWorkerRuntime({++ rt, kind: 'restream', exit: () => {},++ log: { log() {}, warn() {}, error: (m) => errors.push(String(m)) },++ hooks: { activeCount: () => 1 },++ });++ await wr.register({});++ rt.store.workers.stop = async () => { throw new Error('stop boom'); };++ const unhandled = await withUnhandled(() => { wr.stop(); });++ assert.deepStrictEqual(unhandled, [], 'store.workers.stop returns a promise; dropping it surfaces here');++ assert.ok(errors.some((m) => /stop boom/.test(m)), 'the failure is logged');++});++++t('an RTMP admission whose async success handler rejects is logged, not unhandled', async () => {++ // The production call discards the chain's promise; so must this test, or the await would hide the bug.++ const errors = [];++ const log = { log() {}, warn() {}, error: (m) => errors.push(String(m)) };++ const unhandled = await withUnhandled(() => {++ admissionChain(++ Promise.resolve({ sessionId: 'ses_000000000000000000000000' }),++ async () => { throw new Error('finish boom'); },++ () => errors.push('onFailed'),++ log,++ );++ });++ assert.deepStrictEqual(unhandled, [], 'a rejection inside the async success handler must be caught');++ assert.ok(errors.some((m) => /\[rtmp\] admission failed: Error: finish boom/.test(m)), 'it is logged');++ assert.ok(!errors.includes('onFailed'), 'onFailed handles only a rejection of admission itself');++});++++t('the coordinator handles a failing outbox prune: logged, not an unhandled rejection', async () => {++ // The 6-hour timer discards the promise, so a rejection would end the coordinator unhandled.++ const errors = [];++ const unhandled = await withUnhandled(() => {++ pruneOutbox({ prune: async () => { throw new Error('prune boom'); } }, { error: (m) => errors.push(String(m)) });++ });++ assert.deepStrictEqual(unhandled, []);++ assert.ok(errors.some((m) => /prune boom/.test(m)), 'the failure is logged');++ // A synchronous throw (an absent outbox) takes the same path, as the old try/catch did.++ const syncErrors = [];++ const syncUnhandled = await withUnhandled(() => {++ pruneOutbox({ prune: () => { throw new Error('sync boom'); } }, { error: (m) => syncErrors.push(String(m)) });++ });++ assert.deepStrictEqual(syncUnhandled, []);++ assert.ok(syncErrors.some((m) => /sync boom/.test(m)));++});++++t('the restream worker boot awaits start(): a register failure is reported, not left unhandled', async () => {++ const dir = tmpDir();++ const preload = path.join(ROOT, 'test', 'fixtures', 'fail-worker-register.js');++ const p = child('workers/restream-worker.js', { ...await testEnv(dir), NODE_OPTIONS: `--require ${preload}` });++ const r = await p.exited;++ assert.strictEqual(r.code, 1, p.output);++ // Without the await, the boot IIFE resolves and the rejection is unhandled: Node prints its own++ // stack and this process-owned line never appears.++ assert.match(p.output, /\[restream-worker\] failed to start: Error: register boom/);++});++++t.run();@@@ -8,6 +8,15 @@*/const path = require('path');++/**++ * Best-effort housekeeping on a 6-hour timer: `prune` is async and the timer is fire-and-forget, so its rejection++ * must be handled — an unhandled one ends the coordinator (Node 22), delaying output assignment and the events++ * relay. Wrapping the call also turns a synchronous throw (an absent outbox) into the same logged path.++ */++function pruneOutbox(outbox, log = console) {++ return Promise.resolve().then(() => outbox.prune()).catch((err) => log.error(`[coordinator] outbox prune failed (next time): ${err && err.message || err}`));++}++if (require.main === module) {(async () => {require('dotenv').config({ path: process.env.OPENRE_ENV_FILE || path.join(process.cwd(), '.env') });@@@ -23,7 +32,8 @@ if (require.main === module) {const coordinator = createCoordinator({ rt, media, lineage });coordinator.start();console.log(`[coordinator] running (events relay ${rt.events.configured ? 'on' : 'off: EVENTS_URL/OV_OAUTH_CLIENT_SECRET not set'}, recordings ${media.configured ? `on → ${config.media.url} app ${config.media.appId}` : 'off'}, Live lineage ${lineage.enabled ? 'on' : 'off'})`);-- const prune = setInterval(() => { try { rt.events.outbox.prune(); } catch { /* next time */ } }, 6 * 60 * 60 * 1000);++ // prune is async: a bare call would reject unhandled (Node ends the coordinator) instead of being caught.++ const prune = setInterval(() => pruneOutbox(rt.events.outbox, console), 6 * 60 * 60 * 1000);prune.unref();const shutdown = (sig) => {console.log(`[coordinator] ${sig}: stopping`);@@@ -34,3 +44,5 @@ if (require.main === module) {process.on('SIGINT', () => shutdown('SIGINT'));})().catch((err) => { console.error(`[coordinator] failed to start: ${err && err.stack || err}`); process.exit(1); });}++++module.exports = { pruneOutbox };@@@ -117,13 +117,13 @@ if (require.main === module) {exitIfDrill(config, 'openre-restream-worker');const rt = await openRuntime({ config });const worker = createRestreamWorker({ rt });-- worker.start();++ await 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));++ for (const sig of ['SIGTERM', 'SIGINT']) process.on(sig, () => worker.drain(sig).catch((err) => console.error(`[restream-worker] drain failed on ${sig}: ${err && (err.stack || err.message) || err}`)));})().catch((err) => { console.error(`[restream-worker] failed to start: ${err && err.stack || err}`); process.exit(1); });}@@@ -53,6 +53,16 @@ function listenInRange(server, min, max, skip = new Set()) {});}++/**++ * Run `admission` then `onAdmitted`, with `onFailed` for a rejection of `admission` itself. `.then(onOk, onErr)`++ * leaves a rejection of the (async) success handler — `sessions.finish` before the publisher went live — unhandled,++ * and Node 22 ends the worker on an unhandled rejection, dropping every publisher on it. The trailing `.catch`++ * routes both rejection sources to the log.++ */++function admissionChain(admission, onAdmitted, onFailed, log) {++ return admission.then(onAdmitted, onFailed).catch((err) => log.error(`[rtmp] admission failed: ${err.stack || err}`));++}++function createRtmpIngest({ rt, log = console, exit = (code) => process.exit(code) }) {const { store, config } = rt;NmsLogger.setLogType(1); // errors only: the library's info lines would print stream paths@@@ -127,7 +137,7 @@ function createRtmpIngest({ rt, log = console, exit = (code) => process.exit(cod// NetStream.Publish.BadConnection without touching the first stream.if (publishers.has(session.id)) return onPublish(invokeMessage);const streamPath = `/${session.appname}/${String(invokeMessage.streamName || '').split('?')[0]}`;-- admit(session, streamPath).then(async (adm) => {++ admissionChain(admit(session, streamPath), async (adm) => {if (!session.isStarting) {// The publisher left while it was being admitted: its session ends without going live.if (adm.sessionId) await store.sessions.finish(adm.sessionId, { reason: 'publisher_disconnected', actor: `worker:${runtime.me.id}` });@@@ -139,7 +149,7 @@ function createRtmpIngest({ rt, log = console, exit = (code) => process.exit(codlog.error(`[rtmp] admission failed: ${err.stack || err}`);admissions.set(session.id, { error: 'admission failed' });if (session.isStarting) onPublish(invokeMessage);-- });++ }, log);};}@@@ -313,8 +323,8 @@ if (require.main === module) {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));++ for (const sig of ['SIGTERM', 'SIGINT']) process.on(sig, () => worker.drain(sig).catch((err) => console.error(`[rtmp] drain failed on ${sig}: ${err && (err.stack || err.message) || err}`)));})().catch((err) => { console.error(`[rtmp-ingest] failed to start: ${err && err.stack || err}`); process.exit(1); });}--module.exports = { createRtmpIngest, listenInRange };++module.exports = { createRtmpIngest, listenInRange, admissionChain };@@@ -84,14 +84,14 @@ function createWorkerRuntime({ rt, kind, log = console, hooks = {}, exit = (code* them); async ones used to reject unhandled — a failure is now logged, sync or async. */function call(name, ...args) {if (typeof hooks[name] !== 'function') return;-- Promise.resolve().then(() => hooks[name](...args)).catch((err) => log.error(`[${kind}] ${name}: ${err.message}`));++ Promise.resolve().then(() => hooks[name](...args)).catch((err) => log.error(`[${kind}] ${name}: ${err && (err.stack || err.message) || err}`));}function schedule() {clearTimeout(timer);if (stopped) return;timer = setTimeout(async () => {-- try { await beat(); } catch (err) { log.error(`[${kind}] heartbeat failed: ${err.message}`); }++ try { await beat(); } catch (err) { log.error(`[${kind}] heartbeat failed: ${err && (err.stack || err.message) || err}`); }schedule();}, config.workers.heartbeatMs);}@@@ -115,7 +115,7 @@ function createWorkerRuntime({ rt, kind, log = console, hooks = {}, exit = (code// The stopped row is written before the process goes: otherwise the worker stays ready/draining until its lease// expires and new sessions can be routed to a generation that is gone.Promise.resolve().then(() => store.workers.stop(me.id, draining ? 'drained' : 'exit'))-- .catch((err) => log.error(`[${kind}] could not record the stop: ${err.message}`))++ .catch((err) => log.error(`[${kind}] could not record the stop: ${err && (err.stack || err.message) || err}`)).finally(() => settle(hooks.onExit).finally(() => exit(0)));return true;}@@@ -128,7 +128,14 @@ function createWorkerRuntime({ rt, kind, log = console, hooks = {}, exit = (codeexitWhenIdle,get me() { return me; },get draining() { return draining; },-- stop() { stopped = true; clearTimeout(timer); if (me) store.workers.stop(me.id, 'stopped'); },++ stop() {++ stopped = true;++ clearTimeout(timer);++ // A close() caller does not wait for this write, but a rejection must still be handled: an++ // unhandled one ends the process, and the stopped row would be lost without a trace.++ if (me) return store.workers.stop(me.id, 'stopped').catch((err) => log.error(`[${kind}] could not record the stop: ${err && (err.stack || err.message) || err}`));++ return undefined;++ },};}@@@ -41,8 +41,8 @@ async function main(kind) {const config = load();exitIfDrill(config, `openre-${kind}`);const w = createUnportedTransport({ rt: await openRuntime({ config }), kind });-- w.start();-- for (const sig of ['SIGTERM', 'SIGINT']) process.on(sig, () => w.drain(sig));++ await w.start();++ for (const sig of ['SIGTERM', 'SIGINT']) process.on(sig, () => w.drain(sig).catch((err) => console.error(`[${kind}] drain failed on ${sig}: ${err && (err.stack || err.message) || err}`)));}module.exports = { createUnportedTransport, PORT_STATUS, main };
new file mode 100644
new file mode 100644
@@ -0,0 +1,16 @@
@@ -0,0 +1,16 @@
+'use strict';
+/**
+ * NODE_OPTIONS=--require for test/worker-async.test.js: the restream worker's boot IIFE must await
+ * worker.start(), so a failing register is reported by its own `.catch` (exit 1) instead of becoming
+ * an unhandled rejection. openRuntime is replaced with a runtime whose workers.register rejects, so
+ * no database is opened and the child runs the same way under PGlite and PostgreSQL.
+ */
+const store = require('../../server/store');
+const { load } = require('../../server/config');
+
+store.openRuntime = async () => ({
+ config: load(),
+ store: { workers: { register: async () => { throw new Error('register boom'); } } },
+ events: {},
+ db: { close() {} },
+});
new file mode 100644
new file mode 100644
@@ -0,0 +1,43 @@
@@ -0,0 +1,43 @@
+'use strict';
+// Express 4 ignores what an `async (req, res)` handler returns, so a rejection becomes an unhandled
+// rejection and Node ends the API process. server/app.js calls openvibe-shared/metrics instrument(),
+// which installs openvibe-shared/express-async's routeAsyncErrors on the app (one prototype patch, so
+// it covers app-level handlers and every Router alike). This guards that wiring: a store call that
+// rejects answers 500 problem+json and nothing floats.
+const assert = require('assert');
+const { bootApi, request, userToken, suite, OWNER } = require('./helpers');
+
+const t = suite('http-async');
+
+const cookie = `ov_token=${userToken({ subjectId: OWNER })}`;
+
+t('a rejecting store call in an async GET answers 500, with no unhandled rejection', async () => {
+ const api = await bootApi();
+ const seen = [];
+ const onUnhandled = (err) => seen.push(err);
+ process.on('unhandledRejection', onUnhandled);
+ const list = api.rt.store.definitions.list;
+ const get = api.rt.store.sessions.get;
+ try {
+ // server/ui/routes.js `/streams`: a bare async GET on a Router, reached as the signed-in owner.
+ api.rt.store.definitions.list = async () => { throw new Error('db down (definitions.list)'); };
+ const ui = await request(api.base, 'GET', '/streams', { cookie });
+ assert.strictEqual(ui.status, 500, ui.text);
+ assert.match(ui.headers.get('content-type'), /application\/problem\+json/);
+ assert.strictEqual(ui.body.code, 'openre.internal');
+
+ // server/app.js `/play/:file`: a bare async GET on the app itself.
+ api.rt.store.sessions.get = async () => { throw new Error('db down (sessions.get)'); };
+ const play = await request(api.base, 'GET', `/play/${'ses_' + '0'.repeat(26)}.flv`);
+ assert.strictEqual(play.status, 500, play.text);
+ assert.strictEqual(play.body.code, 'openre.internal');
+
+ await new Promise((r) => setTimeout(r, 50));
+ assert.deepStrictEqual(seen.map((e) => String(e && e.message)), [], 'a handler rejection floated unhandled');
+ } finally {
+ api.rt.store.definitions.list = list;
+ api.rt.store.sessions.get = get;
+ process.removeListener('unhandledRejection', onUnhandled);
+ await api.close();
+ }
+});
new file mode 100644
new file mode 100644
@@ -0,0 +1,84 @@
@@ -0,0 +1,84 @@
+'use strict';
+// Async hardening in the workers (the PostgreSQL sweep): a promise the worker runtime or the RTMP
+// admission path used to drop now has a handler. Node 22 ends the process on an unhandled rejection,
+// and on an RTMP worker that would drop every publisher it carries, so each of these is a regression
+// test for one dropped promise — remove the `.catch` and the test sees the rejection.
+const assert = require('assert');
+const path = require('path');
+const { runtime, testEnv, tmpDir, child, suite, ROOT } = require('./helpers');
+const { createWorkerRuntime } = require('../workers/runtime');
+const { admissionChain } = require('../workers/rtmp-ingest');
+const { pruneOutbox } = require('../workers/coordinator');
+
+const t = suite('worker-async');
+
+/** Run fn while collecting the process's unhandled rejections (adding a listener suppresses Node's exit). */
+async function withUnhandled(fn) {
+ const unhandled = [];
+ const onUnhandled = (err) => unhandled.push(err);
+ process.on('unhandledRejection', onUnhandled);
+ try { await fn(); await new Promise((r) => setTimeout(r, 30)); } finally { process.removeListener('unhandledRejection', onUnhandled); }
+ return unhandled;
+}
+
+t('the worker runtime handles a failing stop(): it is logged, not an unhandled rejection', async () => {
+ const rt = await runtime();
+ const errors = [];
+ const wr = createWorkerRuntime({
+ rt, kind: 'restream', exit: () => {},
+ log: { log() {}, warn() {}, error: (m) => errors.push(String(m)) },
+ hooks: { activeCount: () => 1 },
+ });
+ await wr.register({});
+ rt.store.workers.stop = async () => { throw new Error('stop boom'); };
+ const unhandled = await withUnhandled(() => { wr.stop(); });
+ assert.deepStrictEqual(unhandled, [], 'store.workers.stop returns a promise; dropping it surfaces here');
+ assert.ok(errors.some((m) => /stop boom/.test(m)), 'the failure is logged');
+});
+
+t('an RTMP admission whose async success handler rejects is logged, not unhandled', async () => {
+ // The production call discards the chain's promise; so must this test, or the await would hide the bug.
+ const errors = [];
+ const log = { log() {}, warn() {}, error: (m) => errors.push(String(m)) };
+ const unhandled = await withUnhandled(() => {
+ admissionChain(
+ Promise.resolve({ sessionId: 'ses_000000000000000000000000' }),
+ async () => { throw new Error('finish boom'); },
+ () => errors.push('onFailed'),
+ log,
+ );
+ });
+ assert.deepStrictEqual(unhandled, [], 'a rejection inside the async success handler must be caught');
+ assert.ok(errors.some((m) => /\[rtmp\] admission failed: Error: finish boom/.test(m)), 'it is logged');
+ assert.ok(!errors.includes('onFailed'), 'onFailed handles only a rejection of admission itself');
+});
+
+t('the coordinator handles a failing outbox prune: logged, not an unhandled rejection', async () => {
+ // The 6-hour timer discards the promise, so a rejection would end the coordinator unhandled.
+ const errors = [];
+ const unhandled = await withUnhandled(() => {
+ pruneOutbox({ prune: async () => { throw new Error('prune boom'); } }, { error: (m) => errors.push(String(m)) });
+ });
+ assert.deepStrictEqual(unhandled, []);
+ assert.ok(errors.some((m) => /prune boom/.test(m)), 'the failure is logged');
+ // A synchronous throw (an absent outbox) takes the same path, as the old try/catch did.
+ const syncErrors = [];
+ const syncUnhandled = await withUnhandled(() => {
+ pruneOutbox({ prune: () => { throw new Error('sync boom'); } }, { error: (m) => syncErrors.push(String(m)) });
+ });
+ assert.deepStrictEqual(syncUnhandled, []);
+ assert.ok(syncErrors.some((m) => /sync boom/.test(m)));
+});
+
+t('the restream worker boot awaits start(): a register failure is reported, not left unhandled', async () => {
+ const dir = tmpDir();
+ const preload = path.join(ROOT, 'test', 'fixtures', 'fail-worker-register.js');
+ const p = child('workers/restream-worker.js', { ...await testEnv(dir), NODE_OPTIONS: `--require ${preload}` });
+ const r = await p.exited;
+ assert.strictEqual(r.code, 1, p.output);
+ // Without the await, the boot IIFE resolves and the rejection is unhandled: Node prints its own
+ // stack and this process-owned line never appears.
+ assert.match(p.output, /\[restream-worker\] failed to start: Error: register boom/);
+});
+
+t.run();
@@ -8,6 +8,15 @@
@@ -8,6 +8,15 @@
*/
*/
const path = require('path');
const path = require('path');
+/**
+ * Best-effort housekeeping on a 6-hour timer: `prune` is async and the timer is fire-and-forget, so its rejection
+ * must be handled — an unhandled one ends the coordinator (Node 22), delaying output assignment and the events
+ * relay. Wrapping the call also turns a synchronous throw (an absent outbox) into the same logged path.
+ */
+function pruneOutbox(outbox, log = console) {
+ return Promise.resolve().then(() => outbox.prune()).catch((err) => log.error(`[coordinator] outbox prune failed (next time): ${err && err.message || err}`));
+}
+
if (require.main === module) {
if (require.main === module) {
(async () => {
(async () => {
require('dotenv').config({ path: process.env.OPENRE_ENV_FILE || path.join(process.cwd(), '.env') });
require('dotenv').config({ path: process.env.OPENRE_ENV_FILE || path.join(process.cwd(), '.env') });
@@ -23,7 +32,8 @@ if (require.main === module) {
@@ -23,7 +32,8 @@ if (require.main === module) {
const coordinator = createCoordinator({ rt, media, lineage });
const coordinator = createCoordinator({ rt, media, lineage });
coordinator.start();
coordinator.start();
console.log(`[coordinator] running (events relay ${rt.events.configured ? 'on' : 'off: EVENTS_URL/OV_OAUTH_CLIENT_SECRET not set'}, recordings ${media.configured ? `on → ${config.media.url} app ${config.media.appId}` : 'off'}, Live lineage ${lineage.enabled ? 'on' : 'off'})`);
console.log(`[coordinator] running (events relay ${rt.events.configured ? 'on' : 'off: EVENTS_URL/OV_OAUTH_CLIENT_SECRET not set'}, recordings ${media.configured ? `on → ${config.media.url} app ${config.media.appId}` : 'off'}, Live lineage ${lineage.enabled ? 'on' : 'off'})`);
- const prune = setInterval(() => { try { rt.events.outbox.prune(); } catch { /* next time */ } }, 6 * 60 * 60 * 1000);
+ // prune is async: a bare call would reject unhandled (Node ends the coordinator) instead of being caught.
+ const prune = setInterval(() => pruneOutbox(rt.events.outbox, console), 6 * 60 * 60 * 1000);
prune.unref();
prune.unref();
const shutdown = (sig) => {
const shutdown = (sig) => {
console.log(`[coordinator] ${sig}: stopping`);
console.log(`[coordinator] ${sig}: stopping`);
@@ -34,3 +44,5 @@ if (require.main === module) {
@@ -34,3 +44,5 @@ if (require.main === module) {
process.on('SIGINT', () => shutdown('SIGINT'));
process.on('SIGINT', () => shutdown('SIGINT'));
})().catch((err) => { console.error(`[coordinator] failed to start: ${err && err.stack || err}`); process.exit(1); });
})().catch((err) => { console.error(`[coordinator] failed to start: ${err && err.stack || err}`); process.exit(1); });
}
}
+
+module.exports = { pruneOutbox };
@@ -117,13 +117,13 @@ if (require.main === module) {
@@ -117,13 +117,13 @@ if (require.main === module) {
exitIfDrill(config, 'openre-restream-worker');
exitIfDrill(config, 'openre-restream-worker');
const rt = await openRuntime({ config });
const rt = await openRuntime({ config });
const worker = createRestreamWorker({ rt });
const worker = createRestreamWorker({ rt });
- worker.start();
+ await worker.start();
// If this process dies, its ffmpeg children must not keep pushing unsupervised: the outputs are
// 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.
// reassigned to another worker, and two pushes to one ingest make platforms drop both.
process.on('exit', () => {
process.on('exit', () => {
for (const r of worker.runners().values()) { try { if (r.proc) r.proc.kill('SIGKILL'); } catch { /* */ } }
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));
+ for (const sig of ['SIGTERM', 'SIGINT']) process.on(sig, () => worker.drain(sig).catch((err) => console.error(`[restream-worker] drain failed on ${sig}: ${err && (err.stack || err.message) || err}`)));
})().catch((err) => { console.error(`[restream-worker] failed to start: ${err && err.stack || err}`); process.exit(1); });
})().catch((err) => { console.error(`[restream-worker] failed to start: ${err && err.stack || err}`); process.exit(1); });
}
}
@@ -53,6 +53,16 @@ function listenInRange(server, min, max, skip = new Set()) {
@@ -53,6 +53,16 @@ function listenInRange(server, min, max, skip = new Set()) {
});
});
}
}
+/**
+ * Run `admission` then `onAdmitted`, with `onFailed` for a rejection of `admission` itself. `.then(onOk, onErr)`
+ * leaves a rejection of the (async) success handler — `sessions.finish` before the publisher went live — unhandled,
+ * and Node 22 ends the worker on an unhandled rejection, dropping every publisher on it. The trailing `.catch`
+ * routes both rejection sources to the log.
+ */
+function admissionChain(admission, onAdmitted, onFailed, log) {
+ return admission.then(onAdmitted, onFailed).catch((err) => log.error(`[rtmp] admission failed: ${err.stack || err}`));
+}
+
function createRtmpIngest({ rt, log = console, exit = (code) => process.exit(code) }) {
function createRtmpIngest({ rt, log = console, exit = (code) => process.exit(code) }) {
const { store, config } = rt;
const { store, config } = rt;
NmsLogger.setLogType(1); // errors only: the library's info lines would print stream paths
NmsLogger.setLogType(1); // errors only: the library's info lines would print stream paths
@@ -127,7 +137,7 @@ function createRtmpIngest({ rt, log = console, exit = (code) => process.exit(cod
@@ -127,7 +137,7 @@ function createRtmpIngest({ rt, log = console, exit = (code) => process.exit(cod
// NetStream.Publish.BadConnection without touching the first stream.
// NetStream.Publish.BadConnection without touching the first stream.
if (publishers.has(session.id)) return onPublish(invokeMessage);
if (publishers.has(session.id)) return onPublish(invokeMessage);
const streamPath = `/${session.appname}/${String(invokeMessage.streamName || '').split('?')[0]}`;
const streamPath = `/${session.appname}/${String(invokeMessage.streamName || '').split('?')[0]}`;
- admit(session, streamPath).then(async (adm) => {
+ admissionChain(admit(session, streamPath), async (adm) => {
if (!session.isStarting) {
if (!session.isStarting) {
// The publisher left while it was being admitted: its session ends without going live.
// The publisher left while it was being admitted: its session ends without going live.
if (adm.sessionId) await store.sessions.finish(adm.sessionId, { reason: 'publisher_disconnected', actor: `worker:${runtime.me.id}` });
if (adm.sessionId) await store.sessions.finish(adm.sessionId, { reason: 'publisher_disconnected', actor: `worker:${runtime.me.id}` });
@@ -139,7 +149,7 @@ function createRtmpIngest({ rt, log = console, exit = (code) => process.exit(cod
@@ -139,7 +149,7 @@ function createRtmpIngest({ rt, log = console, exit = (code) => process.exit(cod
log.error(`[rtmp] admission failed: ${err.stack || err}`);
log.error(`[rtmp] admission failed: ${err.stack || err}`);
admissions.set(session.id, { error: 'admission failed' });
admissions.set(session.id, { error: 'admission failed' });
if (session.isStarting) onPublish(invokeMessage);
if (session.isStarting) onPublish(invokeMessage);
- });
+ }, log);
};
};
}
}
@@ -313,8 +323,8 @@ if (require.main === module) {
@@ -313,8 +323,8 @@ if (require.main === module) {
console.error(`[rtmp] uncaught exception (worker keeps running): ${err && err.stack || err}`);
console.error(`[rtmp] uncaught exception (worker keeps running): ${err && err.stack || err}`);
if (!started) process.exit(1);
if (!started) process.exit(1);
});
});
- for (const sig of ['SIGTERM', 'SIGINT']) process.on(sig, () => worker.drain(sig));
+ for (const sig of ['SIGTERM', 'SIGINT']) process.on(sig, () => worker.drain(sig).catch((err) => console.error(`[rtmp] drain failed on ${sig}: ${err && (err.stack || err.message) || err}`)));
})().catch((err) => { console.error(`[rtmp-ingest] failed to start: ${err && err.stack || err}`); process.exit(1); });
})().catch((err) => { console.error(`[rtmp-ingest] failed to start: ${err && err.stack || err}`); process.exit(1); });
}
}
-module.exports = { createRtmpIngest, listenInRange };
+module.exports = { createRtmpIngest, listenInRange, admissionChain };
@@ -84,14 +84,14 @@ function createWorkerRuntime({ rt, kind, log = console, hooks = {}, exit = (code
@@ -84,14 +84,14 @@ function createWorkerRuntime({ rt, kind, log = console, hooks = {}, exit = (code
* them); async ones used to reject unhandled — a failure is now logged, sync or async. */
* them); async ones used to reject unhandled — a failure is now logged, sync or async. */
function call(name, ...args) {
function call(name, ...args) {
if (typeof hooks[name] !== 'function') return;
if (typeof hooks[name] !== 'function') return;
- Promise.resolve().then(() => hooks[name](...args)).catch((err) => log.error(`[${kind}] ${name}: ${err.message}`));
+ Promise.resolve().then(() => hooks[name](...args)).catch((err) => log.error(`[${kind}] ${name}: ${err && (err.stack || err.message) || err}`));
}
}
function schedule() {
function schedule() {
clearTimeout(timer);
clearTimeout(timer);
if (stopped) return;
if (stopped) return;
timer = setTimeout(async () => {
timer = setTimeout(async () => {
- try { await beat(); } catch (err) { log.error(`[${kind}] heartbeat failed: ${err.message}`); }
+ try { await beat(); } catch (err) { log.error(`[${kind}] heartbeat failed: ${err && (err.stack || err.message) || err}`); }
schedule();
schedule();
}, config.workers.heartbeatMs);
}, config.workers.heartbeatMs);
}
}
@@ -115,7 +115,7 @@ function createWorkerRuntime({ rt, kind, log = console, hooks = {}, exit = (code
@@ -115,7 +115,7 @@ function createWorkerRuntime({ rt, kind, log = console, hooks = {}, exit = (code
// The stopped row is written before the process goes: otherwise the worker stays ready/draining until its lease
// The stopped row is written before the process goes: otherwise the worker stays ready/draining until its lease
// expires and new sessions can be routed to a generation that is gone.
// expires and new sessions can be routed to a generation that is gone.
Promise.resolve().then(() => store.workers.stop(me.id, draining ? 'drained' : 'exit'))
Promise.resolve().then(() => store.workers.stop(me.id, draining ? 'drained' : 'exit'))
- .catch((err) => log.error(`[${kind}] could not record the stop: ${err.message}`))
+ .catch((err) => log.error(`[${kind}] could not record the stop: ${err && (err.stack || err.message) || err}`))
.finally(() => settle(hooks.onExit).finally(() => exit(0)));
.finally(() => settle(hooks.onExit).finally(() => exit(0)));
return true;
return true;
}
}
@@ -128,7 +128,14 @@ function createWorkerRuntime({ rt, kind, log = console, hooks = {}, exit = (code
@@ -128,7 +128,14 @@ function createWorkerRuntime({ rt, kind, log = console, hooks = {}, exit = (code
exitWhenIdle,
exitWhenIdle,
get me() { return me; },
get me() { return me; },
get draining() { return draining; },
get draining() { return draining; },
- stop() { stopped = true; clearTimeout(timer); if (me) store.workers.stop(me.id, 'stopped'); },
+ stop() {
+ stopped = true;
+ clearTimeout(timer);
+ // A close() caller does not wait for this write, but a rejection must still be handled: an
+ // unhandled one ends the process, and the stopped row would be lost without a trace.
+ if (me) return store.workers.stop(me.id, 'stopped').catch((err) => log.error(`[${kind}] could not record the stop: ${err && (err.stack || err.message) || err}`));
+ return undefined;
+ },
};
};
}
}
@@ -41,8 +41,8 @@ async function main(kind) {
@@ -41,8 +41,8 @@ async function main(kind) {
const config = load();
const config = load();
exitIfDrill(config, `openre-${kind}`);
exitIfDrill(config, `openre-${kind}`);
const w = createUnportedTransport({ rt: await openRuntime({ config }), kind });
const w = createUnportedTransport({ rt: await openRuntime({ config }), kind });
- w.start();
- for (const sig of ['SIGTERM', 'SIGINT']) process.on(sig, () => w.drain(sig));
+ await w.start();
+ for (const sig of ['SIGTERM', 'SIGINT']) process.on(sig, () => w.drain(sig).catch((err) => console.error(`[${kind}] drain failed on ${sig}: ${err && (err.stack || err.message) || err}`)));
}
}
module.exports = { createUnportedTransport, PORT_STATUS, main };
module.exports = { createUnportedTransport, PORT_STATUS, main };