diff --git a/src/api/index.js b/src/api/index.js index 397875b..26dd1dd 100644 --- a/src/api/index.js +++ b/src/api/index.js @@ -13,6 +13,7 @@ import { ensureConnection } from '../db/store.js'; import { closeAll as closeAllSse } from './sse.js'; import { closeAllManagers } from '../services/conversationManager.js'; import { shutdownImageCropWorkers } from '../services/imageCrop.js'; +import { shutdownJavascriptSandboxWorkers } from '../services/javascriptSandbox.js'; import { getOrCreateCert } from '../util/cert.js'; import { createLogger } from '../util/logger.js'; import v1Router from '../v1/index.js'; @@ -242,6 +243,8 @@ export async function stopApi() { // Pooled image-crop workers are live threads and must be terminated too, or the // process stays alive after the last request has been served. await shutdownImageCropWorkers(); + // Same for the pooled sandbox threads that run `javascript_execute` / applets. + await shutdownJavascriptSandboxWorkers(); const closers = []; if (server) { const s = server; diff --git a/src/services/javascriptSandbox.js b/src/services/javascriptSandbox.js new file mode 100644 index 0000000..ad4d443 --- /dev/null +++ b/src/services/javascriptSandbox.js @@ -0,0 +1,406 @@ +import { Worker } from 'worker_threads'; +import { fileURLToPath } from 'url'; +import env from '../config/env.js'; + +// The sandbox worker pays a large fixed startup cost: importing vm2, sharp, exceljs, pdfkit, +// react-canvas-label-editor AND `vfs.js` (which pulls in libdashstore) costs ~700 ms in a fresh +// thread, while the snippet itself runs in ~2 ms. ESM modules are never shared between worker +// threads, so spawning one thread per applet run re-paid that cost on every request. +// +// This pool keeps threads alive and hands them one job at a time. Threads are keyed by VFS root +// (`data/vfs/`), so a reused thread always has the right `vfsDir` baked in and two +// companies can never share a thread. +const WORKER_PATH = fileURLToPath(new URL('../tools/javascript-worker.js', import.meta.url)); +const ROOT_DIR = fileURLToPath(new URL('../..', import.meta.url)); +const WORKER_BOOT_TIMEOUT_MS = 15000; + +// Two threads per company rides out the common "run it, then run it again" loop without queueing. +const MAX_WORKERS_PER_COMPANY = 2; +// ...but the process is shared by every company, so the total stays bounded. Each thread holds +// sharp/libvips plus a NodeVM, so 5 is a deliberate memory ceiling rather than a core count. +const MAX_WORKERS_TOTAL = 5; + +const DEFAULT_BUFFER_LIMIT = 32 * 1024 * 1024; + +/** + * @typedef {object} PoolEntry + * @property {Worker} worker + * @property {number} id Monotonic thread id, used to tag job ids. + * @property {string} poolKey VFS root this thread was spawned for. + * @property {boolean} busy + * @property {number} lastUsed `Date.now()` of the last job handoff (eviction order). + */ + +/** @type {Array} */ +const pool = []; +/** @type {Array<{ resolve: Function, reject: Function, timer: NodeJS.Timeout, poolKey: string, job: object, timeoutMs: number }>} */ +const queue = []; +let workerSeq = 0; +let jobSeq = 0; +let shuttingDown = false; +/** Serializes pool growth so concurrent callers cannot overshoot either cap. */ +let growing = Promise.resolve(); + +function countFor(poolKey) { + return pool.filter((entry) => entry.poolKey === poolKey).length; +} + +function findIdle(poolKey) { + return pool.find((entry) => !entry.busy && entry.poolKey === poolKey); +} + +function dropWorker(entry) { + const index = pool.indexOf(entry); + if (index !== -1) pool.splice(index, 1); +} + +/** + * Reclaims an idle thread belonging to a DIFFERENT company so a queued job can be served. + * + * Without this the total cap would deadlock: with 5 threads spread over three companies, a + * fourth company's job could never be dispatched (no thread for it) and never spawn (cap hit). + * Only genuinely idle threads are evicted, and preferentially ones whose own company has + * nothing waiting. + * + * @param {string} poolKey + * @returns {boolean} whether a thread was reclaimed + */ +function evictIdleOtherCompany(poolKey) { + const candidates = pool.filter((entry) => !entry.busy && entry.poolKey !== poolKey); + if (candidates.length === 0) return false; + + const idleOthers = new Set( + queue.filter((item) => item.poolKey !== poolKey).map((item) => item.poolKey) + ); + candidates.sort((a, b) => { + const aWanted = idleOthers.has(a.poolKey) ? 1 : 0; + const bWanted = idleOthers.has(b.poolKey) ? 1 : 0; + return aWanted - bWanted || a.lastUsed - b.lastUsed; + }); + + const victim = candidates[0]; + dropWorker(victim); + try { victim.worker.terminate(); } catch { /* already closed */ } + return true; +} + +/** + * Spawns a thread for `vfsRoot` and resolves once it has reported `{ ready: true }`. + * + * The worker signals readiness only after its module body (and therefore every import) has + * finished, so a thread is never handed a job while it is still booting — which is what makes + * the reuse actually cheap. + * + * @param {string} vfsRoot VFS root of the company this thread serves (the pool key). + * @param {number} googleAuthId + * @returns {Promise} + */ +function spawnWorker(vfsRoot, googleAuthId) { + const entry = { worker: null, id: ++workerSeq, poolKey: vfsRoot, busy: false, lastUsed: Date.now() }; + + return new Promise((resolve, reject) => { + const bootTimer = setTimeout(() => { + cleanup(); + try { entry.worker?.terminate(); } catch { /* already closed */ } + reject(new Error(`Sandbox worker failed to start within ${WORKER_BOOT_TIMEOUT_MS} ms`)); + }, WORKER_BOOT_TIMEOUT_MS); + + function cleanup() { + clearTimeout(bootTimer); + entry.worker?.off('message', onBooted); + entry.worker?.off('error', onError); + entry.worker?.off('exit', onExit); + } + + function onBooted(msg) { + if (!msg || msg.ready !== true) return; + cleanup(); + resolve(entry); + } + + function onError(err) { + cleanup(); + reject(err instanceof Error ? err : new Error(err?.message || 'Sandbox worker failed to start')); + } + + function onExit(code) { + cleanup(); + reject(new Error(`Sandbox worker exited before becoming ready (code ${code})`)); + } + + try { + entry.worker = new Worker(WORKER_PATH, { + workerData: { + root: ROOT_DIR, + bufferLimit: DEFAULT_BUFFER_LIMIT, + dashbackUrl: `http://127.0.0.1:${env.port}/api`, + vfsDir: vfsRoot, + googleAuthId, + }, + }); + } catch (err) { + cleanup(); + reject(err); + return; + } + + entry.worker.on('message', onBooted); + entry.worker.on('error', onError); + entry.worker.on('exit', onExit); + }); +} + +/** + * Returns an idle thread for `poolKey`, spawning (or evicting) to grow the pool when allowed. + * Resolves to `null` when every permitted thread is busy — the caller then queues. + * + * @param {string} poolKey + * @param {number} googleAuthId + * @returns {Promise} + */ +async function acquireWorker(poolKey, googleAuthId) { + const idle = findIdle(poolKey); + // Reserve SYNCHRONOUSLY, before any `await`. Otherwise two concurrent callers both pass + // this check while the thread is still unmarked, and both post to the same worker. + if (idle) { + idle.busy = true; + idle.lastUsed = Date.now(); + return idle; + } + + if (countFor(poolKey) < MAX_WORKERS_PER_COMPANY) { + growing = growing.then(async () => { + // Re-check inside the serialized block: another caller may have grown the pool. + if (countFor(poolKey) >= MAX_WORKERS_PER_COMPANY) return; + if (pool.length >= MAX_WORKERS_TOTAL && !evictIdleOtherCompany(poolKey)) return; + try { + pool.push(await spawnWorker(poolKey, googleAuthId)); + } catch (err) { + // Fail the oldest queued job for this company rather than leaving it waiting on + // a thread that will never arrive, which would deadlock the queue instead of + // surfacing the boot failure to the caller. + const index = queue.findIndex((item) => item.poolKey === poolKey); + if (index === -1) throw err; + const waiting = queue.splice(index, 1)[0]; + clearTimeout(waiting.timer); + waiting.reject(err); + } + }); + try { + await growing; + } catch { + // Already reported to the queued job; continue so this caller still queues. + } + // Reserve the freshly booted thread atomically; a concurrent waiter that wakes after us + // finds none and queues instead. + const after = findIdle(poolKey); + if (after) { + after.busy = true; + after.lastUsed = Date.now(); + return after; + } + } + + return null; +} + +/** + * Hands one job to one thread and resolves with the worker's report. + * + * Always RESOLVES with a report-shaped object — it never rejects. The tool layer distinguishes + * outcomes by shape (`timedOut`, `error`, or a `result` key), matching what the old one-shot + * worker returned, so timeout/error mapping in the routes keeps working unchanged. + * + * @param {PoolEntry} entry + * @param {object} job Forwarded to the worker verbatim. + * @param {number} timeoutMs + * @returns {Promise} + */ +function runJobOnWorker(entry, job, timeoutMs) { + return new Promise((resolve) => { + const jobId = `${entry.id}:${++jobSeq}`; + let settled = false; + + const timer = setTimeout(() => { + // Terminate first, then report. A snippet stuck in a native call (libvips, fetch) + // would otherwise keep burning the thread after the caller has moved on, and its + // eventual reply would arrive against an already-settled job. + try { entry.worker.terminate(); } catch { /* already closed */ } + dropWorker(entry); + finish({ result: null, stdout: '', consoleLineCount: 0, timedOut: true, timeoutMs }); + }, timeoutMs); + + function cleanup() { + clearTimeout(timer); + entry.worker.off('message', onMessage); + entry.worker.off('error', onError); + entry.worker.off('exit', onExit); + } + + function finish(payload) { + if (settled) return; + settled = true; + cleanup(); + resolve(payload); + } + + function onMessage(msg) { + // Ignore anything that is not THIS job's report. Under `node --watch` the dev + // server's module-loader hook posts `watch:import` notices on the same channel, so + // the first message is not necessarily our reply; job ids are matched, not order. + if (!msg || msg.__sandboxReport !== true || msg.id !== jobId) return; + finish(msg); + } + + function onError(err) { + dropWorker(entry); + finish({ + result: null, + stdout: '', + consoleLineCount: 0, + error: (err && err.message) ? err.message : `Sandbox worker failed: ${err?.stack || err?.name || 'unknown error'}`, + errorName: (err && err.name) || 'Error', + stack: err && typeof err.stack === 'string' ? err.stack : undefined, + }); + } + + function onExit(code) { + dropWorker(entry); + // Reaching here means the thread died WITHOUT reporting — a hard failure, never a + // silent empty success. + console.error(`[sandbox] worker exited without reporting (code ${code})`); + finish({ + result: null, + stdout: '', + consoleLineCount: 0, + error: `Sandbox worker exited unexpectedly without reporting a result (code ${code}).`, + errorName: 'WorkerExit', + }); + } + + entry.worker.on('message', onMessage); + entry.worker.on('error', onError); + entry.worker.on('exit', onExit); + + try { + entry.worker.postMessage({ ...job, id: jobId }); + } catch (err) { + finish({ + result: null, + stdout: '', + consoleLineCount: 0, + error: (err && err.message) || 'Failed to post job to sandbox worker', + errorName: 'PostMessageError', + }); + } + }); +} + +/** + * Starts as many queued jobs as there are idle threads for their company. + */ +function drainQueue() { + for (;;) { + const index = queue.findIndex((item) => findIdle(item.poolKey)); + if (index === -1) return; + + const item = queue.splice(index, 1)[0]; + clearTimeout(item.timer); + const entry = findIdle(item.poolKey); + entry.busy = true; + entry.lastUsed = Date.now(); + + runJobOnWorker(entry, item.job, item.timeoutMs) + .then(item.resolve, item.reject) + .finally(() => { + entry.busy = false; + drainQueue(); + }); + } +} + +/** + * Runs one sandbox job on a pooled thread, spawning a thread if the caps allow and otherwise + * queueing until one frees up. + * + * @param {object} options + * @param {string} options.poolKey VFS root; identifies the company (and the thread). + * @param {number} options.googleAuthId + * @param {string} options.code Source to run. + * @param {number} options.timeoutMs + * @param {string} options.dashbackToken Per-run JWT for the sandbox's `fetch()`. + * @param {object} [options.appletInput] + * @param {string} [options.appletFile] + * @returns {Promise} Worker report (`result` / `timedOut` / `error` shape). + */ +export async function runSandboxJob(options) { + if (shuttingDown) { + throw new Error('Sandbox worker pool is shutting down'); + } + + const { poolKey, googleAuthId, code, timeoutMs, dashbackToken, appletInput, appletFile } = options; + const job = { code, timeoutMs, dashbackToken, appletInput, appletFile, poolKey, googleAuthId }; + + const entry = await acquireWorker(poolKey, googleAuthId); + if (entry) { + try { + return await runJobOnWorker(entry, job, timeoutMs); + } finally { + entry.busy = false; + drainQueue(); + } + } + + return new Promise((resolve, reject) => { + // The queue wait is bounded by the same timeout budget as execution, so a caller cannot + // wait forever behind other companies' jobs. + const timer = setTimeout(() => { + const index = queue.findIndex((item) => item.timer === timer); + if (index !== -1) queue.splice(index, 1); + resolve({ + result: null, + stdout: '', + consoleLineCount: 0, + error: `Sandbox queue timed out after ${timeoutMs} ms (all workers busy).`, + errorName: 'SandboxQueueTimeout', + }); + }, timeoutMs); + + queue.push({ resolve, reject, timer, poolKey, job, timeoutMs }); + }); +} + +/** + * Terminates every pooled thread and fails anything still queued. + * + * Mirrors `shutdownImageCropWorkers`: live worker threads keep the process alive after the last + * request, so `node --watch` hangs on "Waiting for graceful termination..." without this. + * + * @returns {Promise} + */ +export async function shutdownJavascriptSandboxWorkers() { + shuttingDown = true; + + while (queue.length > 0) { + const pending = queue.shift(); + clearTimeout(pending.timer); + pending.resolve({ + result: null, + stdout: '', + consoleLineCount: 0, + error: 'Sandbox worker pool is shutting down', + errorName: 'SandboxShutdown', + }); + } + + const entries = pool.splice(0, pool.length); + await Promise.all( + entries.map((entry) => + Promise.resolve() + .then(() => entry.worker.terminate()) + .catch(() => { /* already closed */ }) + ) + ); + + shuttingDown = false; +} diff --git a/src/tools/javascript-worker.js b/src/tools/javascript-worker.js index 5286d9e..3c74c64 100644 --- a/src/tools/javascript-worker.js +++ b/src/tools/javascript-worker.js @@ -10,18 +10,15 @@ import ExcelJS from 'exceljs'; import PDFDocument from 'pdfkit'; import { createVfs } from './vfs.js'; +// Static, per-THREAD configuration. A pooled thread serves many jobs, so per-job values +// (`code`, `appletInput`, `appletFile`, the request-scoped token and timeout) arrive as +// `runJob` messages instead — see the job loop at the bottom of this file. const { - code, root, - vmTimeout, bufferLimit, - network, dashbackUrl, - dashbackToken, vfsDir, googleAuthId, - appletInput, - appletFile, } = workerData; function resolveDashbackUrl(inputUrl, baseApiUrl) { @@ -425,7 +422,6 @@ function bridgePDFKit(vfs) { } const NETWORK_GLOBALS = { - fetch: createSandboxFetch(dashbackUrl, dashbackToken), // WebCrypto globals (bridged across the vm2 realm boundary — see bridgeCrypto above). // Node's `crypto` MODULE is deliberately not exposed. ...(bridgedCrypto ? { crypto: bridgedCrypto } : {}), @@ -448,46 +444,71 @@ const NETWORK_GLOBALS = { clearTimeout, }; -const lines = []; -const push = (prefix) => (...args) => { - let text; - try { - text = args - .map((a) => { - if (typeof a === 'string') return a; - if (Buffer.isBuffer(a) || ArrayBuffer.isView(a) || (a && a.constructor && a.constructor.name === 'Buffer')) { - return ``; - } - try { - return JSON.stringify(a, (k, v) => { - if (v && typeof v === 'object') { - if (v.type === 'Buffer' && Array.isArray(v.data)) { - return ``; +/** + * Builds the sandbox `fetch` for ONE job. + * + * `fetch` signs its `Authorization` header with the JWT minted for that job's `googleAuthId`, + * so it must not live in a module-level object shared across every company a pooled thread + * happens to serve. + * + * @param {string} dashbackToken + */ +function createRunGlobals(dashbackToken) { + return { + fetch: createSandboxFetch(dashbackUrl, dashbackToken), + ...NETWORK_GLOBALS, + }; +} + +/** + * Builds a console that appends to the given per-job line buffer. + * + * A pooled thread runs many applets, so each job needs its OWN sink — otherwise job N's stdout + * would include job N-1's output. + * + * @param {Array} lines + */ +function createSandboxConsole(lines) { + const push = (prefix) => (...args) => { + let text; + try { + text = args + .map((a) => { + if (typeof a === 'string') return a; + if (Buffer.isBuffer(a) || ArrayBuffer.isView(a) || (a && a.constructor && a.constructor.name === 'Buffer')) { + return ``; + } + try { + return JSON.stringify(a, (k, v) => { + if (v && typeof v === 'object') { + if (v.type === 'Buffer' && Array.isArray(v.data)) { + return ``; + } + if (Buffer.isBuffer(v) || ArrayBuffer.isView(v) || (v.constructor && v.constructor.name === 'Buffer')) { + return ``; + } } - if (Buffer.isBuffer(v) || ArrayBuffer.isView(v) || (v.constructor && v.constructor.name === 'Buffer')) { - return ``; - } - } - return v; - }); - } catch { - return String(a); - } - }) - .join(' '); - } catch { - text = String(args); - } - lines.push(prefix + text); -}; -const sandboxConsole = { - log: push(''), - info: push(''), - debug: push(''), - warn: push('warn: '), - error: push('error: '), - trace: push('trace: '), -}; + return v; + }); + } catch { + return String(a); + } + }) + .join(' '); + } catch { + text = String(args); + } + lines.push(prefix + text); + }; + return { + log: push(''), + info: push(''), + debug: push(''), + warn: push('warn: '), + error: push('error: '), + trace: push('trace: '), + }; +} const MAX_ITEMS = 200; @@ -527,37 +548,53 @@ function sanitize(value, depth = 0) { return { type: t }; } -// Every genuine sandbox report is tagged with `__sandboxReport: true`. -// -// This is REQUIRED, not cosmetic: when the server runs under `node --watch` (see the `dev` -// script), Node installs a module-loader hook that posts `watch:import` notices over the SAME -// parentPort channel the worker reports on. Without this tag the parent cannot tell a real -// report apart from loader chatter and will resolve on the wrong message. -function report(payload) { - const message = { __sandboxReport: true, ...payload }; +/** + * Posts a job report back to the pool. + * + * Every report is tagged with the job id it answers AND `__sandboxReport: true`. + * + * The tag is REQUIRED, not cosmetic: when the server runs under `node --watch` (see the `dev` + * script), Node installs a module-loader hook that posts `watch:import` notices over the SAME + * parentPort channel the worker reports on. The pool matches the `id` to route a reply to the + * right job; without the tag it cannot tell a real report apart from loader chatter. + * + * @param {string} id + * @param {object} payload + */ +function report(id, payload) { try { - parentPort.postMessage(message); + parentPort.postMessage({ __sandboxReport: true, id, ...payload }); } catch { try { - parentPort.postMessage({ - __sandboxReport: true, - result: null, - error: 'Sandbox produced a result that could not be serialized.', - errorName: 'SerializationError', - stdout: lines.join('\n'), - consoleLineCount: lines.length, - }); + parentPort.postMessage({ __sandboxReport: true, id, result: null, error: 'Sandbox produced a result that could not be serialized.', errorName: 'SerializationError' }); } catch { /* thread exiting */ } } } -void (async () => { -try { +/** + * Runs ONE job posted by the pool. Never throws: every outcome is reported via {@link report}. + * + * @param {object} job + * @param {string} job.id Opaque job id; echoed back on the report. + * @param {string} job.code Source to run. + * @param {string} job.dashbackToken Per-job JWT used to sign sandbox `fetch()` calls. + * @param {object} [job.appletInput] Value passed as `applet_input` when `appletFile` is set. + * @param {string} [job.appletFile] VFS path of the applet, which switches on the applet protocol. + */ +async function runJob(job) { + const code = job.code; + const appletInput = job.appletInput; + const appletFile = job.appletFile; + const vfs = createVfs(vfsDir, { googleAuthId }); const bridgedExcel = bridgeExcelJS(vfs); const bridgedPDF = bridgePDFKit(vfs); + // One console sink per job so pooled runs cannot see each other's stdout. + const lines = []; + const sandboxConsole = createSandboxConsole(lines); + const vm = new NodeVM({ console: 'off', sandbox: { @@ -572,7 +609,7 @@ try { pdfkit: bridgedPDF, validateLabelObjects: (objects) => validateObjects(JSON.parse(JSON.stringify(objects))), validateLabelObject: (object) => validateObject(JSON.parse(JSON.stringify(object))), - ...NETWORK_GLOBALS, + ...createRunGlobals(job.dashbackToken), }, require: { root, @@ -581,7 +618,7 @@ try { }, nesting: false, bufferAllocLimit: bufferLimit, - timeout: vmTimeout, + timeout: job.timeoutMs, }); const started = Date.now(); @@ -602,52 +639,67 @@ try { value = value(appletInput); } if (value && typeof value.then === 'function') { - const settleTimeout = Math.max(1000, vmTimeout - 500); + const settleTimeout = Math.max(1000, job.timeoutMs - 500); value = await Promise.race([ - value, - new Promise((_, reject) => - setTimeout(() => reject(new Error('Async snippet did not settle before the timeout.')), settleTimeout) - ), + value, + new Promise((_, reject) => + setTimeout(() => reject(new Error('Async snippet did not settle before the timeout.')), settleTimeout) + ), ]); } } catch (err) { const rawMsg = (err && err.message) || String(err); failed = { - error: rawMsg, + error: rawMsg, errorName: (err && err.name) || 'Error', cause: err && err.cause ? ((err.cause && err.cause.message) || String(err.cause)) : undefined, - stack: err && typeof err.stack === 'string' ? err.stack : undefined, + stack: err && typeof err.stack === 'string' ? err.stack : undefined, }; // Snippets run in a CommonJS wrapper, so bare top-level `await` is a syntax error. - // That is the single most common mistake, so say exactly how to fix it. + // That is the single most common mistake, so say exactly how to fix it. if (err instanceof SyntaxError && /await is only valid|Unexpected reserved word/i.test(rawMsg)) { - failed.error = - 'Top-level `await` is not allowed: snippets run in a CommonJS (non-ESM) wrapper. ' + - 'Wrap async code in an async IIFE and assign the promise to module.exports:\n' + - ' module.exports = (async () => { const res = await fetch("/company"); return res.json(); })();'; -} + failed.error = + 'Top-level `await` is not allowed: snippets run in a CommonJS (non-ESM) wrapper. ' + + 'Wrap async code in an async IIFE and assign the promise to module.exports:\n' + + ' module.exports = (async () => { const res = await fetch("/company"); return res.json(); })();'; + } } if (failed) { console.error('[javascript-worker] snippet threw:', failed.error, '\n', failed.stack || '(no stack)'); - report({ result: null, ...done(), ...failed }); + report(job.id, { result: null, ...done(), ...failed }); } else { // `result` MUST always be present: if the key were absent, the parent's // JSON.stringify() drops it and the tool response degenerates to a bare // echo of the submitted source with no sign of execution. - report({ result: sanitize(value), ...done() }); + report(job.id, { result: sanitize(value), ...done() }); } -} catch (err) { - // Setup-level failure (bad VFS dir, thrown import, etc.) — log it here too, because the - // parent only sees a message and the worker's stderr is otherwise swallowed. - console.error('[javascript-worker] fatal setup error:', (err && err.message) || String(err), '\n', (err && err.stack) || '(no stack)'); - report({ - result: null, - stdout: lines.join('\n'), - consoleLineCount: lines.length, - error: (err && err.message) || String(err), - errorName: (err && err.name) || 'Error', - stack: err && typeof err.stack === 'string' ? err.stack : undefined, - }); } -})(); + +// Pooled job loop. Imports (sharp, exceljs, pdfkit, libdashstore, …) are paid ONCE for the +// lifetime of the thread; each subsequent job only builds a fresh NodeVM and runs the snippet, +// which is the ~2 ms part. Jobs are serialized (one `await` per message) so a pooled thread +// never interleaves two applets against the same VFS. +parentPort.on('message', (job) => { + void (async () => { + try { + await runJob(job); + } catch (err) { + // Setup-level failure (bad VFS dir, thrown bridge, etc.) — log it here too, because + // the pool only sees a message and this thread's stderr is otherwise swallowed. + console.error('[javascript-worker] fatal setup error:', (err && err.message) || String(err), '\n', (err && err.stack) || '(no stack)'); + report(job?.id, { + result: null, + stdout: '', + consoleLineCount: 0, + error: (err && err.message) || String(err), + errorName: (err && err.name) || 'Error', + stack: err && typeof err.stack === 'string' ? err.stack : undefined, + }); + } + })(); +}); + +// Ready handshake. All imports above have resolved by the time this module body executes, so +// signalling now tells the pool it can hand this thread a job without racing the import phase. +parentPort.postMessage({ ready: true }); diff --git a/src/tools/javascript.js b/src/tools/javascript.js index 94988eb..2ac83a8 100644 --- a/src/tools/javascript.js +++ b/src/tools/javascript.js @@ -1,109 +1,33 @@ -import { Worker } from 'worker_threads'; -import { fileURLToPath } from 'url'; import fs from 'fs'; import path from 'path'; import jwt from 'jsonwebtoken'; import { defineTool, ToolError } from 'libllmorbiter'; import env from '../config/env.js'; import { createVfs, prepareVfs, syncVfsToDb } from './vfs.js'; - -const __dirname = path.dirname(fileURLToPath(import.meta.url)); -const ROOT_DIR = path.resolve(__dirname, '../..'); -const WORKER_PATH = fileURLToPath(new URL('./javascript-worker.js', import.meta.url)); +import { runSandboxJob } from '../services/javascriptSandbox.js'; const DEFAULT_TIMEOUT_MS = 20_000; const MAX_TIMEOUT_MS = 60_000; const DEFAULT_MAX_OUTPUT = 20_000; -const DEFAULT_BUFFER_LIMIT = 32 * 1024 * 1024; - -// Keys already logged as benign non-sandbox worker chatter. Under `node --watch` the loader -// emits `watch:import` / `watch:require` notices on the worker's parentPort for EVERY import, -// so warning per message floods the log. We warn once per distinct key instead. -const WATCH_NOISE_WARNED = new Set(); +/** + * Runs one snippet/applet on the shared sandbox pool. + * + * Threads are keyed by VFS root so a reused thread keeps the right company's `vfsDir`, and the + * per-run JWT is minted here because it is scoped to THIS `googleAuthId`. + * + * @param {{ code: string, timeoutMs: number, vfsRoot: string, googleAuthId: number, appletInput?: object, appletFile?: string }} ctx + */ function runInWorker({ code, timeoutMs, vfsRoot, googleAuthId, appletInput, appletFile }) { - return new Promise((resolve) => { - let settled = false; - const token = jwt.sign({ id: googleAuthId }, env.jwtSecret, { expiresIn: '5m' }); - - const worker = new Worker(WORKER_PATH, { - workerData: { - code, - root: ROOT_DIR, - vmTimeout: timeoutMs, - bufferLimit: DEFAULT_BUFFER_LIMIT, - dashbackUrl: `http://127.0.0.1:${env.port}/api`, - dashbackToken: token, - vfsDir: vfsRoot, - googleAuthId, - appletInput, - appletFile, - }, - }); - - const finish = (payload) => { - if (settled) return; - settled = true; - clearTimeout(timer); - try { worker.terminate(); } catch { /* already closed */ } - resolve(payload); - }; - - const timer = setTimeout(() => { - console.error(`[javascript_execute] sandbox worker timed out after ${timeoutMs} ms`); - finish({ result: null, stdout: '', consoleLineCount: 0, timedOut: true, timeoutMs }); - }, timeoutMs); - - worker.on('message', (msg) => { - // Ignore anything that is not a genuine sandbox report. Running the server under - // `node --watch` (the `dev` script) installs a module-loader hook that posts - // `watch:import` notices on this same channel; treating the first message as the - // result made the sandbox appear to run for 0ms and return nothing. - if (!msg || msg.__sandboxReport !== true) { - // Those loader notices fire many times per run, so warn ONCE per key rather - // than on every message — expected, benign noise while running under --watch. - for (const key of Object.keys(msg || {})) { - if (!WATCH_NOISE_WARNED.has(key)) { - WATCH_NOISE_WARNED.add(key); - console.warn( - `[javascript_execute] ignoring non-sandbox worker message "${key}" ` + - '(expected under `node --watch`; logged once per key).' - ); - } - } - return; - } - finish(msg); - }); - // A worker 'error' fires BEFORE any message when the thread fails to boot (e.g. an - // import throws at startup). ALWAYS send a truthy `error` plus the full stack so the - // parent can throw with something actionable; an empty `err.message` must never - // degrade into a silent, result-less success payload. - worker.on('error', (err) => finish({ - result: null, - stdout: '', - consoleLineCount: 0, - error: (err && err.message) - ? err.message - : `Sandbox worker failed to start: ${err?.stack || err?.name || 'unknown error'}`, - errorName: (err && err.name) || 'Error', - stack: err && typeof err.stack === 'string' ? err.stack : undefined, - })); - worker.on('exit', (code) => { - // `finish` is idempotent via `settled`, so a message that already resolved the - // promise is never overwritten. Reaching here means the worker died WITHOUT - // reporting — that is a hard failure, so log it and always carry a truthy error. - if (!settled) { - console.error(`[javascript_execute] sandbox worker exited without reporting (code ${code})`); - } - finish({ - result: null, - stdout: '', - consoleLineCount: 0, - error: `Sandbox worker exited unexpectedly without reporting a result (code ${code}).`, - errorName: 'WorkerExit', - }); - }); + const dashbackToken = jwt.sign({ id: googleAuthId }, env.jwtSecret, { expiresIn: '5m' }); + return runSandboxJob({ + poolKey: vfsRoot, + googleAuthId, + code, + timeoutMs, + dashbackToken, + appletInput, + appletFile, }); }