openvibers/openvibe.network Commits

Add the service-instance report writer and its route (#14)

OP OpenVibers committed 2 Oct 2026 · b6b123c032 · 4 files, +242 −1

local checks ✓ 3/3 · CI ✓ 3/3 · review SHIP by DeepSeek harness

https://github.com/OpenVibers/OpenVibe.Network/pull/14

Changes

ddiff --git a/server/index.js b/server/index.js
iindex f680462a..ceea0d3c 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -476,6 +476,12 @@ const notifyAccountData = async (userId, n) => await notificationService.create(
app.use('/api/v1/resources', offerRouters.pub);
app.use('/internal/resources', offerRouters.internal);
}
++// Service instances (plan T2, docs/t2-cells-and-node-principal.md section 4.1): Host reports what runs on its machines
++// with network.node.report. Mounted before /internal/registry so the cells read does not swallow the report path.
++{
++ const instanceRouters = require('./registry/instances').routers({ guard: require('./identity/principals').guard('network.node.report') });
++ app.use('/internal/registry/instances', instanceRouters.internal);
++}
// Cells and node principals (plan T2, docs/t2-resource-registry.md section 9): the public cell list; a cell's nodes,
// instances and offers for services holding network.registry.read.
{
ddiff --git a/server/registry/cells.js b/server/registry/cells.js
iindex c65d9d5f..32c95fe7 100644
---- a/server/registry/cells.js
++++ b/server/registry/cells.js
@@@ -98,7 +98,7 @@ async function topology(db, id) {
trust: p.trust, status: p.status, health: health.get(p.node_id) || 'unknown',
}));
const instances = (await db.prepare('SELECT * FROM platform_service_instances WHERE cell = ? ORDER BY service, id').all(cell.id)).map((r) => ({
-- id: r.id, service: r.service, version: r.version, node: r.node_id, endpoints: JSON.parse(r.endpoints), state: r.state,
++ id: r.id, service: r.service, version: r.version, cell: r.cell, region: cell.region, node: r.node_id, endpoints: JSON.parse(r.endpoints), state: r.state,
route_weight: Number(r.route_weight), started_at: r.started_at, reported_at: r.reported_at,
}));
const offers = (await db.prepare('SELECT doc FROM platform_resource_offers WHERE cell = ? ORDER BY id').all(cell.id)).map((r) => {
ddiff --git a/server/registry/instances.js b/server/registry/instances.js
nnew file mode 100644
iindex 00000000..d4838761
---- /dev/null
++++ b/server/registry/instances.js
@@@ -0,0 +1,110 @@
++'use strict';
++/**
++ * Service instances (plan T2, docs/t2-cells-and-node-principal.md section 4.1; Contracts
++ * platform.service-instance@1; migrations/0008_cells_and_node_principals.sql): what runs on the platform's machines —
++ * one row per running instance with its cell, node, endpoints and state. Host reports the complete set its inventory
++ * knows (POST /internal/registry/instances/report, network.node.report); an instance absent from a later report of the
++ * same source is set to `stopped`, never deleted. Placement is checked against what Network holds before anything is
++ * written: the cell must exist and not be retired, the node must be a live principal whose home cell is the instance's
++ * cell, and the instance's region must be the cell's region. Endpoints are stored as JSON text; no public route
++ * (instances carry endpoints).
++ */
++const express = require('express');
++
++const CONTRACT = 'platform.service-instance@1';
++const SOURCE = /^[a-z][a-z0-9-]{1,39}$/; // the same rule as network.node-report-request@1's source
++const MAX_INSTANCES = 500;
++
++function ensureSchema(db) { /* the schema is migrations/0008_cells_and_node_principals.sql; nothing is created at runtime */ }
++
++/**
++ * Check a whole report before anything is written: the Network-local envelope, then every instance against the
++ * contract. → null, or the 400 body for the first problem.
++ */
++function check(body) {
++ const { validate } = require('openvibe-contracts');
++ const keys = body && typeof body === 'object' && !Array.isArray(body) ? Object.keys(body) : null;
++ if (!keys || keys.some((k) => k !== 'source' && k !== 'instances') || typeof body.source !== 'string' || !SOURCE.test(body.source)
++ || !Array.isArray(body.instances) || body.instances.length > MAX_INSTANCES) {
++ return { error: `The body must be {source, instances}: source matching ${SOURCE}, at most ${MAX_INSTANCES} instances` };
++ }
++ for (let i = 0; i < body.instances.length; i++) {
++ const v = validate(CONTRACT, body.instances[i]);
++ if (!v.valid) return { error: `instance ${i} does not match ${CONTRACT}`, details: (v.errors || []).slice(0, 5) };
++ }
++ return null;
++}
++
++/**
++ * Check where a batch of instances says it runs, against the cells and node principals Network holds. Throws
++ * RegistryError for the first problem, so a refused report writes nothing.
++ */
++async function checkPlacement(db, instances) {
++ const RegistryError = require('./cells').RegistryError;
++ const cells = new Map((await db.prepare('SELECT id, region, status FROM platform_cells').all()).map((r) => [r.id, r]));
++ const principal = db.prepare('SELECT home_cell, status FROM platform_node_principals WHERE node_id = ?');
++ for (let i = 0; i < instances.length; i++) {
++ const inst = instances[i];
++ const cell = cells.get(inst.cell);
++ if (!cell) throw new RegistryError(400, 'registry.unknown_cell', `instance ${i}: no cell ${inst.cell}`);
++ if (cell.status === 'retired') throw new RegistryError(409, 'registry.cell_retired', `instance ${i}: cell ${inst.cell} is retired`);
++ const p = await principal.get(inst.node);
++ if (!p) throw new RegistryError(400, 'registry.unknown_node', `instance ${i}: no node principal ${inst.node}`);
++ if (p.status === 'revoked') throw new RegistryError(409, 'registry.node_revoked', `instance ${i}: node ${inst.node} was revoked`);
++ if (p.home_cell !== inst.cell) throw new RegistryError(409, 'registry.node_cell_mismatch', `instance ${i}: node ${inst.node} lives in ${p.home_cell}, not ${inst.cell}`);
++ if (cell.region !== inst.region) throw new RegistryError(400, 'registry.region_mismatch', `instance ${i}: cell ${inst.cell} is in ${cell.region}, not ${inst.region}`);
++ }
++}
++
++/**
++ * Apply a report: check the whole batch, then in one transaction upsert every instance by id (route_weight is left
++ * untouched on update — the scheduler owns weights, section 8) and set this source's other instances to `stopped`.
++ * Throws RegistryError, writing nothing, when a check fails. → { instances, stopped }
++ */
++async function report(db, { source, instances } = {}, now = new Date().toISOString()) {
++ ensureSchema(db);
++ const bad = check({ source, instances });
++ if (bad) {
++ const e = new (require('./cells').RegistryError)(400, 'registry.invalid_report', bad.error);
++ e.details = bad.details;
++ throw e;
++ }
++ await checkPlacement(db, instances);
++ const upsert = db.prepare(`INSERT INTO platform_service_instances (id, service, version, cell, node_id, endpoints, state, source, started_at, reported_at)
++ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
++ ON CONFLICT(id) DO UPDATE SET service = excluded.service, version = excluded.version, cell = excluded.cell,
++ node_id = excluded.node_id, endpoints = excluded.endpoints, state = excluded.state, source = excluded.source,
++ started_at = excluded.started_at, reported_at = excluded.reported_at`);
++ let stopped = 0;
++ await db.tx(async () => {
++ for (const inst of instances) {
++ await upsert.run(inst.id, inst.service, inst.version, inst.cell, inst.node, JSON.stringify(inst.endpoints), inst.state, source, inst.started_at, now);
++ }
++ const ids = new Set(instances.map((i) => i.id));
++ for (const row of await db.prepare('SELECT id FROM platform_service_instances WHERE source = ? AND state <> ?').all(source, 'stopped')) {
++ if (ids.has(row.id)) continue;
++ await db.prepare('UPDATE platform_service_instances SET state = ?, reported_at = ? WHERE id = ?').run('stopped', now, row.id);
++ stopped++;
++ }
++ });
++ return { instances: instances.length, stopped };
++}
++
++function routers({ guard }) {
++ const { http } = require('openvibe-contracts');
++ const internal = express.Router();
++ internal.post('/report', guard, express.json({ limit: '256kb' }), async (req, res) => {
++ let out;
++ try {
++ out = await report(req.app.locals.db, req.body || {});
++ } catch (e) {
++ if (e instanceof require('./cells').RegistryError) return http.sendProblem(res, e.status, e.code, { detail: e.message, errors: e.details });
++ throw e;
++ }
++ res.set('Cache-Control', 'no-store').set('X-Instances-Stopped', String(out.stopped))
++ .json({ source: req.body.source, ...out, generated_at: new Date().toISOString() });
++ });
++ return { internal };
++}
++
++module.exports = { CONTRACT, ensureSchema, check, checkPlacement, report, routers };
ddiff --git a/test/registry-instances.test.js b/test/registry-instances.test.js
nnew file mode 100644
iindex 00000000..465c45e4
---- /dev/null
++++ b/test/registry-instances.test.js
@@@ -0,0 +1,125 @@
++'use strict';
++// Service instances (plan T2, docs/t2-cells-and-node-principal.md section 4.1, slice N2; migration 0008): Host reports
++// what runs on its machines (POST /internal/registry/instances/report, network.node.report). The whole batch is
++// validated against platform.service-instance@1 and placed against the cells and node principals Network holds
++// before anything is written; an instance absent from a later report of the same source is set to `stopped`, never
++// deleted. The topology read gains the instance's cell region.
++// node test/registry-instances.test.js
++const assert = require('assert');
++const crypto = require('crypto');
++const http = require('http');
++const express = require('express');
++const { ids, validate } = require('openvibe-contracts');
++const { getDb } = require('../server/db/database');
++const principals = require('../server/identity/principals');
++const cells = require('../server/registry/cells');
++const instances = require('../server/registry/instances');
++
++(async () => {
++const log = console.log; console.log = () => {};
++const db = getDb();
++console.log = log;
++await db.prepare("UPDATE oauth_clients SET client_secret = 'host-secret' WHERE client_id = 'host'").run();
++await db.prepare("UPDATE oauth_clients SET client_secret = 'live-secret' WHERE client_id = 'live'").run();
++
++const keys = crypto.generateKeyPairSync('rsa', { modulusLength: 2048, publicKeyEncoding: { type: 'spki', format: 'pem' }, privateKeyEncoding: { type: 'pkcs8', format: 'pem' } });
++const app = express();
++app.use(express.urlencoded({ extended: true }));
++app.locals.db = db;
++app.locals.config = { jwt: { issuer: 'https://openvibe.network', accessTokenExpiry: '1h' } };
++app.locals.privateKey = keys.privateKey;
++app.locals.publicKey = keys.publicKey;
++app.use('/oauth', require('../server/auth/oauth-routes'));
++// The report route is mounted before /internal/registry, exactly as server/index.js mounts it.
++app.use('/internal/registry/instances', instances.routers({ guard: principals.guard('network.node.report') }).internal);
++const c = cells.routers({ readGuard: principals.guard('network.registry.read') });
++app.use('/internal/registry', c.internal);
++const server = http.createServer(app);
++
++const nod = () => `nod_${ids.ulid()}`;
++const insertPrincipal = (p) => db.prepare(`INSERT INTO platform_node_principals (id, node_id, home_cell, owner_kind, project_id, trust, status, created_by, revoked_at)
++ VALUES (?, ?, ?, ?, ?, ?, ?, 'test', ?)`).run(p.id || nod(), p.node_id, p.home_cell || 'wnam-1', p.owner_kind || 'platform', p.project_id ?? null, p.trust || 'first-party', p.status || 'active', p.revoked_at ?? null);
++const inst = (id, extra = {}) => ({ id, service: 'media', version: '1.4.0', cell: 'wnam-1', node: 'ovh-1', region: 'us-west', endpoints: ['https://media.internal.example'], state: 'ready', started_at: '2026-10-01T00:00:00Z', ...extra });
++const count = async () => (await db.prepare('SELECT COUNT(*) AS n FROM platform_service_instances').get()).n;
++const stored = async (id) => db.prepare('SELECT * FROM platform_service_instances WHERE id = ?').get(id);
++
++try {
++ // ── Setup: two extra regions, an active cell east-1, a retired cell weu-1, and node principals on wnam-1.
++ await db.prepare("INSERT INTO platform_regions (id, country) VALUES ('us-east', 'US'), ('eu-west', 'IE')").run();
++ await db.prepare("INSERT INTO platform_cells (id, region, residency, status) VALUES ('east-1', 'us-east', 'US', 'active')").run();
++ await db.prepare("INSERT INTO platform_cells (id, region, residency, status) VALUES ('weu-1', 'eu-west', 'IE', 'retired')").run();
++ await insertPrincipal({ node_id: 'ovh-1' });
++ await insertPrincipal({ node_id: 'old-1', status: 'revoked', revoked_at: '2026-10-01T00:00:00Z' });
++
++ await new Promise((ok) => server.listen(0, '127.0.0.1', ok));
++ const base = `http://127.0.0.1:${server.address().port}`;
++ const token = async (id, secret) => (await (await fetch(`${base}/oauth/token`, { method: 'POST', headers: { 'content-type': 'application/x-www-form-urlencoded' }, body: new URLSearchParams({ grant_type: 'client_credentials', client_id: id, client_secret: secret, audience: 'openvibe.network' }) })).json()).access_token;
++ const call = (method, p, body, headers = {}) => fetch(`${base}${p}`, { method, headers: { 'content-type': 'application/json', ...headers }, body: body ? JSON.stringify(body) : undefined })
++ .then(async (x) => ({ status: x.status, headers: x.headers, body: await x.json().catch(() => null) }));
++ const report = (source, list, headers) => call('POST', '/internal/registry/instances/report', { source, instances: list }, headers);
++
++ // ── Auth: nobody, and a token without network.node.report, are refused.
++ assert.strictEqual((await report('primary', [] )).status, 403, 'nobody');
++ const live = { authorization: `Bearer ${await token('live', 'live-secret')}` };
++ assert.strictEqual((await report('primary', [], live)).status, 403, 'Live lacks network.node.report');
++ assert.strictEqual(await count(), 0, 'a refused request wrote nothing');
++
++ // ── A bad instance anywhere in the batch fails the whole batch: nothing is written.
++ const host = { authorization: `Bearer ${await token('host', 'host-secret')}` };
++ let x = await report('primary', [inst('media-ok'), inst('media-bad', { state: 'running' })], host);
++ assert.strictEqual(x.status, 400, JSON.stringify(x.body));
++ assert.strictEqual(await count(), 0, 'one bad instance wrote nothing');
++
++ // ── Placement, each with nothing written: unknown node, wrong cell, revoked node, retired cell, wrong region.
++ for (const [bad, status, code] of [
++ [inst('media-unknown', { node: 'ghost-1' }), 400, 'registry.unknown_node'],
++ [inst('media-cell', { cell: 'east-1', node: 'ovh-1', region: 'us-east' }), 409, 'registry.node_cell_mismatch'],
++ [inst('media-revoked', { node: 'old-1' }), 409, 'registry.node_revoked'],
++ [inst('media-retired', { cell: 'weu-1', region: 'eu-west' }), 409, 'registry.cell_retired'],
++ [inst('media-region', { region: 'eu-west' }), 400, 'registry.region_mismatch'],
++ ]) {
++ x = await report('primary', [inst('media-ok'), bad], host);
++ assert.deepStrictEqual([x.status, x.body.code], [status, code], JSON.stringify(x.body));
++ assert.strictEqual(await count(), 0, `${code} wrote nothing`);
++ }
++
++ // ── The good report: upsert, and an update leaves a hand-set route_weight untouched (the scheduler owns it).
++ x = await report('primary', [inst('media-1')], host);
++ assert.strictEqual(x.status, 200, JSON.stringify(x.body));
++ assert.strictEqual(x.headers.get('cache-control'), 'no-store');
++ assert.strictEqual(x.headers.get('x-instances-stopped'), '0');
++ await db.prepare("UPDATE platform_service_instances SET route_weight = 7 WHERE id = 'media-1'").run();
++ x = await report('primary', [inst('media-1', { version: '1.5.0' })], host);
++ assert.strictEqual(x.status, 200, JSON.stringify(x.body));
++ assert.deepStrictEqual([(await stored('media-1')).version, Number((await stored('media-1')).route_weight)], ['1.5.0', 7], 'route_weight survives an update');
++
++ // ── An instance absent from the same source's next report is stopped, never deleted; another source is untouched.
++ x = await report('primary', [inst('media-1'), inst('media-2')], host);
++ assert.strictEqual(x.status, 200, JSON.stringify(x.body));
++ x = await report('secondary', [inst('media-3')], host);
++ assert.strictEqual(x.status, 200, JSON.stringify(x.body));
++ x = await report('primary', [inst('media-1')], host);
++ assert.strictEqual(x.headers.get('x-instances-stopped'), '1');
++ assert.deepStrictEqual([(await stored('media-2')).state, (await stored('media-3')).state], ['stopped', 'ready'], 'only the absent instance of the same source is stopped');
++ assert.ok(await stored('media-2'), 'a stopped instance is kept, never deleted');
++ x = await report('primary', [inst('media-1')], host);
++ assert.strictEqual(x.headers.get('x-instances-stopped'), '0', 'an already-stopped instance is not counted again');
++
++ // ── The topology read lists the instance with its cell and region.
++ x = await call('GET', '/internal/registry/cells/wnam-1', null, live);
++ assert.strictEqual(x.status, 200, JSON.stringify(x.body));
++ const byId = Object.fromEntries(x.body.instances.map((i) => [i.id, i]));
++ assert.deepStrictEqual([byId['media-1'].cell, byId['media-1'].region, byId['media-1'].node, byId['media-1'].state], ['wnam-1', 'us-west', 'ovh-1', 'ready']);
++
++ // ── Every stored row still matches platform.service-instance@1.
++ const regionOf = Object.fromEntries((await db.prepare('SELECT id, region FROM platform_cells').all()).map((r) => [r.id, r.region]));
++ for (const r of await db.prepare('SELECT * FROM platform_service_instances ORDER BY id').all()) {
++ const doc = { id: r.id, service: r.service, version: r.version, cell: r.cell, node: r.node_id, region: regionOf[r.cell], endpoints: JSON.parse(r.endpoints), state: r.state, started_at: r.started_at, route_weight: Number(r.route_weight) };
++ const v = validate('platform.service-instance@1', doc);
++ assert.ok(v.valid, `${r.id}: ${JSON.stringify(v.errors)}`);
++ }
++ console.log('registry-instances: all tests passed');
++} finally {
++ server.close();
++}
++})().catch((err) => { console.error(err); process.exit(1); });