openvibers/openvibe.events Commits

Deliveries are claimed with a database lease (plan T7, ADR-042 decision 3): the claim locks the subscription rows FOR UPDATE SKIP LOCKED and leases each one's next due delivery (lease_owner, lease_until; migration 0002, which also adds ordering_key and carrier for the planner), so any number of workers share the queue with at most one delivery in flight per subscription; the lease is re-checked in the UPDATE's own snapshot, because READ COMMITTED does not re-evaluate the candidate query's NOT EXISTS when another worker commits a lease just before the lock; an expired lease is claimable again (at least once) and a worker that lost its lease does not record its late result. First-party and developer-app candidates are fetched apart so app traffic never crowds first-party out of a claim. test/lease.test.js: two workers, every delivery exactly once and one in flight per subscription (it double-sent about every second run on PostgreSQL before the re-check), lease expiry, a lost lease. npm test and test:pg 18/18.

OP OpenVibers committed 29 Sept 2026 · f7f592b938 · 5 files, +164 −38


Changes

ddiff --git a/migrations/0002_delivery_lease.sql b/migrations/0002_delivery_lease.sql
nnew file mode 100644
iindex 0000000..da7714c
---- /dev/null
++++ b/migrations/0002_delivery_lease.sql
@@@ -0,0 +1,10 @@
++-- phase: expand
++-- ADR-042 decision 3: a delivery is claimed with a database lease, so more than one worker can run without sending the
++-- same delivery twice. lease_owner/lease_until mark the one in flight (at most one per subscription: the claim locks the
++-- subscription row, FOR UPDATE SKIP LOCKED); an expired lease is claimable again (at least once). ordering_key and
++-- carrier are filled by the planner (ADR-042 decisions 4-5); until then they stay NULL.
++ALTER TABLE deliveries ADD COLUMN lease_owner text COLLATE "C";
++ALTER TABLE deliveries ADD COLUMN lease_until bigint;
++ALTER TABLE deliveries ADD COLUMN ordering_key text COLLATE "C";
++ALTER TABLE deliveries ADD COLUMN carrier text COLLATE "C";
++CREATE INDEX idx_deliveries_lease ON deliveries(subscription_id, lease_until) WHERE lease_until IS NOT NULL;
ddiff --git a/server/store.js b/server/store.js
iindex 9273580..052e6b6 100644
---- a/server/store.js
++++ b/server/store.js
@@@ -406,29 +406,65 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
// ── Deliveries ───────────────────────────────────────────
/**
-- * Next due delivery per subscription (at most one each, so per-subscription concurrency stays 1),
-- * ordered by priority class, then seq. `busy` are subscription ids with a delivery in flight.
++ * Claim up to `limit` due deliveries for `owner` until `leaseUntil` (ms): the next due delivery (by priority class,
++ * then seq) of each subscription that has none in flight. The claim locks the subscription rows
++ * (FOR UPDATE SKIP LOCKED), so two workers — in one process or many — never claim the same subscription at once, and
++ * the lease it writes keeps every other worker off that subscription until recordAttempt clears it or it expires
++ * (an expired lease is claimable again: delivery is at least once). `maxApp` caps developer-app subscriptions.
*/
-- async function dueDeliveries(now, limit, busy = [], { firstPartyOnly = false } = {}) {
-- // app_project_id: set for developer-app subscriptions (the worker caps their share of slots).
-- const rows = await db.prepare(`
-- SELECT * FROM (
-- SELECT d.*, s.project_id AS app_project_id, ROW_NUMBER() OVER (PARTITION BY d.subscription_id ORDER BY d.priority, d.seq) AS rn
-- FROM deliveries d JOIN subscriptions s ON s.id = d.subscription_id
-- WHERE d.status IN ('pending', 'failed') AND d.next_attempt_at <= ? AND s.enabled = 1
-- ${firstPartyOnly ? 'AND s.project_id IS NULL' : ''}
-- ) WHERE rn = 1 ORDER BY priority, seq LIMIT ?`).all(now, limit + busy.length);
-- const skip = new Set(busy);
-- return rows.filter(r => !skip.has(r.subscription_id)).slice(0, limit);
-- }
++ const claimDeliveries = async (now, limit, { owner, leaseUntil, maxApp = limit } = {}) => {
++ if (!owner || !(leaseUntil > now) || limit <= 0) return [];
++ return await db.tx(async () => {
++ const due = `d.status IN ('pending', 'failed') AND d.next_attempt_at <= ?`;
++ // First-party and developer-app subscriptions are fetched apart, each up to its own share, then merged by
++ // priority: app subscriptions that outrank first-party ones can never crowd them out of a claim.
++ const candidates = (apps, n) => (n <= 0 ? [] : db.prepare(`
++ SELECT s.id, s.project_id,
++ (SELECT d.priority FROM deliveries d WHERE d.subscription_id = s.id AND ${due} ORDER BY d.priority, d.seq LIMIT 1) AS p,
++ (SELECT d.seq FROM deliveries d WHERE d.subscription_id = s.id AND ${due} ORDER BY d.priority, d.seq LIMIT 1) AS q
++ FROM subscriptions s
++ WHERE s.enabled = 1 AND s.project_id IS ${apps ? 'NOT ' : ''}NULL
++ AND EXISTS (SELECT 1 FROM deliveries d WHERE d.subscription_id = s.id AND ${due})
++ AND NOT EXISTS (SELECT 1 FROM deliveries x WHERE x.subscription_id = s.id AND x.lease_until > ?)
++ ORDER BY p, q LIMIT ?
++ FOR UPDATE OF s SKIP LOCKED`).all(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 = [];
++ let apps = 0;
++ const next = db.prepare(`SELECT d.*, s.project_id AS app_project_id FROM deliveries d JOIN subscriptions s ON s.id = d.subscription_id
++ WHERE d.subscription_id = ? AND d.status IN ('pending', 'failed') AND d.next_attempt_at <= ? ORDER BY d.priority, d.seq LIMIT 1`);
++ // The lease is taken only if nothing of this subscription is leased as of NOW: this statement starts after the
++ // subscription row was locked, so it sees a lease another worker committed after the candidate query's
++ // snapshot (READ COMMITTED does not re-check that query's NOT EXISTS when the locked row itself did not change).
++ const lease = 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 NOT EXISTS (SELECT 1 FROM deliveries x WHERE x.subscription_id = d.subscription_id AND x.lease_until > ?)`);
++ for (const sub of subs) {
++ if (picked.length >= limit) break;
++ if (sub.project_id && apps >= maxApp) continue;
++ const d = await next.get(sub.id, now);
++ if (!d) continue;
++ if ((await lease.run(owner, leaseUntil, d.event_id, d.subscription_id, now)).changes !== 1) continue; // another worker got there first
++ picked.push({ ...d, lease_owner: owner, lease_until: leaseUntil });
++ if (sub.project_id) apps++;
++ }
++ return picked;
++ });
++ };
/**
* Record one delivery attempt. `app` ({ projectId, env, traceId }) marks a developer-app
* subscription: the attempt counts toward the project's events.app.subscribe usage in the same
* transaction.
*/
-- const recordAttempt = async (eventId, subscriptionId, outcome, app = null) => await db.tx(async () => {
++ const recordAttempt = async (eventId, subscriptionId, outcome, app = null, owner = null) => await db.tx(async () => {
const now = clock.now();
++ // The attempt is this worker's only while it still holds the lease: after an expiry another worker may have
++ // claimed the delivery again, and its result is the one to keep (this one returns false and records nothing).
++ if (owner) {
++ const held = await db.prepare('SELECT 1 FROM deliveries WHERE event_id = ? AND subscription_id = ? AND lease_owner = ? FOR UPDATE').get(eventId, subscriptionId, owner);
++ if (!held) return false;
++ }
if (app) {
await usage.record({
projectId: app.projectId, env: app.env, capability: 'events.app.subscribe', unit: 'deliveries', quantity: 1, at: now,
@@@ -437,14 +473,15 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
}
if (outcome.ok) {
await db.prepare(`UPDATE deliveries SET status = 'delivered', attempt = ?, delivered_at = ?, last_status = ?, last_error = NULL,
-- next_attempt_at = NULL, updated_at = ? WHERE event_id = ? AND subscription_id = ?`)
++ next_attempt_at = NULL, lease_owner = NULL, lease_until = NULL, updated_at = ? WHERE event_id = ? AND subscription_id = ?`)
.run(outcome.attempt, now, outcome.status ?? null, now, eventId, subscriptionId);
} else {
-- await db.prepare(`UPDATE deliveries SET status = ?, attempt = ?, last_status = ?, last_error = ?, next_attempt_at = ?, updated_at = ?
-- WHERE event_id = ? AND subscription_id = ?`)
++ await db.prepare(`UPDATE deliveries SET status = ?, attempt = ?, last_status = ?, last_error = ?, next_attempt_at = ?,
++ lease_owner = NULL, lease_until = NULL, updated_at = ? WHERE event_id = ? AND subscription_id = ?`)
.run(outcome.dead ? 'dead' : 'failed', outcome.attempt, outcome.status ?? null, String(outcome.error || '').slice(0, 500),
outcome.dead ? null : outcome.nextAttemptAt, now, eventId, subscriptionId);
}
++ return true;
});
/** Send the closed hours' usage rollups (./usage.js); returns the stored rows. */
@@@ -532,7 +569,7 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
return {
db, insertBatch, redact, getEvent, revokedAt, lastSeq, oldestSeq, firstSeqSince, scan,
createSubscription, getSubscription, listSubscriptions, countSubscriptions, countProjectSubscriptions, setSubscriptionEnabled, rotateSubscriptionSecret,
-- dueDeliveries, recordAttempt, getDelivery, listDeliveries, requeue, deliveryCounts,
++ claimDeliveries, recordAttempt, getDelivery, listDeliveries, requeue, deliveryCounts,
getCheckpoint, setCheckpoint, prune, ping, usage, flushUsage,
};
}
ddiff --git a/server/worker.js b/server/worker.js
iindex c89052f..1776106 100644
---- a/server/worker.js
++++ b/server/worker.js
@@@ -20,6 +20,7 @@
* most maxAppInflight to developer-app endpoints (so apps can never hold every slot).
*/
const crypto = require('crypto');
++const os = require('os');
const { signV2 } = require('../lib/client');
/**
@@@ -44,7 +45,11 @@ const { createGuardedPost } = require('./egress');
// observe (optional): { delivered(seconds, row), attempt(outcome) } — metrics hooks, never required.
function createWorker({ store, config, clock = { now: () => Date.now() }, fetchImpl = globalThis.fetch, appPost = createGuardedPost(), log = console, observe = null }) {
const opts = config.worker;
-- const busy = new Set(); // subscription ids with a delivery in flight
++ // This worker's lease identity (ADR-042 decision 3): claims are leased in the database, so any number of workers, in
++ // one process or many, share the queue without sending a delivery twice. A lease outlives one attempt.
++ const owner = `${os.hostname()}:${process.pid}:${crypto.randomBytes(4).toString('hex')}`;
++ const leaseMs = () => (Number(opts.leaseMs) > 0 ? Number(opts.leaseMs) : Number(opts.timeoutMs || 10000) + 30000);
++ const busy = new Set(); // subscription ids with a delivery in flight here (stats; the lease is what excludes)
const inflight = new Set(); // promises
let appInflight = 0; // of which to developer-app endpoints (at most opts.maxAppInflight)
let timer = null;
@@@ -112,7 +117,8 @@ function createWorker({ store, config, clock = { now: () => Date.now() }, fetchI
}
// A developer app's subscription: the attempt counts toward its project's usage (./usage.js).
const app = sub.project_id ? { projectId: sub.project_id, env: sub.env || 'production', traceId: row.trace_id } : null;
-- await store.recordAttempt(delivery.event_id, delivery.subscription_id, outcome, app);
++ const kept = await store.recordAttempt(delivery.event_id, delivery.subscription_id, outcome, app, owner);
++ if (kept === false) { log.warn(`[worker] ${row.id} -> ${sub.id}: the lease expired before the attempt was recorded; the newer claim's result stands`); return; }
if (observe) {
try {
observe.attempt(outcome.ok ? 'delivered' : outcome.dead ? 'dead' : 'retry');
@@@ -141,22 +147,11 @@ function createWorker({ store, config, clock = { now: () => Date.now() }, fetchI
lastTickAt = Date.now();
const room = opts.maxInflight - inflight.size;
if (room <= 0) return 0;
-- let started = 0;
-- let skippedApp = false;
-- const startAll = (due) => {
-- for (const d of due) {
-- if (started >= room) return;
-- const isApp = Boolean(d.app_project_id);
-- if (isApp && appInflight >= opts.maxAppInflight) { skippedApp = true; continue; }
-- start(d, isApp);
-- started++;
-- }
-- };
-- startAll(await store.dueDeliveries(clock.now(), room, [...busy], { firstPartyOnly: appInflight >= opts.maxAppInflight }));
-- // App deliveries (which can outrank first-party ones by priority) filled the app share: the
-- // rest of the room goes to first-party subscriptions.
-- if (skippedApp && started < room) startAll(await store.dueDeliveries(clock.now(), room - started, [...busy], { firstPartyOnly: true }));
-- return started;
++ const now = clock.now();
++ // Everything claimed here is leased to this worker; developer-app subscriptions get at most their share of slots.
++ const claimed = await store.claimDeliveries(now, room, { owner, leaseUntil: now + leaseMs(), maxApp: Math.max(0, opts.maxAppInflight - appInflight) });
++ for (const d of claimed) start(d, Boolean(d.app_project_id));
++ return claimed.length;
}
function start(d, isApp) {
ddiff --git a/test/delivery.test.js b/test/delivery.test.js
iindex 872c83b..9a439aa 100644
---- a/test/delivery.test.js
++++ b/test/delivery.test.js
@@@ -203,9 +203,11 @@ t('priority classes: critical before important before low', async () => {
await publish(envelope('live', { event_type: 'live.prio.x', priority: 'low' }));
await publish(envelope('live', { event_type: 'live.prio.x', priority: 'important' }));
await publish(envelope('live', { event_type: 'live.prio.x', priority: 'critical' }));
-- const due = await h.store.dueDeliveries(h.clock.now(), 100);
++ // A probe claim (leased for 1 ms) sees each subscription's head; its leases then lapse and the worker claims again.
++ const due = await h.store.claimDeliveries(h.clock.now(), 100, { owner: 'probe', leaseUntil: h.clock.now() + 1 });
assert.strictEqual(due.length, 3, 'one per subscription');
assert.ok(due.every(d => d.priority === 0), 'each subscription\'s head is its critical event');
++ h.clock.advance(2);
await h.worker.drain();
for (const o of orders) assert.deepStrictEqual(o, ['critical', 'important', 'low']);
for (const s of stubs) await s.close();
ddiff --git a/test/lease.test.js b/test/lease.test.js
nnew file mode 100644
iindex 0000000..34a7e56
---- /dev/null
++++ b/test/lease.test.js
@@@ -0,0 +1,82 @@
++'use strict';
++// ADR-042 decision 3: deliveries are claimed with a database lease, so two workers never send one delivery twice and
++// keep at most one delivery in flight per subscription; an expired lease is claimed again (at least once), and the
++// late result of the worker that lost it is not recorded over the newer claim.
++const assert = require('assert');
++const { boot, request, serviceToken, envelope, subscriber, suite, sleep } = require('./helpers');
++const { createWorker } = require('../server/worker');
++
++const t = suite('lease');
++let h;
++const live = serviceToken('live', ['events.event.publish']);
++const silent = { log() {}, warn() {}, error() {} };
++
++async function subscribe(slug, topic, url) {
++ const r = await request(h.base, 'POST', '/api/v1/subscriptions', { token: serviceToken(slug, ['events.subscription.manage']), body: { topic_pattern: topic, endpoint: url } });
++ assert.strictEqual(r.status, 201, r.text);
++ return r.body;
++}
++async function publish(env) {
++ const r = await request(h.base, 'POST', '/api/v1/events', { token: live, body: env });
++ assert.ok(r.status === 201 || r.status === 200, r.text);
++}
++
++t('boot', async () => { h = await boot(); });
++
++t('two workers share the queue: every delivery exactly once, one in flight per subscription', async () => {
++ const sent = new Map();
++ const inFlight = new Map();
++ let perSubPeak = 0;
++ const stubs = [];
++ for (let i = 0; i < 4; i++) {
++ stubs.push(await subscriber(async (c) => 204));
++ }
++ const fetchImpl = async (url, init) => {
++ const key = `${init.headers['X-OpenVibe-Event-Id']}|${init.headers['X-OpenVibe-Subscription-Id']}`;
++ sent.set(key, (sent.get(key) || 0) + 1);
++ const sub = init.headers['X-OpenVibe-Subscription-Id'];
++ inFlight.set(sub, (inFlight.get(sub) || 0) + 1); perSubPeak = Math.max(perSubPeak, inFlight.get(sub));
++ await sleep(3);
++ inFlight.set(sub, inFlight.get(sub) - 1);
++ return new Response(null, { status: 204 });
++ };
++ for (let i = 0; i < 4; i++) await subscribe(`lease${i}`, 'live.lease.*', stubs[i].url);
++ for (let i = 0; i < 10; i++) await publish(envelope('live', { event_type: 'live.lease.x' }));
++ const a = createWorker({ store: h.store, config: h.config, clock: h.clock, fetchImpl, log: silent });
++ const b = createWorker({ store: h.store, config: h.config, clock: h.clock, fetchImpl, log: silent });
++ await Promise.all([a.drain(), b.drain()]);
++ assert.strictEqual(sent.size, 40, 'every (event, subscription) pair was delivered');
++ assert.ok([...sent.values()].every((n) => n === 1), `nothing was sent twice: ${JSON.stringify([...sent.entries()].filter(([, n]) => n > 1))}`);
++ assert.strictEqual(perSubPeak, 1, 'at most one delivery in flight per subscription across both workers');
++ for (const s of stubs) await s.close();
++ for (const sub of await h.store.listSubscriptions()) await h.store.setSubscriptionEnabled(sub.id, false);
++});
++
++t('an expired lease is claimed again; the late result of the lost claim is not recorded', async () => {
++ const stub = await subscriber(() => 204);
++ const sub = await subscribe('leaseexp', 'live.leaseexp.*', stub.url);
++ await publish(envelope('live', { event_type: 'live.leaseexp.x' }));
++ const now = h.clock.now();
++ const [first] = await h.store.claimDeliveries(now, 10, { owner: 'worker-a', leaseUntil: now + 1000 });
++ assert.ok(first, 'worker A claims it');
++ assert.deepStrictEqual(await h.store.claimDeliveries(now, 10, { owner: 'worker-b', leaseUntil: now + 1000 }), [], 'while A holds the lease, B gets nothing');
++ h.clock.advance(1001);
++ const [again] = await h.store.claimDeliveries(h.clock.now(), 10, { owner: 'worker-b', leaseUntil: h.clock.now() + 1000 });
++ assert.strictEqual(again && again.event_id, first.event_id, 'after the lease expires B claims the same delivery');
++ const late = await h.store.recordAttempt(first.event_id, first.subscription_id, { ok: false, attempt: 1, status: 500, error: 'late', nextAttemptAt: h.clock.now() }, null, 'worker-a');
++ assert.strictEqual(late, false, "A's late result is not recorded");
++ assert.strictEqual(await h.store.recordAttempt(again.event_id, again.subscription_id, { ok: true, attempt: 1, status: 204 }, null, 'worker-b'), true);
++ const d = await h.store.getDelivery(first.event_id, first.subscription_id);
++ assert.deepStrictEqual([d.status, d.lease_owner, d.lease_until], ['delivered', null, null], "B's result stands and the lease is released");
++ await stub.close();
++ await h.store.setSubscriptionEnabled(sub.id, false);
++});
++
++t('a claim with no room, or a lease that ends now, claims nothing', async () => {
++ const now = h.clock.now();
++ assert.deepStrictEqual(await h.store.claimDeliveries(now, 5, { owner: 'x', leaseUntil: now }), [], 'a lease that ends now claims nothing');
++ assert.deepStrictEqual(await h.store.claimDeliveries(now, 0, { owner: 'x', leaseUntil: now + 10 }), []);
++});
++
++t('stop', async () => { await h.stop(); });
++t.run();