openvibers/openvibe.events Commits

Carriers, the planner and explain (plan T7, ADR-042 decisions 2 and 5): pg-v1 (the poll, every class) and valkey-v1 (QUEUE: a Valkey stream per carrier class through openvibe-sdk/queue, whose consumers take a delivery with a targeted lease; TOPIC: realtime fan-out across Events processes over pub/sub, deduped by event id), with EWMA and a breaker per adapter. PostgreSQL stays the record and the lease the only thing that sends: a lost, late or duplicated stream item changes latency, never the outcome. server/fabric/planner.js places each (carrier class, ordering key) with openvibe-sdk/placement over resource offers and revisioned rate cards, with hysteresis, and keeps a key with a backlog on its carrier; deliveries record their carrier; delivery_placements (migration 0005) keeps each change of decision (pruned with the retention, the latest per key kept). EVENTS_CARRIERS / EVENTS_CARRIERS_DISABLED choose and switch off adapters (the rollback). at_most_once policies are not retried. Admin explain API: GET /api/v1/placement?event_type=&key= and /api/v1/placement/deliveries/:event/:subscription. A targeted claim locks its subscription row like the poll does (a race test on real PostgreSQL fails without it). npm test 20/21 (fabric's Valkey half needs test:pg) and test:pg 21/21.

OP OpenVibers committed 29 Sept 2026 · d705349ad1 · 19 files, +1044 −28


Changes

ddiff --git a/README.md b/README.md
iindex bcd947c..3982d2d 100644
---- a/README.md
++++ b/README.md
@@@ -114,6 +114,22 @@ Operators: `GET /api/v1/deliveries?status=dead` is the dead-letter queue; `POST
Pull consumers: `GET /api/v1/events?topic=media.vod.*&after_seq=<position>&limit=100` returns `{ events: [{ seq, cursor, event }], next_after_seq, next_cursor, latest_seq }`, plus `gap: { from_seq, to_seq }` when the position is older than retention or from another retention epoch. The opaque cursor ([ADR-042](https://github.com/OpenVibers/OpenVibe.Contracts/blob/main/docs/adr/ADR-042-events-fabric.md) decision 7) goes back in as `after=<cursor>` beside `after_seq`; `next_cursor` is the cursor for the next page, `seq`/`next_after_seq`/`latest_seq` stay for one release. `PUT /api/v1/checkpoints { topic, cursor }` stores a consumer's position here if it has nowhere better (an opaque cursor string is accepted in place of the integer; the answer carries its `epoch`, `carrier` and `next_cursor`).
++## The fabric: carriers, the planner and explain
++
++How an event type is carried is its delivery policy (`events.delivery-policy@1`, [ADR-042](https://github.com/OpenVibers/OpenVibe.Contracts/blob/main/docs/adr/ADR-042-events-fabric.md) decision 1). The carrier class — TOPIC (fan-out, no ack), QUEUE (competing consumers, ack) or STREAM (retained, ordered per key) — is a derived, internal label, never in the envelope. PostgreSQL stays the record: every event and every delivery is committed before any carrier sees it, and the worker's lease is the only thing that makes a send happen. A publish commits, then the chosen carrier is signalled; a lost or duplicated signal changes latency, never the outcome.
++
++Carriers (`server/fabric/carriers`; `EVENTS_CARRIERS`, default `pg-v1,valkey-v1`):
++
++- `pg-v1` — the poll. Always present, every class; the worker's 500 ms tick and its in-process `kick()` carry delivery. It is the floor every delivery falls back to.
++- `valkey-v1` — push, only when `VALKEY_URL` is set. **QUEUE:** after the commit, each new delivery is XADDed to a Valkey stream (`openvibe-sdk/queue`, consumer group `workers`); each Events process claims just that one row (`store.claimDelivery`, the same key-head and inflight-cap rules as the poll) and hands it to the worker's normal send — the item is acked either way, so a lost item only falls to the poll. **TOPIC:** the stored envelopes are PUBLISHed on a Valkey channel tagged with this process's id, and every process re-emits envelopes from other processes to its own SSE clients (deduped by event id), so an event published on one process reaches a browser connected to another. Health is an EWMA plus a breaker; while a breaker is open the planner excludes the carrier with a reason and `pg-v1` delivers.
++- `nats-v1` (STREAM) is not built yet: STREAM is carried by `pg-v1`.
++
++The planner (`server/fabric/planner.js`) calls `openvibe-sdk/placement` `plan()` per (carrier class, ordering key) with one `resource-offer@1` per adapter and the rate cards in `server/fabric/rate-cards.json`, caching decisions with hysteresis and recording a change in `delivery_placements`. A key's backlog stays on one carrier until it drains (a placement is pinned per (subscription, key), not per time window). An adapter named in `EVENTS_CARRIERS_DISABLED` is ineligible at once with reason `disabled by configuration` — the ADR's rollback; an unknown id in `EVENTS_CARRIERS` refuses to start.
++
++Explain (operators only, `events.delivery.admin`, like `/api/v1/deliveries`): `GET /api/v1/placement?event_type=<t>[&key=<k>]` answers the carrier class, the resolved policy, the planner's `platform.placement-result@1` (reasons, and per candidate its eligibility / `excluded_because` / cost / latency) and the last recorded decision; `GET /api/v1/placement/deliveries/:event_id/:subscription_id` answers a delivery's carrier and the decision in force when it was created.
++
++`delivery_semantics: at_most_once` makes a failed attempt final (`dead`, never retried); the default stays `at_least_once`.
++
## Developer apps
Roadmap Wave 20, ADR-014. A developer app gets a token from OpenVibe.Network (`POST /oauth/token`, `grant_type=client_credentials`, `audience=openvibe.events`) with `sub: app:app_<ULID>`, `project_id: prj_<ULID>`, `env: sandbox|production`. It uses the same routes as services, with these capabilities (all `public`, openvibe-contracts ≥ 0.28.0):
ddiff --git a/STATUS.json b/STATUS.json
iindex 6015248..4baeffb 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -30,6 +30,8 @@
"developer-app publish, read and subscribe with sandbox separation, SSRF-guarded webhooks and project quotas",
"hourly project usage rollups (events.usage.recorded)",
"redaction tombstones",
-- "per-actor limits (openvibe-sdk/limits; shared on Valkey)"
++ "per-actor limits (openvibe-sdk/limits; shared on Valkey)",
++ "carrier fabric (ADR-042): the pg-v1 poll beside a valkey-v1 push carrier for QUEUE (streams) and TOPIC (SSE fan-out), with a placement planner, per-key pinning, delivery_placements history and a GET /api/v1/placement explain API; EVENTS_CARRIERS[_DISABLED] switches it",
++ "at_most_once delivery semantics (a failed attempt is final)"
]
}
ddiff --git a/migrations/0005_placements.sql b/migrations/0005_placements.sql
nnew file mode 100644
iindex 0000000..0a5ba38
---- /dev/null
++++ b/migrations/0005_placements.sql
@@@ -0,0 +1,14 @@
++-- phase: expand
++-- ADR-042 decision 5 (plan T7): the planner records a (carrier class, ordering key) decision in delivery_placements
++-- only when it CHANGES, so a past delivery can be explained from history (GET /api/v1/placement). It is advisory
++-- history, pruned with the events retention; deliveries themselves carry the chosen carrier in deliveries.carrier
++-- (0002). ordering_key is nullable: a policy without ordering still gets a decision (per class, no key).
++CREATE TABLE delivery_placements (
++ id bigserial PRIMARY KEY,
++ carrier_class text NOT NULL,
++ ordering_key text,
++ carrier text NOT NULL,
++ result jsonb NOT NULL,
++ decided_at bigint NOT NULL
++);
++CREATE INDEX idx_delivery_placements_lookup ON delivery_placements(carrier_class, ordering_key, decided_at DESC);
ddiff --git a/server/api/placement.js b/server/api/placement.js
nnew file mode 100644
iindex 0000000..1a1561a
---- /dev/null
++++ b/server/api/placement.js
@@@ -0,0 +1,64 @@
++'use strict';
++/**
++ * Explain the fabric (ADR-042 decisions 1 and 5; plan T7) — operators only (events.delivery.admin), behind the same
++ * isAdmin guard and per-actor limits as /api/v1/deliveries.
++ *
++ * GET /api/v1/placement?event_type=<t>[&key=<k>]
++ * -> { event_type, policy, pattern, carrier_class, ordering_key, placement, last_decision }
++ * placement is platform.placement-result@1 exactly as the planner produced it (validated against the contract):
++ * non-empty reasons, and per candidate its eligibility / excluded_because / cost / latency.
++ * last_decision is { carrier, decided_at } from delivery_placements, or null before any decision changed.
++ * GET /api/v1/placement/deliveries/:event_id/:subscription_id
++ * -> { carrier, decision } (the delivery_placements row in force when the delivery was created, or null)
++ * 404 when the delivery is unknown.
++ */
++const express = require('express');
++const { http, validate } = require('openvibe-contracts');
++const { CAPS } = require('../auth');
++const { carrierClass, orderingKey } = require('../fabric/policy');
++
++function placementRouter({ planner, policies, auth, limits, store }) {
++ const router = express.Router();
++ const isAdmin = auth.requireCap(CAPS.admin);
++
++ router.get('/api/v1/placement', isAdmin, limits('events.placement.read'), async (req, res) => {
++ const ctx = req.ov;
++ 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 });
++ }
++ const { policy, pattern } = await policies.resolve(eventType);
++ const klass = carrierClass(policy);
++ const key = req.query.key != null ? String(req.query.key).slice(0, 200) : orderingKey(policy, { event_type: eventType });
++ const { result } = planner.explain(klass, key, policy);
++ const v = validate('platform.placement-result@1', result);
++ if (!v.valid) {
++ return http.sendProblem(res, 500, 'events.internal', { detail: `planner result does not match platform.placement-result@1: ${v.errors.map((e) => `${e.path} ${e.message}`).join('; ')}`, ctx });
++ }
++ const last = await planner.lastDecision(klass, key);
++ res.json({
++ event_type: eventType,
++ policy,
++ pattern,
++ carrier_class: klass,
++ ordering_key: key,
++ placement: result,
++ last_decision: last ? { carrier: last.carrier, decided_at: new Date(last.decided_at).toISOString() } : null,
++ });
++ });
++
++ router.get('/api/v1/placement/deliveries/:event_id/:subscription_id', isAdmin, limits('events.placement.read'), async (req, res) => {
++ const ctx = req.ov;
++ const d = await store.getDelivery(String(req.params.event_id), String(req.params.subscription_id));
++ if (!d) return http.sendProblem(res, 404, 'events.not_found', { detail: 'no such delivery', ctx });
++ const event = await store.getEvent(d.event_id);
++ const { policy } = event ? await policies.resolve(event.event_type) : { policy: null };
++ const klass = policy ? carrierClass(policy) : null;
++ const decision = klass ? await planner.decisionBefore(klass, d.ordering_key, d.created_at) : null;
++ res.json({ carrier: d.carrier || null, decision: decision || null });
++ });
++
++ return router;
++}
++
++module.exports = { placementRouter };
ddiff --git a/server/api/publish.js b/server/api/publish.js
iindex 1f79dde..a44aea0 100644
---- a/server/api/publish.js
++++ b/server/api/publish.js
@@@ -40,7 +40,7 @@ function normalize(input, ctx) {
return env;
}
--function publishRouter({ config, store, auth, worker, realtime, limits }) {
++function publishRouter({ config, store, auth, worker, realtime, limits, fabric = null }) {
const router = express.Router();
// Per-actor limits (server/actor-limits.js), in requests; a request carries up to 100 events.
@@@ -149,6 +149,9 @@ function publishRouter({ config, store, auth, worker, realtime, limits }) {
}
if (out.inserted.length) {
realtime.publish(out.inserted);
++ // A push carrier (ADR-042 decision 5) is signalled after the commit; a lost signal changes latency only —
++ // the worker's poll below is the floor. The local kick() keeps this process's own deliveries fast.
++ if (fabric) await fabric.signalPublish(out.inserted, out.deliveries);
worker.kick();
}
const status = out.inserted.length ? 201 : 200;
ddiff --git a/server/app.js b/server/app.js
iindex 7d5a600..3cdccc7 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -10,11 +10,12 @@ const { publishRouter } = require('./api/publish');
const { subscriptionsRouter } = require('./api/subscriptions');
const { readRouter } = require('./api/read');
const { policiesRouter } = require('./api/policies');
++const { placementRouter } = require('./api/placement');
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, policies = null }) {
++function createApp({ config, store, auth, keys, worker, realtime, metrics, dnsLookup, clock, log = console, valkey = null, policies = null, planner = null, carriers = null }) {
const app = express();
app.disable('x-powered-by');
app.set('trust proxy', 'loopback');
@@@ -76,10 +77,11 @@ function createApp({ config, store, auth, keys, worker, realtime, metrics, dnsLo
// One per-actor limiter for the capability routes below (server/actor-limits.js); health, ready,
// release.json, limits.json, metrics and the realtime stream above are never limited.
const limits = createLimits({ config, clock, metrics, log, valkey });
-- app.use(publishRouter({ config, store, auth, worker, realtime, limits }));
++ app.use(publishRouter({ config, store, auth, worker, realtime, limits, fabric: carriers }));
app.use(subscriptionsRouter({ config, store, auth, dnsLookup, limits }));
app.use(readRouter({ store, auth, worker, limits }));
if (policies) app.use(policiesRouter({ policies, auth, limits }));
++ if (planner) app.use(placementRouter({ planner, policies, auth, limits, store }));
app.get('/', (_req, res) => {
res.type('text/plain').send([
ddiff --git a/server/config.js b/server/config.js
iindex 0d63b16..a2cdef4 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -66,6 +66,13 @@ function load(env = process.env) {
// PostgreSQL (ADR-035): DATABASE_URL serves (PgBouncer), DATABASE_DIRECT_URL migrates (owner role).
db: { url: env.DATABASE_URL || '', directUrl: env.DATABASE_DIRECT_URL || '' },
++ // Carriers (ADR-042 decisions 1-2, 5; server/fabric/carriers): the adapters to instantiate and the ones the
++ // operator switch removes. pg-v1 is always present; valkey-v1 needs VALKEY_URL and is silently absent without
++ // it. A disabled adapter is ineligible at once (reason `disabled by configuration`): the ADR's rollback.
++ carriers: {
++ enabled: list(env.EVENTS_CARRIERS, ['pg-v1', 'valkey-v1']),
++ disabled: list(env.EVENTS_CARRIERS_DISABLED, []),
++ },
// Valkey (ADR-035): per-actor limit counters shared across processes; without it they count in this process.
valkey: { url: env.VALKEY_URL || '', prefix: env.VALKEY_PREFIX || 'ov:events:' },
retentionDays: int(env.EVENTS_RETENTION_DAYS, 30),
ddiff --git a/server/fabric/carriers/index.js b/server/fabric/carriers/index.js
nnew file mode 100644
iindex 0000000..5353bfb
---- /dev/null
++++ b/server/fabric/carriers/index.js
@@@ -0,0 +1,100 @@
++'use strict';
++/**
++ * The carrier registry (ADR-042 decisions 1-2, 5): one adapter per carrier, each with an id, the classes it can
++ * carry, an offer, health, a signal() and a start/stop. pg-v1 is always present and carries every class; valkey-v1
++ * is registered only when VALKEY_URL is set and carries QUEUE and TOPIC. nats-v1 is not built yet (it needs the host
++ * unit), so STREAM is left to pg-v1.
++ *
++ * Config (server/config.js):
++ * EVENTS_CARRIERS ids to instantiate (default pg-v1,valkey-v1). An unknown id refuses to start.
++ * EVENTS_CARRIERS_DISABLED ids excluded at once with reason `disabled by configuration` (the ADR's rollback).
++ *
++ * The planner asks offers(class) for one resource-offer@1 per eligible adapter and forClass(class) for the
++ * candidates to explain (including the excluded ones). Signal dispatch is by the chosen carrier id.
++ */
++const { createPgCarrier } = require('./pg');
++const { createValkeyCarrier, newInstanceId } = require('./valkey');
++
++const KNOWN = ['pg-v1', 'valkey-v1'];
++
++function createCarriers({ config, clock = { now: () => Date.now() }, log = console, valkey = null } = {}) {
++ const enabled = new Set(config.carriers.enabled);
++ for (const id of enabled) {
++ if (!KNOWN.includes(id)) throw new Error(`EVENTS_CARRIERS: unknown carrier adapter "${id}" (known: ${KNOWN.join(', ')})`);
++ }
++ const disabled = new Set(config.carriers.disabled);
++ const off = (id) => !enabled.has(id) || disabled.has(id);
++
++ const adapters = [createPgCarrier({ intervalMs: config.worker.intervalMs, now: () => clock.now() })];
++ if (valkey) {
++ const v = createValkeyCarrier({ valkey, clock, log, instanceId: newInstanceId() });
++ v.disabledReason = off('valkey-v1') ? 'disabled by configuration' : null;
++ adapters.push(v);
++ }
++ for (const a of adapters) if (!a.disabledReason && off(a.id)) a.disabledReason = 'disabled by configuration';
++
++ const get = (id) => adapters.find((a) => a.id === id) || null;
++ const forClass = (klass) => adapters.filter((a) => a.classes.has(klass));
++ const eligible = (klass) => forClass(klass).filter((a) => !a.disabledReason);
++
++ return {
++ adapters,
++ get,
++
++ /** One resource-offer@1 per eligible adapter of `klass`; an unhealthy adapter's health is `down`. */
++ offers(klass, now = clock.now()) {
++ return eligible(klass).map((a) => {
++ const o = a.offer(now);
++ if (!a.healthy()) o.health = { ...o.health, status: 'down', checked_at: new Date(now).toISOString() };
++ return o;
++ });
++ },
++
++ /** The eligible adapters for a class, plus the disabled ones to explain. */
++ forClass,
++ eligible,
++
++ /** Signal one carrier with rows (kind is 'QUEUE' or 'TOPIC'); fail open — never throws to the caller. */
++ async signal(carrierId, rows, kind) {
++ const a = get(carrierId);
++ if (!a || a.disabledReason || !rows || !rows.length) return;
++ try { await a.signal(rows, kind); } catch (err) { log.warn(`[carriers] ${carrierId} signal: ${err.message}`); }
++ },
++
++ /** After a publish commits: QUEUE deliveries to their stream, TOPIC events to their channel. The worker's
++ * kick() beside this is what makes a local process fast; a remote process is woken by the carrier. */
++ async signalPublish(events = [], deliveries = []) {
++ const byQueue = new Map();
++ for (const d of deliveries) {
++ if (d.carrier_class !== 'QUEUE' || !d.carrier) continue;
++ if (!byQueue.has(d.carrier)) byQueue.set(d.carrier, []);
++ byQueue.get(d.carrier).push(d);
++ }
++ for (const [cid, rows] of byQueue) await this.signal(cid, rows, 'QUEUE');
++
++ const byTopic = new Map();
++ for (const e of events) {
++ if (e.carrier_class !== 'TOPIC' || !e.carrier) continue;
++ if (!byTopic.has(e.carrier)) byTopic.set(e.carrier, []);
++ byTopic.get(e.carrier).push(e);
++ }
++ for (const [cid, rows] of byTopic) await this.signal(cid, rows, 'TOPIC');
++ },
++
++ async start(handlers) {
++ for (const a of adapters) { if (!a.disabledReason && a.start) await a.start(handlers); }
++ },
++ async stop() {
++ for (const a of adapters) { if (a.stop) await a.stop().catch(() => {}); }
++ },
++ snapshot() {
++ return adapters.map((a) => ({
++ id: a.id, classes: [...a.classes], disabled: a.disabledReason || null,
++ healthy: a.disabledReason ? false : a.healthy(),
++ signals: a.signals ? a.signals.snapshot() : {},
++ }));
++ },
++ };
++}
++
++module.exports = { createCarriers, KNOWN };
ddiff --git a/server/fabric/carriers/pg.js b/server/fabric/carriers/pg.js
nnew file mode 100644
iindex 0000000..5d8d67f
---- /dev/null
++++ b/server/fabric/carriers/pg.js
@@@ -0,0 +1,37 @@
++'use strict';
++/**
++ * pg-v1 — the poll (ADR-042 decision 2). Always present, every carrier class. A delivery is committed to
++ * `deliveries` before any carrier sees it; the worker's own tick (and its in-process kick()) carry it, so signal()
++ * does nothing. It is the floor a delivery falls back to when another carrier is unhealthy or ineligible.
++ *
++ * The offer prices the poll at its interval: half the interval is the expected extra latency beside a push carrier,
++ * which is what keeps valkey-v1 ahead whenever it is healthy.
++ */
++function createPgCarrier({ intervalMs = 500, now = () => Date.now() } = {}) {
++ const latency = Math.max(1, Math.round(intervalMs / 2));
++ return {
++ id: 'pg-v1',
++ classes: new Set(['TOPIC', 'QUEUE', 'STREAM']),
++ async start() { /* nothing to consume: the worker poll carries it */ },
++ async stop() {},
++ signal() { /* the poll and the local kick() are the signal */ },
++ healthy: () => true,
++ offer() {
++ return {
++ offer_id: 'pg-v1',
++ kind: 'provider',
++ provider: 'postgres',
++ adapter: 'pg-v1',
++ region: 'cell',
++ trust: 'first-party',
++ capabilities: ['events:gateway', 'events:durable', 'events:ordered'],
++ latency_ms: { publish_p95: latency, ack_p95: latency },
++ health: { status: 'up', checked_at: new Date(now()).toISOString() },
++ pricing: { model: 'per-operation', unit: 'delivery-operation', rate_card: 'rc-pg-v1' },
++ updated_at: new Date(now()).toISOString(),
++ };
++ },
++ };
++}
++
++module.exports = { createPgCarrier };
ddiff --git a/server/fabric/carriers/valkey.js b/server/fabric/carriers/valkey.js
nnew file mode 100644
iindex 0000000..2460680
---- /dev/null
++++ b/server/fabric/carriers/valkey.js
@@@ -0,0 +1,127 @@
++'use strict';
++/**
++ * valkey-v1 — push carriers for QUEUE and TOPIC (ADR-042 decisions 1 and 5), only when VALKEY_URL is set.
++ *
++ * QUEUE: after a publish commits, signal(rows) XADDs { event_id, subscription_id } — one entry per new delivery —
++ * to one Valkey stream per carrier class through openvibe-sdk/queue (consumer group `workers`). Every Events process
++ * consumes the group; for each item it asks the worker to claim exactly that one row (the same claimable + cap +
++ * re-check rules as the poll) and sends it if it got it. The item is acked either way (a null claim means someone
++ * else has it, it is not its key's head yet, or it is done: the 500 ms poll will find it when it is due). Nothing
++ * depends on the stream — a lost or duplicated item changes latency, never the outcome.
++ *
++ * TOPIC: signal(rows) PUBLISHes the stored envelopes on a Valkey channel tagged with this process's id. Every
++ * process subscribes and re-emits envelopes from OTHER processes to its own SSE clients, deduped by event id with a
++ * small LRU; visibility and topic matching are applied by the receiving gateway exactly as for local events. So an
++ * event published on one Events process reaches a browser connected to another.
++ *
++ * Health: an EWMA of signal/publish latency and a breaker on consecutive errors (../signals.js). healthy() is false
++ * while any op's breaker is open, which drops valkey-v1 from the plan with a reason; the poll then carries delivery.
++ */
++const os = require('os');
++const crypto = require('crypto');
++const { createQueue } = require('openvibe-sdk/queue');
++const { createPubSub } = require('openvibe-sdk/pubsub');
++const { createSignals } = require('../signals');
++
++const QUEUE_NAME = 'fabric.queue'; // one stream per carrier class (QUEUE)
++const TOPIC_CHANNEL = 'fabric.topic'; // one channel per carrier class (TOPIC)
++const SEEN_MAX = 5000; // event ids remembered for TOPIC dedupe
++
++function createValkeyCarrier({ valkey, clock = { now: () => Date.now() }, log = console, instanceId } = {}) {
++ const signals = createSignals({ now: () => clock.now() });
++ const queue = createQueue({ valkey, name: QUEUE_NAME, group: 'workers', log });
++ const pubsub = createPubSub({ valkey, log });
++ const seen = new Map(); // event id -> first-seen ms (small LRU)
++ let consumer = null;
++ let unsubscribe = null;
++
++ function markSeen(id) {
++ seen.set(id, clock.now());
++ if (seen.size > SEEN_MAX) {
++ const cut = clock.now() - 300000;
++ for (const [k, t] of seen) { if (t < cut) seen.delete(k); if (seen.size <= SEEN_MAX) break; }
++ }
++ }
++
++ async function signalQueue(rows) {
++ if (!consumer) return; // nothing consumes the stream (the delivery worker is off): the poll carries delivery
++ const t0 = clock.now();
++ try {
++ for (const r of rows) {
++ await queue.add({ event_id: r.event_id, subscription_id: r.subscription_id }, { id: `${r.event_id}:${r.subscription_id}` });
++ }
++ signals.record('queue', true, clock.now() - t0);
++ } catch (err) {
++ signals.record('queue', false);
++ log.warn(`[valkey-v1] queue signal: ${err.message}`);
++ }
++ }
++
++ async function signalTopic(rows) {
++ const t0 = clock.now();
++ try {
++ await pubsub.publish(TOPIC_CHANNEL, { from: instanceId, rows });
++ signals.record('topic', true, clock.now() - t0);
++ } catch (err) {
++ signals.record('topic', false);
++ log.warn(`[valkey-v1] topic publish: ${err.message}`);
++ }
++ }
++
++ return {
++ id: 'valkey-v1',
++ classes: new Set(['QUEUE', 'TOPIC']),
++
++ async start(handlers = {}) {
++ // The QUEUE consumer is a delivery path: it runs only when the delivery worker does (EVENTS_WORKER). With
++ // the worker off (a test driving its own worker, or an operator pausing delivery) there is no consumer and
++ // signal() enqueues nothing. The TOPIC subscription always runs: it is the realtime fan-out.
++ if (handlers.consumeQueue !== false) consumer = queue.process(async (data) => {
++ try {
++ if (!handlers.claim || !handlers.deliver) return;
++ const row = await handlers.claim(data && data.event_id, data && data.subscription_id);
++ if (row) await handlers.deliver(row);
++ } catch (err) {
++ log.warn(`[valkey-v1] delivery ${data && data.event_id}: ${err.message}`); // acked anyway: the poll retries
++ }
++ }, { concurrency: 1 }); // in stream order: a key's head before its successors (a non-head claim is acked and falls to the poll)
++ unsubscribe = await pubsub.subscribe(TOPIC_CHANNEL, (msg) => {
++ if (!msg || msg.from === instanceId) return;
++ const rows = (msg.rows || []).filter((r) => { if (seen.has(r.id)) return false; markSeen(r.id); return true; });
++ if (rows.length && handlers.onRemote) { try { handlers.onRemote(rows); } catch (err) { log.warn(`[valkey-v1] remote fan-out: ${err.message}`); } }
++ });
++ },
++
++ async stop() {
++ if (unsubscribe) { await unsubscribe().catch(() => {}); unsubscribe = null; }
++ await pubsub.close();
++ if (consumer) { await consumer.stop(); consumer = null; }
++ },
++
++ signal(rows, kind) { return kind === 'TOPIC' ? signalTopic(rows) : signalQueue(rows); },
++ healthy: () => signals.healthy('queue') && signals.healthy('topic'),
++ signals,
++
++ offer() {
++ const queueMs = signals.ewma('queue') || 2;
++ const topicMs = signals.ewma('topic') || 2;
++ return {
++ offer_id: 'valkey-v1',
++ kind: 'provider',
++ provider: 'valkey',
++ adapter: 'valkey-v1',
++ region: 'cell',
++ trust: 'first-party',
++ capabilities: ['events:gateway'],
++ latency_ms: { publish_p95: topicMs, ack_p95: queueMs },
++ health: { status: 'up', checked_at: new Date(clock.now()).toISOString() },
++ pricing: { model: 'per-operation', unit: 'signal-operation', rate_card: 'rc-valkey-v1' },
++ updated_at: new Date(clock.now()).toISOString(),
++ };
++ },
++ };
++}
++
++function newInstanceId() { return `${os.hostname()}:${process.pid}:${crypto.randomBytes(4).toString('hex')}`; }
++
++module.exports = { createValkeyCarrier, newInstanceId, QUEUE_NAME, TOPIC_CHANNEL };
ddiff --git a/server/fabric/planner.js b/server/fabric/planner.js
nnew file mode 100644
iindex 0000000..51a89c8
---- /dev/null
++++ b/server/fabric/planner.js
@@@ -0,0 +1,118 @@
++'use strict';
++/**
++ * The planner (ADR-042 decision 5): which carrier carries a (carrier class, ordering key). It calls
++ * openvibe-sdk/placement's plan() with requirements `{ kind: 'events.deliver', mobility: 'stateful-partition' }`
++ * plus the policy's latency/durability/ordering, against one resource-offer@1 per carrier adapter and the rate cards
++ * in ./rate-cards.json. plan() filters by hard constraints, then scores by cost and latency with hysteresis (a
++ * candidate replaces the current one only when it is clearly better, the SDK's default minGain) — so a decision does
++ * not flap.
++ *
++ * Decisions are cached per (class, key) in a bounded LRU (10k) and written to `delivery_placements` only when a
++ * decision CHANGES, so a past delivery can be explained from history. The chosen adapter id is written on every
++ * delivery at fan-out (store.js). `pg-v1` is the floor: whenever no candidate is eligible the plan falls back to it.
++ *
++ * Pinning (decision 4) is applied by the store, not here: while a (subscription, key) still has pending/failed
++ * deliveries on carrier X, new deliveries of that key stay on X; the store passes the pinned carrier through.
++ */
++const fs = require('fs');
++const path = require('path');
++const { plan } = require('openvibe-sdk/placement');
++
++const RATE_CARDS = JSON.parse(fs.readFileSync(path.join(__dirname, 'rate-cards.json'), 'utf8'));
++const CACHE_MAX = 10000;
++
++// The policy's routing objective vocabulary maps onto the platform objective enum.
++const OBJECTIVES = { latency_first: 'lowest-latency', cheapest: 'cheapest', balanced: 'balanced', correctness: 'correctness' };
++// A policy class maps to the placement latency class (a label the objective scoring can weigh).
++const LATENCY_CLASS = {
++ realtime_ephemeral: 'interactive', interactive: 'interactive', interactive_durable: 'interactive',
++ domain: 'interactive', critical: 'critical', background: 'background', bulk: 'bulk', webhook: 'interactive',
++};
++
++function requirements(klass, key, policy) {
++ const lat = (policy && policy.latency) || {};
++ const objective = OBJECTIVES[(policy && policy.routing && policy.routing.objective) || 'balanced'] || 'balanced';
++ return {
++ kind: 'events.deliver',
++ mobility: 'stateful-partition',
++ latency_class: LATENCY_CLASS[klass] || 'interactive',
++ objective,
++ partition_key: key || undefined,
++ capabilities: ['events:gateway'],
++ latency_op: 'ack_p95',
++ ...(Number.isFinite(lat.maximum_p95_ms) ? { max_latency_ms: lat.maximum_p95_ms } : {}),
++ units: 1,
++ };
++}
++
++function decisionRow(r) {
++ return r ? { carrier: r.carrier, decided_at: Number(r.decided_at), result: typeof r.result === 'string' ? JSON.parse(r.result) : r.result } : null;
++}
++
++function createPlanner({ registry, db = null, clock = { now: () => Date.now() }, log = console, cacheMax = CACHE_MAX } = {}) {
++ const cache = new Map(); // `${class}\0${key}` -> { carrier, result, at }
++
++ function decide(klass, key, policy, { update }) {
++ const ck = `${klass}\u0000${key || ''}`;
++ const current = cache.get(ck) || null;
++ const now = clock.now();
++ const excluded = [];
++ const offers = [];
++ for (const a of registry.forClass(klass)) {
++ if (a.disabledReason) { excluded.push({ id: a.id, eligible: false, excluded_because: a.disabledReason }); continue; }
++ const o = a.offer(now);
++ if (!a.healthy()) o.health = { ...o.health, status: 'down', checked_at: new Date(now).toISOString() };
++ offers.push(o);
++ }
++ const r = plan(requirements(klass, key, policy), offers, { rateCards: RATE_CARDS, now, current: current ? current.carrier : null });
++ // The offer's health already excludes an unhealthy adapter; name the breaker in the reason.
++ for (const c of r.candidates) {
++ const a = registry.get(c.id);
++ if (a && !a.disabledReason && !a.healthy() && !c.eligible) c.excluded_because = `breaker open (${a.id})`;
++ }
++ for (const e of excluded) if (!r.candidates.some((c) => c.id === e.id)) r.candidates.push(e);
++ if (!r.selected) {
++ const floor = registry.eligible(klass).find((a) => a.id === 'pg-v1') || registry.eligible(klass)[0];
++ r.selected = floor ? floor.id : 'pg-v1';
++ r.reasons.push(`no eligible candidate: falling back to ${r.selected}`);
++ }
++ const carrier = r.selected;
++ let record = null;
++ if (update) {
++ if (!current || current.carrier !== carrier) {
++ record = { carrier_class: klass, ordering_key: key || null, carrier, result: r, decided_at: now };
++ }
++ cache.set(ck, { carrier, result: r, at: now });
++ if (cache.size > cacheMax) { const oldest = cache.keys().next().value; cache.delete(oldest); }
++ }
++ return { carrier, result: r, record };
++ }
++
++ return {
++ rateCards: RATE_CARDS,
++ place: (klass, key, policy) => decide(klass, key, policy, { update: true }),
++ explain: (klass, key, policy) => decide(klass, key, policy, { update: false }),
++ clear: () => cache.clear(),
++ size: () => cache.size,
++
++ /** The most recent recorded decision for a (class, key), or null. */
++ async lastDecision(klass, key) {
++ if (!db) return null;
++ const r = await db.prepare(`SELECT carrier, decided_at, result FROM delivery_placements
++ WHERE carrier_class = ? AND ordering_key IS NOT DISTINCT FROM ? ORDER BY decided_at DESC, id DESC LIMIT 1`)
++ .get(klass, key == null ? null : key);
++ return decisionRow(r);
++ },
++
++ /** The decision in force when a delivery was created: the latest row at or before `ts`, or null. */
++ async decisionBefore(klass, key, ts) {
++ if (!db) return null;
++ const r = await db.prepare(`SELECT carrier, decided_at, result FROM delivery_placements
++ WHERE carrier_class = ? AND ordering_key IS NOT DISTINCT FROM ? AND decided_at <= ? ORDER BY decided_at DESC, id DESC LIMIT 1`)
++ .get(klass, key == null ? null : key, ts);
++ return decisionRow(r);
++ },
++ };
++}
++
++module.exports = { createPlanner, RATE_CARDS };
ddiff --git a/server/fabric/rate-cards.json b/server/fabric/rate-cards.json
nnew file mode 100644
iindex 0000000..2c17a1f
---- /dev/null
++++ b/server/fabric/rate-cards.json
@@@ -0,0 +1,30 @@
++[
++ {
++ "id": "rc-pg-v1",
++ "provider": "postgres",
++ "adapter": "pg-v1",
++ "metric": "delivery-operation",
++ "unit_size": 1,
++ "unit_price_usd": 0.0000005,
++ "free_allowance": 1000000000,
++ "reset_period": "month",
++ "region": "cell",
++ "effective_from": "2026-09-01",
++ "source": "https://openvibe.network/pricing/events",
++ "verified_at": "2026-09-01"
++ },
++ {
++ "id": "rc-valkey-v1",
++ "provider": "valkey",
++ "adapter": "valkey-v1",
++ "metric": "signal-operation",
++ "unit_size": 1,
++ "unit_price_usd": 0.0000002,
++ "free_allowance": 1000000000,
++ "reset_period": "month",
++ "region": "cell",
++ "effective_from": "2026-09-01",
++ "source": "https://openvibe.network/pricing/events",
++ "verified_at": "2026-09-01"
++ }
++]
ddiff --git a/server/fabric/signals.js b/server/fabric/signals.js
nnew file mode 100644
iindex 0000000..82cbe61
---- /dev/null
++++ b/server/fabric/signals.js
@@@ -0,0 +1,57 @@
++'use strict';
++/**
++ * Per-carrier rolling signals and a circuit breaker (ADR-042 decision 5). This is a local copy of the pattern in
++ * OpenVibe.Media's server/placement/signals.js — same idea (an EWMA of signal latency, a breaker on consecutive
++ * errors, half-open after a pause, closed on the next success) — kept here so Events never requires another
++ * repository at runtime. In-memory only: a restart re-learns from the next calls. `now` is injectable for tests.
++ *
++ * record(key, ok, ms) one observation
++ * healthy(key) false while the breaker is open (the planner drops the carrier with a reason)
++ * ewma(key) the current EWMA latency in ms (0 before any sample)
++ */
++const EWMA_ALPHA = 0.3;
++const OPEN_AFTER = 5; // consecutive failures that open the breaker
++const HALF_OPEN_AFTER_MS = 60000;
++
++function createSignals({ alpha = EWMA_ALPHA, openAfter = OPEN_AFTER, halfOpenMs = HALF_OPEN_AFTER_MS, now = () => Date.now() } = {}) {
++ const map = new Map(); // key -> { ewmaMs, streak, openAt, halfOpenAt, trial }
++
++ function get(key) {
++ let s = map.get(key);
++ if (!s) { s = { ewmaMs: 0, streak: 0, openAt: null, halfOpenAt: null, trial: false }; map.set(key, s); }
++ return s;
++ }
++
++ function record(key, ok, ms = 0) {
++ const s = get(key);
++ if (ok) {
++ if (ms > 0) s.ewmaMs = s.ewmaMs ? alpha * ms + (1 - alpha) * s.ewmaMs : ms;
++ s.streak = 0;
++ s.openAt = null; s.halfOpenAt = null; s.trial = false;
++ return;
++ }
++ s.streak += 1;
++ if (s.trial) { s.openAt = now(); s.halfOpenAt = null; s.trial = false; return; } // a failed trial re-opens at once
++ if (!s.openAt && s.streak >= openAfter) { s.openAt = now(); s.halfOpenAt = null; }
++ }
++
++ /** closed (healthy) | open (skipped) | half-open (one trial allowed). */
++ function state(key) {
++ const s = map.get(key);
++ if (!s || !s.openAt) return 'closed';
++ if (s.trial) return 'half-open';
++ if (s.halfOpenAt && now() >= s.halfOpenAt) return 'half-open';
++ if (now() - s.openAt < halfOpenMs) return 'open';
++ s.halfOpenAt = now();
++ return 'half-open';
++ }
++
++ function healthy(key) { return state(key) !== 'open'; }
++ function ewma(key) { const s = map.get(key); return s ? Math.round(s.ewmaMs * 1000) / 1000 : 0; }
++ function snapshot() { const out = {}; for (const k of map.keys()) out[k] = { state: state(k), ewma_ms: ewma(k) }; return out; }
++ function reset() { map.clear(); }
++
++ return { record, state, healthy, ewma, snapshot, reset };
++}
++
++module.exports = { createSignals, EWMA_ALPHA, OPEN_AFTER, HALF_OPEN_AFTER_MS };
ddiff --git a/server/index.js b/server/index.js
iindex fa7f7b5..606ff32 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -10,6 +10,8 @@
const { load } = require('./config');
const { openDb, createStore } = require('./store');
const { createPolicyLibrary } = require('./fabric/policy');
++const { createCarriers } = require('./fabric/carriers');
++const { createPlanner } = require('./fabric/planner');
const { createKeyStore, createAuth } = require('./auth');
const { createWorker } = require('./worker');
const { createRealtime } = require('./realtime');
@@@ -27,7 +29,13 @@ async function start({
// 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 });
++ // Valkey (before the carriers: valkey-v1 needs it). Without VALKEY_URL it is null and no valkey carrier registers.
++ const valkey = config.valkey.url ? require('openvibe-sdk/valkey').createValkey({ url: config.valkey.url, prefix: config.valkey.prefix, log }) : null;
++ // The carrier registry (ADR-042 decision 5) and the planner that chooses among them; the store writes the chosen
++ // carrier on every delivery at fan-out and records a (class, key) decision in delivery_placements when it changes.
++ const carriers = createCarriers({ config, clock, log, valkey });
++ const planner = createPlanner({ registry: carriers, db, clock, log });
++ const store = createStore(db, { clock, maxHops: config.maxHops, usage: config.usage, policies, planner });
const keys = createKeyStore({ urls: [config.networkInternalUrl, config.networkUrl], pem: config.networkPublicKey, fetchImpl, log });
const auth = createAuth({ config, keys, store });
const metrics = createMetrics({ store });
@@@ -36,13 +44,24 @@ async function start({
const worker = createWorker({ store, config, clock, fetchImpl: deliveryFetch, appPost: appPost || createGuardedPost({ lookup: dnsLookup }), log, observe: metrics.observe });
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, policies });
++ const app = createApp({ config, store, auth, keys, worker, realtime, metrics, dnsLookup, clock, log, valkey, policies, planner, carriers });
// The key loads in the background (retrying while Network boots); /api/ready says when it has.
const keyLoaded = keys.start().catch(() => null);
if (config.worker.enabled) worker.start();
realtime.start();
++ // The push carriers consume their streams and re-emit remote TOPIC events to this process's SSE clients
++ // (ADR-042 decision 5). A failure to start a carrier is not fatal: the poll (pg-v1) still carries delivery.
++ try {
++ await carriers.start({
++ consumeQueue: config.worker.enabled,
++ claim: (eventId, subscriptionId) => store.claimDelivery(eventId, subscriptionId, worker.lease()),
++ deliver: (delivery) => worker.deliver(delivery),
++ onRemote: (rows) => realtime.publish(rows),
++ });
++ } catch (err) {
++ log.warn(`[events] carriers did not start: ${err.message}`);
++ }
const prune = async () => {
try {
@@@ -91,6 +110,7 @@ async function start({
clearInterval(pruneTimer);
if (usageTimer) clearInterval(usageTimer);
policies.stop();
++ await carriers.stop().catch(() => {});
realtime.stop();
keys.stop();
await worker.stop();
@@@ -102,7 +122,7 @@ async function start({
if (!givenDb) await db.close();
}
-- return { config, db, store, keys, keyLoaded, auth, worker, realtime, metrics, app, server, policies, close, prune, flushUsage };
++ return { config, db, store, keys, keyLoaded, auth, worker, realtime, metrics, app, server, policies, carriers, planner, valkey, close, prune, flushUsage };
}
if (require.main === module) {
ddiff --git a/server/store.js b/server/store.js
iindex 8f77bee..926c6d7 100644
---- a/server/store.js
++++ b/server/store.js
@@@ -13,6 +13,7 @@ const topics = require('./topics');
const apps = require('./apps');
const redaction = require('./redaction');
const cursor = require('./cursor');
++const { carrierClass, orderingKey } = require('./fabric/policy');
const { createUsage, deliveryCode } = require('./usage');
const PRIORITY_RANK = { critical: 0, important: 1, low: 2 };
@@@ -91,7 +92,7 @@ function subscriptionView(row, { withSecret = false } = {}) {
return out;
}
--function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage: usageConfig = { enabled: true }, policies = null } = {}) {
++function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage: usageConfig = { enabled: true }, policies = null, planner = 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 +131,12 @@ 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)`),
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, ordering_key, created_at, updated_at) VALUES (?, ?, ?, ?, 0, 'pending', ?, ?, ?, ?) ON CONFLICT DO NOTHING`),
++ next_attempt_at, ordering_key, carrier, created_at, updated_at) VALUES (?, ?, ?, ?, 0, 'pending', ?, ?, ?, ?, ?) ON CONFLICT DO NOTHING`),
++ insertPlacement: db.prepare(`INSERT INTO delivery_placements (carrier_class, ordering_key, carrier, result, decided_at)
++ VALUES (?, ?, ?, ?, ?)`),
++ // The carrier a (subscription, ordering key) currently has a backlog on: what pins new deliveries of the key.
++ backlogCarrier: db.prepare(`SELECT carrier FROM deliveries WHERE subscription_id = ? AND ordering_key = ? AND status IN ('pending', 'failed')
++ AND carrier IS NOT NULL ORDER BY seq LIMIT 1`),
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 ?"),
};
@@@ -151,6 +157,8 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
const epoch = (await q.epoch.get()).value; // the retention epoch the stored positions belong to (ADR-042)
const results = [];
const inserted = [];
++ const deliveries = []; // the newly fanned-out deliveries, for the carrier signal after the commit
++ const placementRecords = []; // (class, key) decisions that changed, written in this transaction
const subs = await q.enabledSubs.all();
let recent = null;
let bytes = null;
@@@ -232,11 +240,25 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
// 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;
++ const resolved = policies ? await policies.resolve(env.event_type) : { policy: null };
++ const key = resolved.policy ? orderingKey(resolved.policy, env) : null;
++ const klass = resolved.policy ? carrierClass(resolved.policy) : null;
++ // The planner (ADR-042 decision 5) picks the carrier for this (class, key) and records a change. Its chosen
++ // carrier is written on every delivery; a decision that changes is recorded in delivery_placements once.
++ const decision = (planner && klass) ? planner.place(klass, key, resolved.policy) : null;
++ if (decision && decision.record) placementRecords.push(decision.record);
++ row.carrier_class = klass;
++ row.carrier = decision ? decision.carrier : 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, orderingKey, now, now);
++ if (!apps.deliverable(s, row)) continue;
++ // Pinning (decision 4): while this subscription has a backlog on carrier X for this key, stay on X.
++ let carrier = row.carrier;
++ if (planner && key && carrier) {
++ const pin = await q.backlogCarrier.get(s.id, key);
++ if (pin && pin.carrier) carrier = pin.carrier;
}
++ await q.insertDelivery.run(env.event_id, s.id, seq, PRIORITY_RANK[env.priority], now, key, carrier, now, now);
++ if (carrier) deliveries.push({ event_id: env.event_id, subscription_id: s.id, carrier, carrier_class: klass });
}
await revocationHook(row, now);
results.push({ event_id: env.event_id, seq, duplicate: false, cursor: cursor.encode(seq, epoch) });
@@@ -255,7 +277,11 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
if (project && inserted.length) {
await usage.record({ projectId: project.projectId, env: project.env, capability: 'events.app.publish', unit: 'events', quantity: inserted.length, at: now });
}
-- return { results, inserted };
++ // A (class, key) decision that changed is recorded once, in the same transaction as the deliveries it covers.
++ for (const rec of placementRecords) {
++ await q.insertPlacement.run(rec.carrier_class, rec.ordering_key, rec.carrier, JSON.stringify(rec.result), rec.decided_at);
++ }
++ return { results, inserted, deliveries };
});
/**
@@@ -332,6 +358,11 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
return await q.getEvent.get(id) || null;
}
++ /** The resolved delivery policy (../fabric/policy.js) for an event type, or null when no library is attached. */
++ async function resolvePolicy(eventType) {
++ return policies ? (await policies.resolve(eventType)).policy : null;
++ }
++
async function lastSeq() {
return (await q.lastSeq.get()).value;
}
@@@ -497,6 +528,40 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
});
};
++ /**
++ * Claim exactly one delivery (ADR-042 decision 5; the valkey-v1 QUEUE carrier calls this per stream item): the same
++ * claimable + cap + re-check rules as claimDeliveries, applied to the one (event_id, subscription_id). Returns the
++ * leased row (with app_project_id) or null when it is not the key's head yet, someone else holds it, it is at its
++ * subscription's inflight cap, it is not due, or it is already done. Nothing here depends on the caller: a null
++ * answer just means the 500 ms poll will find the delivery when it is due.
++ */
++ const claimDelivery = async (eventId, subscriptionId, { owner, leaseUntil } = {}) => {
++ const now = clock.now();
++ if (!owner || !(leaseUntil > now)) return null;
++ return await db.tx(async () => {
++ // Claims of one subscription are serialised on its row, as in claimDeliveries: without this lock a
++ // claim racing the poll re-checks its sub-selects against a snapshot taken before the other claim
++ // committed (READ COMMITTED re-evaluates only the updated row), and a delivery could be sent twice or a
++ // subscription exceed max_inflight. A subscription another worker is claiming for is skipped: that
++ // worker, or the next poll, takes the delivery.
++ const locked = await db.prepare('SELECT id FROM subscriptions WHERE id = ? AND enabled = 1 FOR UPDATE SKIP LOCKED').get(subscriptionId);
++ if (!locked) return null;
++ const r = await db.prepare(`UPDATE deliveries d SET lease_owner = ?, lease_until = ? WHERE d.event_id = ? AND d.subscription_id = ?
++ AND 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 > ?)
++ ))
++ AND (SELECT COUNT(*) FROM deliveries x WHERE x.subscription_id = d.subscription_id AND x.lease_until > ?)
++ < (SELECT s.max_inflight FROM subscriptions s WHERE s.id = d.subscription_id AND s.enabled = 1)
++ RETURNING d.*, (SELECT s.project_id FROM subscriptions s WHERE s.id = d.subscription_id) AS app_project_id`)
++ .get(owner, leaseUntil, eventId, subscriptionId, now, now, now);
++ return r || null;
++ });
++ };
++
/**
* 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
@@@ -565,15 +630,21 @@ 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);
// 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', ?, ?, ?, ?)
++ const up = db.prepare(`INSERT INTO deliveries (event_id, subscription_id, seq, priority, attempt, status, next_attempt_at, ordering_key, carrier, 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`);
++ last_error = NULL, last_status = NULL, delivered_at = NULL, carrier = excluded.carrier, updated_at = excluded.updated_at`);
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);
++ const envelope = { event_type: r.event_type, subject: { type: r.subject_type, id: r.subject_id }, actor: JSON.parse(r.actor), payload: JSON.parse(r.payload) };
++ const resolved = policies ? await policies.resolve(r.event_type) : { policy: null };
++ const key = resolved.policy ? orderingKey(resolved.policy, envelope) : null;
++ const klass = resolved.policy ? carrierClass(resolved.policy) : null;
++ // A replay is a fresh fan-out: the planner's current carrier is chosen (and a change recorded).
++ const decision = (planner && klass) ? planner.place(klass, key, resolved.policy) : null;
++ if (decision && decision.record) await q.insertPlacement.run(decision.record.carrier_class, decision.record.ordering_key, decision.record.carrier, JSON.stringify(decision.record.result), decision.record.decided_at);
++ let carrier = decision ? decision.carrier : null;
++ if (planner && key && carrier) { const pin = await q.backlogCarrier.get(sub.id, key); if (pin && pin.carrier) carrier = pin.carrier; }
++ await up.run(r.id, sub.id, r.seq, PRIORITY_RANK[r.priority], now, key, carrier, now, now);
}
return rows.length;
});
@@@ -616,7 +687,13 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
}
const receipts = (await db.prepare('DELETE FROM idempotency_receipts WHERE processed_at < ?')
.run(now - Math.max(receiptRetentionDays, retentionDays) * DAY_MS)).changes;
-- return { events, receipts };
++ // Placement history past the retention goes too, except each (carrier class, key)'s latest decision: it is still
++ // the one in force and explains deliveries made under it.
++ const placements = (await db.prepare(`DELETE FROM delivery_placements p WHERE p.decided_at < ? AND EXISTS (
++ SELECT 1 FROM delivery_placements q WHERE q.carrier_class = p.carrier_class
++ AND q.ordering_key IS NOT DISTINCT FROM p.ordering_key AND q.decided_at > p.decided_at)`)
++ .run(now - retentionDays * DAY_MS)).changes;
++ return { events, receipts, placements };
}
async function ping() {
@@@ -624,9 +701,9 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
}
return {
-- db, insertBatch, redact, getEvent, revokedAt, lastSeq, oldestSeq, epoch, bumpEpoch, firstSeqSince, scan,
++ db, insertBatch, redact, getEvent, resolvePolicy, revokedAt, lastSeq, oldestSeq, epoch, bumpEpoch, firstSeqSince, scan,
createSubscription, getSubscription, listSubscriptions, countSubscriptions, countProjectSubscriptions, setSubscriptionEnabled, rotateSubscriptionSecret,
-- claimDeliveries, recordAttempt, getDelivery, listDeliveries, requeue, deliveryCounts,
++ claimDeliveries, claimDelivery, recordAttempt, getDelivery, listDeliveries, requeue, deliveryCounts,
getCheckpoint, setCheckpoint, prune, ping, usage, flushUsage,
};
}
ddiff --git a/server/worker.js b/server/worker.js
iindex 1776106..1e1ff69 100644
---- a/server/worker.js
++++ b/server/worker.js
@@@ -110,8 +110,14 @@ function createWorker({ store, config, clock = { now: () => Date.now() }, fetchI
} catch (err) {
outcome = { ok: false, attempt, status: null, error: err.name === 'TimeoutError' ? 'timeout' : (err.message || String(err)), permanent: Boolean(err.permanent) };
}
++ // at_most_once (ADR-042 decisions 1 and 5): a failed attempt is final, never retried. The default policy is
++ // at_least_once, so existing subscriptions keep today's retry policy.
++ let atMostOnce = false;
++ if (!outcome.ok && store.resolvePolicy) {
++ try { const pol = await store.resolvePolicy(row.event_type); atMostOnce = Boolean(pol && pol.delivery_semantics === 'at_most_once'); } catch { atMostOnce = false; }
++ }
if (!outcome.ok) {
-- outcome.dead = outcome.permanent || attempt >= maxAttempts;
++ outcome.dead = outcome.permanent || atMostOnce || attempt >= maxAttempts;
if (!outcome.dead) outcome.nextAttemptAt = clock.now() + backoffMs[Math.min(attempt - 1, backoffMs.length - 1)];
if (outcome.dead) log.warn(`[worker] ${row.id} -> ${sub.id} dead after ${attempt} attempts: ${outcome.error}`);
}
@@@ -196,6 +202,9 @@ function createWorker({ store, config, clock = { now: () => Date.now() }, fetchI
start() { if (!running) { running = true; schedule(0); } },
async stop() { running = false; clearTimeout(timer); timer = null; await Promise.allSettled([...inflight]); },
kick() { schedule(0); },
++ // The carrier path (ADR-042 decision 5): claim one row (store.claimDelivery) and send it through the same send().
++ deliver(delivery) { return send(delivery); },
++ lease(now = clock.now()) { return { owner, leaseUntil: now + leaseMs() }; },
drain,
dispatch,
running: () => running,
ddiff --git a/test/fabric.test.js b/test/fabric.test.js
nnew file mode 100644
iindex 0000000..ddb5ae3
---- /dev/null
++++ b/test/fabric.test.js
@@@ -0,0 +1,306 @@
++'use strict';
++// ADR-042 decisions 1-5 (plan T7): carriers, the planner and the explain API, beside the poll. The registry maps
++// every policy class to exactly one carrier class, refuses an unknown adapter id, and honours the disable switch;
++// the planner picks a carrier with hysteresis and records a (class, key) decision that explains a past delivery;
++// valkey-v1 pushes QUEUE deliveries and TOPIC fan-out across processes while pg-v1 (the poll) stays underneath.
++// The Valkey parts need a running Valkey (npm run test:pg); on PGlite they print one `skipped (…)` line.
++const assert = require('assert');
++const { validate, ids } = require('openvibe-contracts');
++const { boot, request, serviceToken, userToken, envelope, subscriber, suite, sleep, manualClock, publicKey } = require('./helpers');
++const { testDb } = require('./db');
++const { load } = require('../server/config');
++const { createWorker } = require('../server/worker');
++const { createCarriers } = require('../server/fabric/carriers');
++const { createPlanner, RATE_CARDS } = require('../server/fabric/planner');
++const { carrierClass } = require('../server/fabric/policy');
++
++const t = suite('fabric');
++const silent = { log() {}, warn() {}, error() {} };
++const HAS_VALKEY = process.env.EVENTS_TEST_STORE === 'pg' && !!process.env.OV_TEST_VALKEY_URL;
++
++const admin = serviceToken('ops', ['events.delivery.admin']);
++const reader = serviceToken('media', ['events.event.read']);
++const live = serviceToken('live', ['events.event.publish']);
++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) => policy({ ordering: { scope: 'key', key } });
++
++// A valkey-shaped stub: the registry only constructs the queue/pubsub adapters from it, never connects in these tests.
++const fakeValkey = {
++ client: { ovQueuePromote() {}, defineCommand() {}, on() {} },
++ key: (...p) => `k:${p.join(':')}`,
++ duplicate() { return { on() {}, subscribe() {}, unsubscribe: async () => {}, disconnect() {} }; },
++};
++
++// A minimal registry for the planner's own tests: two tunable adapters over one class.
++function fakeAdapter(id, ms) {
++ return {
++ id, healthy: () => true, disabledReason: null,
++ offer: () => ({
++ offer_id: id, kind: 'provider', provider: id, region: 'cell', trust: 'first-party',
++ capabilities: ['events:gateway'], latency_ms: { ack_p95: ms }, health: { status: 'up' },
++ pricing: { model: 'per-operation', marginal_usd_per_unit: 0 }, updated_at: new Date(0).toISOString(),
++ }),
++ };
++}
++const fakeRegistry = (adapters) => ({
++ get: (id) => adapters.find((a) => a.id === id) || null,
++ forClass: () => adapters, eligible: () => adapters,
++});
++
++async function subscribe(base, slug, topic, url) {
++ const r = await request(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(base, env) {
++ const r = await request(base, 'POST', '/api/v1/events', { token: live, body: env });
++ assert.ok(r.status === 201 || r.status === 200, r.text);
++ return r.body;
++}
++async function putPolicy(base, pattern, body) {
++ const r = await request(base, 'PUT', `/api/v1/delivery-policies/${pattern}`, { token: admin, body });
++ assert.strictEqual(r.status, 200, r.text);
++ return r.body;
++}
++async function waitFor(pred, ms = 8000) {
++ const until = Date.now() + ms;
++ for (;;) { if (pred()) return; if (Date.now() > until) throw new Error('waitFor timed out'); await sleep(25); }
++}
++
++// ── Registry and policy classes ─────────────────────────────────────────────
++
++t('registry: every class -> one carrier class; unknown id refused; disabled valkey excluded with a reason', async () => {
++ // Every policy class maps to exactly one carrier class (the ADR table; also asserted in ordering.test.js).
++ for (const c of CLASSES) {
++ assert.strictEqual(carrierClass(policy({ class: c })), c === 'realtime_ephemeral' ? 'TOPIC' : 'QUEUE');
++ assert.strictEqual(carrierClass(keyed('subject', c)), 'STREAM');
++ }
++ // An unknown adapter id in config refuses to start.
++ assert.throws(() => createCarriers({ config: load({ NODE_ENV: 'test', EVENTS_CARRIERS: 'pg-v1,nats-v1' }) }), /unknown carrier adapter "nats-v1"/);
++
++ // The rate cards are valid platform.rate-card@1 (the model is exercised with a small per-op figure).
++ for (const card of RATE_CARDS) {
++ const v = validate('platform.rate-card@1', card);
++ assert.ok(v.valid, `${card.id}: ${JSON.stringify(v.errors)}`);
++ }
++
++ // EVENTS_CARRIERS_DISABLED=valkey-v1: excluded at once with reason `disabled by configuration`; pg-v1 chosen.
++ const reg = createCarriers({ config: load({ NODE_ENV: 'test', EVENTS_CARRIERS_DISABLED: 'valkey-v1' }), valkey: fakeValkey });
++ assert.strictEqual(reg.get('valkey-v1').disabledReason, 'disabled by configuration');
++ const planner = createPlanner({ registry: reg });
++ const ex = planner.explain('QUEUE', 'stream:K', policy({ class: 'critical', durability: 'none' }));
++ assert.strictEqual(ex.carrier, 'pg-v1');
++ const vk = ex.result.candidates.find((c) => c.id === 'valkey-v1');
++ assert.strictEqual(vk.eligible, false);
++ assert.strictEqual(vk.excluded_because, 'disabled by configuration');
++ assert.ok(ex.result.reasons.length > 0 && validate('platform.placement-result@1', ex.result).valid);
++});
++
++// ── The planner ─────────────────────────────────────────────────────────────
++
++t('planner: hysteresis holds a placement inside minGain; a much better candidate moves it', async () => {
++ const a = fakeAdapter('a', 10);
++ const b = fakeAdapter('b', 20);
++ const planner = createPlanner({ registry: fakeRegistry([a, b]) });
++ assert.strictEqual(planner.place('QUEUE', 'k', policy()).carrier, 'a', 'the lower-latency candidate wins');
++
++ b.offer = fakeAdapter('b', 9).offer; // 10% better: inside the SDK default minGain (15%)
++ assert.strictEqual(planner.place('QUEUE', 'k', policy()).carrier, 'a', 'hysteresis keeps the current placement');
++
++ b.offer = fakeAdapter('b', 1).offer; // far better: move
++ assert.strictEqual(planner.place('QUEUE', 'k', policy()).carrier, 'b', 'a clearly better candidate wins');
++});
++
++t('planner: an unhealthy valkey-v1 (breaker open) is excluded with a reason and pg-v1 is chosen', async () => {
++ const reg = createCarriers({ config: load({ NODE_ENV: 'test' }), valkey: fakeValkey });
++ const planner = createPlanner({ registry: reg });
++ assert.strictEqual(planner.place('QUEUE', 'k', policy()).carrier, 'valkey-v1', 'valkey-v1 is preferred while healthy');
++ const v = reg.get('valkey-v1');
++ for (let i = 0; i < 5; i++) { v.signals.record('queue', false); v.signals.record('topic', false); }
++ assert.strictEqual(v.healthy(), false, 'the breaker is open after consecutive errors');
++ const ex = planner.explain('QUEUE', 'k', policy());
++ assert.strictEqual(ex.carrier, 'pg-v1', 'pg-v1 delivers when valkey-v1 is down');
++ const cand = ex.result.candidates.find((c) => c.id === 'valkey-v1');
++ assert.strictEqual(cand.eligible, false);
++ assert.match(cand.excluded_because, /breaker open/);
++});
++
++// ── Explain, stored decisions, pinning, at_most_once ────────────────────────
++
++let h;
++// pg-v1 only: these tests are about the planner's choice, the stored decision and pinning, deterministically on the
++// poll; valkey-v1 has its own tests below (and the selection tests above use the registry directly).
++t('boot', async () => { h = await boot({ env: { EVENTS_CARRIERS: 'pg-v1' } }); });
++
++t('explain: shape, a stored decision explains a past delivery, and pinning keeps a backlog key', async () => {
++ const stub = await subscriber(() => 204);
++ const sub = await subscribe(h.base, 'fab', 'live.fab.*', stub.url);
++ // durability optional + an ordering key: the QUEUE class (STREAM needs durability required).
++ await putPolicy(h.base, 'live.fab.*', { class: 'domain', durability: 'optional', delivery_semantics: 'at_least_once', ordering: { scope: 'key', key: 'subject' } });
++
++ // E0 fans out (worker not run): a key with a backlog. Its carrier is forced to a sentinel to prove pinning reads
++ // the backlog's carrier, not the plan's choice.
++ const E0 = envelope('live', { event_type: 'live.fab.x', subject: { type: 'stream', id: 'K' } });
++ await publish(h.base, E0);
++ await h.store.db.prepare("UPDATE deliveries SET carrier = 'sentinel-v1' WHERE event_id = ? AND subscription_id = ?").run(E0.event_id, sub.id);
++
++ const E1 = envelope('live', { event_type: 'live.fab.x', subject: { type: 'stream', id: 'K' } });
++ await publish(h.base, E1);
++ const d1 = await h.store.getDelivery(E1.event_id, sub.id);
++ assert.strictEqual(d1.ordering_key, 'stream:K');
++ assert.strictEqual(d1.carrier, 'sentinel-v1', 'E1 is pinned to the carrier E0 still has a backlog on');
++
++ // The explain endpoint: validated placement-result@1, non-empty reasons, candidates, and the stored decision.
++ const ex = await request(h.base, 'GET', '/api/v1/placement?event_type=live.fab.x&key=stream:K', { token: admin });
++ assert.strictEqual(ex.status, 200, ex.text);
++ assert.strictEqual(ex.body.carrier_class, 'QUEUE');
++ assert.strictEqual(ex.body.ordering_key, 'stream:K');
++ assert.strictEqual(ex.body.pattern, 'live.fab.*');
++ assert.ok(ex.body.placement.reasons.length > 0, 'reasons are non-empty');
++ assert.ok(ex.body.placement.candidates.some((c) => c.id === 'pg-v1' && c.eligible));
++ assert.ok(validate('platform.placement-result@1', ex.body.placement).valid, JSON.stringify(ex.body.placement));
++ assert.ok(ex.body.last_decision && ex.body.last_decision.carrier === 'pg-v1', 'a decision was recorded');
++
++ // The delivery is explained from the decision in force when it was created.
++ const de = await request(h.base, 'GET', `/api/v1/placement/deliveries/${E1.event_id}/${sub.id}`, { token: admin });
++ assert.strictEqual(de.status, 200, de.text);
++ assert.strictEqual(de.body.carrier, 'sentinel-v1');
++ assert.ok(de.body.decision && de.body.decision.carrier === 'pg-v1', 'the recorded decision explains the delivery');
++
++ assert.strictEqual((await request(h.base, 'GET', '/api/v1/placement/deliveries/evt_nope/sub_nope', { token: admin })).status, 404);
++ assert.strictEqual((await request(h.base, 'GET', '/api/v1/placement?event_type=live.fab.x', { token: reader })).status, 403);
++ assert.strictEqual((await request(h.base, 'GET', '/api/v1/placement?event_type=live.fab.x')).status, 401);
++
++ await h.store.setSubscriptionEnabled(sub.id, false);
++ await stub.close();
++});
++
++t('at_most_once: one failed attempt is final (dead, no retry)', async () => {
++ const sub = await subscribe(h.base, 'amo', 'live.amo.*', 'http://127.0.0.1:9/hook');
++ await putPolicy(h.base, 'live.amo.*', { class: 'critical', durability: 'none', delivery_semantics: 'at_most_once' });
++ let attempts = 0;
++ const fetchImpl = async () => { attempts++; return new Response(null, { status: 500 }); };
++ const w = createWorker({ store: h.store, config: h.config, clock: h.clock, fetchImpl, log: silent });
++ const { event_id } = await publish(h.base, envelope('live', { event_type: 'live.amo.x', subject: { type: 'stream', id: 'A' } }));
++ await w.drain();
++ const d = await h.store.getDelivery(event_id, sub.id);
++ assert.strictEqual(attempts, 1, 'only one attempt');
++ assert.strictEqual(d.status, 'dead', 'a failed at_most_once attempt is dead, never retried');
++ assert.strictEqual(d.next_attempt_at, null);
++ await h.store.setSubscriptionEnabled(sub.id, false);
++});
++
++t('prune: placement history past the retention goes, except each key\'s latest decision', async () => {
++ const DAY = 86400000;
++ const now = Date.now() + 400 * DAY;
++ const ins = h.store.db.prepare('INSERT INTO delivery_placements (carrier_class, ordering_key, carrier, result, decided_at) VALUES (?, ?, ?, ?::jsonb, ?)');
++ for (const [key, age] of [['prune:a', 90], ['prune:a', 60], ['prune:a', 45], ['prune:b', 80]]) await ins.run('QUEUE', key, 'pg-v1', '{}', now - age * DAY);
++ const r = await h.store.prune({ retentionDays: 30, now });
++ assert.ok(r.placements >= 2, JSON.stringify(r));
++ const left = await h.store.db.prepare("SELECT ordering_key, decided_at FROM delivery_placements WHERE ordering_key LIKE 'prune:%' ORDER BY ordering_key").all();
++ assert.deepStrictEqual(left.map((x) => [x.ordering_key, Math.round((now - Number(x.decided_at)) / DAY)]), [['prune:a', 45], ['prune:b', 80]]);
++});
++
++t('stop', async () => { await h.stop(); });
++
++// ── valkey-v1 (Valkey only: test:pg) ────────────────────────────────────────
++
++t('valkey-v1 QUEUE: two instances, exactly once, per-key order, fast; a down carrier leaves nothing lost', async () => {
++ if (!HAS_VALKEY) return; // the skip line is printed once, below
++ const shared = await testDb();
++ const clock = manualClock();
++ const prefix = `ov:events-test:fabric:${Date.now()}:`;
++ const env = { VALKEY_URL: process.env.OV_TEST_VALKEY_URL, VALKEY_PREFIX: prefix, EVENTS_WORKER_INTERVAL_MS: '5000' };
++ const a = await boot({ db: shared.db, clock, env, worker: 'on' });
++ const b = await boot({ db: shared.db, clock, env, worker: 'on' });
++ const stub = await subscriber(() => 204);
++ try {
++ const sub = await subscribe(a.base, 'fabq', 'live.q.*', stub.url);
++ await putPolicy(a.base, 'live.q.*', keyed('subject'));
++ const KEYS = ['K0', 'K1', 'K2', 'K3', 'K4'];
++ const byId = new Map();
++ const firstSeq = new Map(KEYS.map((k) => [k, []]));
++ stub.calls.length = 0;
++ const seen = new Map();
++ // Publish 50 events over 5 keys, recording the publish time per event id.
++ const publishedAt = new Map();
++ for (let i = 0; i < 10; i++) {
++ for (const k of KEYS) {
++ const env2 = envelope('live', { event_type: 'live.q.x', subject: { type: 'stream', id: k } });
++ publishedAt.set(env2.event_id, Date.now());
++ await publish(a.base, env2);
++ }
++ }
++ // Wait for every delivery (carrier first; the 5 s poll is the floor).
++ await waitFor(() => {
++ for (const c of stub.calls) {
++ const id = c.body.event.event_id;
++ if (!seen.has(id)) { seen.set(id, { at: Date.now(), seq: c.body.seq }); byId.set(id, (byId.get(id) || 0) + 1); const L = firstSeq.get(c.body.event.subject.id); if (L && !L.includes(c.body.seq)) L.push(c.body.seq); }
++ }
++ return seen.size >= 50;
++ }, 9000);
++ assert.strictEqual(byId.size, 50, 'every event of every key was delivered');
++ assert.ok([...byId.values()].every((n) => n === 1), 'each delivery was sent exactly once');
++ for (const k of KEYS) { const L = firstSeq.get(k); assert.deepStrictEqual(L, [...L].sort((x, y) => x - y), `${k}: first delivery per seq is in order`); }
++ const lats = [...publishedAt.entries()].map(([id, at]) => seen.get(id).at - at).sort((x, y) => x - y);
++ const median = lats[Math.floor(lats.length / 2)];
++ assert.ok(median < 5000, `median publish->send latency ${median}ms is below the 5 s poll interval`);
++
++ // Take the carrier down (consecutive signal errors open the breaker): the remaining deliveries still arrive.
++ for (const inst of [a, b]) { const v = inst.carriers.get('valkey-v1'); for (let i = 0; i < 5; i++) { v.signals.record('queue', false); v.signals.record('topic', false); } assert.strictEqual(v.healthy(), false); }
++ const before = seen.size;
++ for (let i = 0; i < 5; i++) await publish(a.base, envelope('live', { event_type: 'live.q.x', subject: { type: 'stream', id: 'K9' } }));
++ await waitFor(() => {
++ for (const c of stub.calls) { const id = c.body.event.event_id; if (!seen.has(id)) { seen.set(id, { at: Date.now(), seq: c.body.seq }); byId.set(id, (byId.get(id) || 0) + 1); } }
++ return seen.size >= before + 5;
++ }, 12000);
++ assert.ok([...byId.values()].every((n) => n === 1), 'nothing was lost or doubled when the carrier went down');
++ await a.store.setSubscriptionEnabled(sub.id, false);
++ } finally {
++ await stub.close();
++ await a.stop(); await b.stop();
++ await shared.close();
++ }
++});
++
++t('valkey-v1 TOPIC: an event published on one instance reaches an SSE client on another, exactly once', async () => {
++ if (!HAS_VALKEY) return;
++ const shared = await testDb();
++ const clock = manualClock();
++ const prefix = `ov:events-test:fabric:${Date.now()}:`;
++ const env = { VALKEY_URL: process.env.OV_TEST_VALKEY_URL, VALKEY_PREFIX: prefix };
++ const a = await boot({ db: shared.db, clock, env, worker: 'off' });
++ const b = await boot({ db: shared.db, clock, env, worker: 'off' });
++ const { sse } = require('./helpers');
++ const serviceViewer = serviceToken('media', ['events.event.read']);
++ const cService = await sse(b.base, '/realtime/stream?topics=live.rt.*', { headers: { Authorization: `Bearer ${serviceViewer}` } });
++ const U1 = ids.newId('user');
++ const u1 = userToken({ subjectId: U1 });
++ const u2 = userToken({ subjectId: ids.newId('user') });
++ const c1 = await sse(b.base, '/realtime/stream?topics=live.rt.*', { headers: { Authorization: `Bearer ${u1}` } });
++ const c2 = await sse(b.base, '/realtime/stream?topics=live.rt.*', { headers: { Authorization: `Bearer ${u2}` } });
++ try {
++ await putPolicy(a.base, 'live.rt.*', { class: 'realtime_ephemeral', durability: 'none', delivery_semantics: 'at_least_once' });
++ const E = envelope('live', { event_type: 'live.rt.x', visibility: 'internal' });
++ await publish(a.base, E);
++ await cService.waitFor((c) => c.events().some((m) => m.event.event_id === E.event_id), 5000);
++ await sleep(200);
++ assert.strictEqual(cService.events().filter((m) => m.event.event_id === E.event_id).length, 1, 'exactly once on the remote gateway');
++
++ // A subject-visibility event reaches only its subject on the remote gateway.
++ const S = envelope('live', { event_type: 'live.rt.x', visibility: 'subject', subject: { type: 'user', id: U1 }, actor: { type: 'user', id: U1 } });
++ await publish(a.base, S);
++ await c1.waitFor((c) => c.events().some((m) => m.event.event_id === S.event_id), 5000);
++ await sleep(200);
++ assert.strictEqual(c2.events().filter((m) => m.event.event_id === S.event_id).length, 0, 'another subject sees nothing');
++ } finally {
++ cService.close(); c1.close(); c2.close();
++ await a.stop(); await b.stop();
++ await shared.close();
++ }
++});
++
++if (!HAS_VALKEY) console.log('fabric valkey: skipped (VALKEY_URL not set: run npm run test:pg with the test services up)');
++
++t.run();
ddiff --git a/test/helpers.js b/test/helpers.js
iindex 5a41e9e..c93719e 100644
---- a/test/helpers.js
++++ b/test/helpers.js
@@@ -72,7 +72,7 @@ function tmpDir() {
}
/** Boot the service. env overrides go through config.load(); `worker: 'manual'` keeps the loop off. */
--async function boot({ env = {}, clock = manualClock(), worker = 'manual', fetchImpl, deliveryFetch, appPost, dnsLookup } = {}) {
++async function boot({ env = {}, clock = manualClock(), worker = 'manual', fetchImpl, deliveryFetch, appPost, dnsLookup, db: sharedDb = null } = {}) {
const config = load({
NODE_ENV: 'test',
PORT: '0',
@@@ -83,8 +83,9 @@ async function boot({ env = {}, clock = manualClock(), worker = 'manual', fetchI
? { VALKEY_URL: process.env.OV_TEST_VALKEY_URL, VALKEY_PREFIX: `ov:events-test:${process.pid}:${ids.newId('event').slice(-8)}:` } : {}),
...env,
});
-- // One database per boot (PGlite, or EVENTS_TEST_STORE=pg: the containers), dropped when the boot stops.
-- const testdb = await require('./db').testDb();
++ // One database per boot (PGlite, or EVENTS_TEST_STORE=pg: the containers), dropped when the boot stops. A caller
++ // may pass a shared, already-migrated handle (several app instances on one database: the fabric tests).
++ const testdb = sharedDb ? { db: sharedDb, close: async () => {} } : await require('./db').testDb();
const h = await start({ config, db: testdb.db, clock, log: silent, fetchImpl, deliveryFetch, appPost, dnsLookup });
const base = `http://127.0.0.1:${h.server.address().port}`;
return {
ddiff --git a/test/lease.test.js b/test/lease.test.js
iindex 34a7e56..0ee4b4e 100644
---- a/test/lease.test.js
++++ b/test/lease.test.js
@@@ -78,5 +78,31 @@ t('a claim with no room, or a lease that ends now, claims nothing', async () =>
assert.deepStrictEqual(await h.store.claimDeliveries(now, 0, { owner: 'x', leaseUntil: now + 10 }), []);
});
++t('a targeted claim (the valkey-v1 path) racing the poll never double-leases a delivery', async () => {
++ const stub = await subscriber(() => 204);
++ const sub = await subscribe('leaserace', 'live.leaserace.*', stub.url);
++ for (let i = 0; i < 30; i++) await publish(envelope('live', { event_type: 'live.leaserace.x' }));
++ const pending = (await h.store.listDeliveries({ subscriptionId: sub.id, status: 'pending', limit: 100 })).sort((x, y) => Number(x.seq) - Number(y.seq));
++ assert.strictEqual(pending.length, 30);
++ let won = 0;
++ for (const d of pending) {
++ const now = h.clock.now();
++ const [poll, push] = await Promise.all([
++ h.store.claimDeliveries(now, 1, { owner: 'poll', leaseUntil: now + 60000 }),
++ h.store.claimDelivery(d.event_id, sub.id, { owner: 'push', leaseUntil: now + 60000 }),
++ ]);
++ const winners = (poll.length ? 1 : 0) + (push ? 1 : 0);
++ assert.ok(winners <= 1, `delivery ${d.event_id}: both the poll and the targeted claim leased it`);
++ const owner = poll.length ? 'poll' : push ? 'push' : null;
++ const leased = poll.length ? poll[0] : push;
++ if (!leased) continue;
++ won++;
++ assert.strictEqual(await h.store.recordAttempt(leased.event_id, sub.id, { ok: true, attempt: 1, status: 204 }, null, owner), true);
++ }
++ assert.ok(won > 0, 'the claims made progress');
++ await stub.close();
++ await h.store.setSubscriptionEnabled(sub.id, false);
++});
++
t('stop', async () => { await h.stop(); });
t.run();