openvibers/openvibe.events Commits
Delivery counts: one query per moment, kept 15 s in production (#218)
Changes
@@@ -110,6 +110,8 @@ 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),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.@@@ -43,7 +43,7 @@ async function start({// carrier on every delivery at fan-out and records a (class, key) decision in delivery_placements when it changes.const carriers = createCarriers({ config, clock, log, valkey });const planner = createPlanner({ registry: carriers, db, clock, log });-- const store = createStore(db, { clock, maxHops: config.maxHops, usage: config.usage, policies, planner });++ const store = createStore(db, { clock, maxHops: config.maxHops, usage: config.usage, policies, planner, countsCacheMs: config.countsCacheMs });const keys = createNetworkKeys({ network: config.networkInternalUrl, publicKey: config.networkPublicKey, fetch: fetchImpl, log });const auth = createAuth({ config, keys, store });const metrics = createMetrics({ store });@@@ -116,7 +116,7 @@ function subscriptionView(row, { withSecret = false } = {}) {return out;}--function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage: usageConfig = { enabled: true }, policies = null, planner = null } = {}) {++function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage: usageConfig = { enabled: true }, policies = null, planner = null, countsCacheMs = 0 } = {}) {// Project usage rollups (./usage.js): counted in the transactions below, sent as events.usage.recorded.const usage = createUsage(db, { clock, enabled: usageConfig.enabled !== false });const q = {@@@ -714,11 +714,31 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usagereturn rows.length;});-- async function deliveryCounts() {++ // 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.++ 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;}++ async function deliveryCounts({ fresh = false } = {}) {++ const t = Date.now();++ if (!fresh && counted && t - counted.at < COUNTS_MS) return counted.pending ? counted.pending : { ...counted.value };++ const pending = countDeliveries();++ counted = { at: t, pending };++ try {++ const value = await pending;++ if (counted && counted.pending === pending) counted = { at: t, value };++ return { ...value };++ } catch (err) {++ if (counted && counted.pending === pending) counted = null;++ throw err;++ }++ }// ── Checkpoints ──────────────────────────────────────────nnew file mode 100644@@@ -0,0 +1,56 @@++'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;++// without it every call counts, so a test reads its own writes.++const assert = require('assert');++const { boot, suite } = require('./helpers');++const { createStore } = require('../server/store');++++const t = suite('delivery-counts');++let h;++++t('boot', async () => { h = await boot(); });++++t('one query for callers at the same moment, kept for countsCacheMs, fresh on demand', async () => {++ let queries = 0;++ const db = new Proxy(h.db, {++ get(target, key) {++ if (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); } };++ };++ }++ const v = target[key];++ return typeof v === 'function' ? v.bind(target) : v;++ },++ });++ const kept = createStore(db, { clock: h.clock, countsCacheMs: 15_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);++ assert.deepStrictEqual(b, c);++ a.dead = 999;++ assert.notStrictEqual((await kept.deliveryCounts()).dead, 999, 'each caller gets its own copy');++ assert.strictEqual(queries, 1, 'within the window: kept');++ await kept.deliveryCounts({ fresh: true });++ assert.strictEqual(queries, 2, 'fresh: 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.deepStrictEqual(Object.keys(await every.deliveryCounts()).sort(), ['dead', 'delivered', 'failed', 'pending']);++});++++t('the setting: off outside production unless EVENTS_COUNTS_CACHE_MS is set', async () => {++ const { load: loadConfig } = require('../server/config');++ 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', EVENTS_COUNTS_CACHE_MS: '5000' }).countsCacheMs, 5000);++});++++t('close', async () => { await h.close(); });++++t.run();@@@ -275,7 +275,7 @@ t('backpressure: one in flight per subscription, <= maxInflight (20) overall', aawait bp.worker.drain();assert.strictEqual(peak, 20, `peak in flight ${peak}`);assert.strictEqual(perSubPeak, 1);-- assert.strictEqual((await bp.store.deliveryCounts()).delivered, 75);++ assert.strictEqual((await bp.store.deliveryCounts({ fresh: true })).delivered, 75);await bp.stop();});@@@ -232,7 +232,7 @@ t('max_inflight = 2: two keys run together, never two of one key, never three atassert.strictEqual(totalPeak, 2, 'two keys of the subscription went in flight together');assert.strictEqual(keyPeak, 1, 'never two of one key at once');-- assert.strictEqual((await h.store.deliveryCounts()).pending, 0, 'every delivery was sent');++ assert.strictEqual((await h.store.deliveryCounts({ fresh: true })).pending, 0, 'every delivery was sent');await h.store.db.prepare('UPDATE subscriptions SET max_inflight = 1 WHERE id = ?').run(sub.id);await deletePolicy('live.ord4.*');
@@ -110,6 +110,8 @@ function load(env = process.env) {
@@ -110,6 +110,8 @@ 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),
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.
@@ -43,7 +43,7 @@ async function start({
@@ -43,7 +43,7 @@ async function start({
// carrier on every delivery at fan-out and records a (class, key) decision in delivery_placements when it changes.
// carrier on every delivery at fan-out and records a (class, key) decision in delivery_placements when it changes.
const carriers = createCarriers({ config, clock, log, valkey });
const carriers = createCarriers({ config, clock, log, valkey });
const planner = createPlanner({ registry: carriers, db, clock, log });
const planner = createPlanner({ registry: carriers, db, clock, log });
- const store = createStore(db, { clock, maxHops: config.maxHops, usage: config.usage, policies, planner });
+ const store = createStore(db, { clock, maxHops: config.maxHops, usage: config.usage, policies, planner, countsCacheMs: config.countsCacheMs });
const keys = createNetworkKeys({ network: config.networkInternalUrl, publicKey: config.networkPublicKey, fetch: fetchImpl, log });
const keys = createNetworkKeys({ network: config.networkInternalUrl, publicKey: config.networkPublicKey, fetch: fetchImpl, log });
const auth = createAuth({ config, keys, store });
const auth = createAuth({ config, keys, store });
const metrics = createMetrics({ store });
const metrics = createMetrics({ store });
@@ -116,7 +116,7 @@ function subscriptionView(row, { withSecret = false } = {}) {
@@ -116,7 +116,7 @@ function subscriptionView(row, { withSecret = false } = {}) {
return out;
return out;
}
}
-function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage: usageConfig = { enabled: true }, policies = null, planner = null } = {}) {
+function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage: usageConfig = { enabled: true }, policies = null, planner = null, countsCacheMs = 0 } = {}) {
// Project usage rollups (./usage.js): counted in the transactions below, sent as events.usage.recorded.
// Project usage rollups (./usage.js): counted in the transactions below, sent as events.usage.recorded.
const usage = createUsage(db, { clock, enabled: usageConfig.enabled !== false });
const usage = createUsage(db, { clock, enabled: usageConfig.enabled !== false });
const q = {
const q = {
@@ -714,11 +714,31 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
@@ -714,11 +714,31 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
return rows.length;
return rows.length;
});
});
- async function deliveryCounts() {
+ // 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.
+ 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 };
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;
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;
return out;
}
}
+ async function deliveryCounts({ fresh = false } = {}) {
+ const t = Date.now();
+ if (!fresh && counted && t - counted.at < COUNTS_MS) return counted.pending ? counted.pending : { ...counted.value };
+ const pending = countDeliveries();
+ counted = { at: t, pending };
+ try {
+ const value = await pending;
+ if (counted && counted.pending === pending) counted = { at: t, value };
+ return { ...value };
+ } catch (err) {
+ if (counted && counted.pending === pending) counted = null;
+ throw err;
+ }
+ }
// ── Checkpoints ──────────────────────────────────────────
// ── Checkpoints ──────────────────────────────────────────
new file mode 100644
new file mode 100644
@@ -0,0 +1,56 @@
@@ -0,0 +1,56 @@
+'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;
+// without it every call counts, so a test reads its own writes.
+const assert = require('assert');
+const { boot, suite } = require('./helpers');
+const { createStore } = require('../server/store');
+
+const t = suite('delivery-counts');
+let h;
+
+t('boot', async () => { h = await boot(); });
+
+t('one query for callers at the same moment, kept for countsCacheMs, fresh on demand', async () => {
+ let queries = 0;
+ const db = new Proxy(h.db, {
+ get(target, key) {
+ if (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); } };
+ };
+ }
+ const v = target[key];
+ return typeof v === 'function' ? v.bind(target) : v;
+ },
+ });
+ const kept = createStore(db, { clock: h.clock, countsCacheMs: 15_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);
+ assert.deepStrictEqual(b, c);
+ a.dead = 999;
+ assert.notStrictEqual((await kept.deliveryCounts()).dead, 999, 'each caller gets its own copy');
+ assert.strictEqual(queries, 1, 'within the window: kept');
+ await kept.deliveryCounts({ fresh: true });
+ assert.strictEqual(queries, 2, 'fresh: 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.deepStrictEqual(Object.keys(await every.deliveryCounts()).sort(), ['dead', 'delivered', 'failed', 'pending']);
+});
+
+t('the setting: off outside production unless EVENTS_COUNTS_CACHE_MS is set', async () => {
+ const { load: loadConfig } = require('../server/config');
+ 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', EVENTS_COUNTS_CACHE_MS: '5000' }).countsCacheMs, 5000);
+});
+
+t('close', async () => { await h.close(); });
+
+t.run();
@@ -275,7 +275,7 @@ t('backpressure: one in flight per subscription, <= maxInflight (20) overall', a
@@ -275,7 +275,7 @@ t('backpressure: one in flight per subscription, <= maxInflight (20) overall', a
await bp.worker.drain();
await bp.worker.drain();
assert.strictEqual(peak, 20, `peak in flight ${peak}`);
assert.strictEqual(peak, 20, `peak in flight ${peak}`);
assert.strictEqual(perSubPeak, 1);
assert.strictEqual(perSubPeak, 1);
- assert.strictEqual((await bp.store.deliveryCounts()).delivered, 75);
+ assert.strictEqual((await bp.store.deliveryCounts({ fresh: true })).delivered, 75);
await bp.stop();
await bp.stop();
});
});
@@ -232,7 +232,7 @@ t('max_inflight = 2: two keys run together, never two of one key, never three at
@@ -232,7 +232,7 @@ t('max_inflight = 2: two keys run together, never two of one key, never three at
assert.strictEqual(totalPeak, 2, 'two keys of the subscription went in flight together');
assert.strictEqual(totalPeak, 2, 'two keys of the subscription went in flight together');
assert.strictEqual(keyPeak, 1, 'never two of one key at once');
assert.strictEqual(keyPeak, 1, 'never two of one key at once');
- assert.strictEqual((await h.store.deliveryCounts()).pending, 0, 'every delivery was sent');
+ assert.strictEqual((await h.store.deliveryCounts({ fresh: true })).pending, 0, 'every delivery was sent');
await h.store.db.prepare('UPDATE subscriptions SET max_inflight = 1 WHERE id = ?').run(sub.id);
await h.store.db.prepare('UPDATE subscriptions SET max_inflight = 1 WHERE id = ?').run(sub.id);
await deletePolicy('live.ord4.*');
await deletePolicy('live.ord4.*');