openvibers/openvibe.events Commits
perf: one-pass delivery counts kept 60 s (#260)
Changes
@@@ -126,8 +126,9 @@ function load(env = process.env) {host: env.HOST || '127.0.0.1',nodeEnv,isProduction,-- // How long the delivery counts by status are kept (server/store.js deliveryCounts): 15 s in production.-- countsCacheMs: Math.max(0, Number(env.EVENTS_COUNTS_CACHE_MS || (isProduction ? 15000 : 0)) || 0),++ // How long the delivery counts by status are kept (server/store.js deliveryCounts): 60 s in production, so a++ // 30 s scrape reuses the previous answer instead of re-walking 1.35M deliveries (2026-11-08).++ countsCacheMs: Math.max(0, Number(env.EVENTS_COUNTS_CACHE_MS || (isProduction ? 60000 : 0)) || 0),baseUrl,// The product origin is the API origin (plan T7): the home page and its robots.txt/sitemap.xml/// llms.txt are served from baseUrl. EVENTS_SITE_URL overrides it for a deployment that serves them apart.@@@ -719,19 +719,24 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usagereturn rows.length;});-- // The counts by status walk every delivery (a million rows: 42 ms) and were asked for every few seconds (the two-- // metric gauges per scrape, the readiness check and its details, the home and the status API: 414k calls in 35 days,-- // 289 minutes of database time). Callers at the same moment share one query, and in production its answer is kept-- // countsCacheMs (15 s; config.countsCacheMs). Elsewhere every call counts, so a test reads its own writes.++ // The counts by status walk every delivery (1.35M rows on 2026-11-08: 350 ms, 482k heap fetches on the index-only++ // scan) and were asked for every few seconds (the two metric gauges per scrape, the readiness check and its details,++ // the home and the status API: 414k calls in 35 days, 289 minutes of database time). One heap pass computes all four++ // with FILTER (the index-only scan was slower; a status index has nothing to cover), callers at the same moment++ // share one query, and in production its answer is kept countsCacheMs (60 s; config.countsCacheMs). Elsewhere every++ // call counts, so a test reads its own writes.const COUNTS_MS = Math.max(0, Number(countsCacheMs) || 0);let counted = null; // { at, value } or { at, pending }async function countDeliveries() {-- const out = { pending: 0, delivered: 0, failed: 0, dead: 0 };-- for (const r of await db.prepare('SELECT status, COUNT(*) AS n FROM deliveries GROUP BY status').all()) out[r.status] = r.n;-- return out;++ const r = await db.prepare(`SELECT count(*) FILTER (WHERE status = 'pending') AS pending,++ count(*) FILTER (WHERE status = 'delivered') AS delivered,++ count(*) FILTER (WHERE status = 'failed') AS failed,++ count(*) FILTER (WHERE status = 'dead') AS dead++ FROM deliveries`).get();++ return { pending: Number(r.pending), delivered: Number(r.delivered), failed: Number(r.failed), dead: Number(r.dead) };}async function deliveryCounts({ fresh = false } = {}) {-- const t = Date.now();++ const t = clock.now();if (!fresh && counted && t - counted.at < COUNTS_MS) return counted.pending ? counted.pending : { ...counted.value };const pending = countDeliveries();counted = { at: t, pending };@@@ -1,7 +1,7 @@'use strict';// The delivery counts by status (store.deliveryCounts) walk every delivery, and the metrics, readiness, home and status// API asked for them every few seconds (414k calls in 35 days on production, 289 minutes of database time). Callers at--// the same moment share one query; with countsCacheMs (production: 15 s) its answer is kept; fresh: true counts again;++// the same moment share one query; with countsCacheMs (production: 60 s) its answer is kept; fresh: true counts again;// without it every call counts, so a test reads its own writes.const assert = require('assert');const { boot, suite } = require('./helpers');@@@ -19,15 +19,16 @@ t('one query for callers at the same moment, kept for countsCacheMs, fresh on deif (key === 'prepare') {return (sql) => {const st = target.prepare(sql);-- if (!/GROUP BY status/.test(sql)) return st;-- return { ...st, all: async (...a) => { queries += 1; return st.all(...a); } };++ if (!/FILTER \(WHERE status/.test(sql)) return st;++ const counted = (read) => async (...a) => { queries += 1; return read(...a); };++ return { ...st, all: counted(st.all), get: counted(st.get) };};}const v = target[key];return typeof v === 'function' ? v.bind(target) : v;},});-- const kept = createStore(db, { clock: h.clock, countsCacheMs: 15_000 });++ const kept = createStore(db, { clock: h.clock, countsCacheMs: 60_000 });const [a, b, c] = await Promise.all([kept.deliveryCounts(), kept.deliveryCounts(), kept.deliveryCounts()]);assert.strictEqual(queries, 1, 'three callers at once share one query');assert.deepStrictEqual(a, b);@@@ -37,10 +38,16 @@ t('one query for callers at the same moment, kept for countsCacheMs, fresh on deassert.strictEqual(queries, 1, 'within the window: kept');await kept.deliveryCounts({ fresh: true });assert.strictEqual(queries, 2, 'fresh: counted again');++ h.clock.advance(59_000);++ assert.strictEqual((await kept.deliveryCounts()).pending, 0, 'still inside the 60 s window');++ assert.strictEqual(queries, 2, 'within the 60 s window: kept');++ h.clock.advance(1_000);++ await kept.deliveryCounts();++ assert.strictEqual(queries, 3, 'past the 60 s window: counted again');const every = createStore(db, { clock: h.clock });await every.deliveryCounts();await every.deliveryCounts();-- assert.strictEqual(queries, 4, 'without countsCacheMs every call counts');++ assert.strictEqual(queries, 5, 'without countsCacheMs every call counts');assert.deepStrictEqual(Object.keys(await every.deliveryCounts()).sort(), ['dead', 'delivered', 'failed', 'pending']);});@@@ -49,6 +56,7 @@ t('the setting: off outside production unless EVENTS_COUNTS_CACHE_MS is set', asconst base = { DATABASE_URL: 'postgres://x/y', OV_OAUTH_CLIENT_SECRET: 'x', EVENTS_SECRETS: 'x'.repeat(32) };assert.strictEqual(loadConfig({ ...base, NODE_ENV: 'test' }).countsCacheMs, 0);assert.strictEqual(loadConfig({ ...base, NODE_ENV: 'test', EVENTS_COUNTS_CACHE_MS: '5000' }).countsCacheMs, 5000);++ assert.strictEqual(loadConfig({ ...base, NODE_ENV: 'production' }).countsCacheMs, 60_000, 'production keeps the counts 60 s');});t('close', async () => { await h.close(); });
@@ -126,8 +126,9 @@ function load(env = process.env) {
@@ -126,8 +126,9 @@ function load(env = process.env) {
host: env.HOST || '127.0.0.1',
host: env.HOST || '127.0.0.1',
nodeEnv,
nodeEnv,
isProduction,
isProduction,
- // How long the delivery counts by status are kept (server/store.js deliveryCounts): 15 s in production.
- countsCacheMs: Math.max(0, Number(env.EVENTS_COUNTS_CACHE_MS || (isProduction ? 15000 : 0)) || 0),
+ // How long the delivery counts by status are kept (server/store.js deliveryCounts): 60 s in production, so a
+ // 30 s scrape reuses the previous answer instead of re-walking 1.35M deliveries (2026-11-08).
+ countsCacheMs: Math.max(0, Number(env.EVENTS_COUNTS_CACHE_MS || (isProduction ? 60000 : 0)) || 0),
baseUrl,
baseUrl,
// The product origin is the API origin (plan T7): the home page and its robots.txt/sitemap.xml/
// The product origin is the API origin (plan T7): the home page and its robots.txt/sitemap.xml/
// llms.txt are served from baseUrl. EVENTS_SITE_URL overrides it for a deployment that serves them apart.
// llms.txt are served from baseUrl. EVENTS_SITE_URL overrides it for a deployment that serves them apart.
@@ -719,19 +719,24 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
@@ -719,19 +719,24 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
return rows.length;
return rows.length;
});
});
- // The counts by status walk every delivery (a million rows: 42 ms) and were asked for every few seconds (the two
- // metric gauges per scrape, the readiness check and its details, the home and the status API: 414k calls in 35 days,
- // 289 minutes of database time). Callers at the same moment share one query, and in production its answer is kept
- // countsCacheMs (15 s; config.countsCacheMs). Elsewhere every call counts, so a test reads its own writes.
+ // The counts by status walk every delivery (1.35M rows on 2026-11-08: 350 ms, 482k heap fetches on the index-only
+ // scan) and were asked for every few seconds (the two metric gauges per scrape, the readiness check and its details,
+ // the home and the status API: 414k calls in 35 days, 289 minutes of database time). One heap pass computes all four
+ // with FILTER (the index-only scan was slower; a status index has nothing to cover), callers at the same moment
+ // share one query, and in production its answer is kept countsCacheMs (60 s; config.countsCacheMs). Elsewhere every
+ // call counts, so a test reads its own writes.
const COUNTS_MS = Math.max(0, Number(countsCacheMs) || 0);
const COUNTS_MS = Math.max(0, Number(countsCacheMs) || 0);
let counted = null; // { at, value } or { at, pending }
let counted = null; // { at, value } or { at, pending }
async function countDeliveries() {
async function countDeliveries() {
- const out = { pending: 0, delivered: 0, failed: 0, dead: 0 };
- for (const r of await db.prepare('SELECT status, COUNT(*) AS n FROM deliveries GROUP BY status').all()) out[r.status] = r.n;
- return out;
+ const r = await db.prepare(`SELECT count(*) FILTER (WHERE status = 'pending') AS pending,
+ count(*) FILTER (WHERE status = 'delivered') AS delivered,
+ count(*) FILTER (WHERE status = 'failed') AS failed,
+ count(*) FILTER (WHERE status = 'dead') AS dead
+ FROM deliveries`).get();
+ return { pending: Number(r.pending), delivered: Number(r.delivered), failed: Number(r.failed), dead: Number(r.dead) };
}
}
async function deliveryCounts({ fresh = false } = {}) {
async function deliveryCounts({ fresh = false } = {}) {
- const t = Date.now();
+ const t = clock.now();
if (!fresh && counted && t - counted.at < COUNTS_MS) return counted.pending ? counted.pending : { ...counted.value };
if (!fresh && counted && t - counted.at < COUNTS_MS) return counted.pending ? counted.pending : { ...counted.value };
const pending = countDeliveries();
const pending = countDeliveries();
counted = { at: t, pending };
counted = { at: t, pending };
@@ -1,7 +1,7 @@
@@ -1,7 +1,7 @@
'use strict';
'use strict';
// The delivery counts by status (store.deliveryCounts) walk every delivery, and the metrics, readiness, home and status
// The delivery counts by status (store.deliveryCounts) walk every delivery, and the metrics, readiness, home and status
// API asked for them every few seconds (414k calls in 35 days on production, 289 minutes of database time). Callers at
// API asked for them every few seconds (414k calls in 35 days on production, 289 minutes of database time). Callers at
-// the same moment share one query; with countsCacheMs (production: 15 s) its answer is kept; fresh: true counts again;
+// the same moment share one query; with countsCacheMs (production: 60 s) its answer is kept; fresh: true counts again;
// without it every call counts, so a test reads its own writes.
// without it every call counts, so a test reads its own writes.
const assert = require('assert');
const assert = require('assert');
const { boot, suite } = require('./helpers');
const { boot, suite } = require('./helpers');
@@ -19,15 +19,16 @@ t('one query for callers at the same moment, kept for countsCacheMs, fresh on de
@@ -19,15 +19,16 @@ t('one query for callers at the same moment, kept for countsCacheMs, fresh on de
if (key === 'prepare') {
if (key === 'prepare') {
return (sql) => {
return (sql) => {
const st = target.prepare(sql);
const st = target.prepare(sql);
- if (!/GROUP BY status/.test(sql)) return st;
- return { ...st, all: async (...a) => { queries += 1; return st.all(...a); } };
+ if (!/FILTER \(WHERE status/.test(sql)) return st;
+ const counted = (read) => async (...a) => { queries += 1; return read(...a); };
+ return { ...st, all: counted(st.all), get: counted(st.get) };
};
};
}
}
const v = target[key];
const v = target[key];
return typeof v === 'function' ? v.bind(target) : v;
return typeof v === 'function' ? v.bind(target) : v;
},
},
});
});
- const kept = createStore(db, { clock: h.clock, countsCacheMs: 15_000 });
+ const kept = createStore(db, { clock: h.clock, countsCacheMs: 60_000 });
const [a, b, c] = await Promise.all([kept.deliveryCounts(), kept.deliveryCounts(), kept.deliveryCounts()]);
const [a, b, c] = await Promise.all([kept.deliveryCounts(), kept.deliveryCounts(), kept.deliveryCounts()]);
assert.strictEqual(queries, 1, 'three callers at once share one query');
assert.strictEqual(queries, 1, 'three callers at once share one query');
assert.deepStrictEqual(a, b);
assert.deepStrictEqual(a, b);
@@ -37,10 +38,16 @@ t('one query for callers at the same moment, kept for countsCacheMs, fresh on de
@@ -37,10 +38,16 @@ t('one query for callers at the same moment, kept for countsCacheMs, fresh on de
assert.strictEqual(queries, 1, 'within the window: kept');
assert.strictEqual(queries, 1, 'within the window: kept');
await kept.deliveryCounts({ fresh: true });
await kept.deliveryCounts({ fresh: true });
assert.strictEqual(queries, 2, 'fresh: counted again');
assert.strictEqual(queries, 2, 'fresh: counted again');
+ h.clock.advance(59_000);
+ assert.strictEqual((await kept.deliveryCounts()).pending, 0, 'still inside the 60 s window');
+ assert.strictEqual(queries, 2, 'within the 60 s window: kept');
+ h.clock.advance(1_000);
+ await kept.deliveryCounts();
+ assert.strictEqual(queries, 3, 'past the 60 s window: counted again');
const every = createStore(db, { clock: h.clock });
const every = createStore(db, { clock: h.clock });
await every.deliveryCounts();
await every.deliveryCounts();
await every.deliveryCounts();
await every.deliveryCounts();
- assert.strictEqual(queries, 4, 'without countsCacheMs every call counts');
+ assert.strictEqual(queries, 5, 'without countsCacheMs every call counts');
assert.deepStrictEqual(Object.keys(await every.deliveryCounts()).sort(), ['dead', 'delivered', 'failed', 'pending']);
assert.deepStrictEqual(Object.keys(await every.deliveryCounts()).sort(), ['dead', 'delivered', 'failed', 'pending']);
});
});
@@ -49,6 +56,7 @@ t('the setting: off outside production unless EVENTS_COUNTS_CACHE_MS is set', as
@@ -49,6 +56,7 @@ t('the setting: off outside production unless EVENTS_COUNTS_CACHE_MS is set', as
const base = { DATABASE_URL: 'postgres://x/y', OV_OAUTH_CLIENT_SECRET: 'x', EVENTS_SECRETS: 'x'.repeat(32) };
const base = { DATABASE_URL: 'postgres://x/y', OV_OAUTH_CLIENT_SECRET: 'x', EVENTS_SECRETS: 'x'.repeat(32) };
assert.strictEqual(loadConfig({ ...base, NODE_ENV: 'test' }).countsCacheMs, 0);
assert.strictEqual(loadConfig({ ...base, NODE_ENV: 'test' }).countsCacheMs, 0);
assert.strictEqual(loadConfig({ ...base, NODE_ENV: 'test', EVENTS_COUNTS_CACHE_MS: '5000' }).countsCacheMs, 5000);
assert.strictEqual(loadConfig({ ...base, NODE_ENV: 'test', EVENTS_COUNTS_CACHE_MS: '5000' }).countsCacheMs, 5000);
+ assert.strictEqual(loadConfig({ ...base, NODE_ENV: 'production' }).countsCacheMs, 60_000, 'production keeps the counts 60 s');
});
});
t('close', async () => { await h.close(); });
t('close', async () => { await h.close(); });