openvibers/openvibe.community Commits
Move Community's graceful shutdown onto openvibe-sdk/service (#7)
Changes
@@@ -18,7 +18,7 @@"iovalkey": "^0.4.0","multer": "^2.4.0","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.0",++ "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"},@@@ -1207,9 +1207,9 @@}},"node_modules/openvibe-sdk": {-- "version": "0.21.0",-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.21.0",-- "integrity": "sha512-4gB+Lsj6Df1ztxI2ZNvPGUzXpSNV3FsEOP2VkcQ1hQIKdUdwH1BFEC16aQIwOrugmhe1U/CSmCWpPVb9bFu1ag==",++ "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"@@@ -29,7 +29,7 @@"iovalkey": "^0.4.0","multer": "^2.4.0","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.0",++ "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"},ddeleted file mode 100644@@@ -1,117 +0,0 @@--'use strict';--// ═══════════════════════════════════════════════════════════════--// Graceful stop for OpenVibe.Community (roadmap WS-P lifecycle; openvibe-contracts--// manifests/services/community.json → lifecycle.shutdown). systemd sends SIGTERM on a restart or a deploy;--// SIGINT does the same by hand. On the first signal:--//--// 1. `stop` steps run: timers and scans stop taking new work (nothing new starts);--// 2. the HTTP server stops taking connections, idle keep-alive connections are closed, requests in--// flight are answered with `Connection: close`, and open event streams (text/event-stream) are--// closed — EventSource reconnects on its own;--// 3. the requests in flight finish, for at most `drainMs`; whatever is still open then is cut;--// 4. `close` steps run: the relays settle, the database closes;--// 5. the process exits 0.--//--// A hard `deadlineMs` bounds the whole stop (the manifest's deadlineSeconds): past it the process--// exits 1. A second signal while stopping changes nothing.--// ═══════════════════════════════════════════════════════════════----const DRAIN_MS = 4000;--const DEADLINE_MS = 5000;----/**-- * @param {object} o-- * @param {string} o.name log prefix, e.g. 'Community'-- * @param {import('http').Server} o.server-- * @param {Array<Function>} [o.stop] run first, in order (sync or async; a failure is logged)-- * @param {Array<Function>} [o.close] run after the HTTP drain, in order-- * @param {number} [o.drainMs] how long requests in flight may take (default 4000)-- * @param {number} [o.deadlineMs] the whole stop, else exit 1 (default 5000)-- * @param {boolean} [o.signals] false: do not install SIGTERM/SIGINT handlers (tests)-- * @param {(code:number) => void} [o.exit] default process.exit-- * @returns {{ stop(signal?: string): Promise<number>, stopping(): boolean }}-- */--function gracefulStop(o) {-- const { name, server } = o;-- const drainMs = o.drainMs == null ? DRAIN_MS : o.drainMs;-- const deadlineMs = o.deadlineMs == null ? DEADLINE_MS : o.deadlineMs;-- const log = o.log || console;-- const exit = o.exit || ((code) => process.exit(code));-- let started = null;---- // Every response in flight, so the stop can mark them Connection: close and close event streams.-- const inflight = new Set();-- server.prependListener('request', (req, res) => {-- if (started && !res.headersSent) res.setHeader('Connection', 'close');-- inflight.add(res);-- res.on('close', () => inflight.delete(res));-- });---- async function step(fn, phase) {-- try { await fn(); } catch (err) { log.warn(`[${name}] stop: ${phase} step failed: ${err && err.message}`); }-- }---- function isEventStream(res) {-- const ct = String(res.getHeader('content-type') || '');-- return /text\/event-stream/i.test(ct) || (typeof res._header === 'string' && /content-type:\s*text\/event-stream/i.test(res._header));-- }---- function drain() {-- return new Promise((resolve) => {-- let done = false;-- const finish = (cut) => {-- if (done) return;-- done = true;-- clearInterval(sweep);-- clearTimeout(timer);-- resolve(cut);-- };-- server.close(() => finish(0));-- for (const res of inflight) {-- if (isEventStream(res)) res.destroy();-- else if (!res.headersSent) res.setHeader('Connection', 'close');-- }-- // A connection whose response just finished goes idle: close it now rather than after keepAliveTimeout.-- const idle = () => { if (typeof server.closeIdleConnections === 'function') server.closeIdleConnections(); };-- idle();-- const sweep = setInterval(idle, 50);-- const timer = setTimeout(() => {-- const cut = inflight.size;-- if (typeof server.closeAllConnections === 'function') server.closeAllConnections();-- finish(cut || 1);-- }, drainMs);-- });-- }---- function stop(signal = 'stop') {-- if (started) return started;-- started = (async () => {-- const t0 = Date.now();-- log.log(`[${name}] ${signal}: stopping (requests in flight get ${drainMs} ms)`);-- const hard = setTimeout(() => { log.error(`[${name}] stop took longer than ${deadlineMs} ms: exiting 1`); exit(1); }, deadlineMs);-- if (hard.unref) hard.unref();-- for (const fn of o.stop || []) await step(fn, 'stop');-- const cut = await drain();-- if (cut) log.warn(`[${name}] ${cut} request(s) still open after ${drainMs} ms were cut`);-- for (const fn of o.close || []) await step(fn, 'close');-- clearTimeout(hard);-- log.log(`[${name}] stopped in ${Date.now() - t0} ms`);-- exit(0);-- return 0;-- })();-- return started;-- }---- if (o.signals !== false) {-- process.on('SIGTERM', () => { stop('SIGTERM'); });-- process.on('SIGINT', () => { stop('SIGINT'); });-- }-- return { stop, stopping: () => !!started };--}----/** Await `promise`, but no longer than `ms` (a best-effort step inside the deadline). */--function within(ms, promise) {-- return Promise.race([Promise.resolve(promise).catch(() => {}), new Promise((r) => { const t = setTimeout(r, ms); if (t.unref) t.unref(); })]);--}----module.exports = { gracefulStop, within, DRAIN_MS, DEADLINE_MS };@@@ -41,17 +41,18 @@ const { createApp } = require('./app');let searchDocs = null;try { searchDocs = require('./search/documents').createSearchDocuments({ db: require('./db').getDb() }); searchDocs.start(); } catch (err) { console.warn('[Search] documents not started:', err.message); }-- // ── Stop (roadmap WS-P lifecycle; server/graceful.js) ────────++ // ── Stop (roadmap WS-P lifecycle; openvibe-sdk/service) ──────// SIGTERM: the Pulse subscription retries, the community.profile scan and the search-document scans// stop (nothing new starts); the server stops taking connections, closes idle keep-alive ones (it keeps// them 65 s otherwise) and lets requests in flight finish (4 s at most); then the profile scan in// progress, the Discord relay's drain and its Events worker's page (the gateway closes at once) and the// events outbox's send finish (unsent rows stay in their tables for the next start), community.db// closes, and the process exits 0, within the manifest's 5 s.-- const { within } = require('./graceful');++ const { gracefulStop, within } = require('openvibe-sdk/service');let profilesDone = null;-- require('./graceful').gracefulStop({++ gracefulStop({name: 'Community', server,++ drainMs: 4000, deadlineMs: 5000, deadlineExitCode: 1,stop: [() => { if (subscriptions) subscriptions.stop(); },() => { if (profiles) profilesDone = profiles.stop(); },@@@ -1,6 +1,6 @@'use strict';/**-- * Graceful stop (roadmap WS-P lifecycle; server/graceful.js), on the real process with the events++ * Graceful stop (roadmap WS-P lifecycle; openvibe-sdk/service), on the real process with the events* outbox, the Discord relay (with its Events worker and gateway), the Pulse subscriptions and the profile and search scans all on (pointed* at nothing): SIGTERM while a request is in flight and a keep-alive connection sits idle. New* connections are refused, the request is answered (Connection: close), every background worker is
@@ -18,7 +18,7 @@
@@ -18,7 +18,7 @@
"iovalkey": "^0.4.0",
"iovalkey": "^0.4.0",
"multer": "^2.4.0",
"multer": "^2.4.0",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.76.0",
"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.0",
+ "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",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.3.1",
"pg": "^8.23.0"
"pg": "^8.23.0"
},
},
@@ -1207,9 +1207,9 @@
@@ -1207,9 +1207,9 @@
}
}
},
},
"node_modules/openvibe-sdk": {
"node_modules/openvibe-sdk": {
- "version": "0.21.0",
- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.21.0",
- "integrity": "sha512-4gB+Lsj6Df1ztxI2ZNvPGUzXpSNV3FsEOP2VkcQ1hQIKdUdwH1BFEC16aQIwOrugmhe1U/CSmCWpPVb9bFu1ag==",
+ "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"
@@ -29,7 +29,7 @@
@@ -29,7 +29,7 @@
"iovalkey": "^0.4.0",
"iovalkey": "^0.4.0",
"multer": "^2.4.0",
"multer": "^2.4.0",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.76.0",
"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.0",
+ "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",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.3.1",
"pg": "^8.23.0"
"pg": "^8.23.0"
},
},
deleted file mode 100644
deleted file mode 100644
@@ -1,117 +0,0 @@
@@ -1,117 +0,0 @@
-'use strict';
-// ═══════════════════════════════════════════════════════════════
-// Graceful stop for OpenVibe.Community (roadmap WS-P lifecycle; openvibe-contracts
-// manifests/services/community.json → lifecycle.shutdown). systemd sends SIGTERM on a restart or a deploy;
-// SIGINT does the same by hand. On the first signal:
-//
-// 1. `stop` steps run: timers and scans stop taking new work (nothing new starts);
-// 2. the HTTP server stops taking connections, idle keep-alive connections are closed, requests in
-// flight are answered with `Connection: close`, and open event streams (text/event-stream) are
-// closed — EventSource reconnects on its own;
-// 3. the requests in flight finish, for at most `drainMs`; whatever is still open then is cut;
-// 4. `close` steps run: the relays settle, the database closes;
-// 5. the process exits 0.
-//
-// A hard `deadlineMs` bounds the whole stop (the manifest's deadlineSeconds): past it the process
-// exits 1. A second signal while stopping changes nothing.
-// ═══════════════════════════════════════════════════════════════
-
-const DRAIN_MS = 4000;
-const DEADLINE_MS = 5000;
-
-/**
- * @param {object} o
- * @param {string} o.name log prefix, e.g. 'Community'
- * @param {import('http').Server} o.server
- * @param {Array<Function>} [o.stop] run first, in order (sync or async; a failure is logged)
- * @param {Array<Function>} [o.close] run after the HTTP drain, in order
- * @param {number} [o.drainMs] how long requests in flight may take (default 4000)
- * @param {number} [o.deadlineMs] the whole stop, else exit 1 (default 5000)
- * @param {boolean} [o.signals] false: do not install SIGTERM/SIGINT handlers (tests)
- * @param {(code:number) => void} [o.exit] default process.exit
- * @returns {{ stop(signal?: string): Promise<number>, stopping(): boolean }}
- */
-function gracefulStop(o) {
- const { name, server } = o;
- const drainMs = o.drainMs == null ? DRAIN_MS : o.drainMs;
- const deadlineMs = o.deadlineMs == null ? DEADLINE_MS : o.deadlineMs;
- const log = o.log || console;
- const exit = o.exit || ((code) => process.exit(code));
- let started = null;
-
- // Every response in flight, so the stop can mark them Connection: close and close event streams.
- const inflight = new Set();
- server.prependListener('request', (req, res) => {
- if (started && !res.headersSent) res.setHeader('Connection', 'close');
- inflight.add(res);
- res.on('close', () => inflight.delete(res));
- });
-
- async function step(fn, phase) {
- try { await fn(); } catch (err) { log.warn(`[${name}] stop: ${phase} step failed: ${err && err.message}`); }
- }
-
- function isEventStream(res) {
- const ct = String(res.getHeader('content-type') || '');
- return /text\/event-stream/i.test(ct) || (typeof res._header === 'string' && /content-type:\s*text\/event-stream/i.test(res._header));
- }
-
- function drain() {
- return new Promise((resolve) => {
- let done = false;
- const finish = (cut) => {
- if (done) return;
- done = true;
- clearInterval(sweep);
- clearTimeout(timer);
- resolve(cut);
- };
- server.close(() => finish(0));
- for (const res of inflight) {
- if (isEventStream(res)) res.destroy();
- else if (!res.headersSent) res.setHeader('Connection', 'close');
- }
- // A connection whose response just finished goes idle: close it now rather than after keepAliveTimeout.
- const idle = () => { if (typeof server.closeIdleConnections === 'function') server.closeIdleConnections(); };
- idle();
- const sweep = setInterval(idle, 50);
- const timer = setTimeout(() => {
- const cut = inflight.size;
- if (typeof server.closeAllConnections === 'function') server.closeAllConnections();
- finish(cut || 1);
- }, drainMs);
- });
- }
-
- function stop(signal = 'stop') {
- if (started) return started;
- started = (async () => {
- const t0 = Date.now();
- log.log(`[${name}] ${signal}: stopping (requests in flight get ${drainMs} ms)`);
- const hard = setTimeout(() => { log.error(`[${name}] stop took longer than ${deadlineMs} ms: exiting 1`); exit(1); }, deadlineMs);
- if (hard.unref) hard.unref();
- for (const fn of o.stop || []) await step(fn, 'stop');
- const cut = await drain();
- if (cut) log.warn(`[${name}] ${cut} request(s) still open after ${drainMs} ms were cut`);
- for (const fn of o.close || []) await step(fn, 'close');
- clearTimeout(hard);
- log.log(`[${name}] stopped in ${Date.now() - t0} ms`);
- exit(0);
- return 0;
- })();
- return started;
- }
-
- if (o.signals !== false) {
- process.on('SIGTERM', () => { stop('SIGTERM'); });
- process.on('SIGINT', () => { stop('SIGINT'); });
- }
- return { stop, stopping: () => !!started };
-}
-
-/** Await `promise`, but no longer than `ms` (a best-effort step inside the deadline). */
-function within(ms, promise) {
- return Promise.race([Promise.resolve(promise).catch(() => {}), new Promise((r) => { const t = setTimeout(r, ms); if (t.unref) t.unref(); })]);
-}
-
-module.exports = { gracefulStop, within, DRAIN_MS, DEADLINE_MS };
@@ -41,17 +41,18 @@ const { createApp } = require('./app');
@@ -41,17 +41,18 @@ const { createApp } = require('./app');
let searchDocs = null;
let searchDocs = null;
try { searchDocs = require('./search/documents').createSearchDocuments({ db: require('./db').getDb() }); searchDocs.start(); } catch (err) { console.warn('[Search] documents not started:', err.message); }
try { searchDocs = require('./search/documents').createSearchDocuments({ db: require('./db').getDb() }); searchDocs.start(); } catch (err) { console.warn('[Search] documents not started:', err.message); }
- // ── Stop (roadmap WS-P lifecycle; server/graceful.js) ────────
+ // ── Stop (roadmap WS-P lifecycle; openvibe-sdk/service) ──────
// SIGTERM: the Pulse subscription retries, the community.profile scan and the search-document scans
// SIGTERM: the Pulse subscription retries, the community.profile scan and the search-document scans
// stop (nothing new starts); the server stops taking connections, closes idle keep-alive ones (it keeps
// stop (nothing new starts); the server stops taking connections, closes idle keep-alive ones (it keeps
// them 65 s otherwise) and lets requests in flight finish (4 s at most); then the profile scan in
// them 65 s otherwise) and lets requests in flight finish (4 s at most); then the profile scan in
// progress, the Discord relay's drain and its Events worker's page (the gateway closes at once) and the
// progress, the Discord relay's drain and its Events worker's page (the gateway closes at once) and the
// events outbox's send finish (unsent rows stay in their tables for the next start), community.db
// events outbox's send finish (unsent rows stay in their tables for the next start), community.db
// closes, and the process exits 0, within the manifest's 5 s.
// closes, and the process exits 0, within the manifest's 5 s.
- const { within } = require('./graceful');
+ const { gracefulStop, within } = require('openvibe-sdk/service');
let profilesDone = null;
let profilesDone = null;
- require('./graceful').gracefulStop({
+ gracefulStop({
name: 'Community', server,
name: 'Community', server,
+ drainMs: 4000, deadlineMs: 5000, deadlineExitCode: 1,
stop: [
stop: [
() => { if (subscriptions) subscriptions.stop(); },
() => { if (subscriptions) subscriptions.stop(); },
() => { if (profiles) profilesDone = profiles.stop(); },
() => { if (profiles) profilesDone = profiles.stop(); },
@@ -1,6 +1,6 @@
@@ -1,6 +1,6 @@
'use strict';
'use strict';
/**
/**
- * Graceful stop (roadmap WS-P lifecycle; server/graceful.js), on the real process with the events
+ * Graceful stop (roadmap WS-P lifecycle; openvibe-sdk/service), on the real process with the events
* outbox, the Discord relay (with its Events worker and gateway), the Pulse subscriptions and the profile and search scans all on (pointed
* outbox, the Discord relay (with its Events worker and gateway), the Pulse subscriptions and the profile and search scans all on (pointed
* at nothing): SIGTERM while a request is in flight and a keep-alive connection sits idle. New
* at nothing): SIGTERM while a request is in flight and a keep-alive connection sits idle. New
* connections are refused, the request is answered (Connection: close), every background worker is
* connections are refused, the request is answered (Connection: close), every background worker is