openvibers/openvibe.sources Commits
The item count behind metrics and readiness is shared and kept 30 s in production (#189)
Changes
@@@ -20,6 +20,25 @@ const { jwksStatus } = require('openvibe-sdk/auth');const OVERDUE_MS = 15 * 60 * 1000;const STATUSES = ['healthy', 'stale', 'failing', 'never_fetched', 'disabled', 'manual'];++/**++ * 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).++ */++const COUNTS_MS = process.env.NODE_ENV === 'production' ? 30_000 : 0;++function kept(read, keepMs = COUNTS_MS, now = () => Date.now()) {++ if (!(keepMs > 0)) return read;++ let hold = null;++ return () => {++ const t = now();++ if (hold && (hold.pending || t - hold.at < keepMs)) return hold.pending || Promise.resolve(hold.value);++ const pending = Promise.resolve().then(read);++ hold = { at: t, pending };++ return pending.then((value) => { if (hold && hold.pending === pending) hold = { at: t, value }; return value; },++ (err) => { if (hold && hold.pending === pending) hold = null; throw err; });++ };++}++function readers(db, { startedAt }) {// A source waits from when it became due, when it was last changed (a new or re-enabled source// is due at once, from next_due_at 0) or when this process started, whichever is latest.@@@ -36,7 +55,7 @@ function readers(db, { startedAt }) {},lastFetchAt: async () => { const r = await lastRun.get(); return r ? r.finished_at : null; },lastSuccessAt: async () => (await lastSuccess.get()).t,-- items: async () => { const r = await items.get(); return { current: r.n - r.removed, removed: r.removed }; },++ items: kept(async () => { const r = await items.get(); return { current: r.n - r.removed, removed: r.removed }; }),};}@@@ -119,4 +138,4 @@ function registerSourcesGauges(registry, { db, sources, ingest, outbox, now }) {registry.gauge({ name: 'sources_outbox_pending', help: 'Events waiting in the outbox', collect: async () => (await outbox.status()).pending });}--module.exports = { createSourcesReadiness, registerSourcesGauges, OVERDUE_MS };++module.exports = { createSourcesReadiness, registerSourcesGauges, OVERDUE_MS, kept };nnew file mode 100644@@@ -0,0 +1,34 @@++'use strict';++/**++ * The item count behind the metrics and readiness (server/observability.js kept): callers at the same moment share one++ * read; within keepMs the kept answer; past it, read again; a failed read is not kept; keepMs 0 reads every time.++ *++ * node test/kept-counts.test.js++ */++const assert = require('assert');++const { kept } = require('../server/observability');++++(async () => {++ let t = 0;++ 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 [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;++ assert.strictEqual((await k()).current, 2, 'past it: read again');++ fail = true;++ t = 61_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');++ let plain = 0;++ const every = kept(async () => ++plain, 0);++ await every(); await every();++ assert.strictEqual(plain, 2, 'keepMs 0: every call reads');++ console.log('kept counts: all checks passed');++})().catch((err) => { console.error(err); process.exit(1); });
@@ -20,6 +20,25 @@ const { jwksStatus } = require('openvibe-sdk/auth');
@@ -20,6 +20,25 @@ const { jwksStatus } = require('openvibe-sdk/auth');
const OVERDUE_MS = 15 * 60 * 1000;
const OVERDUE_MS = 15 * 60 * 1000;
const STATUSES = ['healthy', 'stale', 'failing', 'never_fetched', 'disabled', 'manual'];
const STATUSES = ['healthy', 'stale', 'failing', 'never_fetched', 'disabled', 'manual'];
+/**
+ * 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).
+ */
+const COUNTS_MS = process.env.NODE_ENV === 'production' ? 30_000 : 0;
+function kept(read, keepMs = COUNTS_MS, now = () => Date.now()) {
+ if (!(keepMs > 0)) return read;
+ let hold = null;
+ return () => {
+ const t = now();
+ if (hold && (hold.pending || t - hold.at < keepMs)) return hold.pending || Promise.resolve(hold.value);
+ const pending = Promise.resolve().then(read);
+ hold = { at: t, pending };
+ return pending.then((value) => { if (hold && hold.pending === pending) hold = { at: t, value }; return value; },
+ (err) => { if (hold && hold.pending === pending) hold = null; throw err; });
+ };
+}
+
function readers(db, { startedAt }) {
function readers(db, { startedAt }) {
// A source waits from when it became due, when it was last changed (a new or re-enabled source
// A source waits from when it became due, when it was last changed (a new or re-enabled source
// is due at once, from next_due_at 0) or when this process started, whichever is latest.
// is due at once, from next_due_at 0) or when this process started, whichever is latest.
@@ -36,7 +55,7 @@ function readers(db, { startedAt }) {
@@ -36,7 +55,7 @@ function readers(db, { startedAt }) {
},
},
lastFetchAt: async () => { const r = await lastRun.get(); return r ? r.finished_at : null; },
lastFetchAt: async () => { const r = await lastRun.get(); return r ? r.finished_at : null; },
lastSuccessAt: async () => (await lastSuccess.get()).t,
lastSuccessAt: async () => (await lastSuccess.get()).t,
- items: async () => { const r = await items.get(); return { current: r.n - r.removed, removed: r.removed }; },
+ items: kept(async () => { const r = await items.get(); return { current: r.n - r.removed, removed: r.removed }; }),
};
};
}
}
@@ -119,4 +138,4 @@ function registerSourcesGauges(registry, { db, sources, ingest, outbox, now }) {
@@ -119,4 +138,4 @@ function registerSourcesGauges(registry, { db, sources, ingest, outbox, now }) {
registry.gauge({ name: 'sources_outbox_pending', help: 'Events waiting in the outbox', collect: async () => (await outbox.status()).pending });
registry.gauge({ name: 'sources_outbox_pending', help: 'Events waiting in the outbox', collect: async () => (await outbox.status()).pending });
}
}
-module.exports = { createSourcesReadiness, registerSourcesGauges, OVERDUE_MS };
+module.exports = { createSourcesReadiness, registerSourcesGauges, OVERDUE_MS, kept };
new file mode 100644
new file mode 100644
@@ -0,0 +1,34 @@
@@ -0,0 +1,34 @@
+'use strict';
+/**
+ * The item count behind the metrics and readiness (server/observability.js kept): callers at the same moment share one
+ * read; within keepMs the kept answer; past it, read again; a failed read is not kept; keepMs 0 reads every time.
+ *
+ * node test/kept-counts.test.js
+ */
+const assert = require('assert');
+const { kept } = require('../server/observability');
+
+(async () => {
+ let t = 0;
+ 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 [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;
+ assert.strictEqual((await k()).current, 2, 'past it: read again');
+ fail = true;
+ t = 61_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');
+ let plain = 0;
+ const every = kept(async () => ++plain, 0);
+ await every(); await every();
+ assert.strictEqual(plain, 2, 'keepMs 0: every call reads');
+ console.log('kept counts: all checks passed');
+})().catch((err) => { console.error(err); process.exit(1); });