openvibers/openvibe.media Commits
Move Media's shutdown onto openvibe-sdk/service gracefulStop with the 70 s drain (#33)
Changes
@@@ -45,7 +45,7 @@ without `DATABASE_URL`, development uses an embedded PGlite database in `data/pgexport and deletion), OpenVibe.Live (`live.lineage.resolve`), OpenVibe.Events (the outbox relay andthe account and revocation subscriptions)- Backblaze B2 and Cloudflare R2 (S3 API) when configured; `ffmpeg`/`ffprobe` on the host--- `openvibe-contracts` v0.79.0, `openvibe-sdk` v0.21.2 (tokens, events outbox, per-actor limits),++- `openvibe-contracts` v0.79.0, `openvibe-sdk` v0.28.0 (tokens, events outbox, per-actor limits),`openvibe-shared` v2.5.0, pinned by release tarball## Capabilities@@@ -115,7 +115,7 @@ server/storage.orphans.scan (monthly storage orphan report, service-wide, report only),storage.move.cleanup (the alert for placement moves whose delete keeps failing, service-wide)client-ip.js trust proxy = loopback; req.ip is the only client address--(openvibe-shared v2.5.0, openvibe-contracts v0.79.0, openvibe-sdk v0.21.2: pinned release tarballs, installed by npm)++(openvibe-shared v2.5.0, openvibe-contracts v0.79.0, openvibe-sdk v0.28.0: pinned release tarballs, installed by npm)scripts/smoke-test.sh end-to-end smoke test (boots a temp instance)scripts/reconcile-objects.js / object-invariant.js / object-drift-report.js / no-good-copy-report.js object-model operator toolsscripts/media-jobs.js list jobs, run the size-invariant scan (dry run by default), approve/cancel proposals@@@ -18,7 +18,7 @@"iovalkey": "^0.4.0","multer": "^1.4.5-lts.1","openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.0",-- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.21.2",++ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.28.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.5.0","pg": "^8.23.0","sharp": "^0.33.4"@@@ -2026,9 +2026,9 @@}},"node_modules/openvibe-sdk": {-- "version": "0.21.2",-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.21.2",-- "integrity": "sha512-k04b9r2FCYVIhb8KLNrr5y43eb1l3UKzYNKl9m/vqMRAjZGh4p3pI2I/8UFU9bosKD8KAS121OO4rH8NUtx4sg==",++ "version": "0.28.0",++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.28.0",++ "integrity": "sha512-jjLisFDWBN+xDVsixa5BJRr3355kRijmvOh1/9rz4s1yVb3wZWCbpTLNcOvXm57ksnRoqK57393W38G0Ey3PFw==","license": "MIT","engines": {"node": ">=20"@@@ -22,7 +22,7 @@"iovalkey": "^0.4.0","multer": "^1.4.5-lts.1","openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.0",-- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.21.2",++ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.28.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.5.0","pg": "^8.23.0","sharp": "^0.33.4"@@@ -27,6 +27,7 @@ if (drill.enabled) {const fs = require('fs');const express = require('express');++const { gracefulStop } = require('openvibe-sdk/service');const config = require('./config');const db = require('./db/database');const auth = require('./auth');@@@ -325,32 +326,35 @@ const ready = (async () => {// A drill whose port is taken stops instead of running unready.if (drill.enabled) server.once('error', (err) => { console.error(`[Drill] HTTP server: ${err.message}`); process.exit(1); });-- let shuttingDown = false;-- function shutdown(signal) {-- if (shuttingDown) return;-- shuttingDown = true;-- console.log(`[Media] ${signal} received — shutting down`);-- try { recorder.stopAll(); } catch { /* */ }-- try { vodStorage.stop(); } catch { /* */ }-- try { healthJob.stop(); } catch { /* */ }-- try { require('./objects/verify-job').stop(); } catch { /* */ }-- try { require('./jobs/worker').stop(); } catch { /* */ }-- try { require('./objects/owner-subject-job').stop(); } catch { /* */ }-- try { auth.stopJwksRefresh(); } catch { /* */ }-- try { require('./events')._reset(); } catch { /* */ }-- try { const v = require('./actor-limits').valkey(); if (v) v.close().catch(() => {}); } catch { /* */ }-- for (const t of timers) clearInterval(t);-- server.close(() => {-- db.close().catch(() => {}).finally(() => process.exit(0));-- });-- // Recordings get STOP_GRACE_MS to flush trailers; don't hang forever.-- setTimeout(() => {-- console.warn('[Media] Forced exit after shutdown grace');-- db.close().catch(() => {}).finally(() => process.exit(0));-- }, 70_000).unref();-- }-- process.on('SIGINT', () => shutdown('SIGINT'));-- process.on('SIGTERM', () => shutdown('SIGTERM'));++ // SIGTERM/SIGINT (openvibe-sdk/service; docs/service.md's handles family): stopping() turns true at++ // once, the recorder stops first, then the background handles stop, the events outbox is reset and the++ // database closes. The recorder is the first `stop` step so it runs before the worker and the pollers,++ // as the hand-written shutdown ran it: stopAll() signals FFmpeg, which then gets VOD_STOP_GRACE_MS to++ // flush trailers. The drain gets 60 s so in-flight uploads and large downloads are not cut at the kit's++ // 4 s default; the whole stop is bounded at 70 s, the systemd unit's TimeoutStopSec (80 s) leaving room++ // (deploy/systemd/openvibe-media.service). Media exits 0 past the deadline, like the 5 s family. A step++ // that throws is logged and the stop goes on: the try/catch the hand-written shutdown carried lives in++ // the kit now.++ gracefulStop({++ name: 'Media',++ server,++ drainMs: 60000,++ stop: [++ () => recorder.stopAll(),++ () => vodStorage.stop(),++ () => healthJob.stop(),++ () => require('./objects/verify-job').stop(),++ () => require('./jobs/worker').stop(),++ () => require('./objects/owner-subject-job').stop(),++ () => auth.stopJwksRefresh(),++ () => require('./events')._reset(),++ () => { const v = require('./actor-limits').valkey(); if (v) v.close().catch(() => {}); },++ () => { for (const t of timers) clearInterval(t); },++ ],++ close: [() => db.close()],++ deadlineMs: 70000,++ deadlineExitCode: 0,++ });return { app, server };})();ready.catch((err) => {@@@ -9,6 +9,16 @@ import { fileURLToPath } from 'node:url';const require = createRequire(import.meta.url);const root = path.join(path.dirname(fileURLToPath(import.meta.url)), '..', '..');const { createTestDb } = require('openvibe-sdk/testing');++// openvibe-sdk >= 0.26 createTestDb holds a lease pg.Client (application_name *-test-lease) whose socket is not unref'd,++// so a test file that ends without process.exit() never exits. Unref it here; the lease still holds while the process lives.++// Drop this once Media pins openvibe-sdk >= 0.31.1, which unrefs the lease itself.++const pg = require(require.resolve('pg', { paths: [require.resolve('openvibe-sdk/testing')] }));++const connect = pg.Client.prototype.connect;++pg.Client.prototype.connect = function (...args) {++ const r = connect.apply(this, args);++ if (String(this.connectionParameters && this.connectionParameters.application_name).endsWith('-test-lease')) this.connection.stream.unref();++ return r;++};const t = await createTestDb({ migrations: path.join(root, 'migrations'), store: process.env.MEDIA_TEST_STORE || 'pglite', service: 'media', max: 4 });// Tests insert rows with small explicit ids. A PostgreSQL identity would continue after them, so generated ids// are set to start at 100000 here, clear of every id a test names.nnew file mode 100644@@@ -0,0 +1,133 @@++'use strict';++/**++ * openvibe-sdk/service in Media (plan B4): the SIGTERM/SIGINT shutdown is the kit's gracefulStop, not++ * the hand-written shutdown()/process.on pair it replaces. It runs the stop steps (the recorder first —++ * stopAll() signalling FFmpeg before the worker and pollers stop — the tiering sweep, health job, verify++ * job, job worker, owner-subject job, JWKS refresher and events outbox stopped, the actor-limits Valkey++ * closed, the timers cleared), drains the server for 60 s so in-flight uploads and large downloads are++ * not cut at the kit's 4 s default, then closes the database; the whole stop is bounded at 70 s like the++ * systemd unit, and Media exits 0 past the deadline. A failing step is logged and the stop goes on.++ * The static half reads server/index.js; the behavioural half drives gracefulStop with the same option++ * shape Media passes, with exports stubbed (/mirrors OpenVibe.Sources/test/service-kit.test.js).++ */++const assert = require('assert');++const fs = require('fs');++const path = require('path');++const http = require('http');++const { gracefulStop } = require('openvibe-sdk/service');++++const source = fs.readFileSync(path.join(__dirname, '../server/index.js'), 'utf8');++const quiet = { log() {}, warn() {}, error() {} };++++// Slice a balanced (...) / [...] starting at `from` (source[from] is the opener), so the static checks++// read the whole gracefulStop call without depending on character offsets.++function balanced(source, from, open, close) {++ let depth = 0;++ for (let i = from; i < source.length; i++) {++ if (source[i] === open) depth++;++ else if (source[i] === close) { depth--; if (depth === 0) return source.slice(from, i + 1); }++ }++ throw new Error(`unbalanced ${open} at ${from}`);++}++function gracefulStopCall(source) {++ const at = source.indexOf('gracefulStop(');++ assert.ok(at > 0, 'server/index.js calls gracefulStop');++ return balanced(source, at + 'gracefulStop'.length, '(', ')');++}++function optionArray(call, name) {++ const m = call.match(new RegExp(`\\b${name}:\\s*\\[`));++ assert.ok(m, `the gracefulStop call has a ${name} array`);++ return balanced(call, m.index + m[0].length - 1, '[', ']');++}++++// A tiny local runner, as in OpenVibe.Sources/test/service-kit.test.js: `node test/service-kit.test.js`++// prints one line per case and exits 1 on the first failure.++function suite(name) {++ const tests = [];++ const t = (n, fn) => tests.push([n, fn]);++ t.run = async () => {++ let failed = 0;++ for (const [n, fn] of tests) {++ try { await fn(); console.log(` ok ${n}`); } catch (err) { failed++; console.log(` FAIL ${n}\n${err.stack}`); }++ }++ console.log(`${name}: ${tests.length - failed}/${tests.length} passed`);++ if (failed) process.exit(1);++ };++ return t;++}++const t = suite('service-kit');++++t('server/index.js imports gracefulStop from openvibe-sdk/service and owns no signal handler', () => {++ assert.match(source, /const \{ gracefulStop \} = require\('openvibe-sdk\/service'\)/);++ assert.doesNotMatch(source, /process\.on\('SIGTERM'/);++ assert.doesNotMatch(source, /process\.on\('SIGINT'/);++ assert.doesNotMatch(source, /function shutdown\(/, 'the hand-written shutdown() is gone');++ assert.doesNotMatch(source, /shuttingDown/, 'the guard that made a second signal a no-op is the kit\'s now');++});++++t("the one stop names Media, stops the recorder first, keeps every handle, and uses the 60 s drain / 70 s deadline / exit 0", () => {++ const call = gracefulStopCall(source);++ assert.match(call, /name: 'Media'/);++ assert.match(call, /drainMs: 60000/, "a 60 s drain (VOD_STOP_GRACE_MS) so in-flight uploads/large downloads are not cut at the kit's 4 s default");++ assert.match(call, /deadlineMs: 70000/, 'the 70 s stop the systemd unit leaves room for (TimeoutStopSec=80)');++ assert.match(call, /deadlineExitCode: 0/, 'Media exits 0 past the deadline, like the 5 s family');++ assert.match(call, /close: \[\(\) => db\.close\(\)\]/);++ assert.doesNotMatch(call, /beforeDrain/, 'the recorder is a stop step now, not beforeDrain (the kit runs beforeDrain after stop)');++++ // The recorder stops first, before every other handle — the order the hand-written shutdown ran.++ const stop = optionArray(call, 'stop');++ const recorderAt = stop.indexOf('recorder.stopAll()');++ assert.ok(recorderAt > 0, 'recorder.stopAll() is a stop step');++ for (const step of [++ 'vodStorage.stop()',++ 'healthJob.stop()',++ "require('./objects/verify-job').stop()",++ "require('./jobs/worker').stop()",++ "require('./objects/owner-subject-job').stop()",++ 'auth.stopJwksRefresh()',++ "require('./events')._reset()",++ "require('./actor-limits').valkey()",++ 'clearInterval(t)',++ ]) {++ const stepAt = stop.indexOf(step);++ assert.ok(stepAt > 0, `the stop still names ${step}`);++ assert.ok(recorderAt < stepAt, `recorder.stopAll() runs before ${step}`);++ }++ assert.doesNotMatch(source, /deadlineExitCode: 1/);++});++++t('the stop runs the recorder first, drains the server, then closes, and exits 0 exactly once', async () => {++ // Media passes no handles, so a step that throws is logged and the stop goes on (the try/catch the++ // old shutdown carried); the exit code stays the deadline's 0 either way.++ for (const fails of [false, true]) {++ const server = http.createServer((_req, res) => res.end('ok'));++ await new Promise((r) => server.listen(0, '127.0.0.1', r));++ const order = [];++ const warns = [];++ let exits = 0;++ let exited = null;++ const kit = gracefulStop({++ name: 'Media', server, signals: false,++ drainMs: 60000,++ stop: [++ () => { order.push(server.listening ? 'recorder while listening' : 'recorder too late'); if (fails) throw new Error('recorder'); },++ () => { order.push('worker'); },++ ],++ close: [() => { order.push(server.listening ? 'close while listening' : 'close after drain'); }],++ deadlineMs: 70000, deadlineExitCode: 0,++ exit: (c) => { exited = c; exits++; },++ log: { log() {}, warn: (m) => warns.push(m), error() {} },++ });++ const [code, again] = await Promise.all([kit.stop('SIGTERM'), kit.stop('SIGINT')]);++ assert.deepStrictEqual(order, ['recorder while listening', 'worker', 'close after drain'], 'the recorder stops first while the server still listens; close runs after the drain');++ assert.strictEqual(code, 0, 'Media exits 0 past the deadline and on a logged step failure');++ assert.strictEqual(again, code, 'a second signal changes nothing');++ assert.strictEqual(exited, 0);++ assert.strictEqual(exits, 1, 'exit is called once');++ assert.strictEqual(kit.stopping(), true);++ assert.strictEqual(server.listening, false);++ if (fails) assert.ok(warns.some((w) => /stop step failed/.test(w)), 'the failing step is logged');++ }++});++++t.run();
@@ -45,7 +45,7 @@ without `DATABASE_URL`, development uses an embedded PGlite database in `data/pg
@@ -45,7 +45,7 @@ without `DATABASE_URL`, development uses an embedded PGlite database in `data/pg
export and deletion), OpenVibe.Live (`live.lineage.resolve`), OpenVibe.Events (the outbox relay and
export and deletion), OpenVibe.Live (`live.lineage.resolve`), OpenVibe.Events (the outbox relay and
the account and revocation subscriptions)
the account and revocation subscriptions)
- Backblaze B2 and Cloudflare R2 (S3 API) when configured; `ffmpeg`/`ffprobe` on the host
- Backblaze B2 and Cloudflare R2 (S3 API) when configured; `ffmpeg`/`ffprobe` on the host
-- `openvibe-contracts` v0.79.0, `openvibe-sdk` v0.21.2 (tokens, events outbox, per-actor limits),
+- `openvibe-contracts` v0.79.0, `openvibe-sdk` v0.28.0 (tokens, events outbox, per-actor limits),
`openvibe-shared` v2.5.0, pinned by release tarball
`openvibe-shared` v2.5.0, pinned by release tarball
## Capabilities
## Capabilities
@@ -115,7 +115,7 @@ server/
@@ -115,7 +115,7 @@ server/
storage.orphans.scan (monthly storage orphan report, service-wide, report only),
storage.orphans.scan (monthly storage orphan report, service-wide, report only),
storage.move.cleanup (the alert for placement moves whose delete keeps failing, service-wide)
storage.move.cleanup (the alert for placement moves whose delete keeps failing, service-wide)
client-ip.js trust proxy = loopback; req.ip is the only client address
client-ip.js trust proxy = loopback; req.ip is the only client address
-(openvibe-shared v2.5.0, openvibe-contracts v0.79.0, openvibe-sdk v0.21.2: pinned release tarballs, installed by npm)
+(openvibe-shared v2.5.0, openvibe-contracts v0.79.0, openvibe-sdk v0.28.0: pinned release tarballs, installed by npm)
scripts/smoke-test.sh end-to-end smoke test (boots a temp instance)
scripts/smoke-test.sh end-to-end smoke test (boots a temp instance)
scripts/reconcile-objects.js / object-invariant.js / object-drift-report.js / no-good-copy-report.js object-model operator tools
scripts/reconcile-objects.js / object-invariant.js / object-drift-report.js / no-good-copy-report.js object-model operator tools
scripts/media-jobs.js list jobs, run the size-invariant scan (dry run by default), approve/cancel proposals
scripts/media-jobs.js list jobs, run the size-invariant scan (dry run by default), approve/cancel proposals
@@ -18,7 +18,7 @@
@@ -18,7 +18,7 @@
"iovalkey": "^0.4.0",
"iovalkey": "^0.4.0",
"multer": "^1.4.5-lts.1",
"multer": "^1.4.5-lts.1",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.0",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.0",
- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.21.2",
+ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.28.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.5.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.5.0",
"pg": "^8.23.0",
"pg": "^8.23.0",
"sharp": "^0.33.4"
"sharp": "^0.33.4"
@@ -2026,9 +2026,9 @@
@@ -2026,9 +2026,9 @@
}
}
},
},
"node_modules/openvibe-sdk": {
"node_modules/openvibe-sdk": {
- "version": "0.21.2",
- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.21.2",
- "integrity": "sha512-k04b9r2FCYVIhb8KLNrr5y43eb1l3UKzYNKl9m/vqMRAjZGh4p3pI2I/8UFU9bosKD8KAS121OO4rH8NUtx4sg==",
+ "version": "0.28.0",
+ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.28.0",
+ "integrity": "sha512-jjLisFDWBN+xDVsixa5BJRr3355kRijmvOh1/9rz4s1yVb3wZWCbpTLNcOvXm57ksnRoqK57393W38G0Ey3PFw==",
"license": "MIT",
"license": "MIT",
"engines": {
"engines": {
"node": ">=20"
"node": ">=20"
@@ -22,7 +22,7 @@
@@ -22,7 +22,7 @@
"iovalkey": "^0.4.0",
"iovalkey": "^0.4.0",
"multer": "^1.4.5-lts.1",
"multer": "^1.4.5-lts.1",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.0",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.0",
- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.21.2",
+ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.28.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.5.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.5.0",
"pg": "^8.23.0",
"pg": "^8.23.0",
"sharp": "^0.33.4"
"sharp": "^0.33.4"
@@ -27,6 +27,7 @@ if (drill.enabled) {
@@ -27,6 +27,7 @@ if (drill.enabled) {
const fs = require('fs');
const fs = require('fs');
const express = require('express');
const express = require('express');
+const { gracefulStop } = require('openvibe-sdk/service');
const config = require('./config');
const config = require('./config');
const db = require('./db/database');
const db = require('./db/database');
const auth = require('./auth');
const auth = require('./auth');
@@ -325,32 +326,35 @@ const ready = (async () => {
@@ -325,32 +326,35 @@ const ready = (async () => {
// A drill whose port is taken stops instead of running unready.
// A drill whose port is taken stops instead of running unready.
if (drill.enabled) server.once('error', (err) => { console.error(`[Drill] HTTP server: ${err.message}`); process.exit(1); });
if (drill.enabled) server.once('error', (err) => { console.error(`[Drill] HTTP server: ${err.message}`); process.exit(1); });
- let shuttingDown = false;
- function shutdown(signal) {
- if (shuttingDown) return;
- shuttingDown = true;
- console.log(`[Media] ${signal} received — shutting down`);
- try { recorder.stopAll(); } catch { /* */ }
- try { vodStorage.stop(); } catch { /* */ }
- try { healthJob.stop(); } catch { /* */ }
- try { require('./objects/verify-job').stop(); } catch { /* */ }
- try { require('./jobs/worker').stop(); } catch { /* */ }
- try { require('./objects/owner-subject-job').stop(); } catch { /* */ }
- try { auth.stopJwksRefresh(); } catch { /* */ }
- try { require('./events')._reset(); } catch { /* */ }
- try { const v = require('./actor-limits').valkey(); if (v) v.close().catch(() => {}); } catch { /* */ }
- for (const t of timers) clearInterval(t);
- server.close(() => {
- db.close().catch(() => {}).finally(() => process.exit(0));
- });
- // Recordings get STOP_GRACE_MS to flush trailers; don't hang forever.
- setTimeout(() => {
- console.warn('[Media] Forced exit after shutdown grace');
- db.close().catch(() => {}).finally(() => process.exit(0));
- }, 70_000).unref();
- }
- process.on('SIGINT', () => shutdown('SIGINT'));
- process.on('SIGTERM', () => shutdown('SIGTERM'));
+ // SIGTERM/SIGINT (openvibe-sdk/service; docs/service.md's handles family): stopping() turns true at
+ // once, the recorder stops first, then the background handles stop, the events outbox is reset and the
+ // database closes. The recorder is the first `stop` step so it runs before the worker and the pollers,
+ // as the hand-written shutdown ran it: stopAll() signals FFmpeg, which then gets VOD_STOP_GRACE_MS to
+ // flush trailers. The drain gets 60 s so in-flight uploads and large downloads are not cut at the kit's
+ // 4 s default; the whole stop is bounded at 70 s, the systemd unit's TimeoutStopSec (80 s) leaving room
+ // (deploy/systemd/openvibe-media.service). Media exits 0 past the deadline, like the 5 s family. A step
+ // that throws is logged and the stop goes on: the try/catch the hand-written shutdown carried lives in
+ // the kit now.
+ gracefulStop({
+ name: 'Media',
+ server,
+ drainMs: 60000,
+ stop: [
+ () => recorder.stopAll(),
+ () => vodStorage.stop(),
+ () => healthJob.stop(),
+ () => require('./objects/verify-job').stop(),
+ () => require('./jobs/worker').stop(),
+ () => require('./objects/owner-subject-job').stop(),
+ () => auth.stopJwksRefresh(),
+ () => require('./events')._reset(),
+ () => { const v = require('./actor-limits').valkey(); if (v) v.close().catch(() => {}); },
+ () => { for (const t of timers) clearInterval(t); },
+ ],
+ close: [() => db.close()],
+ deadlineMs: 70000,
+ deadlineExitCode: 0,
+ });
return { app, server };
return { app, server };
})();
})();
ready.catch((err) => {
ready.catch((err) => {
@@ -9,6 +9,16 @@ import { fileURLToPath } from 'node:url';
@@ -9,6 +9,16 @@ import { fileURLToPath } from 'node:url';
const require = createRequire(import.meta.url);
const require = createRequire(import.meta.url);
const root = path.join(path.dirname(fileURLToPath(import.meta.url)), '..', '..');
const root = path.join(path.dirname(fileURLToPath(import.meta.url)), '..', '..');
const { createTestDb } = require('openvibe-sdk/testing');
const { createTestDb } = require('openvibe-sdk/testing');
+// openvibe-sdk >= 0.26 createTestDb holds a lease pg.Client (application_name *-test-lease) whose socket is not unref'd,
+// so a test file that ends without process.exit() never exits. Unref it here; the lease still holds while the process lives.
+// Drop this once Media pins openvibe-sdk >= 0.31.1, which unrefs the lease itself.
+const pg = require(require.resolve('pg', { paths: [require.resolve('openvibe-sdk/testing')] }));
+const connect = pg.Client.prototype.connect;
+pg.Client.prototype.connect = function (...args) {
+ const r = connect.apply(this, args);
+ if (String(this.connectionParameters && this.connectionParameters.application_name).endsWith('-test-lease')) this.connection.stream.unref();
+ return r;
+};
const t = await createTestDb({ migrations: path.join(root, 'migrations'), store: process.env.MEDIA_TEST_STORE || 'pglite', service: 'media', max: 4 });
const t = await createTestDb({ migrations: path.join(root, 'migrations'), store: process.env.MEDIA_TEST_STORE || 'pglite', service: 'media', max: 4 });
// Tests insert rows with small explicit ids. A PostgreSQL identity would continue after them, so generated ids
// Tests insert rows with small explicit ids. A PostgreSQL identity would continue after them, so generated ids
// are set to start at 100000 here, clear of every id a test names.
// are set to start at 100000 here, clear of every id a test names.
new file mode 100644
new file mode 100644
@@ -0,0 +1,133 @@
@@ -0,0 +1,133 @@
+'use strict';
+/**
+ * openvibe-sdk/service in Media (plan B4): the SIGTERM/SIGINT shutdown is the kit's gracefulStop, not
+ * the hand-written shutdown()/process.on pair it replaces. It runs the stop steps (the recorder first —
+ * stopAll() signalling FFmpeg before the worker and pollers stop — the tiering sweep, health job, verify
+ * job, job worker, owner-subject job, JWKS refresher and events outbox stopped, the actor-limits Valkey
+ * closed, the timers cleared), drains the server for 60 s so in-flight uploads and large downloads are
+ * not cut at the kit's 4 s default, then closes the database; the whole stop is bounded at 70 s like the
+ * systemd unit, and Media exits 0 past the deadline. A failing step is logged and the stop goes on.
+ * The static half reads server/index.js; the behavioural half drives gracefulStop with the same option
+ * shape Media passes, with exports stubbed (/mirrors OpenVibe.Sources/test/service-kit.test.js).
+ */
+const assert = require('assert');
+const fs = require('fs');
+const path = require('path');
+const http = require('http');
+const { gracefulStop } = require('openvibe-sdk/service');
+
+const source = fs.readFileSync(path.join(__dirname, '../server/index.js'), 'utf8');
+const quiet = { log() {}, warn() {}, error() {} };
+
+// Slice a balanced (...) / [...] starting at `from` (source[from] is the opener), so the static checks
+// read the whole gracefulStop call without depending on character offsets.
+function balanced(source, from, open, close) {
+ let depth = 0;
+ for (let i = from; i < source.length; i++) {
+ if (source[i] === open) depth++;
+ else if (source[i] === close) { depth--; if (depth === 0) return source.slice(from, i + 1); }
+ }
+ throw new Error(`unbalanced ${open} at ${from}`);
+}
+function gracefulStopCall(source) {
+ const at = source.indexOf('gracefulStop(');
+ assert.ok(at > 0, 'server/index.js calls gracefulStop');
+ return balanced(source, at + 'gracefulStop'.length, '(', ')');
+}
+function optionArray(call, name) {
+ const m = call.match(new RegExp(`\\b${name}:\\s*\\[`));
+ assert.ok(m, `the gracefulStop call has a ${name} array`);
+ return balanced(call, m.index + m[0].length - 1, '[', ']');
+}
+
+// A tiny local runner, as in OpenVibe.Sources/test/service-kit.test.js: `node test/service-kit.test.js`
+// prints one line per case and exits 1 on the first failure.
+function suite(name) {
+ const tests = [];
+ const t = (n, fn) => tests.push([n, fn]);
+ t.run = async () => {
+ let failed = 0;
+ for (const [n, fn] of tests) {
+ try { await fn(); console.log(` ok ${n}`); } catch (err) { failed++; console.log(` FAIL ${n}\n${err.stack}`); }
+ }
+ console.log(`${name}: ${tests.length - failed}/${tests.length} passed`);
+ if (failed) process.exit(1);
+ };
+ return t;
+}
+const t = suite('service-kit');
+
+t('server/index.js imports gracefulStop from openvibe-sdk/service and owns no signal handler', () => {
+ assert.match(source, /const \{ gracefulStop \} = require\('openvibe-sdk\/service'\)/);
+ assert.doesNotMatch(source, /process\.on\('SIGTERM'/);
+ assert.doesNotMatch(source, /process\.on\('SIGINT'/);
+ assert.doesNotMatch(source, /function shutdown\(/, 'the hand-written shutdown() is gone');
+ assert.doesNotMatch(source, /shuttingDown/, 'the guard that made a second signal a no-op is the kit\'s now');
+});
+
+t("the one stop names Media, stops the recorder first, keeps every handle, and uses the 60 s drain / 70 s deadline / exit 0", () => {
+ const call = gracefulStopCall(source);
+ assert.match(call, /name: 'Media'/);
+ assert.match(call, /drainMs: 60000/, "a 60 s drain (VOD_STOP_GRACE_MS) so in-flight uploads/large downloads are not cut at the kit's 4 s default");
+ assert.match(call, /deadlineMs: 70000/, 'the 70 s stop the systemd unit leaves room for (TimeoutStopSec=80)');
+ assert.match(call, /deadlineExitCode: 0/, 'Media exits 0 past the deadline, like the 5 s family');
+ assert.match(call, /close: \[\(\) => db\.close\(\)\]/);
+ assert.doesNotMatch(call, /beforeDrain/, 'the recorder is a stop step now, not beforeDrain (the kit runs beforeDrain after stop)');
+
+ // The recorder stops first, before every other handle — the order the hand-written shutdown ran.
+ const stop = optionArray(call, 'stop');
+ const recorderAt = stop.indexOf('recorder.stopAll()');
+ assert.ok(recorderAt > 0, 'recorder.stopAll() is a stop step');
+ for (const step of [
+ 'vodStorage.stop()',
+ 'healthJob.stop()',
+ "require('./objects/verify-job').stop()",
+ "require('./jobs/worker').stop()",
+ "require('./objects/owner-subject-job').stop()",
+ 'auth.stopJwksRefresh()',
+ "require('./events')._reset()",
+ "require('./actor-limits').valkey()",
+ 'clearInterval(t)',
+ ]) {
+ const stepAt = stop.indexOf(step);
+ assert.ok(stepAt > 0, `the stop still names ${step}`);
+ assert.ok(recorderAt < stepAt, `recorder.stopAll() runs before ${step}`);
+ }
+ assert.doesNotMatch(source, /deadlineExitCode: 1/);
+});
+
+t('the stop runs the recorder first, drains the server, then closes, and exits 0 exactly once', async () => {
+ // Media passes no handles, so a step that throws is logged and the stop goes on (the try/catch the
+ // old shutdown carried); the exit code stays the deadline's 0 either way.
+ for (const fails of [false, true]) {
+ const server = http.createServer((_req, res) => res.end('ok'));
+ await new Promise((r) => server.listen(0, '127.0.0.1', r));
+ const order = [];
+ const warns = [];
+ let exits = 0;
+ let exited = null;
+ const kit = gracefulStop({
+ name: 'Media', server, signals: false,
+ drainMs: 60000,
+ stop: [
+ () => { order.push(server.listening ? 'recorder while listening' : 'recorder too late'); if (fails) throw new Error('recorder'); },
+ () => { order.push('worker'); },
+ ],
+ close: [() => { order.push(server.listening ? 'close while listening' : 'close after drain'); }],
+ deadlineMs: 70000, deadlineExitCode: 0,
+ exit: (c) => { exited = c; exits++; },
+ log: { log() {}, warn: (m) => warns.push(m), error() {} },
+ });
+ const [code, again] = await Promise.all([kit.stop('SIGTERM'), kit.stop('SIGINT')]);
+ assert.deepStrictEqual(order, ['recorder while listening', 'worker', 'close after drain'], 'the recorder stops first while the server still listens; close runs after the drain');
+ assert.strictEqual(code, 0, 'Media exits 0 past the deadline and on a logged step failure');
+ assert.strictEqual(again, code, 'a second signal changes nothing');
+ assert.strictEqual(exited, 0);
+ assert.strictEqual(exits, 1, 'exit is called once');
+ assert.strictEqual(kit.stopping(), true);
+ assert.strictEqual(server.listening, false);
+ if (fails) assert.ok(warns.some((w) => /stop step failed/.test(w)), 'the failing step is logged');
+ }
+});
+
+t.run();