openvibers/openrestream Commits
OPENRE_DRILL: restore-drill mode. openre-api serves reads from the restored copy and answers 503 openre.drill_read_only to writes, /play/ and sign-in; /api/ready reports mode drill. The coordinator and every transport worker refuse to be created (code OPENRE_DRILL), and their entry points exit before opening the database. The event relay and Media recording requests are off whatever else the env sets. This lets ovhost drill openre start a second instance safely.
Changes
@@@ -24,6 +24,18 @@ function createApp({ rt, auth, keys, log = console, fetchImpl }) {});app.use(cookieParser());++ // Restore drill (OPENRE_DRILL, ovhost drill): serve reads from the restored copy and nothing++ // else. No writes (they would only change the copy, but a drill must not look like it works),++ // no /play/ (it would pull from production's worker on loopback), no sign-in (it would redeem++ // codes at Network).++ if (config.drill) {++ app.use((req, res, next) => {++ const read = req.method === 'GET' || req.method === 'HEAD';++ if (read && !req.path.startsWith('/play/') && !req.path.startsWith('/auth/')) return next();++ return contracts.http.sendProblem(res, 503, 'openre.drill_read_only', { detail: 'this is a restore-drill instance (OPENRE_DRILL): reads only', ctx: req.ov });++ });++ }++app.get('/api/health', (_req, res) => res.json({ status: 'ok', service: 'openre-api', version: pkg.version, release: config.release }));// Ready = the database answers and the Network key is loaded. Worker and coordinator state is@@@ -42,6 +54,7 @@ function createApp({ rt, auth, keys, log = console, fetchImpl }) {} catch { /* reported as empty */ }res.status(ready ? 200 : 503).json({status: ready ? 'ready' : 'not_ready',++ ...(config.drill ? { mode: 'drill' } : {}),checks,store: { engine: 'sqlite', path_configured: Boolean(config.dbPath) },workers,@@@ -31,6 +31,11 @@ function load(env = process.env) {baseUrl: trim(env.BASE_URL || (isProduction ? 'https://openre.stream' : `http://localhost:${port}`)),// A release label for logs and worker generations (deploy.sh sets the git sha).release: env.OPENRE_RELEASE || 'dev',++ // Restore-drill mode (ovhost drill): the process serves reads from a restored copy of the++ // database and nothing else. The coordinator and every transport worker refuse to start, the++ // event relay and Media recording requests are off, the API refuses writes and /play/.++ // Never set in /etc/openvibe/openre.env (scripts/cutover-preflight.js checks that).++ drill: bool(env.OPENRE_DRILL, false),// ── Identity: OpenVibe.Network ──────────────────────────networkUrl: trim(env.OV_NETWORK_URL || 'https://openvibe.network'),@@@ -146,4 +151,24 @@ function load(env = process.env) {};}--module.exports = { load };++/**++ * Processes that lease sessions, run transports or call other services refuse to start in a++ * restore drill (OPENRE_DRILL): the coordinator and every transport worker. Throws with code++ * OPENRE_DRILL; entry points print the message and exit before opening the database.++ */++function assertNotDrill(config, what) {++ if (!config || !config.drill) return;++ const err = new Error(`${what} does not run in a restore drill (OPENRE_DRILL is set): it would lease sessions, run transports or call other services. Only openre-api runs in a drill.`);++ err.code = 'OPENRE_DRILL';++ throw err;++}++++/** Entry points: print why and exit before the database is opened. */++function exitIfDrill(config, what) {++ try { assertNotDrill(config, what); } catch (err) {++ console.error(`[${what}] ${err.message}`);++ process.exit(1);++ }++}++++module.exports = { load, assertNotDrill, exitIfDrill };@@@ -17,9 +17,11 @@* The event relay (outbox → OpenVibe.Events) runs in this process too.*/const { KINDS } = require('./store/workers');++const { assertNotDrill } = require('./config');function createCoordinator({ rt, media, log = console, holder = `coordinator:${process.pid}` }) {const { db, store, config, clock } = rt;++ assertNotDrill(config, 'The session coordinator');const now = () => clock.now();const lease = {take: db.prepare(`INSERT INTO leases (name, holder, expires_at) VALUES ('coordinator', @holder, @exp)@@@ -36,7 +36,8 @@ const TYPES = Object.freeze({* enqueue(envelope) must run inside the transaction that makes the change.*/function createEvents({ config, db, fetchImpl = globalThis.fetch, now = () => Date.now(), log = console }) {-- const configured = Boolean(config.events.url && config.oauth.clientSecret && config.events.publish);++ // A restore drill (OPENRE_DRILL) never publishes: its outbox rows describe a restored copy.++ const configured = Boolean(config.events.url && config.oauth.clientSecret && config.events.publish && !config.drill);let tokenProvider = null;if (configured) {tokenProvider = createServiceTokenClient({@@@ -28,7 +28,8 @@ function createMediaClient({ config, fetchImpl = globalThis.fetch }) {? createServiceTokenClient({ tokenUrl: `${config.networkInternalUrl}/oauth/token`, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, audience: 'openvibe.media', fetch: fetchImpl }): null;-- const configured = config.media.enabled && Boolean(config.media.auth === 'service' ? tokens : config.media.apiKey);++ // A restore drill (OPENRE_DRILL) never asks Media for anything.++ const configured = config.media.enabled && !config.drill && Boolean(config.media.auth === 'service' ? tokens : config.media.apiKey);async function authHeader() {if (tokens) return { Authorization: `Bearer ${await tokens.getToken()}` };nnew file mode 100644@@@ -0,0 +1,105 @@++'use strict';++// OPENRE_DRILL (ovhost restore drills): openre-api serves reads from a restored copy and nothing++// else; the coordinator and every transport worker refuse to start (before opening the database);++// no event relay and no Media calls, whatever else the production env file sets.++const assert = require('assert');++const fs = require('fs');++const path = require('path');++const { load } = require('../server/config');++const { createEvents } = require('../server/events');++const { createMediaClient } = require('../server/media-client');++const { createCoordinator } = require('../server/coordinator');++const { createRtmpIngest } = require('../workers/rtmp-ingest');++const { createRestreamWorker } = require('../workers/restream-worker');++const { createUnportedTransport } = require('../workers/unported-transport');++const { openDb } = require('../server/db');++const { runtime, bootApi, request, serviceToken, tmpDir, testEnv, child, suite, silent } = require('./helpers');++++const t = suite('drill-mode');++// What the production env file sets: with these, a non-drill process would publish and record.++const PROD_LIKE = { EVENTS_URL: 'http://127.0.0.1:9', OV_OAUTH_CLIENT_SECRET: 'x'.repeat(40), MEDIA_URL: 'http://127.0.0.1:9', MEDIA_API_KEY: 'k'.repeat(40) };++const reader = serviceToken('live', ['openre.stream.read', 'openre.stream.write', 'openre.session.read']);++++t('OPENRE_DRILL parses like the other switches and is off by default', () => {++ assert.strictEqual(load({}).drill, false);++ for (const v of ['1', 'true', 'on', 'yes']) assert.strictEqual(load({ OPENRE_DRILL: v }).drill, true, v);++ for (const v of ['', '0', 'false', 'off']) assert.strictEqual(load({ OPENRE_DRILL: v }).drill, false, v);++});++++t('no event relay and no Media client in a drill, even with production credentials', () => {++ const on = load(testEnv(tmpDir(), PROD_LIKE));++ const off = load(testEnv(tmpDir(), { ...PROD_LIKE, OPENRE_DRILL: '1' }));++ const db = openDb(':memory:');++ assert.strictEqual(createEvents({ config: on, db, log: silent }).configured, true, 'control: the same env publishes outside a drill');++ const ev = createEvents({ config: off, db, log: silent });++ assert.strictEqual(ev.configured, false);++ assert.strictEqual(ev.start(), false, 'the relay does not start');++ assert.strictEqual(createMediaClient({ config: on }).configured, true, 'control');++ assert.strictEqual(createMediaClient({ config: off }).configured, false);++ db.close();++});++++t('the coordinator and every transport worker refuse to be created in a drill', () => {++ const rt = runtime({ env: { OPENRE_DRILL: '1' } });++ const drill = (fn) => assert.throws(fn, (err) => err.code === 'OPENRE_DRILL' && /restore drill/.test(err.message));++ drill(() => createCoordinator({ rt, media: null, log: silent }));++ drill(() => createRtmpIngest({ rt, log: silent, exit: () => {} }));++ drill(() => createRestreamWorker({ rt, log: silent, exit: () => {} }));++ for (const kind of ['webrtc-ingest', 'sfu', 'jsmpeg']) drill(() => createUnportedTransport({ rt, kind, log: silent, exit: () => {} }));++ assert.strictEqual(rt.db.prepare('SELECT count(*) AS n FROM workers').get().n, 0, 'no generation registered');++ assert.strictEqual(rt.db.prepare('SELECT count(*) AS n FROM leases').get().n, 0, 'no coordinator lease taken');++});++++t('the entry points exit before opening the database', async () => {++ for (const script of ['workers/coordinator.js', 'workers/rtmp-ingest.js', 'workers/restream-worker.js', 'workers/sfu.js']) {++ const dir = tmpDir();++ const p = child(script, testEnv(dir, { ...PROD_LIKE, OPENRE_DRILL: '1' }));++ const r = await p.exited;++ assert.strictEqual(r.code, 1, `${script} exit code`);++ assert.match(p.output, /does not run in a restore drill/, script);++ assert.ok(!fs.existsSync(path.join(dir, 'openre.db')), `${script} must not create or touch the database`);++ }++});++++t('openre-api in a drill: reads work and match a normal instance; writes, /play/ and sign-in are refused', async () => {++ const dir = tmpDir();++ const normal = await bootApi({ dir });++ const created = await request(normal.base, 'POST', '/api/v1/streams', { token: reader, headers: { 'X-OV-Subject': 'usr_01J0000000000000000000000A' }, body: { title: 'Restored' } });++ assert.strictEqual(created.status, 201);++ const health = await request(normal.base, 'GET', '/api/health');++ const list = await request(normal.base, 'GET', '/api/v1/streams', { token: reader });++ const outboxBefore = normal.rt.db.prepare('SELECT count(*) AS n FROM event_outbox').get().n;++ await normal.close();++++ const drill = await bootApi({ dir, env: { ...PROD_LIKE, OPENRE_DRILL: '1' } });++ try {++ const ready = await request(drill.base, 'GET', '/api/ready');++ assert.strictEqual(ready.status, 200);++ assert.strictEqual(ready.body.mode, 'drill');++ assert.strictEqual(ready.body.events.configured, false);++ const h = await request(drill.base, 'GET', '/api/health');++ assert.strictEqual(h.text, health.text, '/api/health is byte-identical (ovhost compares it)');++ const l = await request(drill.base, 'GET', '/api/v1/streams', { token: reader });++ assert.strictEqual(l.status, 200);++ assert.deepStrictEqual(l.body, list.body, 'reads come from the restored copy');++ const robots = await request(drill.base, 'GET', '/robots.txt');++ assert.strictEqual(robots.status, 200);++++ const write = await request(drill.base, 'POST', '/api/v1/streams', { token: reader, headers: { 'X-OV-Subject': 'usr_01J0000000000000000000000A' }, body: { title: 'x' } });++ assert.strictEqual(write.status, 503);++ assert.strictEqual(write.body.code, 'openre.drill_read_only');++ const patch = await request(drill.base, 'PATCH', `/api/v1/streams/${created.body.stream.id}`, { token: reader, body: { title: 'y' } });++ assert.strictEqual(patch.status, 503);++ const play = await request(drill.base, 'GET', '/play/ses_01J00000000000000000000000.flv');++ assert.strictEqual(play.status, 503);++ const login = await request(drill.base, 'GET', '/auth/login');++ assert.strictEqual(login.status, 503);++ const count = drill.rt.db.prepare('SELECT count(*) AS n FROM stream_definitions').get().n;++ assert.strictEqual(count, 1, 'nothing was written');++ assert.strictEqual(drill.rt.db.prepare('SELECT count(*) AS n FROM event_outbox').get().n, outboxBefore, 'no event enqueued');++ } finally {++ await drill.close();++ }++});++++t.run();@@@ -10,11 +10,12 @@ const path = require('path');if (require.main === module) {require('dotenv').config({ path: process.env.OPENRE_ENV_FILE || path.join(process.cwd(), '.env') });-- const { load } = require('../server/config');++ const { load, exitIfDrill } = require('../server/config');const { openRuntime } = require('../server/store');const { createCoordinator } = require('../server/coordinator');const { createMediaClient } = require('../server/media-client');const config = load();++ exitIfDrill(config, 'openre-session-coordinator');const rt = openRuntime({ config });const media = createMediaClient({ config });const coordinator = createCoordinator({ rt, media });@@@ -110,9 +110,11 @@ function createRestreamWorker({ rt, log = console, exit = (code) => process.exitif (require.main === module) {require('dotenv').config({ path: process.env.OPENRE_ENV_FILE || path.join(process.cwd(), '.env') });-- const { load } = require('../server/config');++ const { load, exitIfDrill } = require('../server/config');const { openRuntime } = require('../server/store');-- const rt = openRuntime({ config: load() });++ const config = load();++ exitIfDrill(config, 'openre-restream-worker');++ const rt = openRuntime({ config });const worker = createRestreamWorker({ rt });worker.start();// If this process dies, its ffmpeg children must not keep pushing unsupervised: the outputs are@@@ -261,9 +261,11 @@ function createRtmpIngest({ rt, log = console, exit = (code) => process.exit(codif (require.main === module) {require('dotenv').config({ path: process.env.OPENRE_ENV_FILE || path.join(process.cwd(), '.env') });-- const { load } = require('../server/config');++ const { load, exitIfDrill } = require('../server/config');const { openRuntime } = require('../server/store');-- const rt = openRuntime({ config: load() });++ const config = load();++ exitIfDrill(config, 'openre-rtmp-ingest');++ const rt = openRuntime({ config });const worker = createRtmpIngest({ rt });let started = false;worker.start().then(() => { started = true; }, (err) => { console.error(`[rtmp] failed to start: ${err.stack || err}`); process.exit(1); });@@@ -18,6 +18,8 @@* 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.*/++const { assertNotDrill } = require('../server/config');++/** Run a cleanup hook, but never let it keep the process from exiting (5 s at most). */function settle(fn) {return Promise.race([@@@ -28,6 +30,7 @@ function settle(fn) {function createWorkerRuntime({ rt, kind, log = console, hooks = {}, exit = (code) => process.exit(code) }) {const { store, config } = rt;++ assertNotDrill(config, `The ${kind} worker`);let me = null;let timer = null;let draining = false;@@@ -36,9 +36,11 @@ function createUnportedTransport({ rt, kind, log = console, exit = (code) => profunction main(kind) {require('dotenv').config({ path: process.env.OPENRE_ENV_FILE || path.join(process.cwd(), '.env') });-- const { load } = require('../server/config');++ const { load, exitIfDrill } = require('../server/config');const { openRuntime } = require('../server/store');-- const w = createUnportedTransport({ rt: openRuntime({ config: load() }), kind });++ const config = load();++ exitIfDrill(config, `openre-${kind}`);++ const w = createUnportedTransport({ rt: openRuntime({ config }), kind });w.start();for (const sig of ['SIGTERM', 'SIGINT']) process.on(sig, () => w.drain(sig));}
@@ -24,6 +24,18 @@ function createApp({ rt, auth, keys, log = console, fetchImpl }) {
@@ -24,6 +24,18 @@ function createApp({ rt, auth, keys, log = console, fetchImpl }) {
});
});
app.use(cookieParser());
app.use(cookieParser());
+ // Restore drill (OPENRE_DRILL, ovhost drill): serve reads from the restored copy and nothing
+ // else. No writes (they would only change the copy, but a drill must not look like it works),
+ // no /play/ (it would pull from production's worker on loopback), no sign-in (it would redeem
+ // codes at Network).
+ if (config.drill) {
+ app.use((req, res, next) => {
+ const read = req.method === 'GET' || req.method === 'HEAD';
+ if (read && !req.path.startsWith('/play/') && !req.path.startsWith('/auth/')) return next();
+ return contracts.http.sendProblem(res, 503, 'openre.drill_read_only', { detail: 'this is a restore-drill instance (OPENRE_DRILL): reads only', ctx: req.ov });
+ });
+ }
+
app.get('/api/health', (_req, res) => res.json({ status: 'ok', service: 'openre-api', version: pkg.version, release: config.release }));
app.get('/api/health', (_req, res) => res.json({ status: 'ok', service: 'openre-api', version: pkg.version, release: config.release }));
// Ready = the database answers and the Network key is loaded. Worker and coordinator state is
// Ready = the database answers and the Network key is loaded. Worker and coordinator state is
@@ -42,6 +54,7 @@ function createApp({ rt, auth, keys, log = console, fetchImpl }) {
@@ -42,6 +54,7 @@ function createApp({ rt, auth, keys, log = console, fetchImpl }) {
} catch { /* reported as empty */ }
} catch { /* reported as empty */ }
res.status(ready ? 200 : 503).json({
res.status(ready ? 200 : 503).json({
status: ready ? 'ready' : 'not_ready',
status: ready ? 'ready' : 'not_ready',
+ ...(config.drill ? { mode: 'drill' } : {}),
checks,
checks,
store: { engine: 'sqlite', path_configured: Boolean(config.dbPath) },
store: { engine: 'sqlite', path_configured: Boolean(config.dbPath) },
workers,
workers,
@@ -31,6 +31,11 @@ function load(env = process.env) {
@@ -31,6 +31,11 @@ function load(env = process.env) {
baseUrl: trim(env.BASE_URL || (isProduction ? 'https://openre.stream' : `http://localhost:${port}`)),
baseUrl: trim(env.BASE_URL || (isProduction ? 'https://openre.stream' : `http://localhost:${port}`)),
// A release label for logs and worker generations (deploy.sh sets the git sha).
// A release label for logs and worker generations (deploy.sh sets the git sha).
release: env.OPENRE_RELEASE || 'dev',
release: env.OPENRE_RELEASE || 'dev',
+ // Restore-drill mode (ovhost drill): the process serves reads from a restored copy of the
+ // database and nothing else. The coordinator and every transport worker refuse to start, the
+ // event relay and Media recording requests are off, the API refuses writes and /play/.
+ // Never set in /etc/openvibe/openre.env (scripts/cutover-preflight.js checks that).
+ drill: bool(env.OPENRE_DRILL, false),
// ── Identity: OpenVibe.Network ──────────────────────────
// ── Identity: OpenVibe.Network ──────────────────────────
networkUrl: trim(env.OV_NETWORK_URL || 'https://openvibe.network'),
networkUrl: trim(env.OV_NETWORK_URL || 'https://openvibe.network'),
@@ -146,4 +151,24 @@ function load(env = process.env) {
@@ -146,4 +151,24 @@ function load(env = process.env) {
};
};
}
}
-module.exports = { load };
+/**
+ * Processes that lease sessions, run transports or call other services refuse to start in a
+ * restore drill (OPENRE_DRILL): the coordinator and every transport worker. Throws with code
+ * OPENRE_DRILL; entry points print the message and exit before opening the database.
+ */
+function assertNotDrill(config, what) {
+ if (!config || !config.drill) return;
+ const err = new Error(`${what} does not run in a restore drill (OPENRE_DRILL is set): it would lease sessions, run transports or call other services. Only openre-api runs in a drill.`);
+ err.code = 'OPENRE_DRILL';
+ throw err;
+}
+
+/** Entry points: print why and exit before the database is opened. */
+function exitIfDrill(config, what) {
+ try { assertNotDrill(config, what); } catch (err) {
+ console.error(`[${what}] ${err.message}`);
+ process.exit(1);
+ }
+}
+
+module.exports = { load, assertNotDrill, exitIfDrill };
@@ -17,9 +17,11 @@
@@ -17,9 +17,11 @@
* The event relay (outbox → OpenVibe.Events) runs in this process too.
* The event relay (outbox → OpenVibe.Events) runs in this process too.
*/
*/
const { KINDS } = require('./store/workers');
const { KINDS } = require('./store/workers');
+const { assertNotDrill } = require('./config');
function createCoordinator({ rt, media, log = console, holder = `coordinator:${process.pid}` }) {
function createCoordinator({ rt, media, log = console, holder = `coordinator:${process.pid}` }) {
const { db, store, config, clock } = rt;
const { db, store, config, clock } = rt;
+ assertNotDrill(config, 'The session coordinator');
const now = () => clock.now();
const now = () => clock.now();
const lease = {
const lease = {
take: db.prepare(`INSERT INTO leases (name, holder, expires_at) VALUES ('coordinator', @holder, @exp)
take: db.prepare(`INSERT INTO leases (name, holder, expires_at) VALUES ('coordinator', @holder, @exp)
@@ -36,7 +36,8 @@ const TYPES = Object.freeze({
@@ -36,7 +36,8 @@ const TYPES = Object.freeze({
* enqueue(envelope) must run inside the transaction that makes the change.
* enqueue(envelope) must run inside the transaction that makes the change.
*/
*/
function createEvents({ config, db, fetchImpl = globalThis.fetch, now = () => Date.now(), log = console }) {
function createEvents({ config, db, fetchImpl = globalThis.fetch, now = () => Date.now(), log = console }) {
- const configured = Boolean(config.events.url && config.oauth.clientSecret && config.events.publish);
+ // A restore drill (OPENRE_DRILL) never publishes: its outbox rows describe a restored copy.
+ const configured = Boolean(config.events.url && config.oauth.clientSecret && config.events.publish && !config.drill);
let tokenProvider = null;
let tokenProvider = null;
if (configured) {
if (configured) {
tokenProvider = createServiceTokenClient({
tokenProvider = createServiceTokenClient({
@@ -28,7 +28,8 @@ function createMediaClient({ config, fetchImpl = globalThis.fetch }) {
@@ -28,7 +28,8 @@ function createMediaClient({ config, fetchImpl = globalThis.fetch }) {
? createServiceTokenClient({ tokenUrl: `${config.networkInternalUrl}/oauth/token`, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, audience: 'openvibe.media', fetch: fetchImpl })
? createServiceTokenClient({ tokenUrl: `${config.networkInternalUrl}/oauth/token`, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, audience: 'openvibe.media', fetch: fetchImpl })
: null;
: null;
- const configured = config.media.enabled && Boolean(config.media.auth === 'service' ? tokens : config.media.apiKey);
+ // A restore drill (OPENRE_DRILL) never asks Media for anything.
+ const configured = config.media.enabled && !config.drill && Boolean(config.media.auth === 'service' ? tokens : config.media.apiKey);
async function authHeader() {
async function authHeader() {
if (tokens) return { Authorization: `Bearer ${await tokens.getToken()}` };
if (tokens) return { Authorization: `Bearer ${await tokens.getToken()}` };
new file mode 100644
new file mode 100644
@@ -0,0 +1,105 @@
@@ -0,0 +1,105 @@
+'use strict';
+// OPENRE_DRILL (ovhost restore drills): openre-api serves reads from a restored copy and nothing
+// else; the coordinator and every transport worker refuse to start (before opening the database);
+// no event relay and no Media calls, whatever else the production env file sets.
+const assert = require('assert');
+const fs = require('fs');
+const path = require('path');
+const { load } = require('../server/config');
+const { createEvents } = require('../server/events');
+const { createMediaClient } = require('../server/media-client');
+const { createCoordinator } = require('../server/coordinator');
+const { createRtmpIngest } = require('../workers/rtmp-ingest');
+const { createRestreamWorker } = require('../workers/restream-worker');
+const { createUnportedTransport } = require('../workers/unported-transport');
+const { openDb } = require('../server/db');
+const { runtime, bootApi, request, serviceToken, tmpDir, testEnv, child, suite, silent } = require('./helpers');
+
+const t = suite('drill-mode');
+// What the production env file sets: with these, a non-drill process would publish and record.
+const PROD_LIKE = { EVENTS_URL: 'http://127.0.0.1:9', OV_OAUTH_CLIENT_SECRET: 'x'.repeat(40), MEDIA_URL: 'http://127.0.0.1:9', MEDIA_API_KEY: 'k'.repeat(40) };
+const reader = serviceToken('live', ['openre.stream.read', 'openre.stream.write', 'openre.session.read']);
+
+t('OPENRE_DRILL parses like the other switches and is off by default', () => {
+ assert.strictEqual(load({}).drill, false);
+ for (const v of ['1', 'true', 'on', 'yes']) assert.strictEqual(load({ OPENRE_DRILL: v }).drill, true, v);
+ for (const v of ['', '0', 'false', 'off']) assert.strictEqual(load({ OPENRE_DRILL: v }).drill, false, v);
+});
+
+t('no event relay and no Media client in a drill, even with production credentials', () => {
+ const on = load(testEnv(tmpDir(), PROD_LIKE));
+ const off = load(testEnv(tmpDir(), { ...PROD_LIKE, OPENRE_DRILL: '1' }));
+ const db = openDb(':memory:');
+ assert.strictEqual(createEvents({ config: on, db, log: silent }).configured, true, 'control: the same env publishes outside a drill');
+ const ev = createEvents({ config: off, db, log: silent });
+ assert.strictEqual(ev.configured, false);
+ assert.strictEqual(ev.start(), false, 'the relay does not start');
+ assert.strictEqual(createMediaClient({ config: on }).configured, true, 'control');
+ assert.strictEqual(createMediaClient({ config: off }).configured, false);
+ db.close();
+});
+
+t('the coordinator and every transport worker refuse to be created in a drill', () => {
+ const rt = runtime({ env: { OPENRE_DRILL: '1' } });
+ const drill = (fn) => assert.throws(fn, (err) => err.code === 'OPENRE_DRILL' && /restore drill/.test(err.message));
+ drill(() => createCoordinator({ rt, media: null, log: silent }));
+ drill(() => createRtmpIngest({ rt, log: silent, exit: () => {} }));
+ drill(() => createRestreamWorker({ rt, log: silent, exit: () => {} }));
+ for (const kind of ['webrtc-ingest', 'sfu', 'jsmpeg']) drill(() => createUnportedTransport({ rt, kind, log: silent, exit: () => {} }));
+ assert.strictEqual(rt.db.prepare('SELECT count(*) AS n FROM workers').get().n, 0, 'no generation registered');
+ assert.strictEqual(rt.db.prepare('SELECT count(*) AS n FROM leases').get().n, 0, 'no coordinator lease taken');
+});
+
+t('the entry points exit before opening the database', async () => {
+ for (const script of ['workers/coordinator.js', 'workers/rtmp-ingest.js', 'workers/restream-worker.js', 'workers/sfu.js']) {
+ const dir = tmpDir();
+ const p = child(script, testEnv(dir, { ...PROD_LIKE, OPENRE_DRILL: '1' }));
+ const r = await p.exited;
+ assert.strictEqual(r.code, 1, `${script} exit code`);
+ assert.match(p.output, /does not run in a restore drill/, script);
+ assert.ok(!fs.existsSync(path.join(dir, 'openre.db')), `${script} must not create or touch the database`);
+ }
+});
+
+t('openre-api in a drill: reads work and match a normal instance; writes, /play/ and sign-in are refused', async () => {
+ const dir = tmpDir();
+ const normal = await bootApi({ dir });
+ const created = await request(normal.base, 'POST', '/api/v1/streams', { token: reader, headers: { 'X-OV-Subject': 'usr_01J0000000000000000000000A' }, body: { title: 'Restored' } });
+ assert.strictEqual(created.status, 201);
+ const health = await request(normal.base, 'GET', '/api/health');
+ const list = await request(normal.base, 'GET', '/api/v1/streams', { token: reader });
+ const outboxBefore = normal.rt.db.prepare('SELECT count(*) AS n FROM event_outbox').get().n;
+ await normal.close();
+
+ const drill = await bootApi({ dir, env: { ...PROD_LIKE, OPENRE_DRILL: '1' } });
+ try {
+ const ready = await request(drill.base, 'GET', '/api/ready');
+ assert.strictEqual(ready.status, 200);
+ assert.strictEqual(ready.body.mode, 'drill');
+ assert.strictEqual(ready.body.events.configured, false);
+ const h = await request(drill.base, 'GET', '/api/health');
+ assert.strictEqual(h.text, health.text, '/api/health is byte-identical (ovhost compares it)');
+ const l = await request(drill.base, 'GET', '/api/v1/streams', { token: reader });
+ assert.strictEqual(l.status, 200);
+ assert.deepStrictEqual(l.body, list.body, 'reads come from the restored copy');
+ const robots = await request(drill.base, 'GET', '/robots.txt');
+ assert.strictEqual(robots.status, 200);
+
+ const write = await request(drill.base, 'POST', '/api/v1/streams', { token: reader, headers: { 'X-OV-Subject': 'usr_01J0000000000000000000000A' }, body: { title: 'x' } });
+ assert.strictEqual(write.status, 503);
+ assert.strictEqual(write.body.code, 'openre.drill_read_only');
+ const patch = await request(drill.base, 'PATCH', `/api/v1/streams/${created.body.stream.id}`, { token: reader, body: { title: 'y' } });
+ assert.strictEqual(patch.status, 503);
+ const play = await request(drill.base, 'GET', '/play/ses_01J00000000000000000000000.flv');
+ assert.strictEqual(play.status, 503);
+ const login = await request(drill.base, 'GET', '/auth/login');
+ assert.strictEqual(login.status, 503);
+ const count = drill.rt.db.prepare('SELECT count(*) AS n FROM stream_definitions').get().n;
+ assert.strictEqual(count, 1, 'nothing was written');
+ assert.strictEqual(drill.rt.db.prepare('SELECT count(*) AS n FROM event_outbox').get().n, outboxBefore, 'no event enqueued');
+ } finally {
+ await drill.close();
+ }
+});
+
+t.run();
@@ -10,11 +10,12 @@ const path = require('path');
@@ -10,11 +10,12 @@ const path = require('path');
if (require.main === module) {
if (require.main === module) {
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') });
- const { load } = require('../server/config');
+ const { load, exitIfDrill } = require('../server/config');
const { openRuntime } = require('../server/store');
const { openRuntime } = require('../server/store');
const { createCoordinator } = require('../server/coordinator');
const { createCoordinator } = require('../server/coordinator');
const { createMediaClient } = require('../server/media-client');
const { createMediaClient } = require('../server/media-client');
const config = load();
const config = load();
+ exitIfDrill(config, 'openre-session-coordinator');
const rt = openRuntime({ config });
const rt = openRuntime({ config });
const media = createMediaClient({ config });
const media = createMediaClient({ config });
const coordinator = createCoordinator({ rt, media });
const coordinator = createCoordinator({ rt, media });
@@ -110,9 +110,11 @@ function createRestreamWorker({ rt, log = console, exit = (code) => process.exit
@@ -110,9 +110,11 @@ function createRestreamWorker({ rt, log = console, exit = (code) => process.exit
if (require.main === module) {
if (require.main === module) {
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') });
- const { load } = require('../server/config');
+ const { load, exitIfDrill } = require('../server/config');
const { openRuntime } = require('../server/store');
const { openRuntime } = require('../server/store');
- const rt = openRuntime({ config: load() });
+ const config = load();
+ exitIfDrill(config, 'openre-restream-worker');
+ const rt = openRuntime({ config });
const worker = createRestreamWorker({ rt });
const worker = createRestreamWorker({ rt });
worker.start();
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
@@ -261,9 +261,11 @@ function createRtmpIngest({ rt, log = console, exit = (code) => process.exit(cod
@@ -261,9 +261,11 @@ function createRtmpIngest({ rt, log = console, exit = (code) => process.exit(cod
if (require.main === module) {
if (require.main === module) {
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') });
- const { load } = require('../server/config');
+ const { load, exitIfDrill } = require('../server/config');
const { openRuntime } = require('../server/store');
const { openRuntime } = require('../server/store');
- const rt = openRuntime({ config: load() });
+ const config = load();
+ exitIfDrill(config, 'openre-rtmp-ingest');
+ const rt = openRuntime({ config });
const worker = createRtmpIngest({ rt });
const worker = createRtmpIngest({ rt });
let started = false;
let started = false;
worker.start().then(() => { started = true; }, (err) => { console.error(`[rtmp] failed to start: ${err.stack || err}`); process.exit(1); });
worker.start().then(() => { started = true; }, (err) => { console.error(`[rtmp] failed to start: ${err.stack || err}`); process.exit(1); });
@@ -18,6 +18,8 @@
@@ -18,6 +18,8 @@
* The heartbeat (every heartbeatMs) is also the lease renewal for every session the worker owns.
* 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.
* A worker never depends on openre-api being up; it needs only the database.
*/
*/
+const { assertNotDrill } = require('../server/config');
+
/** Run a cleanup hook, but never let it keep the process from exiting (5 s at most). */
/** Run a cleanup hook, but never let it keep the process from exiting (5 s at most). */
function settle(fn) {
function settle(fn) {
return Promise.race([
return Promise.race([
@@ -28,6 +30,7 @@ function settle(fn) {
@@ -28,6 +30,7 @@ function settle(fn) {
function createWorkerRuntime({ rt, kind, log = console, hooks = {}, exit = (code) => process.exit(code) }) {
function createWorkerRuntime({ rt, kind, log = console, hooks = {}, exit = (code) => process.exit(code) }) {
const { store, config } = rt;
const { store, config } = rt;
+ assertNotDrill(config, `The ${kind} worker`);
let me = null;
let me = null;
let timer = null;
let timer = null;
let draining = false;
let draining = false;
@@ -36,9 +36,11 @@ function createUnportedTransport({ rt, kind, log = console, exit = (code) => pro
@@ -36,9 +36,11 @@ function createUnportedTransport({ rt, kind, log = console, exit = (code) => pro
function main(kind) {
function main(kind) {
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') });
- const { load } = require('../server/config');
+ const { load, exitIfDrill } = require('../server/config');
const { openRuntime } = require('../server/store');
const { openRuntime } = require('../server/store');
- const w = createUnportedTransport({ rt: openRuntime({ config: load() }), kind });
+ const config = load();
+ exitIfDrill(config, `openre-${kind}`);
+ const w = createUnportedTransport({ rt: openRuntime({ config }), kind });
w.start();
w.start();
for (const sig of ['SIGTERM', 'SIGINT']) process.on(sig, () => w.drain(sig));
for (const sig of ['SIGTERM', 'SIGINT']) process.on(sig, () => w.drain(sig));
}
}