openvibers/openvibe.network Commits

Graceful stop, all of it (WS-P lifecycle follow-up): SIGTERM now also stops what used to run into the exit: the maintenance timers (notifications, email queue, retired-module sweep, session cleanup), the analytics prune, the ecosystem, deploy-drift and library-tag pollers, the frame rank/copy refreshes (frameService.stop()), the profile-event and grant-expiry timers; then, after the HTTP drain, the developer/profile/grant event relay's send in progress is awaited (unsent rows stay in the outbox), analytics flush and close, network.db closes, exit 0 within 10 s. The drain itself (server/graceful.js) now also answers requests in flight with Connection: close and keeps closing connections as they go idle: before, a keep-alive connection whose request was still running at SIGTERM held the exit for the keep-alive timeout after it finished. test/shutdown.test.js: SIGTERM with a request in flight and an idle keep-alive connection: refused afterwards, the request answered with Connection: close, every step run, exit 0 in < 10 s

OP OpenVibers committed 26 Sept 2026 · 631ed6f202 · 4 files, +187 −31


Changes

ddiff --git a/server/frame/service.js b/server/frame/service.js
iindex a052bb75..59dcf576 100644
---- a/server/frame/service.js
++++ b/server/frame/service.js
@@@ -204,20 +204,24 @@ function createFrameService(db, config, analytics, { privateKey = null, issuer =
res.status(204).end();
});
++ let timers = [];
function start() {
const rankAge = Date.now() - ((load('rank') || {}).at || 0), copyAge = Date.now() - ((load('copy') || {}).at || 0);
const t1 = setTimeout(() => refreshRank().catch(() => {}), rankAge > RANK_MS ? 8000 : RANK_MS - rankAge); t1.unref();
const i1 = setInterval(() => refreshRank().catch(() => {}), RANK_MS); i1.unref();
const t2 = setTimeout(() => refreshCopy().catch(() => {}), copyAge > COPY_MS ? 60_000 : COPY_MS - copyAge); t2.unref();
const i2 = setInterval(() => refreshCopy().catch(() => {}), COPY_MS); i2.unref();
++ timers = [t1, i1, t2, i2];
}
++ /** Graceful stop: no further rank or copy refreshes. */
++ function stop() { for (const t of timers) { clearTimeout(t); clearInterval(t); } timers = []; }
/** The open sites in navbar order (most used first), and when that use was last counted (ms, or null: cold start). */
function ranking() {
return { at: rankedAt, order: orderedOpen().map(s => s.id), everyMs: RANK_MS };
}
-- return { router, start, refreshRank, refreshCopy, payloadFor, ranking, _state: () => ({ rank, copy }) };
++ return { router, start, stop, refreshRank, refreshCopy, payloadFor, ranking, _state: () => ({ rank, copy }) };
}
module.exports = { createFrameService, BANNED };
ddiff --git a/server/graceful.js b/server/graceful.js
nnew file mode 100644
iindex 00000000..51c724be
---- /dev/null
++++ b/server/graceful.js
@@@ -0,0 +1,117 @@
++'use strict';
++// ═══════════════════════════════════════════════════════════════
++// Graceful stop for OpenVibe.Network (roadmap WS-P lifecycle; openvibe-contracts
++// manifests/services/network.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, pollers and relays 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 outbox relay settles, analytics flush, 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. 'Network'
++ * @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 };
ddiff --git a/server/index.js b/server/index.js
iindex 2df61230..2c4622f8 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -413,14 +413,16 @@ const ecosystemInternal = require('./registry/ecosystem').internalFromEnv(proces
if (ecosystemInternal.ignored.length) console.warn(`[Registry] ignored (not a loopback URL): ${ecosystemInternal.ignored.join(', ')}`);
const ecosystem = require('./registry/ecosystem').createEcosystemRegistry({ issuer: config.jwt.issuer, internalOverrides: ecosystemInternal.overrides });
// Production drift: each running service's deployed commit against its repository's main (WS-S task 7).
--require('./registry/deploy-drift').createDeployDrift({
++const deployDrift = require('./registry/deploy-drift').createDeployDrift({
services: () => ecosystem.releases().services.filter((r) => r.release).map((r) => ({ id: r.id, release: r.release, repository: ((require('openvibe-contracts').services.manifests.find((m) => m.id === r.id)) || {}).repository })),
token: () => process.env.GITHUB_TOKEN || require('./integrations/github').tokenOf(db) || '',
--}).start();
++});
++deployDrift.start();
app.use(ecosystem.router());
ecosystem.start();
// The released libraries' latest published tags (not the versions Network installs) for the registry.
--require('./registry/library-tags').createLibraryTags({ onUpdate: require('./registry/exposure').setLibraryReleases, token: () => require('./integrations/github').tokenOf(db) }).start();
++const libraryTags = require('./registry/library-tags').createLibraryTags({ onUpdate: require('./registry/exposure').setLibraryReleases, token: () => require('./integrations/github').tokenOf(db) });
++libraryTags.start();
// Operator status: GET /status (server-rendered, noindex), /api/v1/status, /api/v1/status/slo.
app.use(require('./status/routes').createStatusRoutes({ ecosystem }));
// What shipped network-wide: the changelog proxy every site's widget reads, and /updates.
@@@ -755,6 +757,8 @@ app.get(['/admin', '/admin/*'], (req, res) => {
app.use(require('./not-found').notFound);
// ── Start ────────────────────────────────────────────────────
++const timers = []; // the periodic maintenance below, cleared on stop
++let analyticsPrune = null;
const server = app.listen(config.port, config.host, () => {
// The home page renders the Tools catalog it already holds: fetch it now, not on the first visit.
require('./domains/catalog').refresh().catch(() => {});
@@@ -768,14 +772,14 @@ const server = app.listen(config.port, config.host, () => {
// ── Periodic Maintenance ─────────────────────────────────
// Clean expired notifications every hour
-- setInterval(() => notificationService.maintenance(), 60 * 60 * 1000);
++ timers.push(setInterval(() => notificationService.maintenance(), 60 * 60 * 1000));
// Process email queue every 2 minutes
-- setInterval(() => emailService.processQueue(notificationService), 2 * 60 * 1000);
++ timers.push(setInterval(() => emailService.processQueue(notificationService), 2 * 60 * 1000));
// Raw analytics retention (ADR-021), job analytics-prune: events older than 30 days go, in
// bounded batches; hourly/daily rollups stay. First run 5 minutes after boot, then every 24 h.
-- networkAnalytics.schedulePrune(analytics);
++ analyticsPrune = networkAnalytics.schedulePrune(analytics);
// User modules of a retired owning service (onOwnerRemoved, server/identity/modules.js): writes stop at
// once; delete-after-retention records go retentionDays after Network first saw the retirement.
@@@ -785,28 +789,38 @@ const server = app.listen(config.port, config.host, () => {
if (n) console.log(`[Modules] Deleted ${n} record(s) of retired namespace owners`);
} catch (e) { console.warn('[Modules] retired-owner sweep:', e.message); }
};
-- setTimeout(sweepModules, 5 * 60 * 1000);
-- setInterval(sweepModules, 24 * 60 * 60 * 1000);
++ timers.push(setTimeout(sweepModules, 5 * 60 * 1000), setInterval(sweepModules, 24 * 60 * 60 * 1000));
// Clean expired sessions daily
-- setInterval(() => {
++ timers.push(setInterval(() => {
const cleaned = db.prepare("DELETE FROM user_sessions WHERE expires_at < datetime('now') OR is_active = 0").run().changes;
if (cleaned > 0) console.log(`[Sessions] Cleaned ${cleaned} expired sessions`);
-- }, 24 * 60 * 60 * 1000);
++ }, 24 * 60 * 60 * 1000));
});
--// ── Stop (roadmap WS-P lifecycle) ────────────────────────────
--// systemd sends SIGTERM on a restart or deploy: stop taking connections, close idle keep-alive ones,
--// let requests in flight finish (at most 10 s, well inside the unit's stop timeout), then exit. Work
--// kept in the database (outboxes, the email queue) resumes on the next start.
--function shutdown(signal) {
-- if (shutdown.started) return;
-- shutdown.started = true;
-- console.log(`[Network] ${signal}: closing`);
-- server.close(() => process.exit(0));
-- if (typeof server.closeIdleConnections === 'function') server.closeIdleConnections();
-- setTimeout(() => process.exit(0), 10000).unref();
--}
--process.on('SIGTERM', () => shutdown('SIGTERM'));
--process.on('SIGINT', () => shutdown('SIGINT'));
--
++// ── Stop (roadmap WS-P lifecycle; server/graceful.js) ─────────
++// systemd sends SIGTERM on a restart or deploy. The maintenance timers, the registry, drift and library
++// pollers, the frame refreshes, the profile-event and grant-expiry timers stop (nothing new starts); the
++// server stops taking connections, closes idle keep-alive ones and lets requests in flight finish (8 s at
++// most, Connection: close); then the developer/profile/grant event relay finishes its send in progress
++// (unsent rows stay in the outbox), analytics flush and close, network.db closes, and the process exits 0,
++// within 10 s (well inside the unit's stop timeout). The email queue resumes on the next start.
++const { within } = require('./graceful');
++require('./graceful').gracefulStop({
++ name: 'Network', server, drainMs: 8000, deadlineMs: 10000,
++ stop: [
++ () => { for (const t of timers) { clearTimeout(t); clearInterval(t); } },
++ () => { if (analyticsPrune) analyticsPrune.stop(); },
++ () => ecosystem.stop(),
++ () => deployDrift.stop(),
++ () => libraryTags.stop(),
++ () => frameService.stop(),
++ () => require('./identity/profile-events').stop(),
++ () => require('./identity/grants-admin').stop(),
++ ],
++ close: [
++ () => within(1500, require('./developer/event-relay').stopRelay(db)),
++ () => { analytics.destroy(); analytics.db.close(); },
++ () => db.close(),
++ ],
++});
ddiff --git a/test/shutdown.test.js b/test/shutdown.test.js
iindex bd60d03a..f3e853ad 100644
---- a/test/shutdown.test.js
++++ b/test/shutdown.test.js
@@@ -1,6 +1,8 @@
'use strict';
--// A restart drains instead of cutting (roadmap WS-P lifecycle): on SIGTERM the server stops taking
--// connections, closes idle keep-alive ones and exits 0, promptly, even with a keep-alive client connected.
++// A restart drains instead of cutting (roadmap WS-P lifecycle; server/graceful.js): on SIGTERM the server
++// stops taking connections, closes idle keep-alive ones, answers the request in flight (Connection: close),
++// stops its timers, pollers and relay, closes the databases and exits 0 within the manifest's 10 s, promptly,
++// even with a keep-alive client connected.
// node test/shutdown.test.js
const assert = require('assert');
const fs = require('fs');
@@@ -33,12 +35,31 @@ const freePort = () => new Promise((resolve) => { const s = net.createServer();
// A keep-alive client that finished its request and keeps the connection open.
const agent = new http.Agent({ keepAlive: true });
await new Promise((resolve, reject) => http.get({ host: '127.0.0.1', port, path: '/api/health', agent }, (res) => { res.resume(); res.on('end', resolve); }).on('error', reject));
++ // A request in flight: its JSON body is still coming (express.json reads it before any route 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: '/api/graceful-probe', 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 new Promise((r) => setTimeout(r, 200));
const t0 = Date.now();
child.kill('SIGTERM');
-- const r = await Promise.race([exited, new Promise((res) => setTimeout(() => res({ timeout: true }), 8000))]);
-- assert.ok(!r.timeout, 'exits within 8 s with an idle keep-alive connection open');
++ await new Promise((r) => setTimeout(r, 300));
++ const refused = await fetch(`http://127.0.0.1:${port}/api/health`, { signal: AbortSignal.timeout(1000) }).then(() => false, () => true);
++ assert.ok(refused, 'no new connections once stopping');
++ req.end(body.subarray(10));
++ const a = await answered;
++ assert.ok(a.at > t0, 'the request was still in flight at SIGTERM');
++ assert.ok(a.status >= 200 && a.status < 600, `answered, not cut (${a.status})`);
++ assert.strictEqual(a.connection, 'close');
++ const r = await Promise.race([exited, new Promise((res) => setTimeout(() => res({ timeout: true }), 10000))]);
++ assert.ok(!r.timeout, "exits within the manifest's 10 s with an idle keep-alive connection open");
assert.deepStrictEqual([r.code, r.signal], [0, null], `a clean exit, not a kill: ${JSON.stringify(r)}\n${out.slice(-800)}`);
-- assert.ok(/SIGTERM: closing/.test(out), 'says it is closing');
++ assert.ok(/\[Network\] SIGTERM: stopping/.test(out), 'says it is stopping');
++ assert.ok(/\[Network\] stopped in \d+ ms/.test(out), `ran every stop and close step\n${out.slice(-800)}`);
++ assert.ok(!/stop: (stop|close) step failed/.test(out), `no step failed\n${out.slice(-1500)}`);
console.log(`shutdown: exited 0 in ${Date.now() - t0} ms`);
agent.destroy();
} finally {