openvibers/openvibe.live Commits
Live's Events relay stops at shutdown, while the database is open (it published on past the pool's end) (#330)
Changes
@@@ -80,6 +80,8 @@ function toIso(v) {return Number.isNaN(d.getTime()) ? null : d.toISOString();}++let pruneTimer = null;++function init({ eventsUrl = EVENTS_URL, clientSecret = CLIENT_SECRET, fetchImpl, intervalMs } = {}) {if (outbox) return outbox;if (process.env.EVENTS_PUBLISH === 'off' || !eventsUrl || !clientSecret) {@@@ -107,8 +109,8 @@ function init({ eventsUrl = EVENTS_URL, clientSecret = CLIENT_SECRET, fetchImpl,db.getDb().afterCommit(() => { if (outbox) Promise.resolve(outbox.kick()).catch((err) => { stats.lastError = err.message; }); });});outbox.start();-- const prune = setInterval(() => { if (outbox) outbox.prune().catch(() => { /* next time */ }); }, PRUNE_EVERY_MS);-- if (prune.unref) prune.unref();++ pruneTimer = setInterval(() => { if (outbox) outbox.prune().catch(() => { /* next time */ }); }, PRUNE_EVERY_MS);++ if (pruneTimer.unref) pruneTimer.unref();outbox.pending().then((n) => console.log(`[Events] stream lifecycle → ${eventsUrl} (${n} pending)`), () => {});return outbox;}@@@ -142,6 +144,16 @@ async function counts() {return { ...status(), pending: Number(await outbox.pending()), rejected: Number(await outbox.rejected()) };}--function _reset() { if (outbox) outbox.stop(); outbox = null; db.onStreamLifecycle(null); stats.queued = 0; stats.lastError = null; }++/**++ * Shutdown: the relay sends nothing more (it ran on past the pool's end, "Cannot use a pool after calling end on the++ * pool", 2026-10-23) and a send in flight finishes first. Events queued while the process drains stay in the outbox++ * for the next boot: enqueue() still writes, kick() no longer wakes the relay.++ */++function stop() {++ if (pruneTimer) { clearInterval(pruneTimer); pruneTimer = null; }++ return outbox ? outbox.stop() : Promise.resolve();++}++++function _reset() { stop(); outbox = null; db.onStreamLifecycle(null); stats.queued = 0; stats.lastError = null; }--module.exports = { init, enqueue, kick, status, counts, envelopeFor, _reset };++module.exports = { init, enqueue, kick, stop, status, counts, envelopeFor, _reset };@@@ -186,6 +186,8 @@ const stopper = gracefulStop({// No new background runs from here on (the jobs helper's loops); and end SSE streams so// the drain can actually complete instead of always hitting the deadline.() => { require('./utils/jobs').stopAll(); },++ // The Events relay stops while the database is open (events queued from here on wait in the outbox).++ () => require('./events/stream-events').stop(),async () => { if (!drill.enabled && rumRestored) await db.setState('rum_state', JSON.stringify(rum.state())).catch(() => {}); },() => { require('./streaming/live-events').closeAll(); },],@@@ -103,6 +103,17 @@ const stub = http.createServer((req, res) => {assert.strictEqual(await outbox.pending(), 0);assert.strictEqual(published[2].payload.title, 'while down');++ // Shutdown: after stop() the relay sends nothing; a stream that ends while the process drains is queued for++ // the next boot.++ await outbox.flush();++ const sentBefore = published.length;++ await streamEvents.stop();++ await db.endStream(Number(later.lastInsertRowid));++ streamEvents.kick();++ await new Promise((r) => setTimeout(r, 200));++ assert.strictEqual(published.length, sentBefore, 'nothing is published after stop()');++ assert.strictEqual(await outbox.pending(), 1, 'the ended event waits in the outbox');++// A hook failure never blocks going live.db.onStreamLifecycle(async () => { throw new Error('hook broken'); });const log = console.warn; console.warn = () => {};@@@ -111,6 +122,7 @@ const stub = http.createServer((req, res) => {assert.ok(safe.lastInsertRowid);assert.strictEqual(streamEvents.status().enabled, true);++streamEvents._reset();stub.close();console.log('✅ stream lifecycle events: outbox, relay, retry, rollback');
@@ -80,6 +80,8 @@ function toIso(v) {
@@ -80,6 +80,8 @@ function toIso(v) {
return Number.isNaN(d.getTime()) ? null : d.toISOString();
return Number.isNaN(d.getTime()) ? null : d.toISOString();
}
}
+let pruneTimer = null;
+
function init({ eventsUrl = EVENTS_URL, clientSecret = CLIENT_SECRET, fetchImpl, intervalMs } = {}) {
function init({ eventsUrl = EVENTS_URL, clientSecret = CLIENT_SECRET, fetchImpl, intervalMs } = {}) {
if (outbox) return outbox;
if (outbox) return outbox;
if (process.env.EVENTS_PUBLISH === 'off' || !eventsUrl || !clientSecret) {
if (process.env.EVENTS_PUBLISH === 'off' || !eventsUrl || !clientSecret) {
@@ -107,8 +109,8 @@ function init({ eventsUrl = EVENTS_URL, clientSecret = CLIENT_SECRET, fetchImpl,
@@ -107,8 +109,8 @@ function init({ eventsUrl = EVENTS_URL, clientSecret = CLIENT_SECRET, fetchImpl,
db.getDb().afterCommit(() => { if (outbox) Promise.resolve(outbox.kick()).catch((err) => { stats.lastError = err.message; }); });
db.getDb().afterCommit(() => { if (outbox) Promise.resolve(outbox.kick()).catch((err) => { stats.lastError = err.message; }); });
});
});
outbox.start();
outbox.start();
- const prune = setInterval(() => { if (outbox) outbox.prune().catch(() => { /* next time */ }); }, PRUNE_EVERY_MS);
- if (prune.unref) prune.unref();
+ pruneTimer = setInterval(() => { if (outbox) outbox.prune().catch(() => { /* next time */ }); }, PRUNE_EVERY_MS);
+ if (pruneTimer.unref) pruneTimer.unref();
outbox.pending().then((n) => console.log(`[Events] stream lifecycle → ${eventsUrl} (${n} pending)`), () => {});
outbox.pending().then((n) => console.log(`[Events] stream lifecycle → ${eventsUrl} (${n} pending)`), () => {});
return outbox;
return outbox;
}
}
@@ -142,6 +144,16 @@ async function counts() {
@@ -142,6 +144,16 @@ async function counts() {
return { ...status(), pending: Number(await outbox.pending()), rejected: Number(await outbox.rejected()) };
return { ...status(), pending: Number(await outbox.pending()), rejected: Number(await outbox.rejected()) };
}
}
-function _reset() { if (outbox) outbox.stop(); outbox = null; db.onStreamLifecycle(null); stats.queued = 0; stats.lastError = null; }
+/**
+ * Shutdown: the relay sends nothing more (it ran on past the pool's end, "Cannot use a pool after calling end on the
+ * pool", 2026-10-23) and a send in flight finishes first. Events queued while the process drains stay in the outbox
+ * for the next boot: enqueue() still writes, kick() no longer wakes the relay.
+ */
+function stop() {
+ if (pruneTimer) { clearInterval(pruneTimer); pruneTimer = null; }
+ return outbox ? outbox.stop() : Promise.resolve();
+}
+
+function _reset() { stop(); outbox = null; db.onStreamLifecycle(null); stats.queued = 0; stats.lastError = null; }
-module.exports = { init, enqueue, kick, status, counts, envelopeFor, _reset };
+module.exports = { init, enqueue, kick, stop, status, counts, envelopeFor, _reset };
@@ -186,6 +186,8 @@ const stopper = gracefulStop({
@@ -186,6 +186,8 @@ const stopper = gracefulStop({
// No new background runs from here on (the jobs helper's loops); and end SSE streams so
// No new background runs from here on (the jobs helper's loops); and end SSE streams so
// the drain can actually complete instead of always hitting the deadline.
// the drain can actually complete instead of always hitting the deadline.
() => { require('./utils/jobs').stopAll(); },
() => { require('./utils/jobs').stopAll(); },
+ // The Events relay stops while the database is open (events queued from here on wait in the outbox).
+ () => require('./events/stream-events').stop(),
async () => { if (!drill.enabled && rumRestored) await db.setState('rum_state', JSON.stringify(rum.state())).catch(() => {}); },
async () => { if (!drill.enabled && rumRestored) await db.setState('rum_state', JSON.stringify(rum.state())).catch(() => {}); },
() => { require('./streaming/live-events').closeAll(); },
() => { require('./streaming/live-events').closeAll(); },
],
],
@@ -103,6 +103,17 @@ const stub = http.createServer((req, res) => {
@@ -103,6 +103,17 @@ const stub = http.createServer((req, res) => {
assert.strictEqual(await outbox.pending(), 0);
assert.strictEqual(await outbox.pending(), 0);
assert.strictEqual(published[2].payload.title, 'while down');
assert.strictEqual(published[2].payload.title, 'while down');
+ // Shutdown: after stop() the relay sends nothing; a stream that ends while the process drains is queued for
+ // the next boot.
+ await outbox.flush();
+ const sentBefore = published.length;
+ await streamEvents.stop();
+ await db.endStream(Number(later.lastInsertRowid));
+ streamEvents.kick();
+ await new Promise((r) => setTimeout(r, 200));
+ assert.strictEqual(published.length, sentBefore, 'nothing is published after stop()');
+ assert.strictEqual(await outbox.pending(), 1, 'the ended event waits in the outbox');
+
// A hook failure never blocks going live.
// A hook failure never blocks going live.
db.onStreamLifecycle(async () => { throw new Error('hook broken'); });
db.onStreamLifecycle(async () => { throw new Error('hook broken'); });
const log = console.warn; console.warn = () => {};
const log = console.warn; console.warn = () => {};
@@ -111,6 +122,7 @@ const stub = http.createServer((req, res) => {
@@ -111,6 +122,7 @@ const stub = http.createServer((req, res) => {
assert.ok(safe.lastInsertRowid);
assert.ok(safe.lastInsertRowid);
assert.strictEqual(streamEvents.status().enabled, true);
assert.strictEqual(streamEvents.status().enabled, true);
+
streamEvents._reset();
streamEvents._reset();
stub.close();
stub.close();
console.log('✅ stream lifecycle events: outbox, relay, retry, rollback');
console.log('✅ stream lifecycle events: outbox, relay, retry, rollback');