openvibers/openvibe.events Commits

Observability (Track O): GET /metrics (loopback only) with HTTP golden signals by route template, process metrics, release_info and Events gauges (deliveries by status, DLQ depth, latest seq, SSE connections, delivery latency, attempts by outcome); /api/ready built on openvibe-shared/ready (db, network_jwks, delivery_worker required; dlq optional -> degraded); GET /release.json; openvibe-shared v1.3.0

OP OpenVibers committed 23 Sept 2026 · 0c7f9c58dc · 13 files, +348 −29


Changes

ddiff --git a/.env.example b/.env.example
iindex cd30b94..42dc4ca 100644
---- a/.env.example
++++ b/.env.example
@@@ -47,6 +47,8 @@ EVENTS_RECEIPT_RETENTION_DAYS=90
#EVENTS_DELIVERY_TIMEOUT_MS=10000
#EVENTS_MAX_ATTEMPTS=8
#EVENTS_BACKOFF_MS=1000,5000,30000,120000,600000,3600000
++# /api/ready lists `dlq` as degraded (the service stays ready) above this many dead deliveries.
++#EVENTS_DLQ_DEGRADED_AT=100
#EVENTS_MAX_SUBSCRIPTIONS=100
# Hostnames a subscription endpoint may use (SSRF guard). `*.openvibe.*` = openvibe.<tld>
# and its subdomains; IP literals only when listed exactly.
ddiff --git a/README.md b/README.md
iindex 7e92a56..cc47a8b 100644
---- a/README.md
++++ b/README.md
@@@ -22,7 +22,9 @@ npm test # every test/*.test.js, temp databases, no network needed
Node 22 in production (`fnm exec --using=22.22.1 npm test`). Production: `/opt/openvibe.events`, env `/etc/openvibe/events.env`, unit [deploy/systemd/openvibe-events.service](deploy/systemd/openvibe-events.service), store `/var/lib/openvibe-events/events.db`, nginx [deploy/nginx/events.openvibe.network.conf](deploy/nginx/events.openvibe.network.conf) (public vhost exposes `/realtime/stream` and health only; services on the host call `127.0.0.1:4300`).
--`GET /api/health` is liveness. `GET /api/ready` is 200 only when the database answers, the delivery worker runs and the Network signing key has loaded (it retries every 30 s while Network boots).
++`GET /api/health` is liveness. `GET /api/ready` (openvibe-shared/ready) is 200 only when every required check passes — `db` (a real query), `network_jwks` (the Network signing key has loaded; it retries every 30 s while Network boots) and `delivery_worker` (when `EVENTS_WORKER` is on) — and 503 otherwise, with `failed: [...]`. The optional `dlq` check fails once more than `EVENTS_DLQ_DEGRADED_AT` (default 100) deliveries are dead: the service stays ready (200) and reports `status: "degraded"`, `degraded: ["dlq"]`. Each check carries `status`, `required`, `latency_ms`, `checked_at` and, for `dlq`, `detail: { depth, threshold }`; `latest_seq`, `deliveries`, `worker` and `realtime_connections` are still in the body. **Shape change (Track O):** `checks` used to be booleans (`{ db, worker, key }`); they are now objects keyed `db`, `network_jwks`, `delivery_worker`, `dlq`.
++
++`GET /metrics` (Prometheus text, direct loopback callers only — a request carrying `X-Forwarded-For` gets 404, and nginx refuses the path): HTTP golden signals by route template (`http_requests_total`, `http_request_duration_seconds`, `http_requests_in_flight`; SSE sessions excluded), process metrics, `release_info`, and `events_deliveries{status}`, `events_dlq_depth`, `events_latest_seq`, `events_realtime_connections`, `events_delivery_latency_seconds{priority}` (acceptance → successful delivery, retries included), `events_delivery_attempts_total{outcome}`. `GET /release.json` is the deployed release (ADR-016).
## Auth
ddiff --git a/deploy/nginx/events.openvibe.network.conf b/deploy/nginx/events.openvibe.network.conf
iindex c1712bc..fdff4a5 100644
---- a/deploy/nginx/events.openvibe.network.conf
++++ b/deploy/nginx/events.openvibe.network.conf
@@@ -6,7 +6,8 @@
##
## The Node service (port 4300, systemd unit openvibe-events) serves:
## /realtime/stream browser Server-Sent Events (long-lived; must not be buffered)
--## /api/health, /api/ready
++## /api/health, /api/ready, /release.json
++## /metrics Prometheus text, answered only to direct loopback callers (never proxied)
## /api/v1/* service-to-service API. Services on this host call 127.0.0.1:4300
## directly, so the public vhost refuses it; open the location below only
## when a service on another host needs it (it still requires a token).
@@@ -68,8 +69,14 @@ server {
send_timeout 1h;
}
-- location = /api/health { proxy_pass http://127.0.0.1:4300; proxy_set_header Host $host; }
-- location = /api/ready { proxy_pass http://127.0.0.1:4300; proxy_set_header Host $host; }
++ # X-Forwarded-For on every proxied location: the service treats a request without it from
++ # 127.0.0.1 as a local caller (GET /metrics answers only those).
++ location = /api/health { proxy_pass http://127.0.0.1:4300; proxy_set_header Host $host; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; }
++ location = /api/ready { proxy_pass http://127.0.0.1:4300; proxy_set_header Host $host; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; }
++ location = /release.json { proxy_pass http://127.0.0.1:4300; proxy_set_header Host $host; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; }
++
++ # Metrics are for host-local scrapers only (http://127.0.0.1:4300/metrics).
++ location = /metrics { return 404; }
# ── Service API: host-local only (see header) ───────────
location /api/ {
@@@ -85,6 +92,7 @@ server {
location = / {
proxy_pass http://127.0.0.1:4300;
proxy_set_header Host $host;
++ proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
}
location / {
ddiff --git a/package-lock.json b/package-lock.json
iindex aabab88..d4cad30 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -12,7 +12,8 @@
"better-sqlite3": "^11.10.0",
"dotenv": "^16.4.7",
"express": "^4.21.2",
-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.6.0"
++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.6.0",
++ "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.3.0"
},
"engines": {
"node": ">=20.0.0"
@@@ -169,6 +170,12 @@
"ieee754": "^1.1.13"
}
},
++ "node_modules/buffer-equal-constant-time": {
++ "version": "1.0.1",
++ "resolved": "https://registry.npmjs.org/buffer-equal-constant-time/-/buffer-equal-constant-time-1.0.1.tgz",
++ "integrity": "sha512-zRpUiDwd/xk6ADqPMATG8vc9VPrkck7T07OIx0gnjmJAnHnTVXNQG3vfvWNuiZIkwu9KrKdA1iJKfsfTVxE6NA==",
++ "license": "BSD-3-Clause"
++ },
"node_modules/bytes": {
"version": "3.1.2",
"resolved": "https://registry.npmjs.org/bytes/-/bytes-3.1.2.tgz",
@@@ -336,6 +343,15 @@
"node": ">= 0.4"
}
},
++ "node_modules/ecdsa-sig-formatter": {
++ "version": "1.0.11",
++ "resolved": "https://registry.npmjs.org/ecdsa-sig-formatter/-/ecdsa-sig-formatter-1.0.11.tgz",
++ "integrity": "sha512-nagl3RYrbNv6kQkeJIpt6NJZy8twLB/2vtz6yN9Z4vRKHN4/QZJIEbqohALSgwKdnksuY3k5Addp5lg8sVoVcQ==",
++ "license": "Apache-2.0",
++ "dependencies": {
++ "safe-buffer": "^5.0.1"
++ }
++ },
"node_modules/ee-first": {
"version": "1.1.1",
"resolved": "https://registry.npmjs.org/ee-first/-/ee-first-1.1.1.tgz",
@@@ -697,6 +713,97 @@
"integrity": "sha512-NM8/P9n3XjXhIZn1lLhkFaACTOURQXjWhV4BA/RnOv8xvgqtqpAX9IO4mRQxSx1Rlo4tqzeqb0sOlruaOy3dug==",
"license": "MIT"
},
++ "node_modules/jsonwebtoken": {
++ "version": "9.0.3",
++ "resolved": "https://registry.npmjs.org/jsonwebtoken/-/jsonwebtoken-9.0.3.tgz",
++ "integrity": "sha512-MT/xP0CrubFRNLNKvxJ2BYfy53Zkm++5bX9dtuPbqAeQpTVe0MQTFhao8+Cp//EmJp244xt6Drw/GVEGCUj40g==",
++ "license": "MIT",
++ "dependencies": {
++ "jws": "^4.0.1",
++ "lodash.includes": "^4.3.0",
++ "lodash.isboolean": "^3.0.3",
++ "lodash.isinteger": "^4.0.4",
++ "lodash.isnumber": "^3.0.3",
++ "lodash.isplainobject": "^4.0.6",
++ "lodash.isstring": "^4.0.1",
++ "lodash.once": "^4.0.0",
++ "ms": "^2.1.1",
++ "semver": "^7.5.4"
++ },
++ "engines": {
++ "node": ">=12",
++ "npm": ">=6"
++ }
++ },
++ "node_modules/jsonwebtoken/node_modules/ms": {
++ "version": "2.1.3",
++ "resolved": "https://registry.npmjs.org/ms/-/ms-2.1.3.tgz",
++ "integrity": "sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA==",
++ "license": "MIT"
++ },
++ "node_modules/jwa": {
++ "version": "2.0.1",
++ "resolved": "https://registry.npmjs.org/jwa/-/jwa-2.0.1.tgz",
++ "integrity": "sha512-hRF04fqJIP8Abbkq5NKGN0Bbr3JxlQ+qhZufXVr0DvujKy93ZCbXZMHDL4EOtodSbCWxOqR8MS1tXA5hwqCXDg==",
++ "license": "MIT",
++ "dependencies": {
++ "buffer-equal-constant-time": "^1.0.1",
++ "ecdsa-sig-formatter": "1.0.11",
++ "safe-buffer": "^5.0.1"
++ }
++ },
++ "node_modules/jws": {
++ "version": "4.0.1",
++ "resolved": "https://registry.npmjs.org/jws/-/jws-4.0.1.tgz",
++ "integrity": "sha512-EKI/M/yqPncGUUh44xz0PxSidXFr/+r0pA70+gIYhjv+et7yxM+s29Y+VGDkovRofQem0fs7Uvf4+YmAdyRduA==",
++ "license": "MIT",
++ "dependencies": {
++ "jwa": "^2.0.1",
++ "safe-buffer": "^5.0.1"
++ }
++ },
++ "node_modules/lodash.includes": {
++ "version": "4.3.0",
++ "resolved": "https://registry.npmjs.org/lodash.includes/-/lodash.includes-4.3.0.tgz",
++ "integrity": "sha512-W3Bx6mdkRTGtlJISOvVD/lbqjTlPPUDTMnlXZFnVwi9NKJ6tiAk6LVdlhZMm17VZisqhKcgzpO5Wz91PCt5b0w==",
++ "license": "MIT"
++ },
++ "node_modules/lodash.isboolean": {
++ "version": "3.0.3",
++ "resolved": "https://registry.npmjs.org/lodash.isboolean/-/lodash.isboolean-3.0.3.tgz",
++ "integrity": "sha512-Bz5mupy2SVbPHURB98VAcw+aHh4vRV5IPNhILUCsOzRmsTmSQ17jIuqopAentWoehktxGd9e/hbIXq980/1QJg==",
++ "license": "MIT"
++ },
++ "node_modules/lodash.isinteger": {
++ "version": "4.0.4",
++ "resolved": "https://registry.npmjs.org/lodash.isinteger/-/lodash.isinteger-4.0.4.tgz",
++ "integrity": "sha512-DBwtEWN2caHQ9/imiNeEA5ys1JoRtRfY3d7V9wkqtbycnAmTvRRmbHKDV4a0EYc678/dia0jrte4tjYwVBaZUA==",
++ "license": "MIT"
++ },
++ "node_modules/lodash.isnumber": {
++ "version": "3.0.3",
++ "resolved": "https://registry.npmjs.org/lodash.isnumber/-/lodash.isnumber-3.0.3.tgz",
++ "integrity": "sha512-QYqzpfwO3/CWf3XP+Z+tkQsfaLL/EnUlXWVkIk5FUPc4sBdTehEqZONuyRt2P67PXAk+NXmTBcc97zw9t1FQrw==",
++ "license": "MIT"
++ },
++ "node_modules/lodash.isplainobject": {
++ "version": "4.0.6",
++ "resolved": "https://registry.npmjs.org/lodash.isplainobject/-/lodash.isplainobject-4.0.6.tgz",
++ "integrity": "sha512-oSXzaWypCMHkPC3NvBEaPHf0KsA5mvPrOPgQWDsbg8n7orZ290M0BmC/jgRZ4vcJ6DTAhjrsSYgdsW/F+MFOBA==",
++ "license": "MIT"
++ },
++ "node_modules/lodash.isstring": {
++ "version": "4.0.1",
++ "resolved": "https://registry.npmjs.org/lodash.isstring/-/lodash.isstring-4.0.1.tgz",
++ "integrity": "sha512-0wJxfxH1wgO3GrbuP+dTTk7op+6L41QCXbGINEmD+ny/G/eCqGzxyCsh7159S+mgDDcoarnBw6PC1PS5+wUGgw==",
++ "license": "MIT"
++ },
++ "node_modules/lodash.once": {
++ "version": "4.1.1",
++ "resolved": "https://registry.npmjs.org/lodash.once/-/lodash.once-4.1.1.tgz",
++ "integrity": "sha512-Sb487aTOCr9drQVL8pIxOzVhafOjZN9UU54hiN8PU3uAiSV7lx1yYNpbNmex2PK6dSJoNTSJUUswT651yww3Mg==",
++ "license": "MIT"
++ },
"node_modules/math-intrinsics": {
"version": "1.1.0",
"resolved": "https://registry.npmjs.org/math-intrinsics/-/math-intrinsics-1.1.0.tgz",
@@@ -875,6 +982,14 @@
"node": ">=20"
}
},
++ "node_modules/openvibe-shared": {
++ "version": "1.3.0",
++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.3.0",
++ "license": "MIT",
++ "dependencies": {
++ "jsonwebtoken": "^9.0.0"
++ }
++ },
"node_modules/parseurl": {
"version": "1.3.3",
"resolved": "https://registry.npmjs.org/parseurl/-/parseurl-1.3.3.tgz",
ddiff --git a/package.json b/package.json
iindex 7530bca..96284be 100644
---- a/package.json
++++ b/package.json
@@@ -17,7 +17,8 @@
"better-sqlite3": "^11.10.0",
"dotenv": "^16.4.7",
"express": "^4.21.2",
-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.6.0"
++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.6.0",
++ "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.3.0"
},
"engines": {
"node": ">=20.0.0"
ddiff --git a/server/app.js b/server/app.js
iindex 2ea532a..72a70fc 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -1,17 +1,26 @@
'use strict';
--/** Express app: request context, the v1 API, the realtime gateway, health/readiness. */
++/** Express app: request context, the v1 API, the realtime gateway, health/readiness, metrics. */
++const path = require('path');
const express = require('express');
++const { instrument } = require('openvibe-shared/metrics');
++const { createReadiness } = require('openvibe-shared/ready');
++const { createRelease } = require('openvibe-shared/release');
const { http } = require('openvibe-contracts');
const { publishRouter } = require('./api/publish');
const { subscriptionsRouter } = require('./api/subscriptions');
const { readRouter } = require('./api/read');
const pkg = require('../package.json');
--function createApp({ config, store, auth, keys, worker, realtime, log = console }) {
++function createApp({ config, store, auth, keys, worker, realtime, metrics, log = console }) {
const app = express();
app.disable('x-powered-by');
app.set('trust proxy', 'loopback');
++ const release = createRelease({ service: 'events', root: path.join(__dirname, '..') });
++ // HTTP golden signals by route template + GET /metrics (direct loopback callers only). An SSE
++ // connection is a session, not a request, so it is counted by events_realtime_connections instead.
++ instrument(app, { service: 'events', release: release.release, registry: metrics && metrics.registry, skip: (req) => req.path === '/realtime/stream' });
app.use(http.middleware());
++ app.get('/release.json', release.handler);
app.use((req, res, next) => {
res.setHeader('X-Content-Type-Options', 'nosniff');
next();
@@@ -27,20 +36,34 @@ function createApp({ config, store, auth, keys, worker, realtime, log = console
res.json({ status: 'ok', service: 'openvibe-events', version: pkg.version });
});
-- app.get('/api/ready', (_req, res) => {
-- let dbOk = false;
-- try { dbOk = store.ping(); } catch { dbOk = false; }
-- const checks = { db: dbOk, worker: !config.worker.enabled || worker.running(), key: keys.loaded() };
-- const ready = Object.values(checks).every(Boolean);
-- res.status(ready ? 200 : 503).json({
-- status: ready ? 'ready' : 'not_ready',
-- checks,
-- latest_seq: dbOk ? store.lastSeq() : null,
-- deliveries: dbOk ? store.deliveryCounts() : null,
-- worker: worker.stats(),
-- realtime_connections: realtime.count(),
-- });
++ // Readiness (openvibe-shared/ready): 503 only when a required check fails. A DLQ past
++ // EVENTS_DLQ_DEGRADED_AT degrades the service (still ready: new events are accepted and delivered).
++ const checks = [
++ { name: 'db', required: true, check: () => store.ping() },
++ { name: 'network_jwks', required: true, check: () => keys.loaded() || 'Network signing key not loaded yet' },
++ ];
++ if (config.worker.enabled) checks.push({ name: 'delivery_worker', required: true, check: () => worker.running() || 'delivery worker is not running' });
++ checks.push({
++ name: 'dlq', required: false, check: () => {
++ const depth = store.deliveryCounts().dead;
++ return depth > config.dlqDegradedAt
++ ? { ok: false, error: `${depth} dead deliveries (threshold ${config.dlqDegradedAt})`, detail: { depth, threshold: config.dlqDegradedAt } }
++ : { ok: true, detail: { depth, threshold: config.dlqDegradedAt } };
++ },
});
++ const readiness = createReadiness({
++ service: 'events', release: release.release, checks,
++ details: (body) => {
++ const dbOk = body.checks.db.status === 'ok';
++ return {
++ latest_seq: dbOk ? store.lastSeq() : null,
++ deliveries: dbOk ? store.deliveryCounts() : null,
++ worker: { enabled: config.worker.enabled, ...worker.stats() },
++ realtime_connections: realtime.count(),
++ };
++ },
++ });
++ app.get('/api/ready', readiness.handler);
app.use(publishRouter({ config, store, auth, worker, realtime }));
app.use(subscriptionsRouter({ config, store, auth }));
@@@ -55,7 +78,7 @@ function createApp({ config, store, auth, keys, worker, realtime, log = console
' /api/v1/subscriptions webhook subscriptions (events.subscription.manage)',
' /api/v1/deliveries DLQ inspect and replay (events.delivery.admin)',
'GET /realtime/stream?topics=... server-sent events for browsers',
-- 'GET /api/health, /api/ready',
++ 'GET /api/health, /api/ready, /release.json',
'',
'Source: https://github.com/OpenVibers/OpenVibe.Events',
'',
ddiff --git a/server/config.js b/server/config.js
iindex c817aca..aad8ed8 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -81,6 +81,8 @@ function load(env = process.env) {
// Wait after attempt n fails = backoffMs[n-1] (the last value repeats).
backoffMs: list(env.EVENTS_BACKOFF_MS, ['1000', '5000', '30000', '120000', '600000', '3600000']).map(Number),
},
++ // /api/ready reports `dlq` degraded (still ready) once more dead deliveries than this wait for replay.
++ dlqDegradedAt: int(env.EVENTS_DLQ_DEGRADED_AT, 100),
maxSubscriptionsPerConsumer: int(env.EVENTS_MAX_SUBSCRIPTIONS, 100),
// Hostname patterns a subscription endpoint may point at (SSRF guard).
endpointHosts: list(env.EVENTS_ENDPOINT_HOSTS, ['127.0.0.1', '*.openvibe.*']),
ddiff --git a/server/index.js b/server/index.js
iindex bf233b2..0a5e992 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -13,6 +13,7 @@ const { createKeyStore, createAuth } = require('./auth');
const { createWorker } = require('./worker');
const { createRealtime } = require('./realtime');
const { createApp } = require('./app');
++const { createMetrics } = require('./metrics');
async function start({ config, clock = { now: () => Date.now() }, fetchImpl = globalThis.fetch, deliveryFetch = fetchImpl, log = console, listen = true } = {}) {
config = config || load();
@@@ -20,9 +21,11 @@ async function start({ config, clock = { now: () => Date.now() }, fetchImpl = gl
const store = createStore(db, { clock, maxHops: config.maxHops });
const keys = createKeyStore({ urls: [config.networkInternalUrl, config.networkUrl], pem: config.networkPublicKey, fetchImpl, log });
const auth = createAuth({ config, keys });
-- const worker = createWorker({ store, config, clock, fetchImpl: deliveryFetch, log });
++ const metrics = createMetrics({ store });
++ const worker = createWorker({ store, config, clock, fetchImpl: deliveryFetch, log, observe: metrics.observe });
const realtime = createRealtime({ store, auth, config, log });
-- const app = createApp({ config, store, auth, keys, worker, realtime, log });
++ metrics.bind({ realtime });
++ const app = createApp({ config, store, auth, keys, worker, realtime, metrics, log });
// The key loads in the background (retrying while Network boots); /api/ready says when it has.
const keyLoaded = keys.start().catch(() => null);
@@@ -65,7 +68,7 @@ async function start({ config, clock = { now: () => Date.now() }, fetchImpl = gl
db.close();
}
-- return { config, db, store, keys, keyLoaded, auth, worker, realtime, app, server, close, prune };
++ return { config, db, store, keys, keyLoaded, auth, worker, realtime, metrics, app, server, close, prune };
}
if (require.main === module) {
ddiff --git a/server/metrics.js b/server/metrics.js
nnew file mode 100644
iindex 0000000..509f12e
---- /dev/null
++++ b/server/metrics.js
@@@ -0,0 +1,43 @@
++'use strict';
++/**
++ * Prometheus metrics (roadmap Track O): the openvibe-shared registry that app.js instruments with
++ * HTTP golden signals, plus the Events domain gauges read at scrape time and the worker's delivery
++ * hooks. Served on GET /metrics to direct loopback callers only.
++ *
++ * events_deliveries{status} deliveries by status (pending, failed, delivered, dead)
++ * events_dlq_depth dead deliveries waiting for replay
++ * events_latest_seq last sequence number assigned
++ * events_realtime_connections open SSE connections
++ * events_delivery_latency_seconds{priority} acceptance → successful delivery (retries included)
++ * events_delivery_attempts_total{outcome} delivered | retry | dead
++ */
++const { createRegistry } = require('openvibe-shared/metrics');
++
++// Delivery latency spans retries with backoff up to an hour, so the buckets reach well past 10 s.
++const LATENCY_BUCKETS = [0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60, 300, 1800, 3600, 21600];
++
++function createMetrics({ store }) {
++ const registry = createRegistry();
++ let realtime = null;
++
++ registry.gauge({
++ name: 'events_deliveries', help: 'Deliveries by status', labelNames: ['status'],
++ collect: () => Object.entries(store.deliveryCounts()).map(([status, value]) => ({ labels: { status }, value })),
++ });
++ registry.gauge({ name: 'events_dlq_depth', help: 'Dead deliveries (the DLQ) waiting for replay', collect: () => store.deliveryCounts().dead });
++ registry.gauge({ name: 'events_latest_seq', help: 'Last sequence number assigned to an event', collect: () => store.lastSeq() });
++ registry.gauge({ name: 'events_realtime_connections', help: 'Open realtime (SSE) connections', collect: () => (realtime ? realtime.count() : undefined) });
++ const latency = registry.histogram({ name: 'events_delivery_latency_seconds', help: 'Time from acceptance to successful delivery, retries included', labelNames: ['priority'], buckets: LATENCY_BUCKETS });
++ const attempts = registry.counter({ name: 'events_delivery_attempts_total', help: 'Delivery attempts by outcome', labelNames: ['outcome'] });
++
++ return {
++ registry,
++ observe: {
++ delivered: (seconds, row) => latency.observe({ priority: row && row.priority }, seconds),
++ attempt: (outcome) => attempts.inc({ outcome }),
++ },
++ bind(parts) { realtime = parts.realtime || realtime; },
++ };
++}
++
++module.exports = { createMetrics };
ddiff --git a/server/worker.js b/server/worker.js
iindex 29e7ce6..d1fae98 100644
---- a/server/worker.js
++++ b/server/worker.js
@@@ -17,7 +17,8 @@ const { sign } = require('../lib/client');
const { rowToEnvelope } = require('./store');
const { checkEndpoint } = require('./endpoints');
--function createWorker({ store, config, clock = { now: () => Date.now() }, fetchImpl = globalThis.fetch, log = console }) {
++// observe (optional): { delivered(seconds, row), attempt(outcome) } — metrics hooks, never required.
++function createWorker({ store, config, clock = { now: () => Date.now() }, fetchImpl = globalThis.fetch, log = console, observe = null }) {
const opts = config.worker;
const busy = new Set(); // subscription ids with a delivery in flight
const inflight = new Set(); // promises
@@@ -76,6 +77,13 @@ function createWorker({ store, config, clock = { now: () => Date.now() }, fetchI
if (outcome.dead) log.warn(`[worker] ${row.id} -> ${sub.id} dead after ${attempt} attempts: ${outcome.error}`);
}
store.recordAttempt(delivery.event_id, delivery.subscription_id, outcome);
++ if (observe) {
++ try {
++ observe.attempt(outcome.ok ? 'delivered' : outcome.dead ? 'dead' : 'retry');
++ // Acceptance to successful delivery, retries included.
++ if (outcome.ok) observe.delivered(Math.max(0, clock.now() - row.received_at) / 1000, row);
++ } catch { /* metrics never break delivery */ }
++ }
}
/** Start every due delivery there is room for. Returns how many were started. */
ddiff --git a/test/delivery.test.js b/test/delivery.test.js
iindex 75e4960..c90b73b 100644
---- a/test/delivery.test.js
++++ b/test/delivery.test.js
@@@ -204,7 +204,7 @@ t('the real loop delivers without drain()', async () => {
assert.strictEqual(stub.calls.length, 1);
const ready = await request(loop.base, 'GET', '/api/ready');
assert.strictEqual(ready.status, 200);
-- assert.strictEqual(ready.body.checks.worker, true);
++ assert.strictEqual(ready.body.checks.delivery_worker.status, 'ok');
await loop.stop();
await stub.close();
});
ddiff --git a/test/observability.test.js b/test/observability.test.js
nnew file mode 100644
iindex 0000000..dbdc14c
---- /dev/null
++++ b/test/observability.test.js
@@@ -0,0 +1,108 @@
++'use strict';
++// Track O: GET /metrics (loopback only, route templates, Events gauges and delivery histogram) and a
++// truthful /api/ready (required db/key/worker, optional DLQ → degraded but still ready).
++const assert = require('assert');
++const nodeHttp = require('http');
++const { boot, request, serviceToken, envelope, subscriber, suite } = require('./helpers');
++
++const t = suite('observability');
++const live = serviceToken('live', ['events.event.publish']);
++const media = serviceToken('media', ['events.subscription.manage']);
++const reader = serviceToken('games', ['events.event.read']);
++let h;
++
++function get(base, p, headers = {}) {
++ return new Promise((resolve, reject) => nodeHttp.get(base + p, { headers }, (res) => {
++ let b = ''; res.on('data', (d) => { b += d; }); res.on('end', () => resolve({ status: res.statusCode, body: b }));
++ }).on('error', reject));
++}
++
++t('boot with the worker running', async () => { h = await boot({ worker: 'on', env: { EVENTS_DLQ_DEGRADED_AT: '0', EVENTS_MAX_ATTEMPTS: '1' } }); });
++
++t('/api/ready: every check reports status, latency and checked_at; worker is required', async () => {
++ const r = await request(h.base, 'GET', '/api/ready');
++ assert.strictEqual(r.status, 200, r.text);
++ assert.strictEqual(r.body.status, 'ready');
++ assert.strictEqual(r.body.service, 'events');
++ assert.deepStrictEqual(Object.keys(r.body.checks), ['db', 'network_jwks', 'delivery_worker', 'dlq']);
++ for (const [name, c] of Object.entries(r.body.checks)) {
++ assert.strictEqual(c.status, 'ok', name);
++ assert.strictEqual(typeof c.latency_ms, 'number');
++ assert.ok(Date.parse(c.checked_at));
++ }
++ assert.strictEqual(r.body.checks.dlq.required, false);
++ assert.deepStrictEqual(r.body.checks.dlq.detail, { depth: 0, threshold: 0 });
++ assert.ok('latest_seq' in r.body && 'deliveries' in r.body && 'realtime_connections' in r.body);
++ assert.strictEqual(r.body.worker.running, true);
++});
++
++t('deliveries feed the latency histogram and attempt counter; a dead one degrades /api/ready', async () => {
++ const ok = await subscriber(() => 204);
++ const bad = await subscriber(() => 500);
++ await request(h.base, 'POST', '/api/v1/subscriptions', { token: media, body: { topic_pattern: 'live.stream.*', endpoint: ok.url } });
++ await request(h.base, 'POST', '/api/v1/subscriptions', { token: serviceToken('games', ['events.subscription.manage']), body: { topic_pattern: 'live.stream.*', endpoint: bad.url } });
++ const r = await request(h.base, 'POST', '/api/v1/events', { token: live, body: envelope() });
++ assert.strictEqual(r.status, 201, r.text);
++ await h.worker.drain();
++ assert.strictEqual(ok.calls.length, 1);
++ const ready = await request(h.base, 'GET', '/api/ready');
++ assert.strictEqual(ready.status, 200, 'a DLQ over the threshold degrades, it does not make the service unready');
++ assert.strictEqual(ready.body.ready, true);
++ assert.strictEqual(ready.body.status, 'degraded');
++ assert.deepStrictEqual(ready.body.degraded, ['dlq']);
++ assert.strictEqual(ready.body.checks.dlq.detail.depth, 1);
++ await ok.close(); await bad.close();
++});
++
++t('/metrics: loopback only, templates not ids, Events gauges', async () => {
++ const id = h.db.prepare('SELECT id FROM events ORDER BY seq LIMIT 1').get().id;
++ await request(h.base, 'GET', `/api/v1/events/${id}`, { token: reader });
++ await request(h.base, 'GET', '/api/v1/events?topic=live.*&after_seq=0', { token: reader });
++ await request(h.base, 'GET', '/no/such/12345');
++ let m = await get(h.base, '/metrics', { 'X-Forwarded-For': '203.0.113.7' });
++ assert.strictEqual(m.status, 404, 'through a proxy: not found');
++ m = await get(h.base, '/metrics');
++ assert.strictEqual(m.status, 200);
++ const text = m.body;
++ assert.ok(text.includes('route="/api/v1/events/:id"'), 'route template for the single-event read');
++ assert.ok(!text.includes(id), 'no event id in any label');
++ assert.ok(!/route="[^"]*12345/.test(text), 'unmatched paths are not labels');
++ assert.ok(/http_requests_total\{method="GET",route="unmatched",status_class="4xx"\} 1/.test(text));
++ assert.ok(/events_deliveries\{status="delivered"\} 1\n/.test(text));
++ assert.ok(/events_deliveries\{status="dead"\} 1\n/.test(text));
++ assert.ok(/\nevents_dlq_depth 1\n/.test(text));
++ assert.ok(/\nevents_latest_seq 1\n/.test(text));
++ assert.ok(/\nevents_realtime_connections 0\n/.test(text));
++ assert.ok(/events_delivery_latency_seconds_count\{priority="important"\} 1\n/.test(text), text.split('\n').filter(l => l.includes('latency')).join('\n'));
++ assert.ok(/events_delivery_attempts_total\{outcome="delivered"\} 1\n/.test(text));
++ assert.ok(/events_delivery_attempts_total\{outcome="dead"\} 1\n/.test(text));
++ assert.ok(/release_info\{service="events",release="[0-9a-f]{7,12}"\} 1/.test(text));
++ assert.ok(!text.includes('realtime/stream'), 'SSE sessions are not HTTP request samples');
++});
++
++t('release.json', async () => {
++ const r = await request(h.base, 'GET', '/release.json');
++ assert.strictEqual(r.status, 200);
++ assert.strictEqual(r.body.service, 'events');
++});
++
++t('a stopped worker makes the service unready', async () => {
++ await h.worker.stop();
++ const r = await request(h.base, 'GET', '/api/ready');
++ assert.strictEqual(r.status, 503);
++ assert.strictEqual(r.body.ready, false);
++ assert.deepStrictEqual(r.body.failed, ['delivery_worker']);
++ assert.strictEqual(r.body.checks.delivery_worker.error, 'delivery worker is not running');
++});
++
++t('a closed database makes the service unready', async () => {
++ h.db.close();
++ const r = await request(h.base, 'GET', '/api/ready');
++ assert.strictEqual(r.status, 503);
++ assert.ok(r.body.failed.includes('db'));
++ assert.strictEqual(r.body.latest_seq, null);
++ // close() closes the db again; tolerate it.
++ try { await h.close(); } catch { /* already closed */ }
++});
++
++t.run();
ddiff --git a/test/pull-health.test.js b/test/pull-health.test.js
iindex cf23a37..c08d071 100644
---- a/test/pull-health.test.js
++++ b/test/pull-health.test.js
@@@ -93,7 +93,10 @@ t('health and ready', async () => {
assert.deepStrictEqual([r.status, r.body.status, r.body.service], [200, 'ok', 'openvibe-events']);
r = await request(h.base, 'GET', '/api/ready');
assert.strictEqual(r.status, 200);
-- assert.deepStrictEqual(r.body.checks, { db: true, worker: true, key: true });
++ assert.strictEqual(r.body.ready, true);
++ assert.deepStrictEqual(Object.keys(r.body.checks), ['db', 'network_jwks', 'dlq'], 'worker off in this boot: no delivery_worker check');
++ assert.strictEqual(r.body.checks.db.status, 'ok');
++ assert.strictEqual(r.body.checks.network_jwks.status, 'ok');
r = await request(h.base, 'GET', '/nope');
assert.strictEqual(r.status, 404);
assert.strictEqual(r.body.code, 'events.not_found');
@@@ -118,7 +121,8 @@ t('key loader: JWKS {keys:[jwk]} and legacy {public_key}; ready is 503 until it
await x.keyLoaded;
let r = await request(x.base, 'GET', '/api/ready');
assert.strictEqual(r.status, 503);
-- assert.strictEqual(r.body.checks.key, false);
++ assert.strictEqual(r.body.checks.network_jwks.status, 'fail');
++ assert.deepStrictEqual(r.body.failed, ['network_jwks']);
r = await request(x.base, 'POST', '/api/v1/events', { token: live, body: envelope() });
assert.strictEqual(r.status, 503, 'no key yet: service unavailable, not unauthorized');
shape = s;