openvibers/openvibe.sources Commits
perf: rejected outbox index and 120 s item counts (#233)
Changes
nnew file mode 100644@@@ -0,0 +1,7 @@++-- phase: expand++-- no-transaction++-- openvibe-sdk's outbox.status().rejected reads the whole outbox for its count++-- (SELECT count(*) FROM service_outbox WHERE rejected_at IS NOT NULL): 183 ms mean on 2026-11-08.++-- A partial index holds only the rejected rows, so the same count is an index-only scan++-- over an index that is usually empty.++CREATE INDEX CONCURRENTLY IF NOT EXISTS service_outbox_rejected ON service_outbox (rejected_at) WHERE rejected_at IS NOT NULL;@@@ -22,10 +22,11 @@ const STATUSES = ['healthy', 'stale', 'failing', 'never_fetched', 'disabled', 'm/*** A read kept for keepMs and shared by callers at the same moment. The item count walks every item (47 ms, asked on each-- * scrape and readiness probe: 151k calls and 118 minutes of database time from 09-28 to 11-02). 30 s in production,-- * every call elsewhere (tests read their own writes).++ * scrape and readiness probe: 151k calls and 118 minutes of database time from 09-28 to 11-02). A 30 s keep was shorter++ * than the ~38 s between scrapes, so on 2026-11-08 it still ran 2,111 times (213 ms mean, 463 ms cold): 120 s in++ * production, every call elsewhere (tests read their own writes).*/--const COUNTS_MS = process.env.NODE_ENV === 'production' ? 30_000 : 0;++const COUNTS_MS = process.env.NODE_ENV === 'production' ? 120_000 : 0;function kept(read, keepMs = COUNTS_MS, now = () => Date.now()) {if (!(keepMs > 0)) return read;let hold = null;@@@ -13,16 +13,16 @@ const { kept } = require('../server/observability');let reads = 0;let fail = false;const read = async () => { reads += 1; if (fail) throw new Error('db away'); return { current: reads, removed: 0 }; };-- const k = kept(read, 30_000, () => t);++ const k = kept(read, 120_000, () => t);const [a, b] = await Promise.all([k(), k()]);assert.strictEqual(reads, 1, 'two callers at once: one read');assert.deepStrictEqual(a, b);-- t = 29_000;-- assert.strictEqual((await k()).current, 1, 'within the window: kept');-- t = 30_000;++ t = 119_000;++ assert.strictEqual((await k()).current, 1, 'within the 120 s window: kept');++ t = 120_000;assert.strictEqual((await k()).current, 2, 'past it: read again');fail = true;-- t = 61_000;++ t = 241_000;await assert.rejects(k(), /db away/);fail = false;assert.strictEqual((await k()).current, 4, 'a failed read is not kept: the next call reads');nnew file mode 100644@@@ -0,0 +1,34 @@++'use strict';++/**++ * The rejected-events count on /api/ready and the metrics gauges (openvibe-sdk outbox.status()) walks the++ * whole outbox without an index: 183 ms mean on the 2026-11-08 production read. The new migration adds a++ * partial index over the rejected rows only; this proves the count can use it and still answers the same.++ */++const assert = require('assert');++const { testDb } = require('./db');++++(async () => {++ const { db } = await testDb();++ try {++ const indexes = await db.prepare(`SELECT indexname FROM pg_indexes++ WHERE schemaname = current_schema() AND tablename = 'service_outbox'`).all();++ assert.ok(indexes.some((r) => r.indexname === 'service_outbox_rejected'), 'the migration created the partial index');++++ const now = Date.now();++ await db.prepare(`INSERT INTO service_outbox (event_id, envelope, created_at, next_attempt_at) VALUES++ ('evt_sent', '{}', ?, ?), ('evt_rejected', '{}', ?, ?)`).run(now, now, now, now);++ await db.prepare('UPDATE service_outbox SET sent_at = ? WHERE event_id = ?').run(now, 'evt_sent');++ await db.prepare('UPDATE service_outbox SET rejected_at = ? WHERE event_id = ?').run(now, 'evt_rejected');++++ const count = await db.prepare('SELECT count(*) AS n FROM service_outbox WHERE rejected_at IS NOT NULL').get();++ assert.strictEqual(Number(count.n), 1, 'the count still sees the one rejected event');++++ await db.query('SET enable_seqscan = off');++ const plan = (await db.prepare('EXPLAIN SELECT count(*) FROM service_outbox WHERE rejected_at IS NOT NULL').all())++ .map((r) => r['QUERY PLAN']).join('\n');++ assert.match(plan, /service_outbox_rejected/, plan);++ } finally {++ await db.close();++ }++ console.log('ok - the rejected outbox count is an index-only scan over service_outbox_rejected');++})().catch((err) => { console.error(err); process.exit(1); });
new file mode 100644
new file mode 100644
@@ -0,0 +1,7 @@
@@ -0,0 +1,7 @@
+-- phase: expand
+-- no-transaction
+-- openvibe-sdk's outbox.status().rejected reads the whole outbox for its count
+-- (SELECT count(*) FROM service_outbox WHERE rejected_at IS NOT NULL): 183 ms mean on 2026-11-08.
+-- A partial index holds only the rejected rows, so the same count is an index-only scan
+-- over an index that is usually empty.
+CREATE INDEX CONCURRENTLY IF NOT EXISTS service_outbox_rejected ON service_outbox (rejected_at) WHERE rejected_at IS NOT NULL;
@@ -22,10 +22,11 @@ const STATUSES = ['healthy', 'stale', 'failing', 'never_fetched', 'disabled', 'm
@@ -22,10 +22,11 @@ const STATUSES = ['healthy', 'stale', 'failing', 'never_fetched', 'disabled', 'm
/**
/**
* A read kept for keepMs and shared by callers at the same moment. The item count walks every item (47 ms, asked on each
* A read kept for keepMs and shared by callers at the same moment. The item count walks every item (47 ms, asked on each
- * scrape and readiness probe: 151k calls and 118 minutes of database time from 09-28 to 11-02). 30 s in production,
- * every call elsewhere (tests read their own writes).
+ * scrape and readiness probe: 151k calls and 118 minutes of database time from 09-28 to 11-02). A 30 s keep was shorter
+ * than the ~38 s between scrapes, so on 2026-11-08 it still ran 2,111 times (213 ms mean, 463 ms cold): 120 s in
+ * production, every call elsewhere (tests read their own writes).
*/
*/
-const COUNTS_MS = process.env.NODE_ENV === 'production' ? 30_000 : 0;
+const COUNTS_MS = process.env.NODE_ENV === 'production' ? 120_000 : 0;
function kept(read, keepMs = COUNTS_MS, now = () => Date.now()) {
function kept(read, keepMs = COUNTS_MS, now = () => Date.now()) {
if (!(keepMs > 0)) return read;
if (!(keepMs > 0)) return read;
let hold = null;
let hold = null;
@@ -13,16 +13,16 @@ const { kept } = require('../server/observability');
@@ -13,16 +13,16 @@ const { kept } = require('../server/observability');
let reads = 0;
let reads = 0;
let fail = false;
let fail = false;
const read = async () => { reads += 1; if (fail) throw new Error('db away'); return { current: reads, removed: 0 }; };
const read = async () => { reads += 1; if (fail) throw new Error('db away'); return { current: reads, removed: 0 }; };
- const k = kept(read, 30_000, () => t);
+ const k = kept(read, 120_000, () => t);
const [a, b] = await Promise.all([k(), k()]);
const [a, b] = await Promise.all([k(), k()]);
assert.strictEqual(reads, 1, 'two callers at once: one read');
assert.strictEqual(reads, 1, 'two callers at once: one read');
assert.deepStrictEqual(a, b);
assert.deepStrictEqual(a, b);
- t = 29_000;
- assert.strictEqual((await k()).current, 1, 'within the window: kept');
- t = 30_000;
+ t = 119_000;
+ assert.strictEqual((await k()).current, 1, 'within the 120 s window: kept');
+ t = 120_000;
assert.strictEqual((await k()).current, 2, 'past it: read again');
assert.strictEqual((await k()).current, 2, 'past it: read again');
fail = true;
fail = true;
- t = 61_000;
+ t = 241_000;
await assert.rejects(k(), /db away/);
await assert.rejects(k(), /db away/);
fail = false;
fail = false;
assert.strictEqual((await k()).current, 4, 'a failed read is not kept: the next call reads');
assert.strictEqual((await k()).current, 4, 'a failed read is not kept: the next call reads');
new file mode 100644
new file mode 100644
@@ -0,0 +1,34 @@
@@ -0,0 +1,34 @@
+'use strict';
+/**
+ * The rejected-events count on /api/ready and the metrics gauges (openvibe-sdk outbox.status()) walks the
+ * whole outbox without an index: 183 ms mean on the 2026-11-08 production read. The new migration adds a
+ * partial index over the rejected rows only; this proves the count can use it and still answers the same.
+ */
+const assert = require('assert');
+const { testDb } = require('./db');
+
+(async () => {
+ const { db } = await testDb();
+ try {
+ const indexes = await db.prepare(`SELECT indexname FROM pg_indexes
+ WHERE schemaname = current_schema() AND tablename = 'service_outbox'`).all();
+ assert.ok(indexes.some((r) => r.indexname === 'service_outbox_rejected'), 'the migration created the partial index');
+
+ const now = Date.now();
+ await db.prepare(`INSERT INTO service_outbox (event_id, envelope, created_at, next_attempt_at) VALUES
+ ('evt_sent', '{}', ?, ?), ('evt_rejected', '{}', ?, ?)`).run(now, now, now, now);
+ await db.prepare('UPDATE service_outbox SET sent_at = ? WHERE event_id = ?').run(now, 'evt_sent');
+ await db.prepare('UPDATE service_outbox SET rejected_at = ? WHERE event_id = ?').run(now, 'evt_rejected');
+
+ const count = await db.prepare('SELECT count(*) AS n FROM service_outbox WHERE rejected_at IS NOT NULL').get();
+ assert.strictEqual(Number(count.n), 1, 'the count still sees the one rejected event');
+
+ await db.query('SET enable_seqscan = off');
+ const plan = (await db.prepare('EXPLAIN SELECT count(*) FROM service_outbox WHERE rejected_at IS NOT NULL').all())
+ .map((r) => r['QUERY PLAN']).join('\n');
+ assert.match(plan, /service_outbox_rejected/, plan);
+ } finally {
+ await db.close();
+ }
+ console.log('ok - the rejected outbox count is an index-only scan over service_outbox_rejected');
+})().catch((err) => { console.error(err); process.exit(1); });