openvibers/openrestream Commits

Move the API shutdown onto openvibe-sdk/service gracefulStop (#8)

OP OpenVibers committed 2 Oct 2026 · d272503839 · 6 files, +155 −14

checks run by GitHub CI (test, test-pg) · CI ✓ 3/3 · review SHIP by DeepSeek harness · 1 fix round

https://github.com/OpenVibers/OpenRe.Stream/pull/8

Changes

ddiff --git a/README.md b/README.md
iindex 9b6a106..975f654 100644
---- a/README.md
++++ b/README.md
@@@ -32,7 +32,7 @@ The ingest/restream control plane and runtime extracted from OpenVibe.Live. Live
- OpenVibe.Network (JWKS for service tokens and user JWTs; OAuth client `openre`; client-credentials tokens for Events/Media)
- OpenVibe.Events (event relay; optional: rows wait in the outbox)
- OpenVibe.Media (recording requests; optional: sessions work without it)
--- OpenVibe.Contracts v0.76.0 (ids, problem+json, capability checks), OpenVibe.SDK v0.21.3 (`db`, PostgreSQL outbox, token client, per-actor limits), OpenVibe.Shared v1.28.0 (Frame, readiness), pinned by release tarball
++- OpenVibe.Contracts v0.76.0 (ids, problem+json, capability checks), OpenVibe.SDK v0.26.0 (`db`, PostgreSQL outbox, token client, per-actor limits, the `service` kit), OpenVibe.Shared v2.3.1 (Frame, readiness), pinned by release tarball
- **PostgreSQL 18 and Valkey 9** (OpenVibe.Host `roles/data/`, ADR-035): the store every OpenRe process shares, and the API's per-actor limit counters (optional)
- OpenVibe.Live (`live.lineage.resolve`: which channel a Live-linked stream belongs to)
- node-media-server 2.7.4 (the RTMP session code Live runs) and the system ffmpeg
ddiff --git a/STATUS.json b/STATUS.json
iindex fa46b7f..f9d2c7b 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -8,7 +8,7 @@
"port": 4500,
"rtmpPort": 1936,
"domain": "openre.stream",
-- "contracts": "openvibe-contracts v0.76.0 (openvibe-sdk v0.21.3, openvibe-shared v2.3.1)",
++ "contracts": "openvibe-contracts v0.76.0 (openvibe-sdk v0.26.0, openvibe-shared v2.3.1)",
"protocols": {
"rtmp": "ingest ported (openre-rtmp-ingest)",
"restream": "RTMP source to RTMP/RTMPS/SRT destinations ported (openre-restream-worker)",
ddiff --git a/package-lock.json b/package-lock.json
iindex 0fb8e69..35e36dc 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -17,7 +17,7 @@
"mediasoup": "^3.14.14",
"node-media-server": "2.7.4",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.76.0",
-- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.21.3",
++ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.26.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.3.1",
"pg": "^8.23.0",
"sdp-transform": "^3.0.0",
@@@ -1834,9 +1834,9 @@
}
},
"node_modules/openvibe-sdk": {
-- "version": "0.21.3",
-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.21.3",
-- "integrity": "sha512-oAcxznSHKko+7zQCHS2VrSW2gqpXgbagnu+UZ60vrW/QjwtoCvatMZvtgEwpekYFGzpI7xklvCe0EOb1fa1H2g==",
++ "version": "0.26.0",
++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.26.0",
++ "integrity": "sha512-+RK89ZG4WAKIRh89yr3ieIvnyJVNSOI+dsbBqGNcr9UHsJStdxtM75zSDq8cUNpBPQ1X6ElatZrjU9S3s0aEkg==",
"license": "MIT",
"engines": {
"node": ">=20"
ddiff --git a/package.json b/package.json
iindex 004d7e9..0c53392 100644
---- a/package.json
++++ b/package.json
@@@ -33,7 +33,7 @@
"mediasoup": "^3.14.14",
"node-media-server": "2.7.4",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.76.0",
-- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.21.3",
++ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.26.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.3.1",
"pg": "^8.23.0",
"sdp-transform": "^3.0.0",
ddiff --git a/server/index.js b/server/index.js
iindex d592edd..483575b 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -9,6 +9,7 @@
* start() is what the tests use too: it takes a config plus injectable clock/fetch/log (and a database handle).
*/
const path = require('path');
++const { gracefulStop } = require('openvibe-sdk/service');
const { load } = require('./config');
const { openRuntime } = require('./store');
const { createKeyStore, createAuth } = require('./auth');
@@@ -47,13 +48,18 @@ async function start({ config, clock, fetchImpl = globalThis.fetch, log = consol
if (require.main === module) {
require('dotenv').config({ path: process.env.OPENRE_ENV_FILE || path.join(process.cwd(), '.env') });
start().then((h) => {
-- const shutdown = (sig) => {
-- console.log(`[openre-api] ${sig}: shutting down (workers are separate processes and keep running)`);
-- h.close().then(() => process.exit(0), () => process.exit(1));
-- setTimeout(() => process.exit(1), 10000).unref();
-- };
-- process.on('SIGTERM', () => shutdown('SIGTERM'));
-- process.on('SIGINT', () => shutdown('SIGINT'));
++ // SIGTERM/SIGINT (openvibe-sdk/service, docs/service.md's OpenRe.Stream entry): the keys poller stops, requests
++ // in flight get 8 s, then Valkey and the database close in that order. A database close failure exits 1;
++ // past the 10 s deadline the process also exits 1. Workers are separate processes and keep running.
++ gracefulStop({
++ name: 'openre-api',
++ server: h.server,
++ stop: [() => h.keys.stop()],
++ close: [() => h.app.locals.valkey && h.app.locals.valkey.close()],
++ handles: [() => h.rt.db.close()],
++ drainMs: 8000,
++ deadlineMs: 10000,
++ });
}).catch((err) => {
console.error(`[openre-api] failed to start: ${err.stack || err}`);
process.exit(1);
ddiff --git a/test/service-kit.test.js b/test/service-kit.test.js
nnew file mode 100644
iindex 0000000..136c1d1
---- /dev/null
++++ b/test/service-kit.test.js
@@@ -0,0 +1,135 @@
++'use strict';
++/**
++ * openvibe-sdk/service in OpenRe.Stream (plan T1, lane A): the API's SIGTERM/SIGINT shutdown is the kit's
++ * gracefulStop, not a hand-written handler. The keys poller stops, requests in flight drain for 8 s, then the
++ * store closes in today's order (Valkey, then the database); a database close failure exits 1,
++ * a clean stop exits 0, and past the 10 s deadline the process exits 1. The static half reads server/index.js;
++ * the behavioural half drives gracefulStop with the same options OpenRe passes, exits stubbed.
++ */
++const assert = require('assert');
++const fs = require('fs');
++const path = require('path');
++const http = require('http');
++const { gracefulStop } = require('openvibe-sdk/service');
++const { suite } = require('./helpers');
++
++const t = suite('service-kit');
++const quiet = { log() {}, warn() {}, error() {} };
++const source = fs.readFileSync(path.join(__dirname, '../server/index.js'), 'utf8');
++
++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, /setTimeout\(\(\) => process\.exit\(1\), 10000\)/, 'the old 10 s exit-1 timer is gone');
++});
++
++t('the one stop names openre-api, passes the server and the store steps, and uses drainMs 8000 / deadlineMs 10000', () => {
++ const at = source.indexOf('gracefulStop({');
++ const call = source.slice(at, at + 500);
++ assert.match(call, /name: 'openre-api'/);
++ assert.match(call, /server: h\.server/);
++ assert.match(call, /stop: \[\(\) => h\.keys\.stop\(\)\]/, 'the keys poller stops first');
++ assert.match(call, /close: \[\(\) => h\.app\.locals\.valkey && h\.app\.locals\.valkey\.close\(\)\]/);
++ assert.match(call, /handles: \[\(\) => h\.rt\.db\.close\(\)\]/, 'the database closes after Valkey and its failure exits 1');
++ assert.match(call, /drainMs: 8000/);
++ assert.match(call, /deadlineMs: 10000/);
++ assert.doesNotMatch(call, /deadlineExitCode/, 'the default exit 1 past the deadline is kept');
++});
++
++t('the stop steps run once, in order, after the drain; a clean stop exits 0 and a second signal changes nothing', async () => {
++ const server = http.createServer((_req, res) => res.end('ok'));
++ await new Promise((r) => server.listen(0, '127.0.0.1', r));
++ const order = [];
++ const keys = { stop: () => order.push('keys.stop') };
++ const valkey = { close: async () => order.push(server.listening ? 'valkey.close while listening' : 'valkey.close') };
++ const db = { close: async () => order.push('db.close') };
++ let exited = null;
++ let exits = 0;
++ const kit = gracefulStop({
++ name: 'openre-api', server,
++ stop: [() => keys.stop()],
++ close: [() => valkey.close()], handles: [() => db.close()],
++ drainMs: 8000, deadlineMs: 10000, signals: false,
++ exit: (c) => { exited = c; exits++; }, log: quiet,
++ });
++ const [code, again] = await Promise.all([kit.stop('SIGTERM'), kit.stop('SIGINT')]);
++ assert.deepStrictEqual(order, ['keys.stop', 'valkey.close', 'db.close'], 'the stop step first, then the closes once the server has stopped listening');
++ assert.strictEqual(code, 0);
++ assert.strictEqual(again, code, 'a second signal returns the first stop');
++ assert.strictEqual(exited, 0);
++ assert.strictEqual(exits, 1, 'exit is called once');
++ assert.strictEqual(kit.stopping(), true);
++ assert.strictEqual(server.listening, false);
++});
++
++t('a request in flight finishes during the drain, before the store closes', async () => {
++ let received;
++ const inFlight = new Promise((r) => { received = r; });
++ const server = http.createServer((_req, res) => { received(); setTimeout(() => res.end('done'), 80); });
++ await new Promise((r) => server.listen(0, '127.0.0.1', r));
++ const order = [];
++ const kit = gracefulStop({
++ name: 'openre-api', server,
++ stop: [],
++ close: [() => order.push('valkey.close'), () => order.push('db.close')],
++ drainMs: 8000, deadlineMs: 10000, signals: false, exit: () => {}, log: quiet,
++ });
++ const body = new Promise((resolve, reject) => {
++ http.get({ host: '127.0.0.1', port: server.address().port, path: '/' }, (res) => {
++ let s = '';
++ res.on('data', (c) => { s += c; });
++ res.on('end', () => resolve(s));
++ }).on('error', reject);
++ });
++ await inFlight; // the request is in flight when the signal arrives
++ const stopping = kit.stop('SIGTERM');
++ assert.strictEqual(await body, 'done', 'the drain let it finish');
++ assert.strictEqual(await stopping, 0);
++ assert.deepStrictEqual(order, ['valkey.close', 'db.close'], 'the store closed after the drain');
++});
++
++t('a close step that throws is logged, the stop goes on and exits 0', async () => {
++ const order = [];
++ const kit = gracefulStop({
++ name: 'openre-api', server: null,
++ stop: [],
++ close: [() => { order.push('valkey.close'); throw new Error('valkey down'); }, () => order.push('db.close')],
++ drainMs: 8000, deadlineMs: 10000, signals: false, exit: () => {}, log: quiet,
++ });
++ assert.strictEqual(await kit.stop('SIGTERM'), 0);
++ assert.deepStrictEqual(order, ['valkey.close', 'db.close']);
++});
++
++t('a database close failure exits 1 after Valkey closes', async () => {
++ const order = [];
++ const errors = [];
++ let exited = null;
++ const kit = gracefulStop({
++ name: 'openre-api', server: null,
++ stop: [],
++ close: [() => order.push('valkey.close')],
++ handles: [() => { order.push('db.close'); throw new Error('database down'); }],
++ drainMs: 8000, deadlineMs: 10000, signals: false,
++ exit: (c) => { exited = c; }, log: { ...quiet, error: (msg) => errors.push(msg) },
++ });
++ assert.strictEqual(await kit.stop('SIGTERM'), 1);
++ assert.strictEqual(exited, 1);
++ assert.deepStrictEqual(order, ['valkey.close', 'db.close']);
++ assert.match(errors.join('\n'), /database down/);
++});
++
++t('past the deadline the process exits 1, even with a close step stuck', async () => {
++ let exited = null;
++ const kit = gracefulStop({
++ name: 'openre-api', server: null,
++ stop: [],
++ close: [() => new Promise(() => {})],
++ drainMs: 8000, deadlineMs: 40, signals: false,
++ exit: (c) => { exited = c; }, log: quiet,
++ });
++ assert.strictEqual(await kit.stop('SIGTERM'), 1);
++ assert.strictEqual(exited, 1);
++});
++
++t.run();