openvibers/openvibe.events Commits
Delivery policies and per-key ordering (plan T7, ADR-042 decisions 1 and 4): server/fabric/policy.js resolves an event type's events.delivery-policy@1 (the most specific delivery_policies pattern, else the default: domain, durability required, at least once, no ordering, which is exactly today's behaviour), derives its carrier class (STREAM for required durability with an ordering key, TOPIC for realtime_ephemeral, else QUEUE; never in an envelope) and its ordering key (the policy's key resolved against the envelope, else the subject, else the event type; never the seq). Fan-out writes ordering_key on every delivery (a replay keeps the original's). The claim now leases only the head of a key (no lower-seq pending or failed delivery of it, nothing of it leased, so a failing delivery holds its key back) and at most subscriptions.max_inflight per subscription (default 1, as before), re-checking both in the leasing UPDATE (READ COMMITTED); app claims stay under maxApp. Admin API GET/PUT/DELETE /api/v1/delivery-policies[/:pattern] and /resolve (validated against the contract). Migration 0004: delivery_policies, subscriptions.max_inflight, the key index. test/ordering.test.js: policy resolution, order under retries, two workers with a lease expiry, max_inflight 2; 10/10 runs on real PostgreSQL. npm test and test:pg 20/20.
Changes
nnew file mode 100644@@@ -0,0 +1,16 @@++-- phase: expand++-- ADR-042 decisions 1 and 4 (plan T7): the delivery policy is the semantic vocabulary for how an event type is++-- carried; delivery_policies holds one row per event-type pattern (the same matcher subscriptions use). The carrier++-- class (TOPIC/QUEUE/STREAM) is derived, never stored and never in an envelope. ordering_key already exists (0002);++-- a partial index makes "the head of this key" cheap. max_inflight lets a subscription keep several keys in flight++-- (default 1 = today: one delivery in flight per subscription); never two of one key.++CREATE TABLE delivery_policies (++ pattern text COLLATE "C" PRIMARY KEY,++ policy jsonb NOT NULL,++ revision integer NOT NULL DEFAULT 1,++ updated_at bigint NOT NULL,++ updated_by text COLLATE "C"++);++ALTER TABLE subscriptions ADD COLUMN max_inflight integer NOT NULL DEFAULT 1 CHECK (max_inflight BETWEEN 1 AND 16);++CREATE INDEX idx_deliveries_ordering ON deliveries(subscription_id, ordering_key, seq)++ WHERE ordering_key IS NOT NULL AND status IN ('pending', 'failed');nnew file mode 100644@@@ -0,0 +1,77 @@++'use strict';++/**++ * Delivery policies (ADR-042 decisions 1 and 4; plan T7) — operators only (events.delivery.admin), behind the same++ * isAdmin guard and per-actor limits as /api/v1/deliveries.++ *++ * GET /api/v1/delivery-policies -> { policies: [{ pattern, policy, revision, updated_at, updated_by, carrier_class }] }++ * PUT /api/v1/delivery-policies/:pattern body = events.delivery-policy@1 -> the stored row (revision +1)++ * DELETE /api/v1/delivery-policies/:pattern -> 204 (404 when absent)++ * GET /api/v1/delivery-policies/resolve?event_type=<t> -> { event_type, pattern, revision, policy, carrier_class }++ *++ * The pattern is validated with the subscriptions' topic-pattern rule; the body with the contracts validator (an++ * unknown class or an extra field is a 400 carrying the validator's message). carrier_class is the derived label++ * (./../fabric/policy.js); it is answered here and nowhere else.++ */++const express = require('express');++const { http, validate } = require('openvibe-contracts');++const topics = require('../topics');++const { CAPS } = require('../auth');++const { carrierClass } = require('../fabric/policy');++++function policiesRouter({ policies, auth, limits }) {++ const router = express.Router();++ const isAdmin = auth.requireCap(CAPS.admin);++ // Writes are rare operator actions: 30 a minute (like subscription changes); reads take the defaults.++ const change = { minute: 30, hour: 300 };++++ const view = (row) => ({++ pattern: row.pattern,++ policy: row.policy,++ revision: row.revision,++ updated_at: new Date(row.updated_at).toISOString(),++ updated_by: row.updated_by || null,++ carrier_class: carrierClass(row.policy),++ });++++ router.get('/api/v1/delivery-policies', isAdmin, limits('events.delivery-policy.list'), async (_req, res) => {++ res.json({ policies: (await policies.all()).map(view) });++ });++++ router.get('/api/v1/delivery-policies/resolve', isAdmin, limits('events.delivery-policy.read'), async (req, res) => {++ const eventType = String(req.query.event_type || '');++ if (!eventType || eventType.length > 200) {++ return http.sendProblem(res, 400, 'events.bad_request', { detail: 'event_type is required and at most 200 characters', ctx: req.ov });++ }++ const { policy, pattern, revision } = await policies.resolve(eventType);++ res.json({ event_type: eventType, pattern, revision, policy, carrier_class: carrierClass(policy) });++ });++++ router.put('/api/v1/delivery-policies/:pattern', isAdmin, limits('events.delivery-policy.write', change), async (req, res) => {++ const ctx = req.ov;++ const pattern = String(req.params.pattern);++ if (!topics.isValidPattern(pattern)) {++ return http.sendProblem(res, 422, 'events.bad_topic', { detail: 'pattern must be dot-separated segments of [a-z0-9_] or *', ctx });++ }++ const body = req.body && typeof req.body === 'object' && !Array.isArray(req.body) ? req.body : null;++ if (!body) return http.sendProblem(res, 400, 'events.bad_request', { detail: 'body must be an events.delivery-policy@1 object', ctx });++ const v = validate('events.delivery-policy@1', body);++ if (!v.valid) {++ return http.sendProblem(res, 400, 'events.bad_request', { detail: v.errors.map((e) => `${e.path} ${e.message}`).join('; '), ctx });++ }++ const row = await policies.put(pattern, body, req.principal.service || req.principal.sub);++ res.json(view(row));++ });++++ router.delete('/api/v1/delivery-policies/:pattern', isAdmin, limits('events.delivery-policy.write', change), async (req, res) => {++ const pattern = String(req.params.pattern);++ if (!topics.isValidPattern(pattern)) {++ return http.sendProblem(res, 422, 'events.bad_topic', { detail: 'pattern must be dot-separated segments of [a-z0-9_] or *', ctx: req.ov });++ }++ if (!await policies.remove(pattern)) return http.sendProblem(res, 404, 'events.not_found', { detail: 'no such delivery policy', ctx: req.ov });++ res.status(204).end();++ });++++ return router;++}++++module.exports = { policiesRouter };@@@ -9,11 +9,12 @@ const { http } = require('openvibe-contracts');const { publishRouter } = require('./api/publish');const { subscriptionsRouter } = require('./api/subscriptions');const { readRouter } = require('./api/read');++const { policiesRouter } = require('./api/policies');const { mountLimits } = require('./limits');const { createLimits } = require('./actor-limits');const pkg = require('../package.json');--function createApp({ config, store, auth, keys, worker, realtime, metrics, dnsLookup, clock, log = console, valkey = null }) {++function createApp({ config, store, auth, keys, worker, realtime, metrics, dnsLookup, clock, log = console, valkey = null, policies = null }) {const app = express();app.disable('x-powered-by');app.set('trust proxy', 'loopback');@@@ -78,6 +79,7 @@ function createApp({ config, store, auth, keys, worker, realtime, metrics, dnsLoapp.use(publishRouter({ config, store, auth, worker, realtime, limits }));app.use(subscriptionsRouter({ config, store, auth, dnsLookup, limits }));app.use(readRouter({ store, auth, worker, limits }));++ if (policies) app.use(policiesRouter({ policies, auth, limits }));app.get('/', (_req, res) => {res.type('text/plain').send([nnew file mode 100644@@@ -0,0 +1,160 @@++'use strict';++/**++ * The delivery policy (ADR-042 decisions 1 and 4; plan T7): the only semantic vocabulary for how an event type is++ * carried. Policies live in `delivery_policies`, keyed by an event-type pattern using the same matcher subscriptions++ * use for `topic_pattern` (../topics.js). The most specific matching pattern wins: an exact type beats any wildcard;++ * among wildcards the longer literal part wins; ties break by pattern string order (deterministic). With no policy row++ * behaviour is exactly today's DEFAULT_POLICY: no ordering key, one delivery in flight per subscription.++ *++ * The carrier class (TOPIC | QUEUE | STREAM) is an internal, derived label — never a contract enum, never in an++ * envelope. It is answered by the explain/resolve API only.++ *++ * Publisher intent: the publish contract (events.publish-request@1 -> events.event-envelope@1) has no intent field, so++ * none is read here; a class can only be raised by writing a policy. Do not add a field for it.++ */++const topics = require('../topics');++++const DEFAULT_POLICY = { class: 'domain', durability: 'required', delivery_semantics: 'at_least_once' };++const MAX_KEY_LEN = 200;++++/**++ * Which carrier class a policy's deliveries use. STREAM only when durability is required AND the policy has an++ * ordering key (retained, ordered per key); TOPIC for the fan-out class; QUEUE for everything else.++ */++function carrierClass(policy) {++ if (policy && policy.durability === 'required' && policy.ordering && typeof policy.ordering.key === 'string' && policy.ordering.key) return 'STREAM';++ if (policy && policy.class === 'realtime_ephemeral') return 'TOPIC';++ return 'QUEUE';++}++++function subjectForm(envelope) {++ const s = envelope && envelope.subject;++ if (s && typeof s.type === 'string' && s.type && typeof s.id === 'string' && s.id) return `${s.type}:${s.id}`;++ return null;++}++++/** The value at a dotted path (`subject.id`, `payload.a.b`, `actor.id`) or null when it is not there. */++function pathValue(envelope, path) {++ let cur = envelope;++ for (const seg of String(path).split('.')) {++ if (cur == null || typeof cur !== 'object') return null;++ cur = cur[seg];++ }++ return cur;++}++++/**++ * The ordering key of an envelope under a policy, or null when the policy has no ordering (the default). `'subject'`++ * is `<subject.type>:<subject.id>`; a dotted path starting with subject./actor./payload. is that value when it is a++ * string or a finite number; anything unresolvable falls back to the subject form, then to the event type. Capped at++ * 200 characters. Never the seq.++ */++function orderingKey(policy, envelope) {++ if (!policy || !policy.ordering || typeof policy.ordering.key !== 'string' || !policy.ordering.key) return null;++ const key = policy.ordering.key;++ let value = null;++ if (key === 'subject') {++ value = subjectForm(envelope);++ } else if (/^(subject|actor|payload)\./.test(key)) {++ const raw = pathValue(envelope, key);++ if (typeof raw === 'string' || (typeof raw === 'number' && Number.isFinite(raw))) value = String(raw);++ }++ if (value == null || value === '') value = subjectForm(envelope) || (envelope && envelope.event_type) || null;++ if (value == null) return null;++ return value.length > MAX_KEY_LEN ? value.slice(0, MAX_KEY_LEN) : value;++}++++/** Exact patterns beat wildcards; among wildcards the longer literal part wins. */++function specificity(pattern) {++ return pattern.includes('*') ? pattern.replace(/\*/g, '').length : Infinity;++}++++function bestMatch(rows, eventType) {++ let best = null;++ for (const row of rows) {++ if (!row.compiled.test(eventType)) continue;++ if (!best) { best = row; continue; }++ const a = specificity(row.pattern);++ const b = specificity(best.pattern);++ if (a > b || (a === b && row.pattern < best.pattern)) best = row;++ }++ return best;++}++++function normalize(row) {++ const policy = typeof row.policy === 'string' ? JSON.parse(row.policy) : row.policy;++ return { pattern: row.pattern, policy, revision: Number(row.revision), updated_at: row.updated_at, updated_by: row.updated_by || null };++}++++/**++ * @param {object} db the Events database (openvibe-sdk/db handle)++ * @param {{ now(): number }} [o.clock]++ * @param {number} [o.reloadMs] how long a compiled cache is trusted before it is dropped (a second worker process++ * picks up another process's writes; this process also drops it on every write)++ */++function createPolicyLibrary(db, { clock = { now: () => Date.now() }, reloadMs = 60000, log = console } = {}) {++ let cache = null; // { rows, at }++++ async function load() {++ let raw = [];++ try {++ raw = await db.prepare('SELECT pattern, policy, revision, updated_at, updated_by FROM delivery_policies').all();++ } catch (err) {++ log.warn(`[policy] could not load delivery_policies: ${err.message}`);++ raw = [];++ }++ const rows = raw.map((r) => {++ const row = normalize(r);++ row.compiled = topics.compile(row.pattern);++ return row;++ });++ cache = { rows, at: clock.now() };++ return rows;++ }++++ /** Drop the compiled cache; the next resolve reloads from the database. */++ function invalidate() { cache = null; }++++ async function all() {++ if (!cache || clock.now() - cache.at >= reloadMs) await load();++ return cache.rows;++ }++++ /** { policy, pattern, revision }; pattern is null and revision 0 for the default (no matching row). */++ async function resolve(eventType) {++ const rows = await all();++ const m = bestMatch(rows, eventType);++ return m ? { policy: m.policy, pattern: m.pattern, revision: m.revision } : { policy: DEFAULT_POLICY, pattern: null, revision: 0 };++ }++++ /** The ordering key an envelope of `eventType` takes, or null (no policy ordering). */++ async function keyFor(eventType, envelope) {++ const { policy } = await resolve(eventType);++ return orderingKey(policy, envelope);++ }++++ /** Upsert a policy row; revision +1 on an existing pattern. Returns the stored row. */++ async function put(pattern, policy, updatedBy) {++ const row = await db.prepare(`INSERT INTO delivery_policies (pattern, policy, revision, updated_at, updated_by)++ VALUES (?, ?, 1, ?, ?)++ ON CONFLICT(pattern) DO UPDATE SET policy = excluded.policy, revision = delivery_policies.revision + 1,++ updated_at = excluded.updated_at, updated_by = excluded.updated_by++ RETURNING pattern, policy, revision, updated_at, updated_by`).get(pattern, JSON.stringify(policy), clock.now(), updatedBy);++ invalidate();++ return normalize(row);++ }++++ /** Remove a policy row. Returns the number of rows deleted. */++ async function remove(pattern) {++ const changes = (await db.prepare('DELETE FROM delivery_policies WHERE pattern = ?').run(pattern)).changes;++ invalidate();++ return changes;++ }++++ const timer = setInterval(() => { invalidate(); }, reloadMs);++ timer.unref?.();++++ return { resolve, keyFor, all, put, remove, invalidate, reload: load, stop() { clearInterval(timer); } };++}++++module.exports = { DEFAULT_POLICY, carrierClass, orderingKey, specificity, createPolicyLibrary };@@@ -9,6 +9,7 @@*/const { load } = require('./config');const { openDb, createStore } = require('./store');++const { createPolicyLibrary } = require('./fabric/policy');const { createKeyStore, createAuth } = require('./auth');const { createWorker } = require('./worker');const { createRealtime } = require('./realtime');@@@ -23,7 +24,10 @@ async function start({config = config || load();// PostgreSQL (ADR-035): opened and migrated here unless the caller (a test) hands in a migrated handle.const db = givenDb || await openDb(config, { log });-- const store = createStore(db, { clock, maxHops: config.maxHops, usage: config.usage });++ // Delivery policies (ADR-042 decisions 1 and 4): the store reads the resolved ordering key on fan-out, the admin++ // API writes rows and drops the compiled cache. One library per process, shared by both.++ const policies = createPolicyLibrary(db, { clock, log });++ const store = createStore(db, { clock, maxHops: config.maxHops, usage: config.usage, policies });const keys = createKeyStore({ urls: [config.networkInternalUrl, config.networkUrl], pem: config.networkPublicKey, fetchImpl, log });const auth = createAuth({ config, keys, store });const metrics = createMetrics({ store });@@@ -33,7 +37,7 @@ async function start({const realtime = createRealtime({ store, auth, config, clock, log });metrics.bind({ realtime });const valkey = config.valkey.url ? require('openvibe-sdk/valkey').createValkey({ url: config.valkey.url, prefix: config.valkey.prefix, log }) : null;-- const app = createApp({ config, store, auth, keys, worker, realtime, metrics, dnsLookup, clock, log, valkey });++ const app = createApp({ config, store, auth, keys, worker, realtime, metrics, dnsLookup, clock, log, valkey, policies });// The key loads in the background (retrying while Network boots); /api/ready says when it has.const keyLoaded = keys.start().catch(() => null);@@@ -86,6 +90,7 @@ async function start({async function close() {clearInterval(pruneTimer);if (usageTimer) clearInterval(usageTimer);++ policies.stop();realtime.stop();keys.stop();await worker.stop();@@@ -97,7 +102,7 @@ async function start({if (!givenDb) await db.close();}-- return { config, db, store, keys, keyLoaded, auth, worker, realtime, metrics, app, server, close, prune, flushUsage };++ return { config, db, store, keys, keyLoaded, auth, worker, realtime, metrics, app, server, policies, close, prune, flushUsage };}if (require.main === module) {@@@ -91,7 +91,7 @@ function subscriptionView(row, { withSecret = false } = {}) {return out;}--function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage: usageConfig = { enabled: true } } = {}) {++function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage: usageConfig = { enabled: true }, policies = null } = {}) {// 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 = {@@@ -130,7 +130,7 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usageON CONFLICT(consumer, scope) DO UPDATE SET revoked_at = GREATEST(app_revocations.revoked_at, excluded.revoked_at)`),revokedAt: db.prepare('SELECT revoked_at FROM app_revocations WHERE consumer = ? AND scope = ?'),insertDelivery: db.prepare(`INSERT INTO deliveries (event_id, subscription_id, seq, priority, attempt, status,-- next_attempt_at, created_at, updated_at) VALUES (?, ?, ?, ?, 0, 'pending', ?, ?, ?) ON CONFLICT DO NOTHING`),++ next_attempt_at, ordering_key, created_at, updated_at) VALUES (?, ?, ?, ?, 0, 'pending', ?, ?, ?, ?) ON CONFLICT DO NOTHING`),afterSeq: db.prepare('SELECT * FROM events WHERE seq > ? ORDER BY seq LIMIT ?'),afterSeqLike: db.prepare("SELECT * FROM events WHERE seq > ? AND event_type ILIKE ? ESCAPE '\\' ORDER BY seq LIMIT ?"),};@@@ -229,9 +229,13 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usageif (directive) row.redacts = 1;await q.insertEvent.run(row);await q.addReceipt.run(PUBLISH_RECEIPT, env.event_id, now);++ // The ordering key is a property of the event type (ADR-042 decision 4): resolve it once per event and give++ // every delivery of it the same key. No policy row (the default) resolves to null and behaves as before.++ // Publisher intent is not read: the publish contract has no intent field.++ const orderingKey = policies ? await policies.keyFor(env.event_type, env) : null;for (const s of subs) {if (apps.deliverable(s, row)) {-- await q.insertDelivery.run(env.event_id, s.id, seq, PRIORITY_RANK[env.priority], now, now, now);++ await q.insertDelivery.run(env.event_id, s.id, seq, PRIORITY_RANK[env.priority], now, orderingKey, now, now);}}await revocationHook(row, now);@@@ -422,47 +426,72 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage// ── Deliveries ───────────────────────────────────────────/**-- * 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.++ * Claim up to `limit` due deliveries for `owner` until `leaseUntil` (ms): the best claimable delivery (by priority++ * class, then seq) of each subscription that has room. A delivery is claimable when it is due and (ADR-042++ * decision 4) it is the head of its ordering key — no lower-seq pending/failed row of the same (subscription, key)++ * and nothing of that key leased — or it has no key at all. At most `subscriptions.max_inflight` deliveries of a++ * subscription are leased at once (default 1 = today); with more, several keys of one subscription run together,++ * never two of one key. 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++ * until recordAttempt clears it or it expires (an expired lease is claimable again: delivery is at least once).++ * `maxApp` caps developer-app subscriptions.++ *++ * The leasing UPDATE re-checks everything in its own snapshot (READ COMMITTED does not re-evaluate the candidate++ * query's predicates when the locked row itself did not change): the head condition, that nothing of the key is++ * leased, and that the subscription is under its inflight cap. A claim counts only when it changed exactly one row.*/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).++ const claimable = (d) => `(${d}.status IN ('pending', 'failed') AND ${d}.next_attempt_at <= ?++ AND (${d}.ordering_key IS NULL OR (++ 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 NOT EXISTS (SELECT 1 FROM deliveries y WHERE y.subscription_id = ${d}.subscription_id++ AND y.ordering_key = ${d}.ordering_key AND y.lease_until > ?)++ )))`;++ // One binding: `now`, the lease expiry to compare against.++ const leased = (sub) => `(SELECT COUNT(*) FROM deliveries x WHERE x.subscription_id = ${sub} AND x.lease_until > ?)`;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.++ // 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 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++ (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 sWHERE 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 > ?)++ AND 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, n));++ FOR UPDATE OF s SKIP LOCKED`).all(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 = [];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).++ WHERE d.subscription_id = ? AND ${claimable('d')} AND ${leased('d.subscription_id')} < s.max_inflight++ ORDER BY d.priority, d.seq LIMIT 1`);++ // The lease is taken only if it still holds as of NOW (see the method comment): claimable, under the cap,++ // and of a key nothing else is leased on.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 > ?)`);++ AND ${claimable('d')}++ AND ${leased('d.subscription_id')} < (SELECT s.max_inflight FROM subscriptions s WHERE s.id = d.subscription_id)`);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++;++ // 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 (;;) {++ if (picked.length >= limit || (sub.project_id && apps >= maxApp)) break;++ const d = await next.get(sub.id, 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++ picked.push({ ...d, lease_owner: owner, lease_until: leaseUntil });++ if (sub.project_id) apps++;++ }}return picked;});@@@ -525,7 +554,7 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usageconst requeue = async (sub, { fromSeq, eventIds, max = 10000 }) => await db.tx(async () => {const now = clock.now();let rows;-- const cols = 'id, seq, priority, event_type, visibility, project_id, env';++ const cols = 'id, seq, priority, event_type, visibility, project_id, env, subject_type, subject_id, actor, payload';if (Array.isArray(eventIds)) {const get = db.prepare(`SELECT ${cols} FROM events WHERE id = ?`);rows = (await Promise.all(eventIds.map(async id => await get.get(id)))).filter(Boolean);@@@ -535,11 +564,17 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage}// Replay never widens a subscription's scope (app scope, sandbox separation).rows = rows.filter(r => apps.deliverable(sub, r)).slice(0, max);-- const up = db.prepare(`INSERT INTO deliveries (event_id, subscription_id, seq, priority, attempt, status, next_attempt_at, created_at, updated_at)-- VALUES (?, ?, ?, ?, 0, 'pending', ?, ?, ?)++ // ordering_key is written on an insert only: a row that already exists keeps the key its first fan-out gave it.++ const up = db.prepare(`INSERT INTO deliveries (event_id, subscription_id, seq, priority, attempt, status, next_attempt_at, ordering_key, created_at, updated_at)++ VALUES (?, ?, ?, ?, 0, 'pending', ?, ?, ?, ?)ON CONFLICT(event_id, subscription_id) DO UPDATE SET status = 'pending', attempt = 0, next_attempt_at = excluded.next_attempt_at,last_error = NULL, last_status = NULL, delivered_at = NULL, updated_at = excluded.updated_at`);-- for (const r of rows) await up.run(r.id, sub.id, r.seq, PRIORITY_RANK[r.priority], now, now, now);++ for (const r of rows) {++ const key = policies ? await policies.keyFor(r.event_type, {++ event_type: r.event_type, subject: { type: r.subject_type, id: r.subject_id }, actor: JSON.parse(r.actor), payload: JSON.parse(r.payload),++ }) : null;++ await up.run(r.id, sub.id, r.seq, PRIORITY_RANK[r.priority], now, key, now, now);++ }return rows.length;});nnew file mode 100644@@@ -0,0 +1,239 @@++'use strict';++// ADR-042 decisions 1 and 4 (plan T7): the delivery policy is the semantic vocabulary (class -> derived carrier++// class, ordering key resolved against the envelope) and per-key ordering is enforced in the claim: a key's head is++// claimable, its successors wait for it, and a subscription keeps at most max_inflight deliveries leased (one key in++// flight by default). Concurrency bugs show only on real PostgreSQL — npm run test:pg runs this file there.++const assert = require('assert');++const { boot, request, serviceToken, envelope, subscriber, suite, sleep } = require('./helpers');++const { createWorker } = require('../server/worker');++const { DEFAULT_POLICY, carrierClass, orderingKey } = require('../server/fabric/policy');++++const t = suite('ordering');++let h;++const admin = serviceToken('ops', ['events.delivery.admin']);++const reader = serviceToken('media', ['events.event.read']);++const live = serviceToken('live', ['events.event.publish']);++const silent = { log() {}, warn() {}, error() {} };++++const CLASSES = ['realtime_ephemeral', 'interactive', 'interactive_durable', 'domain', 'critical', 'background', 'bulk', 'webhook'];++const policy = (over = {}) => ({ class: 'domain', durability: 'required', delivery_semantics: 'at_least_once', ...over });++const keyed = (key, scope = 'key') => policy({ ordering: { scope, key } });++++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);++}++async function putPolicy(pattern, body) {++ const r = await request(h.base, 'PUT', `/api/v1/delivery-policies/${pattern}`, { token: admin, body });++ assert.strictEqual(r.status, 200, r.text);++ return r.body;++}++const deletePolicy = (pattern) => request(h.base, 'DELETE', `/api/v1/delivery-policies/${pattern}`, { token: admin });++++t('boot', async () => { h = await boot(); });++++t('policy: default, carrier classes, key resolution and the admin API', async () => {++ // Every class maps to exactly one carrier class; durability required + an ordering key is STREAM for all of them.++ for (const c of CLASSES) {++ assert.strictEqual(carrierClass(policy({ class: c })), c === 'realtime_ephemeral' ? 'TOPIC' : 'QUEUE');++ assert.strictEqual(carrierClass(keyed('subject', 'key', c)), 'STREAM', `${c}: required + ordering is STREAM`);++ assert.strictEqual(carrierClass({ class: c, durability: 'optional', delivery_semantics: 'at_least_once', ordering: { scope: 'key', key: 'subject' } }), c === 'realtime_ephemeral' ? 'TOPIC' : 'QUEUE', `${c}: ordering without required durability is not STREAM`);++ }++ // Key resolution: 'subject', dotted paths, fallbacks, the 200-char cap, never the seq; no ordering -> null.++ const env = envelope('live', { subject: { type: 'stream', id: 'str_9' }, actor: { type: 'service', id: 'live' }, payload: { room: 42, empty: '', deep: { x: 'y' } } });++ assert.strictEqual(orderingKey(keyed('subject'), env), 'stream:str_9');++ assert.strictEqual(orderingKey(keyed('payload.room'), env), '42');++ assert.strictEqual(orderingKey(keyed('actor.id'), env), 'live');++ assert.strictEqual(orderingKey(keyed('payload.deep.x'), env), 'y');++ assert.strictEqual(orderingKey(keyed('payload.empty'), env), 'stream:str_9', 'an empty/unresolvable path falls back to the subject');++ assert.strictEqual(orderingKey(keyed('payload.missing'), env), 'stream:str_9');++ assert.strictEqual(orderingKey(keyed('payload.missing'), { event_type: 'live.x.y' }), 'live.x.y', 'no subject: falls back to the event type');++ assert.strictEqual(orderingKey(keyed('payload.big'), { event_type: 'live.x', payload: { big: 'a'.repeat(500) } }).length, 200, 'the key is capped at 200 characters');++ assert.strictEqual(orderingKey(DEFAULT_POLICY, env), null, 'the default policy has no ordering');++ assert.strictEqual(orderingKey({ ordering: { scope: 'key', key: 'paylaod.typo' } }, env), 'stream:str_9', 'a path that is not subject./actor./payload. falls back to the subject');++++ // No rows: the default, and an empty list. Resolve answers the default shape.++ assert.deepStrictEqual(await h.policies.resolve('live.anything'), { policy: DEFAULT_POLICY, pattern: null, revision: 0 });++ const empty = await request(h.base, 'GET', '/api/v1/delivery-policies', { token: admin });++ assert.strictEqual(empty.status, 200);++ assert.deepStrictEqual(empty.body.policies, []);++++ // PUT: revision starts at 1, +1 on a rewrite; updated_by is the admin principal; carrier_class is derived.++ const exact = await putPolicy('live.order.exact', keyed('subject'));++ assert.strictEqual(exact.revision, 1);++ assert.strictEqual(exact.carrier_class, 'STREAM');++ assert.strictEqual(exact.updated_by, 'ops');++ await putPolicy('live.order.*', keyed('subject'));++ await putPolicy('live.*', keyed('subject'));++ const rewrite = await putPolicy('live.order.exact', { class: 'critical', durability: 'none', delivery_semantics: 'at_most_once' });++ assert.strictEqual(rewrite.revision, 2);++ assert.strictEqual(rewrite.carrier_class, 'QUEUE');++++ // Specificity: the exact type beats any wildcard; the longer literal wildcard beats the shorter one.++ const rExact = await request(h.base, 'GET', '/api/v1/delivery-policies/resolve?event_type=live.order.exact', { token: admin });++ assert.deepStrictEqual([rExact.status, rExact.body.pattern, rExact.body.revision, rExact.body.carrier_class], [200, 'live.order.exact', 2, 'QUEUE']);++ const rWild = await request(h.base, 'GET', '/api/v1/delivery-policies/resolve?event_type=live.order.other', { token: admin });++ assert.deepStrictEqual([rWild.body.pattern, rWild.body.carrier_class], ['live.order.*', 'STREAM']);++ const rShort = await request(h.base, 'GET', '/api/v1/delivery-policies/resolve?event_type=live.elsewhere.x', { token: admin });++ assert.strictEqual(rShort.body.pattern, 'live.*', 'the shorter wildcard is the only match');++++ // Ties (equal literal length) break by pattern string order: 'a.*.c.d' beats 'a.b.*.d'.++ await putPolicy('a.b.*.d', keyed('subject'));++ await putPolicy('a.*.c.d', keyed('subject'));++ const tie = await request(h.base, 'GET', '/api/v1/delivery-policies/resolve?event_type=a.b.c.d', { token: admin });++ assert.strictEqual(tie.body.pattern, 'a.*.c.d', 'ties break deterministically by pattern string order');++++ // The resolve endpoint answers the default for an unknown type.++ const none = await request(h.base, 'GET', '/api/v1/delivery-policies/resolve?event_type=nobody.home', { token: admin });++ assert.deepStrictEqual([none.body.pattern, none.body.revision, none.body.carrier_class, none.body.policy], [null, 0, 'QUEUE', DEFAULT_POLICY]);++++ // The validator refuses an unknown class and an extra field with its own message.++ const badClass = await request(h.base, 'PUT', '/api/v1/delivery-policies/live.bad', { token: admin, body: { class: 'warp_speed', durability: 'required', delivery_semantics: 'at_least_once' } });++ assert.strictEqual(badClass.status, 400, badClass.text);++ assert.ok(/class/i.test(badClass.body.detail), badClass.body.detail);++ const badField = await request(h.base, 'PUT', '/api/v1/delivery-policies/live.bad', { token: admin, body: { ...policy(), mode: 'fast' } });++ assert.strictEqual(badField.status, 400, badField.text);++++ // Non-admin: the same answers as the other admin routes (403 for a service without the capability, 401 without a token).++ assert.strictEqual((await request(h.base, 'GET', '/api/v1/delivery-policies', { token: reader })).status, 403);++ assert.strictEqual((await request(h.base, 'GET', '/api/v1/delivery-policies')).status, 401);++ assert.strictEqual((await request(h.base, 'PUT', '/api/v1/delivery-policies/live.bad', { token: reader, body: policy() })).status, 403);++++ // DELETE: 404 when absent, 204 then again 404.++ assert.strictEqual((await deletePolicy('live.bad')).status, 404);++ assert.strictEqual((await deletePolicy('live.order.exact')).status, 204);++ assert.strictEqual((await deletePolicy('live.order.exact')).status, 404);++++ for (const p of ['live.order.*', 'live.*', 'a.b.*.d', 'a.*.c.d']) assert.strictEqual((await deletePolicy(p)).status, 204);++});++++t('per-key order under retries: a successor waits for its head, another key goes meanwhile', async () => {++ const stub = await subscriber(() => 204);++ const sub = await subscribe('ord2', 'live.ord2.*', stub.url);++ await putPolicy('live.ord2.*', keyed('subject'));++ const A1 = envelope('live', { event_type: 'live.ord2.x', subject: { type: 'stream', id: 'A' } });++ const A2 = envelope('live', { event_type: 'live.ord2.x', subject: { type: 'stream', id: 'A' } });++ const B1 = envelope('live', { event_type: 'live.ord2.x', subject: { type: 'stream', id: 'B' } });++ await publish(A1); await publish(A2); await publish(B1);++++ let failA1 = true;++ const attempts = [];++ const fetchImpl = async (url, init) => {++ const ev = JSON.parse(init.body).event;++ let status = 204;++ if (ev.event_id === A1.event_id && failA1) { failA1 = false; status = 500; }++ attempts.push({ id: ev.event_id, status });++ return new Response(null, { status });++ };++ const w = createWorker({ store: h.store, config: h.config, clock: h.clock, fetchImpl, log: silent });++ await w.drain();++ const ids = () => attempts.map((a) => a.id);++ assert.deepStrictEqual(ids(), [A1.event_id, B1.event_id], 'A1 failed and B1 went; A2 (a successor of A1) did not');++ assert.strictEqual(attempts[0].status, 500);++++ h.clock.advance(1001); // A1's backoff elapses++ await w.drain();++ assert.deepStrictEqual(ids(), [A1.event_id, B1.event_id, A1.event_id, A2.event_id], 'A1 is retried and only then A2');++ assert.strictEqual(attempts.filter((a) => a.id === A1.event_id).length, 2);++ assert.strictEqual(attempts.filter((a) => a.id === A2.event_id).length, 1);++++ await deletePolicy('live.ord2.*');++ await stub.close();++ await h.store.setSubscriptionEnabled(sub.id, false);++});++++t('per-key order across two workers with a lease expiry, and never two distinct seqs of one key in flight', async () => {++ const stub = await subscriber(() => 204);++ const sub = await subscribe('ord3', 'live.ord3.*', stub.url);++ await putPolicy('live.ord3.*', keyed('subject'));++++ const keys = ['K0', 'K1', 'K2'];++ const byId = new Map(); // event_id -> deliveries++ const firstSeqs = new Map(keys.map((k) => [k, []])); // key -> seqs in first-delivery order++ const inflight = new Map(); // key -> Set(seq)++ let distinctPeak = 0; // the most distinct seqs of one key ever in flight at once++ const fetchImpl = async (url, init) => {++ const { event, seq } = JSON.parse(init.body);++ const k = event.subject.id;++ byId.set(event.event_id, (byId.get(event.event_id) || 0) + 1);++ const set = inflight.get(k) || new Set();++ inflight.set(k, set);++ set.add(seq);++ distinctPeak = Math.max(distinctPeak, set.size);++ await sleep(2);++ set.delete(seq);++ const list = firstSeqs.get(k);++ if (!list.includes(seq)) list.push(seq);++ return new Response(null, { status: 204 });++ };++++ // 30 events over 3 keys, interleaved so the global seq order is not the per-key order.++ for (let i = 0; i < 10; i++) {++ for (const k of keys) await publish(envelope('live', { event_type: 'live.ord3.x', subject: { type: 'stream', id: k } }));++ }++++ 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 });++ const pa = a.drain();++ await sleep(4);++ h.clock.advance(40001); // whatever worker A holds now has an expired lease: worker B may re-claim it (at least once)++ const pb = b.drain();++ await Promise.all([pa, pb]);++ await a.drain(); await b.drain();++++ assert.strictEqual(byId.size, 30, 'every event of every key was delivered');++ assert.ok([...byId.values()].every((n) => n >= 1), 'every event delivered at least once');++ for (const k of keys) {++ const list = firstSeqs.get(k);++ assert.deepStrictEqual(list, [...list].sort((x, y) => x - y), `${k}: the first delivery of each seq is in seq order (${list})`);++ }++ assert.strictEqual(distinctPeak, 1, 'never two distinct seqs of one key in flight at once');++++ await deletePolicy('live.ord3.*');++ await stub.close();++ await h.store.setSubscriptionEnabled(sub.id, false);++});++++t('max_inflight = 2: two keys run together, never two of one key, never three at once', async () => {++ const stub = await subscriber(() => 204);++ const sub = await subscribe('ord4', 'live.ord4.*', stub.url);++ await putPolicy('live.ord4.*', keyed('subject'));++ await h.store.db.prepare('UPDATE subscriptions SET max_inflight = 2 WHERE id = ?').run(sub.id);++++ const inflight = []; // { key, seq } currently being sent++ let totalPeak = 0;++ let keyPeak = 0;++ const fetchImpl = async (url, init) => {++ const { event, seq } = JSON.parse(init.body);++ const k = event.subject.id;++ inflight.push({ k, seq });++ totalPeak = Math.max(totalPeak, inflight.length);++ keyPeak = Math.max(keyPeak, inflight.filter((x) => x.k === k).length);++ await sleep(4);++ inflight.splice(inflight.findIndex((x) => x.k === k && x.seq === seq), 1);++ return new Response(null, { status: 204 });++ };++++ for (let i = 0; i < 2; i++) {++ for (const k of ['K0', 'K1']) await publish(envelope('live', { event_type: 'live.ord4.x', subject: { type: 'stream', id: k } }));++ }++ const w = createWorker({ store: h.store, config: h.config, clock: h.clock, fetchImpl, log: silent });++ await w.drain();++++ 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((await h.store.deliveryCounts()).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.*');++ await stub.close();++ await h.store.setSubscriptionEnabled(sub.id, false);++});++++t('stop', async () => { await h.stop(); });++t.run();
new file mode 100644
new file mode 100644
@@ -0,0 +1,16 @@
@@ -0,0 +1,16 @@
+-- phase: expand
+-- ADR-042 decisions 1 and 4 (plan T7): the delivery policy is the semantic vocabulary for how an event type is
+-- carried; delivery_policies holds one row per event-type pattern (the same matcher subscriptions use). The carrier
+-- class (TOPIC/QUEUE/STREAM) is derived, never stored and never in an envelope. ordering_key already exists (0002);
+-- a partial index makes "the head of this key" cheap. max_inflight lets a subscription keep several keys in flight
+-- (default 1 = today: one delivery in flight per subscription); never two of one key.
+CREATE TABLE delivery_policies (
+ pattern text COLLATE "C" PRIMARY KEY,
+ policy jsonb NOT NULL,
+ revision integer NOT NULL DEFAULT 1,
+ updated_at bigint NOT NULL,
+ updated_by text COLLATE "C"
+);
+ALTER TABLE subscriptions ADD COLUMN max_inflight integer NOT NULL DEFAULT 1 CHECK (max_inflight BETWEEN 1 AND 16);
+CREATE INDEX idx_deliveries_ordering ON deliveries(subscription_id, ordering_key, seq)
+ WHERE ordering_key IS NOT NULL AND status IN ('pending', 'failed');
new file mode 100644
new file mode 100644
@@ -0,0 +1,77 @@
@@ -0,0 +1,77 @@
+'use strict';
+/**
+ * Delivery policies (ADR-042 decisions 1 and 4; plan T7) — operators only (events.delivery.admin), behind the same
+ * isAdmin guard and per-actor limits as /api/v1/deliveries.
+ *
+ * GET /api/v1/delivery-policies -> { policies: [{ pattern, policy, revision, updated_at, updated_by, carrier_class }] }
+ * PUT /api/v1/delivery-policies/:pattern body = events.delivery-policy@1 -> the stored row (revision +1)
+ * DELETE /api/v1/delivery-policies/:pattern -> 204 (404 when absent)
+ * GET /api/v1/delivery-policies/resolve?event_type=<t> -> { event_type, pattern, revision, policy, carrier_class }
+ *
+ * The pattern is validated with the subscriptions' topic-pattern rule; the body with the contracts validator (an
+ * unknown class or an extra field is a 400 carrying the validator's message). carrier_class is the derived label
+ * (./../fabric/policy.js); it is answered here and nowhere else.
+ */
+const express = require('express');
+const { http, validate } = require('openvibe-contracts');
+const topics = require('../topics');
+const { CAPS } = require('../auth');
+const { carrierClass } = require('../fabric/policy');
+
+function policiesRouter({ policies, auth, limits }) {
+ const router = express.Router();
+ const isAdmin = auth.requireCap(CAPS.admin);
+ // Writes are rare operator actions: 30 a minute (like subscription changes); reads take the defaults.
+ const change = { minute: 30, hour: 300 };
+
+ const view = (row) => ({
+ pattern: row.pattern,
+ policy: row.policy,
+ revision: row.revision,
+ updated_at: new Date(row.updated_at).toISOString(),
+ updated_by: row.updated_by || null,
+ carrier_class: carrierClass(row.policy),
+ });
+
+ router.get('/api/v1/delivery-policies', isAdmin, limits('events.delivery-policy.list'), async (_req, res) => {
+ res.json({ policies: (await policies.all()).map(view) });
+ });
+
+ router.get('/api/v1/delivery-policies/resolve', isAdmin, limits('events.delivery-policy.read'), async (req, res) => {
+ const eventType = String(req.query.event_type || '');
+ if (!eventType || eventType.length > 200) {
+ return http.sendProblem(res, 400, 'events.bad_request', { detail: 'event_type is required and at most 200 characters', ctx: req.ov });
+ }
+ const { policy, pattern, revision } = await policies.resolve(eventType);
+ res.json({ event_type: eventType, pattern, revision, policy, carrier_class: carrierClass(policy) });
+ });
+
+ router.put('/api/v1/delivery-policies/:pattern', isAdmin, limits('events.delivery-policy.write', change), async (req, res) => {
+ const ctx = req.ov;
+ const pattern = String(req.params.pattern);
+ if (!topics.isValidPattern(pattern)) {
+ return http.sendProblem(res, 422, 'events.bad_topic', { detail: 'pattern must be dot-separated segments of [a-z0-9_] or *', ctx });
+ }
+ const body = req.body && typeof req.body === 'object' && !Array.isArray(req.body) ? req.body : null;
+ if (!body) return http.sendProblem(res, 400, 'events.bad_request', { detail: 'body must be an events.delivery-policy@1 object', ctx });
+ const v = validate('events.delivery-policy@1', body);
+ if (!v.valid) {
+ return http.sendProblem(res, 400, 'events.bad_request', { detail: v.errors.map((e) => `${e.path} ${e.message}`).join('; '), ctx });
+ }
+ const row = await policies.put(pattern, body, req.principal.service || req.principal.sub);
+ res.json(view(row));
+ });
+
+ router.delete('/api/v1/delivery-policies/:pattern', isAdmin, limits('events.delivery-policy.write', change), async (req, res) => {
+ const pattern = String(req.params.pattern);
+ if (!topics.isValidPattern(pattern)) {
+ return http.sendProblem(res, 422, 'events.bad_topic', { detail: 'pattern must be dot-separated segments of [a-z0-9_] or *', ctx: req.ov });
+ }
+ if (!await policies.remove(pattern)) return http.sendProblem(res, 404, 'events.not_found', { detail: 'no such delivery policy', ctx: req.ov });
+ res.status(204).end();
+ });
+
+ return router;
+}
+
+module.exports = { policiesRouter };
@@ -9,11 +9,12 @@ const { http } = require('openvibe-contracts');
@@ -9,11 +9,12 @@ const { http } = require('openvibe-contracts');
const { publishRouter } = require('./api/publish');
const { publishRouter } = require('./api/publish');
const { subscriptionsRouter } = require('./api/subscriptions');
const { subscriptionsRouter } = require('./api/subscriptions');
const { readRouter } = require('./api/read');
const { readRouter } = require('./api/read');
+const { policiesRouter } = require('./api/policies');
const { mountLimits } = require('./limits');
const { mountLimits } = require('./limits');
const { createLimits } = require('./actor-limits');
const { createLimits } = require('./actor-limits');
const pkg = require('../package.json');
const pkg = require('../package.json');
-function createApp({ config, store, auth, keys, worker, realtime, metrics, dnsLookup, clock, log = console, valkey = null }) {
+function createApp({ config, store, auth, keys, worker, realtime, metrics, dnsLookup, clock, log = console, valkey = null, policies = null }) {
const app = express();
const app = express();
app.disable('x-powered-by');
app.disable('x-powered-by');
app.set('trust proxy', 'loopback');
app.set('trust proxy', 'loopback');
@@ -78,6 +79,7 @@ function createApp({ config, store, auth, keys, worker, realtime, metrics, dnsLo
@@ -78,6 +79,7 @@ function createApp({ config, store, auth, keys, worker, realtime, metrics, dnsLo
app.use(publishRouter({ config, store, auth, worker, realtime, limits }));
app.use(publishRouter({ config, store, auth, worker, realtime, limits }));
app.use(subscriptionsRouter({ config, store, auth, dnsLookup, limits }));
app.use(subscriptionsRouter({ config, store, auth, dnsLookup, limits }));
app.use(readRouter({ store, auth, worker, limits }));
app.use(readRouter({ store, auth, worker, limits }));
+ if (policies) app.use(policiesRouter({ policies, auth, limits }));
app.get('/', (_req, res) => {
app.get('/', (_req, res) => {
res.type('text/plain').send([
res.type('text/plain').send([
new file mode 100644
new file mode 100644
@@ -0,0 +1,160 @@
@@ -0,0 +1,160 @@
+'use strict';
+/**
+ * The delivery policy (ADR-042 decisions 1 and 4; plan T7): the only semantic vocabulary for how an event type is
+ * carried. Policies live in `delivery_policies`, keyed by an event-type pattern using the same matcher subscriptions
+ * use for `topic_pattern` (../topics.js). The most specific matching pattern wins: an exact type beats any wildcard;
+ * among wildcards the longer literal part wins; ties break by pattern string order (deterministic). With no policy row
+ * behaviour is exactly today's DEFAULT_POLICY: no ordering key, one delivery in flight per subscription.
+ *
+ * The carrier class (TOPIC | QUEUE | STREAM) is an internal, derived label — never a contract enum, never in an
+ * envelope. It is answered by the explain/resolve API only.
+ *
+ * Publisher intent: the publish contract (events.publish-request@1 -> events.event-envelope@1) has no intent field, so
+ * none is read here; a class can only be raised by writing a policy. Do not add a field for it.
+ */
+const topics = require('../topics');
+
+const DEFAULT_POLICY = { class: 'domain', durability: 'required', delivery_semantics: 'at_least_once' };
+const MAX_KEY_LEN = 200;
+
+/**
+ * Which carrier class a policy's deliveries use. STREAM only when durability is required AND the policy has an
+ * ordering key (retained, ordered per key); TOPIC for the fan-out class; QUEUE for everything else.
+ */
+function carrierClass(policy) {
+ if (policy && policy.durability === 'required' && policy.ordering && typeof policy.ordering.key === 'string' && policy.ordering.key) return 'STREAM';
+ if (policy && policy.class === 'realtime_ephemeral') return 'TOPIC';
+ return 'QUEUE';
+}
+
+function subjectForm(envelope) {
+ const s = envelope && envelope.subject;
+ if (s && typeof s.type === 'string' && s.type && typeof s.id === 'string' && s.id) return `${s.type}:${s.id}`;
+ return null;
+}
+
+/** The value at a dotted path (`subject.id`, `payload.a.b`, `actor.id`) or null when it is not there. */
+function pathValue(envelope, path) {
+ let cur = envelope;
+ for (const seg of String(path).split('.')) {
+ if (cur == null || typeof cur !== 'object') return null;
+ cur = cur[seg];
+ }
+ return cur;
+}
+
+/**
+ * The ordering key of an envelope under a policy, or null when the policy has no ordering (the default). `'subject'`
+ * is `<subject.type>:<subject.id>`; a dotted path starting with subject./actor./payload. is that value when it is a
+ * string or a finite number; anything unresolvable falls back to the subject form, then to the event type. Capped at
+ * 200 characters. Never the seq.
+ */
+function orderingKey(policy, envelope) {
+ if (!policy || !policy.ordering || typeof policy.ordering.key !== 'string' || !policy.ordering.key) return null;
+ const key = policy.ordering.key;
+ let value = null;
+ if (key === 'subject') {
+ value = subjectForm(envelope);
+ } else if (/^(subject|actor|payload)\./.test(key)) {
+ const raw = pathValue(envelope, key);
+ if (typeof raw === 'string' || (typeof raw === 'number' && Number.isFinite(raw))) value = String(raw);
+ }
+ if (value == null || value === '') value = subjectForm(envelope) || (envelope && envelope.event_type) || null;
+ if (value == null) return null;
+ return value.length > MAX_KEY_LEN ? value.slice(0, MAX_KEY_LEN) : value;
+}
+
+/** Exact patterns beat wildcards; among wildcards the longer literal part wins. */
+function specificity(pattern) {
+ return pattern.includes('*') ? pattern.replace(/\*/g, '').length : Infinity;
+}
+
+function bestMatch(rows, eventType) {
+ let best = null;
+ for (const row of rows) {
+ if (!row.compiled.test(eventType)) continue;
+ if (!best) { best = row; continue; }
+ const a = specificity(row.pattern);
+ const b = specificity(best.pattern);
+ if (a > b || (a === b && row.pattern < best.pattern)) best = row;
+ }
+ return best;
+}
+
+function normalize(row) {
+ const policy = typeof row.policy === 'string' ? JSON.parse(row.policy) : row.policy;
+ return { pattern: row.pattern, policy, revision: Number(row.revision), updated_at: row.updated_at, updated_by: row.updated_by || null };
+}
+
+/**
+ * @param {object} db the Events database (openvibe-sdk/db handle)
+ * @param {{ now(): number }} [o.clock]
+ * @param {number} [o.reloadMs] how long a compiled cache is trusted before it is dropped (a second worker process
+ * picks up another process's writes; this process also drops it on every write)
+ */
+function createPolicyLibrary(db, { clock = { now: () => Date.now() }, reloadMs = 60000, log = console } = {}) {
+ let cache = null; // { rows, at }
+
+ async function load() {
+ let raw = [];
+ try {
+ raw = await db.prepare('SELECT pattern, policy, revision, updated_at, updated_by FROM delivery_policies').all();
+ } catch (err) {
+ log.warn(`[policy] could not load delivery_policies: ${err.message}`);
+ raw = [];
+ }
+ const rows = raw.map((r) => {
+ const row = normalize(r);
+ row.compiled = topics.compile(row.pattern);
+ return row;
+ });
+ cache = { rows, at: clock.now() };
+ return rows;
+ }
+
+ /** Drop the compiled cache; the next resolve reloads from the database. */
+ function invalidate() { cache = null; }
+
+ async function all() {
+ if (!cache || clock.now() - cache.at >= reloadMs) await load();
+ return cache.rows;
+ }
+
+ /** { policy, pattern, revision }; pattern is null and revision 0 for the default (no matching row). */
+ async function resolve(eventType) {
+ const rows = await all();
+ const m = bestMatch(rows, eventType);
+ return m ? { policy: m.policy, pattern: m.pattern, revision: m.revision } : { policy: DEFAULT_POLICY, pattern: null, revision: 0 };
+ }
+
+ /** The ordering key an envelope of `eventType` takes, or null (no policy ordering). */
+ async function keyFor(eventType, envelope) {
+ const { policy } = await resolve(eventType);
+ return orderingKey(policy, envelope);
+ }
+
+ /** Upsert a policy row; revision +1 on an existing pattern. Returns the stored row. */
+ async function put(pattern, policy, updatedBy) {
+ const row = await db.prepare(`INSERT INTO delivery_policies (pattern, policy, revision, updated_at, updated_by)
+ VALUES (?, ?, 1, ?, ?)
+ ON CONFLICT(pattern) DO UPDATE SET policy = excluded.policy, revision = delivery_policies.revision + 1,
+ updated_at = excluded.updated_at, updated_by = excluded.updated_by
+ RETURNING pattern, policy, revision, updated_at, updated_by`).get(pattern, JSON.stringify(policy), clock.now(), updatedBy);
+ invalidate();
+ return normalize(row);
+ }
+
+ /** Remove a policy row. Returns the number of rows deleted. */
+ async function remove(pattern) {
+ const changes = (await db.prepare('DELETE FROM delivery_policies WHERE pattern = ?').run(pattern)).changes;
+ invalidate();
+ return changes;
+ }
+
+ const timer = setInterval(() => { invalidate(); }, reloadMs);
+ timer.unref?.();
+
+ return { resolve, keyFor, all, put, remove, invalidate, reload: load, stop() { clearInterval(timer); } };
+}
+
+module.exports = { DEFAULT_POLICY, carrierClass, orderingKey, specificity, createPolicyLibrary };
@@ -9,6 +9,7 @@
@@ -9,6 +9,7 @@
*/
*/
const { load } = require('./config');
const { load } = require('./config');
const { openDb, createStore } = require('./store');
const { openDb, createStore } = require('./store');
+const { createPolicyLibrary } = require('./fabric/policy');
const { createKeyStore, createAuth } = require('./auth');
const { createKeyStore, createAuth } = require('./auth');
const { createWorker } = require('./worker');
const { createWorker } = require('./worker');
const { createRealtime } = require('./realtime');
const { createRealtime } = require('./realtime');
@@ -23,7 +24,10 @@ async function start({
@@ -23,7 +24,10 @@ async function start({
config = config || load();
config = config || load();
// PostgreSQL (ADR-035): opened and migrated here unless the caller (a test) hands in a migrated handle.
// PostgreSQL (ADR-035): opened and migrated here unless the caller (a test) hands in a migrated handle.
const db = givenDb || await openDb(config, { log });
const db = givenDb || await openDb(config, { log });
- const store = createStore(db, { clock, maxHops: config.maxHops, usage: config.usage });
+ // Delivery policies (ADR-042 decisions 1 and 4): the store reads the resolved ordering key on fan-out, the admin
+ // API writes rows and drops the compiled cache. One library per process, shared by both.
+ const policies = createPolicyLibrary(db, { clock, log });
+ const store = createStore(db, { clock, maxHops: config.maxHops, usage: config.usage, policies });
const keys = createKeyStore({ urls: [config.networkInternalUrl, config.networkUrl], pem: config.networkPublicKey, fetchImpl, log });
const keys = createKeyStore({ urls: [config.networkInternalUrl, config.networkUrl], pem: config.networkPublicKey, fetchImpl, log });
const auth = createAuth({ config, keys, store });
const auth = createAuth({ config, keys, store });
const metrics = createMetrics({ store });
const metrics = createMetrics({ store });
@@ -33,7 +37,7 @@ async function start({
@@ -33,7 +37,7 @@ async function start({
const realtime = createRealtime({ store, auth, config, clock, log });
const realtime = createRealtime({ store, auth, config, clock, log });
metrics.bind({ realtime });
metrics.bind({ realtime });
const valkey = config.valkey.url ? require('openvibe-sdk/valkey').createValkey({ url: config.valkey.url, prefix: config.valkey.prefix, log }) : null;
const valkey = config.valkey.url ? require('openvibe-sdk/valkey').createValkey({ url: config.valkey.url, prefix: config.valkey.prefix, log }) : null;
- const app = createApp({ config, store, auth, keys, worker, realtime, metrics, dnsLookup, clock, log, valkey });
+ const app = createApp({ config, store, auth, keys, worker, realtime, metrics, dnsLookup, clock, log, valkey, policies });
// The key loads in the background (retrying while Network boots); /api/ready says when it has.
// The key loads in the background (retrying while Network boots); /api/ready says when it has.
const keyLoaded = keys.start().catch(() => null);
const keyLoaded = keys.start().catch(() => null);
@@ -86,6 +90,7 @@ async function start({
@@ -86,6 +90,7 @@ async function start({
async function close() {
async function close() {
clearInterval(pruneTimer);
clearInterval(pruneTimer);
if (usageTimer) clearInterval(usageTimer);
if (usageTimer) clearInterval(usageTimer);
+ policies.stop();
realtime.stop();
realtime.stop();
keys.stop();
keys.stop();
await worker.stop();
await worker.stop();
@@ -97,7 +102,7 @@ async function start({
@@ -97,7 +102,7 @@ async function start({
if (!givenDb) await db.close();
if (!givenDb) await db.close();
}
}
- return { config, db, store, keys, keyLoaded, auth, worker, realtime, metrics, app, server, close, prune, flushUsage };
+ return { config, db, store, keys, keyLoaded, auth, worker, realtime, metrics, app, server, policies, close, prune, flushUsage };
}
}
if (require.main === module) {
if (require.main === module) {
@@ -91,7 +91,7 @@ function subscriptionView(row, { withSecret = false } = {}) {
@@ -91,7 +91,7 @@ function subscriptionView(row, { withSecret = false } = {}) {
return out;
return out;
}
}
-function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage: usageConfig = { enabled: true } } = {}) {
+function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage: usageConfig = { enabled: true }, policies = null } = {}) {
// 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 = {
@@ -130,7 +130,7 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
@@ -130,7 +130,7 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
ON CONFLICT(consumer, scope) DO UPDATE SET revoked_at = GREATEST(app_revocations.revoked_at, excluded.revoked_at)`),
ON CONFLICT(consumer, scope) DO UPDATE SET revoked_at = GREATEST(app_revocations.revoked_at, excluded.revoked_at)`),
revokedAt: db.prepare('SELECT revoked_at FROM app_revocations WHERE consumer = ? AND scope = ?'),
revokedAt: db.prepare('SELECT revoked_at FROM app_revocations WHERE consumer = ? AND scope = ?'),
insertDelivery: db.prepare(`INSERT INTO deliveries (event_id, subscription_id, seq, priority, attempt, status,
insertDelivery: db.prepare(`INSERT INTO deliveries (event_id, subscription_id, seq, priority, attempt, status,
- next_attempt_at, created_at, updated_at) VALUES (?, ?, ?, ?, 0, 'pending', ?, ?, ?) ON CONFLICT DO NOTHING`),
+ next_attempt_at, ordering_key, created_at, updated_at) VALUES (?, ?, ?, ?, 0, 'pending', ?, ?, ?, ?) ON CONFLICT DO NOTHING`),
afterSeq: db.prepare('SELECT * FROM events WHERE seq > ? ORDER BY seq LIMIT ?'),
afterSeq: db.prepare('SELECT * FROM events WHERE seq > ? ORDER BY seq LIMIT ?'),
afterSeqLike: db.prepare("SELECT * FROM events WHERE seq > ? AND event_type ILIKE ? ESCAPE '\\' ORDER BY seq LIMIT ?"),
afterSeqLike: db.prepare("SELECT * FROM events WHERE seq > ? AND event_type ILIKE ? ESCAPE '\\' ORDER BY seq LIMIT ?"),
};
};
@@ -229,9 +229,13 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
@@ -229,9 +229,13 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
if (directive) row.redacts = 1;
if (directive) row.redacts = 1;
await q.insertEvent.run(row);
await q.insertEvent.run(row);
await q.addReceipt.run(PUBLISH_RECEIPT, env.event_id, now);
await q.addReceipt.run(PUBLISH_RECEIPT, env.event_id, now);
+ // The ordering key is a property of the event type (ADR-042 decision 4): resolve it once per event and give
+ // every delivery of it the same key. No policy row (the default) resolves to null and behaves as before.
+ // Publisher intent is not read: the publish contract has no intent field.
+ const orderingKey = policies ? await policies.keyFor(env.event_type, env) : null;
for (const s of subs) {
for (const s of subs) {
if (apps.deliverable(s, row)) {
if (apps.deliverable(s, row)) {
- await q.insertDelivery.run(env.event_id, s.id, seq, PRIORITY_RANK[env.priority], now, now, now);
+ await q.insertDelivery.run(env.event_id, s.id, seq, PRIORITY_RANK[env.priority], now, orderingKey, now, now);
}
}
}
}
await revocationHook(row, now);
await revocationHook(row, now);
@@ -422,47 +426,72 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
@@ -422,47 +426,72 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
// ── Deliveries ───────────────────────────────────────────
// ── Deliveries ───────────────────────────────────────────
/**
/**
- * 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.
+ * Claim up to `limit` due deliveries for `owner` until `leaseUntil` (ms): the best claimable delivery (by priority
+ * class, then seq) of each subscription that has room. A delivery is claimable when it is due and (ADR-042
+ * decision 4) it is the head of its ordering key — no lower-seq pending/failed row of the same (subscription, key)
+ * and nothing of that key leased — or it has no key at all. At most `subscriptions.max_inflight` deliveries of a
+ * subscription are leased at once (default 1 = today); with more, several keys of one subscription run together,
+ * never two of one key. 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
+ * until recordAttempt clears it or it expires (an expired lease is claimable again: delivery is at least once).
+ * `maxApp` caps developer-app subscriptions.
+ *
+ * The leasing UPDATE re-checks everything in its own snapshot (READ COMMITTED does not re-evaluate the candidate
+ * query's predicates when the locked row itself did not change): the head condition, that nothing of the key is
+ * leased, and that the subscription is under its inflight cap. A claim counts only when it changed exactly one row.
*/
*/
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).
+ const claimable = (d) => `(${d}.status IN ('pending', 'failed') AND ${d}.next_attempt_at <= ?
+ AND (${d}.ordering_key IS NULL OR (
+ 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 NOT EXISTS (SELECT 1 FROM deliveries y WHERE y.subscription_id = ${d}.subscription_id
+ AND y.ordering_key = ${d}.ordering_key AND y.lease_until > ?)
+ )))`;
+ // One binding: `now`, the lease expiry to compare against.
+ const leased = (sub) => `(SELECT COUNT(*) FROM deliveries x WHERE x.subscription_id = ${sub} AND x.lease_until > ?)`;
return await db.tx(async () => {
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
// 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.
+ // 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(`
const candidates = (apps, n) => (n <= 0 ? [] : db.prepare(`
SELECT s.id, s.project_id,
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
+ (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
FROM subscriptions s
FROM subscriptions s
WHERE s.enabled = 1 AND s.project_id IS ${apps ? 'NOT ' : ''}NULL
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 > ?)
+ AND EXISTS (SELECT 1 FROM deliveries d WHERE d.subscription_id = s.id AND ${claimable('d')})
+ 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, n));
+ FOR UPDATE OF s SKIP LOCKED`).all(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 = [];
let apps = 0;
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
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).
+ WHERE d.subscription_id = ? AND ${claimable('d')} AND ${leased('d.subscription_id')} < s.max_inflight
+ ORDER BY d.priority, d.seq LIMIT 1`);
+ // The lease is taken only if it still holds as of NOW (see the method comment): claimable, under the cap,
+ // and of a key nothing else is leased on.
const lease = db.prepare(`UPDATE deliveries d SET lease_owner = ?, lease_until = ? WHERE d.event_id = ? AND d.subscription_id = ?
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 > ?)`);
+ AND ${claimable('d')}
+ AND ${leased('d.subscription_id')} < (SELECT s.max_inflight FROM subscriptions s WHERE s.id = d.subscription_id)`);
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;
- 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++;
+ // 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 (;;) {
+ if (picked.length >= limit || (sub.project_id && apps >= maxApp)) break;
+ const d = await next.get(sub.id, 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
+ picked.push({ ...d, lease_owner: owner, lease_until: leaseUntil });
+ if (sub.project_id) apps++;
+ }
}
}
return picked;
return picked;
});
});
@@ -525,7 +554,7 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
@@ -525,7 +554,7 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
const requeue = async (sub, { fromSeq, eventIds, max = 10000 }) => await db.tx(async () => {
const requeue = async (sub, { fromSeq, eventIds, max = 10000 }) => await db.tx(async () => {
const now = clock.now();
const now = clock.now();
let rows;
let rows;
- const cols = 'id, seq, priority, event_type, visibility, project_id, env';
+ const cols = 'id, seq, priority, event_type, visibility, project_id, env, subject_type, subject_id, actor, payload';
if (Array.isArray(eventIds)) {
if (Array.isArray(eventIds)) {
const get = db.prepare(`SELECT ${cols} FROM events WHERE id = ?`);
const get = db.prepare(`SELECT ${cols} FROM events WHERE id = ?`);
rows = (await Promise.all(eventIds.map(async id => await get.get(id)))).filter(Boolean);
rows = (await Promise.all(eventIds.map(async id => await get.get(id)))).filter(Boolean);
@@ -535,11 +564,17 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
@@ -535,11 +564,17 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
}
}
// Replay never widens a subscription's scope (app scope, sandbox separation).
// Replay never widens a subscription's scope (app scope, sandbox separation).
rows = rows.filter(r => apps.deliverable(sub, r)).slice(0, max);
rows = rows.filter(r => apps.deliverable(sub, r)).slice(0, max);
- const up = db.prepare(`INSERT INTO deliveries (event_id, subscription_id, seq, priority, attempt, status, next_attempt_at, created_at, updated_at)
- VALUES (?, ?, ?, ?, 0, 'pending', ?, ?, ?)
+ // ordering_key is written on an insert only: a row that already exists keeps the key its first fan-out gave it.
+ const up = db.prepare(`INSERT INTO deliveries (event_id, subscription_id, seq, priority, attempt, status, next_attempt_at, ordering_key, created_at, updated_at)
+ VALUES (?, ?, ?, ?, 0, 'pending', ?, ?, ?, ?)
ON CONFLICT(event_id, subscription_id) DO UPDATE SET status = 'pending', attempt = 0, next_attempt_at = excluded.next_attempt_at,
ON CONFLICT(event_id, subscription_id) DO UPDATE SET status = 'pending', attempt = 0, next_attempt_at = excluded.next_attempt_at,
last_error = NULL, last_status = NULL, delivered_at = NULL, updated_at = excluded.updated_at`);
last_error = NULL, last_status = NULL, delivered_at = NULL, updated_at = excluded.updated_at`);
- for (const r of rows) await up.run(r.id, sub.id, r.seq, PRIORITY_RANK[r.priority], now, now, now);
+ for (const r of rows) {
+ const key = policies ? await policies.keyFor(r.event_type, {
+ event_type: r.event_type, subject: { type: r.subject_type, id: r.subject_id }, actor: JSON.parse(r.actor), payload: JSON.parse(r.payload),
+ }) : null;
+ await up.run(r.id, sub.id, r.seq, PRIORITY_RANK[r.priority], now, key, now, now);
+ }
return rows.length;
return rows.length;
});
});
new file mode 100644
new file mode 100644
@@ -0,0 +1,239 @@
@@ -0,0 +1,239 @@
+'use strict';
+// ADR-042 decisions 1 and 4 (plan T7): the delivery policy is the semantic vocabulary (class -> derived carrier
+// class, ordering key resolved against the envelope) and per-key ordering is enforced in the claim: a key's head is
+// claimable, its successors wait for it, and a subscription keeps at most max_inflight deliveries leased (one key in
+// flight by default). Concurrency bugs show only on real PostgreSQL — npm run test:pg runs this file there.
+const assert = require('assert');
+const { boot, request, serviceToken, envelope, subscriber, suite, sleep } = require('./helpers');
+const { createWorker } = require('../server/worker');
+const { DEFAULT_POLICY, carrierClass, orderingKey } = require('../server/fabric/policy');
+
+const t = suite('ordering');
+let h;
+const admin = serviceToken('ops', ['events.delivery.admin']);
+const reader = serviceToken('media', ['events.event.read']);
+const live = serviceToken('live', ['events.event.publish']);
+const silent = { log() {}, warn() {}, error() {} };
+
+const CLASSES = ['realtime_ephemeral', 'interactive', 'interactive_durable', 'domain', 'critical', 'background', 'bulk', 'webhook'];
+const policy = (over = {}) => ({ class: 'domain', durability: 'required', delivery_semantics: 'at_least_once', ...over });
+const keyed = (key, scope = 'key') => policy({ ordering: { scope, key } });
+
+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);
+}
+async function putPolicy(pattern, body) {
+ const r = await request(h.base, 'PUT', `/api/v1/delivery-policies/${pattern}`, { token: admin, body });
+ assert.strictEqual(r.status, 200, r.text);
+ return r.body;
+}
+const deletePolicy = (pattern) => request(h.base, 'DELETE', `/api/v1/delivery-policies/${pattern}`, { token: admin });
+
+t('boot', async () => { h = await boot(); });
+
+t('policy: default, carrier classes, key resolution and the admin API', async () => {
+ // Every class maps to exactly one carrier class; durability required + an ordering key is STREAM for all of them.
+ for (const c of CLASSES) {
+ assert.strictEqual(carrierClass(policy({ class: c })), c === 'realtime_ephemeral' ? 'TOPIC' : 'QUEUE');
+ assert.strictEqual(carrierClass(keyed('subject', 'key', c)), 'STREAM', `${c}: required + ordering is STREAM`);
+ assert.strictEqual(carrierClass({ class: c, durability: 'optional', delivery_semantics: 'at_least_once', ordering: { scope: 'key', key: 'subject' } }), c === 'realtime_ephemeral' ? 'TOPIC' : 'QUEUE', `${c}: ordering without required durability is not STREAM`);
+ }
+ // Key resolution: 'subject', dotted paths, fallbacks, the 200-char cap, never the seq; no ordering -> null.
+ const env = envelope('live', { subject: { type: 'stream', id: 'str_9' }, actor: { type: 'service', id: 'live' }, payload: { room: 42, empty: '', deep: { x: 'y' } } });
+ assert.strictEqual(orderingKey(keyed('subject'), env), 'stream:str_9');
+ assert.strictEqual(orderingKey(keyed('payload.room'), env), '42');
+ assert.strictEqual(orderingKey(keyed('actor.id'), env), 'live');
+ assert.strictEqual(orderingKey(keyed('payload.deep.x'), env), 'y');
+ assert.strictEqual(orderingKey(keyed('payload.empty'), env), 'stream:str_9', 'an empty/unresolvable path falls back to the subject');
+ assert.strictEqual(orderingKey(keyed('payload.missing'), env), 'stream:str_9');
+ assert.strictEqual(orderingKey(keyed('payload.missing'), { event_type: 'live.x.y' }), 'live.x.y', 'no subject: falls back to the event type');
+ assert.strictEqual(orderingKey(keyed('payload.big'), { event_type: 'live.x', payload: { big: 'a'.repeat(500) } }).length, 200, 'the key is capped at 200 characters');
+ assert.strictEqual(orderingKey(DEFAULT_POLICY, env), null, 'the default policy has no ordering');
+ assert.strictEqual(orderingKey({ ordering: { scope: 'key', key: 'paylaod.typo' } }, env), 'stream:str_9', 'a path that is not subject./actor./payload. falls back to the subject');
+
+ // No rows: the default, and an empty list. Resolve answers the default shape.
+ assert.deepStrictEqual(await h.policies.resolve('live.anything'), { policy: DEFAULT_POLICY, pattern: null, revision: 0 });
+ const empty = await request(h.base, 'GET', '/api/v1/delivery-policies', { token: admin });
+ assert.strictEqual(empty.status, 200);
+ assert.deepStrictEqual(empty.body.policies, []);
+
+ // PUT: revision starts at 1, +1 on a rewrite; updated_by is the admin principal; carrier_class is derived.
+ const exact = await putPolicy('live.order.exact', keyed('subject'));
+ assert.strictEqual(exact.revision, 1);
+ assert.strictEqual(exact.carrier_class, 'STREAM');
+ assert.strictEqual(exact.updated_by, 'ops');
+ await putPolicy('live.order.*', keyed('subject'));
+ await putPolicy('live.*', keyed('subject'));
+ const rewrite = await putPolicy('live.order.exact', { class: 'critical', durability: 'none', delivery_semantics: 'at_most_once' });
+ assert.strictEqual(rewrite.revision, 2);
+ assert.strictEqual(rewrite.carrier_class, 'QUEUE');
+
+ // Specificity: the exact type beats any wildcard; the longer literal wildcard beats the shorter one.
+ const rExact = await request(h.base, 'GET', '/api/v1/delivery-policies/resolve?event_type=live.order.exact', { token: admin });
+ assert.deepStrictEqual([rExact.status, rExact.body.pattern, rExact.body.revision, rExact.body.carrier_class], [200, 'live.order.exact', 2, 'QUEUE']);
+ const rWild = await request(h.base, 'GET', '/api/v1/delivery-policies/resolve?event_type=live.order.other', { token: admin });
+ assert.deepStrictEqual([rWild.body.pattern, rWild.body.carrier_class], ['live.order.*', 'STREAM']);
+ const rShort = await request(h.base, 'GET', '/api/v1/delivery-policies/resolve?event_type=live.elsewhere.x', { token: admin });
+ assert.strictEqual(rShort.body.pattern, 'live.*', 'the shorter wildcard is the only match');
+
+ // Ties (equal literal length) break by pattern string order: 'a.*.c.d' beats 'a.b.*.d'.
+ await putPolicy('a.b.*.d', keyed('subject'));
+ await putPolicy('a.*.c.d', keyed('subject'));
+ const tie = await request(h.base, 'GET', '/api/v1/delivery-policies/resolve?event_type=a.b.c.d', { token: admin });
+ assert.strictEqual(tie.body.pattern, 'a.*.c.d', 'ties break deterministically by pattern string order');
+
+ // The resolve endpoint answers the default for an unknown type.
+ const none = await request(h.base, 'GET', '/api/v1/delivery-policies/resolve?event_type=nobody.home', { token: admin });
+ assert.deepStrictEqual([none.body.pattern, none.body.revision, none.body.carrier_class, none.body.policy], [null, 0, 'QUEUE', DEFAULT_POLICY]);
+
+ // The validator refuses an unknown class and an extra field with its own message.
+ const badClass = await request(h.base, 'PUT', '/api/v1/delivery-policies/live.bad', { token: admin, body: { class: 'warp_speed', durability: 'required', delivery_semantics: 'at_least_once' } });
+ assert.strictEqual(badClass.status, 400, badClass.text);
+ assert.ok(/class/i.test(badClass.body.detail), badClass.body.detail);
+ const badField = await request(h.base, 'PUT', '/api/v1/delivery-policies/live.bad', { token: admin, body: { ...policy(), mode: 'fast' } });
+ assert.strictEqual(badField.status, 400, badField.text);
+
+ // Non-admin: the same answers as the other admin routes (403 for a service without the capability, 401 without a token).
+ assert.strictEqual((await request(h.base, 'GET', '/api/v1/delivery-policies', { token: reader })).status, 403);
+ assert.strictEqual((await request(h.base, 'GET', '/api/v1/delivery-policies')).status, 401);
+ assert.strictEqual((await request(h.base, 'PUT', '/api/v1/delivery-policies/live.bad', { token: reader, body: policy() })).status, 403);
+
+ // DELETE: 404 when absent, 204 then again 404.
+ assert.strictEqual((await deletePolicy('live.bad')).status, 404);
+ assert.strictEqual((await deletePolicy('live.order.exact')).status, 204);
+ assert.strictEqual((await deletePolicy('live.order.exact')).status, 404);
+
+ for (const p of ['live.order.*', 'live.*', 'a.b.*.d', 'a.*.c.d']) assert.strictEqual((await deletePolicy(p)).status, 204);
+});
+
+t('per-key order under retries: a successor waits for its head, another key goes meanwhile', async () => {
+ const stub = await subscriber(() => 204);
+ const sub = await subscribe('ord2', 'live.ord2.*', stub.url);
+ await putPolicy('live.ord2.*', keyed('subject'));
+ const A1 = envelope('live', { event_type: 'live.ord2.x', subject: { type: 'stream', id: 'A' } });
+ const A2 = envelope('live', { event_type: 'live.ord2.x', subject: { type: 'stream', id: 'A' } });
+ const B1 = envelope('live', { event_type: 'live.ord2.x', subject: { type: 'stream', id: 'B' } });
+ await publish(A1); await publish(A2); await publish(B1);
+
+ let failA1 = true;
+ const attempts = [];
+ const fetchImpl = async (url, init) => {
+ const ev = JSON.parse(init.body).event;
+ let status = 204;
+ if (ev.event_id === A1.event_id && failA1) { failA1 = false; status = 500; }
+ attempts.push({ id: ev.event_id, status });
+ return new Response(null, { status });
+ };
+ const w = createWorker({ store: h.store, config: h.config, clock: h.clock, fetchImpl, log: silent });
+ await w.drain();
+ const ids = () => attempts.map((a) => a.id);
+ assert.deepStrictEqual(ids(), [A1.event_id, B1.event_id], 'A1 failed and B1 went; A2 (a successor of A1) did not');
+ assert.strictEqual(attempts[0].status, 500);
+
+ h.clock.advance(1001); // A1's backoff elapses
+ await w.drain();
+ assert.deepStrictEqual(ids(), [A1.event_id, B1.event_id, A1.event_id, A2.event_id], 'A1 is retried and only then A2');
+ assert.strictEqual(attempts.filter((a) => a.id === A1.event_id).length, 2);
+ assert.strictEqual(attempts.filter((a) => a.id === A2.event_id).length, 1);
+
+ await deletePolicy('live.ord2.*');
+ await stub.close();
+ await h.store.setSubscriptionEnabled(sub.id, false);
+});
+
+t('per-key order across two workers with a lease expiry, and never two distinct seqs of one key in flight', async () => {
+ const stub = await subscriber(() => 204);
+ const sub = await subscribe('ord3', 'live.ord3.*', stub.url);
+ await putPolicy('live.ord3.*', keyed('subject'));
+
+ const keys = ['K0', 'K1', 'K2'];
+ const byId = new Map(); // event_id -> deliveries
+ const firstSeqs = new Map(keys.map((k) => [k, []])); // key -> seqs in first-delivery order
+ const inflight = new Map(); // key -> Set(seq)
+ let distinctPeak = 0; // the most distinct seqs of one key ever in flight at once
+ const fetchImpl = async (url, init) => {
+ const { event, seq } = JSON.parse(init.body);
+ const k = event.subject.id;
+ byId.set(event.event_id, (byId.get(event.event_id) || 0) + 1);
+ const set = inflight.get(k) || new Set();
+ inflight.set(k, set);
+ set.add(seq);
+ distinctPeak = Math.max(distinctPeak, set.size);
+ await sleep(2);
+ set.delete(seq);
+ const list = firstSeqs.get(k);
+ if (!list.includes(seq)) list.push(seq);
+ return new Response(null, { status: 204 });
+ };
+
+ // 30 events over 3 keys, interleaved so the global seq order is not the per-key order.
+ for (let i = 0; i < 10; i++) {
+ for (const k of keys) await publish(envelope('live', { event_type: 'live.ord3.x', subject: { type: 'stream', id: k } }));
+ }
+
+ 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 });
+ const pa = a.drain();
+ await sleep(4);
+ h.clock.advance(40001); // whatever worker A holds now has an expired lease: worker B may re-claim it (at least once)
+ const pb = b.drain();
+ await Promise.all([pa, pb]);
+ await a.drain(); await b.drain();
+
+ assert.strictEqual(byId.size, 30, 'every event of every key was delivered');
+ assert.ok([...byId.values()].every((n) => n >= 1), 'every event delivered at least once');
+ for (const k of keys) {
+ const list = firstSeqs.get(k);
+ assert.deepStrictEqual(list, [...list].sort((x, y) => x - y), `${k}: the first delivery of each seq is in seq order (${list})`);
+ }
+ assert.strictEqual(distinctPeak, 1, 'never two distinct seqs of one key in flight at once');
+
+ await deletePolicy('live.ord3.*');
+ await stub.close();
+ await h.store.setSubscriptionEnabled(sub.id, false);
+});
+
+t('max_inflight = 2: two keys run together, never two of one key, never three at once', async () => {
+ const stub = await subscriber(() => 204);
+ const sub = await subscribe('ord4', 'live.ord4.*', stub.url);
+ await putPolicy('live.ord4.*', keyed('subject'));
+ await h.store.db.prepare('UPDATE subscriptions SET max_inflight = 2 WHERE id = ?').run(sub.id);
+
+ const inflight = []; // { key, seq } currently being sent
+ let totalPeak = 0;
+ let keyPeak = 0;
+ const fetchImpl = async (url, init) => {
+ const { event, seq } = JSON.parse(init.body);
+ const k = event.subject.id;
+ inflight.push({ k, seq });
+ totalPeak = Math.max(totalPeak, inflight.length);
+ keyPeak = Math.max(keyPeak, inflight.filter((x) => x.k === k).length);
+ await sleep(4);
+ inflight.splice(inflight.findIndex((x) => x.k === k && x.seq === seq), 1);
+ return new Response(null, { status: 204 });
+ };
+
+ for (let i = 0; i < 2; i++) {
+ for (const k of ['K0', 'K1']) await publish(envelope('live', { event_type: 'live.ord4.x', subject: { type: 'stream', id: k } }));
+ }
+ const w = createWorker({ store: h.store, config: h.config, clock: h.clock, fetchImpl, log: silent });
+ await w.drain();
+
+ 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((await h.store.deliveryCounts()).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.*');
+ await stub.close();
+ await h.store.setSubscriptionEnabled(sub.id, false);
+});
+
+t('stop', async () => { await h.stop(); });
+t.run();