openvibers/openvibe.sources Commits

Track O: /metrics for direct loopback callers and a truthful /api/ready (openvibe-shared v1.3.0)

OP OpenVibers committed 23 Sept 2026 · 072f52908a · 9 files, +387 −31

Adds openvibe-shared v1.3.0. /metrics (openvibe-shared/metrics): golden signals labelled by route
template, process metrics, release_info, sources_sources{status}, sources_items{state}, the fetch
queue (sources_fetch_due, sources_fetch_oldest_wait_seconds, sources_fetch_in_flight),
sources_last_fetch_timestamp_seconds, sources_last_success_timestamp_seconds and the outbox
backlog; a request carrying X-Forwarded-For, X-Real-IP or CF-Connecting-IP gets 404. /api/ready
(openvibe-shared/ready): the database is required (503 when it fails); the Network key and the
fetcher are optional: a worker that is off or stopped, or a queue with a source waiting more than
15 minutes (counted from when it became due, was changed or the process started), degrades it.
Source counts, runs in flight and the outbox stay in the body. GET /release.json. nginx answers
/metrics with 404. test/observability.test.js; boot() in the test helpers takes a clock.

Changes

ddiff --git a/README.md b/README.md
iindex 8c26826..38272e8 100644
---- a/README.md
++++ b/README.md
@@@ -33,9 +33,14 @@ Node 22 in production (`fnm exec --using=22.22.1 npm test`). Production: `/opt/o
  env `/etc/openvibe/sources.env`, unit [deploy/systemd/openvibe-sources.service](deploy/systemd/openvibe-sources.service),
  store `/var/lib/openvibe-sources/sources.db`, nginx [deploy/nginx/sources.openvibe.network.conf](deploy/nginx/sources.openvibe.network.conf).
  
--`GET /api/health` is liveness. `GET /api/ready` is 200 when the database answers, the Network key
--has loaded and the scheduler runs; it reports source counts by health status, runs in flight and
--the outbox backlog.
++`GET /api/health` is liveness. `GET /api/ready` (openvibe-shared/ready) is 503 only when the
++database fails; a Network key that has not loaded and a fetcher that is off, stopped or behind (a
++source due for more than 15 minutes) degrade it. It reports source counts by health status, runs in
++flight and the outbox backlog. `GET /metrics` (openvibe-shared/metrics) answers direct loopback
++callers only: golden signals by route template, `sources_sources{status}`, `sources_items{state}`,
++the fetch queue (`sources_fetch_due`, `sources_fetch_oldest_wait_seconds`,
++`sources_fetch_in_flight`), `sources_last_fetch_timestamp_seconds` and
++`sources_last_success_timestamp_seconds`.
  
  ## The registry
  
ddiff --git a/deploy/nginx/sources.openvibe.network.conf b/deploy/nginx/sources.openvibe.network.conf
iindex 22a85e5..f62c241 100644
---- a/deploy/nginx/sources.openvibe.network.conf
++++ b/deploy/nginx/sources.openvibe.network.conf
@@@ -6,6 +6,7 @@
  ##
  ## The Node service (port 4720, systemd unit openvibe-sources) serves:
  ##   /api/health, /api/ready
++##   /metrics    Prometheus text, answered only to direct loopback callers (never proxied)
  ##   /api/v1/*   service-to-service API (every route needs a service token). Services on this
  ##               host call 127.0.0.1:4720 directly, so the public vhost refuses it; open the
  ##               location below only when a service on another host needs it.
@@@ -48,6 +49,9 @@ server {
      location = /api/health { proxy_pass http://127.0.0.1:4720; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $remote_addr; proxy_set_header CF-Connecting-IP $remote_addr; }
      location = /api/ready  { proxy_pass http://127.0.0.1:4720; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $remote_addr; proxy_set_header CF-Connecting-IP $remote_addr; }
  
++    # Metrics are for host-local scrapers only (http://127.0.0.1:4720/metrics).
++    location = /metrics { return 404; }
++
      # ── Service API: host-local only (see header) ───────────
      location /api/ {
          return 403;
ddiff --git a/package-lock.json b/package-lock.json
iindex b7d3b97..ff41977 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -13,7 +13,8 @@
          "dotenv": "^16.4.7",
          "express": "^4.21.2",
          "fast-xml-parser": "^4.5.3",
--        "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.13.0"
++        "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.13.0",
++        "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.3.0"
        },
        "devDependencies": {
          "ajv": "^8.20.0",
@@@ -174,6 +175,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",
@@@ -341,6 +348,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",
@@@ -720,6 +736,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",
@@@ -898,6 +1005,15 @@
          "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",
++      "integrity": "sha512-w8yl/5+9eend8I2O/DzAJnEAL2N3GTUY7c7pzK+MLJuLAGD8FmJJQ9gwbgyrq70Snjt3ebrg3TlpORhRAp2kgA==",
++      "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 b2560df..3f1625b 100644
---- a/package.json
++++ b/package.json
@@@ -27,7 +27,8 @@
      "dotenv": "^16.4.7",
      "express": "^4.21.2",
      "fast-xml-parser": "^4.5.3",
--    "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.13.0"
++    "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.13.0",
++    "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.3.0"
    },
    "engines": {
      "node": ">=22.0.0"
ddiff --git a/server/app.js b/server/app.js
iindex f288e44..f07025a 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -1,7 +1,11 @@
  'use strict';
--/** Express app: request context, the v1 API, health/readiness. */
++/** Express app: request context, the v1 API, health/readiness, metrics. */
++const path = require('path');
  const express = require('express');
  const { http } = require('openvibe-contracts');
++const { instrument } = require('openvibe-shared/metrics');
++const { createRelease } = require('openvibe-shared/release');
++const { createSourcesReadiness, registerSourcesGauges } = require('./observability');
  const { sourcesRouter } = require('./api/sources');
  const { itemsRouter } = require('./api/items');
  const pkg = require('../package.json');
@@@ -10,6 +14,11 @@ function createApp({ config, db, registry, items, ingest, scheduler, auth, keys,
      const app = express();
      app.disable('x-powered-by');
      app.set('trust proxy', 'loopback');
++    const release = createRelease({ service: 'sources', root: path.join(__dirname, '..') });
++    // HTTP golden signals by route template, process metrics, release_info and the Sources gauges;
++    // GET /metrics answers direct loopback callers only (Track O).
++    const metrics = instrument(app, { service: 'sources', release: release.release });
++    registerSourcesGauges(metrics.registry, { db, sources: registry, ingest, outbox, now });
      app.use(http.middleware());
      app.use((req, res, next) => {
          res.setHeader('X-Content-Type-Options', 'nosniff');
@@@ -22,27 +31,11 @@ function createApp({ config, db, registry, items, ingest, scheduler, auth, keys,
          res.json({ status: 'ok', service: 'openvibe-sources', version: pkg.version });
      });
  
--    app.get('/api/ready', (_req, res) => {
--        let dbOk = false;
--        try { dbOk = db.prepare('SELECT 1 AS ok').get().ok === 1; } catch { dbOk = false; }
--        const checks = { db: dbOk, key: keys.loaded(), worker: !config.worker.enabled || scheduler.running() };
--        const ready = Object.values(checks).every(Boolean);
--        let sources = null;
--        if (dbOk) {
--            sources = {};
--            for (const r of registry.all()) {
--                const s = registry.health(r).status;
--                sources[s] = (sources[s] || 0) + 1;
--            }
--        }
--        res.status(ready ? 200 : 503).json({
--            status: ready ? 'ready' : 'not_ready',
--            checks,
--            sources,
--            runs_in_flight: ingest.inflight().length,
--            outbox: dbOk ? { pending: outbox.pending(), rejected: outbox.rejected(), relay: config.events.url ? (relay.running() ? 'running' : 'stopped') : 'off (EVENTS_URL unset)' } : null,
--        });
--    });
++    // Readiness (openvibe-shared/ready): 503 only when the database fails; the Network key and the
++    // fetcher (worker running, queue keeping up) are optional and degrade it (see observability.js).
++    const readiness = createSourcesReadiness({ db, keys, config, registry, ingest, scheduler, outbox, relay, now, release: release.release });
++    app.get('/api/ready', readiness.handler);
++    app.get('/release.json', release.handler);
  
      app.use(sourcesRouter({ db, registry, ingest, auth }));
      app.use(itemsRouter({ db, registry, items, auth, relay, now }));
@@@ -56,7 +49,7 @@ function createApp({ config, db, registry, items, ingest, scheduler, auth, keys,
              'GET  /api/v1/sources, /api/v1/sources/:key, /api/v1/sources/:key/runs, /api/v1/runs, /api/v1/health   (sources.source.read)',
              'GET  /api/v1/items?source=&category=&after=, /api/v1/items/:id                                          (sources.item.read)',
              'POST/PATCH/DELETE /api/v1/sources[/:key], POST /api/v1/sources/:key/fetch, /items, DELETE /api/v1/items/:id (sources.source.manage)',
--            'GET  /api/health, /api/ready',
++            'GET  /api/health, /api/ready, /release.json',
              '',
              'Source: https://github.com/OpenVibers/OpenVibe.Sources',
              '',
@@@ -74,6 +67,7 @@ function createApp({ config, db, registry, items, ingest, scheduler, auth, keys,
          return http.sendProblem(res, 500, 'sources.internal', { detail: 'internal error', ctx: req.ov });
      });
  
++    app.locals.metrics = metrics;
      return app;
  }
  
ddiff --git a/server/observability.js b/server/observability.js
nnew file mode 100644
iindex 0000000..c259139
---- /dev/null
++++ b/server/observability.js
@@@ -0,0 +1,111 @@
++'use strict';
++/**
++ * Track O: truthful readiness for GET /api/ready and the Sources gauges on GET /metrics
++ * (openvibe-shared/ready and openvibe-shared/metrics).
++ *
++ *   db            required  a real query on the source registry: without it nothing is served or fetched
++ *   network_jwks  optional  the Network signing key has loaded. Without it no service token can be
++ *                           verified (the API answers 503), but the fetcher keeps ingesting, so it
++ *                           degrades rather than fails
++ *   fetcher       optional  the ingestion worker is on and running, and its queue keeps up: no enabled
++ *                           source has waited longer than OVERDUE_MS. Without it the API still
++ *                           serves what was fetched, but nothing new arrives
++ *
++ * Gauges: sources by health status, items (current and removed), the queue (due sources, runs in
++ * flight), the time of the last finished fetch and of the last successful one, and the outbox backlog.
++ */
++const { createReadiness } = require('openvibe-shared/ready');
++
++const OVERDUE_MS = 15 * 60 * 1000;
++const STATUSES = ['healthy', 'stale', 'failing', 'never_fetched', 'disabled', 'manual'];
++
++function readers(db, { startedAt }) {
++    // A source waits from when it became due, when it was last changed (a new or re-enabled source
++    // is due at once, from next_due_at 0) or when this process started, whichever is latest.
++    const due = db.prepare(`SELECT COUNT(*) AS n, MIN(MAX(next_due_at, not_before, updated_at)) AS oldest FROM sources
++        WHERE enabled = 1 AND type != 'manual' AND next_due_at <= @now AND not_before <= @now`);
++    const lastRun = db.prepare('SELECT finished_at FROM fetch_runs ORDER BY rid DESC LIMIT 1');
++    const lastSuccess = db.prepare('SELECT MAX(last_success_at) AS t FROM sources');
++    const items = db.prepare('SELECT COUNT(*) AS n, COUNT(removed_at) AS removed FROM items');
++    return {
++        /** Enabled fetched sources whose turn has come, and how long the oldest has waited. */
++        queue(t) {
++            const r = due.get({ now: t });
++            return { due: r.n, oldest_wait_ms: r.n ? Math.max(0, t - Math.max(r.oldest, startedAt)) : 0 };
++        },
++        lastFetchAt: () => { const r = lastRun.get(); return r ? r.finished_at : null; },
++        lastSuccessAt: () => lastSuccess.get().t,
++        items: () => { const r = items.get(); return { current: r.n - r.removed, removed: r.removed }; },
++    };
++}
++
++function createSourcesReadiness({ db, keys, config, registry, ingest, scheduler, outbox, relay, now, release = null }) {
++    const read = readers(db, { startedAt: now() });
++    const iso = (v) => (v == null ? null : new Date(v).toISOString());
++    return createReadiness({
++        service: 'sources',
++        release,
++        checks: [
++            { name: 'db', required: true, check: () => { db.prepare('SELECT COUNT(*) AS n FROM sources').get(); return true; } },
++            { name: 'network_jwks', required: false, check: () => keys.loaded() || 'Network signing key not loaded yet: no service token can be verified' },
++            {
++                name: 'fetcher', required: false,
++                check: () => {
++                    const q = read.queue(now());
++                    const detail = {
++                        enabled: config.worker.enabled, running: scheduler.running(), in_flight: ingest.inflight().length,
++                        max_concurrent: config.worker.maxConcurrent, due: q.due, oldest_wait_ms: q.oldest_wait_ms,
++                        last_fetch_at: iso(read.lastFetchAt()),
++                    };
++                    if (!config.worker.enabled) return { ok: false, error: 'the ingestion worker is off (SOURCES_WORKER=off): nothing is fetched', detail };
++                    if (!scheduler.running()) return { ok: false, error: 'the ingestion worker is not running', detail };
++                    if (q.oldest_wait_ms > OVERDUE_MS) return { ok: false, error: `the fetch queue is behind: ${q.due} source(s) due, the oldest for ${Math.round(q.oldest_wait_ms / 60000)} min`, detail };
++                    return { ok: true, detail };
++                },
++            },
++        ],
++        details: (body) => {
++            const dbOk = body.checks.db.status === 'ok';
++            let sources = null;
++            if (dbOk) {
++                sources = {};
++                for (const r of registry.all()) {
++                    const s = registry.health(r).status;
++                    sources[s] = (sources[s] || 0) + 1;
++                }
++            }
++            return {
++                sources,
++                runs_in_flight: ingest.inflight().length,
++                outbox: dbOk ? { pending: outbox.pending(), rejected: outbox.rejected(), relay: config.events.url ? (relay.running() ? 'running' : 'stopped') : 'off (EVENTS_URL unset)' } : null,
++            };
++        },
++    });
++}
++
++/** Sources gauges on the openvibe-shared/metrics registry. */
++function registerSourcesGauges(registry, { db, sources, ingest, outbox, now }) {
++    const read = readers(db, { startedAt: now() });
++    const seconds = (ms) => (ms == null ? null : ms / 1000);
++    registry.gauge({
++        name: 'sources_sources', help: 'Registered sources by health status', labelNames: ['status'],
++        collect: () => {
++            const n = Object.fromEntries(STATUSES.map((s) => [s, 0]));
++            for (const r of sources.all()) { const s = sources.health(r).status; n[s] = (n[s] || 0) + 1; }
++            return Object.entries(n).map(([status, value]) => ({ labels: { status }, value }));
++        },
++    });
++    registry.gauge({
++        name: 'sources_items', help: 'Items held, current and removed', labelNames: ['state'],
++        collect: () => Object.entries(read.items()).map(([state, value]) => ({ labels: { state }, value })),
++    });
++    registry.gauge({ name: 'sources_fetch_due', help: 'Enabled sources whose next fetch is due now (the fetch queue)', collect: () => read.queue(now()).due });
++    registry.gauge({ name: 'sources_fetch_oldest_wait_seconds', help: 'How long the longest-waiting due source has waited', collect: () => read.queue(now()).oldest_wait_ms / 1000 });
++    registry.gauge({ name: 'sources_fetch_in_flight', help: 'Fetch runs in progress', collect: () => ingest.inflight().length });
++    // Left out until there is a fetch: a timestamp of 0 would read as "1970", not "never".
++    registry.gauge({ name: 'sources_last_fetch_timestamp_seconds', help: 'When the last fetch run finished (any outcome), Unix seconds', collect: () => seconds(read.lastFetchAt()) });
++    registry.gauge({ name: 'sources_last_success_timestamp_seconds', help: 'When a source was last fetched successfully, Unix seconds', collect: () => seconds(read.lastSuccessAt()) });
++    registry.gauge({ name: 'sources_outbox_pending', help: 'Events waiting in the outbox', collect: () => outbox.pending() });
++}
++
++module.exports = { createSourcesReadiness, registerSourcesGauges, OVERDUE_MS };
ddiff --git a/test/api.test.js b/test/api.test.js
iindex f2f8f24..4dffc65 100644
---- a/test/api.test.js
++++ b/test/api.test.js
@@@ -147,7 +147,7 @@ t('health lists what needs attention', async () => {
      assert.ok(h.body.attention.some(a => a.key === 'never' && a.status === 'never_fetched'));
      const ready = await api('GET', '/api/ready');
      assert.strictEqual(ready.status, 200);
--    assert.strictEqual(ready.body.checks.db, true);
++    assert.strictEqual(ready.body.checks.db.status, 'ok');
  });
  
  t('done', async () => { await web.close(); await svc.stop(); });
ddiff --git a/test/helpers.js b/test/helpers.js
iindex 41c4db7..1213c3c 100644
---- a/test/helpers.js
++++ b/test/helpers.js
@@@ -40,7 +40,7 @@ function tmpDir() {
  }
  
  /** Boot Sources; the scheduler is off unless worker: 'on'. Loopback is allowlisted for the stubs. */
--async function boot({ env = {}, worker = 'off', lookupImpl, tokenClient } = {}) {
++async function boot({ env = {}, worker = 'off', lookupImpl, tokenClient, now } = {}) {
      const dir = tmpDir();
      const config = load({
          NODE_ENV: 'test',
@@@ -54,7 +54,7 @@ async function boot({ env = {}, worker = 'off', lookupImpl, tokenClient } = {})
          SOURCES_TICK_MS: '50',
          ...env,
      });
--    const h = await start({ config, log: silent, lookupImpl, tokenClient });
++    const h = await start({ config, log: silent, lookupImpl, tokenClient, ...(now ? { now } : {}) });
      const base = `http://127.0.0.1:${h.server.address().port}`;
      return { ...h, base, dir, async stop() { await h.close(); } };
  }
ddiff --git a/test/observability.test.js b/test/observability.test.js
nnew file mode 100644
iindex 0000000..fef1312
---- /dev/null
++++ b/test/observability.test.js
@@@ -0,0 +1,125 @@
++'use strict';
++// Track O: GET /metrics answers direct loopback callers only, labels requests by route template and
++// carries the Sources gauges (sources by status, items, the fetch queue, last fetch); /api/ready is
++// 503 when the database fails, and a fetcher that is off, stopped or behind degrades it.
++const assert = require('assert');
++const nodeHttp = require('http');
++const { boot, request, serviceToken, site, sourceDef, rss, suite } = require('./helpers');
++const { OVERDUE_MS } = require('../server/observability');
++
++const t = suite('observability');
++const ADMIN = serviceToken('network', ['sources.source.manage']);
++const READER = serviceToken('news', ['sources.source.read', 'sources.item.read']);
++let svc;
++let web;
++
++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 (worker off) and fetch one feed of three items by hand', async () => {
++    svc = await boot();
++    await svc.keyLoaded;
++    web = await site({
++        '/robots.txt': () => ({ status: 404 }),
++        '/feed.xml': () => ({ body: rss([1, 2, 3].map(i => ({ guid: `g${i}`, title: `Item ${i}`, link: `https://example.org/${i}` }))) }),
++    });
++    const c = await request(svc.base, 'POST', '/api/v1/sources', { token: ADMIN, body: sourceDef({ key: 'feed-obs', endpoints: [`${web.origin}/feed.xml`] }) });
++    assert.strictEqual(c.status, 201, c.text);
++    const f = await request(svc.base, 'POST', '/api/v1/sources/feed-obs/fetch', { token: ADMIN });
++    assert.deepStrictEqual(f.body.runs.map(r => [r.state, r.items.created]), [['ok', 3]]);
++});
++
++t('/api/ready: db is the only required check; a worker that is off degrades it (still 200)', async () => {
++    const r = await request(svc.base, 'GET', '/api/ready');
++    assert.strictEqual(r.status, 200, r.text);
++    assert.strictEqual(r.body.ready, true);
++    assert.strictEqual(r.body.status, 'degraded');
++    assert.strictEqual(r.body.service, 'sources');
++    assert.deepStrictEqual(Object.keys(r.body.checks), ['db', 'network_jwks', 'fetcher']);
++    assert.deepStrictEqual(r.body.degraded, ['fetcher']);
++    for (const [name, c] of Object.entries(r.body.checks)) {
++        assert.strictEqual(c.required, name === 'db', name);
++        assert.strictEqual(typeof c.latency_ms, 'number');
++        assert.ok(Date.parse(c.checked_at));
++    }
++    assert.match(r.body.checks.fetcher.error, /SOURCES_WORKER=off/);
++    assert.ok(Date.parse(r.body.checks.fetcher.detail.last_fetch_at));
++    assert.deepStrictEqual(r.body.sources, { healthy: 1 });
++});
++
++t('/metrics: 404 through a proxy; route templates and Sources gauges direct', async () => {
++    await request(svc.base, 'GET', '/api/v1/sources/feed-obs', { token: READER });
++    for (const hdr of [{ 'X-Forwarded-For': '203.0.113.7' }, { 'X-Real-IP': '203.0.113.7' }, { 'CF-Connecting-IP': '203.0.113.7' }]) {
++        const m = await get(svc.base, '/metrics', hdr);
++        assert.strictEqual(m.status, 404, JSON.stringify(hdr));
++        assert.ok(!m.body.includes('sources_items'));
++    }
++    const m = await get(svc.base, '/metrics');
++    assert.strictEqual(m.status, 200);
++    const text = m.body;
++    assert.ok(/http_requests_total\{method="POST",route="\/api\/v1\/sources\/:key\/fetch",status_class="2xx"\} 1\n/.test(text), 'route template');
++    assert.ok(/http_requests_total\{method="GET",route="\/api\/v1\/sources\/:key",status_class="2xx"\} 1\n/.test(text));
++    assert.ok(!/route="[^"]*feed-obs/.test(text), 'no source key in any label');
++    assert.ok(/\nprocess_resident_memory_bytes \d+\n/.test(text));
++    assert.ok(/release_info\{service="sources",release="[^"]+"\} 1\n/.test(text));
++    assert.ok(/sources_sources\{status="healthy"\} 1\n/.test(text));
++    assert.ok(/sources_sources\{status="failing"\} 0\n/.test(text));
++    assert.ok(/sources_items\{state="current"\} 3\n/.test(text));
++    assert.ok(/sources_items\{state="removed"\} 0\n/.test(text));
++    assert.ok(/\nsources_fetch_due 0\n/.test(text));
++    assert.ok(/\nsources_fetch_in_flight 0\n/.test(text));
++    const last = text.match(/\nsources_last_fetch_timestamp_seconds ([\d.]+)\n/);
++    assert.ok(last && Math.abs(Number(last[1]) - Date.now() / 1000) < 60, 'last fetch time in Unix seconds');
++    assert.ok(/\nsources_last_success_timestamp_seconds [\d.]+\n/.test(text));
++    assert.ok(/\nsources_outbox_pending \d+\n/.test(text));
++});
++
++t('a broken database makes the service unready (503); /metrics still answers', async () => {
++    svc.db.close();
++    const r = await request(svc.base, 'GET', '/api/ready');
++    assert.strictEqual(r.status, 503, r.text);
++    assert.strictEqual(r.body.ready, false);
++    assert.ok(r.body.failed.includes('db'), r.text);
++    assert.strictEqual(r.body.sources, null);
++    const m = await get(svc.base, '/metrics');
++    assert.strictEqual(m.status, 200);
++    assert.ok(!/sources_items\{/.test(m.body), 'a gauge that cannot be read is left out, not invented');
++    try { await svc.stop(); } catch { /* the database is already closed */ }
++});
++
++t('worker on: ready; a fetch queue that falls behind, or a stopped worker, degrades it', async () => {
++    const clock = { offset: 0 };
++    // No run slot, so nothing due is ever started; the clock moves forward by hand.
++    svc = await boot({ worker: 'on', env: { SOURCES_MAX_CONCURRENT: '0' }, now: () => Date.now() + clock.offset });
++    let r = await request(svc.base, 'GET', '/api/ready');
++    assert.strictEqual(r.body.status, 'ready', r.text);
++    assert.strictEqual(r.body.checks.fetcher.detail.running, true);
++    await request(svc.base, 'POST', '/api/v1/sources', { token: ADMIN, body: sourceDef({ key: 'feed-late', endpoints: [`${web.origin}/feed.xml`] }) });
++    r = await request(svc.base, 'GET', '/api/ready');
++    assert.strictEqual(r.body.status, 'ready', 'a source that just became due is not late');
++    assert.strictEqual(r.body.checks.fetcher.detail.due, 1);
++
++    clock.offset = OVERDUE_MS + 60_000;
++    r = await request(svc.base, 'GET', '/api/ready');
++    assert.strictEqual(r.status, 200, r.text);
++    assert.deepStrictEqual(r.body.degraded, ['fetcher']);
++    assert.match(r.body.checks.fetcher.error, /the fetch queue is behind: 1 source\(s\) due, the oldest for 16 min/);
++    const text = (await get(svc.base, '/metrics')).body;
++    assert.ok(/\nsources_fetch_due 1\n/.test(text));
++    const wait = Number(text.match(/\nsources_fetch_oldest_wait_seconds ([\d.]+)\n/)[1]);
++    assert.ok(wait >= OVERDUE_MS / 1000, String(wait));
++
++    clock.offset = 0;
++    await svc.scheduler.stop();
++    r = await request(svc.base, 'GET', '/api/ready');
++    assert.strictEqual(r.status, 200);
++    assert.deepStrictEqual(r.body.degraded, ['fetcher']);
++    assert.strictEqual(r.body.checks.fetcher.error, 'the ingestion worker is not running');
++    await svc.stop();
++    await web.close();
++});
++
++t.run();