openvibers/openvibe.chat Commits
Move Chat's shutdown onto openvibe-sdk/service gracefulStop (#5)
Changes
@@@ -17,7 +17,7 @@"iovalkey": "^0.4.0","multer": "^2.0.2","openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.80.0",-- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.2",++ "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.4.0","pg": "^8.23.0","ws": "^8.18.0"@@@ -1310,8 +1310,9 @@}},"node_modules/openvibe-sdk": {-- "version": "0.25.2",-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.2",++ "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"@@@ -30,7 +30,7 @@"iovalkey": "^0.4.0","multer": "^2.0.2","openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.80.0",-- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.2",++ "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.4.0","pg": "^8.23.0","ws": "^8.18.0"@@@ -25,6 +25,7 @@ const subscriptions = require('./events/subscriptions');const chatAi = require('./ai/chat-ai');const { limits } = require('./net/actor-limits');const { createValkey } = require('openvibe-sdk/valkey');++const { gracefulStop, within } = require('openvibe-sdk/service');const { createApp } = require('./app');let rejectionsLogged = false;@@@ -73,33 +74,37 @@ async function start() {console.log(`[Chat] listening on http://${config.host}:${server.address().port} (ws /ws/chat${config.calls.enabled ? ', /ws/call' : ''})`);const subs = subscriptions.startAtBoot({ config, port: server.address().port });-- let stopping = false;-- const shutdown = async () => {-- if (stopping) return;-- stopping = true;-- console.log('[Chat] shutting down');-- try {-- await chatServer.broadcastAll({++ // systemd sends SIGTERM (SIGINT by hand); openvibe-sdk/service's gracefulStop takes the signal, runs the stop++ // steps in order (nothing new starts), drains the HTTP server, runs the close steps, then exits 0. drainMs++ // bounds the drain; deadlineMs 8000 is the manifest's lifecycle.shutdown.deadlineSeconds (openvibe-contracts++ // manifests/services/chat.json) and deadlineExitCode 0 keeps the hand-written hard timer's exit 0. The mirror++ // flush stays bounded at 3 s (within). The stop steps keep the old shutdown's exact order; a step that throws++ // is logged and the stop goes on.++ const { stop: shutdown } = gracefulStop({++ name: 'Chat',++ server,++ drainMs: 4000,++ deadlineMs: 8000,++ deadlineExitCode: 0,++ stop: [++ () => chatServer.broadcastAll({type: 'server_restart',message: '⚙️ Chat server restarting — you will be reconnected automatically.',timestamp: new Date().toISOString(),-- });-- } catch { /* non-critical */ }-- ctx.stop();-- relay.stop();-- mirror.stop();-- events.stop();-- chatAi.stop();-- subs.stop();-- try { if (valkey) await valkey.close(); } catch { /* */ }-- try { await Promise.race([mirror.flush(), new Promise((r) => setTimeout(r, 3000))]); } catch { /* */ }-- try { chatServer.close(); } catch { /* */ }-- try { await callServer.close(); } catch { /* */ }-- server.close(async () => { await db.close(); process.exit(0); });-- setTimeout(() => process.exit(0), 5000).unref();-- };-- process.on('SIGTERM', shutdown);-- process.on('SIGINT', shutdown);++ }),++ () => ctx.stop(),++ () => relay.stop(),++ () => mirror.stop(),++ () => events.stop(),++ () => chatAi.stop(),++ () => subs.stop(),++ () => { if (valkey) return valkey.close(); },++ () => within(3000, mirror.flush()),++ () => chatServer.close(),++ () => callServer.close(),++ ],++ close: [() => db.close()],++ });return { server, mirror, relay, bridge, events, callServer, subscriptions: subs, shutdown };}@@@ -342,6 +342,7 @@ async function boot({ env = {} } = {}) {h.eventsRelay = started.relay;h.eventsConsumer = started.events;h.subscriptions = started.subscriptions;++ h.shutdown = started.shutdown;h.port = started.server.address().port;h.base = `http://127.0.0.1:${h.port}`;h.db = require('../server/db/database');nnew file mode 100644@@@ -0,0 +1,105 @@++'use strict';++/**++ * Chat's shutdown moved onto openvibe-sdk/service (plan T1 lane A): start() hands the signal to++ * gracefulStop, whose stop steps keep the old hand-written order and whose close step closes the++ * database. The manifest's lifecycle.shutdown.deadlineSeconds is 8 (openvibe-contracts++ * manifests/services/chat.json), so deadlineMs is 8000, and deadlineExitCode 0 keeps the old++ * hard timer's exit code.++ *++ * The first three checks read server/index.js itself (the entry point cannot be inspected through++ * its exports alone); the rest boot Chat with the harness and drive the returned shutdown with++ * process.exit stubbed, so the stop steps can be observed in order and the database close seen.++ */++const assert = require('assert');++const fs = require('fs');++const path = require('path');++const { boot, suite } = require('./helpers');++++const t = suite('service-kit');++++const src = fs.readFileSync(path.join(__dirname, '..', 'server', 'index.js'), 'utf8');++++t('server/index.js imports gracefulStop and within from openvibe-sdk/service', () => {++ const m = src.match(/const\s*\{([^}]*)\}\s*=\s*require\(['"]openvibe-sdk\/service['"]\)/);++ assert.ok(m, "require('openvibe-sdk/service')'s exports are not destructured");++ assert.ok(/\bgracefulStop\b/.test(m[1]), 'gracefulStop is not imported');++ assert.ok(/\bwithin\b/.test(m[1]), 'within is not imported');++});++++t('server/index.js leaves the signal handlers to the kit', () => {++ assert.ok(!/process\.on\(\s*['"]SIGTERM['"]/.test(src), 'server/index.js still installs its own SIGTERM handler');++ assert.ok(!/process\.on\(\s*['"]SIGINT['"]/.test(src), 'server/index.js still installs its own SIGINT handler');++ assert.ok(/gracefulStop\(\{/.test(src), 'gracefulStop is not called in start()');++ assert.ok(/name:\s*['"]Chat['"]/.test(src), "gracefulStop is not named 'Chat'");++});++++t("the deadline is the manifest's 8 s and the deadline exit code is 0", () => {++ assert.ok(/deadlineMs:\s*8000/.test(src), 'deadlineMs is not 8000');++ assert.ok(/deadlineExitCode:\s*0/.test(src), 'deadlineExitCode is not 0');++ assert.ok(/drainMs:\s*4000/.test(src), 'drainMs is not 4000');++});++++let h;++++t('boot', async () => {++ h = await boot();++ assert.strictEqual(typeof h.shutdown, 'function', 'start() returned a shutdown handle');++});++++t('the returned shutdown runs the stop steps in order and closes the database', async () => {++ const order = [];++ const wrap = (obj, name, tag) => {++ const orig = obj[name].bind(obj);++ obj[name] = (...args) => { order.push(tag); return orig(...args); };++ };++ const chatServer = h.chatServer;++ const callServer = require('../server/calls/call-server');++ const chatAi = require('../server/ai/chat-ai');++ const db = require('../server/db/database');++ wrap(chatServer, 'broadcastAll', 'broadcastAll');++ wrap(h.ctx, 'stop', 'ctx.stop');++ wrap(h.eventsRelay, 'stop', 'relay.stop');++ wrap(h.mirrorRelay, 'stop', 'mirror.stop');++ wrap(h.eventsConsumer, 'stop', 'events.stop');++ wrap(chatAi, 'stop', 'chatAi.stop');++ wrap(h.subscriptions, 'stop', 'subs.stop');++ wrap(h.mirrorRelay, 'flush', 'mirror.flush');++ wrap(chatServer, 'close', 'chatServer.close');++ wrap(callServer, 'close', 'callServer.close');++ wrap(db, 'close', 'db.close');++++ const realExit = process.exit;++ let exitCode = null;++ process.exit = (code) => { exitCode = code; };++ let code;++ try {++ code = await h.shutdown();++ assert.strictEqual(exitCode, 0, 'exit() is called with 0');++ } finally {++ process.exit = realExit;++ }++ assert.strictEqual(code, 0, 'the shutdown resolves with exit code 0');++++ assert.deepStrictEqual(order, [++ 'broadcastAll',++ 'ctx.stop',++ 'relay.stop',++ 'mirror.stop',++ 'events.stop',++ 'chatAi.stop',++ 'subs.stop',++ 'mirror.flush',++ 'chatServer.close',++ 'callServer.close',++ 'db.close',++ ], `the stop steps ran in the old order (saw ${order.join(', ')})`);++++ // The kit starts the stop once: a second call is the same promise, no step runs again.++ const again = await h.shutdown();++ assert.strictEqual(again, 0, 'a second stop resolves with the same code');++ assert.strictEqual(order.length, 11, 'a second stop runs no step again');++});++++// The shutdown step above already stopped the process; detach is a no-op then (and a fallback if a++// check failed before it).++t.run(async () => { try { if (h) await h.detach(); } catch { /* already stopped */ } });
@@ -17,7 +17,7 @@
@@ -17,7 +17,7 @@
"iovalkey": "^0.4.0",
"iovalkey": "^0.4.0",
"multer": "^2.0.2",
"multer": "^2.0.2",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.80.0",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.80.0",
- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.2",
+ "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.4.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.4.0",
"pg": "^8.23.0",
"pg": "^8.23.0",
"ws": "^8.18.0"
"ws": "^8.18.0"
@@ -1310,8 +1310,9 @@
@@ -1310,8 +1310,9 @@
}
}
},
},
"node_modules/openvibe-sdk": {
"node_modules/openvibe-sdk": {
- "version": "0.25.2",
- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.2",
+ "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",
"license": "MIT",
"engines": {
"engines": {
"node": ">=20"
"node": ">=20"
@@ -30,7 +30,7 @@
@@ -30,7 +30,7 @@
"iovalkey": "^0.4.0",
"iovalkey": "^0.4.0",
"multer": "^2.0.2",
"multer": "^2.0.2",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.80.0",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.80.0",
- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.2",
+ "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.4.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.4.0",
"pg": "^8.23.0",
"pg": "^8.23.0",
"ws": "^8.18.0"
"ws": "^8.18.0"
@@ -25,6 +25,7 @@ const subscriptions = require('./events/subscriptions');
@@ -25,6 +25,7 @@ const subscriptions = require('./events/subscriptions');
const chatAi = require('./ai/chat-ai');
const chatAi = require('./ai/chat-ai');
const { limits } = require('./net/actor-limits');
const { limits } = require('./net/actor-limits');
const { createValkey } = require('openvibe-sdk/valkey');
const { createValkey } = require('openvibe-sdk/valkey');
+const { gracefulStop, within } = require('openvibe-sdk/service');
const { createApp } = require('./app');
const { createApp } = require('./app');
let rejectionsLogged = false;
let rejectionsLogged = false;
@@ -73,33 +74,37 @@ async function start() {
@@ -73,33 +74,37 @@ async function start() {
console.log(`[Chat] listening on http://${config.host}:${server.address().port} (ws /ws/chat${config.calls.enabled ? ', /ws/call' : ''})`);
console.log(`[Chat] listening on http://${config.host}:${server.address().port} (ws /ws/chat${config.calls.enabled ? ', /ws/call' : ''})`);
const subs = subscriptions.startAtBoot({ config, port: server.address().port });
const subs = subscriptions.startAtBoot({ config, port: server.address().port });
- let stopping = false;
- const shutdown = async () => {
- if (stopping) return;
- stopping = true;
- console.log('[Chat] shutting down');
- try {
- await chatServer.broadcastAll({
+ // systemd sends SIGTERM (SIGINT by hand); openvibe-sdk/service's gracefulStop takes the signal, runs the stop
+ // steps in order (nothing new starts), drains the HTTP server, runs the close steps, then exits 0. drainMs
+ // bounds the drain; deadlineMs 8000 is the manifest's lifecycle.shutdown.deadlineSeconds (openvibe-contracts
+ // manifests/services/chat.json) and deadlineExitCode 0 keeps the hand-written hard timer's exit 0. The mirror
+ // flush stays bounded at 3 s (within). The stop steps keep the old shutdown's exact order; a step that throws
+ // is logged and the stop goes on.
+ const { stop: shutdown } = gracefulStop({
+ name: 'Chat',
+ server,
+ drainMs: 4000,
+ deadlineMs: 8000,
+ deadlineExitCode: 0,
+ stop: [
+ () => chatServer.broadcastAll({
type: 'server_restart',
type: 'server_restart',
message: '⚙️ Chat server restarting — you will be reconnected automatically.',
message: '⚙️ Chat server restarting — you will be reconnected automatically.',
timestamp: new Date().toISOString(),
timestamp: new Date().toISOString(),
- });
- } catch { /* non-critical */ }
- ctx.stop();
- relay.stop();
- mirror.stop();
- events.stop();
- chatAi.stop();
- subs.stop();
- try { if (valkey) await valkey.close(); } catch { /* */ }
- try { await Promise.race([mirror.flush(), new Promise((r) => setTimeout(r, 3000))]); } catch { /* */ }
- try { chatServer.close(); } catch { /* */ }
- try { await callServer.close(); } catch { /* */ }
- server.close(async () => { await db.close(); process.exit(0); });
- setTimeout(() => process.exit(0), 5000).unref();
- };
- process.on('SIGTERM', shutdown);
- process.on('SIGINT', shutdown);
+ }),
+ () => ctx.stop(),
+ () => relay.stop(),
+ () => mirror.stop(),
+ () => events.stop(),
+ () => chatAi.stop(),
+ () => subs.stop(),
+ () => { if (valkey) return valkey.close(); },
+ () => within(3000, mirror.flush()),
+ () => chatServer.close(),
+ () => callServer.close(),
+ ],
+ close: [() => db.close()],
+ });
return { server, mirror, relay, bridge, events, callServer, subscriptions: subs, shutdown };
return { server, mirror, relay, bridge, events, callServer, subscriptions: subs, shutdown };
}
}
@@ -342,6 +342,7 @@ async function boot({ env = {} } = {}) {
@@ -342,6 +342,7 @@ async function boot({ env = {} } = {}) {
h.eventsRelay = started.relay;
h.eventsRelay = started.relay;
h.eventsConsumer = started.events;
h.eventsConsumer = started.events;
h.subscriptions = started.subscriptions;
h.subscriptions = started.subscriptions;
+ h.shutdown = started.shutdown;
h.port = started.server.address().port;
h.port = started.server.address().port;
h.base = `http://127.0.0.1:${h.port}`;
h.base = `http://127.0.0.1:${h.port}`;
h.db = require('../server/db/database');
h.db = require('../server/db/database');
new file mode 100644
new file mode 100644
@@ -0,0 +1,105 @@
@@ -0,0 +1,105 @@
+'use strict';
+/**
+ * Chat's shutdown moved onto openvibe-sdk/service (plan T1 lane A): start() hands the signal to
+ * gracefulStop, whose stop steps keep the old hand-written order and whose close step closes the
+ * database. The manifest's lifecycle.shutdown.deadlineSeconds is 8 (openvibe-contracts
+ * manifests/services/chat.json), so deadlineMs is 8000, and deadlineExitCode 0 keeps the old
+ * hard timer's exit code.
+ *
+ * The first three checks read server/index.js itself (the entry point cannot be inspected through
+ * its exports alone); the rest boot Chat with the harness and drive the returned shutdown with
+ * process.exit stubbed, so the stop steps can be observed in order and the database close seen.
+ */
+const assert = require('assert');
+const fs = require('fs');
+const path = require('path');
+const { boot, suite } = require('./helpers');
+
+const t = suite('service-kit');
+
+const src = fs.readFileSync(path.join(__dirname, '..', 'server', 'index.js'), 'utf8');
+
+t('server/index.js imports gracefulStop and within from openvibe-sdk/service', () => {
+ const m = src.match(/const\s*\{([^}]*)\}\s*=\s*require\(['"]openvibe-sdk\/service['"]\)/);
+ assert.ok(m, "require('openvibe-sdk/service')'s exports are not destructured");
+ assert.ok(/\bgracefulStop\b/.test(m[1]), 'gracefulStop is not imported');
+ assert.ok(/\bwithin\b/.test(m[1]), 'within is not imported');
+});
+
+t('server/index.js leaves the signal handlers to the kit', () => {
+ assert.ok(!/process\.on\(\s*['"]SIGTERM['"]/.test(src), 'server/index.js still installs its own SIGTERM handler');
+ assert.ok(!/process\.on\(\s*['"]SIGINT['"]/.test(src), 'server/index.js still installs its own SIGINT handler');
+ assert.ok(/gracefulStop\(\{/.test(src), 'gracefulStop is not called in start()');
+ assert.ok(/name:\s*['"]Chat['"]/.test(src), "gracefulStop is not named 'Chat'");
+});
+
+t("the deadline is the manifest's 8 s and the deadline exit code is 0", () => {
+ assert.ok(/deadlineMs:\s*8000/.test(src), 'deadlineMs is not 8000');
+ assert.ok(/deadlineExitCode:\s*0/.test(src), 'deadlineExitCode is not 0');
+ assert.ok(/drainMs:\s*4000/.test(src), 'drainMs is not 4000');
+});
+
+let h;
+
+t('boot', async () => {
+ h = await boot();
+ assert.strictEqual(typeof h.shutdown, 'function', 'start() returned a shutdown handle');
+});
+
+t('the returned shutdown runs the stop steps in order and closes the database', async () => {
+ const order = [];
+ const wrap = (obj, name, tag) => {
+ const orig = obj[name].bind(obj);
+ obj[name] = (...args) => { order.push(tag); return orig(...args); };
+ };
+ const chatServer = h.chatServer;
+ const callServer = require('../server/calls/call-server');
+ const chatAi = require('../server/ai/chat-ai');
+ const db = require('../server/db/database');
+ wrap(chatServer, 'broadcastAll', 'broadcastAll');
+ wrap(h.ctx, 'stop', 'ctx.stop');
+ wrap(h.eventsRelay, 'stop', 'relay.stop');
+ wrap(h.mirrorRelay, 'stop', 'mirror.stop');
+ wrap(h.eventsConsumer, 'stop', 'events.stop');
+ wrap(chatAi, 'stop', 'chatAi.stop');
+ wrap(h.subscriptions, 'stop', 'subs.stop');
+ wrap(h.mirrorRelay, 'flush', 'mirror.flush');
+ wrap(chatServer, 'close', 'chatServer.close');
+ wrap(callServer, 'close', 'callServer.close');
+ wrap(db, 'close', 'db.close');
+
+ const realExit = process.exit;
+ let exitCode = null;
+ process.exit = (code) => { exitCode = code; };
+ let code;
+ try {
+ code = await h.shutdown();
+ assert.strictEqual(exitCode, 0, 'exit() is called with 0');
+ } finally {
+ process.exit = realExit;
+ }
+ assert.strictEqual(code, 0, 'the shutdown resolves with exit code 0');
+
+ assert.deepStrictEqual(order, [
+ 'broadcastAll',
+ 'ctx.stop',
+ 'relay.stop',
+ 'mirror.stop',
+ 'events.stop',
+ 'chatAi.stop',
+ 'subs.stop',
+ 'mirror.flush',
+ 'chatServer.close',
+ 'callServer.close',
+ 'db.close',
+ ], `the stop steps ran in the old order (saw ${order.join(', ')})`);
+
+ // The kit starts the stop once: a second call is the same promise, no step runs again.
+ const again = await h.shutdown();
+ assert.strictEqual(again, 0, 'a second stop resolves with the same code');
+ assert.strictEqual(order.length, 11, 'a second stop runs no step again');
+});
+
+// The shutdown step above already stopped the process; detach is a no-op then (and a fallback if a
+// check failed before it).
+t.run(async () => { try { if (h) await h.detach(); } catch { /* already stopped */ } });