openvibers/openvibe.events Commits
A leased delivery is never claimable: max_inflight N runs N different deliveries (#220)
Changes
@@@ -538,9 +538,13 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage*/const claimDeliveries = async (now, limit, { owner, leaseUntil, maxApp = limit } = {}) => {if (!owner || !(leaseUntil > now) || limit <= 0) return [];-- // Claimable: due, and the head of its key (no lower-seq pending/failed row of it, nothing of it leased). Two-- // bindings, both `now` (the due check and the lease check).++ // Claimable: due, not leased itself, and the head of its key (no lower-seq pending/failed row of it, nothing of++ // it leased). Three bindings, all `now` (the due check, its own lease and the key's leases). Without its own++ // lease check a delivery with no ordering key stayed claimable while in flight, so a subscription with++ // max_inflight above 1 had the same lowest row claimed up to max_inflight times in one pass: Search's++ // deliveries went out six times each on 2026-11-02 (27,428 "lease expired" lines in 15 minutes).const claimable = (d) => `(${d}.status IN ('pending', 'failed') AND ${d}.next_attempt_at <= ?++ AND (${d}.lease_until IS NULL OR ${d}.lease_until <= ?)AND (${d}.ordering_key IS NULL OR (NOT EXISTS (SELECT 1 FROM deliveries x WHERE x.subscription_id = ${d}.subscription_idAND x.ordering_key = ${d}.ordering_key AND x.status IN ('pending', 'failed') AND x.seq < ${d}.seq)@@@ -562,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, n));++ FOR UPDATE OF s SKIP LOCKED`).all(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,9 +586,9 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage// subscription is at its cap or has no claimable delivery left.for (;;) {if (picked.length >= limit || (sub.project_id && apps >= maxApp)) break;-- const d = await next.get(sub.id, now, now, now);++ 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)).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 firstpicked.push({ ...d, lease_owner: owner, lease_until: leaseUntil });if (sub.project_id) apps++;}@@@ -613,6 +617,7 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usageif (!locked) return null;const r = await db.prepare(`UPDATE deliveries d SET lease_owner = ?, lease_until = ? WHERE d.event_id = ? AND d.subscription_id = ?AND d.status IN ('pending', 'failed') AND d.next_attempt_at <= ?++ AND (d.lease_until IS NULL OR d.lease_until <= ?)AND (d.ordering_key IS NULL OR (NOT EXISTS (SELECT 1 FROM deliveries x WHERE x.subscription_id = d.subscription_idAND x.ordering_key = d.ordering_key AND x.status IN ('pending', 'failed') AND x.seq < d.seq)@@@ -622,7 +627,7 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usageAND (SELECT COUNT(*) FROM deliveries x WHERE x.subscription_id = d.subscription_id AND x.lease_until > ?)< (SELECT s.max_inflight FROM subscriptions s WHERE s.id = d.subscription_id AND s.enabled = 1)RETURNING d.*, (SELECT s.project_id FROM subscriptions s WHERE s.id = d.subscription_id) AS app_project_id`)-- .get(owner, leaseUntil, eventId, subscriptionId, now, now, now);++ .get(owner, leaseUntil, eventId, subscriptionId, now, now, now, now);return r || null;});};nnew file mode 100644@@@ -0,0 +1,41 @@++'use strict';++// A subscription with max_inflight above 1 runs that many DIFFERENT deliveries at once, never one delivery several++// times. A delivery with no ordering key was claimable while leased, so one pass claimed the same lowest row up to++// max_inflight times: Search's deliveries went out six times each on 2026-11-02.++const assert = require('assert');++const { boot, request, serviceToken, envelope, subscriber, suite } = require('./helpers');++++const t = suite('inflight-claim');++let h;++++t('boot', async () => { h = await boot(); });++++t('max_inflight 3, no ordering keys: one pass claims three different deliveries, then nothing while they are leased', 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.wide.*', endpoint: stub.url } });++ assert.strictEqual(sub.status, 201, sub.text);++ await h.db.prepare('UPDATE subscriptions SET max_inflight = 3 WHERE id = ?').run(sub.body.id);++ for (let i = 0; i < 5; i++) {++ const r = await request(h.base, 'POST', '/api/v1/events', { token: serviceToken('live', ['events.event.publish']), body: envelope('live', { event_type: 'live.wide.x' }) });++ assert.ok(r.status === 201 || r.status === 200, r.text);++ }++ const now = h.clock.now();++ const first = (await h.store.claimDeliveries(now, 10, { owner: 'worker-a', leaseUntil: now + 60_000 })).filter((d) => d.subscription_id === sub.body.id);++ assert.strictEqual(first.length, 3, 'up to max_inflight');++ assert.strictEqual(new Set(first.map((d) => d.event_id)).size, 3, 'three different deliveries, not one three times');++ const again = (await h.store.claimDeliveries(now + 1000, 10, { owner: 'worker-a', leaseUntil: now + 61_000 })).filter((d) => d.subscription_id === sub.body.id);++ assert.strictEqual(again.length, 0, 'at its cap: nothing more while those are leased');++ // One of them finishes: one more, and never one already leased.++ await h.store.recordAttempt(first[0].event_id, sub.body.id, { ok: true, attempt: 1, status: 204 }, null, 'worker-a');++ const more = (await h.store.claimDeliveries(now + 2000, 10, { owner: 'worker-a', leaseUntil: now + 62_000 })).filter((d) => d.subscription_id === sub.body.id);++ assert.strictEqual(more.length, 1, 'one slot freed, one claimed');++ assert.ok(!first.some((d) => d.event_id === more[0].event_id), 'a delivery that was not leased');++ // The single-delivery claim (the carrier path) refuses one that is leased.++ const leased = first[1];++ assert.strictEqual(await h.store.claimDelivery(leased.event_id, sub.body.id, { owner: 'worker-b', leaseUntil: h.clock.now() + 60_000 }), null, 'a leased delivery is not claimed again');++ await stub.close();++});++++t('close', async () => { await h.close(); });++++t.run();
@@ -538,9 +538,13 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
@@ -538,9 +538,13 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
*/
*/
const claimDeliveries = async (now, limit, { owner, leaseUntil, maxApp = limit } = {}) => {
const claimDeliveries = async (now, limit, { owner, leaseUntil, maxApp = limit } = {}) => {
if (!owner || !(leaseUntil > now) || limit <= 0) return [];
if (!owner || !(leaseUntil > now) || limit <= 0) return [];
- // Claimable: due, and the head of its key (no lower-seq pending/failed row of it, nothing of it leased). Two
- // bindings, both `now` (the due check and the lease check).
+ // Claimable: due, not leased itself, and the head of its key (no lower-seq pending/failed row of it, nothing of
+ // it leased). Three bindings, all `now` (the due check, its own lease and the key's leases). Without its own
+ // lease check a delivery with no ordering key stayed claimable while in flight, so a subscription with
+ // max_inflight above 1 had the same lowest row claimed up to max_inflight times in one pass: Search's
+ // deliveries went out six times each on 2026-11-02 (27,428 "lease expired" lines in 15 minutes).
const claimable = (d) => `(${d}.status IN ('pending', 'failed') AND ${d}.next_attempt_at <= ?
const claimable = (d) => `(${d}.status IN ('pending', 'failed') AND ${d}.next_attempt_at <= ?
+ AND (${d}.lease_until IS NULL OR ${d}.lease_until <= ?)
AND (${d}.ordering_key IS NULL OR (
AND (${d}.ordering_key IS NULL OR (
NOT EXISTS (SELECT 1 FROM deliveries x WHERE x.subscription_id = ${d}.subscription_id
NOT EXISTS (SELECT 1 FROM deliveries x WHERE x.subscription_id = ${d}.subscription_id
AND x.ordering_key = ${d}.ordering_key AND x.status IN ('pending', 'failed') AND x.seq < ${d}.seq)
AND x.ordering_key = ${d}.ordering_key AND x.status IN ('pending', 'failed') AND x.seq < ${d}.seq)
@@ -562,7 +566,7 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
@@ -562,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, n));
+ FOR UPDATE OF s SKIP LOCKED`).all(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,9 +586,9 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
@@ -582,9 +586,9 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
// subscription is at its cap or has no claimable delivery left.
// subscription is at its cap or has no claimable delivery left.
for (;;) {
for (;;) {
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);
+ 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)).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 });
if (sub.project_id) apps++;
if (sub.project_id) apps++;
}
}
@@ -613,6 +617,7 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
@@ -613,6 +617,7 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
if (!locked) return null;
if (!locked) return null;
const r = await db.prepare(`UPDATE deliveries d SET lease_owner = ?, lease_until = ? WHERE d.event_id = ? AND d.subscription_id = ?
const r = await db.prepare(`UPDATE deliveries d SET lease_owner = ?, lease_until = ? WHERE d.event_id = ? AND d.subscription_id = ?
AND d.status IN ('pending', 'failed') AND d.next_attempt_at <= ?
AND d.status IN ('pending', 'failed') AND d.next_attempt_at <= ?
+ AND (d.lease_until IS NULL OR d.lease_until <= ?)
AND (d.ordering_key IS NULL OR (
AND (d.ordering_key IS NULL OR (
NOT EXISTS (SELECT 1 FROM deliveries x WHERE x.subscription_id = d.subscription_id
NOT EXISTS (SELECT 1 FROM deliveries x WHERE x.subscription_id = d.subscription_id
AND x.ordering_key = d.ordering_key AND x.status IN ('pending', 'failed') AND x.seq < d.seq)
AND x.ordering_key = d.ordering_key AND x.status IN ('pending', 'failed') AND x.seq < d.seq)
@@ -622,7 +627,7 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
@@ -622,7 +627,7 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
AND (SELECT COUNT(*) FROM deliveries x WHERE x.subscription_id = d.subscription_id AND x.lease_until > ?)
AND (SELECT COUNT(*) FROM deliveries x WHERE x.subscription_id = d.subscription_id AND x.lease_until > ?)
< (SELECT s.max_inflight FROM subscriptions s WHERE s.id = d.subscription_id AND s.enabled = 1)
< (SELECT s.max_inflight FROM subscriptions s WHERE s.id = d.subscription_id AND s.enabled = 1)
RETURNING d.*, (SELECT s.project_id FROM subscriptions s WHERE s.id = d.subscription_id) AS app_project_id`)
RETURNING d.*, (SELECT s.project_id FROM subscriptions s WHERE s.id = d.subscription_id) AS app_project_id`)
- .get(owner, leaseUntil, eventId, subscriptionId, now, now, now);
+ .get(owner, leaseUntil, eventId, subscriptionId, now, now, now, now);
return r || null;
return r || null;
});
});
};
};
new file mode 100644
new file mode 100644
@@ -0,0 +1,41 @@
@@ -0,0 +1,41 @@
+'use strict';
+// A subscription with max_inflight above 1 runs that many DIFFERENT deliveries at once, never one delivery several
+// times. A delivery with no ordering key was claimable while leased, so one pass claimed the same lowest row up to
+// max_inflight times: Search's deliveries went out six times each on 2026-11-02.
+const assert = require('assert');
+const { boot, request, serviceToken, envelope, subscriber, suite } = require('./helpers');
+
+const t = suite('inflight-claim');
+let h;
+
+t('boot', async () => { h = await boot(); });
+
+t('max_inflight 3, no ordering keys: one pass claims three different deliveries, then nothing while they are leased', 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.wide.*', endpoint: stub.url } });
+ assert.strictEqual(sub.status, 201, sub.text);
+ await h.db.prepare('UPDATE subscriptions SET max_inflight = 3 WHERE id = ?').run(sub.body.id);
+ for (let i = 0; i < 5; i++) {
+ const r = await request(h.base, 'POST', '/api/v1/events', { token: serviceToken('live', ['events.event.publish']), body: envelope('live', { event_type: 'live.wide.x' }) });
+ assert.ok(r.status === 201 || r.status === 200, r.text);
+ }
+ const now = h.clock.now();
+ const first = (await h.store.claimDeliveries(now, 10, { owner: 'worker-a', leaseUntil: now + 60_000 })).filter((d) => d.subscription_id === sub.body.id);
+ assert.strictEqual(first.length, 3, 'up to max_inflight');
+ assert.strictEqual(new Set(first.map((d) => d.event_id)).size, 3, 'three different deliveries, not one three times');
+ const again = (await h.store.claimDeliveries(now + 1000, 10, { owner: 'worker-a', leaseUntil: now + 61_000 })).filter((d) => d.subscription_id === sub.body.id);
+ assert.strictEqual(again.length, 0, 'at its cap: nothing more while those are leased');
+ // One of them finishes: one more, and never one already leased.
+ await h.store.recordAttempt(first[0].event_id, sub.body.id, { ok: true, attempt: 1, status: 204 }, null, 'worker-a');
+ const more = (await h.store.claimDeliveries(now + 2000, 10, { owner: 'worker-a', leaseUntil: now + 62_000 })).filter((d) => d.subscription_id === sub.body.id);
+ assert.strictEqual(more.length, 1, 'one slot freed, one claimed');
+ assert.ok(!first.some((d) => d.event_id === more[0].event_id), 'a delivery that was not leased');
+ // The single-delivery claim (the carrier path) refuses one that is leased.
+ const leased = first[1];
+ assert.strictEqual(await h.store.claimDelivery(leased.event_id, sub.body.id, { owner: 'worker-b', leaseUntil: h.clock.now() + 60_000 }), null, 'a leased delivery is not claimed again');
+ await stub.close();
+});
+
+t('close', async () => { await h.close(); });
+
+t.run();