openvibers/openrestream Commits
Require ready ingest workers for stored stream protocols (#9)
Changes
@@@ -11,8 +11,7 @@ const { createUiRouter } = require('./ui/routes');const pkg = require('../package.json');const SESSION_FLV_RE = /^(ses_[0-9A-HJKMNP-TV-Z]{26})\.flv$/;--// The worker kinds a switched slot needs (scripts/cutover-preflight.js checks the same two).--const WORKER_KINDS = Object.freeze(['rtmp-ingest', 'restream']);++const PROTOCOL_WORKERS = Object.freeze({ rtmp: 'rtmp-ingest', jsmpeg: 'jsmpeg', webrtc: 'webrtc' });function createApp({ rt, auth, keys, log = console, fetchImpl }) {const { config, store, events, db } = rt;@@@ -66,6 +65,17 @@ function createApp({ rt, auth, keys, log = console, fetchImpl }) {// scripts/cutover-preflight.js; a restore drill never publishes, so its relay check is skipped, never ok.const { createReadiness, skip } = require('openvibe-shared/ready');const workerView = async () => (await store.workers.alive()).map(w => ({ kind: w.kind, generation: w.generation, state: w.state, heartbeat_age_ms: Date.now() - w.heartbeat_at }));++ const requiredWorkerKinds = async () => {++ const rows = await db.prepare("SELECT protocols FROM stream_definitions WHERE state != 'archived'").all();++ const kinds = new Set();++ // Before a stream is created, retain the RTMP readiness baseline.++ if (!rows.length) kinds.add('rtmp-ingest');++ for (const row of rows) {++ for (const protocol of JSON.parse(row.protocols)) kinds.add(PROTOCOL_WORKERS[protocol]);++ }++ kinds.add('restream');++ return [...kinds];++ };const leaseView = async () => {const l = await db.prepare("SELECT holder, expires_at FROM leases WHERE name = 'coordinator'").get();return l ? { holder: l.holder, lease_valid: l.expires_at > Date.now() } : null;@@@ -80,10 +90,10 @@ function createApp({ rt, auth, keys, log = console, fetchImpl }) {{ name: 'db', required: true, description: 'PostgreSQL answers a query', check: async () => (await db.prepare('SELECT 1 AS ok').get()).ok === 1 },{ name: 'network_key', required: true, description: 'OpenVibe.Network RS256 key (sign-in and service tokens)', check: () => keys.loaded() || 'Network public key not loaded yet' },{-- name: 'workers', required: false, description: 'a ready rtmp-ingest and restream worker',++ name: 'workers', required: false, description: 'ready workers for stored stream protocols and restream',check: async () => {const alive = await workerView();-- const missing = WORKER_KINDS.filter(k => !alive.some(w => w.kind === k && w.state === 'ready'));++ const missing = (await requiredWorkerKinds()).filter(k => !alive.some(w => w.kind === k && w.state === 'ready'));return missing.length ? `no ready ${missing.join(' or ')} worker` : { ok: true, detail: { ready: alive.filter(w => w.state === 'ready').map(w => `${w.kind}#${w.generation}`) } };},},@@@ -26,6 +26,33 @@ t('boot', async () => {assert.ok(Array.isArray(ready.body.workers) && 'coordinator' in ready.body && 'store' in ready.body, 'the preflight fields stay');});++t('readiness follows the protocol of each stored stream definition', async () => {++ for (const [protocol, kind, otherKind] of [++ ['rtmp', 'rtmp-ingest', 'jsmpeg'],++ ['jsmpeg', 'jsmpeg', 'rtmp-ingest'],++ ['webrtc', 'webrtc', 'rtmp-ingest'],++ ]) {++ const h = await bootApi();++ try {++ await h.rt.store.definitions.create({ owner_subject: OWNER, protocols: [protocol] });++ for (const workerKind of ['restream', otherKind]) {++ const worker = await h.rt.store.workers.register({ kind: workerKind });++ await h.rt.store.workers.ready(worker.id);++ }++ const missing = await request(h.base, 'GET', '/api/ready');++ assert.ok(missing.body.degraded.includes('workers'), `${protocol} needs a ${kind} worker`);++++ const worker = await h.rt.store.workers.register({ kind });++ await h.rt.store.workers.ready(worker.id);++ const ready = await request(h.base, 'GET', '/api/ready');++ assert.ok(!ready.body.degraded.includes('workers'), `${protocol} has its required worker`);++ assert.strictEqual(ready.body.checks.workers.status, 'ok');++ } finally {++ await h.close();++ }++ }++});++t('anonymous and ungranted callers are refused with problem+json', async () => {const r = await request(api.base, 'GET', '/api/v1/streams');assert.strictEqual(r.status, 401);
@@ -11,8 +11,7 @@ const { createUiRouter } = require('./ui/routes');
@@ -11,8 +11,7 @@ const { createUiRouter } = require('./ui/routes');
const pkg = require('../package.json');
const pkg = require('../package.json');
const SESSION_FLV_RE = /^(ses_[0-9A-HJKMNP-TV-Z]{26})\.flv$/;
const SESSION_FLV_RE = /^(ses_[0-9A-HJKMNP-TV-Z]{26})\.flv$/;
-// The worker kinds a switched slot needs (scripts/cutover-preflight.js checks the same two).
-const WORKER_KINDS = Object.freeze(['rtmp-ingest', 'restream']);
+const PROTOCOL_WORKERS = Object.freeze({ rtmp: 'rtmp-ingest', jsmpeg: 'jsmpeg', webrtc: 'webrtc' });
function createApp({ rt, auth, keys, log = console, fetchImpl }) {
function createApp({ rt, auth, keys, log = console, fetchImpl }) {
const { config, store, events, db } = rt;
const { config, store, events, db } = rt;
@@ -66,6 +65,17 @@ function createApp({ rt, auth, keys, log = console, fetchImpl }) {
@@ -66,6 +65,17 @@ function createApp({ rt, auth, keys, log = console, fetchImpl }) {
// scripts/cutover-preflight.js; a restore drill never publishes, so its relay check is skipped, never ok.
// scripts/cutover-preflight.js; a restore drill never publishes, so its relay check is skipped, never ok.
const { createReadiness, skip } = require('openvibe-shared/ready');
const { createReadiness, skip } = require('openvibe-shared/ready');
const workerView = async () => (await store.workers.alive()).map(w => ({ kind: w.kind, generation: w.generation, state: w.state, heartbeat_age_ms: Date.now() - w.heartbeat_at }));
const workerView = async () => (await store.workers.alive()).map(w => ({ kind: w.kind, generation: w.generation, state: w.state, heartbeat_age_ms: Date.now() - w.heartbeat_at }));
+ const requiredWorkerKinds = async () => {
+ const rows = await db.prepare("SELECT protocols FROM stream_definitions WHERE state != 'archived'").all();
+ const kinds = new Set();
+ // Before a stream is created, retain the RTMP readiness baseline.
+ if (!rows.length) kinds.add('rtmp-ingest');
+ for (const row of rows) {
+ for (const protocol of JSON.parse(row.protocols)) kinds.add(PROTOCOL_WORKERS[protocol]);
+ }
+ kinds.add('restream');
+ return [...kinds];
+ };
const leaseView = async () => {
const leaseView = async () => {
const l = await db.prepare("SELECT holder, expires_at FROM leases WHERE name = 'coordinator'").get();
const l = await db.prepare("SELECT holder, expires_at FROM leases WHERE name = 'coordinator'").get();
return l ? { holder: l.holder, lease_valid: l.expires_at > Date.now() } : null;
return l ? { holder: l.holder, lease_valid: l.expires_at > Date.now() } : null;
@@ -80,10 +90,10 @@ function createApp({ rt, auth, keys, log = console, fetchImpl }) {
@@ -80,10 +90,10 @@ function createApp({ rt, auth, keys, log = console, fetchImpl }) {
{ name: 'db', required: true, description: 'PostgreSQL answers a query', check: async () => (await db.prepare('SELECT 1 AS ok').get()).ok === 1 },
{ name: 'db', required: true, description: 'PostgreSQL answers a query', check: async () => (await db.prepare('SELECT 1 AS ok').get()).ok === 1 },
{ name: 'network_key', required: true, description: 'OpenVibe.Network RS256 key (sign-in and service tokens)', check: () => keys.loaded() || 'Network public key not loaded yet' },
{ name: 'network_key', required: true, description: 'OpenVibe.Network RS256 key (sign-in and service tokens)', check: () => keys.loaded() || 'Network public key not loaded yet' },
{
{
- name: 'workers', required: false, description: 'a ready rtmp-ingest and restream worker',
+ name: 'workers', required: false, description: 'ready workers for stored stream protocols and restream',
check: async () => {
check: async () => {
const alive = await workerView();
const alive = await workerView();
- const missing = WORKER_KINDS.filter(k => !alive.some(w => w.kind === k && w.state === 'ready'));
+ const missing = (await requiredWorkerKinds()).filter(k => !alive.some(w => w.kind === k && w.state === 'ready'));
return missing.length ? `no ready ${missing.join(' or ')} worker` : { ok: true, detail: { ready: alive.filter(w => w.state === 'ready').map(w => `${w.kind}#${w.generation}`) } };
return missing.length ? `no ready ${missing.join(' or ')} worker` : { ok: true, detail: { ready: alive.filter(w => w.state === 'ready').map(w => `${w.kind}#${w.generation}`) } };
},
},
},
},
@@ -26,6 +26,33 @@ t('boot', async () => {
@@ -26,6 +26,33 @@ t('boot', async () => {
assert.ok(Array.isArray(ready.body.workers) && 'coordinator' in ready.body && 'store' in ready.body, 'the preflight fields stay');
assert.ok(Array.isArray(ready.body.workers) && 'coordinator' in ready.body && 'store' in ready.body, 'the preflight fields stay');
});
});
+t('readiness follows the protocol of each stored stream definition', async () => {
+ for (const [protocol, kind, otherKind] of [
+ ['rtmp', 'rtmp-ingest', 'jsmpeg'],
+ ['jsmpeg', 'jsmpeg', 'rtmp-ingest'],
+ ['webrtc', 'webrtc', 'rtmp-ingest'],
+ ]) {
+ const h = await bootApi();
+ try {
+ await h.rt.store.definitions.create({ owner_subject: OWNER, protocols: [protocol] });
+ for (const workerKind of ['restream', otherKind]) {
+ const worker = await h.rt.store.workers.register({ kind: workerKind });
+ await h.rt.store.workers.ready(worker.id);
+ }
+ const missing = await request(h.base, 'GET', '/api/ready');
+ assert.ok(missing.body.degraded.includes('workers'), `${protocol} needs a ${kind} worker`);
+
+ const worker = await h.rt.store.workers.register({ kind });
+ await h.rt.store.workers.ready(worker.id);
+ const ready = await request(h.base, 'GET', '/api/ready');
+ assert.ok(!ready.body.degraded.includes('workers'), `${protocol} has its required worker`);
+ assert.strictEqual(ready.body.checks.workers.status, 'ok');
+ } finally {
+ await h.close();
+ }
+ }
+});
+
t('anonymous and ungranted callers are refused with problem+json', async () => {
t('anonymous and ungranted callers are refused with problem+json', async () => {
const r = await request(api.base, 'GET', '/api/v1/streams');
const r = await request(api.base, 'GET', '/api/v1/streams');
assert.strictEqual(r.status, 401);
assert.strictEqual(r.status, 401);