openvibers/openvibe.events Commits
claim: stop at the inflight cap instead of scanning the whole backlog (#261)
Changes
@@@ -558,7 +558,7 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage// priority: app subscriptions that outrank first-party ones can never crowd them out of a claim. Only a// subscription with a CLAIMABLE delivery is a candidate (a backlog of non-heads cannot occupy claim slots).const candidates = (apps, n) => (n <= 0 ? [] : db.prepare(`-- SELECT s.id, s.project_id,++ SELECT s.id, s.project_id, s.max_inflight, ${leased('s.id')} AS leased,(SELECT d.priority FROM deliveries d WHERE d.subscription_id = s.id AND ${claimable('d')} ORDER BY d.priority, d.seq LIMIT 1) AS p,(SELECT d.seq FROM deliveries d WHERE d.subscription_id = s.id AND ${claimable('d')} ORDER BY d.priority, d.seq LIMIT 1) AS qFROM subscriptions s@@@ -566,7 +566,7 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usageAND EXISTS (SELECT 1 FROM deliveries d WHERE d.subscription_id = s.id AND ${claimable('d')})AND ${leased('s.id')} < s.max_inflightORDER BY p, q LIMIT ?-- FOR UPDATE OF s SKIP LOCKED`).all(now, now, now, now, now, now, now, now, now, now, n));++ FOR UPDATE OF s SKIP LOCKED`).all(now, now, now, now, now, now, now, now, now, now, now, n));const subs = [...await candidates(false, limit), ...await candidates(true, Math.min(limit, Math.max(0, maxApp)))].sort((x, y) => Number(x.p) - Number(y.p) || Number(x.q) - Number(y.q));const picked = [];@@@ -582,14 +582,18 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usagefor (const sub of subs) {if (picked.length >= limit) break;if (sub.project_id && apps >= maxApp) continue;-- // Up to (max_inflight − leased) deliveries of this subscription in one pass: `next` returns null when the-- // subscription is at its cap or has no claimable delivery left.-- for (;;) {++ // Up to (max_inflight − leased) deliveries of this subscription in one pass. Stop at the cap here: the++ // old loop asked `next` once more after the last slot was taken, and that lookup could never succeed —++ // its cap filter is false for every row — yet it walked every pending row of the subscription to prove++ // it. On 2026-11-08 that second lookup was half of 111,300 claim lookups and 442 M index tuples a day.++ const room = Math.max(0, Number(sub.max_inflight) - Number(sub.leased));++ for (let claimed = 0; claimed < room;) {if (picked.length >= limit || (sub.project_id && apps >= maxApp)) break;const d = await next.get(sub.id, now, now, now, now);if (!d) break;if ((await lease.run(owner, leaseUntil, d.event_id, d.subscription_id, now, now, now, now)).changes !== 1) break; // another worker got there firstpicked.push({ ...d, lease_owner: owner, lease_until: leaseUntil });++ claimed++;if (sub.project_id) apps++;}}nnew file mode 100644@@@ -0,0 +1,68 @@++'use strict';++/**++ * A claim pass stops when the subscription's inflight cap is reached.++ *++ * The old loop asked for the next delivery once more after the last slot was taken. That lookup could never++ * succeed (its cap filter is false for every row) yet the planner walked every pending row of the subscription++ * to prove it. On production on 2026-11-08 it was half of the 111,300 claim lookups and 442 M index tuples a++ * day (pg_stat_statements: 357 s of DB time, ~794 index tuples per lookup). This test proves the pass does not++ * make that second lookup, and what the lookup costs when the cap is reached (EXPLAIN ANALYZE on PGlite).++ *++ * node test/claim-cap-scan.test.js++ */++const assert = require('assert');++const { boot, request, serviceToken, envelope, subscriber, suite } = require('./helpers');++++const t = suite('claim-cap-scan');++let h;++++t('boot', async () => { h = await boot(); });++++t('a full subscription is not looked up again; the lookup it used to make walks the whole backlog', async () => {++ const stub = await subscriber(() => 204);++ const sub = await request(h.base, 'POST', '/api/v1/subscriptions', {++ token: serviceToken('wide', ['events.subscription.manage']),++ body: { topic_pattern: 'live.cap.*', endpoint: stub.url },++ });++ assert.strictEqual(sub.status, 201, sub.text);++ const subscriptionId = sub.body.id;++ for (let i = 0; i < 60; i++) {++ const r = await request(h.base, 'POST', '/api/v1/events', {++ token: serviceToken('live', ['events.event.publish']),++ body: envelope('live', { event_type: 'live.cap.x' }),++ });++ assert.ok(r.status === 201 || r.status === 200, r.text);++ }++++ // Capture the per-subscription claim lookup (`next` in store.js) the pass prepares.++ const lookups = [];++ const prepare = h.db.prepare.bind(h.db);++ h.db.prepare = (text) => {++ const q = prepare(text);++ if (/FROM deliveries d JOIN subscriptions s/.test(text) && /ORDER BY d.priority, d.seq LIMIT 1/.test(text)) lookups.push(q.source);++ return q;++ };++ const now = h.clock.now();++ let claimed;++ try {++ claimed = (await h.store.claimDeliveries(now, 1, { owner: 'worker-a', leaseUntil: now + 60_000 })).filter((d) => d.subscription_id === subscriptionId);++ } finally {++ h.db.prepare = prepare;++ }++ assert.strictEqual(claimed.length, 1, 'max_inflight 1: one delivery');++ assert.strictEqual(lookups.length, 1, 'one lookup — the claim itself, not a second one to discover the cap');++++ // What the second lookup cost: with the cap reached, every one of the other 59 pending rows is filtered out.++ await h.db.query('SET enable_seqscan = off');++ const plan = (await h.db.many(`EXPLAIN (ANALYZE, BUFFERS) ${lookups[0]}`, [subscriptionId, now, now, now, now]))++ .map((r) => Object.values(r)[0]).join('\n');++ assert.match(plan, /idx_deliveries_claim/, 'the claim index is the plan');++ // With the cap reached the cap filter removes every row the index returned: the query had to walk them all.++ const removed = Math.max(0, ...[...plan.matchAll(/Rows Removed by (?:Join )?Filter: (\d+)/g)].map((m) => Number(m[1])));++ assert.ok(removed >= 50, `the cap-reached lookup walks the whole pending backlog (removed ${removed})`);++ await stub.close();++});++++t('close', async () => { await h.close(); });++++t.run();
@@ -558,7 +558,7 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
@@ -558,7 +558,7 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
// priority: app subscriptions that outrank first-party ones can never crowd them out of a claim. Only a
// priority: app subscriptions that outrank first-party ones can never crowd them out of a claim. Only a
// subscription with a CLAIMABLE delivery is a candidate (a backlog of non-heads cannot occupy claim slots).
// subscription with a CLAIMABLE delivery is a candidate (a backlog of non-heads cannot occupy claim slots).
const candidates = (apps, n) => (n <= 0 ? [] : db.prepare(`
const candidates = (apps, n) => (n <= 0 ? [] : db.prepare(`
- SELECT s.id, s.project_id,
+ SELECT s.id, s.project_id, s.max_inflight, ${leased('s.id')} AS leased,
(SELECT d.priority FROM deliveries d WHERE d.subscription_id = s.id AND ${claimable('d')} ORDER BY d.priority, d.seq LIMIT 1) AS p,
(SELECT d.priority FROM deliveries d WHERE d.subscription_id = s.id AND ${claimable('d')} ORDER BY d.priority, d.seq LIMIT 1) AS p,
(SELECT d.seq FROM deliveries d WHERE d.subscription_id = s.id AND ${claimable('d')} ORDER BY d.priority, d.seq LIMIT 1) AS q
(SELECT d.seq FROM deliveries d WHERE d.subscription_id = s.id AND ${claimable('d')} ORDER BY d.priority, d.seq LIMIT 1) AS q
FROM subscriptions s
FROM subscriptions s
@@ -566,7 +566,7 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
@@ -566,7 +566,7 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
AND EXISTS (SELECT 1 FROM deliveries d WHERE d.subscription_id = s.id AND ${claimable('d')})
AND EXISTS (SELECT 1 FROM deliveries d WHERE d.subscription_id = s.id AND ${claimable('d')})
AND ${leased('s.id')} < s.max_inflight
AND ${leased('s.id')} < s.max_inflight
ORDER BY p, q LIMIT ?
ORDER BY p, q LIMIT ?
- FOR UPDATE OF s SKIP LOCKED`).all(now, now, now, now, now, now, now, now, now, now, n));
+ FOR UPDATE OF s SKIP LOCKED`).all(now, now, now, now, now, now, now, now, now, now, now, n));
const subs = [...await candidates(false, limit), ...await candidates(true, Math.min(limit, Math.max(0, maxApp)))]
const subs = [...await candidates(false, limit), ...await candidates(true, Math.min(limit, Math.max(0, maxApp)))]
.sort((x, y) => Number(x.p) - Number(y.p) || Number(x.q) - Number(y.q));
.sort((x, y) => Number(x.p) - Number(y.p) || Number(x.q) - Number(y.q));
const picked = [];
const picked = [];
@@ -582,14 +582,18 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
@@ -582,14 +582,18 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
for (const sub of subs) {
for (const sub of subs) {
if (picked.length >= limit) break;
if (picked.length >= limit) break;
if (sub.project_id && apps >= maxApp) continue;
if (sub.project_id && apps >= maxApp) continue;
- // Up to (max_inflight − leased) deliveries of this subscription in one pass: `next` returns null when the
- // subscription is at its cap or has no claimable delivery left.
- for (;;) {
+ // Up to (max_inflight − leased) deliveries of this subscription in one pass. Stop at the cap here: the
+ // old loop asked `next` once more after the last slot was taken, and that lookup could never succeed —
+ // its cap filter is false for every row — yet it walked every pending row of the subscription to prove
+ // it. On 2026-11-08 that second lookup was half of 111,300 claim lookups and 442 M index tuples a day.
+ const room = Math.max(0, Number(sub.max_inflight) - Number(sub.leased));
+ for (let claimed = 0; claimed < room;) {
if (picked.length >= limit || (sub.project_id && apps >= maxApp)) break;
if (picked.length >= limit || (sub.project_id && apps >= maxApp)) break;
const d = await next.get(sub.id, now, now, now, now);
const d = await next.get(sub.id, now, now, now, now);
if (!d) break;
if (!d) break;
if ((await lease.run(owner, leaseUntil, d.event_id, d.subscription_id, now, now, now, now)).changes !== 1) break; // another worker got there first
if ((await lease.run(owner, leaseUntil, d.event_id, d.subscription_id, now, now, now, now)).changes !== 1) break; // another worker got there first
picked.push({ ...d, lease_owner: owner, lease_until: leaseUntil });
picked.push({ ...d, lease_owner: owner, lease_until: leaseUntil });
+ claimed++;
if (sub.project_id) apps++;
if (sub.project_id) apps++;
}
}
}
}
new file mode 100644
new file mode 100644
@@ -0,0 +1,68 @@
@@ -0,0 +1,68 @@
+'use strict';
+/**
+ * A claim pass stops when the subscription's inflight cap is reached.
+ *
+ * The old loop asked for the next delivery once more after the last slot was taken. That lookup could never
+ * succeed (its cap filter is false for every row) yet the planner walked every pending row of the subscription
+ * to prove it. On production on 2026-11-08 it was half of the 111,300 claim lookups and 442 M index tuples a
+ * day (pg_stat_statements: 357 s of DB time, ~794 index tuples per lookup). This test proves the pass does not
+ * make that second lookup, and what the lookup costs when the cap is reached (EXPLAIN ANALYZE on PGlite).
+ *
+ * node test/claim-cap-scan.test.js
+ */
+const assert = require('assert');
+const { boot, request, serviceToken, envelope, subscriber, suite } = require('./helpers');
+
+const t = suite('claim-cap-scan');
+let h;
+
+t('boot', async () => { h = await boot(); });
+
+t('a full subscription is not looked up again; the lookup it used to make walks the whole backlog', async () => {
+ const stub = await subscriber(() => 204);
+ const sub = await request(h.base, 'POST', '/api/v1/subscriptions', {
+ token: serviceToken('wide', ['events.subscription.manage']),
+ body: { topic_pattern: 'live.cap.*', endpoint: stub.url },
+ });
+ assert.strictEqual(sub.status, 201, sub.text);
+ const subscriptionId = sub.body.id;
+ for (let i = 0; i < 60; i++) {
+ const r = await request(h.base, 'POST', '/api/v1/events', {
+ token: serviceToken('live', ['events.event.publish']),
+ body: envelope('live', { event_type: 'live.cap.x' }),
+ });
+ assert.ok(r.status === 201 || r.status === 200, r.text);
+ }
+
+ // Capture the per-subscription claim lookup (`next` in store.js) the pass prepares.
+ const lookups = [];
+ const prepare = h.db.prepare.bind(h.db);
+ h.db.prepare = (text) => {
+ const q = prepare(text);
+ if (/FROM deliveries d JOIN subscriptions s/.test(text) && /ORDER BY d.priority, d.seq LIMIT 1/.test(text)) lookups.push(q.source);
+ return q;
+ };
+ const now = h.clock.now();
+ let claimed;
+ try {
+ claimed = (await h.store.claimDeliveries(now, 1, { owner: 'worker-a', leaseUntil: now + 60_000 })).filter((d) => d.subscription_id === subscriptionId);
+ } finally {
+ h.db.prepare = prepare;
+ }
+ assert.strictEqual(claimed.length, 1, 'max_inflight 1: one delivery');
+ assert.strictEqual(lookups.length, 1, 'one lookup — the claim itself, not a second one to discover the cap');
+
+ // What the second lookup cost: with the cap reached, every one of the other 59 pending rows is filtered out.
+ await h.db.query('SET enable_seqscan = off');
+ const plan = (await h.db.many(`EXPLAIN (ANALYZE, BUFFERS) ${lookups[0]}`, [subscriptionId, now, now, now, now]))
+ .map((r) => Object.values(r)[0]).join('\n');
+ assert.match(plan, /idx_deliveries_claim/, 'the claim index is the plan');
+ // With the cap reached the cap filter removes every row the index returned: the query had to walk them all.
+ const removed = Math.max(0, ...[...plan.matchAll(/Rows Removed by (?:Join )?Filter: (\d+)/g)].map((m) => Number(m[1])));
+ assert.ok(removed >= 50, `the cap-reached lookup walks the whole pending backlog (removed ${removed})`);
+ await stub.close();
+});
+
+t('close', async () => { await h.close(); });
+
+t.run();