This commit is contained in:
seb
2026-10-10 06:16:25 +02:00
parent eed10ea38d
commit 67678038c0
4 changed files with 574 additions and 189 deletions

View File

@@ -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;

View File

@@ -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/<companyId>`), 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<PoolEntry>} */
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<PoolEntry>}
*/
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<PoolEntry | null>}
*/
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<object>}
*/
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<object>} 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<void>}
*/
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;
}

View File

@@ -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 `<Buffer ${(a.byteLength ?? a.length ?? 0)} bytes>`;
}
try {
return JSON.stringify(a, (k, v) => {
if (v && typeof v === 'object') {
if (v.type === 'Buffer' && Array.isArray(v.data)) {
return `<Buffer ${v.data.length} bytes>`;
/**
* 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<string>} 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 `<Buffer ${(a.byteLength ?? a.length ?? 0)} bytes>`;
}
try {
return JSON.stringify(a, (k, v) => {
if (v && typeof v === 'object') {
if (v.type === 'Buffer' && Array.isArray(v.data)) {
return `<Buffer ${v.data.length} bytes>`;
}
if (Buffer.isBuffer(v) || ArrayBuffer.isView(v) || (v.constructor && v.constructor.name === 'Buffer')) {
return `<Buffer ${(v.byteLength ?? v.length ?? 0)} bytes>`;
}
}
if (Buffer.isBuffer(v) || ArrayBuffer.isView(v) || (v.constructor && v.constructor.name === 'Buffer')) {
return `<Buffer ${(v.byteLength ?? v.length ?? 0)} bytes>`;
}
}
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 });

View File

@@ -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,
});
}