openvibers/openvibe.tools Commits
Tools anywhere 2: capacity routing, host fallback and job receipts (#289)
Changes
@@@ -174,7 +174,7 @@ function mountJobRoutes(app, o) {if (!ok) { for (const f of files) if (f.path) fs.unlink(f.path, () => {}); return; }}try {-- const { job, replayed } = await system.submit({ owner: w.owner, type: String(b.type || ''), input, files, idempotencyKey: key, ttlMs: ttlMs(req, w), env: w.env || 'production', ipKey: w.caller ? w.caller.ipKey : null, project: w.kind === 'principal' && w.claims && w.claims.actor_type === 'app' ? w.claims.project_id : null, traceId: req.ov && req.ov.traceId });++ const { job, replayed } = await system.submit({ prefer: req.headers['x-tools-prefer'] || 'auto', owner: w.owner, type: String(b.type || ''), input, files, idempotencyKey: key, ttlMs: ttlMs(req, w), env: w.env || 'production', ipKey: w.caller ? w.caller.ipKey : null, project: w.kind === 'principal' && w.claims && w.claims.actor_type === 'app' ? w.claims.project_id : null, traceId: req.ov && req.ov.traceId });res.set('Location', `/api/v1/jobs/${job.id}`);if (replayed) res.set('Idempotent-Replayed', 'true');return res.status(replayed ? 200 : 202).json(await system.view(job));nnew file mode 100644@@@ -0,0 +1,59 @@++'use strict';++const fs = require('node:fs/promises');++const path = require('node:path');++const { MAX_FRAME, bytes, TYPES } = require('../../worker/server/protocol');++// This is a fixed loopback channel, not a URL supplied by a job or a person.++async function requestPlacement(id, signal, fetchImpl = fetch) {++ const port = Number(process.env.TOOLS_GATEWAY_INTERNAL_PORT || 4001);++ if (!Number.isInteger(port) || port < 1 || port > 65535) throw new Error('invalid gateway port');++ const r = await fetchImpl(`http://127.0.0.1:${port}/api/internal/workers/dispatch`, {++ method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ job_id: id }), signal, redirect: 'error',++ });++ if (!r.ok) throw new Error('gateway unavailable');++ let size = 0; const chunks = [];++ for await (const chunk of r.body) { size += chunk.length; if (size > MAX_FRAME) throw new Error('placement response too large'); chunks.push(chunk); }++ return JSON.parse(Buffer.concat(chunks).toString('utf8'));++}++async function executePlaced({ row, def, context, receipt, request = requestPlacement, loadAttempts = async () => [], now = Date.now }) {++ let placement = { mode: 'host', attempts: [] };++ if (TYPES.includes(row.type) && row.execution_prefer !== 'host') {++ try { placement = await request(row.id, context.signal); }++ catch {++ placement.attempts = (await loadAttempts()).map(a => ({ ...a, state: a.state === 'running' ? (context.signal.aborted ? 'cancelled' : 'disconnected') : a.state }));++ receipt.attempts = placement.attempts;++ if (context.signal.aborted) throw new Error('cancelled');++ if (placement.attempts.length >= 2) placement.mode = 'failed';++ }++ }++ receipt.attempts = placement.attempts || [];++ if (placement.mode === 'failed') {++ const last = receipt.attempts.at(-1);++ Object.assign(receipt, { worker: last.worker, location: last.location, took_ms: receipt.attempts.reduce((n, a) => n + a.took_ms, 0) });++ throw new Error('The remote worker could not complete this job');++ }++ if (placement.mode === 'remote') {++ const last = receipt.attempts.at(-1);++ Object.assign(receipt, { worker: last.worker, location: last.location, took_ms: receipt.attempts.reduce((n, a) => n + a.took_ms, 0) });++ const files = [];++ for (const [i, f] of placement.result.files.entries()) {++ const local = path.join(context.outDir, `remote-${i}`);++ await fs.writeFile(local, bytes(f));++ files.push({ path: local, name: path.basename(f.name), mime: f.mime });++ }++ return { files, data: { ...placement.result.data, execution: receipt } };++ }++ const started = now();++ const attempt = { worker: 'host', location: 'host', took_ms: 0, state: 'failed' };++ receipt.attempts.push(attempt);++ Object.assign(receipt, { worker: 'host', location: 'host' });++ try {++ const out = await def.run(context);++ attempt.state = 'succeeded';++ return { ...out, data: { ...(out?.data || {}), execution: receipt } };++ } finally {++ attempt.took_ms = now() - started;++ if (context.signal.aborted) attempt.state = 'cancelled';++ receipt.took_ms = receipt.attempts.reduce((n, a) => n + a.took_ms, 0);++ }++}++module.exports = { executePlaced, requestPlacement };@@@ -28,8 +28,8 @@ function createStore(db, { app } = {}) {if (!db || typeof db.prepare !== 'function') throw new TypeError('createStore needs an openvibe-sdk/db handle');if (!app) throw new TypeError('createStore needs app');const q = {-- insert: db.prepare(`INSERT INTO tool_jobs (id, app, type, type_version, owner, state, input_json, files_json, idempotency_key, request_hash, max_attempts, ttl_ms, created_at, updated_at, env, retry_of, ip_key, tool, project_id, trace_id)-- VALUES (@id, @app, @type, @type_version, @owner, 'queued', @input_json, @files_json, @idempotency_key, @request_hash, @max_attempts, @ttl_ms, @now, @now, @env, @retry_of, @ip_key, @tool, @project_id, @trace_id)`),++ insert: db.prepare(`INSERT INTO tool_jobs (id, app, type, type_version, owner, state, input_json, files_json, idempotency_key, request_hash, max_attempts, ttl_ms, created_at, updated_at, env, retry_of, ip_key, tool, project_id, trace_id, execution_prefer)++ VALUES (@id, @app, @type, @type_version, @owner, 'queued', @input_json, @files_json, @idempotency_key, @request_hash, @max_attempts, @ttl_ms, @now, @now, @env, @retry_of, @ip_key, @tool, @project_id, @trace_id, @execution_prefer)`),markRetried: db.prepare("UPDATE tool_jobs SET retried_by = @next, updated_at = @now WHERE id = @id AND app = @app AND state = 'failed' AND retried_by IS NULL"),get: db.prepare('SELECT * FROM tool_jobs WHERE id = ? AND app = ?'),byIdem: db.prepare('SELECT * FROM tool_jobs WHERE app = @app AND owner = @owner AND idempotency_key = @idempotency_key'),@@@ -82,7 +82,14 @@ function createStore(db, { app } = {}) {db,app,TERMINAL, STATES,-- async insert(row) { await q.insert.run({ app, retry_of: null, ip_key: null, tool: null, project_id: null, trace_id: null, ...row }); },++ async execution(id, jobAttempt) {++ const rows = await db.prepare('SELECT worker_id,location,took_ms,state,at FROM tool_worker_attempts WHERE job_id=? AND job_attempt=? ORDER BY attempt').all(id, jobAttempt);++ return rows.map(a => ({ worker: a.worker_id, location: a.location, took_ms: a.state === 'running' ? Math.max(0, Date.now() - Number(a.at)) : Number(a.took_ms), state: a.state }));++ },++ async recordExecution(id, jobAttempt, attempts) {++ for (const [i, a] of attempts.entries()) await db.prepare('INSERT INTO tool_worker_attempts (job_id,job_attempt,attempt,worker_id,location,took_ms,state,at) VALUES (?,?,?,?,?,?,?,?) ON CONFLICT (job_id,job_attempt,attempt) DO UPDATE SET worker_id=excluded.worker_id,location=excluded.location,took_ms=excluded.took_ms,state=excluded.state,at=excluded.at').run(id, jobAttempt, i, a.worker, a.location, a.took_ms, a.state, Date.now());++ },++ async insert(row) { await q.insert.run({ app, retry_of: null, ip_key: null, tool: null, project_id: null, trace_id: null, execution_prefer: 'auto', ...row }); },byResultMedia: async (owner, mediaId) => q.byResultMedia.all({ app, owner, like: `%"media_id":"${String(mediaId).replace(/[%"\\]/g, '')}"%` }),markRetried: async (id, next, now = Date.now()) => (await q.markRetried.run({ app, id, next, now })).changes === 1,get: async (id) => (await q.get.get(id, app)) || null,@@@ -244,12 +244,13 @@ function createJobSystem(o) {// Developer-app sandbox jobs: fewer at a time, kept briefly, results never leave this server.const SANDBOX_MAX_ACTIVE = Math.max(1, o.sandboxMaxActivePerOwner || 2);const SANDBOX_TTL_MS = 30 * 60 * 1000;-- async function submit({ owner, type, input = {}, files = [], idempotencyKey = null, ttlMs = HOUR, env = 'production', ipKey = null, tool = null, project = null, traceId = null }) {++ async function submit({ owner, type, input = {}, files = [], idempotencyKey = null, ttlMs = HOUR, env = 'production', ipKey = null, tool = null, project = null, traceId = null, prefer = 'auto' }) {if (env !== 'sandbox') env = 'production';if (env === 'sandbox') ttlMs = Math.min(ttlMs, SANDBOX_TTL_MS);if (stopped) throw new JobError(503, 'tools.job.unavailable', 'The job system is shutting down');const discard = () => Promise.all(files.map(f => (f.path ? fsp.unlink(f.path).catch(() => {}) : null)));try {++ if (!['auto', 'home', 'host'].includes(prefer)) throw new JobError(400, 'tools.job.invalid', 'prefer must be auto, home or host');if (!owner) throw new JobError(401, 'tools.job.no_owner', 'No owner for this job');const def = types.get(type);if (!def) throw new JobError(400, 'tools.job.unknown_type', `Unknown job type "${type}". Known: ${[...types.keys()].join(', ')}`);@@@ -264,7 +265,7 @@ function createJobSystem(o) {// Hash the request so a reused key with a different request is caught.const fileHashes = [];for (const f of files) fileHashes.push(f.buffer ? crypto.createHash('sha256').update(f.buffer).digest('hex') : await sha256File(f.path));-- const requestHash = crypto.createHash('sha256').update(canonical({ type, input, files: fileHashes })).digest('hex');++ const requestHash = crypto.createHash('sha256').update(canonical({ type, input, files: fileHashes, ...(prefer !== 'auto' && { prefer }) })).digest('hex');if (idempotencyKey != null) {const prior = await store.byIdempotencyKey(owner, String(idempotencyKey));@@@ -290,7 +291,7 @@ function createJobSystem(o) {try {await store.tx(async (t) => {await store.insert({-- id, type, type_version: def.version, owner, input_json: JSON.stringify(input), files_json: JSON.stringify(stored),++ id, type, type_version: def.version, owner, execution_prefer: prefer, input_json: JSON.stringify(input), files_json: JSON.stringify(stored),idempotency_key: idempotencyKey == null ? null : String(idempotencyKey), request_hash: requestHash,max_attempts: def.maxAttempts, ttl_ms: ttlMs, now: Date.now(), env,ip_key: String(owner).startsWith('session:') && ipKey ? String(ipKey) : null,@@@ -367,6 +368,7 @@ function createJobSystem(o) {const def = types.get(row.type);const outDir = path.join(jobDir(row.id), 'out');let outcome;++ const execution = { worker: 'host', location: 'host', took_ms: 0, attempts: [] };const timer = setTimeout(() => { if (!entry.reason) { entry.reason = 'timeout'; entry.ctrl.abort(new Error('timeout')); } }, def.timeoutMs);if (timer.unref) timer.unref();try {@@@ -375,12 +377,12 @@ function createJobSystem(o) {// would not see an abort that fired before it registered its listener.if (entry.reason) throw new Error(entry.reason);const files = store.parse(row.files_json, []).map(f => ({ name: f.name, mime: f.mime, size: f.size, sha256: f.sha256, path: f._path }));-- const out = await def.run({++ const out = await require('./placement').executePlaced({ row, def, receipt: execution, loadAttempts: () => store.execution(row.id, Number(row.attempts)), context: {job: { id: row.id, type: row.type, attempt: Number(row.attempts), owner: row.owner },input: store.parse(row.input_json, {}), files, outDir,signal: entry.ctrl.signal,progress: progressFn(row.id, entry),-- });++ } });if (stopped) return;if (entry.reason) throw new Error(entry.reason);const result = await storeResult(row, out || {}, entry);@@@ -403,8 +405,15 @@ function createJobSystem(o) {}let recorded = false;try {++ if (!execution.attempts.length) execution.attempts = await store.execution(row.id, Number(row.attempts));++ if (execution.attempts.length) {++ for (const a of execution.attempts) if (a.state === 'running') a.state = outcome.state === 'cancelled' ? 'cancelled' : 'failed';++ const last = execution.attempts.at(-1);++ Object.assign(execution, { worker: last.worker, location: last.location, took_ms: execution.attempts.reduce((n, a) => n + a.took_ms, 0) });++ }++ if (outcome?.error) outcome.error.execution = execution;// The end state and its tools.job.succeeded|failed event commit together (or neither does).-- await store.tx(async (t) => { if (await store.finish(row.id, outcome)) await ended(t, row.id); });++ await store.tx(async (t) => { if (await store.finish(row.id, outcome)) { await store.recordExecution(row.id, Number(row.attempts), execution.attempts); await ended(t, row.id); } });recorded = true;await emit(row.id, EVENT_FOR[outcome.state]);} catch (err) {@@@ -529,7 +538,7 @@ function createJobSystem(o) {await store.insert({id: next, type: row.type, type_version: def.version, owner: row.owner, input_json: row.input_json, files_json: JSON.stringify(stored),idempotency_key: null, request_hash: row.request_hash, max_attempts: def.maxAttempts, ttl_ms: row.ttl_ms, now: Date.now(),-- env: row.env, retry_of: row.id, ip_key: row.ip_key || null, tool: row.tool || null, project_id: row.project_id || null,++ env: row.env, execution_prefer: row.execution_prefer, retry_of: row.id, ip_key: row.ip_key || null, tool: row.tool || null, project_id: row.project_id || null,});await announce.created(t, await store.get(next));});@@@ -38,7 +38,7 @@ function openCors(res) {// along); first-party pages keep the app's own credentialed CORS.const RUN_OR_JOBS_RE = /^\/api\/v1\/(?:tools\/[a-z][a-z0-9-]{0,39}\/run|jobs(?:\/.*)?)\/?$/;const isPublicApiPath = (p) => RUN_OR_JOBS_RE.test(String(p || ''));--const API_HEADERS = 'Authorization, Content-Type, Idempotency-Key, Last-Event-ID, Cache-Control, traceparent, X-OpenVibe-Request-Id';++const API_HEADERS = 'X-Tools-Prefer, Authorization, Content-Type, Idempotency-Key, Last-Event-ID, Cache-Control, traceparent, X-OpenVibe-Request-Id';const API_EXPOSE = 'Location, Idempotent-Replayed, Retry-After, RateLimit-Limit, RateLimit-Remaining, RateLimit-Reset, X-OpenVibe-Request-Id, traceparent, Content-Disposition, Deprecation, Sunset, Link';/**@@@ -410,6 +410,7 @@ function createRunApi(o) {let row, replayed;try {const out = await system.submit({++ prefer: req.headers['x-tools-prefer'] || 'auto',owner, type: d.run.job.type, input: contracts.tools.jobInput(d, input),files: files.map(f => ({ path: f.path, buffer: f.buffer, name: f.originalname || f.name, mime: f.mimetype || f.mime, size: f.size })),idempotencyKey: key, ttlMs: who.kind === 'session' ? 60 * 60 * 1000 : 24 * 60 * 60 * 1000,@@@ -440,6 +441,7 @@ function createRunApi(o) {if (!m) return next();if (req.method === 'OPTIONS') return next(); // the app's CORS answers preflightsif (req.method !== 'POST') { res.setHeader('Allow', 'POST'); return problem(req, res, 405, 'method_not_allowed', 'POST a tools.run-request@1 to run a tool.'); }++ if (req.headers['x-tools-prefer'] && !['auto', 'home', 'host'].includes(req.headers['x-tools-prefer'])) return problem(req, res, 400, 'tools.run.invalid', 'X-Tools-Prefer must be auto, home or host.');const t0 = Date.now();counts.runs++;const snap = o.snapshot();@@@ -67,7 +67,22 @@ let auth = null; // created below; the key check reads it at request timeconst toolsDb = require('../../_shared/db').openToolsDb({ createDb: require('openvibe-sdk/db').createDb, service: 'tools-gateway' });const workerStore = require('./workers/store').createStore(toolsDb.db);const workerHub = require('./workers/hub').createHub({ WebSocketServer: require('ws').WebSocketServer, store: workerStore });--app.get('/api/v1/workers/status', (_req, res) => res.json({ protocol: 1, dry_run: true, transport: 'outbound-websocket', accepts: ['img.process', 'docs.process', 'audio.process'] }));++const workerRouter = require('./workers/router').createRouter({ hub: workerHub, store: workerStore, db: toolsDb.db });++app.post('/api/internal/workers/dispatch', (req, res, next) => {++ if (!require('../../_shared/internal-token').loopbackOnly(req)) return res.sendStatus(404);++ next();++}, express.json({ limit: '1kb' }), async (req, res) => {++ if (!/^job_[0-9A-HJKMNP-TV-Z]{26}$/.test(req.body?.job_id || '')) return res.sendStatus(400);++ const ctrl = new AbortController();++ res.on('close', () => { if (!res.writableFinished) ctrl.abort(); });++ try { res.json(await workerRouter.route(req.body.job_id, ctrl.signal)); }++ catch { if (!res.headersSent) res.status(503).json({ error: 'Worker placement unavailable' }); }++});++++app.get('/api/v1/workers/status', async (_req, res) => {++ try { res.json({ protocol: 1, dry_run: true, transport: 'outbound-websocket', accepts: ['img.process', 'docs.process', 'audio.process'], routing_enabled: !!(await workerStore.routing())?.enabled, strategy: 'owner-order/free-slots/recent-latency', fallback: 'host' }); }++ catch { res.status(503).json({ error: 'Worker registry unavailable' }); }++});const toolsValkey = require('../../_shared/db').openToolsValkey({ createValkey: require('openvibe-sdk/valkey').createValkey });// The account export and deletion (ADR-033, ./account-data.js): what Tools holds about a person, and the erasure// of it. The network sender posts Tools' part of an export back to Network with Tools' own token.@@@ -316,6 +331,14 @@ 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' }); }@@@ -18,15 +18,16 @@ function createHub({ WebSocketServer, store, now = Date.now }) {wss.handleUpgrade(req, socket, head, ws => {const old = connections.get(id);if (old) old.ws.terminate();-- const c = { id, location: worker.location, draining: worker.draining, ws, last: now(), pending: new Map(), capabilities: [], slots: 0, free: 0, load: 0, latency_ms: null };++ const c = { id, location: worker.location, draining: worker.draining, ws, last: now(), pending: new Map(), capabilities: [], tools: [], slots: 0, free: 0, load: 0, latency_ms: null };connections.set(id, c);ws.on('error', () => {});ws.on('message', raw => {let m;try { m = JSON.parse(String(raw)); } catch { ws.close(1008); return; }if (m.type === 'status') {-- if (!Array.isArray(m.capabilities) || m.capabilities.some(x => !TYPES.includes(x)) || !Number.isInteger(m.slots) || m.slots < 1 || m.slots > 8 || !Number.isInteger(m.free) || m.free < 0 || m.free > m.slots || !Number.isFinite(m.load) || m.load < 0) { ws.close(1008); return; }-- Object.assign(c, { capabilities: m.dry ? [] : [...new Set(m.capabilities)], slots: m.slots, free: m.dry ? 0 : m.free, load: m.load, last: now() });++ if (m.tools === undefined) m.tools = []; // Older protocol-1 workers stay connected but receive no jobs.++ if (!Array.isArray(m.tools) || m.tools.length > 128 || m.tools.some(t => typeof t !== 'string' || !/^(img|docs|audio)\.process:[a-z0-9-]{1,40}$/.test(t)) || !Array.isArray(m.capabilities) || m.capabilities.some(x => !TYPES.includes(x)) || !Number.isInteger(m.slots) || m.slots < 1 || m.slots > 8 || !Number.isInteger(m.free) || m.free < 0 || m.free > m.slots || !Number.isFinite(m.load) || m.load < 0) { ws.close(1008); return; }++ Object.assign(c, { tools: m.dry ? [] : m.tools, capabilities: m.dry ? [] : [...new Set(m.capabilities)], slots: m.slots, free: m.dry ? 0 : m.free, load: m.load, last: now() });ws.send(JSON.stringify({ type: 'heartbeat' })); // Also keeps the reverse proxy's upstream read deadline alive.} else if (m.type === 'result') {const p = c.pending.get(m.id);nnew file mode 100644@@@ -0,0 +1,83 @@++'use strict';++const fs = require('node:fs/promises');++const path = require('node:path');++const { MAX_BYTES, TYPES, bytes } = require('../../../worker/server/protocol');++const { detect } = require('../../../_shared/guard/sniff');++const free = c => Math.min(c.free, c.slots - c.pending.size);++function choose(connections, order, type, tool, excluded = new Set()) {++ const eligible = [...connections.values()].filter(c => !excluded.has(c.id) && !c.draining && free(c) > 0 && c.capabilities.includes(type) && c.tools?.includes(`${type}:${tool}`));++ for (const key of [...order, 'host']) {++ const candidates = eligible.filter(c => c.id === key || c.location === key);++ candidates.sort((a, b) => free(b) - free(a) || (a.latency_ms ?? Infinity) - (b.latency_ms ?? Infinity) || a.id.localeCompare(b.id));++ if (candidates.length) return candidates[0];++ if (key === 'host') return null;++ }++ return null;++}++function createRouter({ hub, store, db, now = Date.now }) {++ const active = new Set();++ async function route(id, signal) {++ const host = attempts => ({ mode: 'host', attempts });++ const config = await store.routing();++ if (!config?.enabled) return host([]);++ if (active.has(id)) throw new Error('job already dispatched');++ const row = await db.prepare('SELECT id,type,state,input_json,files_json,execution_prefer,attempts FROM tool_jobs WHERE id=?').get(id);++ if (!row || row.state !== 'running' || !TYPES.includes(row.type) || row.execution_prefer === 'host') return host([]);++ const input = JSON.parse(row.input_json), source = JSON.parse(row.files_json);++ let order; try { order = JSON.parse(config.worker_order); } catch { return host([]); }++ if (!Array.isArray(order) || !order.length) return host([]);++ const tool = String(input.tool || 'convert');++ if (!choose(hub.connections, order, row.type, tool)) return host([]);++ active.add(id);++ try {++ const files = [];++ let total = 0;++ for (const f of source) {++ const stat = await fs.stat(f._path); total += stat.size;++ if (total > MAX_BYTES) return host([]);++ const b = await fs.readFile(f._path);++ const kind = detect(b);++ if (!kind || kind.mime === 'image/svg+xml') return host([]);++ files.push({ name: path.basename(f.name), mime: kind.mime, bytes: b.toString('base64') });++ }++ const excluded = new Set(), attempts = [];++ // One retry after disconnect, including a host retry when no other worker is available.++ for (let n = 0; n < 2; n++) {++ if (signal.aborted) throw new Error('cancelled');++ const worker = choose(hub.connections, order, row.type, tool, excluded);++ if (!worker) return host(attempts);++ excluded.add(worker.id);++ const started = now();++ const attempt = { worker: worker.id, location: worker.location, took_ms: 0, state: 'failed' };++ attempts.push(attempt);++ // Commit the assignment before sending bytes so cancellation/restart receipts retain its location.++ await db.prepare('INSERT INTO tool_worker_attempts (job_id,job_attempt,attempt,worker_id,location,took_ms,state,at) VALUES (?,?,?,?,?,0,?,?) ON CONFLICT DO NOTHING').run(id, Number(row.attempts), n, worker.id, worker.location, 'running', started);++ try {++ if (signal.aborted) throw new Error('cancelled');++ const reply = await hub.dispatch(worker, { type: row.type, input, files }, signal);++ attempt.took_ms = now() - started;++ if (!reply.ok) return reply.error === 'worker.transfer_limit' && n === 0 ? host(attempts) : { mode: 'failed', attempts };++ const result = reply.result;++ if (!result || !Array.isArray(result.files) || result.files.length > 50 || !result.data || typeof result.data !== 'object' || Array.isArray(result.data)) throw new Error('invalid worker result');++ let size = 0;++ for (const f of result.files) {++ size += bytes(f).length;++ if (size > MAX_BYTES || typeof f.name !== 'string' || f.name.length > 200 || typeof f.mime !== 'string' || f.mime.length > 100) throw new Error('invalid worker output');++ f.name = path.basename(f.name);++ }++ attempt.state = 'succeeded';++ return { mode: 'remote', result, attempts };++ } catch (err) {++ attempt.took_ms = now() - started;++ attempt.state = signal.aborted ? 'cancelled' : err.code === 'worker.disconnected' ? 'disconnected' : 'failed';++ if (signal.aborted || err.code !== 'worker.disconnected') return { mode: 'failed', attempts };++ } finally {++ await db.prepare('UPDATE tool_worker_attempts SET took_ms=?,state=?,at=? WHERE job_id=? AND job_attempt=? AND attempt=? AND state=?').run(attempt.took_ms, attempt.state, now(), id, Number(row.attempts), n, 'running');++ }++ }++ return { mode: 'failed', attempts };++ } finally { active.delete(id); }++ }++ return { route };++}++module.exports = { choose, createRouter, free };@@@ -10,6 +10,13 @@ function createStore(db) {await db.prepare('INSERT INTO tools_workers (id,label,location,token_hash,created_at) VALUES (?,?,?,?,?)').run(id, label, location, hash(token), Date.now());return { id, token }; // One response only; listing never selects token_hash.},++ async setRouting({ enabled, order }) {++ if (typeof enabled !== 'boolean' || !Array.isArray(order) || order.length > 128 || !order.length || new Set(order).size !== order.length || order.some(k => typeof k !== 'string' || !/^(home|host|wrk_[a-f0-9]{24})$/.test(k))) throw new Error('invalid routing');++ for (const id of order.filter(k => k.startsWith('wrk_'))) if (!await db.prepare('SELECT id FROM tools_workers WHERE id=? AND revoked=FALSE').get(id)) throw new Error('unknown worker');++ if (!order.includes('host')) order = [...order, 'host'];++ await db.prepare('UPDATE tools_worker_routing SET enabled=?,worker_order=? WHERE id=1').run(enabled, JSON.stringify(order));++ return { enabled, order };++ },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(),@@@ -76,6 +76,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','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',@@@ -6,4 +6,6 @@ A single Node 22 process reuses the image, PDF and audio satellite job handlersOnly img.process, docs.process and audio.process are supported. Each job is capped at 16 MiB of input and output; larger jobs remain on the host. Input types are sniffed again on the worker. SVG and URL-fetching jobs are refused. Tool options are validated by the same satellite definitions. The gateway retains job ownership, quotas and result storage. No caller token crosses the worker channel.--Worker routing is default-off. This first part installs the channel and execution runtime; routing and owner controls follow separately.++Worker routing is default-off. The owner can configure it with GET/PUT `/api/v1/owner/worker-routing` (`{ enabled, order: ["home", "host"] }`); writes require the same-origin owner session. Order entries can also name a worker id. Within a group the gateway uses free slots, then recent latency. A disconnected job gets one attempt elsewhere, including the host when no second worker is available.++++Run requests accept `X-Tools-Prefer: auto|home|host`. `host` keeps the work on the host; `home` is advisory and respects the owner's order. Completed jobs include `result.data.execution` with worker id, location, duration and attempts. Failed/cancelled jobs carry the same receipt in `error.execution`. Inputs over 16 MiB, SVG, unavailable tools and all fetch-based jobs stay on the host. Workers advertise individual operations, so missing ffmpeg/qpdf/poppler remove affected capabilities.@@@ -33,7 +33,7 @@ async function main() {try {const result = await runtime.run({ id: m.id, type: m.jobType, input: m.input, files: m.files });if (ws === connection) send({ type: 'result', id: m.id, ok: true, result });-- } catch { if (ws === connection) send({ type: 'result', id: m.id, ok: false, error: 'worker job failed' }); }++ } catch (err) { if (ws === connection) send({ type: 'result', id: m.id, ok: false, error: err.code === 'worker.transfer_limit' ? 'worker.transfer_limit' : 'worker job failed' }); }status();});ws.on('error', () => {}); // Never log a credential or server-supplied error.@@@ -7,21 +7,25 @@ const { MAX_BYTES, TYPES, bytes } = require('./protocol');const { detect } = require('../../_shared/guard/sniff');const { poolFromEnv } = require('../../_shared/jobs/pool');function definitions() {-- const defs = new Map(), pools = [];++ const defs = new Map(), pools = [], tools = [];++ const hasAudio = ['ffmpeg', 'ffprobe'].every(bin => require('node:child_process').spawnSync(bin, ['-version'], { stdio: 'ignore', timeout: 3000 }).status === 0);for (const app of ['img', 'docs', 'audio']) {try {++ if (app === 'audio' && !hasAudio) continue;++ const catalog = require(`../../${app}/server/tools`);++ tools.push(...catalog.listTools().filter(t => t.available !== false).map(t => `${app}.process:${t.id}`));const process = require(`../../${app}/server/process`);const pool = app === 'audio' ? null : poolFromEnv(app, path.resolve(__dirname, `../../${app}/server/worker.js`), { size: 1, maxQueue: 0 });if (pool) pools.push(pool);process.defineJobs({ define: def => defs.set(def.type, def), JobError: Error }, pool);} catch { /* A missing dependency removes this capability, rather than accepting work it cannot run. */ }}-- return { defs, close: () => Promise.all(pools.map(p => p.close())) };++ return { defs, tools, close: () => Promise.all(pools.map(p => p.close())) };}function createRuntime({ slots = 2, dry = false, registry = definitions(), root = os.tmpdir() } = {}) {if (!Number.isInteger(slots) || slots < 1 || slots > 8) throw new Error('slots must be 1..8');const active = new Map();-- function status() { return { capabilities: dry ? [] : [...registry.defs.keys()].filter(k => TYPES.includes(k)), slots, free: dry ? 0 : slots - active.size, load: os.loadavg()[0], dry }; }++ function status() { return { tools: dry ? [] : (registry.tools || []), capabilities: dry ? [] : [...registry.defs.keys()].filter(k => TYPES.includes(k)), slots, free: dry ? 0 : slots - active.size, load: os.loadavg()[0], dry }; }async function run(job) {const def = registry.defs.get(job.type);if (dry || !TYPES.includes(job.type) || !def || active.size >= slots || active.has(job.id) || !/^[a-f0-9]{32}$/.test(job.id || '')) throw new Error('job refused');@@@ -57,7 +61,7 @@ function createRuntime({ slots = 2, dry = false, registry = definitions(), roottotal = 0;for (const f of result.files || []) {const size = (await fs.stat(f.path)).size; total += size;-- if (total > MAX_BYTES) throw new Error('result exceeds worker transfer limit');++ if (total > MAX_BYTES) throw Object.assign(new Error('result exceeds worker transfer limit'), { code: 'worker.transfer_limit' });output.push({ name: path.basename(f.name), mime: f.mime, bytes: (await fs.readFile(f.path)).toString('base64') });}return { files: output, data: result.data || {} };nnew file mode 100644@@@ -0,0 +1,59 @@++'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 { choose, createRouter } = require('../../gateway/server/workers/router');++const { executePlaced } = require('../../_shared/jobs/placement');++const worker = (id, free = 2, latency_ms = 10) => ({ id, location: 'home', free, slots: 2, latency_ms, pending: new Map(), draining: false, capabilities: ['img.process'], tools: ['img.process:convert'] });++(async () => {++ const a = worker('a'), b = worker('b', 1, 1), connections = new Map([['a', a], ['b', b]]);++ assert.equal(choose(connections, ['home', 'host'], 'img.process', 'convert').id, 'a');++ a.pending.set('busy', {}); assert.equal(choose(connections, ['home', 'host'], 'img.process', 'convert').id, 'b');++ assert.equal(choose(connections, ['a', 'home', 'host'], 'img.process', 'convert').id, 'a');++ assert.equal(choose(connections, ['host', 'home'], 'img.process', 'convert'), null);++ a.draining = true; b.draining = true; assert.equal(choose(connections, ['home'], 'img.process', 'convert'), null);++ a.draining = false; b.draining = false; a.pending.clear();++ const dir = await fs.mkdtemp(path.join(os.tmpdir(), 'tools-routing-test-'));++ try {++ const png = await require('../../img/node_modules/sharp')({ create: { width: 1, height: 1, channels: 3, background: 'red' } }).png().toBuffer();++ const file = path.join(dir, 'input.png'); await fs.writeFile(file, png);++ const row = { id: 'test', type: 'img.process', state: 'running', input_json: '{"tool":"convert"}', files_json: JSON.stringify([{ name: 'input.png', _path: file }]), execution_prefer: 'auto', attempts: 1 };++ let enabled = false, seen = [];++ const hub = { connections, dispatch: async (w, job) => {++ seen.push(w.id); assert.deepEqual(Object.keys(job).sort(), ['files', 'input', 'type']);++ if (w.id === 'a') throw Object.assign(new Error('gone'), { code: 'worker.disconnected' });++ return { ok: true, result: { files: [], data: { converted: true } } };++ } };++ const router = createRouter({ hub, store: { routing: async () => ({ enabled, worker_order: '["home","host"]' }) }, db: { prepare: () => ({ get: async () => row, run: async () => {} }) } });++ const signal = new AbortController().signal;++ assert.equal((await router.route('test', signal)).mode, 'host'); assert.deepEqual(seen, []);++ enabled = true;++ const out = await router.route('test', signal);++ assert.equal(out.mode, 'remote'); assert.deepEqual(seen, ['a', 'b']);++ assert.deepEqual(out.attempts.map(a => a.state), ['disconnected', 'succeeded']);++ seen = []; connections.delete('b');++ const fallback = await router.route('test', signal); assert.equal(fallback.mode, 'host'); assert.equal(fallback.attempts.length, 1);++ connections.set('b', b); hub.dispatch = async w => { seen.push(w.id); throw Object.assign(new Error('gone'), { code: 'worker.disconnected' }); };++ seen = []; assert.equal((await router.route('test', signal)).mode, 'failed'); assert.equal(seen.length, 2);++ hub.dispatch = async () => ({ ok: false, error: 'worker.transfer_limit' });++ assert.equal((await router.route('test', signal)).mode, 'host', 'expanded audio/PDF outputs retry on the host');++ row.type = 'yt.download'; seen = []; assert.equal((await router.route('test', signal)).mode, 'host'); assert.equal(seen.length, 0);++ row.type = 'img.process'; row.execution_prefer = 'host'; assert.equal((await router.route('test', signal)).mode, 'host');++ const receipt = {};++ const result = await executePlaced({ row, def: { run: async () => ({ files: [], data: { local: true } }) }, context: { signal, outDir: dir }, receipt, request: async () => { throw new Error('must not dispatch preferred host'); } });++ assert.equal(result.data.execution.worker, 'host'); assert.equal(receipt.attempts[0].state, 'succeeded');++ assert(Number.isFinite(receipt.took_ms));++ row.execution_prefer = 'auto';++ let clock = 0; const retried = {};++ await executePlaced({ row, def: { run: async () => ({}) }, context: { signal, outDir: dir }, receipt: retried, now: () => ++clock, request: async () => ({ mode: 'host', attempts: [{ worker: 'a', location: 'home', took_ms: 30, state: 'disconnected' }] }) });++ assert.equal(retried.took_ms, 31); assert.equal(retried.attempts.length, 2);++ const recovered = {};++ await executePlaced({ row, def: { run: async () => ({}) }, context: { signal, outDir: dir }, receipt: recovered, request: async () => { throw new Error('gateway restart'); }, loadAttempts: async () => [{ worker: 'a', location: 'home', took_ms: 20, state: 'running' }] });++ assert.deepEqual(recovered.attempts.map(a => a.worker), ['a', 'host']);++ assert.equal(recovered.attempts[0].state, 'disconnected');++ const remote = await executePlaced({ row, def: {}, context: { signal, outDir: dir }, receipt: {}, request: async () => ({ mode: 'remote', attempts: [{ worker: 'b', location: 'home', took_ms: 12, state: 'succeeded' }], result: { files: [{ name: '../../safe.png', mime: 'image/png', bytes: png.toString('base64') }], data: {} } }) });++ assert.equal(remote.files[0].name, 'safe.png'); assert.equal(path.dirname(remote.files[0].path), dir); assert.equal(remote.data.execution.worker, 'b');++ console.log('routing: owner order, capacity reservation, latency, drain, default-off, one retry, host fallback, fetch exclusion and receipts passed');++ } finally { await fs.rm(dir, { recursive: true, force: true }); }++})().catch(e => { console.error(e); process.exitCode = 1; });@@@ -46,7 +46,7 @@ const { createHub } = require('../../gateway/server/workers/hub');const url = `ws://127.0.0.1:${server.address().port}/api/v1/workers/connect`;const ws = new WebSocket(url, { headers });await once(ws, 'open');-- ws.send(JSON.stringify({ type: 'status', capabilities: ['img.process'], slots: 1, free: 1, load: 0 }));++ ws.send(JSON.stringify({ type: 'status', tools: ['img.process:convert'], capabilities: ['img.process'], slots: 1, free: 1, load: 0 }));await new Promise(r => setTimeout(r, 30));assert.equal(hub.stats()[0].free, 1);const replay = new WebSocket(url, { headers });nnew file mode 100644@@@ -0,0 +1,14 @@++-- phase: expand++ALTER TABLE tool_jobs ADD COLUMN IF NOT EXISTS execution_prefer TEXT NOT NULL DEFAULT 'auto';++CREATE TABLE IF NOT EXISTS tool_worker_attempts (++ job_id TEXT NOT NULL REFERENCES tool_jobs(id) ON DELETE CASCADE,++ job_attempt INTEGER NOT NULL,++ attempt INTEGER NOT NULL,++ worker_id TEXT NOT NULL,++ location TEXT NOT NULL,++ took_ms BIGINT NOT NULL,++ state TEXT NOT NULL,++ at BIGINT NOT NULL,++ PRIMARY KEY (job_id, job_attempt, attempt)++);++CREATE INDEX IF NOT EXISTS tool_worker_attempts_day ON tool_worker_attempts(at, worker_id);@@@ -18,7 +18,8 @@"test:discovery": "node test/run.js gateway/test/home-jobs gateway/test/tool-opens gateway/test/discovery gateway/test/run-api _shared/test/job-usage gateway/test/perf-budget","test:worker": "node apps/worker/test/worker.test.js","test:suite": "node --unhandled-rejections=strict test/run.js",-- "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"},"license": "MIT","engines": {
@@ -174,7 +174,7 @@ function mountJobRoutes(app, o) {
@@ -174,7 +174,7 @@ function mountJobRoutes(app, o) {
if (!ok) { for (const f of files) if (f.path) fs.unlink(f.path, () => {}); return; }
if (!ok) { for (const f of files) if (f.path) fs.unlink(f.path, () => {}); return; }
}
}
try {
try {
- const { job, replayed } = await system.submit({ owner: w.owner, type: String(b.type || ''), input, files, idempotencyKey: key, ttlMs: ttlMs(req, w), env: w.env || 'production', ipKey: w.caller ? w.caller.ipKey : null, project: w.kind === 'principal' && w.claims && w.claims.actor_type === 'app' ? w.claims.project_id : null, traceId: req.ov && req.ov.traceId });
+ const { job, replayed } = await system.submit({ prefer: req.headers['x-tools-prefer'] || 'auto', owner: w.owner, type: String(b.type || ''), input, files, idempotencyKey: key, ttlMs: ttlMs(req, w), env: w.env || 'production', ipKey: w.caller ? w.caller.ipKey : null, project: w.kind === 'principal' && w.claims && w.claims.actor_type === 'app' ? w.claims.project_id : null, traceId: req.ov && req.ov.traceId });
res.set('Location', `/api/v1/jobs/${job.id}`);
res.set('Location', `/api/v1/jobs/${job.id}`);
if (replayed) res.set('Idempotent-Replayed', 'true');
if (replayed) res.set('Idempotent-Replayed', 'true');
return res.status(replayed ? 200 : 202).json(await system.view(job));
return res.status(replayed ? 200 : 202).json(await system.view(job));
new file mode 100644
new file mode 100644
@@ -0,0 +1,59 @@
@@ -0,0 +1,59 @@
+'use strict';
+const fs = require('node:fs/promises');
+const path = require('node:path');
+const { MAX_FRAME, bytes, TYPES } = require('../../worker/server/protocol');
+// This is a fixed loopback channel, not a URL supplied by a job or a person.
+async function requestPlacement(id, signal, fetchImpl = fetch) {
+ const port = Number(process.env.TOOLS_GATEWAY_INTERNAL_PORT || 4001);
+ if (!Number.isInteger(port) || port < 1 || port > 65535) throw new Error('invalid gateway port');
+ const r = await fetchImpl(`http://127.0.0.1:${port}/api/internal/workers/dispatch`, {
+ method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ job_id: id }), signal, redirect: 'error',
+ });
+ if (!r.ok) throw new Error('gateway unavailable');
+ let size = 0; const chunks = [];
+ for await (const chunk of r.body) { size += chunk.length; if (size > MAX_FRAME) throw new Error('placement response too large'); chunks.push(chunk); }
+ return JSON.parse(Buffer.concat(chunks).toString('utf8'));
+}
+async function executePlaced({ row, def, context, receipt, request = requestPlacement, loadAttempts = async () => [], now = Date.now }) {
+ let placement = { mode: 'host', attempts: [] };
+ if (TYPES.includes(row.type) && row.execution_prefer !== 'host') {
+ try { placement = await request(row.id, context.signal); }
+ catch {
+ placement.attempts = (await loadAttempts()).map(a => ({ ...a, state: a.state === 'running' ? (context.signal.aborted ? 'cancelled' : 'disconnected') : a.state }));
+ receipt.attempts = placement.attempts;
+ if (context.signal.aborted) throw new Error('cancelled');
+ if (placement.attempts.length >= 2) placement.mode = 'failed';
+ }
+ }
+ receipt.attempts = placement.attempts || [];
+ if (placement.mode === 'failed') {
+ const last = receipt.attempts.at(-1);
+ Object.assign(receipt, { worker: last.worker, location: last.location, took_ms: receipt.attempts.reduce((n, a) => n + a.took_ms, 0) });
+ throw new Error('The remote worker could not complete this job');
+ }
+ if (placement.mode === 'remote') {
+ const last = receipt.attempts.at(-1);
+ Object.assign(receipt, { worker: last.worker, location: last.location, took_ms: receipt.attempts.reduce((n, a) => n + a.took_ms, 0) });
+ const files = [];
+ for (const [i, f] of placement.result.files.entries()) {
+ const local = path.join(context.outDir, `remote-${i}`);
+ await fs.writeFile(local, bytes(f));
+ files.push({ path: local, name: path.basename(f.name), mime: f.mime });
+ }
+ return { files, data: { ...placement.result.data, execution: receipt } };
+ }
+ const started = now();
+ const attempt = { worker: 'host', location: 'host', took_ms: 0, state: 'failed' };
+ receipt.attempts.push(attempt);
+ Object.assign(receipt, { worker: 'host', location: 'host' });
+ try {
+ const out = await def.run(context);
+ attempt.state = 'succeeded';
+ return { ...out, data: { ...(out?.data || {}), execution: receipt } };
+ } finally {
+ attempt.took_ms = now() - started;
+ if (context.signal.aborted) attempt.state = 'cancelled';
+ receipt.took_ms = receipt.attempts.reduce((n, a) => n + a.took_ms, 0);
+ }
+}
+module.exports = { executePlaced, requestPlacement };
@@ -28,8 +28,8 @@ function createStore(db, { app } = {}) {
@@ -28,8 +28,8 @@ function createStore(db, { app } = {}) {
if (!db || typeof db.prepare !== 'function') throw new TypeError('createStore needs an openvibe-sdk/db handle');
if (!db || typeof db.prepare !== 'function') throw new TypeError('createStore needs an openvibe-sdk/db handle');
if (!app) throw new TypeError('createStore needs app');
if (!app) throw new TypeError('createStore needs app');
const q = {
const q = {
- insert: db.prepare(`INSERT INTO tool_jobs (id, app, type, type_version, owner, state, input_json, files_json, idempotency_key, request_hash, max_attempts, ttl_ms, created_at, updated_at, env, retry_of, ip_key, tool, project_id, trace_id)
- VALUES (@id, @app, @type, @type_version, @owner, 'queued', @input_json, @files_json, @idempotency_key, @request_hash, @max_attempts, @ttl_ms, @now, @now, @env, @retry_of, @ip_key, @tool, @project_id, @trace_id)`),
+ insert: db.prepare(`INSERT INTO tool_jobs (id, app, type, type_version, owner, state, input_json, files_json, idempotency_key, request_hash, max_attempts, ttl_ms, created_at, updated_at, env, retry_of, ip_key, tool, project_id, trace_id, execution_prefer)
+ VALUES (@id, @app, @type, @type_version, @owner, 'queued', @input_json, @files_json, @idempotency_key, @request_hash, @max_attempts, @ttl_ms, @now, @now, @env, @retry_of, @ip_key, @tool, @project_id, @trace_id, @execution_prefer)`),
markRetried: db.prepare("UPDATE tool_jobs SET retried_by = @next, updated_at = @now WHERE id = @id AND app = @app AND state = 'failed' AND retried_by IS NULL"),
markRetried: db.prepare("UPDATE tool_jobs SET retried_by = @next, updated_at = @now WHERE id = @id AND app = @app AND state = 'failed' AND retried_by IS NULL"),
get: db.prepare('SELECT * FROM tool_jobs WHERE id = ? AND app = ?'),
get: db.prepare('SELECT * FROM tool_jobs WHERE id = ? AND app = ?'),
byIdem: db.prepare('SELECT * FROM tool_jobs WHERE app = @app AND owner = @owner AND idempotency_key = @idempotency_key'),
byIdem: db.prepare('SELECT * FROM tool_jobs WHERE app = @app AND owner = @owner AND idempotency_key = @idempotency_key'),
@@ -82,7 +82,14 @@ function createStore(db, { app } = {}) {
@@ -82,7 +82,14 @@ function createStore(db, { app } = {}) {
db,
db,
app,
app,
TERMINAL, STATES,
TERMINAL, STATES,
- async insert(row) { await q.insert.run({ app, retry_of: null, ip_key: null, tool: null, project_id: null, trace_id: null, ...row }); },
+ async execution(id, jobAttempt) {
+ const rows = await db.prepare('SELECT worker_id,location,took_ms,state,at FROM tool_worker_attempts WHERE job_id=? AND job_attempt=? ORDER BY attempt').all(id, jobAttempt);
+ return rows.map(a => ({ worker: a.worker_id, location: a.location, took_ms: a.state === 'running' ? Math.max(0, Date.now() - Number(a.at)) : Number(a.took_ms), state: a.state }));
+ },
+ async recordExecution(id, jobAttempt, attempts) {
+ for (const [i, a] of attempts.entries()) await db.prepare('INSERT INTO tool_worker_attempts (job_id,job_attempt,attempt,worker_id,location,took_ms,state,at) VALUES (?,?,?,?,?,?,?,?) ON CONFLICT (job_id,job_attempt,attempt) DO UPDATE SET worker_id=excluded.worker_id,location=excluded.location,took_ms=excluded.took_ms,state=excluded.state,at=excluded.at').run(id, jobAttempt, i, a.worker, a.location, a.took_ms, a.state, Date.now());
+ },
+ async insert(row) { await q.insert.run({ app, retry_of: null, ip_key: null, tool: null, project_id: null, trace_id: null, execution_prefer: 'auto', ...row }); },
byResultMedia: async (owner, mediaId) => q.byResultMedia.all({ app, owner, like: `%"media_id":"${String(mediaId).replace(/[%"\\]/g, '')}"%` }),
byResultMedia: async (owner, mediaId) => q.byResultMedia.all({ app, owner, like: `%"media_id":"${String(mediaId).replace(/[%"\\]/g, '')}"%` }),
markRetried: async (id, next, now = Date.now()) => (await q.markRetried.run({ app, id, next, now })).changes === 1,
markRetried: async (id, next, now = Date.now()) => (await q.markRetried.run({ app, id, next, now })).changes === 1,
get: async (id) => (await q.get.get(id, app)) || null,
get: async (id) => (await q.get.get(id, app)) || null,
@@ -244,12 +244,13 @@ function createJobSystem(o) {
@@ -244,12 +244,13 @@ function createJobSystem(o) {
// Developer-app sandbox jobs: fewer at a time, kept briefly, results never leave this server.
// Developer-app sandbox jobs: fewer at a time, kept briefly, results never leave this server.
const SANDBOX_MAX_ACTIVE = Math.max(1, o.sandboxMaxActivePerOwner || 2);
const SANDBOX_MAX_ACTIVE = Math.max(1, o.sandboxMaxActivePerOwner || 2);
const SANDBOX_TTL_MS = 30 * 60 * 1000;
const SANDBOX_TTL_MS = 30 * 60 * 1000;
- async function submit({ owner, type, input = {}, files = [], idempotencyKey = null, ttlMs = HOUR, env = 'production', ipKey = null, tool = null, project = null, traceId = null }) {
+ async function submit({ owner, type, input = {}, files = [], idempotencyKey = null, ttlMs = HOUR, env = 'production', ipKey = null, tool = null, project = null, traceId = null, prefer = 'auto' }) {
if (env !== 'sandbox') env = 'production';
if (env !== 'sandbox') env = 'production';
if (env === 'sandbox') ttlMs = Math.min(ttlMs, SANDBOX_TTL_MS);
if (env === 'sandbox') ttlMs = Math.min(ttlMs, SANDBOX_TTL_MS);
if (stopped) throw new JobError(503, 'tools.job.unavailable', 'The job system is shutting down');
if (stopped) throw new JobError(503, 'tools.job.unavailable', 'The job system is shutting down');
const discard = () => Promise.all(files.map(f => (f.path ? fsp.unlink(f.path).catch(() => {}) : null)));
const discard = () => Promise.all(files.map(f => (f.path ? fsp.unlink(f.path).catch(() => {}) : null)));
try {
try {
+ if (!['auto', 'home', 'host'].includes(prefer)) throw new JobError(400, 'tools.job.invalid', 'prefer must be auto, home or host');
if (!owner) throw new JobError(401, 'tools.job.no_owner', 'No owner for this job');
if (!owner) throw new JobError(401, 'tools.job.no_owner', 'No owner for this job');
const def = types.get(type);
const def = types.get(type);
if (!def) throw new JobError(400, 'tools.job.unknown_type', `Unknown job type "${type}". Known: ${[...types.keys()].join(', ')}`);
if (!def) throw new JobError(400, 'tools.job.unknown_type', `Unknown job type "${type}". Known: ${[...types.keys()].join(', ')}`);
@@ -264,7 +265,7 @@ function createJobSystem(o) {
@@ -264,7 +265,7 @@ function createJobSystem(o) {
// Hash the request so a reused key with a different request is caught.
// Hash the request so a reused key with a different request is caught.
const fileHashes = [];
const fileHashes = [];
for (const f of files) fileHashes.push(f.buffer ? crypto.createHash('sha256').update(f.buffer).digest('hex') : await sha256File(f.path));
for (const f of files) fileHashes.push(f.buffer ? crypto.createHash('sha256').update(f.buffer).digest('hex') : await sha256File(f.path));
- const requestHash = crypto.createHash('sha256').update(canonical({ type, input, files: fileHashes })).digest('hex');
+ const requestHash = crypto.createHash('sha256').update(canonical({ type, input, files: fileHashes, ...(prefer !== 'auto' && { prefer }) })).digest('hex');
if (idempotencyKey != null) {
if (idempotencyKey != null) {
const prior = await store.byIdempotencyKey(owner, String(idempotencyKey));
const prior = await store.byIdempotencyKey(owner, String(idempotencyKey));
@@ -290,7 +291,7 @@ function createJobSystem(o) {
@@ -290,7 +291,7 @@ function createJobSystem(o) {
try {
try {
await store.tx(async (t) => {
await store.tx(async (t) => {
await store.insert({
await store.insert({
- id, type, type_version: def.version, owner, input_json: JSON.stringify(input), files_json: JSON.stringify(stored),
+ id, type, type_version: def.version, owner, execution_prefer: prefer, input_json: JSON.stringify(input), files_json: JSON.stringify(stored),
idempotency_key: idempotencyKey == null ? null : String(idempotencyKey), request_hash: requestHash,
idempotency_key: idempotencyKey == null ? null : String(idempotencyKey), request_hash: requestHash,
max_attempts: def.maxAttempts, ttl_ms: ttlMs, now: Date.now(), env,
max_attempts: def.maxAttempts, ttl_ms: ttlMs, now: Date.now(), env,
ip_key: String(owner).startsWith('session:') && ipKey ? String(ipKey) : null,
ip_key: String(owner).startsWith('session:') && ipKey ? String(ipKey) : null,
@@ -367,6 +368,7 @@ function createJobSystem(o) {
@@ -367,6 +368,7 @@ function createJobSystem(o) {
const def = types.get(row.type);
const def = types.get(row.type);
const outDir = path.join(jobDir(row.id), 'out');
const outDir = path.join(jobDir(row.id), 'out');
let outcome;
let outcome;
+ const execution = { worker: 'host', location: 'host', took_ms: 0, attempts: [] };
const timer = setTimeout(() => { if (!entry.reason) { entry.reason = 'timeout'; entry.ctrl.abort(new Error('timeout')); } }, def.timeoutMs);
const timer = setTimeout(() => { if (!entry.reason) { entry.reason = 'timeout'; entry.ctrl.abort(new Error('timeout')); } }, def.timeoutMs);
if (timer.unref) timer.unref();
if (timer.unref) timer.unref();
try {
try {
@@ -375,12 +377,12 @@ function createJobSystem(o) {
@@ -375,12 +377,12 @@ function createJobSystem(o) {
// would not see an abort that fired before it registered its listener.
// would not see an abort that fired before it registered its listener.
if (entry.reason) throw new Error(entry.reason);
if (entry.reason) throw new Error(entry.reason);
const files = store.parse(row.files_json, []).map(f => ({ name: f.name, mime: f.mime, size: f.size, sha256: f.sha256, path: f._path }));
const files = store.parse(row.files_json, []).map(f => ({ name: f.name, mime: f.mime, size: f.size, sha256: f.sha256, path: f._path }));
- const out = await def.run({
+ const out = await require('./placement').executePlaced({ row, def, receipt: execution, loadAttempts: () => store.execution(row.id, Number(row.attempts)), context: {
job: { id: row.id, type: row.type, attempt: Number(row.attempts), owner: row.owner },
job: { id: row.id, type: row.type, attempt: Number(row.attempts), owner: row.owner },
input: store.parse(row.input_json, {}), files, outDir,
input: store.parse(row.input_json, {}), files, outDir,
signal: entry.ctrl.signal,
signal: entry.ctrl.signal,
progress: progressFn(row.id, entry),
progress: progressFn(row.id, entry),
- });
+ } });
if (stopped) return;
if (stopped) return;
if (entry.reason) throw new Error(entry.reason);
if (entry.reason) throw new Error(entry.reason);
const result = await storeResult(row, out || {}, entry);
const result = await storeResult(row, out || {}, entry);
@@ -403,8 +405,15 @@ function createJobSystem(o) {
@@ -403,8 +405,15 @@ function createJobSystem(o) {
}
}
let recorded = false;
let recorded = false;
try {
try {
+ if (!execution.attempts.length) execution.attempts = await store.execution(row.id, Number(row.attempts));
+ if (execution.attempts.length) {
+ for (const a of execution.attempts) if (a.state === 'running') a.state = outcome.state === 'cancelled' ? 'cancelled' : 'failed';
+ const last = execution.attempts.at(-1);
+ Object.assign(execution, { worker: last.worker, location: last.location, took_ms: execution.attempts.reduce((n, a) => n + a.took_ms, 0) });
+ }
+ if (outcome?.error) outcome.error.execution = execution;
// The end state and its tools.job.succeeded|failed event commit together (or neither does).
// The end state and its tools.job.succeeded|failed event commit together (or neither does).
- await store.tx(async (t) => { if (await store.finish(row.id, outcome)) await ended(t, row.id); });
+ await store.tx(async (t) => { if (await store.finish(row.id, outcome)) { await store.recordExecution(row.id, Number(row.attempts), execution.attempts); await ended(t, row.id); } });
recorded = true;
recorded = true;
await emit(row.id, EVENT_FOR[outcome.state]);
await emit(row.id, EVENT_FOR[outcome.state]);
} catch (err) {
} catch (err) {
@@ -529,7 +538,7 @@ function createJobSystem(o) {
@@ -529,7 +538,7 @@ function createJobSystem(o) {
await store.insert({
await store.insert({
id: next, type: row.type, type_version: def.version, owner: row.owner, input_json: row.input_json, files_json: JSON.stringify(stored),
id: next, type: row.type, type_version: def.version, owner: row.owner, input_json: row.input_json, files_json: JSON.stringify(stored),
idempotency_key: null, request_hash: row.request_hash, max_attempts: def.maxAttempts, ttl_ms: row.ttl_ms, now: Date.now(),
idempotency_key: null, request_hash: row.request_hash, max_attempts: def.maxAttempts, ttl_ms: row.ttl_ms, now: Date.now(),
- env: row.env, retry_of: row.id, ip_key: row.ip_key || null, tool: row.tool || null, project_id: row.project_id || null,
+ env: row.env, execution_prefer: row.execution_prefer, retry_of: row.id, ip_key: row.ip_key || null, tool: row.tool || null, project_id: row.project_id || null,
});
});
await announce.created(t, await store.get(next));
await announce.created(t, await store.get(next));
});
});
@@ -38,7 +38,7 @@ function openCors(res) {
@@ -38,7 +38,7 @@ function openCors(res) {
// along); first-party pages keep the app's own credentialed CORS.
// along); first-party pages keep the app's own credentialed CORS.
const RUN_OR_JOBS_RE = /^\/api\/v1\/(?:tools\/[a-z][a-z0-9-]{0,39}\/run|jobs(?:\/.*)?)\/?$/;
const RUN_OR_JOBS_RE = /^\/api\/v1\/(?:tools\/[a-z][a-z0-9-]{0,39}\/run|jobs(?:\/.*)?)\/?$/;
const isPublicApiPath = (p) => RUN_OR_JOBS_RE.test(String(p || ''));
const isPublicApiPath = (p) => RUN_OR_JOBS_RE.test(String(p || ''));
-const API_HEADERS = 'Authorization, Content-Type, Idempotency-Key, Last-Event-ID, Cache-Control, traceparent, X-OpenVibe-Request-Id';
+const API_HEADERS = 'X-Tools-Prefer, Authorization, Content-Type, Idempotency-Key, Last-Event-ID, Cache-Control, traceparent, X-OpenVibe-Request-Id';
const API_EXPOSE = 'Location, Idempotent-Replayed, Retry-After, RateLimit-Limit, RateLimit-Remaining, RateLimit-Reset, X-OpenVibe-Request-Id, traceparent, Content-Disposition, Deprecation, Sunset, Link';
const API_EXPOSE = 'Location, Idempotent-Replayed, Retry-After, RateLimit-Limit, RateLimit-Remaining, RateLimit-Reset, X-OpenVibe-Request-Id, traceparent, Content-Disposition, Deprecation, Sunset, Link';
/**
/**
@@ -410,6 +410,7 @@ function createRunApi(o) {
@@ -410,6 +410,7 @@ function createRunApi(o) {
let row, replayed;
let row, replayed;
try {
try {
const out = await system.submit({
const out = await system.submit({
+ prefer: req.headers['x-tools-prefer'] || 'auto',
owner, type: d.run.job.type, input: contracts.tools.jobInput(d, input),
owner, type: d.run.job.type, input: contracts.tools.jobInput(d, input),
files: files.map(f => ({ path: f.path, buffer: f.buffer, name: f.originalname || f.name, mime: f.mimetype || f.mime, size: f.size })),
files: files.map(f => ({ path: f.path, buffer: f.buffer, name: f.originalname || f.name, mime: f.mimetype || f.mime, size: f.size })),
idempotencyKey: key, ttlMs: who.kind === 'session' ? 60 * 60 * 1000 : 24 * 60 * 60 * 1000,
idempotencyKey: key, ttlMs: who.kind === 'session' ? 60 * 60 * 1000 : 24 * 60 * 60 * 1000,
@@ -440,6 +441,7 @@ function createRunApi(o) {
@@ -440,6 +441,7 @@ function createRunApi(o) {
if (!m) return next();
if (!m) return next();
if (req.method === 'OPTIONS') return next(); // the app's CORS answers preflights
if (req.method === 'OPTIONS') return next(); // the app's CORS answers preflights
if (req.method !== 'POST') { res.setHeader('Allow', 'POST'); return problem(req, res, 405, 'method_not_allowed', 'POST a tools.run-request@1 to run a tool.'); }
if (req.method !== 'POST') { res.setHeader('Allow', 'POST'); return problem(req, res, 405, 'method_not_allowed', 'POST a tools.run-request@1 to run a tool.'); }
+ if (req.headers['x-tools-prefer'] && !['auto', 'home', 'host'].includes(req.headers['x-tools-prefer'])) return problem(req, res, 400, 'tools.run.invalid', 'X-Tools-Prefer must be auto, home or host.');
const t0 = Date.now();
const t0 = Date.now();
counts.runs++;
counts.runs++;
const snap = o.snapshot();
const snap = o.snapshot();
@@ -67,7 +67,22 @@ let auth = null; // created below; the key check reads it at request time
@@ -67,7 +67,22 @@ let auth = null; // created below; the key check reads it at request time
const toolsDb = require('../../_shared/db').openToolsDb({ createDb: require('openvibe-sdk/db').createDb, service: 'tools-gateway' });
const toolsDb = require('../../_shared/db').openToolsDb({ createDb: require('openvibe-sdk/db').createDb, service: 'tools-gateway' });
const workerStore = require('./workers/store').createStore(toolsDb.db);
const workerStore = require('./workers/store').createStore(toolsDb.db);
const workerHub = require('./workers/hub').createHub({ WebSocketServer: require('ws').WebSocketServer, store: workerStore });
const workerHub = require('./workers/hub').createHub({ WebSocketServer: require('ws').WebSocketServer, store: workerStore });
-app.get('/api/v1/workers/status', (_req, res) => res.json({ protocol: 1, dry_run: true, transport: 'outbound-websocket', accepts: ['img.process', 'docs.process', 'audio.process'] }));
+const workerRouter = require('./workers/router').createRouter({ hub: workerHub, store: workerStore, db: toolsDb.db });
+app.post('/api/internal/workers/dispatch', (req, res, next) => {
+ if (!require('../../_shared/internal-token').loopbackOnly(req)) return res.sendStatus(404);
+ next();
+}, express.json({ limit: '1kb' }), async (req, res) => {
+ if (!/^job_[0-9A-HJKMNP-TV-Z]{26}$/.test(req.body?.job_id || '')) return res.sendStatus(400);
+ const ctrl = new AbortController();
+ res.on('close', () => { if (!res.writableFinished) ctrl.abort(); });
+ try { res.json(await workerRouter.route(req.body.job_id, ctrl.signal)); }
+ catch { if (!res.headersSent) res.status(503).json({ error: 'Worker placement unavailable' }); }
+});
+
+app.get('/api/v1/workers/status', async (_req, res) => {
+ try { res.json({ protocol: 1, dry_run: true, transport: 'outbound-websocket', accepts: ['img.process', 'docs.process', 'audio.process'], routing_enabled: !!(await workerStore.routing())?.enabled, strategy: 'owner-order/free-slots/recent-latency', fallback: 'host' }); }
+ catch { res.status(503).json({ error: 'Worker registry unavailable' }); }
+});
const toolsValkey = require('../../_shared/db').openToolsValkey({ createValkey: require('openvibe-sdk/valkey').createValkey });
const toolsValkey = require('../../_shared/db').openToolsValkey({ createValkey: require('openvibe-sdk/valkey').createValkey });
// The account export and deletion (ADR-033, ./account-data.js): what Tools holds about a person, and the erasure
// The account export and deletion (ADR-033, ./account-data.js): what Tools holds about a person, and the erasure
// of it. The network sender posts Tools' part of an export back to Network with Tools' own token.
// of it. The network sender posts Tools' part of an export back to Network with Tools' own token.
@@ -316,6 +331,14 @@ app.locals.config = config;
@@ -316,6 +331,14 @@ 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) => {
app.post('/api/v1/owner/workers', workerOwner, express.json({ limit: '4kb' }), async (req, res) => {
try { res.status(201).json(await workerStore.mint(req.body || {})); }
try { res.status(201).json(await workerStore.mint(req.body || {})); }
catch { res.status(400).json({ error: 'Could not create worker' }); }
catch { res.status(400).json({ error: 'Could not create worker' }); }
@@ -18,15 +18,16 @@ function createHub({ WebSocketServer, store, now = Date.now }) {
@@ -18,15 +18,16 @@ function createHub({ WebSocketServer, store, now = Date.now }) {
wss.handleUpgrade(req, socket, head, ws => {
wss.handleUpgrade(req, socket, head, ws => {
const old = connections.get(id);
const old = connections.get(id);
if (old) old.ws.terminate();
if (old) old.ws.terminate();
- const c = { id, location: worker.location, draining: worker.draining, ws, last: now(), pending: new Map(), capabilities: [], slots: 0, free: 0, load: 0, latency_ms: null };
+ const c = { id, location: worker.location, draining: worker.draining, ws, last: now(), pending: new Map(), capabilities: [], tools: [], slots: 0, free: 0, load: 0, latency_ms: null };
connections.set(id, c);
connections.set(id, c);
ws.on('error', () => {});
ws.on('error', () => {});
ws.on('message', raw => {
ws.on('message', raw => {
let m;
let m;
try { m = JSON.parse(String(raw)); } catch { ws.close(1008); return; }
try { m = JSON.parse(String(raw)); } catch { ws.close(1008); return; }
if (m.type === 'status') {
if (m.type === 'status') {
- if (!Array.isArray(m.capabilities) || m.capabilities.some(x => !TYPES.includes(x)) || !Number.isInteger(m.slots) || m.slots < 1 || m.slots > 8 || !Number.isInteger(m.free) || m.free < 0 || m.free > m.slots || !Number.isFinite(m.load) || m.load < 0) { ws.close(1008); return; }
- Object.assign(c, { capabilities: m.dry ? [] : [...new Set(m.capabilities)], slots: m.slots, free: m.dry ? 0 : m.free, load: m.load, last: now() });
+ if (m.tools === undefined) m.tools = []; // Older protocol-1 workers stay connected but receive no jobs.
+ if (!Array.isArray(m.tools) || m.tools.length > 128 || m.tools.some(t => typeof t !== 'string' || !/^(img|docs|audio)\.process:[a-z0-9-]{1,40}$/.test(t)) || !Array.isArray(m.capabilities) || m.capabilities.some(x => !TYPES.includes(x)) || !Number.isInteger(m.slots) || m.slots < 1 || m.slots > 8 || !Number.isInteger(m.free) || m.free < 0 || m.free > m.slots || !Number.isFinite(m.load) || m.load < 0) { ws.close(1008); return; }
+ Object.assign(c, { tools: m.dry ? [] : m.tools, capabilities: m.dry ? [] : [...new Set(m.capabilities)], slots: m.slots, free: m.dry ? 0 : m.free, load: m.load, last: now() });
ws.send(JSON.stringify({ type: 'heartbeat' })); // Also keeps the reverse proxy's upstream read deadline alive.
ws.send(JSON.stringify({ type: 'heartbeat' })); // Also keeps the reverse proxy's upstream read deadline alive.
} else if (m.type === 'result') {
} else if (m.type === 'result') {
const p = c.pending.get(m.id);
const p = c.pending.get(m.id);
new file mode 100644
new file mode 100644
@@ -0,0 +1,83 @@
@@ -0,0 +1,83 @@
+'use strict';
+const fs = require('node:fs/promises');
+const path = require('node:path');
+const { MAX_BYTES, TYPES, bytes } = require('../../../worker/server/protocol');
+const { detect } = require('../../../_shared/guard/sniff');
+const free = c => Math.min(c.free, c.slots - c.pending.size);
+function choose(connections, order, type, tool, excluded = new Set()) {
+ const eligible = [...connections.values()].filter(c => !excluded.has(c.id) && !c.draining && free(c) > 0 && c.capabilities.includes(type) && c.tools?.includes(`${type}:${tool}`));
+ for (const key of [...order, 'host']) {
+ const candidates = eligible.filter(c => c.id === key || c.location === key);
+ candidates.sort((a, b) => free(b) - free(a) || (a.latency_ms ?? Infinity) - (b.latency_ms ?? Infinity) || a.id.localeCompare(b.id));
+ if (candidates.length) return candidates[0];
+ if (key === 'host') return null;
+ }
+ return null;
+}
+function createRouter({ hub, store, db, now = Date.now }) {
+ const active = new Set();
+ async function route(id, signal) {
+ const host = attempts => ({ mode: 'host', attempts });
+ const config = await store.routing();
+ if (!config?.enabled) return host([]);
+ if (active.has(id)) throw new Error('job already dispatched');
+ const row = await db.prepare('SELECT id,type,state,input_json,files_json,execution_prefer,attempts FROM tool_jobs WHERE id=?').get(id);
+ if (!row || row.state !== 'running' || !TYPES.includes(row.type) || row.execution_prefer === 'host') return host([]);
+ const input = JSON.parse(row.input_json), source = JSON.parse(row.files_json);
+ let order; try { order = JSON.parse(config.worker_order); } catch { return host([]); }
+ if (!Array.isArray(order) || !order.length) return host([]);
+ const tool = String(input.tool || 'convert');
+ if (!choose(hub.connections, order, row.type, tool)) return host([]);
+ active.add(id);
+ try {
+ const files = [];
+ let total = 0;
+ for (const f of source) {
+ const stat = await fs.stat(f._path); total += stat.size;
+ if (total > MAX_BYTES) return host([]);
+ const b = await fs.readFile(f._path);
+ const kind = detect(b);
+ if (!kind || kind.mime === 'image/svg+xml') return host([]);
+ files.push({ name: path.basename(f.name), mime: kind.mime, bytes: b.toString('base64') });
+ }
+ const excluded = new Set(), attempts = [];
+ // One retry after disconnect, including a host retry when no other worker is available.
+ for (let n = 0; n < 2; n++) {
+ if (signal.aborted) throw new Error('cancelled');
+ const worker = choose(hub.connections, order, row.type, tool, excluded);
+ if (!worker) return host(attempts);
+ excluded.add(worker.id);
+ const started = now();
+ const attempt = { worker: worker.id, location: worker.location, took_ms: 0, state: 'failed' };
+ attempts.push(attempt);
+ // Commit the assignment before sending bytes so cancellation/restart receipts retain its location.
+ await db.prepare('INSERT INTO tool_worker_attempts (job_id,job_attempt,attempt,worker_id,location,took_ms,state,at) VALUES (?,?,?,?,?,0,?,?) ON CONFLICT DO NOTHING').run(id, Number(row.attempts), n, worker.id, worker.location, 'running', started);
+ try {
+ if (signal.aborted) throw new Error('cancelled');
+ const reply = await hub.dispatch(worker, { type: row.type, input, files }, signal);
+ attempt.took_ms = now() - started;
+ if (!reply.ok) return reply.error === 'worker.transfer_limit' && n === 0 ? host(attempts) : { mode: 'failed', attempts };
+ const result = reply.result;
+ if (!result || !Array.isArray(result.files) || result.files.length > 50 || !result.data || typeof result.data !== 'object' || Array.isArray(result.data)) throw new Error('invalid worker result');
+ let size = 0;
+ for (const f of result.files) {
+ size += bytes(f).length;
+ if (size > MAX_BYTES || typeof f.name !== 'string' || f.name.length > 200 || typeof f.mime !== 'string' || f.mime.length > 100) throw new Error('invalid worker output');
+ f.name = path.basename(f.name);
+ }
+ attempt.state = 'succeeded';
+ return { mode: 'remote', result, attempts };
+ } catch (err) {
+ attempt.took_ms = now() - started;
+ attempt.state = signal.aborted ? 'cancelled' : err.code === 'worker.disconnected' ? 'disconnected' : 'failed';
+ if (signal.aborted || err.code !== 'worker.disconnected') return { mode: 'failed', attempts };
+ } finally {
+ await db.prepare('UPDATE tool_worker_attempts SET took_ms=?,state=?,at=? WHERE job_id=? AND job_attempt=? AND attempt=? AND state=?').run(attempt.took_ms, attempt.state, now(), id, Number(row.attempts), n, 'running');
+ }
+ }
+ return { mode: 'failed', attempts };
+ } finally { active.delete(id); }
+ }
+ return { route };
+}
+module.exports = { choose, createRouter, free };
@@ -10,6 +10,13 @@ function createStore(db) {
@@ -10,6 +10,13 @@ function createStore(db) {
await db.prepare('INSERT INTO tools_workers (id,label,location,token_hash,created_at) VALUES (?,?,?,?,?)').run(id, label, location, hash(token), Date.now());
await db.prepare('INSERT INTO tools_workers (id,label,location,token_hash,created_at) VALUES (?,?,?,?,?)').run(id, label, location, hash(token), Date.now());
return { id, token }; // One response only; listing never selects token_hash.
return { id, token }; // One response only; listing never selects token_hash.
},
},
+ async setRouting({ enabled, order }) {
+ if (typeof enabled !== 'boolean' || !Array.isArray(order) || order.length > 128 || !order.length || new Set(order).size !== order.length || order.some(k => typeof k !== 'string' || !/^(home|host|wrk_[a-f0-9]{24})$/.test(k))) throw new Error('invalid routing');
+ for (const id of order.filter(k => k.startsWith('wrk_'))) if (!await db.prepare('SELECT id FROM tools_workers WHERE id=? AND revoked=FALSE').get(id)) throw new Error('unknown worker');
+ if (!order.includes('host')) order = [...order, 'host'];
+ await db.prepare('UPDATE tools_worker_routing SET enabled=?,worker_order=? WHERE id=1').run(enabled, JSON.stringify(order));
+ return { enabled, order };
+ },
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(),
@@ -76,6 +76,7 @@ function rawGet(port, p, headers = {}) {
@@ -76,6 +76,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',
'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',
@@ -6,4 +6,6 @@ A single Node 22 process reuses the image, PDF and audio satellite job handlers
@@ -6,4 +6,6 @@ A single Node 22 process reuses the image, PDF and audio satellite job handlers
Only img.process, docs.process and audio.process are supported. Each job is capped at 16 MiB of input and output; larger jobs remain on the host. Input types are sniffed again on the worker. SVG and URL-fetching jobs are refused. Tool options are validated by the same satellite definitions. The gateway retains job ownership, quotas and result storage. No caller token crosses the worker channel.
Only img.process, docs.process and audio.process are supported. Each job is capped at 16 MiB of input and output; larger jobs remain on the host. Input types are sniffed again on the worker. SVG and URL-fetching jobs are refused. Tool options are validated by the same satellite definitions. The gateway retains job ownership, quotas and result storage. No caller token crosses the worker channel.
-Worker routing is default-off. This first part installs the channel and execution runtime; routing and owner controls follow separately.
+Worker routing is default-off. The owner can configure it with GET/PUT `/api/v1/owner/worker-routing` (`{ enabled, order: ["home", "host"] }`); writes require the same-origin owner session. Order entries can also name a worker id. Within a group the gateway uses free slots, then recent latency. A disconnected job gets one attempt elsewhere, including the host when no second worker is available.
+
+Run requests accept `X-Tools-Prefer: auto|home|host`. `host` keeps the work on the host; `home` is advisory and respects the owner's order. Completed jobs include `result.data.execution` with worker id, location, duration and attempts. Failed/cancelled jobs carry the same receipt in `error.execution`. Inputs over 16 MiB, SVG, unavailable tools and all fetch-based jobs stay on the host. Workers advertise individual operations, so missing ffmpeg/qpdf/poppler remove affected capabilities.
@@ -33,7 +33,7 @@ async function main() {
@@ -33,7 +33,7 @@ async function main() {
try {
try {
const result = await runtime.run({ id: m.id, type: m.jobType, input: m.input, files: m.files });
const result = await runtime.run({ id: m.id, type: m.jobType, input: m.input, files: m.files });
if (ws === connection) send({ type: 'result', id: m.id, ok: true, result });
if (ws === connection) send({ type: 'result', id: m.id, ok: true, result });
- } catch { if (ws === connection) send({ type: 'result', id: m.id, ok: false, error: 'worker job failed' }); }
+ } catch (err) { if (ws === connection) send({ type: 'result', id: m.id, ok: false, error: err.code === 'worker.transfer_limit' ? 'worker.transfer_limit' : 'worker job failed' }); }
status();
status();
});
});
ws.on('error', () => {}); // Never log a credential or server-supplied error.
ws.on('error', () => {}); // Never log a credential or server-supplied error.
@@ -7,21 +7,25 @@ const { MAX_BYTES, TYPES, bytes } = require('./protocol');
@@ -7,21 +7,25 @@ const { MAX_BYTES, TYPES, bytes } = require('./protocol');
const { detect } = require('../../_shared/guard/sniff');
const { detect } = require('../../_shared/guard/sniff');
const { poolFromEnv } = require('../../_shared/jobs/pool');
const { poolFromEnv } = require('../../_shared/jobs/pool');
function definitions() {
function definitions() {
- const defs = new Map(), pools = [];
+ const defs = new Map(), pools = [], tools = [];
+ const hasAudio = ['ffmpeg', 'ffprobe'].every(bin => require('node:child_process').spawnSync(bin, ['-version'], { stdio: 'ignore', timeout: 3000 }).status === 0);
for (const app of ['img', 'docs', 'audio']) {
for (const app of ['img', 'docs', 'audio']) {
try {
try {
+ if (app === 'audio' && !hasAudio) continue;
+ const catalog = require(`../../${app}/server/tools`);
+ tools.push(...catalog.listTools().filter(t => t.available !== false).map(t => `${app}.process:${t.id}`));
const process = require(`../../${app}/server/process`);
const process = require(`../../${app}/server/process`);
const pool = app === 'audio' ? null : poolFromEnv(app, path.resolve(__dirname, `../../${app}/server/worker.js`), { size: 1, maxQueue: 0 });
const pool = app === 'audio' ? null : poolFromEnv(app, path.resolve(__dirname, `../../${app}/server/worker.js`), { size: 1, maxQueue: 0 });
if (pool) pools.push(pool);
if (pool) pools.push(pool);
process.defineJobs({ define: def => defs.set(def.type, def), JobError: Error }, pool);
process.defineJobs({ define: def => defs.set(def.type, def), JobError: Error }, pool);
} catch { /* A missing dependency removes this capability, rather than accepting work it cannot run. */ }
} catch { /* A missing dependency removes this capability, rather than accepting work it cannot run. */ }
}
}
- return { defs, close: () => Promise.all(pools.map(p => p.close())) };
+ return { defs, tools, close: () => Promise.all(pools.map(p => p.close())) };
}
}
function createRuntime({ slots = 2, dry = false, registry = definitions(), root = os.tmpdir() } = {}) {
function createRuntime({ slots = 2, dry = false, registry = definitions(), root = os.tmpdir() } = {}) {
if (!Number.isInteger(slots) || slots < 1 || slots > 8) throw new Error('slots must be 1..8');
if (!Number.isInteger(slots) || slots < 1 || slots > 8) throw new Error('slots must be 1..8');
const active = new Map();
const active = new Map();
- function status() { return { capabilities: dry ? [] : [...registry.defs.keys()].filter(k => TYPES.includes(k)), slots, free: dry ? 0 : slots - active.size, load: os.loadavg()[0], dry }; }
+ function status() { return { tools: dry ? [] : (registry.tools || []), capabilities: dry ? [] : [...registry.defs.keys()].filter(k => TYPES.includes(k)), slots, free: dry ? 0 : slots - active.size, load: os.loadavg()[0], dry }; }
async function run(job) {
async function run(job) {
const def = registry.defs.get(job.type);
const def = registry.defs.get(job.type);
if (dry || !TYPES.includes(job.type) || !def || active.size >= slots || active.has(job.id) || !/^[a-f0-9]{32}$/.test(job.id || '')) throw new Error('job refused');
if (dry || !TYPES.includes(job.type) || !def || active.size >= slots || active.has(job.id) || !/^[a-f0-9]{32}$/.test(job.id || '')) throw new Error('job refused');
@@ -57,7 +61,7 @@ function createRuntime({ slots = 2, dry = false, registry = definitions(), root
@@ -57,7 +61,7 @@ function createRuntime({ slots = 2, dry = false, registry = definitions(), root
total = 0;
total = 0;
for (const f of result.files || []) {
for (const f of result.files || []) {
const size = (await fs.stat(f.path)).size; total += size;
const size = (await fs.stat(f.path)).size; total += size;
- if (total > MAX_BYTES) throw new Error('result exceeds worker transfer limit');
+ if (total > MAX_BYTES) throw Object.assign(new Error('result exceeds worker transfer limit'), { code: 'worker.transfer_limit' });
output.push({ name: path.basename(f.name), mime: f.mime, bytes: (await fs.readFile(f.path)).toString('base64') });
output.push({ name: path.basename(f.name), mime: f.mime, bytes: (await fs.readFile(f.path)).toString('base64') });
}
}
return { files: output, data: result.data || {} };
return { files: output, data: result.data || {} };
new file mode 100644
new file mode 100644
@@ -0,0 +1,59 @@
@@ -0,0 +1,59 @@
+'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 { choose, createRouter } = require('../../gateway/server/workers/router');
+const { executePlaced } = require('../../_shared/jobs/placement');
+const worker = (id, free = 2, latency_ms = 10) => ({ id, location: 'home', free, slots: 2, latency_ms, pending: new Map(), draining: false, capabilities: ['img.process'], tools: ['img.process:convert'] });
+(async () => {
+ const a = worker('a'), b = worker('b', 1, 1), connections = new Map([['a', a], ['b', b]]);
+ assert.equal(choose(connections, ['home', 'host'], 'img.process', 'convert').id, 'a');
+ a.pending.set('busy', {}); assert.equal(choose(connections, ['home', 'host'], 'img.process', 'convert').id, 'b');
+ assert.equal(choose(connections, ['a', 'home', 'host'], 'img.process', 'convert').id, 'a');
+ assert.equal(choose(connections, ['host', 'home'], 'img.process', 'convert'), null);
+ a.draining = true; b.draining = true; assert.equal(choose(connections, ['home'], 'img.process', 'convert'), null);
+ a.draining = false; b.draining = false; a.pending.clear();
+ const dir = await fs.mkdtemp(path.join(os.tmpdir(), 'tools-routing-test-'));
+ try {
+ const png = await require('../../img/node_modules/sharp')({ create: { width: 1, height: 1, channels: 3, background: 'red' } }).png().toBuffer();
+ const file = path.join(dir, 'input.png'); await fs.writeFile(file, png);
+ const row = { id: 'test', type: 'img.process', state: 'running', input_json: '{"tool":"convert"}', files_json: JSON.stringify([{ name: 'input.png', _path: file }]), execution_prefer: 'auto', attempts: 1 };
+ let enabled = false, seen = [];
+ const hub = { connections, dispatch: async (w, job) => {
+ seen.push(w.id); assert.deepEqual(Object.keys(job).sort(), ['files', 'input', 'type']);
+ if (w.id === 'a') throw Object.assign(new Error('gone'), { code: 'worker.disconnected' });
+ return { ok: true, result: { files: [], data: { converted: true } } };
+ } };
+ const router = createRouter({ hub, store: { routing: async () => ({ enabled, worker_order: '["home","host"]' }) }, db: { prepare: () => ({ get: async () => row, run: async () => {} }) } });
+ const signal = new AbortController().signal;
+ assert.equal((await router.route('test', signal)).mode, 'host'); assert.deepEqual(seen, []);
+ enabled = true;
+ const out = await router.route('test', signal);
+ assert.equal(out.mode, 'remote'); assert.deepEqual(seen, ['a', 'b']);
+ assert.deepEqual(out.attempts.map(a => a.state), ['disconnected', 'succeeded']);
+ seen = []; connections.delete('b');
+ const fallback = await router.route('test', signal); assert.equal(fallback.mode, 'host'); assert.equal(fallback.attempts.length, 1);
+ connections.set('b', b); hub.dispatch = async w => { seen.push(w.id); throw Object.assign(new Error('gone'), { code: 'worker.disconnected' }); };
+ seen = []; assert.equal((await router.route('test', signal)).mode, 'failed'); assert.equal(seen.length, 2);
+ hub.dispatch = async () => ({ ok: false, error: 'worker.transfer_limit' });
+ assert.equal((await router.route('test', signal)).mode, 'host', 'expanded audio/PDF outputs retry on the host');
+ row.type = 'yt.download'; seen = []; assert.equal((await router.route('test', signal)).mode, 'host'); assert.equal(seen.length, 0);
+ row.type = 'img.process'; row.execution_prefer = 'host'; assert.equal((await router.route('test', signal)).mode, 'host');
+ const receipt = {};
+ const result = await executePlaced({ row, def: { run: async () => ({ files: [], data: { local: true } }) }, context: { signal, outDir: dir }, receipt, request: async () => { throw new Error('must not dispatch preferred host'); } });
+ assert.equal(result.data.execution.worker, 'host'); assert.equal(receipt.attempts[0].state, 'succeeded');
+ assert(Number.isFinite(receipt.took_ms));
+ row.execution_prefer = 'auto';
+ let clock = 0; const retried = {};
+ await executePlaced({ row, def: { run: async () => ({}) }, context: { signal, outDir: dir }, receipt: retried, now: () => ++clock, request: async () => ({ mode: 'host', attempts: [{ worker: 'a', location: 'home', took_ms: 30, state: 'disconnected' }] }) });
+ assert.equal(retried.took_ms, 31); assert.equal(retried.attempts.length, 2);
+ const recovered = {};
+ await executePlaced({ row, def: { run: async () => ({}) }, context: { signal, outDir: dir }, receipt: recovered, request: async () => { throw new Error('gateway restart'); }, loadAttempts: async () => [{ worker: 'a', location: 'home', took_ms: 20, state: 'running' }] });
+ assert.deepEqual(recovered.attempts.map(a => a.worker), ['a', 'host']);
+ assert.equal(recovered.attempts[0].state, 'disconnected');
+ const remote = await executePlaced({ row, def: {}, context: { signal, outDir: dir }, receipt: {}, request: async () => ({ mode: 'remote', attempts: [{ worker: 'b', location: 'home', took_ms: 12, state: 'succeeded' }], result: { files: [{ name: '../../safe.png', mime: 'image/png', bytes: png.toString('base64') }], data: {} } }) });
+ assert.equal(remote.files[0].name, 'safe.png'); assert.equal(path.dirname(remote.files[0].path), dir); assert.equal(remote.data.execution.worker, 'b');
+ console.log('routing: owner order, capacity reservation, latency, drain, default-off, one retry, host fallback, fetch exclusion and receipts passed');
+ } finally { await fs.rm(dir, { recursive: true, force: true }); }
+})().catch(e => { console.error(e); process.exitCode = 1; });
@@ -46,7 +46,7 @@ const { createHub } = require('../../gateway/server/workers/hub');
@@ -46,7 +46,7 @@ const { createHub } = require('../../gateway/server/workers/hub');
const url = `ws://127.0.0.1:${server.address().port}/api/v1/workers/connect`;
const url = `ws://127.0.0.1:${server.address().port}/api/v1/workers/connect`;
const ws = new WebSocket(url, { headers });
const ws = new WebSocket(url, { headers });
await once(ws, 'open');
await once(ws, 'open');
- ws.send(JSON.stringify({ type: 'status', capabilities: ['img.process'], slots: 1, free: 1, load: 0 }));
+ ws.send(JSON.stringify({ type: 'status', tools: ['img.process:convert'], capabilities: ['img.process'], slots: 1, free: 1, load: 0 }));
await new Promise(r => setTimeout(r, 30));
await new Promise(r => setTimeout(r, 30));
assert.equal(hub.stats()[0].free, 1);
assert.equal(hub.stats()[0].free, 1);
const replay = new WebSocket(url, { headers });
const replay = new WebSocket(url, { headers });
new file mode 100644
new file mode 100644
@@ -0,0 +1,14 @@
@@ -0,0 +1,14 @@
+-- phase: expand
+ALTER TABLE tool_jobs ADD COLUMN IF NOT EXISTS execution_prefer TEXT NOT NULL DEFAULT 'auto';
+CREATE TABLE IF NOT EXISTS tool_worker_attempts (
+ job_id TEXT NOT NULL REFERENCES tool_jobs(id) ON DELETE CASCADE,
+ job_attempt INTEGER NOT NULL,
+ attempt INTEGER NOT NULL,
+ worker_id TEXT NOT NULL,
+ location TEXT NOT NULL,
+ took_ms BIGINT NOT NULL,
+ state TEXT NOT NULL,
+ at BIGINT NOT NULL,
+ PRIMARY KEY (job_id, job_attempt, attempt)
+);
+CREATE INDEX IF NOT EXISTS tool_worker_attempts_day ON tool_worker_attempts(at, worker_id);
@@ -18,7 +18,8 @@
@@ -18,7 +18,8 @@
"test:discovery": "node test/run.js gateway/test/home-jobs gateway/test/tool-opens gateway/test/discovery gateway/test/run-api _shared/test/job-usage gateway/test/perf-budget",
"test:discovery": "node test/run.js gateway/test/home-jobs gateway/test/tool-opens gateway/test/discovery gateway/test/run-api _shared/test/job-usage gateway/test/perf-budget",
"test:worker": "node apps/worker/test/worker.test.js",
"test:worker": "node apps/worker/test/worker.test.js",
"test:suite": "node --unhandled-rejections=strict test/run.js",
"test:suite": "node --unhandled-rejections=strict test/run.js",
- "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"
},
},
"license": "MIT",
"license": "MIT",
"engines": {
"engines": {