openvibers/openvibe.tools Commits
Expose owner worker fleet, daily counts and drain controls (#296)
Changes
@@@ -332,18 +332,7 @@ app.locals.config = config;app.use('/auth', createAuthRoutes(config, auth));const workerOwner = require('./workers/owner').ownerGuard({ verify: token => auth.verify(token), extractToken, origin: new URL(config.baseUrl).origin });--app.get('/api/v1/owner/worker-routing', workerOwner, async (_req, res) => {-- try { const r = await workerStore.routing(); res.json({ enabled: r.enabled, order: JSON.parse(r.worker_order) }); }-- catch { res.status(503).json({ error: 'Worker registry unavailable' }); }--});--app.put('/api/v1/owner/worker-routing', workerOwner, express.json({ limit: '4kb' }), async (req, res) => {-- try { res.json(await workerStore.setRouting(req.body || {})); }-- catch { res.status(400).json({ error: 'Invalid worker routing' }); }--});--app.post('/api/v1/owner/workers', workerOwner, express.json({ limit: '4kb' }), async (req, res) => {-- try { res.status(201).json(await workerStore.mint(req.body || {})); }-- catch { res.status(400).json({ error: 'Could not create worker' }); }--});++app.use('/api/v1/owner', require('./workers/admin').createAdmin({ express, guard: workerOwner, store: workerStore, hub: workerHub, ports: SATELLITES }));/** 401 unless a valid ov_token is presented. */nnew file mode 100644@@@ -0,0 +1,57 @@++'use strict';++const os = require('node:os');++function createAdmin({ express, guard, store, hub, ports, fetchImpl = fetch, now = Date.now }) {++ const router = express.Router();++ router.use(guard);++ router.use(express.json({ limit: '4kb' }));++ async function host() {++ const samples = await Promise.all(['img', 'docs', 'audio', 'yt'].map(async app => {++ const port = Number(ports[app]);++ if (!Number.isInteger(port) || port < 1 || port > 65535) return null;++ try {++ const r = await fetchImpl(`http://127.0.0.1:${port}/api/health`, { signal: AbortSignal.timeout(1500), redirect: 'error' });++ const jobs = (await r.json()).jobs;++ const active = jobs?.executing ?? jobs?.running;++ return r.ok && Number.isInteger(jobs?.concurrency) && Number.isInteger(active) ? { slots: jobs.concurrency, active } : null;++ } catch { return null; }++ }));++ const known = samples.filter(Boolean), complete = known.length === samples.length;++ return { id: 'host', label: 'Host fallback', location: 'host', online: known.length > 0, draining: false, revoked: false, slots: complete ? known.reduce((n, s) => n + s.slots, 0) : null, free: complete ? known.reduce((n, s) => n + Math.max(0, s.slots - s.active), 0) : null, active_jobs: complete ? known.reduce((n, s) => n + s.active, 0) : null, load: os.loadavg()[0], latency_ms: null };++ }++ router.get('/workers', async (_req, res) => {++ try {++ const day = new Date(now()).toISOString().slice(0, 10);++ const [rows, config, counts, local] = await Promise.all([store.list(), store.routing(), store.daily(day), host()]);++ const online = new Map(hub.stats().map(c => [c.id, c]));++ const daily = new Map(counts.map(c => [c.worker_id, Number(c.jobs)]));++ const workers = [local, ...rows.map(row => ({ ...row, online: online.has(row.id), slots: 0, free: 0, load: null, latency_ms: null, active_jobs: 0, ...online.get(row.id) }))].map(w => ({ ...w, jobs_today: daily.get(w.id) || 0 }));++ res.json({ workers, routing: { enabled: config.enabled, order: JSON.parse(config.worker_order) }, day, counts: 'execution_attempts' });++ } catch { res.status(503).json({ error: 'Worker registry unavailable' }); }++ });++ router.get('/worker-routing', async (_req, res) => {++ try { const r = await store.routing(); res.json({ enabled: r.enabled, order: JSON.parse(r.worker_order) }); }++ catch { res.status(503).json({ error: 'Worker registry unavailable' }); }++ });++ router.put('/worker-routing', async (req, res) => {++ try { res.json(await store.setRouting(req.body || {})); }++ catch { res.status(400).json({ error: 'Invalid worker routing' }); }++ });++ router.post('/workers', async (req, res) => {++ try { res.status(201).json(await store.mint(req.body || {})); }++ catch { res.status(400).json({ error: 'Could not create worker' }); }++ });++ router.put('/workers/:id/drain', async (req, res) => {++ try {++ const row = await store.drain(req.params.id, req.body?.draining);++ const connection = hub.connections.get(row.id);++ if (connection) connection.draining = row.draining;++ res.json(row);++ } catch { res.status(400).json({ error: 'Invalid worker drain request' }); }++ });++ router.delete('/workers/:id', async (req, res) => {++ try { const row = await store.revoke(req.params.id); hub.connections.get(row.id)?.ws.terminate(); res.json({ id: row.id, revoked: true }); }++ catch { res.status(400).json({ error: 'Invalid worker revocation' }); }++ });++ return router;++}++module.exports = { createAdmin };@@@ -83,6 +83,6 @@ function createHub({ WebSocketServer, store, now = Date.now, heartbeatMs = 15000c.ws.send(JSON.stringify({ type: 'job', id, jobType: job.type, input: job.input, files: job.files }));});}-- return { upgrade, connections, dispatch, stats: () => [...connections.values()].map(({ id, location, draining, capabilities, slots, free, load, latency_ms }) => ({ id, location, draining, capabilities, slots, free, load, latency_ms })), close() { clearInterval(sweep); for (const c of connections.values()) c.ws.terminate(); wss.close(); } };++ return { upgrade, connections, dispatch, stats: () => [...connections.values()].map(({ id, location, draining, capabilities, slots, free, load, latency_ms, pending }) => ({ id, location, draining, capabilities, slots, free: Math.max(0, Math.min(free, slots - pending.size)), load, latency_ms, active_jobs: pending.size })), close() { clearInterval(sweep); for (const c of connections.values()) c.ws.terminate(); wss.close(); } };}module.exports = { createHub };@@@ -17,6 +17,19 @@ function createStore(db) {await db.prepare('UPDATE tools_worker_routing SET enabled=?,worker_order=? WHERE id=1').run(enabled, JSON.stringify(order));return { enabled, order };},++ async drain(id, draining) {++ if (!/^wrk_[a-f0-9]{24}$/.test(id) || typeof draining !== 'boolean') throw new Error('invalid drain');++ const row = await db.prepare('UPDATE tools_workers SET draining=? WHERE id=? AND revoked=FALSE RETURNING id,draining').get(draining, id);++ if (!row) throw new Error('unknown worker');++ return row;++ },++ async revoke(id) {++ if (!/^wrk_[a-f0-9]{24}$/.test(id)) throw new Error('invalid worker');++ const row = await db.prepare('UPDATE tools_workers SET revoked=TRUE,draining=TRUE WHERE id=? RETURNING id').get(id);++ if (!row) throw new Error('unknown worker');++ return row;++ },++ daily: day => db.prepare('SELECT worker_id,jobs FROM tools_worker_daily WHERE day=?').all(day),lookup: id => db.prepare('SELECT id,token_hash,location,draining FROM tools_workers WHERE id=? AND revoked=FALSE').get(id),list: () => db.prepare('SELECT id,label,location,draining,revoked,created_at FROM tools_workers ORDER BY created_at').all(),routing: () => db.prepare('SELECT enabled,worker_order FROM tools_worker_routing WHERE id=1').get(),@@@ -77,6 +77,7 @@ function rawGet(port, p, headers = {}) {await check('ratchet: every file that makes an outbound request itself is reviewed', () => {const REVIEWED = {'_shared/jobs/placement.js': 'fixed loopback gateway port and dispatch path; only durable job id, never caller credentials',++ 'gateway/server/workers/admin.js': 'fixed loopback satellite health paths with operator-configured ports; no caller URL','worker/server/index.js': 'operator-configured HTTPS gateway only; fixed status/connect paths, never job input','gateway/server/auth/routes.js': 'Network JWKS, OAuth token and revoke (configured)','gateway/server/index.js': 'OpenVibe.Community paste API (configured; paths held to /api/pastes) and the satellites on loopback',@@@ -8,7 +8,7 @@ URL-fetching jobs (including YouTube, maps and network tools), SVG inputs, and tUse Ubuntu 22.04 or newer. The installer installs ffmpeg, qpdf and Poppler, and Node 22 if Node 22+ is missing. On either Dell Precision 3620, start with the default two slots; the systemd unit limits work to two CPU cores and 2 GiB RAM. Home upload is the bottleneck: keep URL-fetching traffic on the host.--1. In the owner's Network console, OpenVibe.Tools settings, create a worker labelled for this machine with location `home`. Copy its id and one-time token. Only the token hash remains in the gateway database. The worker controls are being added in the owner-view part of this rollout; the owner API is `POST /api/v1/owner/workers`.++1. In the owner's Network console, OpenVibe.Tools settings, create a worker labelled for this machine with location `home`. Copy its id and one-time token. Only the token hash remains in the gateway database. The owner API is `POST /api/v1/owner/workers`.2. Save the token in a private file on the tower (mode 600). Do not put it in a command, shell history, repository or log.3. Download and inspect the installer, then run it with the id and the **path** to that file:nnew file mode 100644@@@ -0,0 +1,61 @@++'use strict';++const assert = require('node:assert/strict');++const fs = require('node:fs/promises');++const os = require('node:os');++const path = require('node:path');++const express = require('../../gateway/node_modules/express');++const { testDb, closeAllTestDbs } = require('../../_shared/test/testdb');++const { createStore } = require('../../gateway/server/workers/store');++const { createAdmin } = require('../../gateway/server/workers/admin');++const { ownerGuard } = require('../../gateway/server/workers/owner');++const { hash } = require('../server/protocol');++(async () => {++ const dir = await fs.mkdtemp(path.join(os.tmpdir(), 'tools-owner-test-'));++ let server;++ try {++ const db = await testDb(dir), store = createStore(db);++ const minted = await store.mint({ label: 'Tower one', location: 'home' });++ const stored = await store.lookup(minted.id);++ assert.equal(stored.token_hash, hash(minted.token)); assert.notEqual(stored.token_hash, minted.token);++ const at = Date.UTC(2026, 10, 10, 12), day = '2026-11-10';++ await db.prepare("INSERT INTO tool_jobs (id,type,owner,state,created_at,updated_at) VALUES ('job-test','img.process','test-owner','running',?,?)").run(at, at);++ const insert = () => db.prepare("INSERT INTO tool_worker_attempts (job_id,job_attempt,attempt,worker_id,location,took_ms,state,at) VALUES ('job-test',1,0,?,'home',10,'succeeded',?) ON CONFLICT DO NOTHING").run(minted.id, at);++ await insert(); await insert();++ assert.equal(Number((await store.daily(day))[0].jobs), 1, 'replayed assignment does not double-count');++ await db.prepare("DELETE FROM tool_jobs WHERE id='job-test'").run();++ assert.equal(Number((await store.daily(day))[0].jobs), 1, 'daily count survives job expiry/cascade');++ let terminated = false;++ const c = { id: minted.id, slots: 2, free: 1, active_jobs: 1, load: 0.2, latency_ms: 35, draining: false, ws: { terminate() { terminated = true; } } };++ const hub = { connections: new Map([[c.id, c]]), stats: () => [{ id: c.id, slots: c.slots, free: c.free, active_jobs: c.active_jobs, load: c.load, latency_ms: c.latency_ms, draining: c.draining }] };++ const guard = ownerGuard({ verify: async t => t === 'owner-test' ? { username: 'test-owner' } : { username: 'someone' }, extractToken: r => r.headers.authorization, origin: 'https://openvibe.tools', username: 'test-owner' });++ const app = express();++ app.use('/api/v1/owner', createAdmin({ express, guard, store, hub, now: () => at, ports: { img: 1, docs: 2, audio: 3, yt: 4 }, fetchImpl: async url => {++ assert.match(url, /^http:\/\/127\.0\.0\.1:[1-4]\/api\/health$/);++ return { ok: true, json: async () => ({ jobs: { concurrency: 2, executing: 1 } }) };++ } }));++ server = await new Promise(r => { const s = app.listen(0, '127.0.0.1', () => r(s)); });++ const base = `http://127.0.0.1:${server.address().port}/api/v1/owner`;++ async function call(p, method = 'GET', body, auth = 'owner-test', origin = 'https://openvibe.tools') {++ return fetch(base + p, { method, headers: { authorization: auth, origin, 'content-type': 'application/json' }, body: body === undefined ? undefined : JSON.stringify(body) });++ }++ let response = await call('/workers');++ assert.equal(response.headers.get('cache-control'), 'no-store');++ let fleet = await response.json();++ assert.equal(fleet.routing.enabled, false); assert.equal(fleet.workers[0].free, 4);++ assert.equal(fleet.workers[1].jobs_today, 1); assert.equal(fleet.workers[1].online, true);++ assert(!JSON.stringify(fleet).includes(minted.token)); assert(!JSON.stringify(fleet).includes(stored.token_hash));++ assert.equal((await call('/workers', 'GET', undefined, 'not-owner')).status, 403);++ assert.equal((await call('/workers', 'POST', { label: 'refused' }, 'owner-test', 'https://elsewhere.example')).status, 403);++ response = await call(`/workers/${minted.id}/drain`, 'PUT', { draining: true });++ assert.equal(response.status, 200); assert.equal(c.draining, true); assert.equal(terminated, false, 'drain keeps existing work alive');++ assert.equal((await call(`/workers/${minted.id}/drain`, 'PUT', { draining: 'true' })).status, 400);++ response = await call('/worker-routing', 'PUT', { enabled: true, order: [minted.id, 'home'] });++ assert.deepEqual((await response.json()).order, [minted.id, 'home', 'host']);++ assert.equal((await call(`/workers/${minted.id}`, 'DELETE')).status, 200);++ assert.equal(terminated, true); assert.equal(await store.lookup(minted.id), undefined);++ console.log('worker owner: hashed credentials, owner/origin guard, fleet/load/slots, persistent UTC counts, immediate drain, routing and revocation passed');++ } finally {++ if (server) await new Promise(r => server.close(r));++ await closeAllTestDbs(); await fs.rm(dir, { recursive: true, force: true });++ }++})().catch(e => { console.error(e); process.exitCode = 1; });nnew file mode 100644@@@ -0,0 +1,23 @@++-- phase: expand++-- Owner infrastructure counters survive expiry of the underlying job and its attempt rows.++CREATE TABLE IF NOT EXISTS tools_worker_daily (++ worker_id TEXT NOT NULL,++ day TEXT NOT NULL,++ jobs BIGINT NOT NULL DEFAULT 0,++ PRIMARY KEY (worker_id, day)++);++LOCK TABLE tool_worker_attempts IN SHARE ROW EXCLUSIVE MODE;++INSERT INTO tools_worker_daily (worker_id, day, jobs)++SELECT worker_id, to_char(to_timestamp(at / 1000.0) AT TIME ZONE 'UTC', 'YYYY-MM-DD'), COUNT(*)++FROM tool_worker_attempts GROUP BY 1, 2++ON CONFLICT DO NOTHING;++CREATE OR REPLACE FUNCTION tools_count_worker_attempt() RETURNS trigger LANGUAGE plpgsql AS $$++BEGIN++ INSERT INTO tools_worker_daily (worker_id, day, jobs)++ VALUES (NEW.worker_id, to_char(to_timestamp(NEW.at / 1000.0) AT TIME ZONE 'UTC', 'YYYY-MM-DD'), 1)++ ON CONFLICT (worker_id, day) DO UPDATE SET jobs = tools_worker_daily.jobs + 1;++ RETURN NEW;++END;++$$;++CREATE TRIGGER tools_worker_attempt_count AFTER INSERT ON tool_worker_attempts++FOR EACH ROW EXECUTE FUNCTION tools_count_worker_attempt();@@@ -2,7 +2,7 @@"name": "openvibe-tools","version": "1.0.0","private": true,-- "description": "OpenVibe.Tools — the tools suite on openvibe.tools and *.openvibe.tools",++ "description": "OpenVibe.Tools \u2014 the tools suite on openvibe.tools and *.openvibe.tools","scripts": {"test": "node test/run.js","test:pg": "DATABASE_URL=\"$OV_TEST_PG_URL\" DATABASE_DIRECT_URL=\"$OV_TEST_PG_DIRECT_URL\" VALKEY_URL=\"$OV_TEST_VALKEY_URL\" node test/run.js",@@@ -21,7 +21,9 @@"test:worker-security": "node apps/gateway/test/security-ssrf.test.js","test:routing": "node apps/worker/test/routing.test.js && node apps/worker/test/worker.test.js && node apps/_shared/test/jobs.test.js && node apps/gateway/test/security-ssrf.test.js","test:worker-docs": "node apps/gateway/test/openapi.test.js",-- "test:worker-setup": "node apps/worker/test/deploy.test.js && node apps/worker/test/worker.test.js && node apps/gateway/test/security-ssrf.test.js"++ "test:worker-setup": "node apps/worker/test/deploy.test.js && node apps/worker/test/worker.test.js && node apps/gateway/test/security-ssrf.test.js",++ "test:worker-owner": "node apps/worker/test/admin.test.js && node apps/worker/test/routing.test.js && node apps/gateway/test/security-ssrf.test.js",++ "test:worker-owner-db": "node apps/worker/test/admin.test.js"},"license": "MIT","engines": {
@@ -332,18 +332,7 @@ app.locals.config = config;
@@ -332,18 +332,7 @@ app.locals.config = config;
app.use('/auth', createAuthRoutes(config, auth));
app.use('/auth', createAuthRoutes(config, auth));
const workerOwner = require('./workers/owner').ownerGuard({ verify: token => auth.verify(token), extractToken, origin: new URL(config.baseUrl).origin });
const workerOwner = require('./workers/owner').ownerGuard({ verify: token => auth.verify(token), extractToken, origin: new URL(config.baseUrl).origin });
-app.get('/api/v1/owner/worker-routing', workerOwner, async (_req, res) => {
- try { const r = await workerStore.routing(); res.json({ enabled: r.enabled, order: JSON.parse(r.worker_order) }); }
- catch { res.status(503).json({ error: 'Worker registry unavailable' }); }
-});
-app.put('/api/v1/owner/worker-routing', workerOwner, express.json({ limit: '4kb' }), async (req, res) => {
- try { res.json(await workerStore.setRouting(req.body || {})); }
- catch { res.status(400).json({ error: 'Invalid worker routing' }); }
-});
-app.post('/api/v1/owner/workers', workerOwner, express.json({ limit: '4kb' }), async (req, res) => {
- try { res.status(201).json(await workerStore.mint(req.body || {})); }
- catch { res.status(400).json({ error: 'Could not create worker' }); }
-});
+app.use('/api/v1/owner', require('./workers/admin').createAdmin({ express, guard: workerOwner, store: workerStore, hub: workerHub, ports: SATELLITES }));
/** 401 unless a valid ov_token is presented. */
/** 401 unless a valid ov_token is presented. */
new file mode 100644
new file mode 100644
@@ -0,0 +1,57 @@
@@ -0,0 +1,57 @@
+'use strict';
+const os = require('node:os');
+function createAdmin({ express, guard, store, hub, ports, fetchImpl = fetch, now = Date.now }) {
+ const router = express.Router();
+ router.use(guard);
+ router.use(express.json({ limit: '4kb' }));
+ async function host() {
+ const samples = await Promise.all(['img', 'docs', 'audio', 'yt'].map(async app => {
+ const port = Number(ports[app]);
+ if (!Number.isInteger(port) || port < 1 || port > 65535) return null;
+ try {
+ const r = await fetchImpl(`http://127.0.0.1:${port}/api/health`, { signal: AbortSignal.timeout(1500), redirect: 'error' });
+ const jobs = (await r.json()).jobs;
+ const active = jobs?.executing ?? jobs?.running;
+ return r.ok && Number.isInteger(jobs?.concurrency) && Number.isInteger(active) ? { slots: jobs.concurrency, active } : null;
+ } catch { return null; }
+ }));
+ const known = samples.filter(Boolean), complete = known.length === samples.length;
+ return { id: 'host', label: 'Host fallback', location: 'host', online: known.length > 0, draining: false, revoked: false, slots: complete ? known.reduce((n, s) => n + s.slots, 0) : null, free: complete ? known.reduce((n, s) => n + Math.max(0, s.slots - s.active), 0) : null, active_jobs: complete ? known.reduce((n, s) => n + s.active, 0) : null, load: os.loadavg()[0], latency_ms: null };
+ }
+ router.get('/workers', async (_req, res) => {
+ try {
+ const day = new Date(now()).toISOString().slice(0, 10);
+ const [rows, config, counts, local] = await Promise.all([store.list(), store.routing(), store.daily(day), host()]);
+ const online = new Map(hub.stats().map(c => [c.id, c]));
+ const daily = new Map(counts.map(c => [c.worker_id, Number(c.jobs)]));
+ const workers = [local, ...rows.map(row => ({ ...row, online: online.has(row.id), slots: 0, free: 0, load: null, latency_ms: null, active_jobs: 0, ...online.get(row.id) }))].map(w => ({ ...w, jobs_today: daily.get(w.id) || 0 }));
+ res.json({ workers, routing: { enabled: config.enabled, order: JSON.parse(config.worker_order) }, day, counts: 'execution_attempts' });
+ } catch { res.status(503).json({ error: 'Worker registry unavailable' }); }
+ });
+ router.get('/worker-routing', async (_req, res) => {
+ try { const r = await store.routing(); res.json({ enabled: r.enabled, order: JSON.parse(r.worker_order) }); }
+ catch { res.status(503).json({ error: 'Worker registry unavailable' }); }
+ });
+ router.put('/worker-routing', async (req, res) => {
+ try { res.json(await store.setRouting(req.body || {})); }
+ catch { res.status(400).json({ error: 'Invalid worker routing' }); }
+ });
+ router.post('/workers', async (req, res) => {
+ try { res.status(201).json(await store.mint(req.body || {})); }
+ catch { res.status(400).json({ error: 'Could not create worker' }); }
+ });
+ router.put('/workers/:id/drain', async (req, res) => {
+ try {
+ const row = await store.drain(req.params.id, req.body?.draining);
+ const connection = hub.connections.get(row.id);
+ if (connection) connection.draining = row.draining;
+ res.json(row);
+ } catch { res.status(400).json({ error: 'Invalid worker drain request' }); }
+ });
+ router.delete('/workers/:id', async (req, res) => {
+ try { const row = await store.revoke(req.params.id); hub.connections.get(row.id)?.ws.terminate(); res.json({ id: row.id, revoked: true }); }
+ catch { res.status(400).json({ error: 'Invalid worker revocation' }); }
+ });
+ return router;
+}
+module.exports = { createAdmin };
@@ -83,6 +83,6 @@ function createHub({ WebSocketServer, store, now = Date.now, heartbeatMs = 15000
@@ -83,6 +83,6 @@ function createHub({ WebSocketServer, store, now = Date.now, heartbeatMs = 15000
c.ws.send(JSON.stringify({ type: 'job', id, jobType: job.type, input: job.input, files: job.files }));
c.ws.send(JSON.stringify({ type: 'job', id, jobType: job.type, input: job.input, files: job.files }));
});
});
}
}
- return { upgrade, connections, dispatch, stats: () => [...connections.values()].map(({ id, location, draining, capabilities, slots, free, load, latency_ms }) => ({ id, location, draining, capabilities, slots, free, load, latency_ms })), close() { clearInterval(sweep); for (const c of connections.values()) c.ws.terminate(); wss.close(); } };
+ return { upgrade, connections, dispatch, stats: () => [...connections.values()].map(({ id, location, draining, capabilities, slots, free, load, latency_ms, pending }) => ({ id, location, draining, capabilities, slots, free: Math.max(0, Math.min(free, slots - pending.size)), load, latency_ms, active_jobs: pending.size })), close() { clearInterval(sweep); for (const c of connections.values()) c.ws.terminate(); wss.close(); } };
}
}
module.exports = { createHub };
module.exports = { createHub };
@@ -17,6 +17,19 @@ function createStore(db) {
@@ -17,6 +17,19 @@ function createStore(db) {
await db.prepare('UPDATE tools_worker_routing SET enabled=?,worker_order=? WHERE id=1').run(enabled, JSON.stringify(order));
await db.prepare('UPDATE tools_worker_routing SET enabled=?,worker_order=? WHERE id=1').run(enabled, JSON.stringify(order));
return { enabled, order };
return { enabled, order };
},
},
+ async drain(id, draining) {
+ if (!/^wrk_[a-f0-9]{24}$/.test(id) || typeof draining !== 'boolean') throw new Error('invalid drain');
+ const row = await db.prepare('UPDATE tools_workers SET draining=? WHERE id=? AND revoked=FALSE RETURNING id,draining').get(draining, id);
+ if (!row) throw new Error('unknown worker');
+ return row;
+ },
+ async revoke(id) {
+ if (!/^wrk_[a-f0-9]{24}$/.test(id)) throw new Error('invalid worker');
+ const row = await db.prepare('UPDATE tools_workers SET revoked=TRUE,draining=TRUE WHERE id=? RETURNING id').get(id);
+ if (!row) throw new Error('unknown worker');
+ return row;
+ },
+ daily: day => db.prepare('SELECT worker_id,jobs FROM tools_worker_daily WHERE day=?').all(day),
lookup: id => db.prepare('SELECT id,token_hash,location,draining FROM tools_workers WHERE id=? AND revoked=FALSE').get(id),
lookup: id => db.prepare('SELECT id,token_hash,location,draining FROM tools_workers WHERE id=? AND revoked=FALSE').get(id),
list: () => db.prepare('SELECT id,label,location,draining,revoked,created_at FROM tools_workers ORDER BY created_at').all(),
list: () => db.prepare('SELECT id,label,location,draining,revoked,created_at FROM tools_workers ORDER BY created_at').all(),
routing: () => db.prepare('SELECT enabled,worker_order FROM tools_worker_routing WHERE id=1').get(),
routing: () => db.prepare('SELECT enabled,worker_order FROM tools_worker_routing WHERE id=1').get(),
@@ -77,6 +77,7 @@ function rawGet(port, p, headers = {}) {
@@ -77,6 +77,7 @@ function rawGet(port, p, headers = {}) {
await check('ratchet: every file that makes an outbound request itself is reviewed', () => {
await check('ratchet: every file that makes an outbound request itself is reviewed', () => {
const REVIEWED = {
const REVIEWED = {
'_shared/jobs/placement.js': 'fixed loopback gateway port and dispatch path; only durable job id, never caller credentials',
'_shared/jobs/placement.js': 'fixed loopback gateway port and dispatch path; only durable job id, never caller credentials',
+ 'gateway/server/workers/admin.js': 'fixed loopback satellite health paths with operator-configured ports; no caller URL',
'worker/server/index.js': 'operator-configured HTTPS gateway only; fixed status/connect paths, never job input',
'worker/server/index.js': 'operator-configured HTTPS gateway only; fixed status/connect paths, never job input',
'gateway/server/auth/routes.js': 'Network JWKS, OAuth token and revoke (configured)',
'gateway/server/auth/routes.js': 'Network JWKS, OAuth token and revoke (configured)',
'gateway/server/index.js': 'OpenVibe.Community paste API (configured; paths held to /api/pastes) and the satellites on loopback',
'gateway/server/index.js': 'OpenVibe.Community paste API (configured; paths held to /api/pastes) and the satellites on loopback',
@@ -8,7 +8,7 @@ URL-fetching jobs (including YouTube, maps and network tools), SVG inputs, and t
@@ -8,7 +8,7 @@ URL-fetching jobs (including YouTube, maps and network tools), SVG inputs, and t
Use Ubuntu 22.04 or newer. The installer installs ffmpeg, qpdf and Poppler, and Node 22 if Node 22+ is missing. On either Dell Precision 3620, start with the default two slots; the systemd unit limits work to two CPU cores and 2 GiB RAM. Home upload is the bottleneck: keep URL-fetching traffic on the host.
Use Ubuntu 22.04 or newer. The installer installs ffmpeg, qpdf and Poppler, and Node 22 if Node 22+ is missing. On either Dell Precision 3620, start with the default two slots; the systemd unit limits work to two CPU cores and 2 GiB RAM. Home upload is the bottleneck: keep URL-fetching traffic on the host.
-1. In the owner's Network console, OpenVibe.Tools settings, create a worker labelled for this machine with location `home`. Copy its id and one-time token. Only the token hash remains in the gateway database. The worker controls are being added in the owner-view part of this rollout; the owner API is `POST /api/v1/owner/workers`.
+1. In the owner's Network console, OpenVibe.Tools settings, create a worker labelled for this machine with location `home`. Copy its id and one-time token. Only the token hash remains in the gateway database. The owner API is `POST /api/v1/owner/workers`.
2. Save the token in a private file on the tower (mode 600). Do not put it in a command, shell history, repository or log.
2. Save the token in a private file on the tower (mode 600). Do not put it in a command, shell history, repository or log.
3. Download and inspect the installer, then run it with the id and the **path** to that file:
3. Download and inspect the installer, then run it with the id and the **path** to that file:
new file mode 100644
new file mode 100644
@@ -0,0 +1,61 @@
@@ -0,0 +1,61 @@
+'use strict';
+const assert = require('node:assert/strict');
+const fs = require('node:fs/promises');
+const os = require('node:os');
+const path = require('node:path');
+const express = require('../../gateway/node_modules/express');
+const { testDb, closeAllTestDbs } = require('../../_shared/test/testdb');
+const { createStore } = require('../../gateway/server/workers/store');
+const { createAdmin } = require('../../gateway/server/workers/admin');
+const { ownerGuard } = require('../../gateway/server/workers/owner');
+const { hash } = require('../server/protocol');
+(async () => {
+ const dir = await fs.mkdtemp(path.join(os.tmpdir(), 'tools-owner-test-'));
+ let server;
+ try {
+ const db = await testDb(dir), store = createStore(db);
+ const minted = await store.mint({ label: 'Tower one', location: 'home' });
+ const stored = await store.lookup(minted.id);
+ assert.equal(stored.token_hash, hash(minted.token)); assert.notEqual(stored.token_hash, minted.token);
+ const at = Date.UTC(2026, 10, 10, 12), day = '2026-11-10';
+ await db.prepare("INSERT INTO tool_jobs (id,type,owner,state,created_at,updated_at) VALUES ('job-test','img.process','test-owner','running',?,?)").run(at, at);
+ const insert = () => db.prepare("INSERT INTO tool_worker_attempts (job_id,job_attempt,attempt,worker_id,location,took_ms,state,at) VALUES ('job-test',1,0,?,'home',10,'succeeded',?) ON CONFLICT DO NOTHING").run(minted.id, at);
+ await insert(); await insert();
+ assert.equal(Number((await store.daily(day))[0].jobs), 1, 'replayed assignment does not double-count');
+ await db.prepare("DELETE FROM tool_jobs WHERE id='job-test'").run();
+ assert.equal(Number((await store.daily(day))[0].jobs), 1, 'daily count survives job expiry/cascade');
+ let terminated = false;
+ const c = { id: minted.id, slots: 2, free: 1, active_jobs: 1, load: 0.2, latency_ms: 35, draining: false, ws: { terminate() { terminated = true; } } };
+ const hub = { connections: new Map([[c.id, c]]), stats: () => [{ id: c.id, slots: c.slots, free: c.free, active_jobs: c.active_jobs, load: c.load, latency_ms: c.latency_ms, draining: c.draining }] };
+ const guard = ownerGuard({ verify: async t => t === 'owner-test' ? { username: 'test-owner' } : { username: 'someone' }, extractToken: r => r.headers.authorization, origin: 'https://openvibe.tools', username: 'test-owner' });
+ const app = express();
+ app.use('/api/v1/owner', createAdmin({ express, guard, store, hub, now: () => at, ports: { img: 1, docs: 2, audio: 3, yt: 4 }, fetchImpl: async url => {
+ assert.match(url, /^http:\/\/127\.0\.0\.1:[1-4]\/api\/health$/);
+ return { ok: true, json: async () => ({ jobs: { concurrency: 2, executing: 1 } }) };
+ } }));
+ server = await new Promise(r => { const s = app.listen(0, '127.0.0.1', () => r(s)); });
+ const base = `http://127.0.0.1:${server.address().port}/api/v1/owner`;
+ async function call(p, method = 'GET', body, auth = 'owner-test', origin = 'https://openvibe.tools') {
+ return fetch(base + p, { method, headers: { authorization: auth, origin, 'content-type': 'application/json' }, body: body === undefined ? undefined : JSON.stringify(body) });
+ }
+ let response = await call('/workers');
+ assert.equal(response.headers.get('cache-control'), 'no-store');
+ let fleet = await response.json();
+ assert.equal(fleet.routing.enabled, false); assert.equal(fleet.workers[0].free, 4);
+ assert.equal(fleet.workers[1].jobs_today, 1); assert.equal(fleet.workers[1].online, true);
+ assert(!JSON.stringify(fleet).includes(minted.token)); assert(!JSON.stringify(fleet).includes(stored.token_hash));
+ assert.equal((await call('/workers', 'GET', undefined, 'not-owner')).status, 403);
+ assert.equal((await call('/workers', 'POST', { label: 'refused' }, 'owner-test', 'https://elsewhere.example')).status, 403);
+ response = await call(`/workers/${minted.id}/drain`, 'PUT', { draining: true });
+ assert.equal(response.status, 200); assert.equal(c.draining, true); assert.equal(terminated, false, 'drain keeps existing work alive');
+ assert.equal((await call(`/workers/${minted.id}/drain`, 'PUT', { draining: 'true' })).status, 400);
+ response = await call('/worker-routing', 'PUT', { enabled: true, order: [minted.id, 'home'] });
+ assert.deepEqual((await response.json()).order, [minted.id, 'home', 'host']);
+ assert.equal((await call(`/workers/${minted.id}`, 'DELETE')).status, 200);
+ assert.equal(terminated, true); assert.equal(await store.lookup(minted.id), undefined);
+ console.log('worker owner: hashed credentials, owner/origin guard, fleet/load/slots, persistent UTC counts, immediate drain, routing and revocation passed');
+ } finally {
+ if (server) await new Promise(r => server.close(r));
+ await closeAllTestDbs(); await fs.rm(dir, { recursive: true, force: true });
+ }
+})().catch(e => { console.error(e); process.exitCode = 1; });
new file mode 100644
new file mode 100644
@@ -0,0 +1,23 @@
@@ -0,0 +1,23 @@
+-- phase: expand
+-- Owner infrastructure counters survive expiry of the underlying job and its attempt rows.
+CREATE TABLE IF NOT EXISTS tools_worker_daily (
+ worker_id TEXT NOT NULL,
+ day TEXT NOT NULL,
+ jobs BIGINT NOT NULL DEFAULT 0,
+ PRIMARY KEY (worker_id, day)
+);
+LOCK TABLE tool_worker_attempts IN SHARE ROW EXCLUSIVE MODE;
+INSERT INTO tools_worker_daily (worker_id, day, jobs)
+SELECT worker_id, to_char(to_timestamp(at / 1000.0) AT TIME ZONE 'UTC', 'YYYY-MM-DD'), COUNT(*)
+FROM tool_worker_attempts GROUP BY 1, 2
+ON CONFLICT DO NOTHING;
+CREATE OR REPLACE FUNCTION tools_count_worker_attempt() RETURNS trigger LANGUAGE plpgsql AS $$
+BEGIN
+ INSERT INTO tools_worker_daily (worker_id, day, jobs)
+ VALUES (NEW.worker_id, to_char(to_timestamp(NEW.at / 1000.0) AT TIME ZONE 'UTC', 'YYYY-MM-DD'), 1)
+ ON CONFLICT (worker_id, day) DO UPDATE SET jobs = tools_worker_daily.jobs + 1;
+ RETURN NEW;
+END;
+$$;
+CREATE TRIGGER tools_worker_attempt_count AFTER INSERT ON tool_worker_attempts
+FOR EACH ROW EXECUTE FUNCTION tools_count_worker_attempt();
@@ -2,7 +2,7 @@
@@ -2,7 +2,7 @@
"name": "openvibe-tools",
"name": "openvibe-tools",
"version": "1.0.0",
"version": "1.0.0",
"private": true,
"private": true,
- "description": "OpenVibe.Tools — the tools suite on openvibe.tools and *.openvibe.tools",
+ "description": "OpenVibe.Tools \u2014 the tools suite on openvibe.tools and *.openvibe.tools",
"scripts": {
"scripts": {
"test": "node test/run.js",
"test": "node test/run.js",
"test:pg": "DATABASE_URL=\"$OV_TEST_PG_URL\" DATABASE_DIRECT_URL=\"$OV_TEST_PG_DIRECT_URL\" VALKEY_URL=\"$OV_TEST_VALKEY_URL\" node test/run.js",
"test:pg": "DATABASE_URL=\"$OV_TEST_PG_URL\" DATABASE_DIRECT_URL=\"$OV_TEST_PG_DIRECT_URL\" VALKEY_URL=\"$OV_TEST_VALKEY_URL\" node test/run.js",
@@ -21,7 +21,9 @@
@@ -21,7 +21,9 @@
"test:worker-security": "node apps/gateway/test/security-ssrf.test.js",
"test:worker-security": "node apps/gateway/test/security-ssrf.test.js",
"test:routing": "node apps/worker/test/routing.test.js && node apps/worker/test/worker.test.js && node apps/_shared/test/jobs.test.js && node apps/gateway/test/security-ssrf.test.js",
"test:routing": "node apps/worker/test/routing.test.js && node apps/worker/test/worker.test.js && node apps/_shared/test/jobs.test.js && node apps/gateway/test/security-ssrf.test.js",
"test:worker-docs": "node apps/gateway/test/openapi.test.js",
"test:worker-docs": "node apps/gateway/test/openapi.test.js",
- "test:worker-setup": "node apps/worker/test/deploy.test.js && node apps/worker/test/worker.test.js && node apps/gateway/test/security-ssrf.test.js"
+ "test:worker-setup": "node apps/worker/test/deploy.test.js && node apps/worker/test/worker.test.js && node apps/gateway/test/security-ssrf.test.js",
+ "test:worker-owner": "node apps/worker/test/admin.test.js && node apps/worker/test/routing.test.js && node apps/gateway/test/security-ssrf.test.js",
+ "test:worker-owner-db": "node apps/worker/test/admin.test.js"
},
},
"license": "MIT",
"license": "MIT",
"engines": {
"engines": {