openvibers/openvibe.community Commits
Graceful stop (WS-P lifecycle follow-up): SIGTERM used to close only the HTTP server (which waited on 65 s keep-alive connections until the 5 s fallback) while the events outbox, the Discord relay, the Pulse subscription retries and the profile and search-document scans kept running into the exit. Now (server/graceful.js): the subscription retries, the community.profile scan and the search-document scans stop; the server stops taking connections, closes idle keep-alive ones and lets requests in flight finish with Connection: close (4 s at most); then the profile scan in progress (it stops between people), the relay's drain and the outbox's send are awaited (unsent rows stay for the next start), community.db closes (db.closeDb) and the process exits 0, within 5 s (exit 1 past it). events.stop(), relay.stop() and profile.stop() resolve when their work in flight ends and start nothing after. test/shutdown.test.js: the real process with every worker on, SIGTERM with a request in flight and an idle keep-alive connection: refused afterwards, the request answered, every worker reported stopped, exit 0 in < 5 s
Changes
@@@ -565,7 +565,8 @@ to appear.```server/-- index.js process entry (listen, shutdown)++ index.js process entry (listen, graceful stop)++ graceful.js SIGTERM: stop timers and scans, drain HTTP (4 s), settle the relays, close the DB, exit 0 (5 s at most)app.js Express app factory: middleware, routesconfig.js env → configdb.js SQLite (better-sqlite3): schema, opened at COMMUNITY_DB_PATH@@@ -475,4 +475,11 @@ function getDb() {return _shared;}--module.exports = { openDb, getDb, registerFunctions, SCHEMA, newThreadAccessId };++/** Graceful stop: close the process-wide database (a later getDb() opens it again). */++function closeDb() {++ const db = _shared;++ _shared = null;++ if (db) try { db.close(); } catch { /* already closed */ }++}++++module.exports = { openDb, getDb, closeDb, registerFunctions, SCHEMA, newThreadAccessId };@@@ -20,6 +20,7 @@ const { createEventsClient, createOutbox } = require('openvibe-sdk/events');const config = require('./config');let outbox = null;++let pruneTimer = null;const stats = { queued: 0, lastError: null };const PRUNE_EVERY_MS = 6 * 60 * 60 * 1000;@@@ -35,8 +36,8 @@ function init(db, { eventsUrl = process.env.EVENTS_URL, clientSecret = config.oa});outbox.ensureSchema();outbox.start();-- const prune = setInterval(() => { try { outbox.prune(); } catch { /* next time */ } }, PRUNE_EVERY_MS);-- if (prune.unref) prune.unref();++ pruneTimer = setInterval(() => { try { outbox.prune(); } catch { /* next time */ } }, PRUNE_EVERY_MS);++ if (pruneTimer.unref) pruneTimer.unref();console.log(`[Events] community → ${eventsUrl} (${outbox.pending()} pending)`);return outbox;}@@@ -122,6 +123,15 @@ function status() {if (!outbox) return { enabled: false };return { enabled: true, pending: outbox.pending(), rejected: outbox.rejected(), queued_since_boot: stats.queued, last_error: stats.lastError };}++/**++ * Graceful stop (server/index.js): no further sends; resolves when the send in progress has finished.++ * Rows written after this (a request that was still finishing) stay in the outbox for the next start.++ */++function stop() {++ if (pruneTimer) clearInterval(pruneTimer);++ pruneTimer = null;++ return outbox ? outbox.stop() : Promise.resolve();++}function _reset() { if (outbox) outbox.stop(); outbox = null; stats.queued = 0; stats.lastError = null; }--module.exports = { init, record, moderationAction, pasteCreated, pasteUpdated, pasteDeleted, threadCreated, postCreated, commentCreated, status, _reset };++module.exports = { init, record, moderationAction, pasteCreated, pasteUpdated, pasteDeleted, threadCreated, postCreated, commentCreated, status, stop, _reset };nnew file mode 100644@@@ -0,0 +1,117 @@++'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 };@@@ -95,20 +95,27 @@ function createProfileModule({ db, config, fetchImpl = globalThis.fetch, log = cchanged('pastes', 'owner_subject', ['created_at', 'updated_at', 'deleted_at']),].join(' UNION ');const subjects = db.prepare(sql).all({ from, to }).map((r) => r.s).filter((s) => USR.test(String(s || '')));-- for (const s of subjects) await push(s);++ for (const s of subjects) { if (stopped) return subjects.length; await push(s); }lastScan = to;return subjects.length;}++ let running = null;++ let stopped = false;function start() {if (!enabled || timer) return false;++ stopped = false;ensureSchema(db);-- timer = setInterval(() => { scan().catch((err) => { stats.lastError = err.message; }); }, SCAN_MS);++ timer = setInterval(() => {++ if (running) return;++ running = scan().catch((err) => { stats.lastError = err.message; }).finally(() => { running = null; });++ }, SCAN_MS);if (timer.unref) timer.unref();log.log && log.log('[Modules] community.profile: scanning every 5 minutes');return true;}-- function stop() { if (timer) clearInterval(timer); timer = null; }++ /** Graceful stop: no further scans; resolves when a scan in progress has finished (it stops between people). */++ function stop() { stopped = true; if (timer) clearInterval(timer); timer = null; return running || Promise.resolve(); }return { enabled, push, scan, start, stop, stats: () => ({ enabled, ...stats }) };}@@@ -23,19 +23,41 @@ const server = app.listen(config.port, config.host, () => {});server.keepAliveTimeout = 65_000;// Subscribe Pulse to public activity at Events (idempotent; off without EVENTS_URL / COMMUNITY_EVENTS_SECRET).++let subscriptions = null;try {const secret = String(process.env.COMMUNITY_EVENTS_SECRET || '').split(',')[0].trim();-- require('./pulse/consumer').startSubscriptions({ config, port: config.port, secret });++ subscriptions = require('./pulse/consumer').startSubscriptions({ config, port: config.port, secret });} catch (err) { console.warn('[Pulse consumer] not subscribed:', err.message); }// community.profile on Network (Contracts 0.41.0): a 5-minute scan of changed authors (off without the client secret).--try { require('./identity/profile-module').createProfileModule({ db: require('./db').getDb(), config }).start(); } catch (err) { console.warn('[Modules] community.profile not started:', err.message); }++let profiles = null;++try { profiles = require('./identity/profile-module').createProfileModule({ db: require('./db').getDb(), config }); profiles.start(); } catch (err) { console.warn('[Modules] community.profile not started:', err.message); }// Public threads and pastes in OpenVibe.Search (WS-O task 10): community.index_document.* through the outbox (off without EVENTS_URL).--try { require('./search/documents').createSearchDocuments({ db: require('./db').getDb() }).start(); } catch (err) { console.warn('[Search] documents not started:', err.message); }++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); }--function shutdown(signal) {-- console.log(`[Community] ${signal} — closing`);-- server.close(() => process.exit(0));-- setTimeout(() => process.exit(0), 5000).unref();--}--process.on('SIGTERM', () => shutdown('SIGTERM'));--process.on('SIGINT', () => shutdown('SIGINT'));++// ── Stop (roadmap WS-P lifecycle; server/graceful.js) ────────++// 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 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');++let profilesDone = null;++require('./graceful').gracefulStop({++ name: 'Community', server,++ stop: [++ () => { if (subscriptions) subscriptions.stop(); },++ () => { if (profiles) profilesDone = profiles.stop(); },++ () => { if (searchDocs) searchDocs.stop(); },++ ],++ close: [++ () => within(1000, profilesDone),++ () => within(1000, app.locals.relay && app.locals.relay.stop()),++ () => within(1500, require('./events').stop()),++ () => {++ const ev = require('./events').status();++ console.log(`[Community] stopped: subscriptions ${subscriptions ? 'stopped' : 'off'}, profile scan ${profiles && profiles.enabled ? 'stopped' : 'off'}, search scans ${searchDocs ? 'stopped' : 'off'}, relay ${config.discordRelay.enabled ? 'stopped' : 'off'}, outbox ${ev.enabled ? `stopped (${ev.pending} pending)` : 'off'}`);++ require('./db').closeDb();++ },++ ],++});@@@ -147,11 +147,15 @@ function startSubscriptions({ config, port, secret, eventsUrl = process.env.EVEN};const delays = [0, 10_000, 60_000, 5 * 60_000, 15 * 60_000];let i = 0;-- const run = () => attempt().catch((err) => {-- if (++i < delays.length) { const t = setTimeout(run, delays[i]); if (t.unref) t.unref(); } else log.warn('[Pulse consumer] subscriptions not created:', err.message);-- });-- const t = setTimeout(run, delays[0]); if (t.unref) t.unref();-- return { topics: TOPICS, endpoint };++ let timer = null;++ let stopped = false;++ const run = () => { timer = null; if (stopped) return; attempt().catch((err) => {++ if (stopped) return;++ if (++i < delays.length) { timer = setTimeout(run, delays[i]); if (timer.unref) timer.unref(); } else log.warn('[Pulse consumer] subscriptions not created:', err.message);++ }); };++ timer = setTimeout(run, delays[0]); if (timer.unref) timer.unref();++ // Graceful stop: no further attempts (they run again at the next start).++ return { topics: TOPICS, endpoint, stop() { stopped = true; if (timer) clearTimeout(timer); timer = null; } };}module.exports = { createPulseConsumer, startSubscriptions, itemFor, TOPICS };@@@ -43,6 +43,7 @@ function createDiscordRelay({ db, config = {}, env = process.env, fetchImpl = glconst refAllowed = (name) => typeof name === 'string' && ENV_NAME.test(name) && (allowed ? allowed.has(name) : WEBHOOK_VAR.test(name));let timer = null;let draining = null;++ let stopped = false;function deliveryKey(threadId, mappingId) { return `thread:${threadId}:mapping:${mappingId}`; }@@@ -136,6 +137,7 @@ function createDiscordRelay({ db, config = {}, env = process.env, fetchImpl = gl/** Send every due delivery once. One drain at a time. → { delivered, retry, failed } */function drain() {if (draining) return draining;++ if (stopped) return Promise.resolve({ delivered: 0, retry: 0, failed: 0 });draining = (async () => {const summary = { delivered: 0, retry: 0, failed: 0 };if (!enabled) return summary;@@@ -149,11 +151,13 @@ function createDiscordRelay({ db, config = {}, env = process.env, fetchImpl = glfunction start() {if (!enabled || timer) return;++ stopped = false;timer = setInterval(() => { drain().catch((err) => console.warn('[Relay] drain failed:', err.message)); }, pollMs);if (timer.unref) timer.unref();kick();}-- function stop() { if (timer) clearInterval(timer); timer = null; }++ /** Graceful stop: no further drains; resolves when the drain in progress has finished (the rest stay pending). */++ function stop() { stopped = true; if (timer) clearInterval(timer); timer = null; return draining || Promise.resolve(); }// ── admin (staff) ────────────────────────────────────────function shapeMapping(m) {nnew file mode 100644@@@ -0,0 +1,84 @@++'use strict';++/**++ * Graceful stop (roadmap WS-P lifecycle; server/graceful.js), on the real process with the events++ * outbox, the Discord relay, 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++ * stopped, community.db is closed, and the process exits 0 within the manifest's 5 s; the idle++ * keep-alive connection (65 s otherwise) does not hold the stop.++ */++const assert = require('assert');++const fs = require('fs');++const http = require('http');++const net = require('net');++const os = require('os');++const path = require('path');++const { spawn } = require('child_process');++++const sleep = (ms) => new Promise((r) => setTimeout(r, ms));++const freePort = () => new Promise((resolve) => { const s = net.createServer(); s.listen(0, '127.0.0.1', () => { const { port } = s.address(); s.close(() => resolve(port)); }); });++++(async () => {++ const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-community-stop-'));++ const port = await freePort();++ const child = spawn(process.execPath, [path.join(__dirname, '..', 'server', 'index.js')], {++ cwd: path.join(__dirname, '..'),++ env: {++ ...process.env, PORT: String(port), HOST: '127.0.0.1', NODE_ENV: 'test', COMMUNITY_DB_PATH: path.join(dir, 'community.db'),++ EVENTS_URL: 'http://127.0.0.1:9', OV_NETWORK_INTERNAL_URL: 'http://127.0.0.1:9', OV_NETWORK_URL: 'http://127.0.0.1:9',++ OV_OAUTH_CLIENT_SECRET: 'test-secret-not-real', COMMUNITY_EVENTS_SECRET: 'e'.repeat(40), DISCORD_RELAY_ENABLED: '1',++ },++ stdio: ['ignore', 'pipe', 'pipe'],++ });++ let out = '';++ child.stdout.on('data', (d) => { out += d; });++ child.stderr.on('data', (d) => { out += d; });++ const exited = new Promise((r) => child.on('exit', (code, signal) => r({ code, signal })));++ const base = `http://127.0.0.1:${port}`;++ try {++ let up = false;++ for (let i = 0; i < 150 && !up; i++) {++ up = await fetch(`${base}/api/health`).then((r) => r.ok).catch(() => false);++ if (!up) await sleep(100);++ }++ assert.ok(up, `the server did not start:\n${out}`);++ assert.match(out, /\[Events\] community →/, 'the outbox is on');++++ // An idle keep-alive connection (the server keeps them 65 s).++ const agent = new http.Agent({ keepAlive: true });++ await new Promise((resolve, reject) => http.get(`${base}/api/health`, { agent }, (res) => { res.resume(); res.on('end', resolve); }).on('error', reject));++++ // A request in flight: POST /internal/events reads its whole body before it answers.++ const body = Buffer.from(JSON.stringify({ probe: 'graceful-stop', pad: 'x'.repeat(64) }));++ const req = http.request({ host: '127.0.0.1', port, path: '/internal/events', method: 'POST', agent: false, headers: { 'Content-Type': 'application/json', 'Content-Length': body.length } });++ const answered = new Promise((resolve, reject) => {++ req.on('response', (res) => { res.resume(); res.on('end', () => resolve({ status: res.statusCode, connection: res.headers.connection, at: Date.now() })); });++ req.on('error', reject);++ });++ req.write(body.subarray(0, 10));++ await sleep(200);++++ const t0 = Date.now();++ child.kill('SIGTERM');++ await sleep(300);++ const refused = await fetch(`${base}/api/health`, { signal: AbortSignal.timeout(1000) }).then(() => false, () => true);++ assert.ok(refused, 'no new connections once stopping');++ req.end(body.subarray(10));++ const r = await answered;++ assert.ok(r.at > t0, 'the request was still in flight at SIGTERM');++ assert.strictEqual(r.status, 401, 'answered (an unsigned delivery), not cut');++ assert.strictEqual(r.connection, 'close');++++ const { code, signal } = await exited;++ const ms = Date.now() - t0;++ assert.strictEqual(code, 0, `exit 0 (got ${code} ${signal})\n${out.slice(-2000)}`);++ assert.ok(ms < 5000, `within the manifest's 5 s (${ms} ms)`);++ assert.match(out, /\[Community\] stopped: subscriptions stopped, profile scan stopped, search scans stopped, relay stopped, outbox stopped \(\d+ pending\)/, out.slice(-2000));++ assert.match(out, /\[Community\] stopped in \d+ ms/);++ agent.destroy();++ console.log(`shutdown: SIGTERM → exit 0 in ${ms} ms, the request in flight answered, every worker stopped`);++ } finally {++ if (child.exitCode === null && child.signalCode === null) child.kill('SIGKILL');++ fs.rmSync(dir, { recursive: true, force: true });++ }++})().catch((err) => { console.error(err); process.exitCode = 1; });
@@ -565,7 +565,8 @@ to appear.
@@ -565,7 +565,8 @@ to appear.
```
```
server/
server/
- index.js process entry (listen, shutdown)
+ index.js process entry (listen, graceful stop)
+ graceful.js SIGTERM: stop timers and scans, drain HTTP (4 s), settle the relays, close the DB, exit 0 (5 s at most)
app.js Express app factory: middleware, routes
app.js Express app factory: middleware, routes
config.js env → config
config.js env → config
db.js SQLite (better-sqlite3): schema, opened at COMMUNITY_DB_PATH
db.js SQLite (better-sqlite3): schema, opened at COMMUNITY_DB_PATH
@@ -475,4 +475,11 @@ function getDb() {
@@ -475,4 +475,11 @@ function getDb() {
return _shared;
return _shared;
}
}
-module.exports = { openDb, getDb, registerFunctions, SCHEMA, newThreadAccessId };
+/** Graceful stop: close the process-wide database (a later getDb() opens it again). */
+function closeDb() {
+ const db = _shared;
+ _shared = null;
+ if (db) try { db.close(); } catch { /* already closed */ }
+}
+
+module.exports = { openDb, getDb, closeDb, registerFunctions, SCHEMA, newThreadAccessId };
@@ -20,6 +20,7 @@ const { createEventsClient, createOutbox } = require('openvibe-sdk/events');
@@ -20,6 +20,7 @@ const { createEventsClient, createOutbox } = require('openvibe-sdk/events');
const config = require('./config');
const config = require('./config');
let outbox = null;
let outbox = null;
+let pruneTimer = null;
const stats = { queued: 0, lastError: null };
const stats = { queued: 0, lastError: null };
const PRUNE_EVERY_MS = 6 * 60 * 60 * 1000;
const PRUNE_EVERY_MS = 6 * 60 * 60 * 1000;
@@ -35,8 +36,8 @@ function init(db, { eventsUrl = process.env.EVENTS_URL, clientSecret = config.oa
@@ -35,8 +36,8 @@ function init(db, { eventsUrl = process.env.EVENTS_URL, clientSecret = config.oa
});
});
outbox.ensureSchema();
outbox.ensureSchema();
outbox.start();
outbox.start();
- const prune = setInterval(() => { try { outbox.prune(); } catch { /* next time */ } }, PRUNE_EVERY_MS);
- if (prune.unref) prune.unref();
+ pruneTimer = setInterval(() => { try { outbox.prune(); } catch { /* next time */ } }, PRUNE_EVERY_MS);
+ if (pruneTimer.unref) pruneTimer.unref();
console.log(`[Events] community → ${eventsUrl} (${outbox.pending()} pending)`);
console.log(`[Events] community → ${eventsUrl} (${outbox.pending()} pending)`);
return outbox;
return outbox;
}
}
@@ -122,6 +123,15 @@ function status() {
@@ -122,6 +123,15 @@ function status() {
if (!outbox) return { enabled: false };
if (!outbox) return { enabled: false };
return { enabled: true, pending: outbox.pending(), rejected: outbox.rejected(), queued_since_boot: stats.queued, last_error: stats.lastError };
return { enabled: true, pending: outbox.pending(), rejected: outbox.rejected(), queued_since_boot: stats.queued, last_error: stats.lastError };
}
}
+/**
+ * Graceful stop (server/index.js): no further sends; resolves when the send in progress has finished.
+ * Rows written after this (a request that was still finishing) stay in the outbox for the next start.
+ */
+function stop() {
+ if (pruneTimer) clearInterval(pruneTimer);
+ pruneTimer = null;
+ return outbox ? outbox.stop() : Promise.resolve();
+}
function _reset() { if (outbox) outbox.stop(); outbox = null; stats.queued = 0; stats.lastError = null; }
function _reset() { if (outbox) outbox.stop(); outbox = null; stats.queued = 0; stats.lastError = null; }
-module.exports = { init, record, moderationAction, pasteCreated, pasteUpdated, pasteDeleted, threadCreated, postCreated, commentCreated, status, _reset };
+module.exports = { init, record, moderationAction, pasteCreated, pasteUpdated, pasteDeleted, threadCreated, postCreated, commentCreated, status, stop, _reset };
new file mode 100644
new file mode 100644
@@ -0,0 +1,117 @@
@@ -0,0 +1,117 @@
+'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 };
@@ -95,20 +95,27 @@ function createProfileModule({ db, config, fetchImpl = globalThis.fetch, log = c
@@ -95,20 +95,27 @@ function createProfileModule({ db, config, fetchImpl = globalThis.fetch, log = c
changed('pastes', 'owner_subject', ['created_at', 'updated_at', 'deleted_at']),
changed('pastes', 'owner_subject', ['created_at', 'updated_at', 'deleted_at']),
].join(' UNION ');
].join(' UNION ');
const subjects = db.prepare(sql).all({ from, to }).map((r) => r.s).filter((s) => USR.test(String(s || '')));
const subjects = db.prepare(sql).all({ from, to }).map((r) => r.s).filter((s) => USR.test(String(s || '')));
- for (const s of subjects) await push(s);
+ for (const s of subjects) { if (stopped) return subjects.length; await push(s); }
lastScan = to;
lastScan = to;
return subjects.length;
return subjects.length;
}
}
+ let running = null;
+ let stopped = false;
function start() {
function start() {
if (!enabled || timer) return false;
if (!enabled || timer) return false;
+ stopped = false;
ensureSchema(db);
ensureSchema(db);
- timer = setInterval(() => { scan().catch((err) => { stats.lastError = err.message; }); }, SCAN_MS);
+ timer = setInterval(() => {
+ if (running) return;
+ running = scan().catch((err) => { stats.lastError = err.message; }).finally(() => { running = null; });
+ }, SCAN_MS);
if (timer.unref) timer.unref();
if (timer.unref) timer.unref();
log.log && log.log('[Modules] community.profile: scanning every 5 minutes');
log.log && log.log('[Modules] community.profile: scanning every 5 minutes');
return true;
return true;
}
}
- function stop() { if (timer) clearInterval(timer); timer = null; }
+ /** Graceful stop: no further scans; resolves when a scan in progress has finished (it stops between people). */
+ function stop() { stopped = true; if (timer) clearInterval(timer); timer = null; return running || Promise.resolve(); }
return { enabled, push, scan, start, stop, stats: () => ({ enabled, ...stats }) };
return { enabled, push, scan, start, stop, stats: () => ({ enabled, ...stats }) };
}
}
@@ -23,19 +23,41 @@ const server = app.listen(config.port, config.host, () => {
@@ -23,19 +23,41 @@ const server = app.listen(config.port, config.host, () => {
});
});
server.keepAliveTimeout = 65_000;
server.keepAliveTimeout = 65_000;
// Subscribe Pulse to public activity at Events (idempotent; off without EVENTS_URL / COMMUNITY_EVENTS_SECRET).
// Subscribe Pulse to public activity at Events (idempotent; off without EVENTS_URL / COMMUNITY_EVENTS_SECRET).
+let subscriptions = null;
try {
try {
const secret = String(process.env.COMMUNITY_EVENTS_SECRET || '').split(',')[0].trim();
const secret = String(process.env.COMMUNITY_EVENTS_SECRET || '').split(',')[0].trim();
- require('./pulse/consumer').startSubscriptions({ config, port: config.port, secret });
+ subscriptions = require('./pulse/consumer').startSubscriptions({ config, port: config.port, secret });
} catch (err) { console.warn('[Pulse consumer] not subscribed:', err.message); }
} catch (err) { console.warn('[Pulse consumer] not subscribed:', err.message); }
// community.profile on Network (Contracts 0.41.0): a 5-minute scan of changed authors (off without the client secret).
// community.profile on Network (Contracts 0.41.0): a 5-minute scan of changed authors (off without the client secret).
-try { require('./identity/profile-module').createProfileModule({ db: require('./db').getDb(), config }).start(); } catch (err) { console.warn('[Modules] community.profile not started:', err.message); }
+let profiles = null;
+try { profiles = require('./identity/profile-module').createProfileModule({ db: require('./db').getDb(), config }); profiles.start(); } catch (err) { console.warn('[Modules] community.profile not started:', err.message); }
// Public threads and pastes in OpenVibe.Search (WS-O task 10): community.index_document.* through the outbox (off without EVENTS_URL).
// Public threads and pastes in OpenVibe.Search (WS-O task 10): community.index_document.* through the outbox (off without EVENTS_URL).
-try { require('./search/documents').createSearchDocuments({ db: require('./db').getDb() }).start(); } catch (err) { console.warn('[Search] documents not started:', err.message); }
+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); }
-function shutdown(signal) {
- console.log(`[Community] ${signal} — closing`);
- server.close(() => process.exit(0));
- setTimeout(() => process.exit(0), 5000).unref();
-}
-process.on('SIGTERM', () => shutdown('SIGTERM'));
-process.on('SIGINT', () => shutdown('SIGINT'));
+// ── Stop (roadmap WS-P lifecycle; server/graceful.js) ────────
+// 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 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');
+let profilesDone = null;
+require('./graceful').gracefulStop({
+ name: 'Community', server,
+ stop: [
+ () => { if (subscriptions) subscriptions.stop(); },
+ () => { if (profiles) profilesDone = profiles.stop(); },
+ () => { if (searchDocs) searchDocs.stop(); },
+ ],
+ close: [
+ () => within(1000, profilesDone),
+ () => within(1000, app.locals.relay && app.locals.relay.stop()),
+ () => within(1500, require('./events').stop()),
+ () => {
+ const ev = require('./events').status();
+ console.log(`[Community] stopped: subscriptions ${subscriptions ? 'stopped' : 'off'}, profile scan ${profiles && profiles.enabled ? 'stopped' : 'off'}, search scans ${searchDocs ? 'stopped' : 'off'}, relay ${config.discordRelay.enabled ? 'stopped' : 'off'}, outbox ${ev.enabled ? `stopped (${ev.pending} pending)` : 'off'}`);
+ require('./db').closeDb();
+ },
+ ],
+});
@@ -147,11 +147,15 @@ function startSubscriptions({ config, port, secret, eventsUrl = process.env.EVEN
@@ -147,11 +147,15 @@ function startSubscriptions({ config, port, secret, eventsUrl = process.env.EVEN
};
};
const delays = [0, 10_000, 60_000, 5 * 60_000, 15 * 60_000];
const delays = [0, 10_000, 60_000, 5 * 60_000, 15 * 60_000];
let i = 0;
let i = 0;
- const run = () => attempt().catch((err) => {
- if (++i < delays.length) { const t = setTimeout(run, delays[i]); if (t.unref) t.unref(); } else log.warn('[Pulse consumer] subscriptions not created:', err.message);
- });
- const t = setTimeout(run, delays[0]); if (t.unref) t.unref();
- return { topics: TOPICS, endpoint };
+ let timer = null;
+ let stopped = false;
+ const run = () => { timer = null; if (stopped) return; attempt().catch((err) => {
+ if (stopped) return;
+ if (++i < delays.length) { timer = setTimeout(run, delays[i]); if (timer.unref) timer.unref(); } else log.warn('[Pulse consumer] subscriptions not created:', err.message);
+ }); };
+ timer = setTimeout(run, delays[0]); if (timer.unref) timer.unref();
+ // Graceful stop: no further attempts (they run again at the next start).
+ return { topics: TOPICS, endpoint, stop() { stopped = true; if (timer) clearTimeout(timer); timer = null; } };
}
}
module.exports = { createPulseConsumer, startSubscriptions, itemFor, TOPICS };
module.exports = { createPulseConsumer, startSubscriptions, itemFor, TOPICS };
@@ -43,6 +43,7 @@ function createDiscordRelay({ db, config = {}, env = process.env, fetchImpl = gl
@@ -43,6 +43,7 @@ function createDiscordRelay({ db, config = {}, env = process.env, fetchImpl = gl
const refAllowed = (name) => typeof name === 'string' && ENV_NAME.test(name) && (allowed ? allowed.has(name) : WEBHOOK_VAR.test(name));
const refAllowed = (name) => typeof name === 'string' && ENV_NAME.test(name) && (allowed ? allowed.has(name) : WEBHOOK_VAR.test(name));
let timer = null;
let timer = null;
let draining = null;
let draining = null;
+ let stopped = false;
function deliveryKey(threadId, mappingId) { return `thread:${threadId}:mapping:${mappingId}`; }
function deliveryKey(threadId, mappingId) { return `thread:${threadId}:mapping:${mappingId}`; }
@@ -136,6 +137,7 @@ function createDiscordRelay({ db, config = {}, env = process.env, fetchImpl = gl
@@ -136,6 +137,7 @@ function createDiscordRelay({ db, config = {}, env = process.env, fetchImpl = gl
/** Send every due delivery once. One drain at a time. → { delivered, retry, failed } */
/** Send every due delivery once. One drain at a time. → { delivered, retry, failed } */
function drain() {
function drain() {
if (draining) return draining;
if (draining) return draining;
+ if (stopped) return Promise.resolve({ delivered: 0, retry: 0, failed: 0 });
draining = (async () => {
draining = (async () => {
const summary = { delivered: 0, retry: 0, failed: 0 };
const summary = { delivered: 0, retry: 0, failed: 0 };
if (!enabled) return summary;
if (!enabled) return summary;
@@ -149,11 +151,13 @@ function createDiscordRelay({ db, config = {}, env = process.env, fetchImpl = gl
@@ -149,11 +151,13 @@ function createDiscordRelay({ db, config = {}, env = process.env, fetchImpl = gl
function start() {
function start() {
if (!enabled || timer) return;
if (!enabled || timer) return;
+ stopped = false;
timer = setInterval(() => { drain().catch((err) => console.warn('[Relay] drain failed:', err.message)); }, pollMs);
timer = setInterval(() => { drain().catch((err) => console.warn('[Relay] drain failed:', err.message)); }, pollMs);
if (timer.unref) timer.unref();
if (timer.unref) timer.unref();
kick();
kick();
}
}
- function stop() { if (timer) clearInterval(timer); timer = null; }
+ /** Graceful stop: no further drains; resolves when the drain in progress has finished (the rest stay pending). */
+ function stop() { stopped = true; if (timer) clearInterval(timer); timer = null; return draining || Promise.resolve(); }
// ── admin (staff) ────────────────────────────────────────
// ── admin (staff) ────────────────────────────────────────
function shapeMapping(m) {
function shapeMapping(m) {
new file mode 100644
new file mode 100644
@@ -0,0 +1,84 @@
@@ -0,0 +1,84 @@
+'use strict';
+/**
+ * Graceful stop (roadmap WS-P lifecycle; server/graceful.js), on the real process with the events
+ * outbox, the Discord relay, 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
+ * stopped, community.db is closed, and the process exits 0 within the manifest's 5 s; the idle
+ * keep-alive connection (65 s otherwise) does not hold the stop.
+ */
+const assert = require('assert');
+const fs = require('fs');
+const http = require('http');
+const net = require('net');
+const os = require('os');
+const path = require('path');
+const { spawn } = require('child_process');
+
+const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
+const freePort = () => new Promise((resolve) => { const s = net.createServer(); s.listen(0, '127.0.0.1', () => { const { port } = s.address(); s.close(() => resolve(port)); }); });
+
+(async () => {
+ const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-community-stop-'));
+ const port = await freePort();
+ const child = spawn(process.execPath, [path.join(__dirname, '..', 'server', 'index.js')], {
+ cwd: path.join(__dirname, '..'),
+ env: {
+ ...process.env, PORT: String(port), HOST: '127.0.0.1', NODE_ENV: 'test', COMMUNITY_DB_PATH: path.join(dir, 'community.db'),
+ EVENTS_URL: 'http://127.0.0.1:9', OV_NETWORK_INTERNAL_URL: 'http://127.0.0.1:9', OV_NETWORK_URL: 'http://127.0.0.1:9',
+ OV_OAUTH_CLIENT_SECRET: 'test-secret-not-real', COMMUNITY_EVENTS_SECRET: 'e'.repeat(40), DISCORD_RELAY_ENABLED: '1',
+ },
+ stdio: ['ignore', 'pipe', 'pipe'],
+ });
+ let out = '';
+ child.stdout.on('data', (d) => { out += d; });
+ child.stderr.on('data', (d) => { out += d; });
+ const exited = new Promise((r) => child.on('exit', (code, signal) => r({ code, signal })));
+ const base = `http://127.0.0.1:${port}`;
+ try {
+ let up = false;
+ for (let i = 0; i < 150 && !up; i++) {
+ up = await fetch(`${base}/api/health`).then((r) => r.ok).catch(() => false);
+ if (!up) await sleep(100);
+ }
+ assert.ok(up, `the server did not start:\n${out}`);
+ assert.match(out, /\[Events\] community →/, 'the outbox is on');
+
+ // An idle keep-alive connection (the server keeps them 65 s).
+ const agent = new http.Agent({ keepAlive: true });
+ await new Promise((resolve, reject) => http.get(`${base}/api/health`, { agent }, (res) => { res.resume(); res.on('end', resolve); }).on('error', reject));
+
+ // A request in flight: POST /internal/events reads its whole body before it answers.
+ const body = Buffer.from(JSON.stringify({ probe: 'graceful-stop', pad: 'x'.repeat(64) }));
+ const req = http.request({ host: '127.0.0.1', port, path: '/internal/events', method: 'POST', agent: false, headers: { 'Content-Type': 'application/json', 'Content-Length': body.length } });
+ const answered = new Promise((resolve, reject) => {
+ req.on('response', (res) => { res.resume(); res.on('end', () => resolve({ status: res.statusCode, connection: res.headers.connection, at: Date.now() })); });
+ req.on('error', reject);
+ });
+ req.write(body.subarray(0, 10));
+ await sleep(200);
+
+ const t0 = Date.now();
+ child.kill('SIGTERM');
+ await sleep(300);
+ const refused = await fetch(`${base}/api/health`, { signal: AbortSignal.timeout(1000) }).then(() => false, () => true);
+ assert.ok(refused, 'no new connections once stopping');
+ req.end(body.subarray(10));
+ const r = await answered;
+ assert.ok(r.at > t0, 'the request was still in flight at SIGTERM');
+ assert.strictEqual(r.status, 401, 'answered (an unsigned delivery), not cut');
+ assert.strictEqual(r.connection, 'close');
+
+ const { code, signal } = await exited;
+ const ms = Date.now() - t0;
+ assert.strictEqual(code, 0, `exit 0 (got ${code} ${signal})\n${out.slice(-2000)}`);
+ assert.ok(ms < 5000, `within the manifest's 5 s (${ms} ms)`);
+ assert.match(out, /\[Community\] stopped: subscriptions stopped, profile scan stopped, search scans stopped, relay stopped, outbox stopped \(\d+ pending\)/, out.slice(-2000));
+ assert.match(out, /\[Community\] stopped in \d+ ms/);
+ agent.destroy();
+ console.log(`shutdown: SIGTERM → exit 0 in ${ms} ms, the request in flight answered, every worker stopped`);
+ } finally {
+ if (child.exitCode === null && child.signalCode === null) child.kill('SIGKILL');
+ fs.rmSync(dir, { recursive: true, force: true });
+ }
+})().catch((err) => { console.error(err); process.exitCode = 1; });