openvibers/openvibe.ai Commits
Graceful stop stops the ai.run.* relay (WS-P lifecycle follow-up): close() (SIGTERM) now, after runs.drain() has queued the last run events, stops the events outbox and awaits its send in progress (events.stop(); unsent rows stay for the next start) and clears its prune timer, before the database closes; it used to keep relaying until the exit. test/events.test.js: close() leaves the relay stopped and the unsent rows in the outbox
Changes
@@@ -16,6 +16,7 @@ const { createServiceTokenClient } = require('openvibe-sdk/auth');const { createEventsClient, createOutbox } = require('openvibe-sdk/events');let outbox = null;++let pruneTimer = null;const stats = { queued: 0, lastError: null };const PRUNE_EVERY_MS = 6 * 60 * 60 * 1000;@@@ -32,8 +33,8 @@ function init(db, { eventsUrl = process.env.EVENTS_URL, clientSecret = process.e});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();log.log(`[Events] ai → ${eventsUrl} (${outbox.pending()} pending)`);return outbox;}@@@ -92,6 +93,17 @@ 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 (start().close(), after runs.drain()): the relay stops and nothing more is queued; resolves++ * when the send in progress has finished. Unsent rows stay in the outbox for the next start.++ */++async function stop() {++ if (pruneTimer) clearInterval(pruneTimer);++ pruneTimer = null;++ const o = outbox;++ outbox = null;++ if (o) await o.stop();++}function _reset() { if (outbox) outbox.stop(); outbox = null; stats.queued = 0; stats.lastError = null; }--module.exports = { init, runChanged, payloadOf, status, _reset };++module.exports = { init, runChanged, payloadOf, status, stop, _reset };@@@ -79,6 +79,7 @@ async function start({ config, clock = { now: () => Date.now() }, fetchImpl = glawait new Promise(resolve => server.close(() => resolve()));}await runs.drain();++ await require('./events').stop(); // after the runs: their last ai.run.* rows are queued firstconst w = pool.adapter('whisper');if (w && w.adapter.killActive) w.adapter.killActive();db.close();@@@ -60,5 +60,16 @@ t.test('a failed run queues ai.run.failed with its error code; a restart announcvalid(ev);});--t.test('stop', async () => { require('../server/events')._reset(); await h.stop(); });++t.test('close() stops the ai.run.* relay (unsent rows stay in the outbox)', async () => {++ const events = require('../server/events');++ const pending = events.status().pending;++ assert.ok(pending > 0, 'rows are waiting (Events is unreachable)');++ await h.stop();++ assert.strictEqual(events.status().enabled, false, 'the relay is stopped');++ const Database = require('better-sqlite3');++ const db = new Database(require('path').join(h.dir, 'ai.db'), { readonly: true });++ assert.strictEqual(db.prepare('SELECT COUNT(*) AS n FROM event_outbox WHERE sent_at IS NULL AND rejected_at IS NULL').get().n, pending, 'and its rows wait for the next start');++ db.close();++ events._reset();++});t.run();
@@ -16,6 +16,7 @@ const { createServiceTokenClient } = require('openvibe-sdk/auth');
@@ -16,6 +16,7 @@ const { createServiceTokenClient } = require('openvibe-sdk/auth');
const { createEventsClient, createOutbox } = require('openvibe-sdk/events');
const { createEventsClient, createOutbox } = require('openvibe-sdk/events');
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;
@@ -32,8 +33,8 @@ function init(db, { eventsUrl = process.env.EVENTS_URL, clientSecret = process.e
@@ -32,8 +33,8 @@ function init(db, { eventsUrl = process.env.EVENTS_URL, clientSecret = process.e
});
});
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();
log.log(`[Events] ai → ${eventsUrl} (${outbox.pending()} pending)`);
log.log(`[Events] ai → ${eventsUrl} (${outbox.pending()} pending)`);
return outbox;
return outbox;
}
}
@@ -92,6 +93,17 @@ function status() {
@@ -92,6 +93,17 @@ 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 (start().close(), after runs.drain()): the relay stops and nothing more is queued; resolves
+ * when the send in progress has finished. Unsent rows stay in the outbox for the next start.
+ */
+async function stop() {
+ if (pruneTimer) clearInterval(pruneTimer);
+ pruneTimer = null;
+ const o = outbox;
+ outbox = null;
+ if (o) await o.stop();
+}
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, runChanged, payloadOf, status, _reset };
+module.exports = { init, runChanged, payloadOf, status, stop, _reset };
@@ -79,6 +79,7 @@ async function start({ config, clock = { now: () => Date.now() }, fetchImpl = gl
@@ -79,6 +79,7 @@ async function start({ config, clock = { now: () => Date.now() }, fetchImpl = gl
await new Promise(resolve => server.close(() => resolve()));
await new Promise(resolve => server.close(() => resolve()));
}
}
await runs.drain();
await runs.drain();
+ await require('./events').stop(); // after the runs: their last ai.run.* rows are queued first
const w = pool.adapter('whisper');
const w = pool.adapter('whisper');
if (w && w.adapter.killActive) w.adapter.killActive();
if (w && w.adapter.killActive) w.adapter.killActive();
db.close();
db.close();
@@ -60,5 +60,16 @@ t.test('a failed run queues ai.run.failed with its error code; a restart announc
@@ -60,5 +60,16 @@ t.test('a failed run queues ai.run.failed with its error code; a restart announc
valid(ev);
valid(ev);
});
});
-t.test('stop', async () => { require('../server/events')._reset(); await h.stop(); });
+t.test('close() stops the ai.run.* relay (unsent rows stay in the outbox)', async () => {
+ const events = require('../server/events');
+ const pending = events.status().pending;
+ assert.ok(pending > 0, 'rows are waiting (Events is unreachable)');
+ await h.stop();
+ assert.strictEqual(events.status().enabled, false, 'the relay is stopped');
+ const Database = require('better-sqlite3');
+ const db = new Database(require('path').join(h.dir, 'ai.db'), { readonly: true });
+ assert.strictEqual(db.prepare('SELECT COUNT(*) AS n FROM event_outbox WHERE sent_at IS NULL AND rejected_at IS NULL').get().n, pending, 'and its rows wait for the next start');
+ db.close();
+ events._reset();
+});
t.run();
t.run();