openvibers/openvibe.tools Commits
Jobs publish their lifecycle to OpenVibe.Events: tools.job.created (submit or retry; actor the owner), tools.job.started (claim), tools.job.succeeded and tools.job.failed (tool error, timeout, or failed on restart), each validated against openvibe-contracts tools.job.*@1 and written to an event_outbox table in the satellite's jobs.db (openvibe-sdk createOutbox) in the same SQLite transaction as the state change, including claim(); the relay posts to EVENTS_URL/api/v1/events with the tools client's token (events.event.publish on openvibe.events). Payloads carry ids, type, owner (null for a browser session), state, attempts, times, result locations (index, mime, size, sha256, local or media_id) and the error with server paths and the job's file names removed; never inputs, file names, output data or sessions. Cancelled jobs, progress, requeues and sandbox app jobs are not announced. Inert without EVENTS_URL (EVENTS_PUBLISH=off or no OV_OAUTH_CLIENT_SECRET also keep it off): no table, nothing written. A failed outbox write rolls the transition back; a finish that cannot commit leaves the job running with its inputs for boot recovery. apps/_shared/test/job-events.test.js covers one event per transition, rollbacks, sandbox, cancel, retry, restart, leaks, contract validation and the relay
Changes
@@@ -107,14 +107,27 @@ reload reattaches.`media.object.upload` and `media.object.read` for audience `openvibe.media`, namespace `tools`, and a `tools`tenant in Media. Without them (or if an upload fails) results stay on local disk and the file says`storage: "local"`. Default: `local`.++- **Lifecycle events to OpenVibe.Events** when `EVENTS_URL` is set (`apps/_shared/jobs/events.js`):++ `tools.job.created` (submit or retry; actor the owner), `tools.job.started` (a worker claimed it; a job++ re-queued after a restart is announced again when it starts), `tools.job.succeeded` and `tools.job.failed`++ (payloads `openvibe-contracts` `tools.job.*@1`, validated before they are queued; subject `job <id>`, visibility++ `internal`, source `tools`). Each is written to an `event_outbox` table in the satellite's `jobs.db` (openvibe-sdk++ `createOutbox`) in the same SQLite transaction as the state change, so an event exists exactly when its++ transition committed, and relayed to `EVENTS_URL/api/v1/events` (at least once; Events dedupes on `event_id`).++ Payloads carry ids, type, owner, state, attempts, times, where result files are (index, mime, size, sha256,++ `local` or a Media `media_id`) and the error (`status`, `code`, `detail` with server paths and the job's file++ names taken out) — never the input, file names, output data or a browser session: a session's job has++ `owner: null`. Cancelled jobs, progress and sandbox app jobs are not announced; an Idempotency-Key replay++ creates no job and no event. The relay uses the `tools` client (`OV_OAUTH_CLIENT_ID`, `OV_OAUTH_CLIENT_SECRET`)++ with `events.event.publish` on audience `openvibe.events`. Without `EVENTS_URL` (or with `EVENTS_PUBLISH=off`, or++ no client secret) nothing is written or sent. `GET /api/health` shows `jobs.events` (pending, rejected).Environment (all in `/etc/openvibe/tools.env`): `TOOLS_JOBS_CONCURRENCY`, `TOOLS_JOBS_CONCURRENCY_<APP>`,`TOOLS_JOBS_MAX_ACTIVE`, `TOOLS_JOB_RESULTS`, `TOOLS_MEDIA_NAMESPACE`, `OV_MEDIA_INTERNAL_URL`, `OV_MEDIA_URL`,--`OV_NETWORK_INTERNAL_URL`, `OV_OAUTH_CLIENT_ID`, `OV_OAUTH_CLIENT_SECRET`.++`OV_NETWORK_INTERNAL_URL`, `OV_OAUTH_CLIENT_ID`, `OV_OAUTH_CLIENT_SECRET`, `EVENTS_URL`, `EVENTS_PUBLISH`,++`EVENTS_RELAY_INTERVAL_MS`.--Not done yet: job lifecycle events are not published to OpenVibe.Events (openvibe-contracts registers no--`tools.job.*` event types, and Tools does not depend on openvibe-sdk yet); quotas for external developer apps are--the per-owner limits above, not a Codes-issued quota.++Not done yet: quotas for external developer apps are the per-owner limits above, not a Codes-issued quota.## Canonical hosts and the service registrynnew file mode 100644@@@ -0,0 +1,207 @@++'use strict';++// ═══════════════════════════════════════════════════════════════++// Job lifecycle → OpenVibe.Events (roadmap Wave 11, ADR-004).++//++// queued (submit, retry) → tools.job.created priority low actor: the owner, else service:tools++// running (claim) → tools.job.started priority low actor: service:tools++// succeeded → tools.job.succeeded priority important actor: service:tools++// failed (tool, timeout, boot) → tools.job.failed priority important actor: service:tools++//++// Each event is written to the satellite's own `event_outbox` table (openvibe-sdk createOutbox) in++// the SQLite transaction that records the transition, so it exists if and only if the transition++// committed; the relay posts it to OpenVibe.Events afterwards (at least once; Events dedupes on++// event_id). A cancelled job, a progress tick and a requeue after a restart are not announced (a++// requeued job is announced again when it starts). Sandbox jobs (developer-app sandbox tokens) are++// never announced.++//++// Payloads (openvibe-contracts tools.job.*@1, validated before they are enqueued) carry ids, type,++// owner, state, times, where result files are and the error; never the job's input, file names,++// output data or a browser session: a session-owned job has owner null.++//++// Inert unless EVENTS_URL is set (and OV_OAUTH_CLIENT_SECRET, for the tools service token with++// events.event.publish on openvibe.events): no outbox table, nothing written, nothing relayed.++// EVENTS_PUBLISH=off turns it off with EVENTS_URL set.++// ═══════════════════════════════════════════════════════════════++++const SOURCE = 'tools';++const SERVICE_ACTOR = Object.freeze({ type: 'service', id: SOURCE });++// Owner prefixes in tool_jobs.owner → SubjectRef types. `session:<hash>` (a browser) maps to null.++const OWNER_TYPES = { user: 'user', svc: 'service', app: 'app', mod: 'mod' };++const PRIORITY = { 'tools.job.created': 'low', 'tools.job.started': 'low', 'tools.job.succeeded': 'important', 'tools.job.failed': 'important' };++++const iso = (ms) => (ms == null ? null : new Date(ms).toISOString());++const parse = (s, fallback) => { if (s == null) return fallback; try { return JSON.parse(s); } catch { return fallback; } };++++/** 'user:usr_…' | 'svc:<slug>' | 'app:app_…' | 'mod:mod_…' → SubjectRef; anything else (a session) → null. */++function ownerRef(contracts, owner) {++ const s = String(owner || '');++ const i = s.indexOf(':');++ if (i < 1) return null;++ const type = OWNER_TYPES[s.slice(0, i)];++ const id = s.slice(i + 1);++ return type && contracts.ids.isSubjectId(type, id) ? { type, id } : null;++}++++/** An error detail for the network: the job's own file names are taken out as well as server paths. */++function scrubDetail(detail, row) {++ let out = String(detail == null ? '' : detail);++ const names = new Set();++ for (const f of parse(row.files_json, [])) {++ if (f && f.name && String(f.name).length >= 3) names.add(String(f.name));++ if (f && f._path) { const base = String(f._path).split(/[/\\]/).pop(); if (base.length >= 3) names.add(base); }++ }++ for (const n of [...names].sort((a, b) => b.length - a.length)) out = out.split(n).join('[file]');++ return out.slice(0, 500);++}++++/** tool_jobs row → the payload of `eventType` (tools.job.*@1). */++function payloadFor(eventType, row, { contracts, service, referenced = false }) {++ const p = {++ job_id: row.id,++ service,++ type: row.type,++ type_version: row.type_version,++ state: row.state,++ owner: ownerRef(contracts, row.owner),++ attempts: row.attempts,++ max_attempts: row.max_attempts,++ created_at: iso(row.created_at),++ retry_of: row.retry_of || null,++ };++ if (eventType === 'tools.job.created') return p;++ p.started_at = iso(row.started_at);++ if (eventType === 'tools.job.started') return p;++ p.finished_at = iso(row.finished_at);++ p.expires_at = referenced ? null : iso(row.expires_at);++ if (eventType === 'tools.job.succeeded') {++ const result = parse(row.result_json, null) || {};++ p.result = {++ files: (result.files || []).map((f, index) => ({++ index,++ mime: String(f.mime || 'application/octet-stream'),++ size: Number(f.size) || 0,++ sha256: f.sha256,++ storage: f.storage === 'media' ? 'media' : 'local',++ media: f.storage === 'media' && f.media && f.media.media_id ? { media_id: f.media.media_id, role: f.media.role || 'output' } : null,++ })),++ };++ return p;++ }++ const error = parse(row.error_json, null) || {};++ p.error = {++ status: Number.isInteger(error.status) ? error.status : 500,++ code: /^tools\.[a-z0-9_.]+$/.test(String(error.code || '')) ? error.code : 'tools.job.failed',++ detail: scrubDetail(error.detail || error.title || 'The job failed', row),++ };++ p.retryable = !!row.retryable;++ return p;++}++++/**++ * The announcer the job system calls inside its transactions. With no outbox every call is a no-op.++ *++ * @param {object} o++ * @param {object} o.contracts require('openvibe-contracts') (v0.30.0+ knows tools.job.*@1)++ * @param {string} o.service the satellite ('img' | 'audio' | 'docs' | …) → payload.service++ * @param {object|null} o.outbox openvibe-sdk createOutbox(db, …) on the jobs database, or null++ * @param {(id) => number} [o.referenceCount]++ * @param {object} [o.log]++ */++function createJobEvents({ contracts, service, outbox = null, referenceCount = () => 0, log = console }) {++ const stats = { queued: 0, invalid: 0, lastInvalid: null };++++ /**++ * Enqueue `eventType` for this row (just read back inside the caller's transaction). A payload that++ * fails its contract is logged and skipped, never thrown: a bug here must not stop jobs. An outbox++ * write error does throw, so the caller's transaction rolls back with it.++ */++ function announce(eventType, row) {++ if (!outbox || !row || row.env === 'sandbox') return null;++ let payload;++ try {++ payload = payloadFor(eventType, row, { contracts, service, referenced: eventType !== 'tools.job.created' && eventType !== 'tools.job.started' && referenceCount(row.id) > 0 });++ const v = contracts.validate(`${eventType}@1`, payload);++ if (!v.valid) throw new Error(v.errors.map(e => `${e.path} ${e.message}`).join('; '));++ } catch (err) {++ stats.invalid++;++ stats.lastInvalid = `${eventType} ${row.id}: ${err.message}`;++ log.error(`[Jobs] ${eventType} for ${row.id} not announced (payload does not match the contract): ${err.message}`);++ return null;++ }++ const env = outbox.enqueue({++ event_type: eventType,++ actor: eventType === 'tools.job.created' && payload.owner ? payload.owner : SERVICE_ACTOR,++ subject: { type: 'job', id: row.id },++ visibility: 'internal',++ priority: PRIORITY[eventType],++ payload,++ });++ stats.queued++;++ outbox.kick(); // a timer: the relay runs after this transaction commits++ return env;++ }++++ return {++ enabled: !!outbox,++ created: (row) => announce('tools.job.created', row),++ started: (row) => announce('tools.job.started', row),++ /** succeeded → tools.job.succeeded, failed → tools.job.failed; cancelled is not announced. */++ finished: (row) => (row && row.state === 'succeeded' ? announce('tools.job.succeeded', row)++ : row && row.state === 'failed' ? announce('tools.job.failed', row) : null),++ status() {++ if (!outbox) return { enabled: false };++ return { enabled: true, pending: outbox.pending(), rejected: outbox.rejected(), queued_since_boot: stats.queued, invalid: stats.invalid, last_invalid: stats.lastInvalid };++ },++ };++}++++/**++ * The outbox from the environment → { outbox, reason } (outbox null = inert).++ *++ * EVENTS_URL OpenVibe.Events on this host, e.g. http://127.0.0.1:4300 (unset = off)++ * EVENTS_PUBLISH=off off even with EVENTS_URL set++ * OV_OAUTH_CLIENT_ID the tools client (default tools): events.event.publish on openvibe.events++ * OV_OAUTH_CLIENT_SECRET++ * OV_NETWORK_INTERNAL_URL token endpoint host (default http://127.0.0.1:4000)++ * EVENTS_RELAY_INTERVAL_MS relay poll (default 2000)++ *++ * @param {object} o++ * @param {object} o.db the jobs database (better-sqlite3) — the outbox table lives beside tool_jobs++ * @param {object} [o.sdk] require('openvibe-sdk')++ * @param {object} [o.env]++ * @param {Function} [o.fetch]++ * @param {object} [o.log]++ */++function outboxFromEnv({ db, sdk, env = process.env, fetch: fetchImpl, log = console }) {++ const url = String(env.EVENTS_URL || '').trim().replace(/\/+$/, '');++ if (!url) return { outbox: null, reason: 'EVENTS_URL is not set' };++ if (String(env.EVENTS_PUBLISH || '').toLowerCase() === 'off') return { outbox: null, reason: 'EVENTS_PUBLISH=off' };++ if (!sdk || !sdk.events || !sdk.auth) return { outbox: null, reason: 'openvibe-sdk is not installed in this app' };++ const secret = env.OV_OAUTH_CLIENT_SECRET || '';++ if (!secret) return { outbox: null, reason: 'EVENTS_URL is set but OV_OAUTH_CLIENT_SECRET is not' };++ const network = String(env.OV_NETWORK_INTERNAL_URL || 'http://127.0.0.1:4000').replace(/\/+$/, '');++ const tokens = sdk.auth.createServiceTokenClient({++ tokenUrl: `${network}/oauth/token`,++ clientId: env.OV_OAUTH_CLIENT_ID || 'tools',++ clientSecret: secret,++ scope: { 'openvibe.events': 'events.event.publish' },++ ...(fetchImpl ? { fetch: fetchImpl } : {}),++ });++ const client = sdk.createClient({ baseUrls: { events: url }, tokenProvider: tokens, retries: 0, ...(fetchImpl ? { fetch: fetchImpl } : {}) });++ let lastError = null;++ const interval = parseInt(env.EVENTS_RELAY_INTERVAL_MS, 10);++ const outbox = sdk.events.createOutbox(db, {++ events: sdk.events.createEventsClient(client, { source: SOURCE }),++ intervalMs: Number.isFinite(interval) && interval >= 100 ? interval : 2000,++ onError: (err) => {++ const msg = err && err.message;++ if (msg !== lastError) log.warn('[Jobs] events relay: publish failed (will retry):', msg);++ lastError = msg;++ },++ });++ outbox.ensureSchema();++ outbox.url = url;++ return { outbox, reason: null };++}++++module.exports = { createJobEvents, outboxFromEnv, payloadFor, ownerRef, scrubDetail, SOURCE };@@@ -1,9 +1,9 @@'use strict';// ═══════════════════════════════════════════════════════════════// Tools job runtime (roadmap Wave 11). Shared by the satellites the way apps/_shared always is:--// required by relative path, no dependencies of its own — each app passes its better-sqlite3 and--// openvibe-contracts. See system.js (lifecycle), http.js (routes), media.js (results in Media),--// client.js (browser helper).++// required by relative path, no dependencies of its own — each app passes its better-sqlite3,++// openvibe-contracts and openvibe-sdk. See system.js (lifecycle), http.js (routes), media.js (results++// in Media), events.js (tools.job.* to OpenVibe.Events), client.js (browser helper).//// Environment (one /etc/openvibe/tools.env for every unit):// TOOLS_JOBS_CONCURRENCY jobs running at once per satellite (default 2)@@@ -14,7 +14,10 @@// OV_MEDIA_INTERNAL_URL Media on this host (default http://127.0.0.1:4100; MEDIA_URL is read as a fallback)// OV_MEDIA_URL Media's public origin, for the CSP of previews (default https://openvibe.media)// OV_NETWORK_INTERNAL_URL token endpoint host (default http://127.0.0.1:4000)--// OV_OAUTH_CLIENT_ID / OV_OAUTH_CLIENT_SECRET the tools client (client_credentials for Media)++// OV_OAUTH_CLIENT_ID / OV_OAUTH_CLIENT_SECRET the tools client (client_credentials for Media and Events)++// EVENTS_URL OpenVibe.Events on this host; unset = job events are not published++// EVENTS_PUBLISH=off keep them off with EVENTS_URL set++// EVENTS_RELAY_INTERVAL_MS outbox relay poll (default 2000)// ═══════════════════════════════════════════════════════════════const fs = require('fs');@@@ -22,6 +25,7 @@ const path = require('path');const { createJobSystem, JobError } = require('./system');const { mountJobRoutes, createOwnerResolver, contentDisposition, CAPS } = require('./http');const { createMediaResults } = require('./media');++const { createJobEvents, outboxFromEnv } = require('./events');function envInt(name, fallback) {const v = parseInt(process.env[name], 10);@@@ -65,6 +69,7 @@ function mediaOrigin(env = process.env) {* @param {string} o.dataDir* @param {Function} o.Database require('better-sqlite3')* @param {object} o.contracts require('openvibe-contracts')++ * @param {object} [o.sdk] require('openvibe-sdk') — the outbox relay to OpenVibe.Events* @param {() => string|null} o.getPublicKey* @param {string} o.issuer* @param {(system) => void} o.define registers the app's job types@@@ -77,18 +82,33 @@ function setupJobs(o) {const db = new o.Database(path.join(o.dataDir, 'jobs.db'));const results = mediaFromEnv(o.contracts);if (results.reason && String(process.env.TOOLS_JOB_RESULTS || '').toLowerCase() === 'media') console.warn(`[Jobs] ${o.service}: results stay local — ${results.reason}`);++ // tools.job.* → OpenVibe.Events through an outbox table in this same jobs.db (inert without EVENTS_URL).++ const events = outboxFromEnv({ db, sdk: o.sdk });++ if (events.reason && process.env.EVENTS_URL) console.warn(`[Jobs] ${o.service}: job events are not published — ${events.reason}`);const system = createJobSystem({db, contracts: o.contracts, service: o.service, dataDir: o.dataDir,concurrency: envInt(`TOOLS_JOBS_CONCURRENCY_${o.service.toUpperCase()}`, envInt('TOOLS_JOBS_CONCURRENCY', 2)),maxActivePerOwner: envInt('TOOLS_JOBS_MAX_ACTIVE', 10) || 10,media: results.media,++ outbox: events.outbox,});o.define(system);const resolveOwner = createOwnerResolver({ contracts: o.contracts, getPublicKey: o.getPublicKey, issuer: o.issuer });mountJobRoutes(o.app, { system, contracts: o.contracts, resolveOwner, receive: o.receive, limiters: o.limiters, defaults: o.defaults });system.start();++ let pruneOutbox = null;++ if (events.outbox) {++ events.outbox.start();++ pruneOutbox = setInterval(() => { try { events.outbox.prune(); } catch { /* next time */ } }, 6 * 60 * 60 * 1000);++ if (pruneOutbox.unref) pruneOutbox.unref();++ console.log(`[Jobs] ${o.service}: job events → ${events.outbox.url} (${events.outbox.pending()} pending)`);++ }system.db = db;-- system.close = () => { system.stop(); try { db.close(); } catch { /* already closed */ } };++ system.close = () => {++ system.stop();++ if (events.outbox) { events.outbox.stop(); clearInterval(pruneOutbox); }++ try { db.close(); } catch { /* already closed */ }++ };return system;}@@@ -113,4 +133,4 @@ function readyChecks(getSystem, env = process.env) {}];}--module.exports = { setupJobs, readyChecks, createJobSystem, mountJobRoutes, createOwnerResolver, createMediaResults, mediaFromEnv, mediaOrigin, contentDisposition, JobError, CAPS };++module.exports = { setupJobs, readyChecks, createJobSystem, mountJobRoutes, createOwnerResolver, createMediaResults, mediaFromEnv, mediaOrigin, contentDisposition, JobError, CAPS, createJobEvents, outboxFromEnv };@@@ -20,6 +20,10 @@// (retry_of → the failed one, which records retried_by). Retrying the same failed job again// returns that same new job, so the call is idempotent.//++// Events: with an outbox (./events, openvibe-sdk createOutbox on this same database), created,++// started, succeeded and failed are also announced to OpenVibe.Events as tools.job.*, each written++// in the transaction that records the transition. Without one, nothing is announced.++//// No dependencies of its own: the app passes its better-sqlite3 handle and openvibe-contracts.// ═══════════════════════════════════════════════════════════════@@@ -28,6 +32,7 @@ const fsp = require('fs/promises');const path = require('path');const crypto = require('crypto');const { createStore, TERMINAL } = require('./store');++const { createJobEvents } = require('./events');const HOUR = 60 * 60 * 1000;const PRUNE_RECHECK_MS = 24 * HOUR; // an expired job the pruner could not finish is looked at again after this@@@ -71,6 +76,7 @@ const scrub = (msg) => String(msg || '').replace(/(?:\/[\w.-]+){2,}/g, '[file]')* @param {number} [o.concurrency=2]* @param {number} [o.maxActivePerOwner=10]* @param {object} [o.media] result store from ./media (null = keep results on local disk)++ * @param {object} [o.outbox] openvibe-sdk outbox on `db` (./events outboxFromEnv); null = no platform events* @param {number} [o.pruneIntervalMs=300000]* @param {number} [o.progressThrottleMs=250]*/@@@ -85,6 +91,7 @@ function createJobSystem(o) {const throttleMs = o.progressThrottleMs == null ? 250 : o.progressThrottleMs;const root = path.resolve(o.dataDir, 'jobs');fs.mkdirSync(root, { recursive: true });++ const announce = createJobEvents({ contracts, service: o.service, outbox: o.outbox || null, referenceCount: (id) => store.referenceCount(id), log });const types = new Map();const active = new Map(); // id → { ctrl, reason, last: { pct, msg, at } }@@@ -234,6 +241,7 @@ function createJobSystem(o) {idempotency_key: idempotencyKey == null ? null : String(idempotencyKey), request_hash: requestHash,max_attempts: def.maxAttempts, ttl_ms: ttlMs, now: Date.now(), env,});++ announce.created(store.get(id));})();} catch (err) {await fsp.rm(jobDir(id), { recursive: true, force: true });@@@ -262,7 +270,19 @@ function createJobSystem(o) {while (active.size < concurrency) {const [row] = store.nextQueued(1);if (!row) break;-- if (!store.claim(row.id)) continue;++ // The claim and its tools.job.started event commit together (or neither does).++ let claimed;++ try {++ claimed = store.transaction(() => {++ if (!store.claim(row.id)) return false;++ announce.started(store.get(row.id));++ return true;++ })();++ } catch (err) {++ log.error(`[Jobs] could not start ${row.id}:`, err.message);++ break; // it stays queued; the next kick tries again++ }++ if (!claimed) continue;const entry = { ctrl: new AbortController(), reason: null, last: { pct: -1, msg: null, at: 0 } };active.set(row.id, entry);emit(row.id, 'job.running');@@@ -320,14 +340,20 @@ function createJobSystem(o) {} finally {clearTimeout(timer);}++ let recorded = false;try {-- store.transaction(() => store.finish(row.id, outcome))();++ // The end state and its tools.job.succeeded|failed event commit together (or neither does).++ store.transaction(() => { if (store.finish(row.id, outcome)) announce.finished(store.get(row.id)); })();++ recorded = true;emit(row.id, EVENT_FOR[outcome.state]);++ } catch (err) {++ // Nothing committed: the row is still 'running' with its inputs, which the next start() recovers.++ log.error(`[Jobs] could not record the end of ${row.id}:`, err.message);} finally {active.delete(row.id);// A failed job keeps its inputs (until it expires) so it can be retried.-- if (outcome.state !== 'failed') fsp.rm(path.join(jobDir(row.id), 'in'), { recursive: true, force: true }).catch(() => {});-- if (outcome.state !== 'succeeded') fsp.rm(outDir, { recursive: true, force: true }).catch(() => {});++ if (recorded && outcome.state !== 'failed') fsp.rm(path.join(jobDir(row.id), 'in'), { recursive: true, force: true }).catch(() => {});++ if (recorded && outcome.state !== 'succeeded') fsp.rm(outDir, { recursive: true, force: true }).catch(() => {});kick();}}@@@ -437,6 +463,7 @@ function createJobSystem(o) {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,});++ announce.created(store.get(next));})();} catch (err) {if (moved) { try { fs.renameSync(newIn, oldIn); } catch { /* best effort */ } }@@@ -478,6 +505,11 @@ function createJobSystem(o) {}// ── Boot recovery, pruning ─────────────────────────────────++ /** A job the restart ended as failed: the row and its tools.job.failed event in one transaction. */++ function failRecovered(id, outcome) {++ store.transaction(() => { if (store.finish(id, outcome)) announce.finished(store.get(id)); })();++ }++function recover() {let requeued = 0, failed = 0;for (const row of store.running()) {@@@ -486,14 +518,14 @@ function createJobSystem(o) {store.finish(row.id, { state: 'cancelled', error: problem(409, 'tools.job.cancelled', 'The job was cancelled') });emit(row.id, 'job.cancelled');} else if (!def) {-- store.finish(row.id, { state: 'failed', error: problem(500, 'tools.job.unknown_type', `This service no longer runs "${row.type}" jobs`) });++ failRecovered(row.id, { state: 'failed', error: problem(500, 'tools.job.unknown_type', `This service no longer runs "${row.type}" jobs`) });emit(row.id, 'job.failed'); failed++;} else if (def.onRestart === 'requeue' && row.attempts < row.max_attempts) {store.requeue(row.id);emit(row.id, 'job.queued'); requeued++;} else {const retryable = def.onRestart === 'fail';-- store.finish(row.id, {++ failRecovered(row.id, {state: 'failed', retryable,error: problem(503, 'tools.job.interrupted', retryable ? 'The service restarted while this job was running; submit it again' : `The service restarted during each of ${row.attempts} attempts`),});@@@ -584,7 +616,8 @@ function createJobSystem(o) {types: () => [...types.keys()],media,// running (after the spread) is the store's count of rows in 'running'; executing is this process's.-- stats: () => ({ running: active.size, concurrency, ...store.counts(), executing: active.size, results: media ? 'media' : 'local' }),++ stats: () => ({ running: active.size, concurrency, ...store.counts(), executing: active.size, results: media ? 'media' : 'local', events: announce.status() }),++ outbox: o.outbox || null,/** True between start() and stop(): the worker picks up queued jobs. */isRunning: () => started && !stopped,store,nnew file mode 100644@@@ -0,0 +1,358 @@++'use strict';++// tools.job.* → OpenVibe.Events (roadmap Wave 11): every transition writes exactly one event to the++// jobs database's outbox in the transaction that records it; a rolled-back transition writes none;++// sandbox jobs and cancellations write none; payloads carry no input, file name, output data or++// browser session and validate against openvibe-contracts; the relay posts to EVENTS_URL with the++// tools service token; and without EVENTS_URL nothing is set up at all.++const assert = require('assert');++const fs = require('fs');++const os = require('os');++const path = require('path');++const { dep } = require('./deps');++++const Database = dep('better-sqlite3');++const contracts = dep('openvibe-contracts');++const sdk = dep('openvibe-sdk');++const jobs = require('../jobs');++const { outboxFromEnv, payloadFor, ownerRef } = require('../jobs/events');++++const USER = 'usr_01JAAAAAAAAAAAAAAAAAAAAAAA';++const APP = 'app_01JCCCCCCCCCCCCCCCCCCCCCCC';++const SESSION = 'session:0123456789abcdef0123456789abcdef01234567';++const SECRET_INPUT = 'secret-input-text-xyzzy';++const SECRET_FILE = 'holiday-photo-of-alice.txt';++const silent = { log() {}, warn() {}, error() {} };++const sleep = (ms) => new Promise(r => setTimeout(r, ms));++async function until(fn, what, ms = 5000) {++ const t0 = Date.now();++ for (;;) {++ const v = await fn();++ if (v) return v;++ if (Date.now() - t0 > ms) throw new Error(`timed out waiting for ${what}`);++ await sleep(10);++ }++}++++const gates = new Map();++function gate(name) {++ if (!gates.has(name)) { let open; const promise = new Promise(r => { open = r; }); gates.set(name, { promise, open }); }++ return gates.get(name);++}++const flaky = new Set();++function define(system) {++ system.define({++ type: 'test.upper', maxFiles: 1,++ async run({ input, files, outDir, signal }) {++ if (input.gate) await Promise.race([gate(input.gate).promise, new Promise((_, rej) => signal.addEventListener('abort', () => rej(new Error('aborted'))))]);++ const text = files.length ? fs.readFileSync(files[0].path, 'utf8') : input.text || '';++ const out = path.join(outDir, 'secret-output-name.txt');++ fs.writeFileSync(out, text.toUpperCase());++ return { files: [{ path: out, name: 'secret-output-name.txt', mime: 'text/plain' }], data: { echoed: input.text || null } };++ },++ });++ // Its error names the input file (by name and by server path), which must not reach the network.++ system.define({++ type: 'test.boom', maxFiles: 1,++ async run({ files }) { throw new Error(`Cannot read ${files[0].name} at ${files[0].path}`); },++ });++ system.define({++ type: 'test.flaky', maxFiles: 1,++ async run({ input, files, outDir }) {++ if (flaky.has(input.tag)) throw Object.assign(new Error('upstream hiccup'), { retryable: true });++ const out = path.join(outDir, 'f.txt');++ fs.writeFileSync(out, fs.readFileSync(files[0].path));++ return { files: [{ path: out, name: 'f.txt', mime: 'text/plain' }] };++ },++ });++ system.define({++ type: 'test.fragile', onRestart: 'fail',++ async run({ input, signal }) {++ await Promise.race([gate(input.gate).promise, new Promise((_, rej) => signal.addEventListener('abort', () => rej(new Error('aborted'))))]);++ return { data: {} };++ },++ });++}++++// ── A fake Network token endpoint + OpenVibe.Events, as a fetch ──++function fakeNetwork() {++ const calls = { token: [], publish: [] };++ let seq = 0;++ async function fetchImpl(url, init = {}) {++ const u = new URL(String(url));++ const reply = (status, body) => new Response(JSON.stringify(body), { status, headers: { 'content-type': 'application/json' } });++ if (u.host === 'network.test' && u.pathname === '/oauth/token') {++ const form = Object.fromEntries(new URLSearchParams(String(init.body)));++ calls.token.push(form);++ return reply(200, { access_token: 'events-token', token_type: 'Bearer', expires_in: 300, scope: form.scope });++ }++ if (u.host === 'events.test' && u.pathname === '/api/v1/events' && init.method === 'POST') {++ const headers = new Headers(init.headers);++ const body = JSON.parse(String(init.body));++ const list = body.events || [body];++ calls.publish.push({ authorization: headers.get('authorization'), events: list });++ const results = list.map(e => ({ event_id: e.event_id, seq: ++seq, duplicate: false }));++ return reply(201, body.events ? { results } : results[0]);++ }++ return reply(404, { code: 'not_found' });++ }++ return { fetch: fetchImpl, calls };++}++++const ENV = { EVENTS_URL: 'http://events.test/', OV_OAUTH_CLIENT_SECRET: 'tools-secret', OV_NETWORK_INTERNAL_URL: 'http://network.test' };++++function satellite(dir, { wrap, concurrency = 1, net } = {}) {++ fs.mkdirSync(dir, { recursive: true });++ const db = new Database(path.join(dir, 'jobs.db'));++ const { outbox, reason } = outboxFromEnv({ db, sdk, env: ENV, fetch: net.fetch, log: silent });++ assert.ok(outbox, `outbox configured (${reason})`);++ const system = jobs.createJobSystem({ db, contracts, service: 'img', dataDir: dir, concurrency, outbox: wrap ? wrap(outbox) : outbox, progressThrottleMs: 0, log: silent });++ define(system);++ system.start();++ const envelopes = (jobId) => db.prepare('SELECT envelope FROM event_outbox ORDER BY id').all().map(r => JSON.parse(r.envelope))++ .filter(e => !jobId || e.subject.id === jobId);++ return { db, system, outbox, envelopes, close() { system.stop(); outbox.stop(); db.close(); } };++}++++const types = (list) => list.map(e => e.event_type);++function checkContracts(env) {++ const v = contracts.validate(`${env.event_type}@1`, env.payload);++ assert.ok(v.valid, `${env.event_type} payload validates: ${JSON.stringify(v.errors)}`);++ const e = contracts.validate('events.event-envelope@1', env);++ assert.ok(e.valid, `${env.event_type} envelope validates: ${JSON.stringify(e.errors)}`);++ assert.strictEqual(env.source, 'tools');++ assert.deepStrictEqual(env.subject, { type: 'job', id: env.payload.job_id });++ assert.strictEqual(env.visibility, 'internal');++ assert.strictEqual(env.payload.service, 'img');++}++function noLeaks(list, extra = []) {++ const text = JSON.stringify(list);++ for (const s of [SECRET_INPUT, SECRET_FILE, 'secret-output-name', 'session:', SESSION.slice(8), 'echoed', '/tmp', os.tmpdir(), ...extra]) {++ assert.ok(!text.includes(s), `no "${s}" in published events`);++ }++ for (const env of list) {++ for (const k of ['input', 'input_json', 'files', 'files_json', 'data', 'idempotency_key', 'request_hash']) {++ assert.ok(!(k in env.payload), `${env.event_type} has no ${k}`);++ }++ }++}++const settled = (s, id, state) => until(() => { const r = s.system.get(id); return r && r.state === state && r; }, `${id} ${state}`);++++(async () => {++ const root = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-job-events-'));++ try {++ // ── Inert without EVENTS_URL ─────────────────────────────++ {++ const db = new Database(':memory:');++ assert.deepStrictEqual(outboxFromEnv({ db, sdk, env: {} }), { outbox: null, reason: 'EVENTS_URL is not set' });++ assert.strictEqual(outboxFromEnv({ db, sdk, env: { EVENTS_URL: 'http://events.test' } }).outbox, null, 'no secret, no outbox');++ assert.strictEqual(outboxFromEnv({ db, sdk, env: { ...ENV, EVENTS_PUBLISH: 'off' } }).outbox, null, 'EVENTS_PUBLISH=off');++ assert.strictEqual(outboxFromEnv({ db, sdk: null, env: ENV }).outbox, null, 'no sdk, no outbox');++ const dir = path.join(root, 'inert');++ fs.mkdirSync(dir);++ const system = jobs.createJobSystem({ db, contracts, service: 'img', dataDir: dir, log: silent });++ define(system);++ system.start();++ const { job } = await system.submit({ owner: `user:${USER}`, type: 'test.upper', input: { text: 'x' } });++ await until(() => system.get(job.id).state === 'succeeded', 'inert job');++ assert.strictEqual(db.prepare("SELECT COUNT(*) AS n FROM sqlite_master WHERE name = 'event_outbox'").get().n, 0, 'no outbox table without EVENTS_URL');++ assert.deepStrictEqual(system.stats().events, { enabled: false });++ system.stop(); db.close();++ }++++ const net = fakeNetwork();++ // An outbox that can be told to fail right after it wrote a row (a failing write inside the transaction).++ let failOn = null;++ const wrap = (outbox) => ({++ ...outbox,++ enqueue(envelope, opts) {++ const env = outbox.enqueue(envelope, opts);++ if (failOn === envelope.event_type) throw new Error('disk I/O error (test)');++ return env;++ },++ });++ let s = satellite(path.join(root, 'a'), { net, wrap });++++ // ── A signed-in person's job: created, started, succeeded — once each ──++ let { job } = await s.system.submit({ owner: `user:${USER}`, type: 'test.upper', input: { text: SECRET_INPUT }, files: [{ buffer: Buffer.from('hello'), name: SECRET_FILE, mime: 'text/plain' }] });++ await settled(s, job.id, 'succeeded');++ let list = s.envelopes(job.id);++ assert.deepStrictEqual(types(list), ['tools.job.created', 'tools.job.started', 'tools.job.succeeded']);++ list.forEach(checkContracts);++ noLeaks(list);++ assert.deepStrictEqual(list[0].actor, { type: 'user', id: USER }, 'created: the owner acts');++ assert.deepStrictEqual(list[1].actor, { type: 'service', id: 'tools' });++ assert.deepStrictEqual(list[2].actor, { type: 'service', id: 'tools' });++ assert.deepStrictEqual(list.map(e => e.priority), ['low', 'low', 'important']);++ assert.deepStrictEqual(list.map(e => e.payload.state), ['queued', 'running', 'succeeded']);++ assert.deepStrictEqual(list[0].payload.owner, { type: 'user', id: USER });++ assert.strictEqual(list[0].payload.retry_of, null);++ assert.strictEqual(list[1].payload.attempts, 1);++ const done = list[2].payload;++ assert.ok(done.started_at && done.finished_at && done.expires_at);++ assert.deepStrictEqual(Object.keys(done.result), ['files']);++ assert.strictEqual(done.result.files.length, 1);++ assert.deepStrictEqual(Object.keys(done.result.files[0]).sort(), ['index', 'media', 'mime', 'sha256', 'size', 'storage']);++ assert.strictEqual(done.result.files[0].storage, 'local');++ assert.strictEqual(done.result.files[0].media, null);++ assert.strictEqual(done.result.files[0].size, 5);++ assert.deepStrictEqual(s.system.stats().events.enabled, true);++++ // ── A failure whose message names the input file: failed once, name and path scrubbed ──++ ({ job } = await s.system.submit({ owner: `user:${USER}`, type: 'test.boom', input: { text: SECRET_INPUT }, files: [{ buffer: Buffer.from('x'), name: SECRET_FILE, mime: 'text/plain' }] }));++ await settled(s, job.id, 'failed');++ list = s.envelopes(job.id);++ assert.deepStrictEqual(types(list), ['tools.job.created', 'tools.job.started', 'tools.job.failed']);++ list.forEach(checkContracts);++ noLeaks(list);++ const failed = list[2].payload;++ assert.strictEqual(failed.error.code, 'tools.job.failed');++ assert.strictEqual(failed.error.status, 422);++ assert.match(failed.error.detail, /^Cannot read \[file\] at \[file\]$/);++ assert.strictEqual(failed.retryable, false);++ assert.deepStrictEqual(Object.keys(failed.error).sort(), ['code', 'detail', 'status']);++++ // ── A browser session's job: owner null, the service acts, the session never appears ──++ ({ job } = await s.system.submit({ owner: SESSION, type: 'test.upper', input: { text: SECRET_INPUT } }));++ await settled(s, job.id, 'succeeded');++ list = s.envelopes(job.id);++ assert.deepStrictEqual(types(list), ['tools.job.created', 'tools.job.started', 'tools.job.succeeded']);++ list.forEach(checkContracts);++ noLeaks(list);++ for (const e of list) { assert.strictEqual(e.payload.owner, null); assert.deepStrictEqual(e.actor, { type: 'service', id: 'tools' }); }++++ // ── Principals: a service and a production app are subjects ──++ ({ job } = await s.system.submit({ owner: 'svc:live', type: 'test.upper', input: {} }));++ await settled(s, job.id, 'succeeded');++ list = s.envelopes(job.id);++ list.forEach(checkContracts);++ assert.deepStrictEqual(list[0].actor, { type: 'service', id: 'live' });++ assert.deepStrictEqual(list[0].payload.owner, { type: 'service', id: 'live' });++ assert.deepStrictEqual(ownerRef(contracts, `app:${APP}`), { type: 'app', id: APP });++ assert.strictEqual(ownerRef(contracts, 'app:nope'), null);++++ // ── Sandbox app jobs are never announced ─────────────────++ const before = s.envelopes().length;++ ({ job } = await s.system.submit({ owner: `app:${APP}`, type: 'test.upper', input: { text: 'sandbox' }, env: 'sandbox' }));++ await settled(s, job.id, 'succeeded');++ const sandboxFail = (await s.system.submit({ owner: `app:${APP}`, type: 'test.boom', files: [{ buffer: Buffer.from('x'), name: 'a.txt' }], env: 'sandbox' })).job;++ await settled(s, sandboxFail.id, 'failed');++ assert.strictEqual(s.envelopes(job.id).length + s.envelopes(sandboxFail.id).length, 0, 'sandbox jobs emit nothing');++ assert.strictEqual(s.envelopes().length, before);++++ // ── Retry: a new created (retry_of) + its run; the failed job gets nothing new ──++ flaky.add('t1');++ const first = (await s.system.submit({ owner: `user:${USER}`, type: 'test.flaky', input: { tag: 't1' }, files: [{ buffer: Buffer.from('abc'), name: SECRET_FILE }] })).job;++ await settled(s, first.id, 'failed');++ assert.strictEqual(s.envelopes(first.id).at(-1).payload.retryable, true);++ flaky.delete('t1');++ const retried = s.system.retry(first.id).job;++ await settled(s, retried.id, 'succeeded');++ assert.deepStrictEqual(types(s.envelopes(first.id)), ['tools.job.created', 'tools.job.started', 'tools.job.failed'], 'the retried job gets no new event');++ list = s.envelopes(retried.id);++ assert.deepStrictEqual(types(list), ['tools.job.created', 'tools.job.started', 'tools.job.succeeded']);++ list.forEach(checkContracts);++ noLeaks(list);++ assert.strictEqual(list[0].payload.retry_of, first.id);++ assert.strictEqual(s.system.retry(first.id).replayed, true);++ assert.strictEqual(s.envelopes(retried.id).length, 3, 'a replayed retry creates no event');++++ // ── An Idempotency-Key replay creates no job and no event ──++ const k1 = (await s.system.submit({ owner: `user:${USER}`, type: 'test.upper', input: { text: 'k' }, idempotencyKey: 'key-00000001' })).job;++ await settled(s, k1.id, 'succeeded');++ const k2 = await s.system.submit({ owner: `user:${USER}`, type: 'test.upper', input: { text: 'k' }, idempotencyKey: 'key-00000001' });++ assert.strictEqual(k2.replayed, true);++ assert.strictEqual(s.envelopes(k1.id).length, 3);++++ // ── Cancelled jobs are not announced (neither queued nor running) ──++ const blocker = (await s.system.submit({ owner: `user:${USER}`, type: 'test.upper', input: { gate: 'g1' } })).job;++ await settled(s, blocker.id, 'running');++ const queued = (await s.system.submit({ owner: `user:${USER}`, type: 'test.upper', input: { text: 'q' } })).job;++ assert.strictEqual(s.system.cancel(queued.id).changed, true);++ assert.deepStrictEqual(types(s.envelopes(queued.id)), ['tools.job.created']);++ s.system.cancel(blocker.id);++ await settled(s, blocker.id, 'cancelled');++ assert.deepStrictEqual(types(s.envelopes(blocker.id)), ['tools.job.created', 'tools.job.started']);++++ // ── Rollbacks: a failed outbox write undoes the transition, and no event is left ──++ const count = () => s.envelopes().length;++ let n = count();++ failOn = 'tools.job.created';++ await assert.rejects(s.system.submit({ owner: `user:${USER}`, type: 'test.upper', input: { text: 'rollback' } }), /disk I\/O error/);++ failOn = null;++ assert.strictEqual(count(), n, 'rolled-back submit: no event');++ assert.strictEqual(s.db.prepare("SELECT COUNT(*) AS n FROM tool_jobs WHERE input_json LIKE '%rollback%'").get().n, 0, 'rolled-back submit: no job');++++ // claim(): the job stays queued and nothing says it started.++ failOn = 'tools.job.started';++ const stuck = (await s.system.submit({ owner: `user:${USER}`, type: 'test.upper', input: { text: 'claim' } })).job;++ await sleep(50);++ assert.strictEqual(s.system.get(stuck.id).state, 'queued', 'rolled-back claim: still queued');++ assert.strictEqual(s.system.get(stuck.id).attempts, 0);++ assert.deepStrictEqual(types(s.envelopes(stuck.id)), ['tools.job.created'], 'rolled-back claim: no started event');++ failOn = null;++ // The next kick (any submit) starts it: exactly one started.++ const kicker = (await s.system.submit({ owner: `user:${USER}`, type: 'test.upper', input: { text: 'kick' } })).job;++ await settled(s, stuck.id, 'succeeded');++ await settled(s, kicker.id, 'succeeded');++ assert.deepStrictEqual(types(s.envelopes(stuck.id)), ['tools.job.created', 'tools.job.started', 'tools.job.succeeded']);++++ // finish(): the job stays running (recovered on the next start) and nothing says it ended.++ failOn = 'tools.job.succeeded';++ const unfinished = (await s.system.submit({ owner: `user:${USER}`, type: 'test.upper', input: { text: 'finish' } })).job;++ await until(() => s.envelopes(unfinished.id).length === 2 && s.system.stats().executing === 0, 'finish attempt');++ await sleep(30);++ assert.strictEqual(s.system.get(unfinished.id).state, 'running', 'rolled-back finish: still running');++ assert.deepStrictEqual(types(s.envelopes(unfinished.id)), ['tools.job.created', 'tools.job.started']);++ failOn = null;++++ // ── Restart: the stuck job is re-queued (not announced) and started again (announced again);++ // a running onRestart:'fail' job fails once with tools.job.interrupted ──++ const fragile = (await s.system.submit({ owner: `user:${USER}`, type: 'test.fragile', input: { gate: 'never' } })).job;++ await settled(s, fragile.id, 'running');++ s.close();++ s = satellite(path.join(root, 'a'), { net });++ await settled(s, unfinished.id, 'succeeded');++ assert.deepStrictEqual(types(s.envelopes(unfinished.id)), ['tools.job.created', 'tools.job.started', 'tools.job.started', 'tools.job.succeeded']);++ assert.strictEqual(s.envelopes(unfinished.id)[2].payload.attempts, 2);++ const fr = s.envelopes(fragile.id);++ assert.deepStrictEqual(types(fr), ['tools.job.created', 'tools.job.started', 'tools.job.failed']);++ fr.forEach(checkContracts);++ assert.strictEqual(fr[2].payload.error.code, 'tools.job.interrupted');++ assert.strictEqual(fr[2].payload.retryable, true);++ assert.ok(fr[2].payload.started_at);++++ // ── Every event so far: valid, and never a secret ──++ const all = s.envelopes();++ all.forEach(checkContracts);++ noLeaks(all);++ assert.strictEqual(new Set(all.map(e => e.event_id)).size, all.length, 'unique event ids');++++ // ── The relay: POST EVENTS_URL/api/v1/events with the tools token for openvibe.events ──++ assert.strictEqual(s.outbox.pending(), all.length);++ const r = await s.outbox.flush();++ assert.strictEqual(r.sent, all.length);++ assert.strictEqual(s.outbox.pending(), 0);++ const published = net.calls.publish.flatMap(c => c.events);++ assert.deepStrictEqual(published.map(e => e.event_id), all.map(e => e.event_id), 'published in order, once');++ assert.ok(net.calls.publish.every(c => c.authorization === 'Bearer events-token'));++ assert.deepStrictEqual(net.calls.token[0], { grant_type: 'client_credentials', client_id: 'tools', client_secret: 'tools-secret', audience: 'openvibe.events', scope: 'events.event.publish' });++ s.close();++++ // ── Payload builder: a result in Media is { media_id, role } only ──++ const row = {++ id: `job_${contracts.ids.ulid()}`, type: 'img.process', type_version: 1, owner: `user:${USER}`, state: 'succeeded', attempts: 1, max_attempts: 3,++ created_at: Date.now() - 1000, started_at: Date.now() - 500, finished_at: Date.now(), expires_at: Date.now() + 3600000, retry_of: null, env: 'production',++ files_json: JSON.stringify([{ name: SECRET_FILE }]),++ result_json: JSON.stringify({ files: [{ name: 'secret-output-name.webp', mime: 'image/webp', size: 12, sha256: 'a'.repeat(64), storage: 'media', media: { media_id: `med_${contracts.ids.ulid()}`, role: 'output', namespace: 'tools', size_bytes: 12, content_hash: 'a'.repeat(64), mime_type: 'image/webp', status: 'ready' } }], data: { width: 3 } }),++ };++ const p = payloadFor('tools.job.succeeded', row, { contracts, service: 'img' });++ assert.ok(contracts.validate('tools.job.succeeded@1', p).valid);++ assert.deepStrictEqual(Object.keys(p.result.files[0].media).sort(), ['media_id', 'role']);++ assert.strictEqual(payloadFor('tools.job.succeeded', row, { contracts, service: 'img', referenced: true }).expires_at, null, 'a referenced result does not expire');++ assert.ok(!JSON.stringify(p).includes('secret-output-name'));++++ console.log('job events (tools.job.* outbox): all checks passed');++ } finally {++ fs.rmSync(root, { recursive: true, force: true });++ }++})().catch((err) => { console.error(err); process.exit(1); });@@@ -27,6 +27,7 @@ const { buildOptions, describe, defineJobs } = require('./process');const { hostGuard, ownHost } = require('../../_shared/host-role');const jobsRuntime = require('../../_shared/jobs');const contracts = require('openvibe-contracts');++const sdk = require('openvibe-sdk');// ── Analytics ────────────────────────────────────────────────const Database = require('better-sqlite3');@@@ -204,7 +205,7 @@ app.post('/api/process', burstLimiter, processLimiter, uploadSingle, async (req,// followed over SSE (ffmpeg's own progress), cancellable (ffmpeg is killed), reattachable by id// after a reload or a restart (apps/_shared/jobs).const jobs = jobsRuntime.setupJobs({-- app, service: 'audio', dataDir: path.resolve(__dirname, '..', config.dataDir), Database, contracts,++ app, service: 'audio', dataDir: path.resolve(__dirname, '..', config.dataDir), Database, contracts, sdk,getPublicKey: auth.getPublicKey, issuer: auth.ISSUER,define: defineJobs,receive: uploadSingle,@@@ -25,6 +25,7 @@ const { buildOptions, defineJobs } = require('./process');const { hostGuard, ownHost } = require('../../_shared/host-role');const jobsRuntime = require('../../_shared/jobs');const contracts = require('openvibe-contracts');++const sdk = require('openvibe-sdk');// ── Analytics ────────────────────────────────────────────────const Database = require('better-sqlite3');@@@ -251,7 +252,7 @@ app.post('/api/process/multi', burstLimiter, processLimiter, uploadMultiple, asy// The same operations as /api/process and /api/process/multi, asynchronous and durable: accepted// into data/jobs.db, followed over SSE, reattachable by id after a reload or a restart.const jobs = jobsRuntime.setupJobs({-- app, service: 'docs', dataDir: path.resolve(__dirname, '..', config.dataDir), Database, contracts,++ app, service: 'docs', dataDir: path.resolve(__dirname, '..', config.dataDir), Database, contracts, sdk,getPublicKey: auth.getPublicKey, issuer: auth.ISSUER,define: defineJobs,receive: uploadAny,@@@ -43,3 +43,8 @@ INTERNAL_API_KEY=change-me-to-a-random-64-char-string#TOOLS_MEDIA_NAMESPACE=tools#OV_MEDIA_INTERNAL_URL=http://127.0.0.1:4100#OV_MEDIA_URL=https://openvibe.media++# tools.job.created|started|succeeded|failed → OpenVibe.Events (off while unset). Needs OV_OAUTH_CLIENT_SECRET and++# the Network grant [tools, events.event.publish, openvibe.events].++#EVENTS_URL=http://127.0.0.1:4300++#EVENTS_PUBLISH=off++#EVENTS_RELAY_INTERVAL_MS=2000@@@ -25,6 +25,7 @@ const { buildOptions, describe, processBuffer, defineJobs } = require('./processconst { hostGuard, ownHost } = require('../../_shared/host-role');const jobsRuntime = require('../../_shared/jobs');const contracts = require('openvibe-contracts');++const sdk = require('openvibe-sdk');// ── Analytics ────────────────────────────────────────────────const Database = require('better-sqlite3');@@@ -201,7 +202,7 @@ app.post('/api/process/direct', burstLimiter, processLimiter, uploadSingle, asyn// The same operation as /api/process, asynchronous and durable: accepted into data/jobs.db,// followed over SSE, reattachable by id after a reload or a restart (apps/_shared/jobs).const jobs = jobsRuntime.setupJobs({-- app, service: 'img', dataDir: path.resolve(__dirname, '..', config.dataDir), Database, contracts,++ app, service: 'img', dataDir: path.resolve(__dirname, '..', config.dataDir), Database, contracts, sdk,getPublicKey: auth.getPublicKey, issuer: auth.ISSUER,define: defineJobs,receive: uploadSingle,@@@ -28,11 +28,11 @@ for app in apps/*/; dofidonedone--# Apps that run jobs (apps/_shared/jobs, required by relative path) hand it their own better-sqlite3 and--# openvibe-contracts: check both load under this Node (a native-module ABI mismatch shows up here, not in--# a crash loop) and that the shared runtime itself loads.++# Apps that run jobs (apps/_shared/jobs, required by relative path) hand it their own better-sqlite3,++# openvibe-contracts and openvibe-sdk (the tools.job.* outbox relay): check they load under this Node (a++# native-module ABI mismatch shows up here, not in a crash loop) and that the shared runtime itself loads.for app in img audio docs; do-- (cd "apps/$app" && node -e "const D=require('better-sqlite3'); new D(':memory:').close(); require('openvibe-contracts'); require('../_shared/jobs')") \++ (cd "apps/$app" && node -e "const D=require('better-sqlite3'); new D(':memory:').close(); require('openvibe-contracts'); require('openvibe-sdk'); require('../_shared/jobs')") \|| { echo "ABORT: apps/$app cannot load the jobs runtime; nothing restarted" >&2; exit 1; }doneUNITS=$(systemctl list-unit-files 'openvibe-tools*' --no-legend | awk '{print $1}')@@@ -1,5 +1,5 @@{-- "version": "0.2.0",++ "version": "0.3.0","contractRanges": {"openvibe-contracts": ">=0.7.0 <1.0.0"},@@@ -17,10 +17,15 @@"tools.job.read","tools.job.cancel"],-- "eventsProduced": [],++ "eventsProduced": [++ "tools.job.created",++ "tools.job.started",++ "tools.job.succeeded",++ "tools.job.failed"++ ],"eventsConsumed": [],"namespacesOwned": ["tools.*"],-- "notes": "Wave 11: asynchronous jobs on the img, audio and docs satellites (POST/GET/DELETE /api/v1/jobs, SSE events). Job lifecycle is not yet published to OpenVibe.Events, so eventsProduced stays empty. Tools consumes media.object.upload and media.object.read (namespace tools) when TOOLS_JOB_RESULTS=media."++ "notes": "Wave 11: asynchronous jobs on the img, audio and docs satellites (POST/GET/DELETE /api/v1/jobs, SSE events). Job lifecycle events tools.job.created|started|succeeded|failed (openvibe-contracts v0.30.1 payloads) go to OpenVibe.Events through an outbox in each satellite's jobs.db when EVENTS_URL is set. Tools consumes media.object.upload and media.object.read (namespace tools) when TOOLS_JOB_RESULTS=media."}
@@ -107,14 +107,27 @@ reload reattaches.
@@ -107,14 +107,27 @@ reload reattaches.
`media.object.upload` and `media.object.read` for audience `openvibe.media`, namespace `tools`, and a `tools`
`media.object.upload` and `media.object.read` for audience `openvibe.media`, namespace `tools`, and a `tools`
tenant in Media. Without them (or if an upload fails) results stay on local disk and the file says
tenant in Media. Without them (or if an upload fails) results stay on local disk and the file says
`storage: "local"`. Default: `local`.
`storage: "local"`. Default: `local`.
+- **Lifecycle events to OpenVibe.Events** when `EVENTS_URL` is set (`apps/_shared/jobs/events.js`):
+ `tools.job.created` (submit or retry; actor the owner), `tools.job.started` (a worker claimed it; a job
+ re-queued after a restart is announced again when it starts), `tools.job.succeeded` and `tools.job.failed`
+ (payloads `openvibe-contracts` `tools.job.*@1`, validated before they are queued; subject `job <id>`, visibility
+ `internal`, source `tools`). Each is written to an `event_outbox` table in the satellite's `jobs.db` (openvibe-sdk
+ `createOutbox`) in the same SQLite transaction as the state change, so an event exists exactly when its
+ transition committed, and relayed to `EVENTS_URL/api/v1/events` (at least once; Events dedupes on `event_id`).
+ Payloads carry ids, type, owner, state, attempts, times, where result files are (index, mime, size, sha256,
+ `local` or a Media `media_id`) and the error (`status`, `code`, `detail` with server paths and the job's file
+ names taken out) — never the input, file names, output data or a browser session: a session's job has
+ `owner: null`. Cancelled jobs, progress and sandbox app jobs are not announced; an Idempotency-Key replay
+ creates no job and no event. The relay uses the `tools` client (`OV_OAUTH_CLIENT_ID`, `OV_OAUTH_CLIENT_SECRET`)
+ with `events.event.publish` on audience `openvibe.events`. Without `EVENTS_URL` (or with `EVENTS_PUBLISH=off`, or
+ no client secret) nothing is written or sent. `GET /api/health` shows `jobs.events` (pending, rejected).
Environment (all in `/etc/openvibe/tools.env`): `TOOLS_JOBS_CONCURRENCY`, `TOOLS_JOBS_CONCURRENCY_<APP>`,
Environment (all in `/etc/openvibe/tools.env`): `TOOLS_JOBS_CONCURRENCY`, `TOOLS_JOBS_CONCURRENCY_<APP>`,
`TOOLS_JOBS_MAX_ACTIVE`, `TOOLS_JOB_RESULTS`, `TOOLS_MEDIA_NAMESPACE`, `OV_MEDIA_INTERNAL_URL`, `OV_MEDIA_URL`,
`TOOLS_JOBS_MAX_ACTIVE`, `TOOLS_JOB_RESULTS`, `TOOLS_MEDIA_NAMESPACE`, `OV_MEDIA_INTERNAL_URL`, `OV_MEDIA_URL`,
-`OV_NETWORK_INTERNAL_URL`, `OV_OAUTH_CLIENT_ID`, `OV_OAUTH_CLIENT_SECRET`.
+`OV_NETWORK_INTERNAL_URL`, `OV_OAUTH_CLIENT_ID`, `OV_OAUTH_CLIENT_SECRET`, `EVENTS_URL`, `EVENTS_PUBLISH`,
+`EVENTS_RELAY_INTERVAL_MS`.
-Not done yet: job lifecycle events are not published to OpenVibe.Events (openvibe-contracts registers no
-`tools.job.*` event types, and Tools does not depend on openvibe-sdk yet); quotas for external developer apps are
-the per-owner limits above, not a Codes-issued quota.
+Not done yet: quotas for external developer apps are the per-owner limits above, not a Codes-issued quota.
## Canonical hosts and the service registry
## Canonical hosts and the service registry
new file mode 100644
new file mode 100644
@@ -0,0 +1,207 @@
@@ -0,0 +1,207 @@
+'use strict';
+// ═══════════════════════════════════════════════════════════════
+// Job lifecycle → OpenVibe.Events (roadmap Wave 11, ADR-004).
+//
+// queued (submit, retry) → tools.job.created priority low actor: the owner, else service:tools
+// running (claim) → tools.job.started priority low actor: service:tools
+// succeeded → tools.job.succeeded priority important actor: service:tools
+// failed (tool, timeout, boot) → tools.job.failed priority important actor: service:tools
+//
+// Each event is written to the satellite's own `event_outbox` table (openvibe-sdk createOutbox) in
+// the SQLite transaction that records the transition, so it exists if and only if the transition
+// committed; the relay posts it to OpenVibe.Events afterwards (at least once; Events dedupes on
+// event_id). A cancelled job, a progress tick and a requeue after a restart are not announced (a
+// requeued job is announced again when it starts). Sandbox jobs (developer-app sandbox tokens) are
+// never announced.
+//
+// Payloads (openvibe-contracts tools.job.*@1, validated before they are enqueued) carry ids, type,
+// owner, state, times, where result files are and the error; never the job's input, file names,
+// output data or a browser session: a session-owned job has owner null.
+//
+// Inert unless EVENTS_URL is set (and OV_OAUTH_CLIENT_SECRET, for the tools service token with
+// events.event.publish on openvibe.events): no outbox table, nothing written, nothing relayed.
+// EVENTS_PUBLISH=off turns it off with EVENTS_URL set.
+// ═══════════════════════════════════════════════════════════════
+
+const SOURCE = 'tools';
+const SERVICE_ACTOR = Object.freeze({ type: 'service', id: SOURCE });
+// Owner prefixes in tool_jobs.owner → SubjectRef types. `session:<hash>` (a browser) maps to null.
+const OWNER_TYPES = { user: 'user', svc: 'service', app: 'app', mod: 'mod' };
+const PRIORITY = { 'tools.job.created': 'low', 'tools.job.started': 'low', 'tools.job.succeeded': 'important', 'tools.job.failed': 'important' };
+
+const iso = (ms) => (ms == null ? null : new Date(ms).toISOString());
+const parse = (s, fallback) => { if (s == null) return fallback; try { return JSON.parse(s); } catch { return fallback; } };
+
+/** 'user:usr_…' | 'svc:<slug>' | 'app:app_…' | 'mod:mod_…' → SubjectRef; anything else (a session) → null. */
+function ownerRef(contracts, owner) {
+ const s = String(owner || '');
+ const i = s.indexOf(':');
+ if (i < 1) return null;
+ const type = OWNER_TYPES[s.slice(0, i)];
+ const id = s.slice(i + 1);
+ return type && contracts.ids.isSubjectId(type, id) ? { type, id } : null;
+}
+
+/** An error detail for the network: the job's own file names are taken out as well as server paths. */
+function scrubDetail(detail, row) {
+ let out = String(detail == null ? '' : detail);
+ const names = new Set();
+ for (const f of parse(row.files_json, [])) {
+ if (f && f.name && String(f.name).length >= 3) names.add(String(f.name));
+ if (f && f._path) { const base = String(f._path).split(/[/\\]/).pop(); if (base.length >= 3) names.add(base); }
+ }
+ for (const n of [...names].sort((a, b) => b.length - a.length)) out = out.split(n).join('[file]');
+ return out.slice(0, 500);
+}
+
+/** tool_jobs row → the payload of `eventType` (tools.job.*@1). */
+function payloadFor(eventType, row, { contracts, service, referenced = false }) {
+ const p = {
+ job_id: row.id,
+ service,
+ type: row.type,
+ type_version: row.type_version,
+ state: row.state,
+ owner: ownerRef(contracts, row.owner),
+ attempts: row.attempts,
+ max_attempts: row.max_attempts,
+ created_at: iso(row.created_at),
+ retry_of: row.retry_of || null,
+ };
+ if (eventType === 'tools.job.created') return p;
+ p.started_at = iso(row.started_at);
+ if (eventType === 'tools.job.started') return p;
+ p.finished_at = iso(row.finished_at);
+ p.expires_at = referenced ? null : iso(row.expires_at);
+ if (eventType === 'tools.job.succeeded') {
+ const result = parse(row.result_json, null) || {};
+ p.result = {
+ files: (result.files || []).map((f, index) => ({
+ index,
+ mime: String(f.mime || 'application/octet-stream'),
+ size: Number(f.size) || 0,
+ sha256: f.sha256,
+ storage: f.storage === 'media' ? 'media' : 'local',
+ media: f.storage === 'media' && f.media && f.media.media_id ? { media_id: f.media.media_id, role: f.media.role || 'output' } : null,
+ })),
+ };
+ return p;
+ }
+ const error = parse(row.error_json, null) || {};
+ p.error = {
+ status: Number.isInteger(error.status) ? error.status : 500,
+ code: /^tools\.[a-z0-9_.]+$/.test(String(error.code || '')) ? error.code : 'tools.job.failed',
+ detail: scrubDetail(error.detail || error.title || 'The job failed', row),
+ };
+ p.retryable = !!row.retryable;
+ return p;
+}
+
+/**
+ * The announcer the job system calls inside its transactions. With no outbox every call is a no-op.
+ *
+ * @param {object} o
+ * @param {object} o.contracts require('openvibe-contracts') (v0.30.0+ knows tools.job.*@1)
+ * @param {string} o.service the satellite ('img' | 'audio' | 'docs' | …) → payload.service
+ * @param {object|null} o.outbox openvibe-sdk createOutbox(db, …) on the jobs database, or null
+ * @param {(id) => number} [o.referenceCount]
+ * @param {object} [o.log]
+ */
+function createJobEvents({ contracts, service, outbox = null, referenceCount = () => 0, log = console }) {
+ const stats = { queued: 0, invalid: 0, lastInvalid: null };
+
+ /**
+ * Enqueue `eventType` for this row (just read back inside the caller's transaction). A payload that
+ * fails its contract is logged and skipped, never thrown: a bug here must not stop jobs. An outbox
+ * write error does throw, so the caller's transaction rolls back with it.
+ */
+ function announce(eventType, row) {
+ if (!outbox || !row || row.env === 'sandbox') return null;
+ let payload;
+ try {
+ payload = payloadFor(eventType, row, { contracts, service, referenced: eventType !== 'tools.job.created' && eventType !== 'tools.job.started' && referenceCount(row.id) > 0 });
+ const v = contracts.validate(`${eventType}@1`, payload);
+ if (!v.valid) throw new Error(v.errors.map(e => `${e.path} ${e.message}`).join('; '));
+ } catch (err) {
+ stats.invalid++;
+ stats.lastInvalid = `${eventType} ${row.id}: ${err.message}`;
+ log.error(`[Jobs] ${eventType} for ${row.id} not announced (payload does not match the contract): ${err.message}`);
+ return null;
+ }
+ const env = outbox.enqueue({
+ event_type: eventType,
+ actor: eventType === 'tools.job.created' && payload.owner ? payload.owner : SERVICE_ACTOR,
+ subject: { type: 'job', id: row.id },
+ visibility: 'internal',
+ priority: PRIORITY[eventType],
+ payload,
+ });
+ stats.queued++;
+ outbox.kick(); // a timer: the relay runs after this transaction commits
+ return env;
+ }
+
+ return {
+ enabled: !!outbox,
+ created: (row) => announce('tools.job.created', row),
+ started: (row) => announce('tools.job.started', row),
+ /** succeeded → tools.job.succeeded, failed → tools.job.failed; cancelled is not announced. */
+ finished: (row) => (row && row.state === 'succeeded' ? announce('tools.job.succeeded', row)
+ : row && row.state === 'failed' ? announce('tools.job.failed', row) : null),
+ status() {
+ if (!outbox) return { enabled: false };
+ return { enabled: true, pending: outbox.pending(), rejected: outbox.rejected(), queued_since_boot: stats.queued, invalid: stats.invalid, last_invalid: stats.lastInvalid };
+ },
+ };
+}
+
+/**
+ * The outbox from the environment → { outbox, reason } (outbox null = inert).
+ *
+ * EVENTS_URL OpenVibe.Events on this host, e.g. http://127.0.0.1:4300 (unset = off)
+ * EVENTS_PUBLISH=off off even with EVENTS_URL set
+ * OV_OAUTH_CLIENT_ID the tools client (default tools): events.event.publish on openvibe.events
+ * OV_OAUTH_CLIENT_SECRET
+ * OV_NETWORK_INTERNAL_URL token endpoint host (default http://127.0.0.1:4000)
+ * EVENTS_RELAY_INTERVAL_MS relay poll (default 2000)
+ *
+ * @param {object} o
+ * @param {object} o.db the jobs database (better-sqlite3) — the outbox table lives beside tool_jobs
+ * @param {object} [o.sdk] require('openvibe-sdk')
+ * @param {object} [o.env]
+ * @param {Function} [o.fetch]
+ * @param {object} [o.log]
+ */
+function outboxFromEnv({ db, sdk, env = process.env, fetch: fetchImpl, log = console }) {
+ const url = String(env.EVENTS_URL || '').trim().replace(/\/+$/, '');
+ if (!url) return { outbox: null, reason: 'EVENTS_URL is not set' };
+ if (String(env.EVENTS_PUBLISH || '').toLowerCase() === 'off') return { outbox: null, reason: 'EVENTS_PUBLISH=off' };
+ if (!sdk || !sdk.events || !sdk.auth) return { outbox: null, reason: 'openvibe-sdk is not installed in this app' };
+ const secret = env.OV_OAUTH_CLIENT_SECRET || '';
+ if (!secret) return { outbox: null, reason: 'EVENTS_URL is set but OV_OAUTH_CLIENT_SECRET is not' };
+ const network = String(env.OV_NETWORK_INTERNAL_URL || 'http://127.0.0.1:4000').replace(/\/+$/, '');
+ const tokens = sdk.auth.createServiceTokenClient({
+ tokenUrl: `${network}/oauth/token`,
+ clientId: env.OV_OAUTH_CLIENT_ID || 'tools',
+ clientSecret: secret,
+ scope: { 'openvibe.events': 'events.event.publish' },
+ ...(fetchImpl ? { fetch: fetchImpl } : {}),
+ });
+ const client = sdk.createClient({ baseUrls: { events: url }, tokenProvider: tokens, retries: 0, ...(fetchImpl ? { fetch: fetchImpl } : {}) });
+ let lastError = null;
+ const interval = parseInt(env.EVENTS_RELAY_INTERVAL_MS, 10);
+ const outbox = sdk.events.createOutbox(db, {
+ events: sdk.events.createEventsClient(client, { source: SOURCE }),
+ intervalMs: Number.isFinite(interval) && interval >= 100 ? interval : 2000,
+ onError: (err) => {
+ const msg = err && err.message;
+ if (msg !== lastError) log.warn('[Jobs] events relay: publish failed (will retry):', msg);
+ lastError = msg;
+ },
+ });
+ outbox.ensureSchema();
+ outbox.url = url;
+ return { outbox, reason: null };
+}
+
+module.exports = { createJobEvents, outboxFromEnv, payloadFor, ownerRef, scrubDetail, SOURCE };
@@ -1,9 +1,9 @@
@@ -1,9 +1,9 @@
'use strict';
'use strict';
// ═══════════════════════════════════════════════════════════════
// ═══════════════════════════════════════════════════════════════
// Tools job runtime (roadmap Wave 11). Shared by the satellites the way apps/_shared always is:
// Tools job runtime (roadmap Wave 11). Shared by the satellites the way apps/_shared always is:
-// required by relative path, no dependencies of its own — each app passes its better-sqlite3 and
-// openvibe-contracts. See system.js (lifecycle), http.js (routes), media.js (results in Media),
-// client.js (browser helper).
+// required by relative path, no dependencies of its own — each app passes its better-sqlite3,
+// openvibe-contracts and openvibe-sdk. See system.js (lifecycle), http.js (routes), media.js (results
+// in Media), events.js (tools.job.* to OpenVibe.Events), client.js (browser helper).
//
//
// Environment (one /etc/openvibe/tools.env for every unit):
// Environment (one /etc/openvibe/tools.env for every unit):
// TOOLS_JOBS_CONCURRENCY jobs running at once per satellite (default 2)
// TOOLS_JOBS_CONCURRENCY jobs running at once per satellite (default 2)
@@ -14,7 +14,10 @@
@@ -14,7 +14,10 @@
// OV_MEDIA_INTERNAL_URL Media on this host (default http://127.0.0.1:4100; MEDIA_URL is read as a fallback)
// OV_MEDIA_INTERNAL_URL Media on this host (default http://127.0.0.1:4100; MEDIA_URL is read as a fallback)
// OV_MEDIA_URL Media's public origin, for the CSP of previews (default https://openvibe.media)
// OV_MEDIA_URL Media's public origin, for the CSP of previews (default https://openvibe.media)
// OV_NETWORK_INTERNAL_URL token endpoint host (default http://127.0.0.1:4000)
// OV_NETWORK_INTERNAL_URL token endpoint host (default http://127.0.0.1:4000)
-// OV_OAUTH_CLIENT_ID / OV_OAUTH_CLIENT_SECRET the tools client (client_credentials for Media)
+// OV_OAUTH_CLIENT_ID / OV_OAUTH_CLIENT_SECRET the tools client (client_credentials for Media and Events)
+// EVENTS_URL OpenVibe.Events on this host; unset = job events are not published
+// EVENTS_PUBLISH=off keep them off with EVENTS_URL set
+// EVENTS_RELAY_INTERVAL_MS outbox relay poll (default 2000)
// ═══════════════════════════════════════════════════════════════
// ═══════════════════════════════════════════════════════════════
const fs = require('fs');
const fs = require('fs');
@@ -22,6 +25,7 @@ const path = require('path');
@@ -22,6 +25,7 @@ const path = require('path');
const { createJobSystem, JobError } = require('./system');
const { createJobSystem, JobError } = require('./system');
const { mountJobRoutes, createOwnerResolver, contentDisposition, CAPS } = require('./http');
const { mountJobRoutes, createOwnerResolver, contentDisposition, CAPS } = require('./http');
const { createMediaResults } = require('./media');
const { createMediaResults } = require('./media');
+const { createJobEvents, outboxFromEnv } = require('./events');
function envInt(name, fallback) {
function envInt(name, fallback) {
const v = parseInt(process.env[name], 10);
const v = parseInt(process.env[name], 10);
@@ -65,6 +69,7 @@ function mediaOrigin(env = process.env) {
@@ -65,6 +69,7 @@ function mediaOrigin(env = process.env) {
* @param {string} o.dataDir
* @param {string} o.dataDir
* @param {Function} o.Database require('better-sqlite3')
* @param {Function} o.Database require('better-sqlite3')
* @param {object} o.contracts require('openvibe-contracts')
* @param {object} o.contracts require('openvibe-contracts')
+ * @param {object} [o.sdk] require('openvibe-sdk') — the outbox relay to OpenVibe.Events
* @param {() => string|null} o.getPublicKey
* @param {() => string|null} o.getPublicKey
* @param {string} o.issuer
* @param {string} o.issuer
* @param {(system) => void} o.define registers the app's job types
* @param {(system) => void} o.define registers the app's job types
@@ -77,18 +82,33 @@ function setupJobs(o) {
@@ -77,18 +82,33 @@ function setupJobs(o) {
const db = new o.Database(path.join(o.dataDir, 'jobs.db'));
const db = new o.Database(path.join(o.dataDir, 'jobs.db'));
const results = mediaFromEnv(o.contracts);
const results = mediaFromEnv(o.contracts);
if (results.reason && String(process.env.TOOLS_JOB_RESULTS || '').toLowerCase() === 'media') console.warn(`[Jobs] ${o.service}: results stay local — ${results.reason}`);
if (results.reason && String(process.env.TOOLS_JOB_RESULTS || '').toLowerCase() === 'media') console.warn(`[Jobs] ${o.service}: results stay local — ${results.reason}`);
+ // tools.job.* → OpenVibe.Events through an outbox table in this same jobs.db (inert without EVENTS_URL).
+ const events = outboxFromEnv({ db, sdk: o.sdk });
+ if (events.reason && process.env.EVENTS_URL) console.warn(`[Jobs] ${o.service}: job events are not published — ${events.reason}`);
const system = createJobSystem({
const system = createJobSystem({
db, contracts: o.contracts, service: o.service, dataDir: o.dataDir,
db, contracts: o.contracts, service: o.service, dataDir: o.dataDir,
concurrency: envInt(`TOOLS_JOBS_CONCURRENCY_${o.service.toUpperCase()}`, envInt('TOOLS_JOBS_CONCURRENCY', 2)),
concurrency: envInt(`TOOLS_JOBS_CONCURRENCY_${o.service.toUpperCase()}`, envInt('TOOLS_JOBS_CONCURRENCY', 2)),
maxActivePerOwner: envInt('TOOLS_JOBS_MAX_ACTIVE', 10) || 10,
maxActivePerOwner: envInt('TOOLS_JOBS_MAX_ACTIVE', 10) || 10,
media: results.media,
media: results.media,
+ outbox: events.outbox,
});
});
o.define(system);
o.define(system);
const resolveOwner = createOwnerResolver({ contracts: o.contracts, getPublicKey: o.getPublicKey, issuer: o.issuer });
const resolveOwner = createOwnerResolver({ contracts: o.contracts, getPublicKey: o.getPublicKey, issuer: o.issuer });
mountJobRoutes(o.app, { system, contracts: o.contracts, resolveOwner, receive: o.receive, limiters: o.limiters, defaults: o.defaults });
mountJobRoutes(o.app, { system, contracts: o.contracts, resolveOwner, receive: o.receive, limiters: o.limiters, defaults: o.defaults });
system.start();
system.start();
+ let pruneOutbox = null;
+ if (events.outbox) {
+ events.outbox.start();
+ pruneOutbox = setInterval(() => { try { events.outbox.prune(); } catch { /* next time */ } }, 6 * 60 * 60 * 1000);
+ if (pruneOutbox.unref) pruneOutbox.unref();
+ console.log(`[Jobs] ${o.service}: job events → ${events.outbox.url} (${events.outbox.pending()} pending)`);
+ }
system.db = db;
system.db = db;
- system.close = () => { system.stop(); try { db.close(); } catch { /* already closed */ } };
+ system.close = () => {
+ system.stop();
+ if (events.outbox) { events.outbox.stop(); clearInterval(pruneOutbox); }
+ try { db.close(); } catch { /* already closed */ }
+ };
return system;
return system;
}
}
@@ -113,4 +133,4 @@ function readyChecks(getSystem, env = process.env) {
@@ -113,4 +133,4 @@ function readyChecks(getSystem, env = process.env) {
}];
}];
}
}
-module.exports = { setupJobs, readyChecks, createJobSystem, mountJobRoutes, createOwnerResolver, createMediaResults, mediaFromEnv, mediaOrigin, contentDisposition, JobError, CAPS };
+module.exports = { setupJobs, readyChecks, createJobSystem, mountJobRoutes, createOwnerResolver, createMediaResults, mediaFromEnv, mediaOrigin, contentDisposition, JobError, CAPS, createJobEvents, outboxFromEnv };
@@ -20,6 +20,10 @@
@@ -20,6 +20,10 @@
// (retry_of → the failed one, which records retried_by). Retrying the same failed job again
// (retry_of → the failed one, which records retried_by). Retrying the same failed job again
// returns that same new job, so the call is idempotent.
// returns that same new job, so the call is idempotent.
//
//
+// Events: with an outbox (./events, openvibe-sdk createOutbox on this same database), created,
+// started, succeeded and failed are also announced to OpenVibe.Events as tools.job.*, each written
+// in the transaction that records the transition. Without one, nothing is announced.
+//
// No dependencies of its own: the app passes its better-sqlite3 handle and openvibe-contracts.
// No dependencies of its own: the app passes its better-sqlite3 handle and openvibe-contracts.
// ═══════════════════════════════════════════════════════════════
// ═══════════════════════════════════════════════════════════════
@@ -28,6 +32,7 @@ const fsp = require('fs/promises');
@@ -28,6 +32,7 @@ const fsp = require('fs/promises');
const path = require('path');
const path = require('path');
const crypto = require('crypto');
const crypto = require('crypto');
const { createStore, TERMINAL } = require('./store');
const { createStore, TERMINAL } = require('./store');
+const { createJobEvents } = require('./events');
const HOUR = 60 * 60 * 1000;
const HOUR = 60 * 60 * 1000;
const PRUNE_RECHECK_MS = 24 * HOUR; // an expired job the pruner could not finish is looked at again after this
const PRUNE_RECHECK_MS = 24 * HOUR; // an expired job the pruner could not finish is looked at again after this
@@ -71,6 +76,7 @@ const scrub = (msg) => String(msg || '').replace(/(?:\/[\w.-]+){2,}/g, '[file]')
@@ -71,6 +76,7 @@ const scrub = (msg) => String(msg || '').replace(/(?:\/[\w.-]+){2,}/g, '[file]')
* @param {number} [o.concurrency=2]
* @param {number} [o.concurrency=2]
* @param {number} [o.maxActivePerOwner=10]
* @param {number} [o.maxActivePerOwner=10]
* @param {object} [o.media] result store from ./media (null = keep results on local disk)
* @param {object} [o.media] result store from ./media (null = keep results on local disk)
+ * @param {object} [o.outbox] openvibe-sdk outbox on `db` (./events outboxFromEnv); null = no platform events
* @param {number} [o.pruneIntervalMs=300000]
* @param {number} [o.pruneIntervalMs=300000]
* @param {number} [o.progressThrottleMs=250]
* @param {number} [o.progressThrottleMs=250]
*/
*/
@@ -85,6 +91,7 @@ function createJobSystem(o) {
@@ -85,6 +91,7 @@ function createJobSystem(o) {
const throttleMs = o.progressThrottleMs == null ? 250 : o.progressThrottleMs;
const throttleMs = o.progressThrottleMs == null ? 250 : o.progressThrottleMs;
const root = path.resolve(o.dataDir, 'jobs');
const root = path.resolve(o.dataDir, 'jobs');
fs.mkdirSync(root, { recursive: true });
fs.mkdirSync(root, { recursive: true });
+ const announce = createJobEvents({ contracts, service: o.service, outbox: o.outbox || null, referenceCount: (id) => store.referenceCount(id), log });
const types = new Map();
const types = new Map();
const active = new Map(); // id → { ctrl, reason, last: { pct, msg, at } }
const active = new Map(); // id → { ctrl, reason, last: { pct, msg, at } }
@@ -234,6 +241,7 @@ function createJobSystem(o) {
@@ -234,6 +241,7 @@ function createJobSystem(o) {
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,
});
});
+ announce.created(store.get(id));
})();
})();
} catch (err) {
} catch (err) {
await fsp.rm(jobDir(id), { recursive: true, force: true });
await fsp.rm(jobDir(id), { recursive: true, force: true });
@@ -262,7 +270,19 @@ function createJobSystem(o) {
@@ -262,7 +270,19 @@ function createJobSystem(o) {
while (active.size < concurrency) {
while (active.size < concurrency) {
const [row] = store.nextQueued(1);
const [row] = store.nextQueued(1);
if (!row) break;
if (!row) break;
- if (!store.claim(row.id)) continue;
+ // The claim and its tools.job.started event commit together (or neither does).
+ let claimed;
+ try {
+ claimed = store.transaction(() => {
+ if (!store.claim(row.id)) return false;
+ announce.started(store.get(row.id));
+ return true;
+ })();
+ } catch (err) {
+ log.error(`[Jobs] could not start ${row.id}:`, err.message);
+ break; // it stays queued; the next kick tries again
+ }
+ if (!claimed) continue;
const entry = { ctrl: new AbortController(), reason: null, last: { pct: -1, msg: null, at: 0 } };
const entry = { ctrl: new AbortController(), reason: null, last: { pct: -1, msg: null, at: 0 } };
active.set(row.id, entry);
active.set(row.id, entry);
emit(row.id, 'job.running');
emit(row.id, 'job.running');
@@ -320,14 +340,20 @@ function createJobSystem(o) {
@@ -320,14 +340,20 @@ function createJobSystem(o) {
} finally {
} finally {
clearTimeout(timer);
clearTimeout(timer);
}
}
+ let recorded = false;
try {
try {
- store.transaction(() => store.finish(row.id, outcome))();
+ // The end state and its tools.job.succeeded|failed event commit together (or neither does).
+ store.transaction(() => { if (store.finish(row.id, outcome)) announce.finished(store.get(row.id)); })();
+ recorded = true;
emit(row.id, EVENT_FOR[outcome.state]);
emit(row.id, EVENT_FOR[outcome.state]);
+ } catch (err) {
+ // Nothing committed: the row is still 'running' with its inputs, which the next start() recovers.
+ log.error(`[Jobs] could not record the end of ${row.id}:`, err.message);
} finally {
} finally {
active.delete(row.id);
active.delete(row.id);
// A failed job keeps its inputs (until it expires) so it can be retried.
// A failed job keeps its inputs (until it expires) so it can be retried.
- if (outcome.state !== 'failed') fsp.rm(path.join(jobDir(row.id), 'in'), { recursive: true, force: true }).catch(() => {});
- if (outcome.state !== 'succeeded') fsp.rm(outDir, { recursive: true, force: true }).catch(() => {});
+ if (recorded && outcome.state !== 'failed') fsp.rm(path.join(jobDir(row.id), 'in'), { recursive: true, force: true }).catch(() => {});
+ if (recorded && outcome.state !== 'succeeded') fsp.rm(outDir, { recursive: true, force: true }).catch(() => {});
kick();
kick();
}
}
}
}
@@ -437,6 +463,7 @@ function createJobSystem(o) {
@@ -437,6 +463,7 @@ function createJobSystem(o) {
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,
env: row.env, retry_of: row.id,
});
});
+ announce.created(store.get(next));
})();
})();
} catch (err) {
} catch (err) {
if (moved) { try { fs.renameSync(newIn, oldIn); } catch { /* best effort */ } }
if (moved) { try { fs.renameSync(newIn, oldIn); } catch { /* best effort */ } }
@@ -478,6 +505,11 @@ function createJobSystem(o) {
@@ -478,6 +505,11 @@ function createJobSystem(o) {
}
}
// ── Boot recovery, pruning ─────────────────────────────────
// ── Boot recovery, pruning ─────────────────────────────────
+ /** A job the restart ended as failed: the row and its tools.job.failed event in one transaction. */
+ function failRecovered(id, outcome) {
+ store.transaction(() => { if (store.finish(id, outcome)) announce.finished(store.get(id)); })();
+ }
+
function recover() {
function recover() {
let requeued = 0, failed = 0;
let requeued = 0, failed = 0;
for (const row of store.running()) {
for (const row of store.running()) {
@@ -486,14 +518,14 @@ function createJobSystem(o) {
@@ -486,14 +518,14 @@ function createJobSystem(o) {
store.finish(row.id, { state: 'cancelled', error: problem(409, 'tools.job.cancelled', 'The job was cancelled') });
store.finish(row.id, { state: 'cancelled', error: problem(409, 'tools.job.cancelled', 'The job was cancelled') });
emit(row.id, 'job.cancelled');
emit(row.id, 'job.cancelled');
} else if (!def) {
} else if (!def) {
- store.finish(row.id, { state: 'failed', error: problem(500, 'tools.job.unknown_type', `This service no longer runs "${row.type}" jobs`) });
+ failRecovered(row.id, { state: 'failed', error: problem(500, 'tools.job.unknown_type', `This service no longer runs "${row.type}" jobs`) });
emit(row.id, 'job.failed'); failed++;
emit(row.id, 'job.failed'); failed++;
} else if (def.onRestart === 'requeue' && row.attempts < row.max_attempts) {
} else if (def.onRestart === 'requeue' && row.attempts < row.max_attempts) {
store.requeue(row.id);
store.requeue(row.id);
emit(row.id, 'job.queued'); requeued++;
emit(row.id, 'job.queued'); requeued++;
} else {
} else {
const retryable = def.onRestart === 'fail';
const retryable = def.onRestart === 'fail';
- store.finish(row.id, {
+ failRecovered(row.id, {
state: 'failed', retryable,
state: 'failed', retryable,
error: problem(503, 'tools.job.interrupted', retryable ? 'The service restarted while this job was running; submit it again' : `The service restarted during each of ${row.attempts} attempts`),
error: problem(503, 'tools.job.interrupted', retryable ? 'The service restarted while this job was running; submit it again' : `The service restarted during each of ${row.attempts} attempts`),
});
});
@@ -584,7 +616,8 @@ function createJobSystem(o) {
@@ -584,7 +616,8 @@ function createJobSystem(o) {
types: () => [...types.keys()],
types: () => [...types.keys()],
media,
media,
// running (after the spread) is the store's count of rows in 'running'; executing is this process's.
// running (after the spread) is the store's count of rows in 'running'; executing is this process's.
- stats: () => ({ running: active.size, concurrency, ...store.counts(), executing: active.size, results: media ? 'media' : 'local' }),
+ stats: () => ({ running: active.size, concurrency, ...store.counts(), executing: active.size, results: media ? 'media' : 'local', events: announce.status() }),
+ outbox: o.outbox || null,
/** True between start() and stop(): the worker picks up queued jobs. */
/** True between start() and stop(): the worker picks up queued jobs. */
isRunning: () => started && !stopped,
isRunning: () => started && !stopped,
store,
store,
new file mode 100644
new file mode 100644
@@ -0,0 +1,358 @@
@@ -0,0 +1,358 @@
+'use strict';
+// tools.job.* → OpenVibe.Events (roadmap Wave 11): every transition writes exactly one event to the
+// jobs database's outbox in the transaction that records it; a rolled-back transition writes none;
+// sandbox jobs and cancellations write none; payloads carry no input, file name, output data or
+// browser session and validate against openvibe-contracts; the relay posts to EVENTS_URL with the
+// tools service token; and without EVENTS_URL nothing is set up at all.
+const assert = require('assert');
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+const { dep } = require('./deps');
+
+const Database = dep('better-sqlite3');
+const contracts = dep('openvibe-contracts');
+const sdk = dep('openvibe-sdk');
+const jobs = require('../jobs');
+const { outboxFromEnv, payloadFor, ownerRef } = require('../jobs/events');
+
+const USER = 'usr_01JAAAAAAAAAAAAAAAAAAAAAAA';
+const APP = 'app_01JCCCCCCCCCCCCCCCCCCCCCCC';
+const SESSION = 'session:0123456789abcdef0123456789abcdef01234567';
+const SECRET_INPUT = 'secret-input-text-xyzzy';
+const SECRET_FILE = 'holiday-photo-of-alice.txt';
+const silent = { log() {}, warn() {}, error() {} };
+const sleep = (ms) => new Promise(r => setTimeout(r, ms));
+async function until(fn, what, ms = 5000) {
+ const t0 = Date.now();
+ for (;;) {
+ const v = await fn();
+ if (v) return v;
+ if (Date.now() - t0 > ms) throw new Error(`timed out waiting for ${what}`);
+ await sleep(10);
+ }
+}
+
+const gates = new Map();
+function gate(name) {
+ if (!gates.has(name)) { let open; const promise = new Promise(r => { open = r; }); gates.set(name, { promise, open }); }
+ return gates.get(name);
+}
+const flaky = new Set();
+function define(system) {
+ system.define({
+ type: 'test.upper', maxFiles: 1,
+ async run({ input, files, outDir, signal }) {
+ if (input.gate) await Promise.race([gate(input.gate).promise, new Promise((_, rej) => signal.addEventListener('abort', () => rej(new Error('aborted'))))]);
+ const text = files.length ? fs.readFileSync(files[0].path, 'utf8') : input.text || '';
+ const out = path.join(outDir, 'secret-output-name.txt');
+ fs.writeFileSync(out, text.toUpperCase());
+ return { files: [{ path: out, name: 'secret-output-name.txt', mime: 'text/plain' }], data: { echoed: input.text || null } };
+ },
+ });
+ // Its error names the input file (by name and by server path), which must not reach the network.
+ system.define({
+ type: 'test.boom', maxFiles: 1,
+ async run({ files }) { throw new Error(`Cannot read ${files[0].name} at ${files[0].path}`); },
+ });
+ system.define({
+ type: 'test.flaky', maxFiles: 1,
+ async run({ input, files, outDir }) {
+ if (flaky.has(input.tag)) throw Object.assign(new Error('upstream hiccup'), { retryable: true });
+ const out = path.join(outDir, 'f.txt');
+ fs.writeFileSync(out, fs.readFileSync(files[0].path));
+ return { files: [{ path: out, name: 'f.txt', mime: 'text/plain' }] };
+ },
+ });
+ system.define({
+ type: 'test.fragile', onRestart: 'fail',
+ async run({ input, signal }) {
+ await Promise.race([gate(input.gate).promise, new Promise((_, rej) => signal.addEventListener('abort', () => rej(new Error('aborted'))))]);
+ return { data: {} };
+ },
+ });
+}
+
+// ── A fake Network token endpoint + OpenVibe.Events, as a fetch ──
+function fakeNetwork() {
+ const calls = { token: [], publish: [] };
+ let seq = 0;
+ async function fetchImpl(url, init = {}) {
+ const u = new URL(String(url));
+ const reply = (status, body) => new Response(JSON.stringify(body), { status, headers: { 'content-type': 'application/json' } });
+ if (u.host === 'network.test' && u.pathname === '/oauth/token') {
+ const form = Object.fromEntries(new URLSearchParams(String(init.body)));
+ calls.token.push(form);
+ return reply(200, { access_token: 'events-token', token_type: 'Bearer', expires_in: 300, scope: form.scope });
+ }
+ if (u.host === 'events.test' && u.pathname === '/api/v1/events' && init.method === 'POST') {
+ const headers = new Headers(init.headers);
+ const body = JSON.parse(String(init.body));
+ const list = body.events || [body];
+ calls.publish.push({ authorization: headers.get('authorization'), events: list });
+ const results = list.map(e => ({ event_id: e.event_id, seq: ++seq, duplicate: false }));
+ return reply(201, body.events ? { results } : results[0]);
+ }
+ return reply(404, { code: 'not_found' });
+ }
+ return { fetch: fetchImpl, calls };
+}
+
+const ENV = { EVENTS_URL: 'http://events.test/', OV_OAUTH_CLIENT_SECRET: 'tools-secret', OV_NETWORK_INTERNAL_URL: 'http://network.test' };
+
+function satellite(dir, { wrap, concurrency = 1, net } = {}) {
+ fs.mkdirSync(dir, { recursive: true });
+ const db = new Database(path.join(dir, 'jobs.db'));
+ const { outbox, reason } = outboxFromEnv({ db, sdk, env: ENV, fetch: net.fetch, log: silent });
+ assert.ok(outbox, `outbox configured (${reason})`);
+ const system = jobs.createJobSystem({ db, contracts, service: 'img', dataDir: dir, concurrency, outbox: wrap ? wrap(outbox) : outbox, progressThrottleMs: 0, log: silent });
+ define(system);
+ system.start();
+ const envelopes = (jobId) => db.prepare('SELECT envelope FROM event_outbox ORDER BY id').all().map(r => JSON.parse(r.envelope))
+ .filter(e => !jobId || e.subject.id === jobId);
+ return { db, system, outbox, envelopes, close() { system.stop(); outbox.stop(); db.close(); } };
+}
+
+const types = (list) => list.map(e => e.event_type);
+function checkContracts(env) {
+ const v = contracts.validate(`${env.event_type}@1`, env.payload);
+ assert.ok(v.valid, `${env.event_type} payload validates: ${JSON.stringify(v.errors)}`);
+ const e = contracts.validate('events.event-envelope@1', env);
+ assert.ok(e.valid, `${env.event_type} envelope validates: ${JSON.stringify(e.errors)}`);
+ assert.strictEqual(env.source, 'tools');
+ assert.deepStrictEqual(env.subject, { type: 'job', id: env.payload.job_id });
+ assert.strictEqual(env.visibility, 'internal');
+ assert.strictEqual(env.payload.service, 'img');
+}
+function noLeaks(list, extra = []) {
+ const text = JSON.stringify(list);
+ for (const s of [SECRET_INPUT, SECRET_FILE, 'secret-output-name', 'session:', SESSION.slice(8), 'echoed', '/tmp', os.tmpdir(), ...extra]) {
+ assert.ok(!text.includes(s), `no "${s}" in published events`);
+ }
+ for (const env of list) {
+ for (const k of ['input', 'input_json', 'files', 'files_json', 'data', 'idempotency_key', 'request_hash']) {
+ assert.ok(!(k in env.payload), `${env.event_type} has no ${k}`);
+ }
+ }
+}
+const settled = (s, id, state) => until(() => { const r = s.system.get(id); return r && r.state === state && r; }, `${id} ${state}`);
+
+(async () => {
+ const root = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-job-events-'));
+ try {
+ // ── Inert without EVENTS_URL ─────────────────────────────
+ {
+ const db = new Database(':memory:');
+ assert.deepStrictEqual(outboxFromEnv({ db, sdk, env: {} }), { outbox: null, reason: 'EVENTS_URL is not set' });
+ assert.strictEqual(outboxFromEnv({ db, sdk, env: { EVENTS_URL: 'http://events.test' } }).outbox, null, 'no secret, no outbox');
+ assert.strictEqual(outboxFromEnv({ db, sdk, env: { ...ENV, EVENTS_PUBLISH: 'off' } }).outbox, null, 'EVENTS_PUBLISH=off');
+ assert.strictEqual(outboxFromEnv({ db, sdk: null, env: ENV }).outbox, null, 'no sdk, no outbox');
+ const dir = path.join(root, 'inert');
+ fs.mkdirSync(dir);
+ const system = jobs.createJobSystem({ db, contracts, service: 'img', dataDir: dir, log: silent });
+ define(system);
+ system.start();
+ const { job } = await system.submit({ owner: `user:${USER}`, type: 'test.upper', input: { text: 'x' } });
+ await until(() => system.get(job.id).state === 'succeeded', 'inert job');
+ assert.strictEqual(db.prepare("SELECT COUNT(*) AS n FROM sqlite_master WHERE name = 'event_outbox'").get().n, 0, 'no outbox table without EVENTS_URL');
+ assert.deepStrictEqual(system.stats().events, { enabled: false });
+ system.stop(); db.close();
+ }
+
+ const net = fakeNetwork();
+ // An outbox that can be told to fail right after it wrote a row (a failing write inside the transaction).
+ let failOn = null;
+ const wrap = (outbox) => ({
+ ...outbox,
+ enqueue(envelope, opts) {
+ const env = outbox.enqueue(envelope, opts);
+ if (failOn === envelope.event_type) throw new Error('disk I/O error (test)');
+ return env;
+ },
+ });
+ let s = satellite(path.join(root, 'a'), { net, wrap });
+
+ // ── A signed-in person's job: created, started, succeeded — once each ──
+ let { job } = await s.system.submit({ owner: `user:${USER}`, type: 'test.upper', input: { text: SECRET_INPUT }, files: [{ buffer: Buffer.from('hello'), name: SECRET_FILE, mime: 'text/plain' }] });
+ await settled(s, job.id, 'succeeded');
+ let list = s.envelopes(job.id);
+ assert.deepStrictEqual(types(list), ['tools.job.created', 'tools.job.started', 'tools.job.succeeded']);
+ list.forEach(checkContracts);
+ noLeaks(list);
+ assert.deepStrictEqual(list[0].actor, { type: 'user', id: USER }, 'created: the owner acts');
+ assert.deepStrictEqual(list[1].actor, { type: 'service', id: 'tools' });
+ assert.deepStrictEqual(list[2].actor, { type: 'service', id: 'tools' });
+ assert.deepStrictEqual(list.map(e => e.priority), ['low', 'low', 'important']);
+ assert.deepStrictEqual(list.map(e => e.payload.state), ['queued', 'running', 'succeeded']);
+ assert.deepStrictEqual(list[0].payload.owner, { type: 'user', id: USER });
+ assert.strictEqual(list[0].payload.retry_of, null);
+ assert.strictEqual(list[1].payload.attempts, 1);
+ const done = list[2].payload;
+ assert.ok(done.started_at && done.finished_at && done.expires_at);
+ assert.deepStrictEqual(Object.keys(done.result), ['files']);
+ assert.strictEqual(done.result.files.length, 1);
+ assert.deepStrictEqual(Object.keys(done.result.files[0]).sort(), ['index', 'media', 'mime', 'sha256', 'size', 'storage']);
+ assert.strictEqual(done.result.files[0].storage, 'local');
+ assert.strictEqual(done.result.files[0].media, null);
+ assert.strictEqual(done.result.files[0].size, 5);
+ assert.deepStrictEqual(s.system.stats().events.enabled, true);
+
+ // ── A failure whose message names the input file: failed once, name and path scrubbed ──
+ ({ job } = await s.system.submit({ owner: `user:${USER}`, type: 'test.boom', input: { text: SECRET_INPUT }, files: [{ buffer: Buffer.from('x'), name: SECRET_FILE, mime: 'text/plain' }] }));
+ await settled(s, job.id, 'failed');
+ list = s.envelopes(job.id);
+ assert.deepStrictEqual(types(list), ['tools.job.created', 'tools.job.started', 'tools.job.failed']);
+ list.forEach(checkContracts);
+ noLeaks(list);
+ const failed = list[2].payload;
+ assert.strictEqual(failed.error.code, 'tools.job.failed');
+ assert.strictEqual(failed.error.status, 422);
+ assert.match(failed.error.detail, /^Cannot read \[file\] at \[file\]$/);
+ assert.strictEqual(failed.retryable, false);
+ assert.deepStrictEqual(Object.keys(failed.error).sort(), ['code', 'detail', 'status']);
+
+ // ── A browser session's job: owner null, the service acts, the session never appears ──
+ ({ job } = await s.system.submit({ owner: SESSION, type: 'test.upper', input: { text: SECRET_INPUT } }));
+ await settled(s, job.id, 'succeeded');
+ list = s.envelopes(job.id);
+ assert.deepStrictEqual(types(list), ['tools.job.created', 'tools.job.started', 'tools.job.succeeded']);
+ list.forEach(checkContracts);
+ noLeaks(list);
+ for (const e of list) { assert.strictEqual(e.payload.owner, null); assert.deepStrictEqual(e.actor, { type: 'service', id: 'tools' }); }
+
+ // ── Principals: a service and a production app are subjects ──
+ ({ job } = await s.system.submit({ owner: 'svc:live', type: 'test.upper', input: {} }));
+ await settled(s, job.id, 'succeeded');
+ list = s.envelopes(job.id);
+ list.forEach(checkContracts);
+ assert.deepStrictEqual(list[0].actor, { type: 'service', id: 'live' });
+ assert.deepStrictEqual(list[0].payload.owner, { type: 'service', id: 'live' });
+ assert.deepStrictEqual(ownerRef(contracts, `app:${APP}`), { type: 'app', id: APP });
+ assert.strictEqual(ownerRef(contracts, 'app:nope'), null);
+
+ // ── Sandbox app jobs are never announced ─────────────────
+ const before = s.envelopes().length;
+ ({ job } = await s.system.submit({ owner: `app:${APP}`, type: 'test.upper', input: { text: 'sandbox' }, env: 'sandbox' }));
+ await settled(s, job.id, 'succeeded');
+ const sandboxFail = (await s.system.submit({ owner: `app:${APP}`, type: 'test.boom', files: [{ buffer: Buffer.from('x'), name: 'a.txt' }], env: 'sandbox' })).job;
+ await settled(s, sandboxFail.id, 'failed');
+ assert.strictEqual(s.envelopes(job.id).length + s.envelopes(sandboxFail.id).length, 0, 'sandbox jobs emit nothing');
+ assert.strictEqual(s.envelopes().length, before);
+
+ // ── Retry: a new created (retry_of) + its run; the failed job gets nothing new ──
+ flaky.add('t1');
+ const first = (await s.system.submit({ owner: `user:${USER}`, type: 'test.flaky', input: { tag: 't1' }, files: [{ buffer: Buffer.from('abc'), name: SECRET_FILE }] })).job;
+ await settled(s, first.id, 'failed');
+ assert.strictEqual(s.envelopes(first.id).at(-1).payload.retryable, true);
+ flaky.delete('t1');
+ const retried = s.system.retry(first.id).job;
+ await settled(s, retried.id, 'succeeded');
+ assert.deepStrictEqual(types(s.envelopes(first.id)), ['tools.job.created', 'tools.job.started', 'tools.job.failed'], 'the retried job gets no new event');
+ list = s.envelopes(retried.id);
+ assert.deepStrictEqual(types(list), ['tools.job.created', 'tools.job.started', 'tools.job.succeeded']);
+ list.forEach(checkContracts);
+ noLeaks(list);
+ assert.strictEqual(list[0].payload.retry_of, first.id);
+ assert.strictEqual(s.system.retry(first.id).replayed, true);
+ assert.strictEqual(s.envelopes(retried.id).length, 3, 'a replayed retry creates no event');
+
+ // ── An Idempotency-Key replay creates no job and no event ──
+ const k1 = (await s.system.submit({ owner: `user:${USER}`, type: 'test.upper', input: { text: 'k' }, idempotencyKey: 'key-00000001' })).job;
+ await settled(s, k1.id, 'succeeded');
+ const k2 = await s.system.submit({ owner: `user:${USER}`, type: 'test.upper', input: { text: 'k' }, idempotencyKey: 'key-00000001' });
+ assert.strictEqual(k2.replayed, true);
+ assert.strictEqual(s.envelopes(k1.id).length, 3);
+
+ // ── Cancelled jobs are not announced (neither queued nor running) ──
+ const blocker = (await s.system.submit({ owner: `user:${USER}`, type: 'test.upper', input: { gate: 'g1' } })).job;
+ await settled(s, blocker.id, 'running');
+ const queued = (await s.system.submit({ owner: `user:${USER}`, type: 'test.upper', input: { text: 'q' } })).job;
+ assert.strictEqual(s.system.cancel(queued.id).changed, true);
+ assert.deepStrictEqual(types(s.envelopes(queued.id)), ['tools.job.created']);
+ s.system.cancel(blocker.id);
+ await settled(s, blocker.id, 'cancelled');
+ assert.deepStrictEqual(types(s.envelopes(blocker.id)), ['tools.job.created', 'tools.job.started']);
+
+ // ── Rollbacks: a failed outbox write undoes the transition, and no event is left ──
+ const count = () => s.envelopes().length;
+ let n = count();
+ failOn = 'tools.job.created';
+ await assert.rejects(s.system.submit({ owner: `user:${USER}`, type: 'test.upper', input: { text: 'rollback' } }), /disk I\/O error/);
+ failOn = null;
+ assert.strictEqual(count(), n, 'rolled-back submit: no event');
+ assert.strictEqual(s.db.prepare("SELECT COUNT(*) AS n FROM tool_jobs WHERE input_json LIKE '%rollback%'").get().n, 0, 'rolled-back submit: no job');
+
+ // claim(): the job stays queued and nothing says it started.
+ failOn = 'tools.job.started';
+ const stuck = (await s.system.submit({ owner: `user:${USER}`, type: 'test.upper', input: { text: 'claim' } })).job;
+ await sleep(50);
+ assert.strictEqual(s.system.get(stuck.id).state, 'queued', 'rolled-back claim: still queued');
+ assert.strictEqual(s.system.get(stuck.id).attempts, 0);
+ assert.deepStrictEqual(types(s.envelopes(stuck.id)), ['tools.job.created'], 'rolled-back claim: no started event');
+ failOn = null;
+ // The next kick (any submit) starts it: exactly one started.
+ const kicker = (await s.system.submit({ owner: `user:${USER}`, type: 'test.upper', input: { text: 'kick' } })).job;
+ await settled(s, stuck.id, 'succeeded');
+ await settled(s, kicker.id, 'succeeded');
+ assert.deepStrictEqual(types(s.envelopes(stuck.id)), ['tools.job.created', 'tools.job.started', 'tools.job.succeeded']);
+
+ // finish(): the job stays running (recovered on the next start) and nothing says it ended.
+ failOn = 'tools.job.succeeded';
+ const unfinished = (await s.system.submit({ owner: `user:${USER}`, type: 'test.upper', input: { text: 'finish' } })).job;
+ await until(() => s.envelopes(unfinished.id).length === 2 && s.system.stats().executing === 0, 'finish attempt');
+ await sleep(30);
+ assert.strictEqual(s.system.get(unfinished.id).state, 'running', 'rolled-back finish: still running');
+ assert.deepStrictEqual(types(s.envelopes(unfinished.id)), ['tools.job.created', 'tools.job.started']);
+ failOn = null;
+
+ // ── Restart: the stuck job is re-queued (not announced) and started again (announced again);
+ // a running onRestart:'fail' job fails once with tools.job.interrupted ──
+ const fragile = (await s.system.submit({ owner: `user:${USER}`, type: 'test.fragile', input: { gate: 'never' } })).job;
+ await settled(s, fragile.id, 'running');
+ s.close();
+ s = satellite(path.join(root, 'a'), { net });
+ await settled(s, unfinished.id, 'succeeded');
+ assert.deepStrictEqual(types(s.envelopes(unfinished.id)), ['tools.job.created', 'tools.job.started', 'tools.job.started', 'tools.job.succeeded']);
+ assert.strictEqual(s.envelopes(unfinished.id)[2].payload.attempts, 2);
+ const fr = s.envelopes(fragile.id);
+ assert.deepStrictEqual(types(fr), ['tools.job.created', 'tools.job.started', 'tools.job.failed']);
+ fr.forEach(checkContracts);
+ assert.strictEqual(fr[2].payload.error.code, 'tools.job.interrupted');
+ assert.strictEqual(fr[2].payload.retryable, true);
+ assert.ok(fr[2].payload.started_at);
+
+ // ── Every event so far: valid, and never a secret ──
+ const all = s.envelopes();
+ all.forEach(checkContracts);
+ noLeaks(all);
+ assert.strictEqual(new Set(all.map(e => e.event_id)).size, all.length, 'unique event ids');
+
+ // ── The relay: POST EVENTS_URL/api/v1/events with the tools token for openvibe.events ──
+ assert.strictEqual(s.outbox.pending(), all.length);
+ const r = await s.outbox.flush();
+ assert.strictEqual(r.sent, all.length);
+ assert.strictEqual(s.outbox.pending(), 0);
+ const published = net.calls.publish.flatMap(c => c.events);
+ assert.deepStrictEqual(published.map(e => e.event_id), all.map(e => e.event_id), 'published in order, once');
+ assert.ok(net.calls.publish.every(c => c.authorization === 'Bearer events-token'));
+ assert.deepStrictEqual(net.calls.token[0], { grant_type: 'client_credentials', client_id: 'tools', client_secret: 'tools-secret', audience: 'openvibe.events', scope: 'events.event.publish' });
+ s.close();
+
+ // ── Payload builder: a result in Media is { media_id, role } only ──
+ const row = {
+ id: `job_${contracts.ids.ulid()}`, type: 'img.process', type_version: 1, owner: `user:${USER}`, state: 'succeeded', attempts: 1, max_attempts: 3,
+ created_at: Date.now() - 1000, started_at: Date.now() - 500, finished_at: Date.now(), expires_at: Date.now() + 3600000, retry_of: null, env: 'production',
+ files_json: JSON.stringify([{ name: SECRET_FILE }]),
+ result_json: JSON.stringify({ files: [{ name: 'secret-output-name.webp', mime: 'image/webp', size: 12, sha256: 'a'.repeat(64), storage: 'media', media: { media_id: `med_${contracts.ids.ulid()}`, role: 'output', namespace: 'tools', size_bytes: 12, content_hash: 'a'.repeat(64), mime_type: 'image/webp', status: 'ready' } }], data: { width: 3 } }),
+ };
+ const p = payloadFor('tools.job.succeeded', row, { contracts, service: 'img' });
+ assert.ok(contracts.validate('tools.job.succeeded@1', p).valid);
+ assert.deepStrictEqual(Object.keys(p.result.files[0].media).sort(), ['media_id', 'role']);
+ assert.strictEqual(payloadFor('tools.job.succeeded', row, { contracts, service: 'img', referenced: true }).expires_at, null, 'a referenced result does not expire');
+ assert.ok(!JSON.stringify(p).includes('secret-output-name'));
+
+ console.log('job events (tools.job.* outbox): all checks passed');
+ } finally {
+ fs.rmSync(root, { recursive: true, force: true });
+ }
+})().catch((err) => { console.error(err); process.exit(1); });
@@ -27,6 +27,7 @@ const { buildOptions, describe, defineJobs } = require('./process');
@@ -27,6 +27,7 @@ const { buildOptions, describe, defineJobs } = require('./process');
const { hostGuard, ownHost } = require('../../_shared/host-role');
const { hostGuard, ownHost } = require('../../_shared/host-role');
const jobsRuntime = require('../../_shared/jobs');
const jobsRuntime = require('../../_shared/jobs');
const contracts = require('openvibe-contracts');
const contracts = require('openvibe-contracts');
+const sdk = require('openvibe-sdk');
// ── Analytics ────────────────────────────────────────────────
// ── Analytics ────────────────────────────────────────────────
const Database = require('better-sqlite3');
const Database = require('better-sqlite3');
@@ -204,7 +205,7 @@ app.post('/api/process', burstLimiter, processLimiter, uploadSingle, async (req,
@@ -204,7 +205,7 @@ app.post('/api/process', burstLimiter, processLimiter, uploadSingle, async (req,
// followed over SSE (ffmpeg's own progress), cancellable (ffmpeg is killed), reattachable by id
// followed over SSE (ffmpeg's own progress), cancellable (ffmpeg is killed), reattachable by id
// after a reload or a restart (apps/_shared/jobs).
// after a reload or a restart (apps/_shared/jobs).
const jobs = jobsRuntime.setupJobs({
const jobs = jobsRuntime.setupJobs({
- app, service: 'audio', dataDir: path.resolve(__dirname, '..', config.dataDir), Database, contracts,
+ app, service: 'audio', dataDir: path.resolve(__dirname, '..', config.dataDir), Database, contracts, sdk,
getPublicKey: auth.getPublicKey, issuer: auth.ISSUER,
getPublicKey: auth.getPublicKey, issuer: auth.ISSUER,
define: defineJobs,
define: defineJobs,
receive: uploadSingle,
receive: uploadSingle,
@@ -25,6 +25,7 @@ const { buildOptions, defineJobs } = require('./process');
@@ -25,6 +25,7 @@ const { buildOptions, defineJobs } = require('./process');
const { hostGuard, ownHost } = require('../../_shared/host-role');
const { hostGuard, ownHost } = require('../../_shared/host-role');
const jobsRuntime = require('../../_shared/jobs');
const jobsRuntime = require('../../_shared/jobs');
const contracts = require('openvibe-contracts');
const contracts = require('openvibe-contracts');
+const sdk = require('openvibe-sdk');
// ── Analytics ────────────────────────────────────────────────
// ── Analytics ────────────────────────────────────────────────
const Database = require('better-sqlite3');
const Database = require('better-sqlite3');
@@ -251,7 +252,7 @@ app.post('/api/process/multi', burstLimiter, processLimiter, uploadMultiple, asy
@@ -251,7 +252,7 @@ app.post('/api/process/multi', burstLimiter, processLimiter, uploadMultiple, asy
// The same operations as /api/process and /api/process/multi, asynchronous and durable: accepted
// The same operations as /api/process and /api/process/multi, asynchronous and durable: accepted
// into data/jobs.db, followed over SSE, reattachable by id after a reload or a restart.
// into data/jobs.db, followed over SSE, reattachable by id after a reload or a restart.
const jobs = jobsRuntime.setupJobs({
const jobs = jobsRuntime.setupJobs({
- app, service: 'docs', dataDir: path.resolve(__dirname, '..', config.dataDir), Database, contracts,
+ app, service: 'docs', dataDir: path.resolve(__dirname, '..', config.dataDir), Database, contracts, sdk,
getPublicKey: auth.getPublicKey, issuer: auth.ISSUER,
getPublicKey: auth.getPublicKey, issuer: auth.ISSUER,
define: defineJobs,
define: defineJobs,
receive: uploadAny,
receive: uploadAny,
@@ -43,3 +43,8 @@ INTERNAL_API_KEY=change-me-to-a-random-64-char-string
@@ -43,3 +43,8 @@ INTERNAL_API_KEY=change-me-to-a-random-64-char-string
#TOOLS_MEDIA_NAMESPACE=tools
#TOOLS_MEDIA_NAMESPACE=tools
#OV_MEDIA_INTERNAL_URL=http://127.0.0.1:4100
#OV_MEDIA_INTERNAL_URL=http://127.0.0.1:4100
#OV_MEDIA_URL=https://openvibe.media
#OV_MEDIA_URL=https://openvibe.media
+# tools.job.created|started|succeeded|failed → OpenVibe.Events (off while unset). Needs OV_OAUTH_CLIENT_SECRET and
+# the Network grant [tools, events.event.publish, openvibe.events].
+#EVENTS_URL=http://127.0.0.1:4300
+#EVENTS_PUBLISH=off
+#EVENTS_RELAY_INTERVAL_MS=2000
@@ -25,6 +25,7 @@ const { buildOptions, describe, processBuffer, defineJobs } = require('./process
@@ -25,6 +25,7 @@ const { buildOptions, describe, processBuffer, defineJobs } = require('./process
const { hostGuard, ownHost } = require('../../_shared/host-role');
const { hostGuard, ownHost } = require('../../_shared/host-role');
const jobsRuntime = require('../../_shared/jobs');
const jobsRuntime = require('../../_shared/jobs');
const contracts = require('openvibe-contracts');
const contracts = require('openvibe-contracts');
+const sdk = require('openvibe-sdk');
// ── Analytics ────────────────────────────────────────────────
// ── Analytics ────────────────────────────────────────────────
const Database = require('better-sqlite3');
const Database = require('better-sqlite3');
@@ -201,7 +202,7 @@ app.post('/api/process/direct', burstLimiter, processLimiter, uploadSingle, asyn
@@ -201,7 +202,7 @@ app.post('/api/process/direct', burstLimiter, processLimiter, uploadSingle, asyn
// The same operation as /api/process, asynchronous and durable: accepted into data/jobs.db,
// The same operation as /api/process, asynchronous and durable: accepted into data/jobs.db,
// followed over SSE, reattachable by id after a reload or a restart (apps/_shared/jobs).
// followed over SSE, reattachable by id after a reload or a restart (apps/_shared/jobs).
const jobs = jobsRuntime.setupJobs({
const jobs = jobsRuntime.setupJobs({
- app, service: 'img', dataDir: path.resolve(__dirname, '..', config.dataDir), Database, contracts,
+ app, service: 'img', dataDir: path.resolve(__dirname, '..', config.dataDir), Database, contracts, sdk,
getPublicKey: auth.getPublicKey, issuer: auth.ISSUER,
getPublicKey: auth.getPublicKey, issuer: auth.ISSUER,
define: defineJobs,
define: defineJobs,
receive: uploadSingle,
receive: uploadSingle,
@@ -28,11 +28,11 @@ for app in apps/*/; do
@@ -28,11 +28,11 @@ for app in apps/*/; do
fi
fi
done
done
done
done
-# Apps that run jobs (apps/_shared/jobs, required by relative path) hand it their own better-sqlite3 and
-# openvibe-contracts: check both load under this Node (a native-module ABI mismatch shows up here, not in
-# a crash loop) and that the shared runtime itself loads.
+# Apps that run jobs (apps/_shared/jobs, required by relative path) hand it their own better-sqlite3,
+# openvibe-contracts and openvibe-sdk (the tools.job.* outbox relay): check they load under this Node (a
+# native-module ABI mismatch shows up here, not in a crash loop) and that the shared runtime itself loads.
for app in img audio docs; do
for app in img audio docs; do
- (cd "apps/$app" && node -e "const D=require('better-sqlite3'); new D(':memory:').close(); require('openvibe-contracts'); require('../_shared/jobs')") \
+ (cd "apps/$app" && node -e "const D=require('better-sqlite3'); new D(':memory:').close(); require('openvibe-contracts'); require('openvibe-sdk'); require('../_shared/jobs')") \
|| { echo "ABORT: apps/$app cannot load the jobs runtime; nothing restarted" >&2; exit 1; }
|| { echo "ABORT: apps/$app cannot load the jobs runtime; nothing restarted" >&2; exit 1; }
done
done
UNITS=$(systemctl list-unit-files 'openvibe-tools*' --no-legend | awk '{print $1}')
UNITS=$(systemctl list-unit-files 'openvibe-tools*' --no-legend | awk '{print $1}')
@@ -1,5 +1,5 @@
@@ -1,5 +1,5 @@
{
{
- "version": "0.2.0",
+ "version": "0.3.0",
"contractRanges": {
"contractRanges": {
"openvibe-contracts": ">=0.7.0 <1.0.0"
"openvibe-contracts": ">=0.7.0 <1.0.0"
},
},
@@ -17,10 +17,15 @@
@@ -17,10 +17,15 @@
"tools.job.read",
"tools.job.read",
"tools.job.cancel"
"tools.job.cancel"
],
],
- "eventsProduced": [],
+ "eventsProduced": [
+ "tools.job.created",
+ "tools.job.started",
+ "tools.job.succeeded",
+ "tools.job.failed"
+ ],
"eventsConsumed": [],
"eventsConsumed": [],
"namespacesOwned": [
"namespacesOwned": [
"tools.*"
"tools.*"
],
],
- "notes": "Wave 11: asynchronous jobs on the img, audio and docs satellites (POST/GET/DELETE /api/v1/jobs, SSE events). Job lifecycle is not yet published to OpenVibe.Events, so eventsProduced stays empty. Tools consumes media.object.upload and media.object.read (namespace tools) when TOOLS_JOB_RESULTS=media."
+ "notes": "Wave 11: asynchronous jobs on the img, audio and docs satellites (POST/GET/DELETE /api/v1/jobs, SSE events). Job lifecycle events tools.job.created|started|succeeded|failed (openvibe-contracts v0.30.1 payloads) go to OpenVibe.Events through an outbox in each satellite's jobs.db when EVENTS_URL is set. Tools consumes media.object.upload and media.object.read (namespace tools) when TOOLS_JOB_RESULTS=media."
}
}