openvibers/openvibe.reviews Commits
Boot listens first: the Search reconcile runs after listen, one short transaction per entity (6,800 entities took 30 s before listen and the deploy of #84 was rolled back past ovhost's readiness wait) (#85)
Changes
@@@ -30,7 +30,7 @@ const { createPgTelemetrySink, startTelemetryRetention } = require('openvibe-sdk* its own database (start's `givenDb`), which stays open. `exit` and `signals` are injectable so a test can watch* the stop and no test process installs signal handlers.*/--function createLifecycle({ server, db, valkey = null, platform, timers = [], telemetry = null, retention = null, closeDb = true, exit, signals, log, subscriptions = null } = {}) {++function createLifecycle({ server, db, valkey = null, platform, timers = [], telemetry = null, retention = null, closeDb = true, exit, signals, log, subscriptions = null, background = null } = {}) {return gracefulStop({name: 'Reviews', server, log, drainMs: 8000, deadlineMs: 10000, deadlineExitCode: 0, exit, signals,stop: [@@@ -40,6 +40,7 @@ function createLifecycle({ server, db, valkey = null, platform, timers = [], tel],close: [async () => { if (retention) await retention.stop(); if (telemetry) await telemetry.stop(); },++ async () => { if (background) await background(); },() => { if (closeDb) return db.close(); },() => { if (valkey) return valkey.close(); },],@@@ -76,8 +77,15 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch: null);const app = createApp({ config, telemetry, svc, viewers, platform, sources, sync, keys, db, log, rateLimits, fetchImpl, limitsNow, indexnow, accountData, accountSend });-- // Search holds whatever the current rules say (idempotent: unchanged documents are not re-sent).-- try { await svc.reconcileIndex(); } catch (err) { log.warn(`[Reviews] index reconcile: ${err.message}`); }++ // Search holds whatever the current rules say (idempotent: unchanged documents are not re-sent). A running service++ // does it after it listens: 6,800 entities took 30 s on 2026-10-18, past ovhost's readiness wait, and the deploy++ // was rolled back. Without workers (tests, one-off scripts) it is done before start() returns.++ let stopping = false;++ let reconciling = null;++ const reconcile = () => svc.reconcileIndex({ stopped: () => stopping })++ .then((r) => { if (r.sent) log.log(`[Reviews] index reconcile: ${r.sent} document(s) re-sent`); })++ .catch((err) => log.warn(`[Reviews] index reconcile: ${err.message}`));++ if (!workers) await reconcile();const timers = [];if (workers) {@@@ -99,6 +107,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetchserver.keepAliveTimeout = 65000;log.log(`[Reviews] ${config.nodeEnv} on http://${config.host}:${config.port} → ${config.baseUrl}`);}++ if (workers) reconciling = reconcile();// The two account subscriptions at OpenVibe.Events, created when missing; off without workers, EVENTS_URL,// REVIEWS_EVENTS_SECRET or the client secret.let subscriptions = null;@@@ -109,7 +118,8 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch});}-- const lifecycle = createLifecycle({ server, db, telemetry, retention, platform, timers, closeDb: !givenDb, signals, exit, log, subscriptions });++ const lifecycle = createLifecycle({ server, db, telemetry, retention, platform, timers, closeDb: !givenDb, signals, exit, log, subscriptions,++ background: async () => { stopping = true; if (reconciling) await reconciling; } });async function stop() {return lifecycle.stop('stop');@@@ -2093,17 +2093,20 @@ function createReviewsService({ db, stores, outbox, config, now = () => Date.now// Search reconciliation ---------------------------------------------------------------/** Stamp + enqueue the entity's current document on the caller's handle (used by tests/ops). */syncIndex: syncEntity,-- async reconcileIndex() {-- return await tx(async () => {-- let sent = 0;-- for (const { id } of await q.allEntityIds.all()) {-- if (await sequencer.current('reviews', 'entity', id) == null && (await q.entity.get(id)).state !== 'active') continue;++ /** Re-send every entity's document that differs from what Search holds; one short transaction per entity, so a++ * running service is never held up behind it. `stopped()` ends it early (a shutdown). */++ async reconcileIndex({ stopped = () => false } = {}) {++ let sent = 0;++ for (const { id } of await q.allEntityIds.all()) {++ if (stopped()) break;++ sent += await tx(async () => {const before = await sequencer.current('reviews', 'entity', id);++ if (before == null && (await q.entity.get(id)).state !== 'active') return 0;const doc = await syncEntity(id);-- if (doc && doc.revision !== before) sent++;-- }-- return { sent };-- });++ return doc && doc.revision !== before ? 1 : 0;++ });++ }++ return { sent };},async stats() {@@@ -117,3 +117,22 @@ t("the outbox row and the state change share the caller's transaction (a rollbacassert.strictEqual((await h.db.prepare('SELECT name FROM review_entities WHERE id = ?').get(e.id)).name, row.name, 'the state change rolled back with it');} finally { await h.stop(); }});++++t('the index reconcile re-sends only what changed, entity by entity, stops when asked, and runs after the server listens', async () => {++ const h = await boot();++ try {++ const e = await indexableEntity(h);++ assert.deepStrictEqual(await h.svc.reconcileIndex(), { sent: 0 }, 'nothing changed since it was sent');++ // Something the rules now say differently (a rename made without the service): the reconcile catches it.++ await h.db.prepare('UPDATE review_entities SET name = ? WHERE id = ?').run('Portal 2 (renamed)', e.id);++ assert.deepStrictEqual(await h.svc.reconcileIndex({ stopped: () => true }), { sent: 0 }, 'a shutdown ends it before the next entity');++ assert.deepStrictEqual(await h.svc.reconcileIndex(), { sent: 1 });++ const last = (await outbox(h)).filter((env) => env.event_type === 'reviews.index_document.upserted').pop();++ assert.match(JSON.stringify(last.payload), /Portal 2 \(renamed\)/);++ assert.deepStrictEqual(await h.svc.reconcileIndex(), { sent: 0 }, 'idempotent');++ } finally { await h.stop(); }++ // A running service listens first (6,800 entities took 30 s at boot on 2026-10-18, past the deploy's readiness wait).++ const src = fs.readFileSync(path.join(__dirname, '..', 'server', 'index.js'), 'utf8');++ assert.ok(src.indexOf('if (workers) reconciling = reconcile();') > src.indexOf('app.listen('), 'with workers, the reconcile starts after listen');++ assert.match(src, /background: async \(\) => \{ stopping = true; if \(reconciling\) await reconciling; \}/, 'and a shutdown waits for it before the database closes');++});
@@ -30,7 +30,7 @@ const { createPgTelemetrySink, startTelemetryRetention } = require('openvibe-sdk
@@ -30,7 +30,7 @@ const { createPgTelemetrySink, startTelemetryRetention } = require('openvibe-sdk
* its own database (start's `givenDb`), which stays open. `exit` and `signals` are injectable so a test can watch
* its own database (start's `givenDb`), which stays open. `exit` and `signals` are injectable so a test can watch
* the stop and no test process installs signal handlers.
* the stop and no test process installs signal handlers.
*/
*/
-function createLifecycle({ server, db, valkey = null, platform, timers = [], telemetry = null, retention = null, closeDb = true, exit, signals, log, subscriptions = null } = {}) {
+function createLifecycle({ server, db, valkey = null, platform, timers = [], telemetry = null, retention = null, closeDb = true, exit, signals, log, subscriptions = null, background = null } = {}) {
return gracefulStop({
return gracefulStop({
name: 'Reviews', server, log, drainMs: 8000, deadlineMs: 10000, deadlineExitCode: 0, exit, signals,
name: 'Reviews', server, log, drainMs: 8000, deadlineMs: 10000, deadlineExitCode: 0, exit, signals,
stop: [
stop: [
@@ -40,6 +40,7 @@ function createLifecycle({ server, db, valkey = null, platform, timers = [], tel
@@ -40,6 +40,7 @@ function createLifecycle({ server, db, valkey = null, platform, timers = [], tel
],
],
close: [
close: [
async () => { if (retention) await retention.stop(); if (telemetry) await telemetry.stop(); },
async () => { if (retention) await retention.stop(); if (telemetry) await telemetry.stop(); },
+ async () => { if (background) await background(); },
() => { if (closeDb) return db.close(); },
() => { if (closeDb) return db.close(); },
() => { if (valkey) return valkey.close(); },
() => { if (valkey) return valkey.close(); },
],
],
@@ -76,8 +77,15 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
@@ -76,8 +77,15 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
: null);
: null);
const app = createApp({ config, telemetry, svc, viewers, platform, sources, sync, keys, db, log, rateLimits, fetchImpl, limitsNow, indexnow, accountData, accountSend });
const app = createApp({ config, telemetry, svc, viewers, platform, sources, sync, keys, db, log, rateLimits, fetchImpl, limitsNow, indexnow, accountData, accountSend });
- // Search holds whatever the current rules say (idempotent: unchanged documents are not re-sent).
- try { await svc.reconcileIndex(); } catch (err) { log.warn(`[Reviews] index reconcile: ${err.message}`); }
+ // Search holds whatever the current rules say (idempotent: unchanged documents are not re-sent). A running service
+ // does it after it listens: 6,800 entities took 30 s on 2026-10-18, past ovhost's readiness wait, and the deploy
+ // was rolled back. Without workers (tests, one-off scripts) it is done before start() returns.
+ let stopping = false;
+ let reconciling = null;
+ const reconcile = () => svc.reconcileIndex({ stopped: () => stopping })
+ .then((r) => { if (r.sent) log.log(`[Reviews] index reconcile: ${r.sent} document(s) re-sent`); })
+ .catch((err) => log.warn(`[Reviews] index reconcile: ${err.message}`));
+ if (!workers) await reconcile();
const timers = [];
const timers = [];
if (workers) {
if (workers) {
@@ -99,6 +107,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
@@ -99,6 +107,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
server.keepAliveTimeout = 65000;
server.keepAliveTimeout = 65000;
log.log(`[Reviews] ${config.nodeEnv} on http://${config.host}:${config.port} → ${config.baseUrl}`);
log.log(`[Reviews] ${config.nodeEnv} on http://${config.host}:${config.port} → ${config.baseUrl}`);
}
}
+ if (workers) reconciling = reconcile();
// The two account subscriptions at OpenVibe.Events, created when missing; off without workers, EVENTS_URL,
// The two account subscriptions at OpenVibe.Events, created when missing; off without workers, EVENTS_URL,
// REVIEWS_EVENTS_SECRET or the client secret.
// REVIEWS_EVENTS_SECRET or the client secret.
let subscriptions = null;
let subscriptions = null;
@@ -109,7 +118,8 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
@@ -109,7 +118,8 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
});
});
}
}
- const lifecycle = createLifecycle({ server, db, telemetry, retention, platform, timers, closeDb: !givenDb, signals, exit, log, subscriptions });
+ const lifecycle = createLifecycle({ server, db, telemetry, retention, platform, timers, closeDb: !givenDb, signals, exit, log, subscriptions,
+ background: async () => { stopping = true; if (reconciling) await reconciling; } });
async function stop() {
async function stop() {
return lifecycle.stop('stop');
return lifecycle.stop('stop');
@@ -2093,17 +2093,20 @@ function createReviewsService({ db, stores, outbox, config, now = () => Date.now
@@ -2093,17 +2093,20 @@ function createReviewsService({ db, stores, outbox, config, now = () => Date.now
// Search reconciliation ---------------------------------------------------------------
// Search reconciliation ---------------------------------------------------------------
/** Stamp + enqueue the entity's current document on the caller's handle (used by tests/ops). */
/** Stamp + enqueue the entity's current document on the caller's handle (used by tests/ops). */
syncIndex: syncEntity,
syncIndex: syncEntity,
- async reconcileIndex() {
- return await tx(async () => {
- let sent = 0;
- for (const { id } of await q.allEntityIds.all()) {
- if (await sequencer.current('reviews', 'entity', id) == null && (await q.entity.get(id)).state !== 'active') continue;
+ /** Re-send every entity's document that differs from what Search holds; one short transaction per entity, so a
+ * running service is never held up behind it. `stopped()` ends it early (a shutdown). */
+ async reconcileIndex({ stopped = () => false } = {}) {
+ let sent = 0;
+ for (const { id } of await q.allEntityIds.all()) {
+ if (stopped()) break;
+ sent += await tx(async () => {
const before = await sequencer.current('reviews', 'entity', id);
const before = await sequencer.current('reviews', 'entity', id);
+ if (before == null && (await q.entity.get(id)).state !== 'active') return 0;
const doc = await syncEntity(id);
const doc = await syncEntity(id);
- if (doc && doc.revision !== before) sent++;
- }
- return { sent };
- });
+ return doc && doc.revision !== before ? 1 : 0;
+ });
+ }
+ return { sent };
},
},
async stats() {
async stats() {
@@ -117,3 +117,22 @@ t("the outbox row and the state change share the caller's transaction (a rollbac
@@ -117,3 +117,22 @@ t("the outbox row and the state change share the caller's transaction (a rollbac
assert.strictEqual((await h.db.prepare('SELECT name FROM review_entities WHERE id = ?').get(e.id)).name, row.name, 'the state change rolled back with it');
assert.strictEqual((await h.db.prepare('SELECT name FROM review_entities WHERE id = ?').get(e.id)).name, row.name, 'the state change rolled back with it');
} finally { await h.stop(); }
} finally { await h.stop(); }
});
});
+
+t('the index reconcile re-sends only what changed, entity by entity, stops when asked, and runs after the server listens', async () => {
+ const h = await boot();
+ try {
+ const e = await indexableEntity(h);
+ assert.deepStrictEqual(await h.svc.reconcileIndex(), { sent: 0 }, 'nothing changed since it was sent');
+ // Something the rules now say differently (a rename made without the service): the reconcile catches it.
+ await h.db.prepare('UPDATE review_entities SET name = ? WHERE id = ?').run('Portal 2 (renamed)', e.id);
+ assert.deepStrictEqual(await h.svc.reconcileIndex({ stopped: () => true }), { sent: 0 }, 'a shutdown ends it before the next entity');
+ assert.deepStrictEqual(await h.svc.reconcileIndex(), { sent: 1 });
+ const last = (await outbox(h)).filter((env) => env.event_type === 'reviews.index_document.upserted').pop();
+ assert.match(JSON.stringify(last.payload), /Portal 2 \(renamed\)/);
+ assert.deepStrictEqual(await h.svc.reconcileIndex(), { sent: 0 }, 'idempotent');
+ } finally { await h.stop(); }
+ // A running service listens first (6,800 entities took 30 s at boot on 2026-10-18, past the deploy's readiness wait).
+ const src = fs.readFileSync(path.join(__dirname, '..', 'server', 'index.js'), 'utf8');
+ assert.ok(src.indexOf('if (workers) reconciling = reconcile();') > src.indexOf('app.listen('), 'with workers, the reconcile starts after listen');
+ assert.match(src, /background: async \(\) => \{ stopping = true; if \(reconciling\) await reconciling; \}/, 'and a shutdown waits for it before the database closes');
+});