openvibers/openvibe.tools Commits
Worker threads for sharp and pdf-lib (S5): apps/_shared/jobs/pool.js runs img's image operations and docs' pdf-lib tools in worker_threads, so a large merge or conversion never blocks the event loop (/api/health, the job routes and every other request keep answering). TOOLS_WORKERS (2, per app TOOLS_WORKERS_<APP>) run at once, shared by the jobs and the synchronous /api/process endpoints, which now submit to the pool and wait; TOOLS_WORKER_QUEUE (64) more wait, then 503 tools.busy with Retry-After. Each worker has a V8 heap limit (TOOLS_WORKER_MEMORY_MB, 512): a run that needs more fails with 413 tools.input.too_large and its worker is replaced. A run past its timeout (the job's own, or the descriptor's limits.timeoutMs on the sync endpoints: 504 tools.run.timeout), a cancelled job or a sync client that went away has its worker terminated at once, which pdf-lib and sharp never allowed before. Buffers move by transfer (never a pooled slab), tool errors keep status/code/extra/guardReason, idle workers stop after TOOLS_WORKER_IDLE_MS; qpdf and poppler tools stay on the main thread (their child processes have their own timeouts). /api/health shows workers. The older per-route limiters no longer count liveness and readiness probes (the guard's quota already skipped them). Tests: pool.test.js (off the event loop, size cap, queue and refusal, buffers, error fields, timeout and abort terminate, heap limit, crash, idle stop, env sizing) and docs pool.test.js on a real process (health stays under 250 ms during a 5000-page merge; a merge over a 32 MB heap fails 413 and the next job succeeds)
Changes
@@@ -108,7 +108,7 @@ routes answer on every host of the img, audio and docs satellites:|---|---|| `POST /api/v1/jobs` | Submit: multipart `type`, `input` (JSON text), `file` / `files`, or JSON `{ type, input }`. `Idempotency-Key` header (8–200 chars). → `202` + `Location`; a repeat with the same key and request → `200` + `Idempotent-Replayed: true` (same job); same key, different request → `409 tools.job.idempotency_conflict`. || `GET /api/v1/jobs/:id` | The job: `state` (`queued`, `running`, `succeeded`, `failed`, `cancelled`), `progress {percent, message}`, `attempts`, `result {files, data}`, `error` (problem+json), `retryable`, `expires_at`. |--| `DELETE /api/v1/jobs/:id` | Cancel: queued → `200` cancelled at once; running → `202` (signalled; ffmpeg is killed, sharp/pdf-lib work finishes and is discarded) and ends `cancelled`; finished → `409 tools.job.already_finished`. |++| `DELETE /api/v1/jobs/:id` | Cancel: queued → `200` cancelled at once; running → `202` (signalled; ffmpeg is killed, a sharp/pdf-lib worker thread is terminated) and ends `cancelled`; finished → `409 tools.job.already_finished`. || `GET /api/v1/jobs/:id/events` | SSE: `job.queued`, `job.running`, `job.progress`, `job.cancel_requested`, `job.succeeded`, `job.failed`, `job.cancelled`, each with an `id`. Reconnecting with `Last-Event-ID` (or `?last_event_id=`) replays only later events; a finished job with nothing newer answers `204`. || `GET /api/v1/jobs/:id/files/:n` | A result file (attachment); `?inline=1` for previews. || `POST /api/v1/jobs/:id/retry` | Retry a **failed** job: a new job with the same type, input and files (`retry_of` → the failed one, which gets `retried_by`) → `202` + `Location`. Idempotent: asking again returns that same retry with `200` + `Idempotent-Replayed: true`. Not failed → `409 tools.job.not_failed`; inputs gone → `410 tools.job.inputs_gone`. A failed job keeps its input files until it expires. |@@@ -129,6 +129,15 @@ reload reattaches.`ov_token` subject), an app/service principal (Network client-credentials token, audience `openvibe.tools`,capabilities `tools.job.create|read|cancel`), or else this browser's `ov_tools_jobs` cookie (only its hash isstored). Anyone else gets `404 tools.job.not_found`.++- **Worker threads.** sharp (img) and pdf-lib (docs) never run on the event loop: `apps/_shared/jobs/pool.js`++ runs them in `worker_threads`, `TOOLS_WORKERS` at once per satellite (default 2; `TOOLS_WORKERS_<APP>`), shared++ by the jobs and the synchronous `/api/process` endpoints (which submit to the pool and wait), with++ `TOOLS_WORKER_QUEUE` (64) more waiting before `503 tools.busy`. Each worker has a V8 heap limit++ (`TOOLS_WORKER_MEMORY_MB`, 512; `TOOLS_WORKER_MEMORY_MB_<APP>`): a run that needs more fails with++ `413 tools.input.too_large` and its worker is replaced. A run past its timeout (the job's, or the descriptor's++ `limits.timeoutMs` on the sync endpoints: `504 tools.run.timeout`), a cancelled job or a sync client that went++ away has its worker terminated at once. Idle workers stop after `TOOLS_WORKER_IDLE_MS` (60 s). qpdf and poppler++ tools stay on the main thread (their child processes have their own timeouts). `GET /api/health` shows `workers`.- **Bounded.** `TOOLS_JOBS_CONCURRENCY` jobs run at once per satellite (default 2; `TOOLS_JOBS_CONCURRENCY_<APP>`overrides), `TOOLS_JOBS_MAX_ACTIVE` unfinished jobs per owner (default 10, then `429`), and all browser sessionsfrom one address together `TOOLS_JOBS_MAX_ACTIVE_PER_ADDRESS` (default 3 × that; jobs keep only the guard's hashed@@@ -163,7 +172,8 @@ reload reattaches.with `events.event.publish` on audience `openvibe.events`. Without `EVENTS_URL` (or with `EVENTS_PUBLISH=off`, orno 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_WORKERS`, `TOOLS_WORKERS_<APP>`, `TOOLS_WORKER_MEMORY_MB`,++`TOOLS_WORKER_MEMORY_MB_<APP>`, `TOOLS_WORKER_QUEUE`, `TOOLS_WORKER_IDLE_MS`, `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`, `EVENTS_URL`, `EVENTS_PUBLISH`,`EVENTS_RELAY_INTERVAL_MS`.@@@ -193,8 +203,10 @@ One module used by the gateway and every satellite, driven by each tool's descri`RateLimit-Limit`/`-Remaining`/`-Reset`; a refusal is `429` problem+json `tools.quota.exceeded` with `Retry-After`and `quota_class`, `scope`, `tier`.- **Heavy work.** Synchronous `/api/process` (img, audio, docs) holds one of `TOOLS_SYNC_CONCURRENCY` slots-- (default 2; `TOOLS_SYNC_QUEUE` may wait, 8, for `TOOLS_SYNC_WAIT_MS`, 30 s), else `503 tools.busy`. Audio's sync-- path is killed at the descriptor's `timeoutMs` (`504 tools.run.timeout`) or when the client goes away.++ (default 2; `TOOLS_SYNC_QUEUE` may wait, 8, for `TOOLS_SYNC_WAIT_MS`, 30 s), else `503 tools.busy`. Every sync++ path is stopped at the descriptor's `timeoutMs` (`504 tools.run.timeout`) or when the client goes away: audio's++ ffmpeg is killed, img's and docs' worker thread is terminated (Jobs → Worker threads). Liveness and readiness++ probes (`GET /api/health`, `/api/ready`) are never rate limited.- **Uploads.** The bytes are checked against the descriptor's `files.accept` (`415 tools.file.unsupported_type`;a misleading extension is corrected); docs uploads go to disk. Tools whose descriptor says `auth.anonymous: false`(audio, PDF) need a session, a sign-in or a token (`401 tools.session_required`). Images: no input over@@@ -389,7 +389,7 @@ function createGuard(o = {}) {standardHeaders: headers,legacyHeaders: false,keyGenerator: (req) => { const c = caller(req); return signed(c) || c.tier === 'sandbox' ? c.key : c.ipKey; },-- skip: () => enforcing,++ skip: (req) => enforcing || isProbe(req), // liveness and readiness probes are never limitedmessage: { error: message },});}nnew file mode 100644@@@ -0,0 +1,32 @@++'use strict';++// One worker thread of ./pool.js: requires the app's worker module once, then runs one message at a++// time (op → module[op](payload)) and answers { id, ok, result } or { id, ok: false, error }. Result++// buffers are transferred back, never copied; an error keeps the fields the HTTP and job layers read.++const { parentPort, workerData } = require('worker_threads');++const { transferables } = require('./pool');++++const mod = require(workerData.module);++const FIELDS = ['name', 'message', 'status', 'code', 'expose', 'retryable', 'guardReason', 'detail', 'extra'];++++function plain(err) {++ const out = {};++ for (const k of FIELDS) {++ const v = err && err[k];++ if (v === undefined) continue;++ try { out[k] = k === 'extra' ? JSON.parse(JSON.stringify(v)) : v; } catch { /* not clonable: left out */ }++ }++ if (!out.message) out.message = String(err || 'The tool failed');++ return out;++}++++parentPort.on('message', async ({ id, op, payload }) => {++ try {++ const fn = mod[op];++ if (typeof fn !== 'function') throw Object.assign(new Error(`The worker module has no ${op}()`), { status: 500 });++ const result = await fn(payload);++ // Some native buffers (sharp's) cannot be detached: those are copied instead.++ try { parentPort.postMessage({ id, ok: true, result }, [...transferables(result)]); } catch { parentPort.postMessage({ id, ok: true, result }); }++ } catch (err) {++ parentPort.postMessage({ id, ok: false, error: plain(err) });++ }++});nnew file mode 100644@@@ -0,0 +1,269 @@++'use strict';++// ═══════════════════════════════════════════════════════════════++// A worker_threads pool for the CPU-bound tools (pdf-lib in docs, sharp in img): the work leaves the++// event loop, so /api/health, the job routes and every other request stay responsive while a large++// merge or conversion runs, and each run is bounded.++//++// const pool = createPool({ name: 'docs', module: require.resolve('./worker'), size: 2, memoryMb: 512 });++// const out = await pool.run('process', { tool, input, options }, { timeoutMs, signal, transfer: true });++//++// `module` exports async functions (op → fn(payload)); each worker thread requires it once. At most++// `size` runs at once (the concurrency cap every caller shares: jobs and the synchronous endpoints++// alike); more wait in order, up to `maxQueue`, beyond which run() refuses at once (503 tools.busy,++// Retry-After). Each worker has V8 resourceLimits (`memoryMb` old generation): a run that needs more++// fails with 413 tools.input.too_large and its worker is replaced; nothing else is affected. A run++// past its `timeoutMs`, or whose `signal` aborts (a cancelled job, a client that went away), has its++// worker terminated at once — pdf-lib and sharp cannot be interrupted any other way — and fails with++// 504 tools.run.timeout or tools.job.cancelled. Workers are started on demand and stopped after++// `idleMs` without work, so an idle satellite holds no extra heap.++//++// Buffers cross by copy, or by transfer when asked (transfer: true moves the payload's own++// ArrayBuffers; results always come back transferred) — never a pooled Buffer's shared slab. A++// Uint8Array in the result comes back as a Buffer.++//++// Errors a tool throws keep what the HTTP and job layers read: message, status, code, expose,++// retryable, guardReason, detail, extra.++//++// Environment (read by poolFromEnv): TOOLS_WORKERS (default 2), TOOLS_WORKERS_<APP>,++// TOOLS_WORKER_MEMORY_MB (default 512), TOOLS_WORKER_MEMORY_MB_<APP>, TOOLS_WORKER_QUEUE (default 64),++// TOOLS_WORKER_IDLE_MS (default 60000).++// No dependencies (node:worker_threads).++// ═══════════════════════════════════════════════════════════════++++const path = require('path');++const { Worker } = require('worker_threads');++++const WORKER = path.join(__dirname, 'pool-worker.js');++const ERROR_FIELDS = ['status', 'code', 'expose', 'retryable', 'guardReason', 'detail', 'extra'];++++function fail(message, fields) {++ return Object.assign(new Error(message), fields);++}++++/** The ArrayBuffers a value can hand over without touching anyone else's memory (depth-limited walk). */++function transferables(value, out = new Set(), depth = 0) {++ if (!value || depth > 4) return out;++ if (ArrayBuffer.isView(value)) {++ const ab = value.buffer;++ // A small Buffer shares Node's 8 KB pool slab with other Buffers: never detach that.++ if (ab instanceof ArrayBuffer && value.byteOffset === 0 && value.byteLength === ab.byteLength) out.add(ab);++ return out;++ }++ if (Array.isArray(value)) { for (const v of value) transferables(v, out, depth + 1); return out; }++ if (typeof value === 'object') for (const v of Object.values(value)) transferables(v, out, depth + 1);++ return out;++}++++/** Uint8Arrays (what a Buffer becomes after structured clone) back to Buffers, depth-limited. */++function revive(value, depth = 0) {++ if (!value || depth > 4) return value;++ if (value instanceof Uint8Array && !Buffer.isBuffer(value)) return Buffer.from(value.buffer, value.byteOffset, value.byteLength);++ if (Array.isArray(value)) { for (let i = 0; i < value.length; i++) value[i] = revive(value[i], depth + 1); return value; }++ if (typeof value === 'object' && Object.getPrototypeOf(value) === Object.prototype) {++ for (const k of Object.keys(value)) value[k] = revive(value[k], depth + 1);++ }++ return value;++}++++/**++ * @param {object} o++ * @param {string} o.module absolute path of the module the workers run++ * @param {string} [o.name] for errors and stats++ * @param {number} [o.size=2] runs at once++ * @param {number} [o.memoryMb=512] V8 old-generation limit per worker++ * @param {number} [o.maxQueue=64] runs waiting before run() refuses (0 = no waiting at all)++ * @param {number} [o.idleMs=60000] a worker without work for this long is stopped++ * @param {object} [o.env] the workers' environment (default: this process's)++ */++function createPool(o) {++ if (!o || !o.module) throw new TypeError('createPool needs the worker module');++ const name = o.name || path.basename(o.module, '.js');++ const size = Math.max(1, Number.isFinite(o.size) ? Math.floor(o.size) : 2);++ const memoryMb = Math.max(16, Number.isFinite(o.memoryMb) ? Math.floor(o.memoryMb) : 512);++ const maxQueue = Math.max(0, Number.isFinite(o.maxQueue) ? Math.floor(o.maxQueue) : 64);++ const idleMs = Number.isFinite(o.idleMs) ? o.idleMs : 60_000;++ const resourceLimits = { maxOldGenerationSizeMb: memoryMb, maxYoungGenerationSizeMb: Math.min(64, Math.max(8, Math.round(memoryMb / 8))) };++++ const workers = new Set(); // { worker, task, idleTimer, ready }++ const queue = []; // tasks waiting for a worker++ const counts = { completed: 0, failed: 0, killed: 0, out_of_memory: 0, timed_out: 0, cancelled: 0, refused: 0, started: 0 };++ let seq = 0, closed = false;++++ function spawn() {++ const w = { worker: null, task: null, idleTimer: null };++ w.worker = new Worker(WORKER, { workerData: { module: o.module }, resourceLimits, env: o.env || process.env, stdout: false, stderr: false });++ w.worker.unref(); // an idle worker never keeps the process alive; a busy one does (start())++ counts.started++;++ w.worker.on('message', (msg) => settle(w, msg));++ w.worker.on('error', (err) => {++ // An uncaught throw or the heap limit: the worker is gone either way.++ const t = w.task;++ w.task = null;++ if (!t) return;++ if (err && err.code === 'ERR_WORKER_OUT_OF_MEMORY') {++ counts.out_of_memory++;++ finish(t, fail(`This file needs more memory than a tool may use here (${memoryMb} MB); try a smaller file or fewer pages.`, { status: 413, code: 'tools.input.too_large', expose: true, guardReason: 'memory' }));++ } else {++ finish(t, fail('The tool stopped unexpectedly while working on this file.', { status: 500, code: 'tools.job.failed', cause: err }));++ }++ });++ w.worker.on('exit', () => {++ workers.delete(w);++ clearTimeout(w.idleTimer);++ if (w.task) { const t = w.task; w.task = null; finish(t, fail('The tool stopped unexpectedly while working on this file.', { status: 500, code: 'tools.job.failed' })); }++ dispatch();++ });++ workers.add(w);++ return w;++ }++++ function kill(w) {++ w.task = null;++ workers.delete(w);++ clearTimeout(w.idleTimer);++ counts.killed++;++ w.worker.terminate().catch(() => {});++ }++++ function finish(t, err, result) {++ if (t.done) return;++ t.done = true;++ clearTimeout(t.timer);++ if (t.signal && t.onAbort) t.signal.removeEventListener('abort', t.onAbort);++ if (err) { counts.failed++; t.reject(err); } else { counts.completed++; t.resolve(result); }++ setImmediate(dispatch);++ }++++ function settle(w, msg) {++ const t = w.task;++ if (!t || !msg || msg.id !== t.id) return;++ w.task = null;++ idle(w);++ if (msg.ok) return finish(t, null, revive(msg.result));++ const e = msg.error || {};++ const err = new Error(e.message || 'The tool failed');++ for (const k of ERROR_FIELDS) if (e[k] !== undefined) err[k] = e[k];++ if (e.name && e.name !== 'Error') err.name = e.name;++ return finish(t, err);++ }++++ function idle(w) {++ clearTimeout(w.idleTimer);++ w.worker.unref();++ if (idleMs > 0) {++ w.idleTimer = setTimeout(() => { if (!w.task && !queue.length) { workers.delete(w); w.worker.terminate().catch(() => {}); } }, idleMs);++ if (w.idleTimer.unref) w.idleTimer.unref();++ }++ }++++ function start(w, t) {++ clearTimeout(w.idleTimer);++ w.worker.ref();++ w.task = t;++ t.worker = w;++ if (t.timeoutMs > 0) {++ t.timer = setTimeout(() => {++ if (w.task !== t) return;++ counts.timed_out++;++ kill(w);++ finish(t, fail(`The tool ran longer than ${Math.round(t.timeoutMs / 1000)} s and was stopped.`, { status: 504, code: 'tools.run.timeout', expose: true, guardReason: 'timeout' }));++ }, t.timeoutMs);++ if (t.timer.unref) t.timer.unref();++ }++ try {++ try { w.worker.postMessage({ id: t.id, op: t.op, payload: t.payload }, t.transfer ? [...transferables(t.payload)] : []); } catch (err) {++ if (!t.transfer) throw err;++ w.worker.postMessage({ id: t.id, op: t.op, payload: t.payload }); // a buffer that cannot be detached: copied++ }++ } catch (err) {++ w.task = null;++ idle(w);++ finish(t, fail(`The input could not be handed to the tool (${err.message})`, { status: 500, code: 'tools.job.failed' }));++ }++ t.payload = null;++ }++++ function dispatch() {++ if (closed) return;++ while (queue.length) {++ let w = [...workers].find(x => !x.task);++ if (!w) {++ if (workers.size >= size) return;++ w = spawn();++ }++ start(w, queue.shift());++ }++ }++++ /**++ * Run `op` of the worker module with `payload`.++ * @param {string} op++ * @param {*} payload structured-clonable++ * @param {object} [opts]++ * @param {number} [opts.timeoutMs] stop the worker after this long (0 = no limit of its own)++ * @param {AbortSignal} [opts.signal] stop it when this aborts (cancel, client gone)++ * @param {boolean} [opts.transfer] move the payload's buffers instead of copying them++ */++ function run(op, payload, opts = {}) {++ return new Promise((resolve, reject) => {++ if (closed) return reject(fail('The worker pool is shutting down', { status: 503, code: 'tools.busy', retryAfter: 5 }));++ if (opts.signal && opts.signal.aborted) return reject(fail('cancelled', { status: 409, code: 'tools.job.cancelled' }));++ const busy = [...workers].filter(w => w.task).length;++ if (busy >= size && queue.length >= maxQueue) {++ counts.refused++;++ return reject(fail('The server is busy with other files right now; try again in a moment.', { status: 503, code: 'tools.busy', retryAfter: 10, expose: true, guardReason: 'busy.workers' }));++ }++ const t = { id: ++seq, op, payload, resolve, reject, timeoutMs: opts.timeoutMs || 0, signal: opts.signal || null, transfer: !!opts.transfer, done: false };++ if (t.signal) {++ t.onAbort = () => {++ if (t.done) return;++ counts.cancelled++;++ const i = queue.indexOf(t);++ if (i >= 0) queue.splice(i, 1);++ else if (t.worker && t.worker.task === t) kill(t.worker);++ finish(t, fail('cancelled', { status: 409, code: 'tools.job.cancelled' }));++ };++ t.signal.addEventListener('abort', t.onAbort, { once: true });++ }++ queue.push(t);++ dispatch();++ });++ }++++ function stats() {++ const busy = [...workers].filter(w => w.task).length;++ return { name, size, memory_mb: memoryMb, workers: workers.size, busy, queued: queue.length, max_queue: maxQueue, ...counts };++ }++++ /** Stop every worker; waiting runs fail with 503. */++ async function close() {++ closed = true;++ for (const t of queue.splice(0)) finish(t, fail('The worker pool is shutting down', { status: 503, code: 'tools.busy', retryAfter: 5 }));++ await Promise.all([...workers].map(w => { const t = w.task; w.task = null; if (t) finish(t, fail('The worker pool is shutting down', { status: 503, code: 'tools.busy', retryAfter: 5 })); clearTimeout(w.idleTimer); return w.worker.terminate().catch(() => {}); }));++ workers.clear();++ }++++ return { run, stats, close, size, memoryMb, maxQueue };++}++++function envInt(env, names, fallback) {++ for (const n of names) {++ const v = parseInt(env[n], 10);++ if (Number.isFinite(v) && v >= 0) return v;++ }++ return fallback;++}++++/** A pool sized from the environment (see the header) for `app` ('img', 'docs', 'gateway'). */++function poolFromEnv(app, module, { env = process.env, size, memoryMb, maxQueue, idleMs } = {}) {++ const A = String(app).toUpperCase();++ return createPool({++ name: app, module, env,++ size: envInt(env, [`TOOLS_WORKERS_${A}`, 'TOOLS_WORKERS'], size != null ? size : 2) || 1,++ memoryMb: envInt(env, [`TOOLS_WORKER_MEMORY_MB_${A}`, 'TOOLS_WORKER_MEMORY_MB'], memoryMb != null ? memoryMb : 512),++ maxQueue: envInt(env, [`TOOLS_WORKER_QUEUE_${A}`, 'TOOLS_WORKER_QUEUE'], maxQueue != null ? maxQueue : 64),++ idleMs: envInt(env, ['TOOLS_WORKER_IDLE_MS'], idleMs != null ? idleMs : 60_000),++ });++}++++module.exports = { createPool, poolFromEnv, transferables, revive };nnew file mode 100644@@@ -0,0 +1,16 @@++'use strict';++// A worker module for pool.test.js: each op exercises one thing the pool must survive.++const { threadId } = require('worker_threads');++++const spin = (ms) => { const end = Date.now() + ms; let x = 0; while (Date.now() < end) x += Math.random(); return x; };++++module.exports = {++ echo: async (payload) => ({ payload, threadId }),++ double: async ({ buf }) => ({ out: Buffer.concat([Buffer.from(buf), Buffer.from(buf)]), small: Buffer.from('ok'), threadId }),++ spin: async ({ ms }) => { spin(ms); return { threadId }; },++ // Keeps every chunk reachable until the heap limit ends the thread.++ hog: async () => { const keep = []; for (;;) keep.push(new Array(1e6).fill(Math.random())); },++ fail: async ({ status, code, extra }) => { throw Object.assign(new Error('refused on purpose'), { status, code, expose: true, extra, guardReason: 'test' }); },++ crash: async () => { setTimeout(() => { throw new Error('boom'); }, 1); return new Promise(() => {}); },++ env: async ({ name }) => ({ value: process.env[name] || null }),++};nnew file mode 100644@@@ -0,0 +1,106 @@++'use strict';++// The worker pool (apps/_shared/jobs/pool.js): work runs off the event loop, at most `size` at once++// with the rest queued in order and refused past maxQueue (503 tools.busy), buffers come back as++// Buffers, a tool's error keeps its status/code/extra, a run past its timeout or whose signal aborts++// has its worker terminated (504 tools.run.timeout, tools.job.cancelled), a run over the heap limit++// fails cleanly with 413 tools.input.too_large, a crashed worker fails its run, and after each of++// those the pool keeps working with fresh workers. Idle workers stop.++const assert = require('assert');++const path = require('path');++const { createPool, poolFromEnv, transferables, revive } = require('../jobs/pool');++++const MODULE = path.join(__dirname, 'fixtures', 'pool-module.js');++const rejects = (p) => p.then(() => { throw new Error('expected a rejection'); }, (err) => err);++++(async () => {++ const pool = createPool({ name: 'test', module: MODULE, size: 2, memoryMb: 48, maxQueue: 3, idleMs: 300 });++ try {++ // ── Runs, and buffers come back as Buffers ──++ const r = await pool.run('echo', { a: 1, list: [1, 2] });++ assert.deepStrictEqual(r.payload, { a: 1, list: [1, 2] });++ assert.ok(r.threadId > 0, 'ran in a worker thread');++ const src = Buffer.alloc(64 * 1024, 7);++ const d = await pool.run('double', { buf: src }, { transfer: true });++ assert.ok(Buffer.isBuffer(d.out) && d.out.length === 128 * 1024 && d.out[5] === 7, 'a Buffer back');++ assert.ok(Buffer.isBuffer(d.small) && d.small.toString() === 'ok', 'a small (pooled) Buffer back too');++ assert.strictEqual(src.length, 0, 'transfer: the input moved instead of being copied');++ const pooled = Buffer.from('tiny');++ assert.strictEqual(transferables({ pooled }).size, 0, 'a pooled Buffer is never detached');++ assert.ok(Buffer.isBuffer(revive({ x: new Uint8Array([1]) }).x));++++ // ── The event loop stays free while a worker is busy ──++ const t0 = Date.now();++ const busy = pool.run('spin', { ms: 600 });++ let ticks = 0;++ const ticker = setInterval(() => ticks++, 20);++ await busy;++ clearInterval(ticker);++ assert.ok(Date.now() - t0 >= 550);++ assert.ok(ticks >= 15, `timers kept firing during 600 ms of work (${ticks})`);++++ // ── At most `size` at once, the rest queued; past maxQueue refused ──++ const started = Date.now();++ const runs = [0, 1, 2, 3, 4].map(() => pool.run('spin', { ms: 200 }));++ const refused = await rejects(pool.run('spin', { ms: 1 }));++ assert.deepStrictEqual([refused.status, refused.code, refused.retryAfter], [503, 'tools.busy', 10], '2 running + 3 queued: the next is refused');++ const threads = await Promise.all(runs);++ const took = Date.now() - started;++ assert.ok(took >= 550, `five 200 ms runs two at a time take three rounds (${took} ms)`);++ assert.ok(new Set(threads.map(x => x.threadId)).size <= 2, 'never more than two workers');++ assert.strictEqual(pool.stats().refused, 1);++++ // ── A tool's error keeps its fields ──++ const e = await rejects(pool.run('fail', { status: 422, code: 'tools.pdf.wrong_password', extra: { a: 1 } }));++ assert.deepStrictEqual([e.message, e.status, e.code, e.expose, e.guardReason, e.extra], ['refused on purpose', 422, 'tools.pdf.wrong_password', true, 'test', { a: 1 }]);++++ // ── Timeout: the worker is terminated, the pool carries on ──++ const before = pool.stats().killed;++ const to = await rejects(pool.run('spin', { ms: 5000 }, { timeoutMs: 150 }));++ assert.deepStrictEqual([to.status, to.code, to.guardReason], [504, 'tools.run.timeout', 'timeout']);++ assert.strictEqual(pool.stats().killed, before + 1, 'its worker was terminated');++ assert.ok((await pool.run('echo', 1)).threadId > 0, 'a fresh worker takes the next run');++++ // ── Abort (a cancelled job, a client that left): running and queued ──++ const ac = new AbortController();++ const running = rejects(pool.run('spin', { ms: 5000 }, { signal: ac.signal }));++ const other = pool.run('spin', { ms: 5000 }, { timeoutMs: 100 }).catch(x => x);++ const queued = rejects(pool.run('spin', { ms: 5000 }, { signal: ac.signal }));++ await new Promise(r => setTimeout(r, 50));++ ac.abort();++ for (const x of await Promise.all([running, queued])) assert.deepStrictEqual([x.status, x.code], [409, 'tools.job.cancelled']);++ await other;++ const pre = new AbortController(); pre.abort();++ assert.strictEqual((await rejects(pool.run('echo', 1, { signal: pre.signal }))).code, 'tools.job.cancelled', 'already aborted: never starts');++++ // ── Over the heap limit: fails cleanly, the pool carries on ──++ const oom = await rejects(pool.run('hog', {}, { timeoutMs: 20000 }));++ assert.deepStrictEqual([oom.status, oom.code, oom.guardReason], [413, 'tools.input.too_large', 'memory'], oom.message);++ assert.match(oom.message, /48 MB/);++ assert.strictEqual(pool.stats().out_of_memory, 1);++ assert.ok((await pool.run('echo', 2)).threadId > 0, 'still working after a worker ran out of memory');++++ // ── A crash fails its run and nothing else ──++ const crash = await rejects(pool.run('crash', {}));++ assert.deepStrictEqual([crash.status, crash.code], [500, 'tools.job.failed']);++ assert.strictEqual((await rejects(pool.run('nope', {}))).status, 500, 'an unknown op is an error, not a hang');++ assert.deepStrictEqual((await pool.run('echo', 3)).payload, 3);++++ // ── Idle workers stop ──++ await new Promise(r => setTimeout(r, 700));++ assert.strictEqual(pool.stats().workers, 0, 'no worker kept after idleMs');++ assert.deepStrictEqual((await pool.run('echo', 4)).payload, 4, 'and one starts again on demand');++++ // ── Sizing from the environment ──++ const p2 = poolFromEnv('docs', MODULE, { env: { TOOLS_WORKERS: '3', TOOLS_WORKERS_DOCS: '1', TOOLS_WORKER_MEMORY_MB: '256', TOOLS_WORKER_QUEUE: '5', MARK: 'x' } });++ assert.deepStrictEqual([p2.size, p2.memoryMb, p2.maxQueue], [1, 256, 5], 'the per-app override wins');++ assert.deepStrictEqual(await p2.run('env', { name: 'MARK' }), { value: 'x' }, 'workers see the environment they were given');++ await p2.close();++ const p3 = poolFromEnv('img', MODULE, { env: {} });++ assert.deepStrictEqual([p3.size, p3.memoryMb, p3.maxQueue], [2, 512, 64], 'defaults');++ await p3.close();++ assert.strictEqual((await rejects(p3.run('echo', 1))).code, 'tools.busy', 'a closed pool refuses');++ } finally {++ await pool.close();++ }++ console.log('pool: off the event loop, size cap + queue + refusal, buffers, error fields, timeout/abort terminate, heap limit 413, crash, idle stop, env sizing: all checks passed');++})().catch((err) => { console.error(err); process.exit(1); });@@@ -84,7 +84,7 @@ function tool(id, extra = {}) {files: { ...o.files, maxBytes: MAX_BYTES },output: { kind: 'file', mime: o.mime, schema: RESULT },limits: { timeoutMs: TIMEOUT_MS, ...(op !== 'img2pdf' && { maxPages: o.maxPages || MAX_PAGES }) },-- // PDF work runs on the satellite's main thread today: a browser session, a person or a token.++ // Heavy work (a worker thread each, apps/_shared/jobs/pool.js): a browser session, a person or a token.auth: { anonymous: false, capability: 'tools.tool.run' },quotaClass: 'tools-job', cost: o.cost, egress: false,...(o.requires && { requires: o.requires }),@@@ -20,7 +20,8 @@ const { resolveContext, DOMAIN_MAP } = require('./domain-map');const { getTool, listTools, BINARIES } = require('./tools');const { uploadSingle, uploadMultiple, uploadAny } = require('./middleware/upload');const retention = require('./retention/manager');--const { buildOptions, defineJobs } = require('./process');++const { buildOptions, defineJobs, runTool } = require('./process');++const { poolFromEnv } = require('../../_shared/jobs/pool');const { hostGuard, ownHost } = require('../../_shared/host-role');const { createToolsApi, exceptRegistry } = require('../../_shared/tools/http');const { createLocalRegistry, requiresStatus } = require('../../_shared/tools/local');@@@ -54,6 +55,19 @@ const { apiLimiter, processLimiter, burstLimiter } = guard.legacyLimiters(rateLiconst hostTool = (req) => (guard.tool(req.ctx.toolId) ? req.ctx.toolId : (guard.toolForJob('docs.process', req.ctx.defaultOp || 'merge', null) || { id: null }).id);/** The tool a parsed request runs: its `tool` (operation) on this host. */const toolOf = (req) => { const op = String((req.body && req.body.tool) || req.ctx.defaultOp || ''); const d = op ? guard.toolForJob('docs.process', op, req.ctx.toolId) : null; return d ? d.id : hostTool(req); };++// ── Worker pool (apps/_shared/jobs/pool.js) ─────────────────++// pdf-lib runs in worker threads, never on the event loop: TOOLS_WORKERS (2) at once for the++// synchronous endpoints and the jobs together, each with a heap limit (TOOLS_WORKER_MEMORY_MB, 512)++// and terminated at its timeout, on cancel, or when the client goes away.++const pool = poolFromEnv('docs', path.join(__dirname, 'worker.js'));++/** The sync endpoints' bounds for a run: the tool's descriptor timeout, and the client leaving. */++function syncRun(req, res) {++ const ac = new AbortController();++ res.on('close', () => { if (!res.writableFinished) ac.abort(); });++ const d = guard.tool(toolOf(req));++ return { signal: ac.signal, timeoutMs: (d && d.limits && d.limits.timeoutMs) || 10 * 60 * 1000 };++}++/** Read the uploaded files (on disk) and delete them. */async function readUploads(files) {try { return await Promise.all(files.map(f => fsp.readFile(f.path))); } finally { for (const f of files) fs.unlink(f.path, () => {}); }@@@ -151,7 +165,10 @@ app.use((req, _res, next) => {/** A tool's error on the synchronous endpoints: 503 problem when a tool is not set up, else { error }. */function sendToolError(req, res, err, label) {if (guard.toolRefused(req, res, err, toolOf(req))) return undefined;-- if (err.status === 503) return contracts.http.sendProblem(res, 503, err.code || 'tools.unavailable', { detail: err.message, ctx: req.ov });++ if (err.status === 503) {++ if (err.retryAfter) res.set('Retry-After', String(err.retryAfter));++ return contracts.http.sendProblem(res, 503, err.code || 'tools.unavailable', { detail: err.message, ctx: req.ov });++ }if (!err.expose) console.error(`[${label}] Error:`, err.message);const status = Number.isInteger(err.status) && err.status >= 400 && err.status < 500 ? err.status : 422;return res.status(status).json({ error: err.message || 'Document processing failed', ...(err.code && { code: err.code }) });@@@ -160,7 +177,7 @@ function sendToolError(req, res, err, label) {// Health checkapp.get('/api/health', (_req, res) => {const stats = retention.getStats();-- res.json({ status: 'ok', service: 'openvibe-docs', version: '1.0.0', files: stats, jobs: jobs.stats() });++ res.json({ status: 'ok', service: 'openvibe-docs', version: '1.0.0', files: stats, jobs: jobs.stats(), workers: pool.stats() });});// Domain context (frontend calls this on load to get branding). It also starts this browser's session@@@ -193,7 +210,7 @@ app.post('/api/info', burstLimiter, processLimiter, guard.toolQuota(hostTool), utry {const tool = getTool('metadata');const [buffer] = await readUploads([req.file]);-- const result = await tool.handler(buffer, { mode: 'view' });++ const result = await runTool(pool, tool, buffer, { mode: 'view' }, syncRun(req, res));res.json({ success: true, ...result });} catch (err) {sendToolError(req, res, err, 'Info');@@@ -209,9 +226,9 @@ app.post('/api/process', burstLimiter, processLimiter, guard.toolQuota(hostTool)const tool = getTool(toolId);if (!tool) { fs.unlink(req.file.path, () => {}); return res.status(400).json({ error: `Unknown tool: ${toolId}` }); }-- // Same options and code path as the docs.process job++ // Same options and code path as the docs.process job (the worker pool, its cap and its limits)const [buffer] = await readUploads([req.file]);-- const result = await tool.handler(buffer, buildOptions(req.body, req.ctx, false));++ const result = await runTool(pool, tool, buffer, buildOptions(req.body, req.ctx, false), syncRun(req, res));// Some tools return view-only results (like metadata view)if (result.viewOnly) {@@@ -260,7 +277,7 @@ app.post('/api/process/multi', burstLimiter, processLimiter, guard.toolQuota(hos}const buffers = await readUploads(req.files);-- const result = await tool.handler(buffers, buildOptions(req.body, req.ctx, true));++ const result = await runTool(pool, tool, buffers, buildOptions(req.body, req.ctx, true), syncRun(req, res));// Save outputconst firstName = req.files[0]?.originalname || 'output';@@@ -296,7 +313,7 @@ app.post('/api/process/multi', burstLimiter, processLimiter, guard.toolQuota(hosconst jobs = jobsRuntime.setupJobs({app, service: 'docs', dataDir: path.resolve(__dirname, '..', config.dataDir), Database, contracts, sdk,getPublicKey: guard.keys.get, issuer: config.networkUrl, guard,-- define: defineJobs,++ define: (system) => defineJobs(system, pool),receive: uploadAny,limiters: [burstLimiter, processLimiter, guard.toolQuota(hostTool)],jobTool: (req, type, input) => { const op = String((input && input.tool) || req.ctx.defaultOp || ''); const d = op ? guard.toolForJob('docs.process', op, req.ctx.toolId) : null; return d ? d.id : null; },@@@ -382,6 +399,7 @@ function shutdown() {analytics.destroy();analyticsDb.close();jobs.close(); // running jobs stay 'running' in data/jobs.db; the next boot re-queues them++ pool.close();guard.close();server.close(() => process.exit(0));setTimeout(() => process.exit(1), 5000);@@@ -2,7 +2,9 @@// ═══════════════════════════════════════════════════════════════// Docs.OpenVibe — one document operation, shared by the synchronous endpoints--// (/api/process, /api/process/multi) and the docs.process job (see defineJobs).++// (/api/process, /api/process/multi, /api/info) and the docs.process job (see defineJobs). Both run++// it through runTool(): pdf-lib work goes to the worker pool (./worker.js), so the two share one++// concurrency cap, one heap limit and terminate-on-timeout or cancel.// ═══════════════════════════════════════════════════════════════const fs = require('fs');@@@ -17,6 +19,20 @@ const SINGLE_KEYS = ['format', 'quality', 'angle', 'pages', 'order', 'ranges', ''level', 'defaultFormat'];const MULTI_KEYS = ['order', 'pageSize', 'quality', 'format'];++/**++ * Run a tool: in the worker pool when it is pdf-lib work, on this thread when it shells out to qpdf++ * or poppler (their child processes have their own timeouts). Without a pool (tests), inline.++ * @param {object|null} pool apps/_shared/jobs/pool.js++ * @param {object} tool tools/index.js entry++ * @param {Buffer|Buffer[]} input++ * @param {object} options++ * @param {{ timeoutMs?: number, signal?: AbortSignal }} [opts]++ */++function runTool(pool, tool, input, options, opts = {}) {++ if (!pool || (tool.requires && tool.requires.length)) return tool.handler(input, options);++ return pool.run('process', { tool: tool.id, input, options }, { ...opts, transfer: true });++}++/** Tool options from a request body or a job input. */function buildOptions(src, ctx, multi) {const options = {};@@@ -43,7 +59,7 @@ function describe(toolId, result, fileCount) {}/** Job type docs.process: input { tool, …options } + one PDF, or several files for merge / img2pdf. */--function defineJobs(system) {++function defineJobs(system, pool = null) {const known = new Set(listTools().map(t => t.id));system.define({type: 'docs.process',@@@ -72,7 +88,8 @@ function defineJobs(system) {for (const f of files) buffers.push(await fsp.readFile(f.path));if (signal.aborted) throw new Error('cancelled');progress(20, 'Processing');-- const result = await tool.handler(tool.multiFile ? buffers : buffers[0], buildOptions(input, null, tool.multiFile));++ // Cancel and the job's timeout abort `signal`: a worker is terminated at once.++ const result = await runTool(pool, tool, tool.multiFile ? buffers : buffers[0], buildOptions(input, null, tool.multiFile), { signal });if (result.viewOnly) return { files: [], data: describe(toolId, result) };progress(90, 'Saving');const first = files[0].name || 'document';@@@ -84,4 +101,4 @@ function defineJobs(system) {});}--module.exports = { buildOptions, describe, defineJobs, SINGLE_KEYS, MULTI_KEYS };++module.exports = { buildOptions, describe, defineJobs, runTool, SINGLE_KEYS, MULTI_KEYS };nnew file mode 100644@@@ -0,0 +1,22 @@++'use strict';++// ═══════════════════════════════════════════════════════════════++// Docs.OpenVibe — what a worker thread of the pool runs (apps/_shared/jobs/pool.js).++//++// pdf-lib work (merge, split, compress, rotate, reorder, watermark, metadata, and img2pdf with sharp)++// happens here, off the event loop: the synchronous endpoints and the docs.process job both send it++// through the pool (process.js runTool), so they share its concurrency cap, its heap limit and its++// terminate-on-timeout. The tools that shell out to qpdf or poppler (protect, unlock, pdf2img) keep++// running on the main thread: their work is in a child process with its own timeout already.++// ═══════════════════════════════════════════════════════════════++++const { getTool } = require('./tools');++++module.exports = {++ /** { tool, input: Buffer | Buffer[], options } → the tool's result ({ buffer, ext, mime, … }). */++ async process({ tool, input, options }) {++ const t = getTool(tool);++ if (!t) throw Object.assign(new Error(`Unknown tool: ${tool}`), { status: 400, code: 'tools.job.invalid' });++ const buffers = Array.isArray(input) ? input.map(b => Buffer.from(b.buffer, b.byteOffset, b.byteLength)) : Buffer.from(input.buffer, input.byteOffset, input.byteLength);++ return t.handler(buffers, options || {});++ },++};nnew file mode 100644@@@ -0,0 +1,103 @@++'use strict';++// Docs.OpenVibe's worker pool on a real process (apps/_shared/jobs/pool.js, S5):++// • /api/health keeps answering in milliseconds while a large synchronous merge runs (pdf-lib is++// in a worker thread, not on the event loop), and the merge still comes back whole++// • a merge job that needs more heap than TOOLS_WORKER_MEMORY_MB fails cleanly: the job ends++// failed with 413 tools.input.too_large, the process stays up, and the next job succeeds++const assert = require('assert');++const fs = require('fs');++const os = require('os');++const path = require('path');++const { PDFDocument, StandardFonts } = require('pdf-lib');++const { startApp } = require('../../_shared/test/spawn');++++const sleep = (ms) => new Promise(r => setTimeout(r, ms));++++async function bigPdf(pages) {++ const doc = await PDFDocument.create();++ const font = await doc.embedFont(StandardFonts.Helvetica);++ for (let i = 0; i < pages; i++) doc.addPage([300, 300]).drawText(`Page ${i + 1} ${'x'.repeat(200)}`, { x: 10, y: 150, size: 8, font });++ return Buffer.from(await doc.save());++}++++async function finished(app, id, cookie) {++ for (let i = 0; i < 300; i++) {++ const job = await (await fetch(`${app.base}/api/v1/jobs/${id}`, { headers: { cookie } })).json();++ if (['succeeded', 'failed', 'cancelled'].includes(job.state)) return job;++ await sleep(100);++ }++ throw new Error(`job ${id} did not finish`);++}++++function mergeForm(files, job) {++ const fd = new FormData();++ if (job) { fd.append('type', 'docs.process'); fd.append('input', JSON.stringify({ tool: 'merge' })); } else fd.append('tool', 'merge');++ files.forEach((b, i) => fd.append('files', new Blob([b], { type: 'application/pdf' }), `part-${i}.pdf`));++ return fd;++}++++(async () => {++ const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-docs-pool-'));++ const envFor = (n) => ({ DATA_DIR: path.join(tmp, n), UPLOADS_DIR: path.join(tmp, n, 'uploads'), OUTPUT_DIR: path.join(tmp, n, 'output'), PDF_MAX_PAGES: '100000' });++ const big = await bigPdf(2500);++ const apps = [];++ try {++ // ── Health stays fast during a large merge ──++ const app = await startApp('docs', envFor('a'));++ apps.push(app);++ const ctx = await fetch(`${app.base}/api/context`);++ const cookie = ctx.headers.get('set-cookie').split(';')[0];++ const t0 = Date.now();++ const merging = fetch(`${app.base}/api/process/multi`, { method: 'POST', body: mergeForm([big, big]), headers: { cookie } }).then(async r => ({ status: r.status, body: await r.json(), at: Date.now() }));++ const latencies = [];++ let done = null;++ merging.then(r => { done = r; });++ await sleep(150); // the upload is in and pdf-lib is working++ while (!done) {++ const s = Date.now();++ const h = await fetch(`${app.base}/api/health`);++ assert.strictEqual(h.status, 200);++ const body = await h.json();++ latencies.push(Date.now() - s);++ if (!done && latencies.length === 3) assert.strictEqual(body.workers.busy, 1, 'the merge is in a worker thread');++ await sleep(40);++ }++ const took = done.at - t0;++ assert.strictEqual(done.status, 200, JSON.stringify(done.body));++ assert.strictEqual(done.body.pageCount, 5000, 'every page of both documents');++ assert.ok(took > 700, `the merge is large enough to measure (${took} ms)`);++ assert.ok(latencies.length >= 5, `health was asked during the merge (${latencies.length} times)`);++ const worst = Math.max(...latencies);++ assert.ok(worst < 250, `/api/health stayed fast during a ${took} ms merge (worst ${worst} ms)`);++ const health = await (await fetch(`${app.base}/api/health`)).json();++ assert.strictEqual(health.workers.completed, 1);++ assert.strictEqual(health.workers.busy, 0);++++ // ── Over the heap limit: the job fails cleanly and the next one works ──++ const small = await startApp('docs', { ...envFor('b'), TOOLS_WORKER_MEMORY_MB: '32' });++ apps.push(small);++ const c2 = (await fetch(`${small.base}/api/context`)).headers.get('set-cookie').split(';')[0];++ let r = await fetch(`${small.base}/api/v1/jobs`, { method: 'POST', body: mergeForm([big, big, big], true), headers: { cookie: c2 } });++ assert.strictEqual(r.status, 202, await r.clone().text());++ const oom = await finished(small, (await r.json()).id, c2);++ assert.strictEqual(oom.state, 'failed');++ assert.deepStrictEqual([oom.error.status, oom.error.code], [413, 'tools.input.too_large'], JSON.stringify(oom.error));++ assert.match(oom.error.detail, /32 MB/);++ assert.strictEqual(oom.retryable, false);++ assert.strictEqual((await fetch(`${small.base}/api/health`)).status, 200, 'the process is still up');++ const tiny = await bigPdf(2);++ r = await fetch(`${small.base}/api/v1/jobs`, { method: 'POST', body: mergeForm([tiny, tiny], true), headers: { cookie: c2 } });++ const next = await finished(small, (await r.json()).id, c2);++ assert.strictEqual(next.state, 'succeeded', JSON.stringify(next.error));++ assert.strictEqual(next.result.data.pageCount, 4);++ const stats = (await (await fetch(`${small.base}/api/health`)).json()).workers;++ assert.strictEqual(stats.out_of_memory, 1);++ // The synchronous endpoint says the same (413, the legacy { error, code } shape).++ r = await fetch(`${small.base}/api/process/multi`, { method: 'POST', body: mergeForm([big, big, big]), headers: { cookie: c2 } });++ assert.strictEqual(r.status, 413);++ assert.strictEqual((await r.json()).code, 'tools.input.too_large');++ } finally {++ for (const a of apps) await a.kill('SIGTERM');++ }++ console.log('docs pool: health fast during a 5000-page merge in a worker; heap limit fails the job (413) and the next one succeeds');++})().catch((err) => { console.error(err); process.exit(1); });@@@ -19,7 +19,8 @@ const { resolveContext, DOMAIN_MAP } = require('./domain-map');const { getTool, listTools } = require('./tools');const { uploadSingle } = require('./middleware/upload');const retention = require('./retention/manager');--const { buildOptions, describe, processBuffer, defineJobs } = require('./process');++const { buildOptions, describe, runProcess, defineJobs } = require('./process');++const { poolFromEnv } = require('../../_shared/jobs/pool');const codec = require('./tools/codec');const { hostGuard, ownHost } = require('../../_shared/host-role');const { createToolsApi, exceptRegistry } = require('../../_shared/tools/http');@@@ -54,6 +55,30 @@ const hostTool = (req) => (guard.tool(req.ctx.toolId) ? req.ctx.toolId : (guard./** The tool a parsed request runs: its `tool` (operation) on this host. */const toolOf = (req) => { const d = guard.toolForJob('img.process', String((req.body && req.body.tool) || req.ctx.defaultOp || 'convert'), req.ctx.toolId); return d ? d.id : hostTool(req); };++// ── Worker pool (apps/_shared/jobs/pool.js) ─────────────────++// sharp and the BMP/ICO codecs run in worker threads, never on the event loop: TOOLS_WORKERS (2) at++// once for the synchronous endpoints and the jobs together, each with a heap limit++// (TOOLS_WORKER_MEMORY_MB, 512) and terminated at its timeout, on cancel, or when the client goes away.++const pool = poolFromEnv('img', path.join(__dirname, 'worker.js'));++/** The sync endpoints' bounds for a run: the tool's descriptor timeout, and the client leaving. */++function syncRun(req, res) {++ const ac = new AbortController();++ res.on('close', () => { if (!res.writableFinished) ac.abort(); });++ const d = guard.tool(toolOf(req));++ return { signal: ac.signal, timeoutMs: (d && d.limits && d.limits.timeoutMs) || 5 * 60 * 1000, transfer: true };++}++/** A tool's failure on the synchronous endpoints: its own 503 (with Retry-After when busy), else { error }. */++function syncError(req, res, err, label) {++ if (guard.toolRefused(req, res, err, toolOf(req))) return;++ if (err.status === 503) {++ if (err.retryAfter) res.set('Retry-After', String(err.retryAfter));++ return contracts.http.sendProblem(res, 503, err.code, { detail: err.message, ctx: req.ov });++ }++ if (err.code === 'tools.job.cancelled') return; // the client went away++ console.error(`[${label}] Error:`, err.message);++ res.status(422).json({ error: err.message || 'Image processing failed' });++}++const app = express();// What this deploy runs (ADR-016); the shared navbar's release-watch polls it on every tool host.const release = require('openvibe-shared/release').createRelease({ service: 'tools', root: require('path').join(__dirname, '..', '..', '..') });@@@ -146,7 +171,7 @@ app.use((req, _res, next) => {// Health checkapp.get('/api/health', (_req, res) => {const stats = retention.getStats();-- res.json({ status: 'ok', service: 'openvibe-img', version: '1.0.0', files: stats, jobs: jobs.stats() });++ res.json({ status: 'ok', service: 'openvibe-img', version: '1.0.0', files: stats, jobs: jobs.stats(), workers: pool.stats() });});// Domain context (frontend calls this on load to get branding). It also starts this browser's session@@@ -185,8 +210,8 @@ app.post('/api/process', burstLimiter, processLimiter, guard.toolQuota(hostTool)return res.status(400).json({ error: `Unknown tool: ${toolId}` });}-- // Execute the tool (the same code the img.process job runs)-- const result = await processBuffer(req.file.buffer, toolId, buildOptions(req.body, req.ctx, req.file.mimetype));++ // Execute the tool (the same code, pool and limits the img.process job uses)++ const result = await runProcess(pool, req.file.buffer, toolId, buildOptions(req.body, req.ctx, req.file.mimetype), syncRun(req, res));// Save to retentionconst saved = retention.saveOutput(@@@ -199,10 +224,7 @@ app.post('/api/process', burstLimiter, processLimiter, guard.toolQuota(hostTool)res.json({ success: true, download: saved, ...describe(toolId, result) });} catch (err) {-- if (guard.toolRefused(req, res, err, toolOf(req))) return;-- if (err.status === 503) return contracts.http.sendProblem(res, 503, err.code, { detail: err.message, ctx: req.ov });-- console.error('[Process] Error:', err.message);-- res.status(422).json({ error: err.message || 'Image processing failed' });++ syncError(req, res, err, 'Process');}});@@@ -213,7 +235,7 @@ app.post('/api/process/direct', burstLimiter, processLimiter, guard.toolQuota(hoconst tool = getTool(toolId);if (!tool) return res.status(400).json({ error: `Unknown tool: ${toolId}` });-- const result = await processBuffer(req.file.buffer, toolId, buildOptions(req.body, req.ctx, req.file.mimetype));++ const result = await runProcess(pool, req.file.buffer, toolId, buildOptions(req.body, req.ctx, req.file.mimetype), syncRun(req, res));const baseName = path.basename(req.file.originalname, path.extname(req.file.originalname));res.set({@@@ -223,10 +245,7 @@ app.post('/api/process/direct', burstLimiter, processLimiter, guard.toolQuota(ho});res.send(result.buffer);} catch (err) {-- if (guard.toolRefused(req, res, err, toolOf(req))) return;-- if (err.status === 503) return contracts.http.sendProblem(res, 503, err.code, { detail: err.message, ctx: req.ov });-- console.error('[Process/Direct] Error:', err.message);-- res.status(422).json({ error: err.message || 'Image processing failed' });++ syncError(req, res, err, 'Process/Direct');}});@@@ -236,7 +255,7 @@ app.post('/api/process/direct', burstLimiter, processLimiter, guard.toolQuota(hoconst jobs = jobsRuntime.setupJobs({app, service: 'img', dataDir: path.resolve(__dirname, '..', config.dataDir), Database, contracts, sdk,getPublicKey: guard.keys.get, issuer: config.networkUrl, guard,-- define: defineJobs,++ define: (system) => defineJobs(system, pool),receive: uploadSingle,limiters: [burstLimiter, processLimiter, guard.toolQuota(hostTool)],jobTool: (req, type, input) => { const d = guard.toolForJob('img.process', String((input && input.tool) || req.ctx.defaultOp || 'convert'), req.ctx.toolId); return d ? d.id : null; },@@@ -320,6 +339,7 @@ function shutdown() {analytics.destroy();analyticsDb.close();jobs.close(); // running jobs stay 'running' in data/jobs.db; the next boot re-queues them++ pool.close();guard.close();server.close(() => process.exit(0));setTimeout(() => process.exit(1), 5000);@@@ -2,7 +2,9 @@// ═══════════════════════════════════════════════════════════════// Img.OpenVibe — one image operation, shared by the synchronous endpoints--// (/api/process, /api/process/direct) and the img.process job (server/jobs.js).++// (/api/process, /api/process/direct) and the img.process job (defineJobs). Both run it through++// runProcess(): the worker pool (./worker.js), so they share one concurrency cap, one heap limit and++// terminate-on-timeout or cancel.// ═══════════════════════════════════════════════════════════════const fs = require('fs');@@@ -41,8 +43,19 @@ async function processBuffer(buffer, toolId, options) {return tool.handler(buffer, options);}++/**++ * Run an operation in the worker pool (inline without one: tests).++ * @param {object|null} pool apps/_shared/jobs/pool.js++ * @param {{ timeoutMs?: number, signal?: AbortSignal }} [opts]++ */++function runProcess(pool, buffer, toolId, options, opts = {}) {++ if (!pool) return processBuffer(buffer, toolId, options);++ if (!getTool(toolId)) return Promise.reject(Object.assign(new Error(`Unknown tool: ${toolId}`), { code: 'tools.job.invalid', status: 400 }));++ return pool.run('process', { tool: toolId, buffer, options }, opts);++}++/** Job type img.process: input { tool, format?, quality?, width?, … } + one image file. */--function defineJobs(system) {++function defineJobs(system, pool = null) {const known = new Set(listTools().map(t => t.id));system.define({type: 'img.process',@@@ -64,7 +77,8 @@ function defineJobs(system) {const buffer = await fsp.readFile(src.path);if (signal.aborted) throw new Error('cancelled');progress(20, 'Processing');-- const result = await processBuffer(buffer, toolId, buildOptions(input, null, src.mime));++ // Cancel and the job's timeout abort `signal`: a worker is terminated at once.++ const result = await runProcess(pool, buffer, toolId, buildOptions(input, null, src.mime), { signal, transfer: true });progress(90, 'Saving');const base = path.basename(src.name || 'image', path.extname(src.name || '')) || 'openvibeimg-output';const name = `${base}.${result.ext}`;@@@ -75,4 +89,4 @@ function defineJobs(system) {});}--module.exports = { buildOptions, describe, processBuffer, defineJobs, OPTION_KEYS };++module.exports = { buildOptions, describe, processBuffer, runProcess, defineJobs, OPTION_KEYS };nnew file mode 100644@@@ -0,0 +1,17 @@++'use strict';++// ═══════════════════════════════════════════════════════════════++// Img.OpenVibe — what a worker thread of the pool runs (apps/_shared/jobs/pool.js).++//++// Every image operation (sharp, and the JavaScript BMP/ICO codecs) happens here, off the event loop:++// the synchronous endpoints and the img.process job both send it through the pool (process.js++// runProcess), so they share its concurrency cap, its heap limit and terminate-on-timeout or cancel.++// ═══════════════════════════════════════════════════════════════++++const { processBuffer } = require('./process');++++module.exports = {++ /** { tool, buffer, options } → { buffer, ext, mime, … } */++ async process({ tool, buffer, options }) {++ return processBuffer(Buffer.from(buffer.buffer, buffer.byteOffset, buffer.byteLength), tool, options || {});++ },++};
@@ -108,7 +108,7 @@ routes answer on every host of the img, audio and docs satellites:
@@ -108,7 +108,7 @@ routes answer on every host of the img, audio and docs satellites:
|---|---|
|---|---|
| `POST /api/v1/jobs` | Submit: multipart `type`, `input` (JSON text), `file` / `files`, or JSON `{ type, input }`. `Idempotency-Key` header (8–200 chars). → `202` + `Location`; a repeat with the same key and request → `200` + `Idempotent-Replayed: true` (same job); same key, different request → `409 tools.job.idempotency_conflict`. |
| `POST /api/v1/jobs` | Submit: multipart `type`, `input` (JSON text), `file` / `files`, or JSON `{ type, input }`. `Idempotency-Key` header (8–200 chars). → `202` + `Location`; a repeat with the same key and request → `200` + `Idempotent-Replayed: true` (same job); same key, different request → `409 tools.job.idempotency_conflict`. |
| `GET /api/v1/jobs/:id` | The job: `state` (`queued`, `running`, `succeeded`, `failed`, `cancelled`), `progress {percent, message}`, `attempts`, `result {files, data}`, `error` (problem+json), `retryable`, `expires_at`. |
| `GET /api/v1/jobs/:id` | The job: `state` (`queued`, `running`, `succeeded`, `failed`, `cancelled`), `progress {percent, message}`, `attempts`, `result {files, data}`, `error` (problem+json), `retryable`, `expires_at`. |
-| `DELETE /api/v1/jobs/:id` | Cancel: queued → `200` cancelled at once; running → `202` (signalled; ffmpeg is killed, sharp/pdf-lib work finishes and is discarded) and ends `cancelled`; finished → `409 tools.job.already_finished`. |
+| `DELETE /api/v1/jobs/:id` | Cancel: queued → `200` cancelled at once; running → `202` (signalled; ffmpeg is killed, a sharp/pdf-lib worker thread is terminated) and ends `cancelled`; finished → `409 tools.job.already_finished`. |
| `GET /api/v1/jobs/:id/events` | SSE: `job.queued`, `job.running`, `job.progress`, `job.cancel_requested`, `job.succeeded`, `job.failed`, `job.cancelled`, each with an `id`. Reconnecting with `Last-Event-ID` (or `?last_event_id=`) replays only later events; a finished job with nothing newer answers `204`. |
| `GET /api/v1/jobs/:id/events` | SSE: `job.queued`, `job.running`, `job.progress`, `job.cancel_requested`, `job.succeeded`, `job.failed`, `job.cancelled`, each with an `id`. Reconnecting with `Last-Event-ID` (or `?last_event_id=`) replays only later events; a finished job with nothing newer answers `204`. |
| `GET /api/v1/jobs/:id/files/:n` | A result file (attachment); `?inline=1` for previews. |
| `GET /api/v1/jobs/:id/files/:n` | A result file (attachment); `?inline=1` for previews. |
| `POST /api/v1/jobs/:id/retry` | Retry a **failed** job: a new job with the same type, input and files (`retry_of` → the failed one, which gets `retried_by`) → `202` + `Location`. Idempotent: asking again returns that same retry with `200` + `Idempotent-Replayed: true`. Not failed → `409 tools.job.not_failed`; inputs gone → `410 tools.job.inputs_gone`. A failed job keeps its input files until it expires. |
| `POST /api/v1/jobs/:id/retry` | Retry a **failed** job: a new job with the same type, input and files (`retry_of` → the failed one, which gets `retried_by`) → `202` + `Location`. Idempotent: asking again returns that same retry with `200` + `Idempotent-Replayed: true`. Not failed → `409 tools.job.not_failed`; inputs gone → `410 tools.job.inputs_gone`. A failed job keeps its input files until it expires. |
@@ -129,6 +129,15 @@ reload reattaches.
@@ -129,6 +129,15 @@ reload reattaches.
`ov_token` subject), an app/service principal (Network client-credentials token, audience `openvibe.tools`,
`ov_token` subject), an app/service principal (Network client-credentials token, audience `openvibe.tools`,
capabilities `tools.job.create|read|cancel`), or else this browser's `ov_tools_jobs` cookie (only its hash is
capabilities `tools.job.create|read|cancel`), or else this browser's `ov_tools_jobs` cookie (only its hash is
stored). Anyone else gets `404 tools.job.not_found`.
stored). Anyone else gets `404 tools.job.not_found`.
+- **Worker threads.** sharp (img) and pdf-lib (docs) never run on the event loop: `apps/_shared/jobs/pool.js`
+ runs them in `worker_threads`, `TOOLS_WORKERS` at once per satellite (default 2; `TOOLS_WORKERS_<APP>`), shared
+ by the jobs and the synchronous `/api/process` endpoints (which submit to the pool and wait), with
+ `TOOLS_WORKER_QUEUE` (64) more waiting before `503 tools.busy`. Each worker has a V8 heap limit
+ (`TOOLS_WORKER_MEMORY_MB`, 512; `TOOLS_WORKER_MEMORY_MB_<APP>`): a run that needs more fails with
+ `413 tools.input.too_large` and its worker is replaced. A run past its timeout (the job's, or the descriptor's
+ `limits.timeoutMs` on the sync endpoints: `504 tools.run.timeout`), a cancelled job or a sync client that went
+ away has its worker terminated at once. Idle workers stop after `TOOLS_WORKER_IDLE_MS` (60 s). qpdf and poppler
+ tools stay on the main thread (their child processes have their own timeouts). `GET /api/health` shows `workers`.
- **Bounded.** `TOOLS_JOBS_CONCURRENCY` jobs run at once per satellite (default 2; `TOOLS_JOBS_CONCURRENCY_<APP>`
- **Bounded.** `TOOLS_JOBS_CONCURRENCY` jobs run at once per satellite (default 2; `TOOLS_JOBS_CONCURRENCY_<APP>`
overrides), `TOOLS_JOBS_MAX_ACTIVE` unfinished jobs per owner (default 10, then `429`), and all browser sessions
overrides), `TOOLS_JOBS_MAX_ACTIVE` unfinished jobs per owner (default 10, then `429`), and all browser sessions
from one address together `TOOLS_JOBS_MAX_ACTIVE_PER_ADDRESS` (default 3 × that; jobs keep only the guard's hashed
from one address together `TOOLS_JOBS_MAX_ACTIVE_PER_ADDRESS` (default 3 × that; jobs keep only the guard's hashed
@@ -163,7 +172,8 @@ reload reattaches.
@@ -163,7 +172,8 @@ reload reattaches.
with `events.event.publish` on audience `openvibe.events`. Without `EVENTS_URL` (or with `EVENTS_PUBLISH=off`, or
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).
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_WORKERS`, `TOOLS_WORKERS_<APP>`, `TOOLS_WORKER_MEMORY_MB`,
+`TOOLS_WORKER_MEMORY_MB_<APP>`, `TOOLS_WORKER_QUEUE`, `TOOLS_WORKER_IDLE_MS`, `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`, `EVENTS_URL`, `EVENTS_PUBLISH`,
`OV_NETWORK_INTERNAL_URL`, `OV_OAUTH_CLIENT_ID`, `OV_OAUTH_CLIENT_SECRET`, `EVENTS_URL`, `EVENTS_PUBLISH`,
`EVENTS_RELAY_INTERVAL_MS`.
`EVENTS_RELAY_INTERVAL_MS`.
@@ -193,8 +203,10 @@ One module used by the gateway and every satellite, driven by each tool's descri
@@ -193,8 +203,10 @@ One module used by the gateway and every satellite, driven by each tool's descri
`RateLimit-Limit`/`-Remaining`/`-Reset`; a refusal is `429` problem+json `tools.quota.exceeded` with `Retry-After`
`RateLimit-Limit`/`-Remaining`/`-Reset`; a refusal is `429` problem+json `tools.quota.exceeded` with `Retry-After`
and `quota_class`, `scope`, `tier`.
and `quota_class`, `scope`, `tier`.
- **Heavy work.** Synchronous `/api/process` (img, audio, docs) holds one of `TOOLS_SYNC_CONCURRENCY` slots
- **Heavy work.** Synchronous `/api/process` (img, audio, docs) holds one of `TOOLS_SYNC_CONCURRENCY` slots
- (default 2; `TOOLS_SYNC_QUEUE` may wait, 8, for `TOOLS_SYNC_WAIT_MS`, 30 s), else `503 tools.busy`. Audio's sync
- path is killed at the descriptor's `timeoutMs` (`504 tools.run.timeout`) or when the client goes away.
+ (default 2; `TOOLS_SYNC_QUEUE` may wait, 8, for `TOOLS_SYNC_WAIT_MS`, 30 s), else `503 tools.busy`. Every sync
+ path is stopped at the descriptor's `timeoutMs` (`504 tools.run.timeout`) or when the client goes away: audio's
+ ffmpeg is killed, img's and docs' worker thread is terminated (Jobs → Worker threads). Liveness and readiness
+ probes (`GET /api/health`, `/api/ready`) are never rate limited.
- **Uploads.** The bytes are checked against the descriptor's `files.accept` (`415 tools.file.unsupported_type`;
- **Uploads.** The bytes are checked against the descriptor's `files.accept` (`415 tools.file.unsupported_type`;
a misleading extension is corrected); docs uploads go to disk. Tools whose descriptor says `auth.anonymous: false`
a misleading extension is corrected); docs uploads go to disk. Tools whose descriptor says `auth.anonymous: false`
(audio, PDF) need a session, a sign-in or a token (`401 tools.session_required`). Images: no input over
(audio, PDF) need a session, a sign-in or a token (`401 tools.session_required`). Images: no input over
@@ -389,7 +389,7 @@ function createGuard(o = {}) {
@@ -389,7 +389,7 @@ function createGuard(o = {}) {
standardHeaders: headers,
standardHeaders: headers,
legacyHeaders: false,
legacyHeaders: false,
keyGenerator: (req) => { const c = caller(req); return signed(c) || c.tier === 'sandbox' ? c.key : c.ipKey; },
keyGenerator: (req) => { const c = caller(req); return signed(c) || c.tier === 'sandbox' ? c.key : c.ipKey; },
- skip: () => enforcing,
+ skip: (req) => enforcing || isProbe(req), // liveness and readiness probes are never limited
message: { error: message },
message: { error: message },
});
});
}
}
new file mode 100644
new file mode 100644
@@ -0,0 +1,32 @@
@@ -0,0 +1,32 @@
+'use strict';
+// One worker thread of ./pool.js: requires the app's worker module once, then runs one message at a
+// time (op → module[op](payload)) and answers { id, ok, result } or { id, ok: false, error }. Result
+// buffers are transferred back, never copied; an error keeps the fields the HTTP and job layers read.
+const { parentPort, workerData } = require('worker_threads');
+const { transferables } = require('./pool');
+
+const mod = require(workerData.module);
+const FIELDS = ['name', 'message', 'status', 'code', 'expose', 'retryable', 'guardReason', 'detail', 'extra'];
+
+function plain(err) {
+ const out = {};
+ for (const k of FIELDS) {
+ const v = err && err[k];
+ if (v === undefined) continue;
+ try { out[k] = k === 'extra' ? JSON.parse(JSON.stringify(v)) : v; } catch { /* not clonable: left out */ }
+ }
+ if (!out.message) out.message = String(err || 'The tool failed');
+ return out;
+}
+
+parentPort.on('message', async ({ id, op, payload }) => {
+ try {
+ const fn = mod[op];
+ if (typeof fn !== 'function') throw Object.assign(new Error(`The worker module has no ${op}()`), { status: 500 });
+ const result = await fn(payload);
+ // Some native buffers (sharp's) cannot be detached: those are copied instead.
+ try { parentPort.postMessage({ id, ok: true, result }, [...transferables(result)]); } catch { parentPort.postMessage({ id, ok: true, result }); }
+ } catch (err) {
+ parentPort.postMessage({ id, ok: false, error: plain(err) });
+ }
+});
new file mode 100644
new file mode 100644
@@ -0,0 +1,269 @@
@@ -0,0 +1,269 @@
+'use strict';
+// ═══════════════════════════════════════════════════════════════
+// A worker_threads pool for the CPU-bound tools (pdf-lib in docs, sharp in img): the work leaves the
+// event loop, so /api/health, the job routes and every other request stay responsive while a large
+// merge or conversion runs, and each run is bounded.
+//
+// const pool = createPool({ name: 'docs', module: require.resolve('./worker'), size: 2, memoryMb: 512 });
+// const out = await pool.run('process', { tool, input, options }, { timeoutMs, signal, transfer: true });
+//
+// `module` exports async functions (op → fn(payload)); each worker thread requires it once. At most
+// `size` runs at once (the concurrency cap every caller shares: jobs and the synchronous endpoints
+// alike); more wait in order, up to `maxQueue`, beyond which run() refuses at once (503 tools.busy,
+// Retry-After). Each worker has V8 resourceLimits (`memoryMb` old generation): a run that needs more
+// fails with 413 tools.input.too_large and its worker is replaced; nothing else is affected. A run
+// past its `timeoutMs`, or whose `signal` aborts (a cancelled job, a client that went away), has its
+// worker terminated at once — pdf-lib and sharp cannot be interrupted any other way — and fails with
+// 504 tools.run.timeout or tools.job.cancelled. Workers are started on demand and stopped after
+// `idleMs` without work, so an idle satellite holds no extra heap.
+//
+// Buffers cross by copy, or by transfer when asked (transfer: true moves the payload's own
+// ArrayBuffers; results always come back transferred) — never a pooled Buffer's shared slab. A
+// Uint8Array in the result comes back as a Buffer.
+//
+// Errors a tool throws keep what the HTTP and job layers read: message, status, code, expose,
+// retryable, guardReason, detail, extra.
+//
+// Environment (read by poolFromEnv): TOOLS_WORKERS (default 2), TOOLS_WORKERS_<APP>,
+// TOOLS_WORKER_MEMORY_MB (default 512), TOOLS_WORKER_MEMORY_MB_<APP>, TOOLS_WORKER_QUEUE (default 64),
+// TOOLS_WORKER_IDLE_MS (default 60000).
+// No dependencies (node:worker_threads).
+// ═══════════════════════════════════════════════════════════════
+
+const path = require('path');
+const { Worker } = require('worker_threads');
+
+const WORKER = path.join(__dirname, 'pool-worker.js');
+const ERROR_FIELDS = ['status', 'code', 'expose', 'retryable', 'guardReason', 'detail', 'extra'];
+
+function fail(message, fields) {
+ return Object.assign(new Error(message), fields);
+}
+
+/** The ArrayBuffers a value can hand over without touching anyone else's memory (depth-limited walk). */
+function transferables(value, out = new Set(), depth = 0) {
+ if (!value || depth > 4) return out;
+ if (ArrayBuffer.isView(value)) {
+ const ab = value.buffer;
+ // A small Buffer shares Node's 8 KB pool slab with other Buffers: never detach that.
+ if (ab instanceof ArrayBuffer && value.byteOffset === 0 && value.byteLength === ab.byteLength) out.add(ab);
+ return out;
+ }
+ if (Array.isArray(value)) { for (const v of value) transferables(v, out, depth + 1); return out; }
+ if (typeof value === 'object') for (const v of Object.values(value)) transferables(v, out, depth + 1);
+ return out;
+}
+
+/** Uint8Arrays (what a Buffer becomes after structured clone) back to Buffers, depth-limited. */
+function revive(value, depth = 0) {
+ if (!value || depth > 4) return value;
+ if (value instanceof Uint8Array && !Buffer.isBuffer(value)) return Buffer.from(value.buffer, value.byteOffset, value.byteLength);
+ if (Array.isArray(value)) { for (let i = 0; i < value.length; i++) value[i] = revive(value[i], depth + 1); return value; }
+ if (typeof value === 'object' && Object.getPrototypeOf(value) === Object.prototype) {
+ for (const k of Object.keys(value)) value[k] = revive(value[k], depth + 1);
+ }
+ return value;
+}
+
+/**
+ * @param {object} o
+ * @param {string} o.module absolute path of the module the workers run
+ * @param {string} [o.name] for errors and stats
+ * @param {number} [o.size=2] runs at once
+ * @param {number} [o.memoryMb=512] V8 old-generation limit per worker
+ * @param {number} [o.maxQueue=64] runs waiting before run() refuses (0 = no waiting at all)
+ * @param {number} [o.idleMs=60000] a worker without work for this long is stopped
+ * @param {object} [o.env] the workers' environment (default: this process's)
+ */
+function createPool(o) {
+ if (!o || !o.module) throw new TypeError('createPool needs the worker module');
+ const name = o.name || path.basename(o.module, '.js');
+ const size = Math.max(1, Number.isFinite(o.size) ? Math.floor(o.size) : 2);
+ const memoryMb = Math.max(16, Number.isFinite(o.memoryMb) ? Math.floor(o.memoryMb) : 512);
+ const maxQueue = Math.max(0, Number.isFinite(o.maxQueue) ? Math.floor(o.maxQueue) : 64);
+ const idleMs = Number.isFinite(o.idleMs) ? o.idleMs : 60_000;
+ const resourceLimits = { maxOldGenerationSizeMb: memoryMb, maxYoungGenerationSizeMb: Math.min(64, Math.max(8, Math.round(memoryMb / 8))) };
+
+ const workers = new Set(); // { worker, task, idleTimer, ready }
+ const queue = []; // tasks waiting for a worker
+ const counts = { completed: 0, failed: 0, killed: 0, out_of_memory: 0, timed_out: 0, cancelled: 0, refused: 0, started: 0 };
+ let seq = 0, closed = false;
+
+ function spawn() {
+ const w = { worker: null, task: null, idleTimer: null };
+ w.worker = new Worker(WORKER, { workerData: { module: o.module }, resourceLimits, env: o.env || process.env, stdout: false, stderr: false });
+ w.worker.unref(); // an idle worker never keeps the process alive; a busy one does (start())
+ counts.started++;
+ w.worker.on('message', (msg) => settle(w, msg));
+ w.worker.on('error', (err) => {
+ // An uncaught throw or the heap limit: the worker is gone either way.
+ const t = w.task;
+ w.task = null;
+ if (!t) return;
+ if (err && err.code === 'ERR_WORKER_OUT_OF_MEMORY') {
+ counts.out_of_memory++;
+ finish(t, fail(`This file needs more memory than a tool may use here (${memoryMb} MB); try a smaller file or fewer pages.`, { status: 413, code: 'tools.input.too_large', expose: true, guardReason: 'memory' }));
+ } else {
+ finish(t, fail('The tool stopped unexpectedly while working on this file.', { status: 500, code: 'tools.job.failed', cause: err }));
+ }
+ });
+ w.worker.on('exit', () => {
+ workers.delete(w);
+ clearTimeout(w.idleTimer);
+ if (w.task) { const t = w.task; w.task = null; finish(t, fail('The tool stopped unexpectedly while working on this file.', { status: 500, code: 'tools.job.failed' })); }
+ dispatch();
+ });
+ workers.add(w);
+ return w;
+ }
+
+ function kill(w) {
+ w.task = null;
+ workers.delete(w);
+ clearTimeout(w.idleTimer);
+ counts.killed++;
+ w.worker.terminate().catch(() => {});
+ }
+
+ function finish(t, err, result) {
+ if (t.done) return;
+ t.done = true;
+ clearTimeout(t.timer);
+ if (t.signal && t.onAbort) t.signal.removeEventListener('abort', t.onAbort);
+ if (err) { counts.failed++; t.reject(err); } else { counts.completed++; t.resolve(result); }
+ setImmediate(dispatch);
+ }
+
+ function settle(w, msg) {
+ const t = w.task;
+ if (!t || !msg || msg.id !== t.id) return;
+ w.task = null;
+ idle(w);
+ if (msg.ok) return finish(t, null, revive(msg.result));
+ const e = msg.error || {};
+ const err = new Error(e.message || 'The tool failed');
+ for (const k of ERROR_FIELDS) if (e[k] !== undefined) err[k] = e[k];
+ if (e.name && e.name !== 'Error') err.name = e.name;
+ return finish(t, err);
+ }
+
+ function idle(w) {
+ clearTimeout(w.idleTimer);
+ w.worker.unref();
+ if (idleMs > 0) {
+ w.idleTimer = setTimeout(() => { if (!w.task && !queue.length) { workers.delete(w); w.worker.terminate().catch(() => {}); } }, idleMs);
+ if (w.idleTimer.unref) w.idleTimer.unref();
+ }
+ }
+
+ function start(w, t) {
+ clearTimeout(w.idleTimer);
+ w.worker.ref();
+ w.task = t;
+ t.worker = w;
+ if (t.timeoutMs > 0) {
+ t.timer = setTimeout(() => {
+ if (w.task !== t) return;
+ counts.timed_out++;
+ kill(w);
+ finish(t, fail(`The tool ran longer than ${Math.round(t.timeoutMs / 1000)} s and was stopped.`, { status: 504, code: 'tools.run.timeout', expose: true, guardReason: 'timeout' }));
+ }, t.timeoutMs);
+ if (t.timer.unref) t.timer.unref();
+ }
+ try {
+ try { w.worker.postMessage({ id: t.id, op: t.op, payload: t.payload }, t.transfer ? [...transferables(t.payload)] : []); } catch (err) {
+ if (!t.transfer) throw err;
+ w.worker.postMessage({ id: t.id, op: t.op, payload: t.payload }); // a buffer that cannot be detached: copied
+ }
+ } catch (err) {
+ w.task = null;
+ idle(w);
+ finish(t, fail(`The input could not be handed to the tool (${err.message})`, { status: 500, code: 'tools.job.failed' }));
+ }
+ t.payload = null;
+ }
+
+ function dispatch() {
+ if (closed) return;
+ while (queue.length) {
+ let w = [...workers].find(x => !x.task);
+ if (!w) {
+ if (workers.size >= size) return;
+ w = spawn();
+ }
+ start(w, queue.shift());
+ }
+ }
+
+ /**
+ * Run `op` of the worker module with `payload`.
+ * @param {string} op
+ * @param {*} payload structured-clonable
+ * @param {object} [opts]
+ * @param {number} [opts.timeoutMs] stop the worker after this long (0 = no limit of its own)
+ * @param {AbortSignal} [opts.signal] stop it when this aborts (cancel, client gone)
+ * @param {boolean} [opts.transfer] move the payload's buffers instead of copying them
+ */
+ function run(op, payload, opts = {}) {
+ return new Promise((resolve, reject) => {
+ if (closed) return reject(fail('The worker pool is shutting down', { status: 503, code: 'tools.busy', retryAfter: 5 }));
+ if (opts.signal && opts.signal.aborted) return reject(fail('cancelled', { status: 409, code: 'tools.job.cancelled' }));
+ const busy = [...workers].filter(w => w.task).length;
+ if (busy >= size && queue.length >= maxQueue) {
+ counts.refused++;
+ return reject(fail('The server is busy with other files right now; try again in a moment.', { status: 503, code: 'tools.busy', retryAfter: 10, expose: true, guardReason: 'busy.workers' }));
+ }
+ const t = { id: ++seq, op, payload, resolve, reject, timeoutMs: opts.timeoutMs || 0, signal: opts.signal || null, transfer: !!opts.transfer, done: false };
+ if (t.signal) {
+ t.onAbort = () => {
+ if (t.done) return;
+ counts.cancelled++;
+ const i = queue.indexOf(t);
+ if (i >= 0) queue.splice(i, 1);
+ else if (t.worker && t.worker.task === t) kill(t.worker);
+ finish(t, fail('cancelled', { status: 409, code: 'tools.job.cancelled' }));
+ };
+ t.signal.addEventListener('abort', t.onAbort, { once: true });
+ }
+ queue.push(t);
+ dispatch();
+ });
+ }
+
+ function stats() {
+ const busy = [...workers].filter(w => w.task).length;
+ return { name, size, memory_mb: memoryMb, workers: workers.size, busy, queued: queue.length, max_queue: maxQueue, ...counts };
+ }
+
+ /** Stop every worker; waiting runs fail with 503. */
+ async function close() {
+ closed = true;
+ for (const t of queue.splice(0)) finish(t, fail('The worker pool is shutting down', { status: 503, code: 'tools.busy', retryAfter: 5 }));
+ await Promise.all([...workers].map(w => { const t = w.task; w.task = null; if (t) finish(t, fail('The worker pool is shutting down', { status: 503, code: 'tools.busy', retryAfter: 5 })); clearTimeout(w.idleTimer); return w.worker.terminate().catch(() => {}); }));
+ workers.clear();
+ }
+
+ return { run, stats, close, size, memoryMb, maxQueue };
+}
+
+function envInt(env, names, fallback) {
+ for (const n of names) {
+ const v = parseInt(env[n], 10);
+ if (Number.isFinite(v) && v >= 0) return v;
+ }
+ return fallback;
+}
+
+/** A pool sized from the environment (see the header) for `app` ('img', 'docs', 'gateway'). */
+function poolFromEnv(app, module, { env = process.env, size, memoryMb, maxQueue, idleMs } = {}) {
+ const A = String(app).toUpperCase();
+ return createPool({
+ name: app, module, env,
+ size: envInt(env, [`TOOLS_WORKERS_${A}`, 'TOOLS_WORKERS'], size != null ? size : 2) || 1,
+ memoryMb: envInt(env, [`TOOLS_WORKER_MEMORY_MB_${A}`, 'TOOLS_WORKER_MEMORY_MB'], memoryMb != null ? memoryMb : 512),
+ maxQueue: envInt(env, [`TOOLS_WORKER_QUEUE_${A}`, 'TOOLS_WORKER_QUEUE'], maxQueue != null ? maxQueue : 64),
+ idleMs: envInt(env, ['TOOLS_WORKER_IDLE_MS'], idleMs != null ? idleMs : 60_000),
+ });
+}
+
+module.exports = { createPool, poolFromEnv, transferables, revive };
new file mode 100644
new file mode 100644
@@ -0,0 +1,16 @@
@@ -0,0 +1,16 @@
+'use strict';
+// A worker module for pool.test.js: each op exercises one thing the pool must survive.
+const { threadId } = require('worker_threads');
+
+const spin = (ms) => { const end = Date.now() + ms; let x = 0; while (Date.now() < end) x += Math.random(); return x; };
+
+module.exports = {
+ echo: async (payload) => ({ payload, threadId }),
+ double: async ({ buf }) => ({ out: Buffer.concat([Buffer.from(buf), Buffer.from(buf)]), small: Buffer.from('ok'), threadId }),
+ spin: async ({ ms }) => { spin(ms); return { threadId }; },
+ // Keeps every chunk reachable until the heap limit ends the thread.
+ hog: async () => { const keep = []; for (;;) keep.push(new Array(1e6).fill(Math.random())); },
+ fail: async ({ status, code, extra }) => { throw Object.assign(new Error('refused on purpose'), { status, code, expose: true, extra, guardReason: 'test' }); },
+ crash: async () => { setTimeout(() => { throw new Error('boom'); }, 1); return new Promise(() => {}); },
+ env: async ({ name }) => ({ value: process.env[name] || null }),
+};
new file mode 100644
new file mode 100644
@@ -0,0 +1,106 @@
@@ -0,0 +1,106 @@
+'use strict';
+// The worker pool (apps/_shared/jobs/pool.js): work runs off the event loop, at most `size` at once
+// with the rest queued in order and refused past maxQueue (503 tools.busy), buffers come back as
+// Buffers, a tool's error keeps its status/code/extra, a run past its timeout or whose signal aborts
+// has its worker terminated (504 tools.run.timeout, tools.job.cancelled), a run over the heap limit
+// fails cleanly with 413 tools.input.too_large, a crashed worker fails its run, and after each of
+// those the pool keeps working with fresh workers. Idle workers stop.
+const assert = require('assert');
+const path = require('path');
+const { createPool, poolFromEnv, transferables, revive } = require('../jobs/pool');
+
+const MODULE = path.join(__dirname, 'fixtures', 'pool-module.js');
+const rejects = (p) => p.then(() => { throw new Error('expected a rejection'); }, (err) => err);
+
+(async () => {
+ const pool = createPool({ name: 'test', module: MODULE, size: 2, memoryMb: 48, maxQueue: 3, idleMs: 300 });
+ try {
+ // ── Runs, and buffers come back as Buffers ──
+ const r = await pool.run('echo', { a: 1, list: [1, 2] });
+ assert.deepStrictEqual(r.payload, { a: 1, list: [1, 2] });
+ assert.ok(r.threadId > 0, 'ran in a worker thread');
+ const src = Buffer.alloc(64 * 1024, 7);
+ const d = await pool.run('double', { buf: src }, { transfer: true });
+ assert.ok(Buffer.isBuffer(d.out) && d.out.length === 128 * 1024 && d.out[5] === 7, 'a Buffer back');
+ assert.ok(Buffer.isBuffer(d.small) && d.small.toString() === 'ok', 'a small (pooled) Buffer back too');
+ assert.strictEqual(src.length, 0, 'transfer: the input moved instead of being copied');
+ const pooled = Buffer.from('tiny');
+ assert.strictEqual(transferables({ pooled }).size, 0, 'a pooled Buffer is never detached');
+ assert.ok(Buffer.isBuffer(revive({ x: new Uint8Array([1]) }).x));
+
+ // ── The event loop stays free while a worker is busy ──
+ const t0 = Date.now();
+ const busy = pool.run('spin', { ms: 600 });
+ let ticks = 0;
+ const ticker = setInterval(() => ticks++, 20);
+ await busy;
+ clearInterval(ticker);
+ assert.ok(Date.now() - t0 >= 550);
+ assert.ok(ticks >= 15, `timers kept firing during 600 ms of work (${ticks})`);
+
+ // ── At most `size` at once, the rest queued; past maxQueue refused ──
+ const started = Date.now();
+ const runs = [0, 1, 2, 3, 4].map(() => pool.run('spin', { ms: 200 }));
+ const refused = await rejects(pool.run('spin', { ms: 1 }));
+ assert.deepStrictEqual([refused.status, refused.code, refused.retryAfter], [503, 'tools.busy', 10], '2 running + 3 queued: the next is refused');
+ const threads = await Promise.all(runs);
+ const took = Date.now() - started;
+ assert.ok(took >= 550, `five 200 ms runs two at a time take three rounds (${took} ms)`);
+ assert.ok(new Set(threads.map(x => x.threadId)).size <= 2, 'never more than two workers');
+ assert.strictEqual(pool.stats().refused, 1);
+
+ // ── A tool's error keeps its fields ──
+ const e = await rejects(pool.run('fail', { status: 422, code: 'tools.pdf.wrong_password', extra: { a: 1 } }));
+ assert.deepStrictEqual([e.message, e.status, e.code, e.expose, e.guardReason, e.extra], ['refused on purpose', 422, 'tools.pdf.wrong_password', true, 'test', { a: 1 }]);
+
+ // ── Timeout: the worker is terminated, the pool carries on ──
+ const before = pool.stats().killed;
+ const to = await rejects(pool.run('spin', { ms: 5000 }, { timeoutMs: 150 }));
+ assert.deepStrictEqual([to.status, to.code, to.guardReason], [504, 'tools.run.timeout', 'timeout']);
+ assert.strictEqual(pool.stats().killed, before + 1, 'its worker was terminated');
+ assert.ok((await pool.run('echo', 1)).threadId > 0, 'a fresh worker takes the next run');
+
+ // ── Abort (a cancelled job, a client that left): running and queued ──
+ const ac = new AbortController();
+ const running = rejects(pool.run('spin', { ms: 5000 }, { signal: ac.signal }));
+ const other = pool.run('spin', { ms: 5000 }, { timeoutMs: 100 }).catch(x => x);
+ const queued = rejects(pool.run('spin', { ms: 5000 }, { signal: ac.signal }));
+ await new Promise(r => setTimeout(r, 50));
+ ac.abort();
+ for (const x of await Promise.all([running, queued])) assert.deepStrictEqual([x.status, x.code], [409, 'tools.job.cancelled']);
+ await other;
+ const pre = new AbortController(); pre.abort();
+ assert.strictEqual((await rejects(pool.run('echo', 1, { signal: pre.signal }))).code, 'tools.job.cancelled', 'already aborted: never starts');
+
+ // ── Over the heap limit: fails cleanly, the pool carries on ──
+ const oom = await rejects(pool.run('hog', {}, { timeoutMs: 20000 }));
+ assert.deepStrictEqual([oom.status, oom.code, oom.guardReason], [413, 'tools.input.too_large', 'memory'], oom.message);
+ assert.match(oom.message, /48 MB/);
+ assert.strictEqual(pool.stats().out_of_memory, 1);
+ assert.ok((await pool.run('echo', 2)).threadId > 0, 'still working after a worker ran out of memory');
+
+ // ── A crash fails its run and nothing else ──
+ const crash = await rejects(pool.run('crash', {}));
+ assert.deepStrictEqual([crash.status, crash.code], [500, 'tools.job.failed']);
+ assert.strictEqual((await rejects(pool.run('nope', {}))).status, 500, 'an unknown op is an error, not a hang');
+ assert.deepStrictEqual((await pool.run('echo', 3)).payload, 3);
+
+ // ── Idle workers stop ──
+ await new Promise(r => setTimeout(r, 700));
+ assert.strictEqual(pool.stats().workers, 0, 'no worker kept after idleMs');
+ assert.deepStrictEqual((await pool.run('echo', 4)).payload, 4, 'and one starts again on demand');
+
+ // ── Sizing from the environment ──
+ const p2 = poolFromEnv('docs', MODULE, { env: { TOOLS_WORKERS: '3', TOOLS_WORKERS_DOCS: '1', TOOLS_WORKER_MEMORY_MB: '256', TOOLS_WORKER_QUEUE: '5', MARK: 'x' } });
+ assert.deepStrictEqual([p2.size, p2.memoryMb, p2.maxQueue], [1, 256, 5], 'the per-app override wins');
+ assert.deepStrictEqual(await p2.run('env', { name: 'MARK' }), { value: 'x' }, 'workers see the environment they were given');
+ await p2.close();
+ const p3 = poolFromEnv('img', MODULE, { env: {} });
+ assert.deepStrictEqual([p3.size, p3.memoryMb, p3.maxQueue], [2, 512, 64], 'defaults');
+ await p3.close();
+ assert.strictEqual((await rejects(p3.run('echo', 1))).code, 'tools.busy', 'a closed pool refuses');
+ } finally {
+ await pool.close();
+ }
+ console.log('pool: off the event loop, size cap + queue + refusal, buffers, error fields, timeout/abort terminate, heap limit 413, crash, idle stop, env sizing: all checks passed');
+})().catch((err) => { console.error(err); process.exit(1); });
@@ -84,7 +84,7 @@ function tool(id, extra = {}) {
@@ -84,7 +84,7 @@ function tool(id, extra = {}) {
files: { ...o.files, maxBytes: MAX_BYTES },
files: { ...o.files, maxBytes: MAX_BYTES },
output: { kind: 'file', mime: o.mime, schema: RESULT },
output: { kind: 'file', mime: o.mime, schema: RESULT },
limits: { timeoutMs: TIMEOUT_MS, ...(op !== 'img2pdf' && { maxPages: o.maxPages || MAX_PAGES }) },
limits: { timeoutMs: TIMEOUT_MS, ...(op !== 'img2pdf' && { maxPages: o.maxPages || MAX_PAGES }) },
- // PDF work runs on the satellite's main thread today: a browser session, a person or a token.
+ // Heavy work (a worker thread each, apps/_shared/jobs/pool.js): a browser session, a person or a token.
auth: { anonymous: false, capability: 'tools.tool.run' },
auth: { anonymous: false, capability: 'tools.tool.run' },
quotaClass: 'tools-job', cost: o.cost, egress: false,
quotaClass: 'tools-job', cost: o.cost, egress: false,
...(o.requires && { requires: o.requires }),
...(o.requires && { requires: o.requires }),
@@ -20,7 +20,8 @@ const { resolveContext, DOMAIN_MAP } = require('./domain-map');
@@ -20,7 +20,8 @@ const { resolveContext, DOMAIN_MAP } = require('./domain-map');
const { getTool, listTools, BINARIES } = require('./tools');
const { getTool, listTools, BINARIES } = require('./tools');
const { uploadSingle, uploadMultiple, uploadAny } = require('./middleware/upload');
const { uploadSingle, uploadMultiple, uploadAny } = require('./middleware/upload');
const retention = require('./retention/manager');
const retention = require('./retention/manager');
-const { buildOptions, defineJobs } = require('./process');
+const { buildOptions, defineJobs, runTool } = require('./process');
+const { poolFromEnv } = require('../../_shared/jobs/pool');
const { hostGuard, ownHost } = require('../../_shared/host-role');
const { hostGuard, ownHost } = require('../../_shared/host-role');
const { createToolsApi, exceptRegistry } = require('../../_shared/tools/http');
const { createToolsApi, exceptRegistry } = require('../../_shared/tools/http');
const { createLocalRegistry, requiresStatus } = require('../../_shared/tools/local');
const { createLocalRegistry, requiresStatus } = require('../../_shared/tools/local');
@@ -54,6 +55,19 @@ const { apiLimiter, processLimiter, burstLimiter } = guard.legacyLimiters(rateLi
@@ -54,6 +55,19 @@ const { apiLimiter, processLimiter, burstLimiter } = guard.legacyLimiters(rateLi
const hostTool = (req) => (guard.tool(req.ctx.toolId) ? req.ctx.toolId : (guard.toolForJob('docs.process', req.ctx.defaultOp || 'merge', null) || { id: null }).id);
const hostTool = (req) => (guard.tool(req.ctx.toolId) ? req.ctx.toolId : (guard.toolForJob('docs.process', req.ctx.defaultOp || 'merge', null) || { id: null }).id);
/** The tool a parsed request runs: its `tool` (operation) on this host. */
/** The tool a parsed request runs: its `tool` (operation) on this host. */
const toolOf = (req) => { const op = String((req.body && req.body.tool) || req.ctx.defaultOp || ''); const d = op ? guard.toolForJob('docs.process', op, req.ctx.toolId) : null; return d ? d.id : hostTool(req); };
const toolOf = (req) => { const op = String((req.body && req.body.tool) || req.ctx.defaultOp || ''); const d = op ? guard.toolForJob('docs.process', op, req.ctx.toolId) : null; return d ? d.id : hostTool(req); };
+// ── Worker pool (apps/_shared/jobs/pool.js) ─────────────────
+// pdf-lib runs in worker threads, never on the event loop: TOOLS_WORKERS (2) at once for the
+// synchronous endpoints and the jobs together, each with a heap limit (TOOLS_WORKER_MEMORY_MB, 512)
+// and terminated at its timeout, on cancel, or when the client goes away.
+const pool = poolFromEnv('docs', path.join(__dirname, 'worker.js'));
+/** The sync endpoints' bounds for a run: the tool's descriptor timeout, and the client leaving. */
+function syncRun(req, res) {
+ const ac = new AbortController();
+ res.on('close', () => { if (!res.writableFinished) ac.abort(); });
+ const d = guard.tool(toolOf(req));
+ return { signal: ac.signal, timeoutMs: (d && d.limits && d.limits.timeoutMs) || 10 * 60 * 1000 };
+}
+
/** Read the uploaded files (on disk) and delete them. */
/** Read the uploaded files (on disk) and delete them. */
async function readUploads(files) {
async function readUploads(files) {
try { return await Promise.all(files.map(f => fsp.readFile(f.path))); } finally { for (const f of files) fs.unlink(f.path, () => {}); }
try { return await Promise.all(files.map(f => fsp.readFile(f.path))); } finally { for (const f of files) fs.unlink(f.path, () => {}); }
@@ -151,7 +165,10 @@ app.use((req, _res, next) => {
@@ -151,7 +165,10 @@ app.use((req, _res, next) => {
/** A tool's error on the synchronous endpoints: 503 problem when a tool is not set up, else { error }. */
/** A tool's error on the synchronous endpoints: 503 problem when a tool is not set up, else { error }. */
function sendToolError(req, res, err, label) {
function sendToolError(req, res, err, label) {
if (guard.toolRefused(req, res, err, toolOf(req))) return undefined;
if (guard.toolRefused(req, res, err, toolOf(req))) return undefined;
- if (err.status === 503) return contracts.http.sendProblem(res, 503, err.code || 'tools.unavailable', { detail: err.message, ctx: req.ov });
+ if (err.status === 503) {
+ if (err.retryAfter) res.set('Retry-After', String(err.retryAfter));
+ return contracts.http.sendProblem(res, 503, err.code || 'tools.unavailable', { detail: err.message, ctx: req.ov });
+ }
if (!err.expose) console.error(`[${label}] Error:`, err.message);
if (!err.expose) console.error(`[${label}] Error:`, err.message);
const status = Number.isInteger(err.status) && err.status >= 400 && err.status < 500 ? err.status : 422;
const status = Number.isInteger(err.status) && err.status >= 400 && err.status < 500 ? err.status : 422;
return res.status(status).json({ error: err.message || 'Document processing failed', ...(err.code && { code: err.code }) });
return res.status(status).json({ error: err.message || 'Document processing failed', ...(err.code && { code: err.code }) });
@@ -160,7 +177,7 @@ function sendToolError(req, res, err, label) {
@@ -160,7 +177,7 @@ function sendToolError(req, res, err, label) {
// Health check
// Health check
app.get('/api/health', (_req, res) => {
app.get('/api/health', (_req, res) => {
const stats = retention.getStats();
const stats = retention.getStats();
- res.json({ status: 'ok', service: 'openvibe-docs', version: '1.0.0', files: stats, jobs: jobs.stats() });
+ res.json({ status: 'ok', service: 'openvibe-docs', version: '1.0.0', files: stats, jobs: jobs.stats(), workers: pool.stats() });
});
});
// Domain context (frontend calls this on load to get branding). It also starts this browser's session
// Domain context (frontend calls this on load to get branding). It also starts this browser's session
@@ -193,7 +210,7 @@ app.post('/api/info', burstLimiter, processLimiter, guard.toolQuota(hostTool), u
@@ -193,7 +210,7 @@ app.post('/api/info', burstLimiter, processLimiter, guard.toolQuota(hostTool), u
try {
try {
const tool = getTool('metadata');
const tool = getTool('metadata');
const [buffer] = await readUploads([req.file]);
const [buffer] = await readUploads([req.file]);
- const result = await tool.handler(buffer, { mode: 'view' });
+ const result = await runTool(pool, tool, buffer, { mode: 'view' }, syncRun(req, res));
res.json({ success: true, ...result });
res.json({ success: true, ...result });
} catch (err) {
} catch (err) {
sendToolError(req, res, err, 'Info');
sendToolError(req, res, err, 'Info');
@@ -209,9 +226,9 @@ app.post('/api/process', burstLimiter, processLimiter, guard.toolQuota(hostTool)
@@ -209,9 +226,9 @@ app.post('/api/process', burstLimiter, processLimiter, guard.toolQuota(hostTool)
const tool = getTool(toolId);
const tool = getTool(toolId);
if (!tool) { fs.unlink(req.file.path, () => {}); return res.status(400).json({ error: `Unknown tool: ${toolId}` }); }
if (!tool) { fs.unlink(req.file.path, () => {}); return res.status(400).json({ error: `Unknown tool: ${toolId}` }); }
- // Same options and code path as the docs.process job
+ // Same options and code path as the docs.process job (the worker pool, its cap and its limits)
const [buffer] = await readUploads([req.file]);
const [buffer] = await readUploads([req.file]);
- const result = await tool.handler(buffer, buildOptions(req.body, req.ctx, false));
+ const result = await runTool(pool, tool, buffer, buildOptions(req.body, req.ctx, false), syncRun(req, res));
// Some tools return view-only results (like metadata view)
// Some tools return view-only results (like metadata view)
if (result.viewOnly) {
if (result.viewOnly) {
@@ -260,7 +277,7 @@ app.post('/api/process/multi', burstLimiter, processLimiter, guard.toolQuota(hos
@@ -260,7 +277,7 @@ app.post('/api/process/multi', burstLimiter, processLimiter, guard.toolQuota(hos
}
}
const buffers = await readUploads(req.files);
const buffers = await readUploads(req.files);
- const result = await tool.handler(buffers, buildOptions(req.body, req.ctx, true));
+ const result = await runTool(pool, tool, buffers, buildOptions(req.body, req.ctx, true), syncRun(req, res));
// Save output
// Save output
const firstName = req.files[0]?.originalname || 'output';
const firstName = req.files[0]?.originalname || 'output';
@@ -296,7 +313,7 @@ app.post('/api/process/multi', burstLimiter, processLimiter, guard.toolQuota(hos
@@ -296,7 +313,7 @@ app.post('/api/process/multi', burstLimiter, processLimiter, guard.toolQuota(hos
const jobs = jobsRuntime.setupJobs({
const jobs = jobsRuntime.setupJobs({
app, service: 'docs', dataDir: path.resolve(__dirname, '..', config.dataDir), Database, contracts, sdk,
app, service: 'docs', dataDir: path.resolve(__dirname, '..', config.dataDir), Database, contracts, sdk,
getPublicKey: guard.keys.get, issuer: config.networkUrl, guard,
getPublicKey: guard.keys.get, issuer: config.networkUrl, guard,
- define: defineJobs,
+ define: (system) => defineJobs(system, pool),
receive: uploadAny,
receive: uploadAny,
limiters: [burstLimiter, processLimiter, guard.toolQuota(hostTool)],
limiters: [burstLimiter, processLimiter, guard.toolQuota(hostTool)],
jobTool: (req, type, input) => { const op = String((input && input.tool) || req.ctx.defaultOp || ''); const d = op ? guard.toolForJob('docs.process', op, req.ctx.toolId) : null; return d ? d.id : null; },
jobTool: (req, type, input) => { const op = String((input && input.tool) || req.ctx.defaultOp || ''); const d = op ? guard.toolForJob('docs.process', op, req.ctx.toolId) : null; return d ? d.id : null; },
@@ -382,6 +399,7 @@ function shutdown() {
@@ -382,6 +399,7 @@ function shutdown() {
analytics.destroy();
analytics.destroy();
analyticsDb.close();
analyticsDb.close();
jobs.close(); // running jobs stay 'running' in data/jobs.db; the next boot re-queues them
jobs.close(); // running jobs stay 'running' in data/jobs.db; the next boot re-queues them
+ pool.close();
guard.close();
guard.close();
server.close(() => process.exit(0));
server.close(() => process.exit(0));
setTimeout(() => process.exit(1), 5000);
setTimeout(() => process.exit(1), 5000);
@@ -2,7 +2,9 @@
@@ -2,7 +2,9 @@
// ═══════════════════════════════════════════════════════════════
// ═══════════════════════════════════════════════════════════════
// Docs.OpenVibe — one document operation, shared by the synchronous endpoints
// Docs.OpenVibe — one document operation, shared by the synchronous endpoints
-// (/api/process, /api/process/multi) and the docs.process job (see defineJobs).
+// (/api/process, /api/process/multi, /api/info) and the docs.process job (see defineJobs). Both run
+// it through runTool(): pdf-lib work goes to the worker pool (./worker.js), so the two share one
+// concurrency cap, one heap limit and terminate-on-timeout or cancel.
// ═══════════════════════════════════════════════════════════════
// ═══════════════════════════════════════════════════════════════
const fs = require('fs');
const fs = require('fs');
@@ -17,6 +19,20 @@ const SINGLE_KEYS = ['format', 'quality', 'angle', 'pages', 'order', 'ranges', '
@@ -17,6 +19,20 @@ const SINGLE_KEYS = ['format', 'quality', 'angle', 'pages', 'order', 'ranges', '
'level', 'defaultFormat'];
'level', 'defaultFormat'];
const MULTI_KEYS = ['order', 'pageSize', 'quality', 'format'];
const MULTI_KEYS = ['order', 'pageSize', 'quality', 'format'];
+/**
+ * Run a tool: in the worker pool when it is pdf-lib work, on this thread when it shells out to qpdf
+ * or poppler (their child processes have their own timeouts). Without a pool (tests), inline.
+ * @param {object|null} pool apps/_shared/jobs/pool.js
+ * @param {object} tool tools/index.js entry
+ * @param {Buffer|Buffer[]} input
+ * @param {object} options
+ * @param {{ timeoutMs?: number, signal?: AbortSignal }} [opts]
+ */
+function runTool(pool, tool, input, options, opts = {}) {
+ if (!pool || (tool.requires && tool.requires.length)) return tool.handler(input, options);
+ return pool.run('process', { tool: tool.id, input, options }, { ...opts, transfer: true });
+}
+
/** Tool options from a request body or a job input. */
/** Tool options from a request body or a job input. */
function buildOptions(src, ctx, multi) {
function buildOptions(src, ctx, multi) {
const options = {};
const options = {};
@@ -43,7 +59,7 @@ function describe(toolId, result, fileCount) {
@@ -43,7 +59,7 @@ function describe(toolId, result, fileCount) {
}
}
/** Job type docs.process: input { tool, …options } + one PDF, or several files for merge / img2pdf. */
/** Job type docs.process: input { tool, …options } + one PDF, or several files for merge / img2pdf. */
-function defineJobs(system) {
+function defineJobs(system, pool = null) {
const known = new Set(listTools().map(t => t.id));
const known = new Set(listTools().map(t => t.id));
system.define({
system.define({
type: 'docs.process',
type: 'docs.process',
@@ -72,7 +88,8 @@ function defineJobs(system) {
@@ -72,7 +88,8 @@ function defineJobs(system) {
for (const f of files) buffers.push(await fsp.readFile(f.path));
for (const f of files) buffers.push(await fsp.readFile(f.path));
if (signal.aborted) throw new Error('cancelled');
if (signal.aborted) throw new Error('cancelled');
progress(20, 'Processing');
progress(20, 'Processing');
- const result = await tool.handler(tool.multiFile ? buffers : buffers[0], buildOptions(input, null, tool.multiFile));
+ // Cancel and the job's timeout abort `signal`: a worker is terminated at once.
+ const result = await runTool(pool, tool, tool.multiFile ? buffers : buffers[0], buildOptions(input, null, tool.multiFile), { signal });
if (result.viewOnly) return { files: [], data: describe(toolId, result) };
if (result.viewOnly) return { files: [], data: describe(toolId, result) };
progress(90, 'Saving');
progress(90, 'Saving');
const first = files[0].name || 'document';
const first = files[0].name || 'document';
@@ -84,4 +101,4 @@ function defineJobs(system) {
@@ -84,4 +101,4 @@ function defineJobs(system) {
});
});
}
}
-module.exports = { buildOptions, describe, defineJobs, SINGLE_KEYS, MULTI_KEYS };
+module.exports = { buildOptions, describe, defineJobs, runTool, SINGLE_KEYS, MULTI_KEYS };
new file mode 100644
new file mode 100644
@@ -0,0 +1,22 @@
@@ -0,0 +1,22 @@
+'use strict';
+// ═══════════════════════════════════════════════════════════════
+// Docs.OpenVibe — what a worker thread of the pool runs (apps/_shared/jobs/pool.js).
+//
+// pdf-lib work (merge, split, compress, rotate, reorder, watermark, metadata, and img2pdf with sharp)
+// happens here, off the event loop: the synchronous endpoints and the docs.process job both send it
+// through the pool (process.js runTool), so they share its concurrency cap, its heap limit and its
+// terminate-on-timeout. The tools that shell out to qpdf or poppler (protect, unlock, pdf2img) keep
+// running on the main thread: their work is in a child process with its own timeout already.
+// ═══════════════════════════════════════════════════════════════
+
+const { getTool } = require('./tools');
+
+module.exports = {
+ /** { tool, input: Buffer | Buffer[], options } → the tool's result ({ buffer, ext, mime, … }). */
+ async process({ tool, input, options }) {
+ const t = getTool(tool);
+ if (!t) throw Object.assign(new Error(`Unknown tool: ${tool}`), { status: 400, code: 'tools.job.invalid' });
+ const buffers = Array.isArray(input) ? input.map(b => Buffer.from(b.buffer, b.byteOffset, b.byteLength)) : Buffer.from(input.buffer, input.byteOffset, input.byteLength);
+ return t.handler(buffers, options || {});
+ },
+};
new file mode 100644
new file mode 100644
@@ -0,0 +1,103 @@
@@ -0,0 +1,103 @@
+'use strict';
+// Docs.OpenVibe's worker pool on a real process (apps/_shared/jobs/pool.js, S5):
+// • /api/health keeps answering in milliseconds while a large synchronous merge runs (pdf-lib is
+// in a worker thread, not on the event loop), and the merge still comes back whole
+// • a merge job that needs more heap than TOOLS_WORKER_MEMORY_MB fails cleanly: the job ends
+// failed with 413 tools.input.too_large, the process stays up, and the next job succeeds
+const assert = require('assert');
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+const { PDFDocument, StandardFonts } = require('pdf-lib');
+const { startApp } = require('../../_shared/test/spawn');
+
+const sleep = (ms) => new Promise(r => setTimeout(r, ms));
+
+async function bigPdf(pages) {
+ const doc = await PDFDocument.create();
+ const font = await doc.embedFont(StandardFonts.Helvetica);
+ for (let i = 0; i < pages; i++) doc.addPage([300, 300]).drawText(`Page ${i + 1} ${'x'.repeat(200)}`, { x: 10, y: 150, size: 8, font });
+ return Buffer.from(await doc.save());
+}
+
+async function finished(app, id, cookie) {
+ for (let i = 0; i < 300; i++) {
+ const job = await (await fetch(`${app.base}/api/v1/jobs/${id}`, { headers: { cookie } })).json();
+ if (['succeeded', 'failed', 'cancelled'].includes(job.state)) return job;
+ await sleep(100);
+ }
+ throw new Error(`job ${id} did not finish`);
+}
+
+function mergeForm(files, job) {
+ const fd = new FormData();
+ if (job) { fd.append('type', 'docs.process'); fd.append('input', JSON.stringify({ tool: 'merge' })); } else fd.append('tool', 'merge');
+ files.forEach((b, i) => fd.append('files', new Blob([b], { type: 'application/pdf' }), `part-${i}.pdf`));
+ return fd;
+}
+
+(async () => {
+ const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-docs-pool-'));
+ const envFor = (n) => ({ DATA_DIR: path.join(tmp, n), UPLOADS_DIR: path.join(tmp, n, 'uploads'), OUTPUT_DIR: path.join(tmp, n, 'output'), PDF_MAX_PAGES: '100000' });
+ const big = await bigPdf(2500);
+ const apps = [];
+ try {
+ // ── Health stays fast during a large merge ──
+ const app = await startApp('docs', envFor('a'));
+ apps.push(app);
+ const ctx = await fetch(`${app.base}/api/context`);
+ const cookie = ctx.headers.get('set-cookie').split(';')[0];
+ const t0 = Date.now();
+ const merging = fetch(`${app.base}/api/process/multi`, { method: 'POST', body: mergeForm([big, big]), headers: { cookie } }).then(async r => ({ status: r.status, body: await r.json(), at: Date.now() }));
+ const latencies = [];
+ let done = null;
+ merging.then(r => { done = r; });
+ await sleep(150); // the upload is in and pdf-lib is working
+ while (!done) {
+ const s = Date.now();
+ const h = await fetch(`${app.base}/api/health`);
+ assert.strictEqual(h.status, 200);
+ const body = await h.json();
+ latencies.push(Date.now() - s);
+ if (!done && latencies.length === 3) assert.strictEqual(body.workers.busy, 1, 'the merge is in a worker thread');
+ await sleep(40);
+ }
+ const took = done.at - t0;
+ assert.strictEqual(done.status, 200, JSON.stringify(done.body));
+ assert.strictEqual(done.body.pageCount, 5000, 'every page of both documents');
+ assert.ok(took > 700, `the merge is large enough to measure (${took} ms)`);
+ assert.ok(latencies.length >= 5, `health was asked during the merge (${latencies.length} times)`);
+ const worst = Math.max(...latencies);
+ assert.ok(worst < 250, `/api/health stayed fast during a ${took} ms merge (worst ${worst} ms)`);
+ const health = await (await fetch(`${app.base}/api/health`)).json();
+ assert.strictEqual(health.workers.completed, 1);
+ assert.strictEqual(health.workers.busy, 0);
+
+ // ── Over the heap limit: the job fails cleanly and the next one works ──
+ const small = await startApp('docs', { ...envFor('b'), TOOLS_WORKER_MEMORY_MB: '32' });
+ apps.push(small);
+ const c2 = (await fetch(`${small.base}/api/context`)).headers.get('set-cookie').split(';')[0];
+ let r = await fetch(`${small.base}/api/v1/jobs`, { method: 'POST', body: mergeForm([big, big, big], true), headers: { cookie: c2 } });
+ assert.strictEqual(r.status, 202, await r.clone().text());
+ const oom = await finished(small, (await r.json()).id, c2);
+ assert.strictEqual(oom.state, 'failed');
+ assert.deepStrictEqual([oom.error.status, oom.error.code], [413, 'tools.input.too_large'], JSON.stringify(oom.error));
+ assert.match(oom.error.detail, /32 MB/);
+ assert.strictEqual(oom.retryable, false);
+ assert.strictEqual((await fetch(`${small.base}/api/health`)).status, 200, 'the process is still up');
+ const tiny = await bigPdf(2);
+ r = await fetch(`${small.base}/api/v1/jobs`, { method: 'POST', body: mergeForm([tiny, tiny], true), headers: { cookie: c2 } });
+ const next = await finished(small, (await r.json()).id, c2);
+ assert.strictEqual(next.state, 'succeeded', JSON.stringify(next.error));
+ assert.strictEqual(next.result.data.pageCount, 4);
+ const stats = (await (await fetch(`${small.base}/api/health`)).json()).workers;
+ assert.strictEqual(stats.out_of_memory, 1);
+ // The synchronous endpoint says the same (413, the legacy { error, code } shape).
+ r = await fetch(`${small.base}/api/process/multi`, { method: 'POST', body: mergeForm([big, big, big]), headers: { cookie: c2 } });
+ assert.strictEqual(r.status, 413);
+ assert.strictEqual((await r.json()).code, 'tools.input.too_large');
+ } finally {
+ for (const a of apps) await a.kill('SIGTERM');
+ }
+ console.log('docs pool: health fast during a 5000-page merge in a worker; heap limit fails the job (413) and the next one succeeds');
+})().catch((err) => { console.error(err); process.exit(1); });
@@ -19,7 +19,8 @@ const { resolveContext, DOMAIN_MAP } = require('./domain-map');
@@ -19,7 +19,8 @@ const { resolveContext, DOMAIN_MAP } = require('./domain-map');
const { getTool, listTools } = require('./tools');
const { getTool, listTools } = require('./tools');
const { uploadSingle } = require('./middleware/upload');
const { uploadSingle } = require('./middleware/upload');
const retention = require('./retention/manager');
const retention = require('./retention/manager');
-const { buildOptions, describe, processBuffer, defineJobs } = require('./process');
+const { buildOptions, describe, runProcess, defineJobs } = require('./process');
+const { poolFromEnv } = require('../../_shared/jobs/pool');
const codec = require('./tools/codec');
const codec = require('./tools/codec');
const { hostGuard, ownHost } = require('../../_shared/host-role');
const { hostGuard, ownHost } = require('../../_shared/host-role');
const { createToolsApi, exceptRegistry } = require('../../_shared/tools/http');
const { createToolsApi, exceptRegistry } = require('../../_shared/tools/http');
@@ -54,6 +55,30 @@ const hostTool = (req) => (guard.tool(req.ctx.toolId) ? req.ctx.toolId : (guard.
@@ -54,6 +55,30 @@ const hostTool = (req) => (guard.tool(req.ctx.toolId) ? req.ctx.toolId : (guard.
/** The tool a parsed request runs: its `tool` (operation) on this host. */
/** The tool a parsed request runs: its `tool` (operation) on this host. */
const toolOf = (req) => { const d = guard.toolForJob('img.process', String((req.body && req.body.tool) || req.ctx.defaultOp || 'convert'), req.ctx.toolId); return d ? d.id : hostTool(req); };
const toolOf = (req) => { const d = guard.toolForJob('img.process', String((req.body && req.body.tool) || req.ctx.defaultOp || 'convert'), req.ctx.toolId); return d ? d.id : hostTool(req); };
+// ── Worker pool (apps/_shared/jobs/pool.js) ─────────────────
+// sharp and the BMP/ICO codecs run in worker threads, never on the event loop: TOOLS_WORKERS (2) at
+// once for the synchronous endpoints and the jobs together, each with a heap limit
+// (TOOLS_WORKER_MEMORY_MB, 512) and terminated at its timeout, on cancel, or when the client goes away.
+const pool = poolFromEnv('img', path.join(__dirname, 'worker.js'));
+/** The sync endpoints' bounds for a run: the tool's descriptor timeout, and the client leaving. */
+function syncRun(req, res) {
+ const ac = new AbortController();
+ res.on('close', () => { if (!res.writableFinished) ac.abort(); });
+ const d = guard.tool(toolOf(req));
+ return { signal: ac.signal, timeoutMs: (d && d.limits && d.limits.timeoutMs) || 5 * 60 * 1000, transfer: true };
+}
+/** A tool's failure on the synchronous endpoints: its own 503 (with Retry-After when busy), else { error }. */
+function syncError(req, res, err, label) {
+ if (guard.toolRefused(req, res, err, toolOf(req))) return;
+ if (err.status === 503) {
+ if (err.retryAfter) res.set('Retry-After', String(err.retryAfter));
+ return contracts.http.sendProblem(res, 503, err.code, { detail: err.message, ctx: req.ov });
+ }
+ if (err.code === 'tools.job.cancelled') return; // the client went away
+ console.error(`[${label}] Error:`, err.message);
+ res.status(422).json({ error: err.message || 'Image processing failed' });
+}
+
const app = express();
const app = express();
// What this deploy runs (ADR-016); the shared navbar's release-watch polls it on every tool host.
// What this deploy runs (ADR-016); the shared navbar's release-watch polls it on every tool host.
const release = require('openvibe-shared/release').createRelease({ service: 'tools', root: require('path').join(__dirname, '..', '..', '..') });
const release = require('openvibe-shared/release').createRelease({ service: 'tools', root: require('path').join(__dirname, '..', '..', '..') });
@@ -146,7 +171,7 @@ app.use((req, _res, next) => {
@@ -146,7 +171,7 @@ app.use((req, _res, next) => {
// Health check
// Health check
app.get('/api/health', (_req, res) => {
app.get('/api/health', (_req, res) => {
const stats = retention.getStats();
const stats = retention.getStats();
- res.json({ status: 'ok', service: 'openvibe-img', version: '1.0.0', files: stats, jobs: jobs.stats() });
+ res.json({ status: 'ok', service: 'openvibe-img', version: '1.0.0', files: stats, jobs: jobs.stats(), workers: pool.stats() });
});
});
// Domain context (frontend calls this on load to get branding). It also starts this browser's session
// Domain context (frontend calls this on load to get branding). It also starts this browser's session
@@ -185,8 +210,8 @@ app.post('/api/process', burstLimiter, processLimiter, guard.toolQuota(hostTool)
@@ -185,8 +210,8 @@ app.post('/api/process', burstLimiter, processLimiter, guard.toolQuota(hostTool)
return res.status(400).json({ error: `Unknown tool: ${toolId}` });
return res.status(400).json({ error: `Unknown tool: ${toolId}` });
}
}
- // Execute the tool (the same code the img.process job runs)
- const result = await processBuffer(req.file.buffer, toolId, buildOptions(req.body, req.ctx, req.file.mimetype));
+ // Execute the tool (the same code, pool and limits the img.process job uses)
+ const result = await runProcess(pool, req.file.buffer, toolId, buildOptions(req.body, req.ctx, req.file.mimetype), syncRun(req, res));
// Save to retention
// Save to retention
const saved = retention.saveOutput(
const saved = retention.saveOutput(
@@ -199,10 +224,7 @@ app.post('/api/process', burstLimiter, processLimiter, guard.toolQuota(hostTool)
@@ -199,10 +224,7 @@ app.post('/api/process', burstLimiter, processLimiter, guard.toolQuota(hostTool)
res.json({ success: true, download: saved, ...describe(toolId, result) });
res.json({ success: true, download: saved, ...describe(toolId, result) });
} catch (err) {
} catch (err) {
- if (guard.toolRefused(req, res, err, toolOf(req))) return;
- if (err.status === 503) return contracts.http.sendProblem(res, 503, err.code, { detail: err.message, ctx: req.ov });
- console.error('[Process] Error:', err.message);
- res.status(422).json({ error: err.message || 'Image processing failed' });
+ syncError(req, res, err, 'Process');
}
}
});
});
@@ -213,7 +235,7 @@ app.post('/api/process/direct', burstLimiter, processLimiter, guard.toolQuota(ho
@@ -213,7 +235,7 @@ app.post('/api/process/direct', burstLimiter, processLimiter, guard.toolQuota(ho
const tool = getTool(toolId);
const tool = getTool(toolId);
if (!tool) return res.status(400).json({ error: `Unknown tool: ${toolId}` });
if (!tool) return res.status(400).json({ error: `Unknown tool: ${toolId}` });
- const result = await processBuffer(req.file.buffer, toolId, buildOptions(req.body, req.ctx, req.file.mimetype));
+ const result = await runProcess(pool, req.file.buffer, toolId, buildOptions(req.body, req.ctx, req.file.mimetype), syncRun(req, res));
const baseName = path.basename(req.file.originalname, path.extname(req.file.originalname));
const baseName = path.basename(req.file.originalname, path.extname(req.file.originalname));
res.set({
res.set({
@@ -223,10 +245,7 @@ app.post('/api/process/direct', burstLimiter, processLimiter, guard.toolQuota(ho
@@ -223,10 +245,7 @@ app.post('/api/process/direct', burstLimiter, processLimiter, guard.toolQuota(ho
});
});
res.send(result.buffer);
res.send(result.buffer);
} catch (err) {
} catch (err) {
- if (guard.toolRefused(req, res, err, toolOf(req))) return;
- if (err.status === 503) return contracts.http.sendProblem(res, 503, err.code, { detail: err.message, ctx: req.ov });
- console.error('[Process/Direct] Error:', err.message);
- res.status(422).json({ error: err.message || 'Image processing failed' });
+ syncError(req, res, err, 'Process/Direct');
}
}
});
});
@@ -236,7 +255,7 @@ app.post('/api/process/direct', burstLimiter, processLimiter, guard.toolQuota(ho
@@ -236,7 +255,7 @@ app.post('/api/process/direct', burstLimiter, processLimiter, guard.toolQuota(ho
const jobs = jobsRuntime.setupJobs({
const jobs = jobsRuntime.setupJobs({
app, service: 'img', dataDir: path.resolve(__dirname, '..', config.dataDir), Database, contracts, sdk,
app, service: 'img', dataDir: path.resolve(__dirname, '..', config.dataDir), Database, contracts, sdk,
getPublicKey: guard.keys.get, issuer: config.networkUrl, guard,
getPublicKey: guard.keys.get, issuer: config.networkUrl, guard,
- define: defineJobs,
+ define: (system) => defineJobs(system, pool),
receive: uploadSingle,
receive: uploadSingle,
limiters: [burstLimiter, processLimiter, guard.toolQuota(hostTool)],
limiters: [burstLimiter, processLimiter, guard.toolQuota(hostTool)],
jobTool: (req, type, input) => { const d = guard.toolForJob('img.process', String((input && input.tool) || req.ctx.defaultOp || 'convert'), req.ctx.toolId); return d ? d.id : null; },
jobTool: (req, type, input) => { const d = guard.toolForJob('img.process', String((input && input.tool) || req.ctx.defaultOp || 'convert'), req.ctx.toolId); return d ? d.id : null; },
@@ -320,6 +339,7 @@ function shutdown() {
@@ -320,6 +339,7 @@ function shutdown() {
analytics.destroy();
analytics.destroy();
analyticsDb.close();
analyticsDb.close();
jobs.close(); // running jobs stay 'running' in data/jobs.db; the next boot re-queues them
jobs.close(); // running jobs stay 'running' in data/jobs.db; the next boot re-queues them
+ pool.close();
guard.close();
guard.close();
server.close(() => process.exit(0));
server.close(() => process.exit(0));
setTimeout(() => process.exit(1), 5000);
setTimeout(() => process.exit(1), 5000);
@@ -2,7 +2,9 @@
@@ -2,7 +2,9 @@
// ═══════════════════════════════════════════════════════════════
// ═══════════════════════════════════════════════════════════════
// Img.OpenVibe — one image operation, shared by the synchronous endpoints
// Img.OpenVibe — one image operation, shared by the synchronous endpoints
-// (/api/process, /api/process/direct) and the img.process job (server/jobs.js).
+// (/api/process, /api/process/direct) and the img.process job (defineJobs). Both run it through
+// runProcess(): the worker pool (./worker.js), so they share one concurrency cap, one heap limit and
+// terminate-on-timeout or cancel.
// ═══════════════════════════════════════════════════════════════
// ═══════════════════════════════════════════════════════════════
const fs = require('fs');
const fs = require('fs');
@@ -41,8 +43,19 @@ async function processBuffer(buffer, toolId, options) {
@@ -41,8 +43,19 @@ async function processBuffer(buffer, toolId, options) {
return tool.handler(buffer, options);
return tool.handler(buffer, options);
}
}
+/**
+ * Run an operation in the worker pool (inline without one: tests).
+ * @param {object|null} pool apps/_shared/jobs/pool.js
+ * @param {{ timeoutMs?: number, signal?: AbortSignal }} [opts]
+ */
+function runProcess(pool, buffer, toolId, options, opts = {}) {
+ if (!pool) return processBuffer(buffer, toolId, options);
+ if (!getTool(toolId)) return Promise.reject(Object.assign(new Error(`Unknown tool: ${toolId}`), { code: 'tools.job.invalid', status: 400 }));
+ return pool.run('process', { tool: toolId, buffer, options }, opts);
+}
+
/** Job type img.process: input { tool, format?, quality?, width?, … } + one image file. */
/** Job type img.process: input { tool, format?, quality?, width?, … } + one image file. */
-function defineJobs(system) {
+function defineJobs(system, pool = null) {
const known = new Set(listTools().map(t => t.id));
const known = new Set(listTools().map(t => t.id));
system.define({
system.define({
type: 'img.process',
type: 'img.process',
@@ -64,7 +77,8 @@ function defineJobs(system) {
@@ -64,7 +77,8 @@ function defineJobs(system) {
const buffer = await fsp.readFile(src.path);
const buffer = await fsp.readFile(src.path);
if (signal.aborted) throw new Error('cancelled');
if (signal.aborted) throw new Error('cancelled');
progress(20, 'Processing');
progress(20, 'Processing');
- const result = await processBuffer(buffer, toolId, buildOptions(input, null, src.mime));
+ // Cancel and the job's timeout abort `signal`: a worker is terminated at once.
+ const result = await runProcess(pool, buffer, toolId, buildOptions(input, null, src.mime), { signal, transfer: true });
progress(90, 'Saving');
progress(90, 'Saving');
const base = path.basename(src.name || 'image', path.extname(src.name || '')) || 'openvibeimg-output';
const base = path.basename(src.name || 'image', path.extname(src.name || '')) || 'openvibeimg-output';
const name = `${base}.${result.ext}`;
const name = `${base}.${result.ext}`;
@@ -75,4 +89,4 @@ function defineJobs(system) {
@@ -75,4 +89,4 @@ function defineJobs(system) {
});
});
}
}
-module.exports = { buildOptions, describe, processBuffer, defineJobs, OPTION_KEYS };
+module.exports = { buildOptions, describe, processBuffer, runProcess, defineJobs, OPTION_KEYS };
new file mode 100644
new file mode 100644
@@ -0,0 +1,17 @@
@@ -0,0 +1,17 @@
+'use strict';
+// ═══════════════════════════════════════════════════════════════
+// Img.OpenVibe — what a worker thread of the pool runs (apps/_shared/jobs/pool.js).
+//
+// Every image operation (sharp, and the JavaScript BMP/ICO codecs) happens here, off the event loop:
+// the synchronous endpoints and the img.process job both send it through the pool (process.js
+// runProcess), so they share its concurrency cap, its heap limit and terminate-on-timeout or cancel.
+// ═══════════════════════════════════════════════════════════════
+
+const { processBuffer } = require('./process');
+
+module.exports = {
+ /** { tool, buffer, options } → { buffer, ext, mime, … } */
+ async process({ tool, buffer, options }) {
+ return processBuffer(Buffer.from(buffer.buffer, buffer.byteOffset, buffer.byteLength), tool, options || {});
+ },
+};