mirror of
https://github.com/simstudioai/sim.git
synced 2026-09-02 07:30:01 +08:00
fix(execution): run pptx/docx/pdf generation inside isolated-vm sandbox
Retires the legacy doc-worker.cjs / pptx-worker.cjs pipeline that ran user
DSL via node:vm + full require() in the same UID/PID namespace as the main
Next.js process. User code now runs inside the existing isolated-vm pool
(V8 isolate, no process / require / fs, no /proc/1/environ reachability).
Introduces a first-class SandboxTask abstraction under apps/sim/sandbox-tasks/
that mirrors apps/sim/background/ — one file per task, central typed
registry, kebab-case ids. Adding a new thing that runs in the isolate is
one file plus one registry entry.
Runtime additions in lib/execution/:
- task-mode execution in isolated-vm-worker.cjs: load pre-built library
bundles, run task bootstrap, run user code, run finalize, transfer
Uint8Array result as base64 via IPC
- named broker IPC bridge (generalizes the existing fetch bridge) with
args size, result size, and per-execution call caps
- cooperative AbortSignal support: cancel IPC disposes the isolate, pool
slot is freed, pending broker-call timers are swept
- compiled scripts + references explicitly released per execution
- isolate.isDisposed used for cancellation detection (no error-string
substring matching)
Library bundles (pptxgenjs, docx, pdf-lib) are built into isolate-safe
IIFE bundles by apps/sim/lib/execution/sandbox/bundles/build.ts and
committed; next.config.ts / trigger.config.ts / Dockerfile updated to
ship them instead of the deleted dist/*-worker.cjs artifacts.
Call sites migrated:
- app/api/workspaces/[id]/pptx/preview/route.ts
- app/api/files/serve/[...path]/route.ts (+ test mock)
- lib/copilot/tools/server/files/{workspace-file,edit-content}.ts
All pass owner key user:<userId> for per-user pool fairness + distributed
lease accounting.
Made-with: Cursor
This commit is contained in:
Vikhyath Mondreti
@@ -75,10 +75,12 @@ vi.mock('@/lib/uploads/utils/file-utils', () => ({
|
||||
|
||||
vi.mock('@/lib/uploads/setup.server', () => ({}))
|
||||
|
||||
vi.mock('@/lib/execution/doc-vm', () => ({
|
||||
generatePdfFromCode: vi.fn().mockResolvedValue(Buffer.from('%PDF-compiled')),
|
||||
generateDocxFromCode: vi.fn().mockResolvedValue(Buffer.from('PK\x03\x04compiled')),
|
||||
generatePptxFromCode: vi.fn().mockResolvedValue(Buffer.from('PK\x03\x04compiled')),
|
||||
vi.mock('@/lib/execution/sandbox/run-task', () => ({
|
||||
runSandboxTask: vi
|
||||
.fn()
|
||||
.mockImplementation(async (taskId: string) =>
|
||||
taskId === 'pdf-generate' ? Buffer.from('%PDF-compiled') : Buffer.from('PK\x03\x04compiled')
|
||||
),
|
||||
}))
|
||||
|
||||
vi.mock('@/lib/uploads/contexts/workspace/workspace-file-manager', () => ({
|
||||
|
||||
@@ -4,11 +4,7 @@ import { createLogger } from '@sim/logger'
|
||||
import type { NextRequest } from 'next/server'
|
||||
import { NextResponse } from 'next/server'
|
||||
import { checkSessionOrInternalAuth } from '@/lib/auth/hybrid'
|
||||
import {
|
||||
generateDocxFromCode,
|
||||
generatePdfFromCode,
|
||||
generatePptxFromCode,
|
||||
} from '@/lib/execution/doc-vm'
|
||||
import { runSandboxTask } from '@/lib/execution/sandbox/run-task'
|
||||
import { CopilotFiles, isUsingCloudStorage } from '@/lib/uploads'
|
||||
import type { StorageContext } from '@/lib/uploads/config'
|
||||
import { parseWorkspaceFileKey } from '@/lib/uploads/contexts/workspace/workspace-file-manager'
|
||||
@@ -22,6 +18,7 @@ import {
|
||||
findLocalFile,
|
||||
getContentType,
|
||||
} from '@/app/api/files/utils'
|
||||
import type { SandboxTaskId } from '@/sandbox-tasks/registry'
|
||||
|
||||
const logger = createLogger('FilesServeAPI')
|
||||
|
||||
@@ -30,24 +27,24 @@ const PDF_MAGIC = Buffer.from([0x25, 0x50, 0x44, 0x46, 0x2d]) // %PDF-
|
||||
|
||||
interface CompilableFormat {
|
||||
magic: Buffer
|
||||
compile: (code: string, workspaceId: string) => Promise<Buffer>
|
||||
taskId: SandboxTaskId
|
||||
contentType: string
|
||||
}
|
||||
|
||||
const COMPILABLE_FORMATS: Record<string, CompilableFormat> = {
|
||||
'.pptx': {
|
||||
magic: ZIP_MAGIC,
|
||||
compile: generatePptxFromCode,
|
||||
taskId: 'pptx-generate',
|
||||
contentType: 'application/vnd.openxmlformats-officedocument.presentationml.presentation',
|
||||
},
|
||||
'.docx': {
|
||||
magic: ZIP_MAGIC,
|
||||
compile: generateDocxFromCode,
|
||||
taskId: 'docx-generate',
|
||||
contentType: 'application/vnd.openxmlformats-officedocument.wordprocessingml.document',
|
||||
},
|
||||
'.pdf': {
|
||||
magic: PDF_MAGIC,
|
||||
compile: generatePdfFromCode,
|
||||
taskId: 'pdf-generate',
|
||||
contentType: 'application/pdf',
|
||||
},
|
||||
}
|
||||
@@ -65,8 +62,9 @@ function compiledCacheSet(key: string, buffer: Buffer): void {
|
||||
async function compileDocumentIfNeeded(
|
||||
buffer: Buffer,
|
||||
filename: string,
|
||||
workspaceId?: string,
|
||||
raw?: boolean
|
||||
workspaceId: string | undefined,
|
||||
raw: boolean,
|
||||
ownerKey: string | undefined
|
||||
): Promise<{ buffer: Buffer; contentType: string }> {
|
||||
if (raw) return { buffer, contentType: getContentType(filename) }
|
||||
|
||||
@@ -90,7 +88,11 @@ async function compileDocumentIfNeeded(
|
||||
return { buffer: cached, contentType: format.contentType }
|
||||
}
|
||||
|
||||
const compiled = await format.compile(code, workspaceId || '')
|
||||
const compiled = await runSandboxTask(
|
||||
format.taskId,
|
||||
{ code, workspaceId: workspaceId || '' },
|
||||
{ ownerKey }
|
||||
)
|
||||
compiledCacheSet(cacheKey, compiled)
|
||||
return { buffer: compiled, contentType: format.contentType }
|
||||
}
|
||||
@@ -173,6 +175,7 @@ async function handleLocalFile(
|
||||
userId: string,
|
||||
raw: boolean
|
||||
): Promise<NextResponse> {
|
||||
const ownerKey = `user:${userId}`
|
||||
try {
|
||||
const contextParam: StorageContext | undefined = inferContextFromKey(filename) as
|
||||
| StorageContext
|
||||
@@ -205,7 +208,8 @@ async function handleLocalFile(
|
||||
rawBuffer,
|
||||
displayName,
|
||||
workspaceId,
|
||||
raw
|
||||
raw,
|
||||
ownerKey
|
||||
)
|
||||
|
||||
logger.info('Local file served', { userId, filename, size: fileBuffer.length })
|
||||
@@ -227,6 +231,7 @@ async function handleCloudProxy(
|
||||
userId: string,
|
||||
raw = false
|
||||
): Promise<NextResponse> {
|
||||
const ownerKey = `user:${userId}`
|
||||
try {
|
||||
const context = inferContextFromKey(cloudKey)
|
||||
logger.info(`Inferred context: ${context} from key pattern: ${cloudKey}`)
|
||||
@@ -262,7 +267,8 @@ async function handleCloudProxy(
|
||||
rawBuffer,
|
||||
displayName,
|
||||
workspaceId,
|
||||
raw
|
||||
raw,
|
||||
ownerKey
|
||||
)
|
||||
|
||||
logger.info('Cloud file served', {
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
import { createLogger } from '@sim/logger'
|
||||
import { type NextRequest, NextResponse } from 'next/server'
|
||||
import { getSession } from '@/lib/auth'
|
||||
import { generatePptxFromCode } from '@/lib/execution/doc-vm'
|
||||
import { runSandboxTask } from '@/lib/execution/sandbox/run-task'
|
||||
import { verifyWorkspaceMembership } from '@/app/api/workflows/utils'
|
||||
|
||||
export const dynamic = 'force-dynamic'
|
||||
@@ -44,7 +44,11 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
|
||||
return NextResponse.json({ error: 'code exceeds maximum size' }, { status: 413 })
|
||||
}
|
||||
|
||||
const buffer = await generatePptxFromCode(code, workspaceId, req.signal)
|
||||
const buffer = await runSandboxTask(
|
||||
'pptx-generate',
|
||||
{ code, workspaceId },
|
||||
{ ownerKey: `user:${session.user.id}`, signal: req.signal }
|
||||
)
|
||||
|
||||
return new NextResponse(new Uint8Array(buffer), {
|
||||
status: 200,
|
||||
|
||||
@@ -4,12 +4,9 @@ import {
|
||||
type BaseServerTool,
|
||||
type ServerToolContext,
|
||||
} from '@/lib/copilot/tools/server/base-tool'
|
||||
import {
|
||||
generateDocxFromCode,
|
||||
generatePdfFromCode,
|
||||
generatePptxFromCode,
|
||||
} from '@/lib/execution/doc-vm'
|
||||
import { runSandboxTask } from '@/lib/execution/sandbox/run-task'
|
||||
import { updateWorkspaceFileContent } from '@/lib/uploads/contexts/workspace/workspace-file-manager'
|
||||
import type { SandboxTaskId } from '@/sandbox-tasks/registry'
|
||||
import { consumeLatestFileIntent } from './file-intent-store'
|
||||
import { inferContentType } from './workspace-file'
|
||||
|
||||
@@ -29,7 +26,7 @@ function getDocumentFormatInfo(fileName: string): {
|
||||
isDoc: boolean
|
||||
formatName?: string
|
||||
sourceMime?: string
|
||||
generator?: (code: string, workspaceId: string, signal?: AbortSignal) => Promise<Buffer>
|
||||
taskId?: SandboxTaskId
|
||||
} {
|
||||
const lowerName = fileName.toLowerCase()
|
||||
if (lowerName.endsWith('.pptx')) {
|
||||
@@ -37,7 +34,7 @@ function getDocumentFormatInfo(fileName: string): {
|
||||
isDoc: true,
|
||||
formatName: 'PPTX',
|
||||
sourceMime: 'text/x-pptxgenjs',
|
||||
generator: generatePptxFromCode,
|
||||
taskId: 'pptx-generate',
|
||||
}
|
||||
}
|
||||
if (lowerName.endsWith('.docx')) {
|
||||
@@ -45,7 +42,7 @@ function getDocumentFormatInfo(fileName: string): {
|
||||
isDoc: true,
|
||||
formatName: 'DOCX',
|
||||
sourceMime: 'text/x-docxjs',
|
||||
generator: generateDocxFromCode,
|
||||
taskId: 'docx-generate',
|
||||
}
|
||||
}
|
||||
if (lowerName.endsWith('.pdf')) {
|
||||
@@ -53,7 +50,7 @@ function getDocumentFormatInfo(fileName: string): {
|
||||
isDoc: true,
|
||||
formatName: 'PDF',
|
||||
sourceMime: 'text/x-pdflibjs',
|
||||
generator: generatePdfFromCode,
|
||||
taskId: 'pdf-generate',
|
||||
}
|
||||
}
|
||||
return { isDoc: false }
|
||||
@@ -240,7 +237,11 @@ export const editContentServerTool: BaseServerTool<EditContentArgs, EditContentR
|
||||
|
||||
if (docInfo.isDoc) {
|
||||
try {
|
||||
await docInfo.generator!(finalContent, workspaceId)
|
||||
await runSandboxTask(
|
||||
docInfo.taskId!,
|
||||
{ code: finalContent, workspaceId },
|
||||
{ ownerKey: `user:${context.userId}` }
|
||||
)
|
||||
} catch (err) {
|
||||
const msg = err instanceof Error ? err.message : String(err)
|
||||
return {
|
||||
|
||||
@@ -5,11 +5,7 @@ import {
|
||||
type BaseServerTool,
|
||||
type ServerToolContext,
|
||||
} from '@/lib/copilot/tools/server/base-tool'
|
||||
import {
|
||||
generateDocxFromCode,
|
||||
generatePdfFromCode,
|
||||
generatePptxFromCode,
|
||||
} from '@/lib/execution/doc-vm'
|
||||
import { runSandboxTask } from '@/lib/execution/sandbox/run-task'
|
||||
import {
|
||||
deleteWorkspaceFile,
|
||||
downloadWorkspaceFile as downloadWsFile,
|
||||
@@ -18,6 +14,7 @@ import {
|
||||
renameWorkspaceFile,
|
||||
uploadWorkspaceFile,
|
||||
} from '@/lib/uploads/contexts/workspace/workspace-file-manager'
|
||||
import type { SandboxTaskId } from '@/sandbox-tasks/registry'
|
||||
import { storeFileIntent } from './file-intent-store'
|
||||
|
||||
const logger = createLogger('WorkspaceFileServerTool')
|
||||
@@ -108,7 +105,7 @@ function getDocumentFormatInfo(fileName: string): {
|
||||
isDoc: boolean
|
||||
formatName?: 'PPTX' | 'DOCX' | 'PDF'
|
||||
sourceMime?: string
|
||||
generator?: (code: string, workspaceId: string, signal?: AbortSignal) => Promise<Buffer>
|
||||
taskId?: SandboxTaskId
|
||||
} {
|
||||
const lowerName = fileName.toLowerCase()
|
||||
if (lowerName.endsWith('.pptx')) {
|
||||
@@ -116,7 +113,7 @@ function getDocumentFormatInfo(fileName: string): {
|
||||
isDoc: true,
|
||||
formatName: 'PPTX',
|
||||
sourceMime: PPTX_SOURCE_MIME,
|
||||
generator: generatePptxFromCode,
|
||||
taskId: 'pptx-generate',
|
||||
}
|
||||
}
|
||||
if (lowerName.endsWith('.docx')) {
|
||||
@@ -124,7 +121,7 @@ function getDocumentFormatInfo(fileName: string): {
|
||||
isDoc: true,
|
||||
formatName: 'DOCX',
|
||||
sourceMime: DOCX_SOURCE_MIME,
|
||||
generator: generateDocxFromCode,
|
||||
taskId: 'docx-generate',
|
||||
}
|
||||
}
|
||||
if (lowerName.endsWith('.pdf')) {
|
||||
@@ -132,7 +129,7 @@ function getDocumentFormatInfo(fileName: string): {
|
||||
isDoc: true,
|
||||
formatName: 'PDF',
|
||||
sourceMime: PDF_SOURCE_MIME,
|
||||
generator: generatePdfFromCode,
|
||||
taskId: 'pdf-generate',
|
||||
}
|
||||
}
|
||||
return { isDoc: false }
|
||||
@@ -201,7 +198,11 @@ export const workspaceFileServerTool: BaseServerTool<WorkspaceFileArgs, Workspac
|
||||
let contentType = inferContentType(fileName, explicitType)
|
||||
if (docInfo.isDoc) {
|
||||
try {
|
||||
await docInfo.generator!(content, workspaceId)
|
||||
await runSandboxTask(
|
||||
docInfo.taskId!,
|
||||
{ code: content, workspaceId },
|
||||
{ ownerKey: `user:${context.userId}` }
|
||||
)
|
||||
} catch (err) {
|
||||
const msg = err instanceof Error ? err.message : String(err)
|
||||
return {
|
||||
|
||||
@@ -232,6 +232,9 @@ export const env = createEnv({
|
||||
IVM_DISTRIBUTED_LEASE_MIN_TTL_MS: z.string().optional().default('120000'), // Min TTL for distributed in-flight leases (ms)
|
||||
IVM_QUEUE_TIMEOUT_MS: z.string().optional().default('300000'), // Max queue wait before rejection (ms)
|
||||
IVM_MAX_EXECUTIONS_PER_WORKER: z.string().optional().default('500'), // Max lifetime executions before worker is recycled
|
||||
IVM_MAX_BROKER_ARGS_JSON_CHARS: z.string().optional().default('262144'), // Max JSON payload size for sandbox task broker args (isolate→host)
|
||||
IVM_MAX_BROKER_RESULT_JSON_CHARS: z.string().optional().default('16777216'),// Max JSON payload size for sandbox task broker results (host→isolate)
|
||||
IVM_MAX_BROKERS_PER_EXECUTION: z.string().optional().default('1000'), // Max broker calls per sandbox task execution
|
||||
|
||||
// Knowledge Base Processing Configuration - Shared across all processing methods
|
||||
KB_CONFIG_MAX_DURATION: z.number().optional().default(600), // Max processing duration in seconds (10 minutes)
|
||||
|
||||
@@ -1,211 +0,0 @@
|
||||
/**
|
||||
* Sandboxed document generation via subprocess.
|
||||
*
|
||||
* Supports pptx (pptxgenjs), docx (docx), and pdf (pdf-lib).
|
||||
* User code runs in a separate Node.js child process. File access is brokered
|
||||
* via IPC -- the subprocess never touches the database directly.
|
||||
*/
|
||||
|
||||
import { type ChildProcess, spawn } from 'node:child_process'
|
||||
import fs from 'node:fs'
|
||||
import path from 'node:path'
|
||||
import { fileURLToPath } from 'node:url'
|
||||
import { createLogger } from '@sim/logger'
|
||||
import {
|
||||
downloadWorkspaceFile,
|
||||
getWorkspaceFile,
|
||||
} from '@/lib/uploads/contexts/workspace/workspace-file-manager'
|
||||
|
||||
const logger = createLogger('DocVMExecution')
|
||||
|
||||
export type DocumentFormat = 'pptx' | 'docx' | 'pdf'
|
||||
|
||||
const WORKER_STARTUP_TIMEOUT_MS = 10_000
|
||||
const GENERATION_TIMEOUT_MS = 60_000
|
||||
const MAX_STDERR = 4096
|
||||
|
||||
type WorkerMessage =
|
||||
| { type: 'ready' }
|
||||
| { type: 'result'; data: string }
|
||||
| { type: 'error'; message: string }
|
||||
| { type: 'getFile'; fileReqId: number; fileId: string }
|
||||
|
||||
const currentDir = path.dirname(fileURLToPath(import.meta.url))
|
||||
let cachedWorkerPath: string | undefined
|
||||
|
||||
function getWorkerPath(): string {
|
||||
if (cachedWorkerPath) return cachedWorkerPath
|
||||
const candidates = [
|
||||
path.join(currentDir, '..', '..', 'dist', 'doc-worker.cjs'),
|
||||
path.join(currentDir, 'doc-worker.cjs'),
|
||||
path.join(process.cwd(), 'apps', 'sim', 'dist', 'doc-worker.cjs'),
|
||||
path.join(process.cwd(), 'apps', 'sim', 'lib', 'execution', 'doc-worker.cjs'),
|
||||
path.join(process.cwd(), 'dist', 'doc-worker.cjs'),
|
||||
path.join(process.cwd(), 'lib', 'execution', 'doc-worker.cjs'),
|
||||
]
|
||||
const found = candidates.find((p) => fs.existsSync(p))
|
||||
if (!found) throw new Error(`doc-worker.cjs not found at any of: ${candidates.join(', ')}`)
|
||||
cachedWorkerPath = found
|
||||
return found
|
||||
}
|
||||
|
||||
export async function generateDocumentFromCode(
|
||||
format: DocumentFormat,
|
||||
code: string,
|
||||
workspaceId: string,
|
||||
signal?: AbortSignal
|
||||
): Promise<Buffer> {
|
||||
return new Promise<Buffer>((resolve, reject) => {
|
||||
let proc: ChildProcess | null = null
|
||||
let settled = false
|
||||
let startupTimer: ReturnType<typeof setTimeout> | null = null
|
||||
let generationTimer: ReturnType<typeof setTimeout> | null = null
|
||||
|
||||
function done(err: Error): void
|
||||
function done(err: undefined, result: Buffer): void
|
||||
function done(err: Error | undefined, result?: Buffer): void {
|
||||
if (settled) return
|
||||
settled = true
|
||||
if (startupTimer) clearTimeout(startupTimer)
|
||||
if (generationTimer) clearTimeout(generationTimer)
|
||||
try {
|
||||
proc?.removeAllListeners()
|
||||
proc?.kill()
|
||||
} catch {
|
||||
// Ignore -- process may have already exited
|
||||
}
|
||||
if (err) reject(err)
|
||||
else resolve(result as Buffer)
|
||||
}
|
||||
|
||||
if (signal?.aborted) {
|
||||
reject(new Error(`${format.toUpperCase()} generation cancelled`))
|
||||
return
|
||||
}
|
||||
|
||||
signal?.addEventListener(
|
||||
'abort',
|
||||
() => done(new Error(`${format.toUpperCase()} generation cancelled`)),
|
||||
{ once: true }
|
||||
)
|
||||
|
||||
try {
|
||||
proc = spawn('node', [getWorkerPath()], {
|
||||
stdio: ['ignore', 'pipe', 'pipe', 'ipc'],
|
||||
serialization: 'json',
|
||||
env: { PATH: process.env.PATH ?? '' } as unknown as NodeJS.ProcessEnv,
|
||||
})
|
||||
} catch (err) {
|
||||
done(err instanceof Error ? err : new Error(String(err)))
|
||||
return
|
||||
}
|
||||
|
||||
let stderrData = ''
|
||||
proc.stderr?.on('data', (chunk: Buffer) => {
|
||||
if (stderrData.length < MAX_STDERR) {
|
||||
stderrData += chunk.toString()
|
||||
if (stderrData.length > MAX_STDERR) stderrData = stderrData.slice(0, MAX_STDERR)
|
||||
}
|
||||
})
|
||||
|
||||
startupTimer = setTimeout(() => {
|
||||
logger.error(`${format} worker failed to start within timeout`)
|
||||
done(new Error(`${format.toUpperCase()} worker failed to start`))
|
||||
}, WORKER_STARTUP_TIMEOUT_MS)
|
||||
|
||||
proc.on('exit', (code) => {
|
||||
if (!settled) {
|
||||
logger.error(`${format} worker exited unexpectedly`, {
|
||||
code,
|
||||
stderr: stderrData.slice(0, 500),
|
||||
})
|
||||
done(new Error(`${format.toUpperCase()} worker exited unexpectedly (code ${code})`))
|
||||
}
|
||||
})
|
||||
|
||||
proc.on('error', (err) => {
|
||||
logger.error(`${format} worker process error`, { error: err.message })
|
||||
done(err)
|
||||
})
|
||||
|
||||
proc.on('message', (rawMsg: unknown) => {
|
||||
const msg = rawMsg as WorkerMessage
|
||||
|
||||
if (msg.type === 'ready') {
|
||||
if (startupTimer) {
|
||||
clearTimeout(startupTimer)
|
||||
startupTimer = null
|
||||
}
|
||||
generationTimer = setTimeout(() => {
|
||||
logger.error(`${format} generation timed out`)
|
||||
done(new Error(`${format.toUpperCase()} generation timed out`))
|
||||
}, GENERATION_TIMEOUT_MS)
|
||||
proc!.send({ type: 'generate', format, code })
|
||||
return
|
||||
}
|
||||
|
||||
if (msg.type === 'result') {
|
||||
done(undefined, Buffer.from(msg.data, 'base64'))
|
||||
return
|
||||
}
|
||||
|
||||
if (msg.type === 'error') {
|
||||
done(new Error(msg.message))
|
||||
return
|
||||
}
|
||||
|
||||
if (msg.type === 'getFile') {
|
||||
handleFileRequest(proc!, workspaceId, msg).catch((err) => {
|
||||
logger.error(`Failed to handle file request from ${format} worker`, {
|
||||
fileId: msg.fileId,
|
||||
error: err instanceof Error ? err.message : String(err),
|
||||
})
|
||||
if (proc && !settled) {
|
||||
try {
|
||||
proc.send({
|
||||
type: 'fileResult',
|
||||
fileReqId: msg.fileReqId,
|
||||
error: err instanceof Error ? err.message : 'File fetch failed',
|
||||
})
|
||||
} catch {
|
||||
// Ignore -- process may have died
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
async function handleFileRequest(
|
||||
proc: ChildProcess,
|
||||
workspaceId: string,
|
||||
msg: Extract<WorkerMessage, { type: 'getFile' }>
|
||||
): Promise<void> {
|
||||
const record = await getWorkspaceFile(workspaceId, msg.fileId)
|
||||
if (!record) {
|
||||
proc.send({
|
||||
type: 'fileResult',
|
||||
fileReqId: msg.fileReqId,
|
||||
error: `File not found: ${msg.fileId}`,
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
const buffer = await downloadWorkspaceFile(record)
|
||||
const mime = record.type || 'image/png'
|
||||
proc.send({
|
||||
type: 'fileResult',
|
||||
fileReqId: msg.fileReqId,
|
||||
data: `data:${mime};base64,${buffer.toString('base64')}`,
|
||||
})
|
||||
}
|
||||
|
||||
export const generatePptxFromCode = (code: string, workspaceId: string, signal?: AbortSignal) =>
|
||||
generateDocumentFromCode('pptx', code, workspaceId, signal)
|
||||
|
||||
export const generateDocxFromCode = (code: string, workspaceId: string, signal?: AbortSignal) =>
|
||||
generateDocumentFromCode('docx', code, workspaceId, signal)
|
||||
|
||||
export const generatePdfFromCode = (code: string, workspaceId: string, signal?: AbortSignal) =>
|
||||
generateDocumentFromCode('pdf', code, workspaceId, signal)
|
||||
@@ -1,162 +0,0 @@
|
||||
/**
|
||||
* Generic document generation worker.
|
||||
* Runs in a separate Node.js process, communicates with parent via IPC.
|
||||
* Supports pptx (pptxgenjs), docx (docx), and pdf (pdf-lib).
|
||||
*/
|
||||
|
||||
'use strict'
|
||||
|
||||
const vm = require('node:vm')
|
||||
|
||||
const EXECUTION_TIMEOUT_MS = 30_000
|
||||
const FILE_REQUEST_TIMEOUT_MS = 30_000
|
||||
|
||||
const FORMATS = {
|
||||
pptx: {
|
||||
setup() {
|
||||
const PptxGenJS = require('pptxgenjs')
|
||||
const pptx = new PptxGenJS()
|
||||
return { globals: { pptx }, pptx }
|
||||
},
|
||||
async serialize(ctx) {
|
||||
const output = await ctx.pptx.write({ outputType: 'nodebuffer' })
|
||||
return Buffer.from(output)
|
||||
},
|
||||
},
|
||||
docx: {
|
||||
setup() {
|
||||
const docx = require('docx')
|
||||
const _sections = []
|
||||
return { globals: { docx, addSection: (s) => _sections.push(s) }, _sections, docx }
|
||||
},
|
||||
async serialize(ctx) {
|
||||
if (ctx.globals.doc) {
|
||||
return ctx.docx.Packer.toBuffer(ctx.globals.doc)
|
||||
}
|
||||
if (ctx._sections.length > 0) {
|
||||
const doc = new ctx.docx.Document({ sections: ctx._sections })
|
||||
return ctx.docx.Packer.toBuffer(doc)
|
||||
}
|
||||
throw new Error(
|
||||
'No document created. Use addSection({ children: [...] }) for chunked writes, or set doc = new docx.Document({...}) for a single write.'
|
||||
)
|
||||
},
|
||||
},
|
||||
pdf: {
|
||||
async setup() {
|
||||
const PDFLib = require('pdf-lib')
|
||||
const pdf = await PDFLib.PDFDocument.create()
|
||||
|
||||
async function embedImage(dataUri) {
|
||||
const base64 = dataUri.split(',')[1]
|
||||
const bytes = Buffer.from(base64, 'base64')
|
||||
const mime = dataUri.split(';')[0].split(':')[1] || ''
|
||||
if (mime.includes('png')) return pdf.embedPng(bytes)
|
||||
return pdf.embedJpg(bytes)
|
||||
}
|
||||
|
||||
return { globals: { PDFLib, pdf, embedImage }, pdf }
|
||||
},
|
||||
async serialize(ctx) {
|
||||
const pdf = ctx.globals.pdf
|
||||
if (!pdf)
|
||||
throw new Error(
|
||||
'No PDF document. Use the injected pdf object or load one with PDFLib.PDFDocument.load().'
|
||||
)
|
||||
const bytes = await pdf.save()
|
||||
return Buffer.from(bytes)
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
const pendingFileRequests = new Map()
|
||||
let fileRequestCounter = 0
|
||||
|
||||
function sendToParent(msg) {
|
||||
if (process.send && process.connected) {
|
||||
process.send(msg)
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
process.on('message', async (msg) => {
|
||||
if (msg.type === 'generate') {
|
||||
await handleGenerate(msg)
|
||||
} else if (msg.type === 'fileResult') {
|
||||
handleFileResult(msg)
|
||||
}
|
||||
})
|
||||
|
||||
async function handleGenerate(msg) {
|
||||
const { code, format } = msg
|
||||
|
||||
try {
|
||||
const formatConfig = FORMATS[format]
|
||||
if (!formatConfig) throw new Error(`Unknown document format: ${format}`)
|
||||
|
||||
const ctx = await formatConfig.setup()
|
||||
|
||||
const getFileBase64 = (fileId) =>
|
||||
new Promise((resolve, reject) => {
|
||||
if (typeof fileId !== 'string' || fileId.length === 0) {
|
||||
reject(new Error('fileId must be a non-empty string'))
|
||||
return
|
||||
}
|
||||
|
||||
const fileReqId = ++fileRequestCounter
|
||||
const timeout = setTimeout(() => {
|
||||
if (pendingFileRequests.has(fileReqId)) {
|
||||
pendingFileRequests.delete(fileReqId)
|
||||
reject(new Error(`File request timed out for fileId: ${fileId}`))
|
||||
}
|
||||
}, FILE_REQUEST_TIMEOUT_MS)
|
||||
|
||||
pendingFileRequests.set(fileReqId, { resolve, reject, timeout })
|
||||
|
||||
if (!sendToParent({ type: 'getFile', fileReqId, fileId })) {
|
||||
clearTimeout(timeout)
|
||||
pendingFileRequests.delete(fileReqId)
|
||||
reject(new Error('Parent process disconnected'))
|
||||
}
|
||||
})
|
||||
|
||||
const sandbox = Object.create(null)
|
||||
Object.assign(sandbox, ctx.globals)
|
||||
sandbox.getFileBase64 = getFileBase64
|
||||
|
||||
vm.createContext(sandbox)
|
||||
|
||||
const promise = vm.runInContext(`(async () => { ${code} })()`, sandbox, {
|
||||
timeout: EXECUTION_TIMEOUT_MS,
|
||||
filename: `${format}-code.js`,
|
||||
})
|
||||
await promise
|
||||
|
||||
ctx.globals = sandbox
|
||||
|
||||
const output = await formatConfig.serialize(ctx)
|
||||
const base64 = Buffer.from(output).toString('base64')
|
||||
sendToParent({ type: 'result', data: base64 })
|
||||
} catch (err) {
|
||||
const message = err instanceof Error ? err.message : String(err)
|
||||
sendToParent({ type: 'error', message })
|
||||
}
|
||||
}
|
||||
|
||||
function handleFileResult(msg) {
|
||||
const { fileReqId, data, error } = msg
|
||||
const pending = pendingFileRequests.get(fileReqId)
|
||||
if (!pending) return
|
||||
|
||||
clearTimeout(pending.timeout)
|
||||
pendingFileRequests.delete(fileReqId)
|
||||
|
||||
if (error) {
|
||||
pending.reject(new Error(error))
|
||||
} else {
|
||||
pending.resolve(data)
|
||||
}
|
||||
}
|
||||
|
||||
sendToParent({ type: 'ready' })
|
||||
@@ -4,15 +4,47 @@
|
||||
*/
|
||||
|
||||
const ivm = require('isolated-vm')
|
||||
const fs = require('node:fs')
|
||||
const path = require('node:path')
|
||||
|
||||
const USER_CODE_START_LINE = 4
|
||||
const pendingFetches = new Map()
|
||||
let fetchIdCounter = 0
|
||||
const pendingBrokerCalls = new Map()
|
||||
let brokerIdCounter = 0
|
||||
const FETCH_TIMEOUT_MS = 300000 // 5 minutes
|
||||
const BROKER_TIMEOUT_MS = 300000
|
||||
const MAX_STDOUT_CHARS = Number.parseInt(process.env.IVM_MAX_STDOUT_CHARS || '', 10) || 200000
|
||||
const MAX_FETCH_OPTIONS_JSON_CHARS =
|
||||
Number.parseInt(process.env.IVM_MAX_FETCH_OPTIONS_JSON_CHARS || '', 10) || 256 * 1024
|
||||
|
||||
const SANDBOX_BUNDLE_DIR = path.join(__dirname, 'sandbox', 'bundles')
|
||||
const SANDBOX_BUNDLE_FILES = {
|
||||
pptxgenjs: 'pptxgenjs.cjs',
|
||||
docx: 'docx.cjs',
|
||||
'pdf-lib': 'pdf-lib.cjs',
|
||||
}
|
||||
const bundleSourceCache = new Map()
|
||||
const activeIsolates = new Map()
|
||||
|
||||
function getBundleSource(bundleName) {
|
||||
const cached = bundleSourceCache.get(bundleName)
|
||||
if (cached) return cached
|
||||
const fileName = SANDBOX_BUNDLE_FILES[bundleName]
|
||||
if (!fileName) {
|
||||
throw new Error(`Unknown sandbox bundle: ${bundleName}`)
|
||||
}
|
||||
const bundlePath = path.join(SANDBOX_BUNDLE_DIR, fileName)
|
||||
if (!fs.existsSync(bundlePath)) {
|
||||
throw new Error(
|
||||
`Sandbox bundle not found at ${bundlePath}. Run \`bun run build:sandbox-bundles\`.`
|
||||
)
|
||||
}
|
||||
const source = fs.readFileSync(bundlePath, 'utf-8')
|
||||
bundleSourceCache.set(bundleName, { source, fileName })
|
||||
return bundleSourceCache.get(bundleName)
|
||||
}
|
||||
|
||||
function stringifyLogValue(value) {
|
||||
if (typeof value !== 'object' || value === null) {
|
||||
return String(value)
|
||||
@@ -113,7 +145,7 @@ function convertToCompatibleError(errorInfo, userCode) {
|
||||
/**
|
||||
* Execute code in isolated-vm
|
||||
*/
|
||||
async function executeCode(request) {
|
||||
async function executeCode(request, executionId) {
|
||||
const { code, params, envVars, contextVariables, timeoutMs, requestId } = request
|
||||
const stdoutChunks = []
|
||||
let stdoutLength = 0
|
||||
@@ -152,6 +184,7 @@ async function executeCode(request) {
|
||||
|
||||
try {
|
||||
isolate = new ivm.Isolate({ memoryLimit: 128 })
|
||||
if (executionId !== undefined) activeIsolates.set(executionId, isolate)
|
||||
context = await isolate.createContext()
|
||||
const jail = context.global
|
||||
|
||||
@@ -343,6 +376,17 @@ async function executeCode(request) {
|
||||
stack: err.stack,
|
||||
}
|
||||
|
||||
// Host sent a `cancel` IPC which called `isolate.dispose()`. Any
|
||||
// in-flight compileScript/run then throws; detect that authoritatively
|
||||
// via the isolate flag rather than fuzzy-matching the error message.
|
||||
if (isolate && isolate.isDisposed) {
|
||||
return {
|
||||
result: null,
|
||||
stdout,
|
||||
error: { message: 'Execution cancelled', name: 'AbortError' },
|
||||
}
|
||||
}
|
||||
|
||||
if (err.message.includes('Script execution timed out')) {
|
||||
return {
|
||||
result: null,
|
||||
@@ -393,16 +437,339 @@ async function executeCode(request) {
|
||||
isolate.dispose()
|
||||
} catch {}
|
||||
}
|
||||
if (executionId !== undefined) activeIsolates.delete(executionId)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Task-mode execution. Loads pre-built library bundles into the isolate,
|
||||
* exposes host-side brokers as isolate globals under `__brokers.<name>(args)`,
|
||||
* runs the task bootstrap (which installs friendly names on globalThis),
|
||||
* executes user code, then runs `finalize` (must return a Uint8Array). The
|
||||
* resulting bytes are returned as base64 in `bytesBase64`.
|
||||
*/
|
||||
async function executeTask(request, executionId) {
|
||||
const { code, timeoutMs, task } = request
|
||||
const stdoutChunks = []
|
||||
let stdoutLength = 0
|
||||
let stdoutTruncated = false
|
||||
let isolate = null
|
||||
|
||||
const appendStdout = (line) => {
|
||||
if (stdoutTruncated || !line) return
|
||||
const remaining = MAX_STDOUT_CHARS - stdoutLength
|
||||
if (remaining <= 0) {
|
||||
stdoutTruncated = true
|
||||
stdoutChunks.push('[stdout truncated]\n')
|
||||
return
|
||||
}
|
||||
if (line.length <= remaining) {
|
||||
stdoutChunks.push(line)
|
||||
stdoutLength += line.length
|
||||
return
|
||||
}
|
||||
stdoutChunks.push(line.slice(0, remaining))
|
||||
stdoutChunks.push('\n[stdout truncated]\n')
|
||||
stdoutLength = MAX_STDOUT_CHARS
|
||||
stdoutTruncated = true
|
||||
}
|
||||
|
||||
let context = null
|
||||
const releaseables = []
|
||||
|
||||
try {
|
||||
isolate = new ivm.Isolate({ memoryLimit: 128 })
|
||||
if (executionId !== undefined) activeIsolates.set(executionId, isolate)
|
||||
context = await isolate.createContext()
|
||||
const jail = context.global
|
||||
|
||||
await jail.set('global', jail.derefInto())
|
||||
|
||||
const logCallback = new ivm.Callback((...args) => {
|
||||
const message = args.map((arg) => stringifyLogValue(arg)).join(' ')
|
||||
appendStdout(`${message}\n`)
|
||||
})
|
||||
releaseables.push(logCallback)
|
||||
await jail.set('__log', logCallback)
|
||||
|
||||
const errorCallback = new ivm.Callback((...args) => {
|
||||
const message = args.map((arg) => stringifyLogValue(arg)).join(' ')
|
||||
appendStdout(`ERROR: ${message}\n`)
|
||||
})
|
||||
releaseables.push(errorCallback)
|
||||
await jail.set('__error', errorCallback)
|
||||
|
||||
const brokerRef = new ivm.Reference(async (brokerName, argsJson) => {
|
||||
return new Promise((resolve) => {
|
||||
const brokerId = ++brokerIdCounter
|
||||
const timeout = setTimeout(() => {
|
||||
if (pendingBrokerCalls.has(brokerId)) {
|
||||
pendingBrokerCalls.delete(brokerId)
|
||||
resolve(JSON.stringify({ error: `Broker "${brokerName}" timed out` }))
|
||||
}
|
||||
}, BROKER_TIMEOUT_MS)
|
||||
pendingBrokerCalls.set(brokerId, { resolve, timeout, executionId })
|
||||
if (process.send && process.connected) {
|
||||
process.send({ type: 'broker', brokerId, executionId, brokerName, argsJson })
|
||||
} else {
|
||||
clearTimeout(timeout)
|
||||
pendingBrokerCalls.delete(brokerId)
|
||||
resolve(JSON.stringify({ error: 'Parent process disconnected' }))
|
||||
}
|
||||
})
|
||||
})
|
||||
releaseables.push(brokerRef)
|
||||
await jail.set('__brokerRef', brokerRef)
|
||||
|
||||
const consoleBootstrap = `
|
||||
// Capture log callbacks in a closure so later hardening can unset the
|
||||
// raw __log / __error globals without breaking console in user code.
|
||||
(() => {
|
||||
const __log = globalThis.__log;
|
||||
const __error = globalThis.__error;
|
||||
globalThis.console = {
|
||||
log: (...args) => __log(...args),
|
||||
error: (...args) => __error(...args),
|
||||
warn: (...args) => __log('WARN:', ...args),
|
||||
info: (...args) => __log(...args),
|
||||
debug: (...args) => __log(...args),
|
||||
};
|
||||
})();
|
||||
`
|
||||
const consoleScript = await isolate.compileScript(consoleBootstrap)
|
||||
releaseables.push(consoleScript)
|
||||
await consoleScript.run(context)
|
||||
|
||||
for (const bundleName of task.bundles) {
|
||||
const { source, fileName } = getBundleSource(bundleName)
|
||||
const bundleScript = await isolate.compileScript(source, { filename: `sandbox/${fileName}` })
|
||||
releaseables.push(bundleScript)
|
||||
await bundleScript.run(context, { timeout: timeoutMs })
|
||||
}
|
||||
|
||||
const brokerNamesJson = JSON.stringify(task.brokers)
|
||||
const brokerInstallScript = `
|
||||
(() => {
|
||||
// Capture the bridge reference in a closure so hardening can unset the
|
||||
// global without breaking already-installed brokers.
|
||||
const __ref = globalThis.__brokerRef;
|
||||
globalThis.__brokers = globalThis.__brokers || {};
|
||||
for (const name of ${brokerNamesJson}) {
|
||||
globalThis.__brokers[name] = async (args) => {
|
||||
const argsJson = args === undefined ? undefined : JSON.stringify(args);
|
||||
const responseJson = await __ref.apply(
|
||||
undefined,
|
||||
[name, argsJson],
|
||||
{ result: { promise: true } }
|
||||
);
|
||||
let response;
|
||||
try { response = JSON.parse(responseJson); } catch { throw new Error('Invalid broker response'); }
|
||||
if (response.error) throw new Error(response.error);
|
||||
return response.resultJson === undefined || response.resultJson === null
|
||||
? null
|
||||
: JSON.parse(response.resultJson);
|
||||
};
|
||||
}
|
||||
})();
|
||||
`
|
||||
const brokerScript = await isolate.compileScript(brokerInstallScript)
|
||||
releaseables.push(brokerScript)
|
||||
await brokerScript.run(context)
|
||||
|
||||
const bootstrapScript = await isolate.compileScript(
|
||||
`(async () => { ${task.bootstrap} })()`,
|
||||
{ filename: `sandbox/${task.id}/bootstrap.js` }
|
||||
)
|
||||
releaseables.push(bootstrapScript)
|
||||
await bootstrapScript.run(context, { timeout: timeoutMs, promise: true })
|
||||
|
||||
const hardenScript = await isolate.compileScript(`
|
||||
// Remove host-provided bridges + isolated-vm escape globals before user
|
||||
// code runs. Leave the library polyfills (Buffer, process, etc.) alone —
|
||||
// bundles have already captured what they need and user code calling into
|
||||
// them would break if we stripped these.
|
||||
const undefined_globals = [
|
||||
'Isolate', 'Context', 'Script', 'Module', 'Callback', 'Reference',
|
||||
'ExternalCopy', '__dirname', '__filename', '__brokerRef',
|
||||
'__log', '__error'
|
||||
];
|
||||
for (const name of undefined_globals) {
|
||||
try {
|
||||
Object.defineProperty(globalThis, name, {
|
||||
value: undefined, writable: false, configurable: false
|
||||
});
|
||||
} catch {}
|
||||
}
|
||||
`)
|
||||
releaseables.push(hardenScript)
|
||||
await hardenScript.run(context)
|
||||
|
||||
const wrappedUserCode = `
|
||||
(async () => {
|
||||
try {
|
||||
await (async () => {
|
||||
${code}
|
||||
})();
|
||||
return JSON.stringify({ success: true });
|
||||
} catch (error) {
|
||||
return JSON.stringify({
|
||||
success: false,
|
||||
errorInfo: {
|
||||
message: error && error.message ? error.message : String(error),
|
||||
name: error && error.name ? error.name : 'Error',
|
||||
stack: error && error.stack ? error.stack : '',
|
||||
},
|
||||
});
|
||||
}
|
||||
})()
|
||||
`
|
||||
const userScript = await isolate.compileScript(wrappedUserCode, {
|
||||
filename: 'user-function.js',
|
||||
})
|
||||
releaseables.push(userScript)
|
||||
const userResultJson = await userScript.run(context, { timeout: timeoutMs, promise: true })
|
||||
|
||||
let userResult
|
||||
try {
|
||||
userResult = JSON.parse(userResultJson)
|
||||
} catch {
|
||||
userResult = { success: false, errorInfo: { message: 'Invalid user result', name: 'Error' } }
|
||||
}
|
||||
|
||||
if (!userResult.success) {
|
||||
return {
|
||||
result: null,
|
||||
stdout: stdoutChunks.join(''),
|
||||
error: convertToCompatibleError(userResult.errorInfo, code),
|
||||
}
|
||||
}
|
||||
|
||||
const finalizeWrapped = `
|
||||
(async () => {
|
||||
const __bytes = await (async () => {
|
||||
${task.finalize}
|
||||
})();
|
||||
if (!__bytes) {
|
||||
throw new Error('Task finalize returned nothing; expected a Uint8Array');
|
||||
}
|
||||
const __u8 = __bytes instanceof Uint8Array
|
||||
? __bytes
|
||||
: ArrayBuffer.isView(__bytes)
|
||||
? new Uint8Array(__bytes.buffer, __bytes.byteOffset, __bytes.byteLength)
|
||||
: new Uint8Array(__bytes);
|
||||
// Inline base64 encoding (no Buffer dep; works even if polyfill stripped).
|
||||
const __alphabet = 'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/';
|
||||
let __out = '';
|
||||
const __len = __u8.length;
|
||||
for (let __i = 0; __i < __len; __i += 3) {
|
||||
const __b0 = __u8[__i];
|
||||
const __b1 = __i + 1 < __len ? __u8[__i + 1] : 0;
|
||||
const __b2 = __i + 2 < __len ? __u8[__i + 2] : 0;
|
||||
__out += __alphabet[__b0 >> 2];
|
||||
__out += __alphabet[((__b0 & 0x03) << 4) | (__b1 >> 4)];
|
||||
__out += __i + 1 < __len ? __alphabet[((__b1 & 0x0f) << 2) | (__b2 >> 6)] : '=';
|
||||
__out += __i + 2 < __len ? __alphabet[__b2 & 0x3f] : '=';
|
||||
}
|
||||
return __out;
|
||||
})()
|
||||
`
|
||||
const finalizeScript = await isolate.compileScript(finalizeWrapped, {
|
||||
filename: `sandbox/${task.id}/finalize.js`,
|
||||
})
|
||||
releaseables.push(finalizeScript)
|
||||
const bytesBase64 = await finalizeScript.run(context, { timeout: timeoutMs, promise: true })
|
||||
|
||||
return {
|
||||
result: null,
|
||||
stdout: stdoutChunks.join(''),
|
||||
bytesBase64,
|
||||
}
|
||||
} catch (err) {
|
||||
const stdout = stdoutChunks.join('')
|
||||
if (err instanceof Error) {
|
||||
const errorInfo = { message: err.message, name: err.name, stack: err.stack }
|
||||
// Cancellation: host sent `cancel` IPC which called `isolate.dispose()`.
|
||||
// Detect authoritatively via the isolate flag so we don't depend on
|
||||
// isolated-vm's internal error wording.
|
||||
if (isolate && isolate.isDisposed) {
|
||||
return {
|
||||
result: null,
|
||||
stdout,
|
||||
error: { message: 'Execution cancelled', name: 'AbortError' },
|
||||
}
|
||||
}
|
||||
if (err.message && err.message.includes('Script execution timed out')) {
|
||||
return {
|
||||
result: null,
|
||||
stdout,
|
||||
error: {
|
||||
message: `Execution timed out after ${timeoutMs}ms`,
|
||||
name: 'TimeoutError',
|
||||
},
|
||||
}
|
||||
}
|
||||
return {
|
||||
result: null,
|
||||
stdout,
|
||||
error: convertToCompatibleError(errorInfo, code),
|
||||
}
|
||||
}
|
||||
return {
|
||||
result: null,
|
||||
stdout,
|
||||
error: { message: String(err), name: 'Error' },
|
||||
}
|
||||
} finally {
|
||||
for (const obj of releaseables) {
|
||||
if (obj) {
|
||||
try {
|
||||
obj.release()
|
||||
} catch {}
|
||||
}
|
||||
}
|
||||
if (context) {
|
||||
try {
|
||||
context.release()
|
||||
} catch {}
|
||||
}
|
||||
if (isolate) {
|
||||
try {
|
||||
isolate.dispose()
|
||||
} catch {}
|
||||
}
|
||||
if (executionId !== undefined) activeIsolates.delete(executionId)
|
||||
}
|
||||
}
|
||||
|
||||
process.on('message', async (msg) => {
|
||||
try {
|
||||
if (msg.type === 'execute') {
|
||||
const result = await executeCode(msg.request)
|
||||
const result = msg.request.task
|
||||
? await executeTask(msg.request, msg.executionId)
|
||||
: await executeCode(msg.request, msg.executionId)
|
||||
if (process.send && process.connected) {
|
||||
process.send({ type: 'result', executionId: msg.executionId, result })
|
||||
}
|
||||
} else if (msg.type === 'cancel') {
|
||||
// Host asked us to abort this execution. Disposing the isolate causes
|
||||
// the in-flight compileScript/run to throw; the surrounding try/catch
|
||||
// in execute{Code,Task} detects `isolate.isDisposed` and converts that
|
||||
// into an AbortError result, which the host still processes for cleanup.
|
||||
const iso = activeIsolates.get(msg.executionId)
|
||||
if (iso) {
|
||||
try {
|
||||
iso.dispose()
|
||||
} catch {}
|
||||
}
|
||||
// Release any pending broker-call bookkeeping tied to this execution
|
||||
// so its timers + Map entries don't linger up to BROKER_TIMEOUT_MS.
|
||||
for (const [brokerId, pending] of pendingBrokerCalls) {
|
||||
if (pending.executionId === msg.executionId) {
|
||||
clearTimeout(pending.timeout)
|
||||
pendingBrokerCalls.delete(brokerId)
|
||||
pending.resolve(JSON.stringify({ error: 'Execution cancelled' }))
|
||||
}
|
||||
}
|
||||
} else if (msg.type === 'fetchResponse') {
|
||||
const pending = pendingFetches.get(msg.fetchId)
|
||||
if (pending) {
|
||||
@@ -410,6 +777,13 @@ process.on('message', async (msg) => {
|
||||
pendingFetches.delete(msg.fetchId)
|
||||
pending.resolve(msg.response)
|
||||
}
|
||||
} else if (msg.type === 'brokerResponse') {
|
||||
const pending = pendingBrokerCalls.get(msg.brokerId)
|
||||
if (pending) {
|
||||
clearTimeout(pending.timeout)
|
||||
pendingBrokerCalls.delete(msg.brokerId)
|
||||
pending.resolve(JSON.stringify({ error: msg.error, resultJson: msg.resultJson }))
|
||||
}
|
||||
}
|
||||
} catch (err) {
|
||||
if (msg.type === 'execute' && process.send && process.connected) {
|
||||
|
||||
@@ -35,12 +35,42 @@ export interface IsolatedVMExecutionRequest {
|
||||
requestId: string
|
||||
ownerKey?: string
|
||||
ownerWeight?: number
|
||||
/**
|
||||
* Task-mode execution. When set, the worker loads pre-built library bundles,
|
||||
* runs the task `bootstrap`, executes user `code`, then evaluates `finalize`
|
||||
* (must return a `Uint8Array`). The bytes are returned in
|
||||
* `IsolatedVMExecutionResult.bytesBase64`.
|
||||
*/
|
||||
task?: IsolatedVMTaskRequest
|
||||
}
|
||||
|
||||
export interface IsolatedVMTaskRequest {
|
||||
id: string
|
||||
bundles: string[]
|
||||
bootstrap: string
|
||||
brokers: string[]
|
||||
finalize: string
|
||||
}
|
||||
|
||||
/**
|
||||
* Host-side broker handler invoked when isolate code calls a broker.
|
||||
* Registered per-request via `executeInIsolatedVM(..., { brokers })`.
|
||||
*/
|
||||
export type IsolatedVMBrokerHandler = (args: unknown) => Promise<unknown>
|
||||
|
||||
export interface IsolatedVMExecutionOptions {
|
||||
/** Broker name → handler. Must cover every broker listed in `request.task.brokers`. */
|
||||
brokers?: Record<string, IsolatedVMBrokerHandler>
|
||||
/** Cancel the execution early. Broadcasts a cancellation error to the caller. */
|
||||
signal?: AbortSignal
|
||||
}
|
||||
|
||||
export interface IsolatedVMExecutionResult {
|
||||
result: unknown
|
||||
stdout: string
|
||||
error?: IsolatedVMError
|
||||
/** Populated in task mode: the `finalize` result as base64-encoded bytes. */
|
||||
bytesBase64?: string
|
||||
}
|
||||
|
||||
export interface IsolatedVMError {
|
||||
@@ -71,6 +101,10 @@ const DISTRIBUTED_MAX_INFLIGHT_PER_OWNER =
|
||||
MAX_ACTIVE_PER_OWNER + MAX_QUEUED_PER_OWNER
|
||||
const DISTRIBUTED_LEASE_MIN_TTL_MS = Number.parseInt(env.IVM_DISTRIBUTED_LEASE_MIN_TTL_MS) || 120000
|
||||
const MAX_EXECUTIONS_PER_WORKER = Number.parseInt(env.IVM_MAX_EXECUTIONS_PER_WORKER) || 500
|
||||
const MAX_BROKER_ARGS_JSON_CHARS = Number.parseInt(env.IVM_MAX_BROKER_ARGS_JSON_CHARS) || 262_144
|
||||
const MAX_BROKER_RESULT_JSON_CHARS =
|
||||
Number.parseInt(env.IVM_MAX_BROKER_RESULT_JSON_CHARS) || 16_777_216
|
||||
const MAX_BROKERS_PER_EXECUTION = Number.parseInt(env.IVM_MAX_BROKERS_PER_EXECUTION) || 1000
|
||||
const DISTRIBUTED_KEY_PREFIX = 'ivm:fair:v1:owner'
|
||||
const LEASE_REDIS_DEADLINE_MS = 200
|
||||
const QUEUE_RETRY_DELAY_MS = 1000
|
||||
@@ -80,6 +114,11 @@ interface PendingExecution {
|
||||
resolve: (result: IsolatedVMExecutionResult) => void
|
||||
timeout: ReturnType<typeof setTimeout>
|
||||
ownerKey: string
|
||||
brokers?: Record<string, IsolatedVMBrokerHandler>
|
||||
/** Set when the caller aborts. Broker dispatches and the final result stop resolving the promise. */
|
||||
cancelled: boolean
|
||||
/** Number of broker calls made so far for this execution. */
|
||||
brokerCallCount: number
|
||||
}
|
||||
|
||||
interface WorkerInfo {
|
||||
@@ -100,6 +139,21 @@ interface QueuedExecution {
|
||||
req: IsolatedVMExecutionRequest
|
||||
resolve: (result: IsolatedVMExecutionResult) => void
|
||||
queueTimeout: ReturnType<typeof setTimeout>
|
||||
brokers?: Record<string, IsolatedVMBrokerHandler>
|
||||
state: ExecutionState
|
||||
}
|
||||
|
||||
/**
|
||||
* Mutable per-execution bookkeeping shared between the outer Promise, the queue
|
||||
* entry (if queued), and the worker dispatch entry. Lets the AbortSignal listener
|
||||
* locate the right worker/queue slot and mark it cancelled without racing
|
||||
* against the queue-to-worker handoff.
|
||||
*/
|
||||
interface ExecutionState {
|
||||
cancelled: boolean
|
||||
queueId?: number
|
||||
workerId?: number
|
||||
execId?: number
|
||||
}
|
||||
|
||||
interface QueueNode {
|
||||
@@ -523,6 +577,117 @@ function scheduleDrainRetry() {
|
||||
}, QUEUE_RETRY_DELAY_MS)
|
||||
}
|
||||
|
||||
function handleBrokerMessage(
|
||||
workerInfo: WorkerInfo | undefined,
|
||||
msg: Record<string, unknown>
|
||||
): void {
|
||||
if (!workerInfo) return
|
||||
const brokerId = msg.brokerId as number
|
||||
const executionId = msg.executionId as number
|
||||
const brokerName = msg.brokerName as string
|
||||
const argsJson = msg.argsJson as string | undefined
|
||||
|
||||
const sendResponse = (payload: Record<string, unknown>) => {
|
||||
try {
|
||||
workerInfo.process.send({ type: 'brokerResponse', brokerId, ...payload })
|
||||
} catch (err) {
|
||||
logger.error('Failed to send broker response to worker', {
|
||||
err,
|
||||
brokerId,
|
||||
brokerName,
|
||||
workerId: workerInfo.id,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
const logReject = (reason: string, extra?: Record<string, unknown>) => {
|
||||
logger.warn('Sandbox broker call rejected', {
|
||||
reason,
|
||||
brokerName,
|
||||
executionId,
|
||||
workerId: workerInfo.id,
|
||||
...extra,
|
||||
})
|
||||
}
|
||||
|
||||
const pending = workerInfo.pendingExecutions.get(executionId)
|
||||
if (!pending) {
|
||||
sendResponse({ error: 'Execution no longer active' })
|
||||
return
|
||||
}
|
||||
|
||||
if (pending.cancelled) {
|
||||
sendResponse({ error: 'Execution cancelled' })
|
||||
return
|
||||
}
|
||||
|
||||
if (argsJson && argsJson.length > MAX_BROKER_ARGS_JSON_CHARS) {
|
||||
logReject('args_too_large', { argsJsonLength: argsJson.length })
|
||||
sendResponse({
|
||||
error: `Broker args exceed maximum size (${MAX_BROKER_ARGS_JSON_CHARS} chars)`,
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
pending.brokerCallCount++
|
||||
if (pending.brokerCallCount > MAX_BROKERS_PER_EXECUTION) {
|
||||
logReject('rate_limit', { brokerCallCount: pending.brokerCallCount })
|
||||
sendResponse({
|
||||
error: `Broker call limit exceeded (${MAX_BROKERS_PER_EXECUTION} per execution)`,
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
const handler = pending.brokers?.[brokerName]
|
||||
if (!handler) {
|
||||
logReject('unknown_broker')
|
||||
sendResponse({ error: `Broker "${brokerName}" is not available for this execution` })
|
||||
return
|
||||
}
|
||||
|
||||
let args: unknown
|
||||
if (argsJson) {
|
||||
try {
|
||||
args = JSON.parse(argsJson)
|
||||
} catch {
|
||||
logReject('invalid_args_json')
|
||||
sendResponse({ error: 'Invalid broker args JSON' })
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
Promise.resolve()
|
||||
.then(() => handler(args))
|
||||
.then((resultValue) => {
|
||||
if (pending.cancelled) {
|
||||
sendResponse({ error: 'Execution cancelled' })
|
||||
return
|
||||
}
|
||||
let resultJson: string
|
||||
try {
|
||||
resultJson = JSON.stringify(resultValue ?? null)
|
||||
} catch {
|
||||
logReject('result_not_serializable')
|
||||
sendResponse({ error: 'Broker result is not JSON-serializable' })
|
||||
return
|
||||
}
|
||||
if (resultJson.length > MAX_BROKER_RESULT_JSON_CHARS) {
|
||||
logReject('result_too_large', { resultJsonLength: resultJson.length })
|
||||
sendResponse({
|
||||
error: `Broker result exceeds maximum size (${MAX_BROKER_RESULT_JSON_CHARS} chars)`,
|
||||
})
|
||||
return
|
||||
}
|
||||
sendResponse({ resultJson })
|
||||
})
|
||||
.catch((err) => {
|
||||
logReject('handler_threw', {
|
||||
error: err instanceof Error ? err.message : String(err),
|
||||
})
|
||||
sendResponse({ error: err instanceof Error ? err.message : String(err) })
|
||||
})
|
||||
}
|
||||
|
||||
function handleWorkerMessage(workerId: number, message: unknown) {
|
||||
if (typeof message !== 'object' || message === null) return
|
||||
const msg = message as Record<string, unknown>
|
||||
@@ -554,12 +719,22 @@ function handleWorkerMessage(workerId: number, message: unknown) {
|
||||
} else {
|
||||
resetWorkerIdleTimeout(workerId)
|
||||
}
|
||||
pending.resolve(msg.result as IsolatedVMExecutionResult)
|
||||
// If the caller aborted, the outer Promise is already resolved with
|
||||
// AbortError. Still run all the bookkeeping above so pool counters stay
|
||||
// accurate; just skip re-resolving.
|
||||
if (!pending.cancelled) {
|
||||
pending.resolve(msg.result as IsolatedVMExecutionResult)
|
||||
}
|
||||
drainQueue()
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
if (msg.type === 'broker') {
|
||||
handleBrokerMessage(workerInfo, msg)
|
||||
return
|
||||
}
|
||||
|
||||
if (msg.type === 'fetch') {
|
||||
const { fetchId, requestId, url, optionsJson } = msg as {
|
||||
fetchId: number
|
||||
@@ -855,9 +1030,25 @@ function dispatchToWorker(
|
||||
workerInfo: WorkerInfo,
|
||||
ownerState: OwnerState,
|
||||
req: IsolatedVMExecutionRequest,
|
||||
resolve: (result: IsolatedVMExecutionResult) => void
|
||||
resolve: (result: IsolatedVMExecutionResult) => void,
|
||||
state: ExecutionState,
|
||||
brokers?: Record<string, IsolatedVMBrokerHandler>
|
||||
) {
|
||||
// Caller may have aborted between acquireWorker() and dispatch. Skip the
|
||||
// round-trip entirely and let the abort listener handle settlement.
|
||||
if (state.cancelled) {
|
||||
resolve({
|
||||
result: null,
|
||||
stdout: '',
|
||||
error: { message: 'Execution cancelled', name: 'AbortError' },
|
||||
})
|
||||
drainQueue()
|
||||
return
|
||||
}
|
||||
|
||||
const execId = ++executionIdCounter
|
||||
state.workerId = workerInfo.id
|
||||
state.execId = execId
|
||||
|
||||
if (workerInfo.idleTimeout) {
|
||||
clearTimeout(workerInfo.idleTimeout)
|
||||
@@ -891,7 +1082,14 @@ function dispatchToWorker(
|
||||
drainQueue()
|
||||
}, req.timeoutMs + 1000)
|
||||
|
||||
workerInfo.pendingExecutions.set(execId, { resolve, timeout, ownerKey: ownerState.ownerKey })
|
||||
workerInfo.pendingExecutions.set(execId, {
|
||||
resolve,
|
||||
timeout,
|
||||
ownerKey: ownerState.ownerKey,
|
||||
brokers,
|
||||
cancelled: false,
|
||||
brokerCallCount: 0,
|
||||
})
|
||||
workerInfo.activeExecutions++
|
||||
totalActiveExecutions++
|
||||
ownerState.activeExecutions++
|
||||
@@ -923,7 +1121,9 @@ function dispatchToWorker(
|
||||
function enqueueExecution(
|
||||
ownerState: OwnerState,
|
||||
req: IsolatedVMExecutionRequest,
|
||||
resolve: (result: IsolatedVMExecutionResult) => void
|
||||
resolve: (result: IsolatedVMExecutionResult) => void,
|
||||
state: ExecutionState,
|
||||
brokers?: Record<string, IsolatedVMBrokerHandler>
|
||||
) {
|
||||
if (queueLength() >= MAX_QUEUE_SIZE) {
|
||||
resolve({
|
||||
@@ -963,12 +1163,15 @@ function enqueueExecution(
|
||||
})
|
||||
}, QUEUE_TIMEOUT_MS)
|
||||
|
||||
state.queueId = queueId
|
||||
pushQueuedExecution(ownerState, {
|
||||
id: queueId,
|
||||
ownerKey: ownerState.ownerKey,
|
||||
req,
|
||||
resolve,
|
||||
queueTimeout,
|
||||
brokers,
|
||||
state,
|
||||
})
|
||||
logger.info('Execution queued', {
|
||||
queueLength: queueLength(),
|
||||
@@ -1014,7 +1217,9 @@ function drainQueue() {
|
||||
continue
|
||||
}
|
||||
clearTimeout(queued.queueTimeout)
|
||||
dispatchToWorker(worker, owner, queued.req, queued.resolve)
|
||||
// Clearing queueId: from here on, abort must reach the worker, not the queue.
|
||||
queued.state.queueId = undefined
|
||||
dispatchToWorker(worker, owner, queued.req, queued.resolve, queued.state, queued.brokers)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1022,11 +1227,39 @@ function drainQueue() {
|
||||
* Execute JavaScript code in an isolated V8 isolate via Node.js subprocess.
|
||||
*/
|
||||
export async function executeInIsolatedVM(
|
||||
req: IsolatedVMExecutionRequest
|
||||
req: IsolatedVMExecutionRequest,
|
||||
options?: IsolatedVMExecutionOptions
|
||||
): Promise<IsolatedVMExecutionResult> {
|
||||
const ownerKey = normalizeOwnerKey(req.ownerKey)
|
||||
const ownerWeight = normalizeOwnerWeight(req.ownerWeight)
|
||||
const ownerState = getOrCreateOwnerState(ownerKey, ownerWeight)
|
||||
const brokers = options?.brokers
|
||||
const signal = options?.signal
|
||||
|
||||
if (signal?.aborted) {
|
||||
maybeCleanupOwner(ownerKey)
|
||||
return {
|
||||
result: null,
|
||||
stdout: '',
|
||||
error: { message: 'Execution cancelled', name: 'AbortError' },
|
||||
}
|
||||
}
|
||||
|
||||
if (req.task) {
|
||||
for (const brokerName of req.task.brokers) {
|
||||
if (!brokers?.[brokerName]) {
|
||||
maybeCleanupOwner(ownerKey)
|
||||
return {
|
||||
result: null,
|
||||
stdout: '',
|
||||
error: {
|
||||
message: `Task "${req.task.id}" requires broker "${brokerName}" but none was provided`,
|
||||
name: 'Error',
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const distributedLeaseId = `${req.requestId}:${Date.now()}:${Math.random().toString(36).slice(2, 10)}`
|
||||
const leaseAcquireResult = await tryAcquireDistributedLease(
|
||||
@@ -1060,35 +1293,77 @@ export async function executeInIsolatedVM(
|
||||
})
|
||||
}
|
||||
|
||||
const state: ExecutionState = { cancelled: false }
|
||||
|
||||
return new Promise<IsolatedVMExecutionResult>((resolve) => {
|
||||
let abortListener: (() => void) | null = null
|
||||
let resolved = false
|
||||
const resolveWithRelease = (result: IsolatedVMExecutionResult) => {
|
||||
if (resolved) return
|
||||
resolved = true
|
||||
if (abortListener && signal) {
|
||||
signal.removeEventListener('abort', abortListener)
|
||||
}
|
||||
releaseLease()
|
||||
resolve(result)
|
||||
}
|
||||
|
||||
if (signal) {
|
||||
abortListener = () => {
|
||||
state.cancelled = true
|
||||
// If queued, drop the entry immediately and free the slot.
|
||||
if (state.queueId !== undefined) {
|
||||
const removed = removeQueuedExecutionById(state.queueId)
|
||||
if (removed) clearTimeout(removed.queueTimeout)
|
||||
state.queueId = undefined
|
||||
}
|
||||
// If dispatched, mark the pending entry cancelled and ask the worker to
|
||||
// dispose its isolate so the pool slot can be released. The worker will
|
||||
// emit a `result` shortly after, which runs the normal counter cleanup.
|
||||
if (state.workerId !== undefined && state.execId !== undefined) {
|
||||
const wi = workers.get(state.workerId)
|
||||
const pending = wi?.pendingExecutions.get(state.execId)
|
||||
if (pending) pending.cancelled = true
|
||||
if (wi) {
|
||||
try {
|
||||
wi.process.send({ type: 'cancel', executionId: state.execId })
|
||||
} catch (err) {
|
||||
logger.warn('Failed to send cancel to worker', { err, workerId: state.workerId })
|
||||
}
|
||||
}
|
||||
}
|
||||
resolveWithRelease({
|
||||
result: null,
|
||||
stdout: '',
|
||||
error: { message: 'Execution cancelled', name: 'AbortError' },
|
||||
})
|
||||
}
|
||||
signal.addEventListener('abort', abortListener, { once: true })
|
||||
}
|
||||
|
||||
if (
|
||||
totalActiveExecutions >= MAX_CONCURRENT ||
|
||||
ownerState.activeExecutions >= MAX_ACTIVE_PER_OWNER
|
||||
) {
|
||||
enqueueExecution(ownerState, req, resolveWithRelease)
|
||||
enqueueExecution(ownerState, req, resolveWithRelease, state, brokers)
|
||||
return
|
||||
}
|
||||
|
||||
acquireWorker()
|
||||
.then((workerInfo) => {
|
||||
if (!workerInfo) {
|
||||
enqueueExecution(ownerState, req, resolveWithRelease)
|
||||
enqueueExecution(ownerState, req, resolveWithRelease, state, brokers)
|
||||
return
|
||||
}
|
||||
|
||||
dispatchToWorker(workerInfo, ownerState, req, resolveWithRelease)
|
||||
dispatchToWorker(workerInfo, ownerState, req, resolveWithRelease, state, brokers)
|
||||
if (queueLength() > 0) {
|
||||
drainQueue()
|
||||
}
|
||||
})
|
||||
.catch((error) => {
|
||||
logger.error('Failed to acquire worker for execution', { error, ownerKey })
|
||||
enqueueExecution(ownerState, req, resolveWithRelease)
|
||||
enqueueExecution(ownerState, req, resolveWithRelease, state, brokers)
|
||||
})
|
||||
}).finally(() => {
|
||||
releaseLease()
|
||||
|
||||
@@ -1,198 +0,0 @@
|
||||
/**
|
||||
* Sandboxed PPTX generation via subprocess.
|
||||
*
|
||||
* User code runs in a separate Node.js child process. File access is brokered
|
||||
* via IPC — the subprocess never touches the database directly.
|
||||
*/
|
||||
|
||||
import { type ChildProcess, spawn } from 'node:child_process'
|
||||
import fs from 'node:fs'
|
||||
import path from 'node:path'
|
||||
import { fileURLToPath } from 'node:url'
|
||||
import { createLogger } from '@sim/logger'
|
||||
import {
|
||||
downloadWorkspaceFile,
|
||||
getWorkspaceFile,
|
||||
} from '@/lib/uploads/contexts/workspace/workspace-file-manager'
|
||||
|
||||
const logger = createLogger('PptxVMExecution')
|
||||
|
||||
const WORKER_STARTUP_TIMEOUT_MS = 10_000
|
||||
const GENERATION_TIMEOUT_MS = 60_000
|
||||
const MAX_STDERR = 4096
|
||||
|
||||
type WorkerMessage =
|
||||
| { type: 'ready' }
|
||||
| { type: 'result'; data: string }
|
||||
| { type: 'error'; message: string }
|
||||
| { type: 'getFile'; fileReqId: number; fileId: string }
|
||||
|
||||
const currentDir = path.dirname(fileURLToPath(import.meta.url))
|
||||
let cachedWorkerPath: string | undefined
|
||||
|
||||
function getWorkerPath(): string {
|
||||
if (cachedWorkerPath) return cachedWorkerPath
|
||||
const candidates = [
|
||||
path.join(currentDir, '..', '..', 'dist', 'pptx-worker.cjs'),
|
||||
path.join(currentDir, 'pptx-worker.cjs'),
|
||||
path.join(process.cwd(), 'apps', 'sim', 'dist', 'pptx-worker.cjs'),
|
||||
path.join(process.cwd(), 'apps', 'sim', 'lib', 'execution', 'pptx-worker.cjs'),
|
||||
path.join(process.cwd(), 'dist', 'pptx-worker.cjs'),
|
||||
path.join(process.cwd(), 'lib', 'execution', 'pptx-worker.cjs'),
|
||||
]
|
||||
const found = candidates.find((p) => fs.existsSync(p))
|
||||
if (!found) throw new Error(`pptx-worker.cjs not found at any of: ${candidates.join(', ')}`)
|
||||
cachedWorkerPath = found
|
||||
return found
|
||||
}
|
||||
|
||||
/**
|
||||
* Generate a PPTX file by executing AI-generated PptxGenJS code in a sandboxed
|
||||
* subprocess. File resources referenced by the code are fetched from workspace
|
||||
* storage by the main process and delivered to the worker via IPC.
|
||||
*/
|
||||
export async function generatePptxFromCode(
|
||||
code: string,
|
||||
workspaceId: string,
|
||||
signal?: AbortSignal
|
||||
): Promise<Buffer> {
|
||||
return new Promise<Buffer>((resolve, reject) => {
|
||||
let proc: ChildProcess | null = null
|
||||
let settled = false
|
||||
let startupTimer: ReturnType<typeof setTimeout> | null = null
|
||||
let generationTimer: ReturnType<typeof setTimeout> | null = null
|
||||
|
||||
function done(err: Error): void
|
||||
function done(err: undefined, result: Buffer): void
|
||||
function done(err: Error | undefined, result?: Buffer): void {
|
||||
if (settled) return
|
||||
settled = true
|
||||
if (startupTimer) clearTimeout(startupTimer)
|
||||
if (generationTimer) clearTimeout(generationTimer)
|
||||
try {
|
||||
proc?.removeAllListeners()
|
||||
proc?.kill()
|
||||
} catch {
|
||||
// Ignore — process may have already exited
|
||||
}
|
||||
if (err) reject(err)
|
||||
else resolve(result as Buffer)
|
||||
}
|
||||
|
||||
if (signal?.aborted) {
|
||||
reject(new Error('PPTX generation cancelled'))
|
||||
return
|
||||
}
|
||||
|
||||
signal?.addEventListener('abort', () => done(new Error('PPTX generation cancelled')), {
|
||||
once: true,
|
||||
})
|
||||
|
||||
try {
|
||||
proc = spawn('node', [getWorkerPath()], {
|
||||
stdio: ['ignore', 'pipe', 'pipe', 'ipc'],
|
||||
serialization: 'json',
|
||||
env: { PATH: process.env.PATH ?? '' } as unknown as NodeJS.ProcessEnv,
|
||||
})
|
||||
} catch (err) {
|
||||
done(err instanceof Error ? err : new Error(String(err)))
|
||||
return
|
||||
}
|
||||
|
||||
let stderrData = ''
|
||||
proc.stderr?.on('data', (chunk: Buffer) => {
|
||||
if (stderrData.length < MAX_STDERR) {
|
||||
stderrData += chunk.toString()
|
||||
if (stderrData.length > MAX_STDERR) stderrData = stderrData.slice(0, MAX_STDERR)
|
||||
}
|
||||
})
|
||||
|
||||
startupTimer = setTimeout(() => {
|
||||
logger.error('PPTX worker failed to start within timeout')
|
||||
done(new Error('PPTX worker failed to start'))
|
||||
}, WORKER_STARTUP_TIMEOUT_MS)
|
||||
|
||||
proc.on('exit', (code) => {
|
||||
if (!settled) {
|
||||
logger.error('PPTX worker exited unexpectedly', { code, stderr: stderrData.slice(0, 500) })
|
||||
done(new Error(`PPTX worker exited unexpectedly (code ${code})`))
|
||||
}
|
||||
})
|
||||
|
||||
proc.on('error', (err) => {
|
||||
logger.error('PPTX worker process error', { error: err.message })
|
||||
done(err)
|
||||
})
|
||||
|
||||
proc.on('message', (rawMsg: unknown) => {
|
||||
const msg = rawMsg as WorkerMessage
|
||||
|
||||
if (msg.type === 'ready') {
|
||||
if (startupTimer) {
|
||||
clearTimeout(startupTimer)
|
||||
startupTimer = null
|
||||
}
|
||||
generationTimer = setTimeout(() => {
|
||||
logger.error('PPTX generation timed out')
|
||||
done(new Error('PPTX generation timed out'))
|
||||
}, GENERATION_TIMEOUT_MS)
|
||||
proc!.send({ type: 'generate', code })
|
||||
return
|
||||
}
|
||||
|
||||
if (msg.type === 'result') {
|
||||
done(undefined, Buffer.from(msg.data, 'base64'))
|
||||
return
|
||||
}
|
||||
|
||||
if (msg.type === 'error') {
|
||||
done(new Error(msg.message))
|
||||
return
|
||||
}
|
||||
|
||||
if (msg.type === 'getFile') {
|
||||
handleFileRequest(proc!, workspaceId, msg).catch((err) => {
|
||||
logger.error('Failed to handle file request from PPTX worker', {
|
||||
fileId: msg.fileId,
|
||||
error: err instanceof Error ? err.message : String(err),
|
||||
})
|
||||
if (proc && !settled) {
|
||||
try {
|
||||
proc.send({
|
||||
type: 'fileResult',
|
||||
fileReqId: msg.fileReqId,
|
||||
error: err instanceof Error ? err.message : 'File fetch failed',
|
||||
})
|
||||
} catch {
|
||||
// Ignore — process may have died
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
async function handleFileRequest(
|
||||
proc: ChildProcess,
|
||||
workspaceId: string,
|
||||
msg: Extract<WorkerMessage, { type: 'getFile' }>
|
||||
): Promise<void> {
|
||||
const record = await getWorkspaceFile(workspaceId, msg.fileId)
|
||||
if (!record) {
|
||||
proc.send({
|
||||
type: 'fileResult',
|
||||
fileReqId: msg.fileReqId,
|
||||
error: `File not found: ${msg.fileId}`,
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
const buffer = await downloadWorkspaceFile(record)
|
||||
const mime = record.type || 'image/png'
|
||||
proc.send({
|
||||
type: 'fileResult',
|
||||
fileReqId: msg.fileReqId,
|
||||
data: `data:${mime};base64,${buffer.toString('base64')}`,
|
||||
})
|
||||
}
|
||||
@@ -1,99 +0,0 @@
|
||||
/**
|
||||
* Node.js worker for sandboxed PPTX generation.
|
||||
* Runs in a separate Node.js process, communicates with parent via IPC.
|
||||
*/
|
||||
|
||||
'use strict'
|
||||
|
||||
const vm = require('node:vm')
|
||||
const PptxGenJS = require('pptxgenjs')
|
||||
|
||||
const EXECUTION_TIMEOUT_MS = 30_000
|
||||
const FILE_REQUEST_TIMEOUT_MS = 30_000
|
||||
|
||||
const pendingFileRequests = new Map()
|
||||
let fileRequestCounter = 0
|
||||
|
||||
function sendToParent(msg) {
|
||||
if (process.send && process.connected) {
|
||||
process.send(msg)
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
process.on('message', async (msg) => {
|
||||
if (msg.type === 'generate') {
|
||||
await handleGenerate(msg)
|
||||
} else if (msg.type === 'fileResult') {
|
||||
handleFileResult(msg)
|
||||
}
|
||||
})
|
||||
|
||||
async function handleGenerate(msg) {
|
||||
const { code } = msg
|
||||
|
||||
try {
|
||||
const pptx = new PptxGenJS()
|
||||
|
||||
const getFileBase64 = (fileId) =>
|
||||
new Promise((resolve, reject) => {
|
||||
if (typeof fileId !== 'string' || fileId.length === 0) {
|
||||
reject(new Error('fileId must be a non-empty string'))
|
||||
return
|
||||
}
|
||||
|
||||
const fileReqId = ++fileRequestCounter
|
||||
const timeout = setTimeout(() => {
|
||||
if (pendingFileRequests.has(fileReqId)) {
|
||||
pendingFileRequests.delete(fileReqId)
|
||||
reject(new Error(`File request timed out for fileId: ${fileId}`))
|
||||
}
|
||||
}, FILE_REQUEST_TIMEOUT_MS)
|
||||
|
||||
pendingFileRequests.set(fileReqId, { resolve, reject, timeout })
|
||||
|
||||
if (!sendToParent({ type: 'getFile', fileReqId, fileId })) {
|
||||
clearTimeout(timeout)
|
||||
pendingFileRequests.delete(fileReqId)
|
||||
reject(new Error('Parent process disconnected'))
|
||||
}
|
||||
})
|
||||
|
||||
const sandbox = Object.create(null)
|
||||
sandbox.pptx = pptx
|
||||
sandbox.getFileBase64 = getFileBase64
|
||||
|
||||
vm.createContext(sandbox)
|
||||
|
||||
const promise = vm.runInContext(`(async () => { ${code} })()`, sandbox, {
|
||||
timeout: EXECUTION_TIMEOUT_MS,
|
||||
filename: 'pptx-code.js',
|
||||
})
|
||||
await promise
|
||||
|
||||
const output = await pptx.write({ outputType: 'nodebuffer' })
|
||||
const base64 = Buffer.from(output).toString('base64')
|
||||
sendToParent({ type: 'result', data: base64 })
|
||||
} catch (err) {
|
||||
const message = err instanceof Error ? err.message : String(err)
|
||||
sendToParent({ type: 'error', message })
|
||||
}
|
||||
}
|
||||
|
||||
function handleFileResult(msg) {
|
||||
const { fileReqId, data, error } = msg
|
||||
const pending = pendingFileRequests.get(fileReqId)
|
||||
if (!pending) return
|
||||
|
||||
clearTimeout(pending.timeout)
|
||||
pendingFileRequests.delete(fileReqId)
|
||||
|
||||
if (error) {
|
||||
pending.reject(new Error(error))
|
||||
} else {
|
||||
pending.resolve(data)
|
||||
}
|
||||
}
|
||||
|
||||
sendToParent({ type: 'ready' })
|
||||
@@ -0,0 +1,47 @@
|
||||
import { createLogger } from '@sim/logger'
|
||||
import type { SandboxBroker } from '@/lib/execution/sandbox/types'
|
||||
import {
|
||||
downloadWorkspaceFile,
|
||||
getWorkspaceFile,
|
||||
} from '@/lib/uploads/contexts/workspace/workspace-file-manager'
|
||||
|
||||
const logger = createLogger('SandboxWorkspaceFileBroker')
|
||||
|
||||
interface WorkspaceFileArgs {
|
||||
fileId: string
|
||||
}
|
||||
|
||||
interface WorkspaceFileResult {
|
||||
dataUri: string
|
||||
}
|
||||
|
||||
/**
|
||||
* Host-side broker that resolves a workspace file id into a base64 data URI.
|
||||
*
|
||||
* Exposed to isolate code through `__brokers.workspaceFile(fileId)` and wrapped
|
||||
* by the task bootstrap as `getFileBase64(fileId)`.
|
||||
*/
|
||||
export const workspaceFileBroker: SandboxBroker<WorkspaceFileArgs, WorkspaceFileResult> = {
|
||||
name: 'workspaceFile',
|
||||
async handle(ctx, args) {
|
||||
if (!args || typeof args.fileId !== 'string' || args.fileId.length === 0) {
|
||||
throw new Error('workspaceFile broker requires a non-empty fileId')
|
||||
}
|
||||
if (!ctx.workspaceId) {
|
||||
throw new Error('workspaceFile broker requires a workspaceId')
|
||||
}
|
||||
|
||||
const record = await getWorkspaceFile(ctx.workspaceId, args.fileId)
|
||||
if (!record) {
|
||||
logger.warn('Workspace file not found for sandbox broker', {
|
||||
workspaceId: ctx.workspaceId,
|
||||
fileId: args.fileId,
|
||||
})
|
||||
throw new Error(`File not found: ${args.fileId}`)
|
||||
}
|
||||
|
||||
const buffer = await downloadWorkspaceFile(record)
|
||||
const mime = record.type || 'image/png'
|
||||
return { dataUri: `data:${mime};base64,${buffer.toString('base64')}` }
|
||||
},
|
||||
}
|
||||
@@ -0,0 +1,127 @@
|
||||
/**
|
||||
* Isolate-side polyfills. Must run BEFORE the library imports (and before
|
||||
* `process/browser` in particular), because `process/browser` captures
|
||||
* `setTimeout` at module-init time.
|
||||
*
|
||||
* Only imported from bundle entries in `build.ts`; not for direct use
|
||||
* elsewhere.
|
||||
*/
|
||||
|
||||
type TimerFn = (fn: () => void) => number
|
||||
|
||||
const g = globalThis as unknown as Record<string, unknown>
|
||||
|
||||
if (typeof g.global === 'undefined') g.global = globalThis
|
||||
if (typeof g.globalThis === 'undefined') (g as Record<string, unknown>).globalThis = globalThis
|
||||
|
||||
const microtask: TimerFn = (fn) => {
|
||||
try {
|
||||
Promise.resolve().then(fn)
|
||||
} catch {
|
||||
fn()
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
if (typeof g.setTimeout === 'undefined') {
|
||||
g.setTimeout = microtask
|
||||
g.clearTimeout = () => {}
|
||||
}
|
||||
if (typeof g.setImmediate === 'undefined') {
|
||||
g.setImmediate = microtask
|
||||
g.clearImmediate = () => {}
|
||||
}
|
||||
if (typeof g.setInterval === 'undefined') {
|
||||
g.setInterval = () => 0
|
||||
g.clearInterval = () => {}
|
||||
}
|
||||
if (typeof g.queueMicrotask === 'undefined') {
|
||||
g.queueMicrotask = (fn: () => void) => {
|
||||
Promise.resolve().then(fn)
|
||||
}
|
||||
}
|
||||
|
||||
if (typeof g.TextEncoder === 'undefined') {
|
||||
g.TextEncoder = class TextEncoder {
|
||||
get encoding() {
|
||||
return 'utf-8'
|
||||
}
|
||||
encode(input?: string): Uint8Array {
|
||||
const str = String(input == null ? '' : input)
|
||||
const bytes: number[] = []
|
||||
for (let i = 0; i < str.length; i++) {
|
||||
let code = str.charCodeAt(i)
|
||||
if (code >= 0xd800 && code <= 0xdbff && i + 1 < str.length) {
|
||||
const next = str.charCodeAt(i + 1)
|
||||
if (next >= 0xdc00 && next <= 0xdfff) {
|
||||
code = 0x10000 + ((code - 0xd800) << 10) + (next - 0xdc00)
|
||||
i++
|
||||
}
|
||||
}
|
||||
if (code < 0x80) {
|
||||
bytes.push(code)
|
||||
} else if (code < 0x800) {
|
||||
bytes.push(0xc0 | (code >> 6), 0x80 | (code & 0x3f))
|
||||
} else if (code < 0x10000) {
|
||||
bytes.push(0xe0 | (code >> 12), 0x80 | ((code >> 6) & 0x3f), 0x80 | (code & 0x3f))
|
||||
} else {
|
||||
bytes.push(
|
||||
0xf0 | (code >> 18),
|
||||
0x80 | ((code >> 12) & 0x3f),
|
||||
0x80 | ((code >> 6) & 0x3f),
|
||||
0x80 | (code & 0x3f)
|
||||
)
|
||||
}
|
||||
}
|
||||
return new Uint8Array(bytes)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (typeof g.TextDecoder === 'undefined') {
|
||||
g.TextDecoder = class TextDecoder {
|
||||
private _label: string
|
||||
constructor(label?: string) {
|
||||
this._label = (label || 'utf-8').toLowerCase()
|
||||
}
|
||||
get encoding() {
|
||||
return this._label
|
||||
}
|
||||
decode(input?: BufferSource): string {
|
||||
if (!input) return ''
|
||||
const bytes =
|
||||
input instanceof Uint8Array
|
||||
? input
|
||||
: ArrayBuffer.isView(input)
|
||||
? new Uint8Array(input.buffer, input.byteOffset, input.byteLength)
|
||||
: new Uint8Array(input as ArrayBuffer)
|
||||
let out = ''
|
||||
let i = 0
|
||||
while (i < bytes.length) {
|
||||
const b1 = bytes[i++]
|
||||
if (b1 < 0x80) {
|
||||
out += String.fromCharCode(b1)
|
||||
} else if (b1 < 0xc0) {
|
||||
out += '\ufffd'
|
||||
} else if (b1 < 0xe0) {
|
||||
const b2 = bytes[i++] & 0x3f
|
||||
out += String.fromCharCode(((b1 & 0x1f) << 6) | b2)
|
||||
} else if (b1 < 0xf0) {
|
||||
const b2 = bytes[i++] & 0x3f
|
||||
const b3 = bytes[i++] & 0x3f
|
||||
out += String.fromCharCode(((b1 & 0x0f) << 12) | (b2 << 6) | b3)
|
||||
} else {
|
||||
const b2 = bytes[i++] & 0x3f
|
||||
const b3 = bytes[i++] & 0x3f
|
||||
const b4 = bytes[i++] & 0x3f
|
||||
let cp = ((b1 & 0x07) << 18) | (b2 << 12) | (b3 << 6) | b4
|
||||
cp -= 0x10000
|
||||
out += String.fromCharCode(0xd800 + (cp >> 10), 0xdc00 + (cp & 0x3ff))
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export {}
|
||||
@@ -0,0 +1,131 @@
|
||||
#!/usr/bin/env bun
|
||||
/**
|
||||
* Builds isolate-compatible bundles for the document-generation libraries.
|
||||
*
|
||||
* Each library is bundled with `target=browser, format=iife` so it can be
|
||||
* evaluated inside a V8 isolate that has no Node APIs (`require`, `process`,
|
||||
* `fs`). The emitted files attach their exports to `globalThis.__bundles[name]`
|
||||
* and are checked in so production images don't need the bundler at runtime.
|
||||
*
|
||||
* Run via: `bun run build:sandbox-bundles`.
|
||||
*/
|
||||
|
||||
import { mkdirSync, rmSync, writeFileSync } from 'node:fs'
|
||||
import { dirname, join } from 'node:path'
|
||||
import { fileURLToPath } from 'node:url'
|
||||
|
||||
interface BunBuildResult {
|
||||
success: boolean
|
||||
logs: unknown[]
|
||||
outputs: Array<{ text: () => Promise<string> }>
|
||||
}
|
||||
interface BunBuildOptions {
|
||||
entrypoints: string[]
|
||||
target: string
|
||||
format: string
|
||||
minify: boolean
|
||||
sourcemap: string
|
||||
root: string
|
||||
}
|
||||
declare const Bun: { build: (opts: BunBuildOptions) => Promise<BunBuildResult> }
|
||||
|
||||
const HERE = dirname(fileURLToPath(import.meta.url))
|
||||
const BUNDLES_DIR = HERE
|
||||
const ENTRIES_DIR = join(HERE, '.entries')
|
||||
const APP_SIM_ROOT = join(HERE, '..', '..', '..', '..')
|
||||
|
||||
interface BundleSpec {
|
||||
/** Key on `globalThis.__bundles`. */
|
||||
name: string
|
||||
/** Short filename written under `bundles/<file>.cjs`. */
|
||||
outFile: string
|
||||
/** Source of the entry file bun will bundle. */
|
||||
entry: string
|
||||
}
|
||||
|
||||
const POLYFILLS_PATH = join(HERE, '_polyfills.ts')
|
||||
const POLYFILL_PRELUDE = `
|
||||
// Isolate-side polyfills must execute BEFORE any other import (process/browser
|
||||
// captures setTimeout at module-init time). Keep this as the first import.
|
||||
import '${POLYFILLS_PATH}'
|
||||
import { Buffer as __BufferPolyfill } from 'buffer'
|
||||
import * as __processPolyfill from 'process/browser'
|
||||
if (typeof globalThis.Buffer === 'undefined') globalThis.Buffer = __BufferPolyfill
|
||||
if (typeof globalThis.process === 'undefined') globalThis.process = __processPolyfill
|
||||
`
|
||||
|
||||
const BUNDLES: ReadonlyArray<BundleSpec> = [
|
||||
{
|
||||
name: 'pdf-lib',
|
||||
outFile: 'pdf-lib.cjs',
|
||||
entry: `
|
||||
${POLYFILL_PRELUDE}
|
||||
import * as mod from 'pdf-lib'
|
||||
globalThis.__bundles = globalThis.__bundles || {}
|
||||
globalThis.__bundles['pdf-lib'] = mod
|
||||
`,
|
||||
},
|
||||
{
|
||||
name: 'docx',
|
||||
outFile: 'docx.cjs',
|
||||
entry: `
|
||||
${POLYFILL_PRELUDE}
|
||||
import * as mod from 'docx'
|
||||
globalThis.__bundles = globalThis.__bundles || {}
|
||||
globalThis.__bundles['docx'] = mod
|
||||
`,
|
||||
},
|
||||
{
|
||||
name: 'pptxgenjs',
|
||||
outFile: 'pptxgenjs.cjs',
|
||||
entry: `
|
||||
${POLYFILL_PRELUDE}
|
||||
import PptxGenJS from 'pptxgenjs'
|
||||
globalThis.__bundles = globalThis.__bundles || {}
|
||||
globalThis.__bundles['pptxgenjs'] = PptxGenJS
|
||||
`,
|
||||
},
|
||||
]
|
||||
|
||||
async function main(): Promise<void> {
|
||||
rmSync(ENTRIES_DIR, { recursive: true, force: true })
|
||||
mkdirSync(ENTRIES_DIR, { recursive: true })
|
||||
mkdirSync(BUNDLES_DIR, { recursive: true })
|
||||
|
||||
for (const spec of BUNDLES) {
|
||||
const entryPath = join(ENTRIES_DIR, `${spec.name}.entry.ts`)
|
||||
writeFileSync(entryPath, spec.entry, 'utf-8')
|
||||
|
||||
const result = await Bun.build({
|
||||
entrypoints: [entryPath],
|
||||
target: 'browser',
|
||||
format: 'iife',
|
||||
minify: true,
|
||||
sourcemap: 'none',
|
||||
root: APP_SIM_ROOT,
|
||||
})
|
||||
|
||||
if (!result.success) {
|
||||
for (const log of result.logs) {
|
||||
console.error(log)
|
||||
}
|
||||
throw new Error(`Failed to build sandbox bundle: ${spec.name}`)
|
||||
}
|
||||
|
||||
if (result.outputs.length === 0) {
|
||||
throw new Error(`No output produced for sandbox bundle: ${spec.name}`)
|
||||
}
|
||||
|
||||
const code = await result.outputs[0].text()
|
||||
const banner = `// sandbox bundle: ${spec.name}\n// generated by apps/sim/lib/execution/sandbox/bundles/build.ts\n// do not edit by hand. run \`bun run build:sandbox-bundles\` to regenerate.\n`
|
||||
writeFileSync(join(BUNDLES_DIR, spec.outFile), banner + code, 'utf-8')
|
||||
console.log(`built ${spec.outFile} (${code.length.toLocaleString()} chars)`)
|
||||
}
|
||||
|
||||
rmSync(ENTRIES_DIR, { recursive: true, force: true })
|
||||
}
|
||||
|
||||
main().catch((err) => {
|
||||
console.error(err)
|
||||
process.exit(1)
|
||||
})
|
||||
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
@@ -0,0 +1,27 @@
|
||||
import type { SandboxTask, SandboxTaskInput } from '@/lib/execution/sandbox/types'
|
||||
|
||||
/**
|
||||
* Helper that preserves the task's input type through declaration.
|
||||
* Mirrors the `task(...)` / `defineConfig(...)` pattern used elsewhere in the
|
||||
* codebase so sandbox tasks look familiar next to trigger.dev tasks.
|
||||
*/
|
||||
export function defineSandboxTask<TInput extends SandboxTaskInput = SandboxTaskInput>(
|
||||
task: SandboxTask<TInput>
|
||||
): SandboxTask<TInput> {
|
||||
if (!task.id || !/^[a-z][a-z0-9-]*$/.test(task.id)) {
|
||||
throw new Error(`Sandbox task id must be kebab-case: got "${task.id}"`)
|
||||
}
|
||||
const brokerNames = new Set<string>()
|
||||
for (const broker of task.brokers) {
|
||||
if (!/^[a-zA-Z_$][a-zA-Z0-9_$]*$/.test(broker.name)) {
|
||||
throw new Error(
|
||||
`Sandbox broker name must be a valid JS identifier: got "${broker.name}" on task "${task.id}"`
|
||||
)
|
||||
}
|
||||
if (brokerNames.has(broker.name)) {
|
||||
throw new Error(`Duplicate broker name "${broker.name}" on task "${task.id}"`)
|
||||
}
|
||||
brokerNames.add(broker.name)
|
||||
}
|
||||
return task
|
||||
}
|
||||
@@ -0,0 +1,94 @@
|
||||
import { createLogger } from '@sim/logger'
|
||||
import { generateShortId } from '@/lib/core/utils/uuid'
|
||||
import {
|
||||
executeInIsolatedVM,
|
||||
type IsolatedVMBrokerHandler,
|
||||
type IsolatedVMExecutionRequest,
|
||||
} from '@/lib/execution/isolated-vm'
|
||||
import type { SandboxBrokerContext, SandboxTaskInput } from '@/lib/execution/sandbox/types'
|
||||
import { getSandboxTask, type SandboxTaskId } from '@/sandbox-tasks/registry'
|
||||
|
||||
const logger = createLogger('SandboxRunTask')
|
||||
|
||||
export interface RunSandboxTaskOptions {
|
||||
/**
|
||||
* Owner key used by the isolated-vm pool for fairness + distributed leases.
|
||||
* Typically `user:<userId>` or `workspace:<workspaceId>`.
|
||||
*/
|
||||
ownerKey?: string
|
||||
/** Optional AbortSignal to cancel the execution early. */
|
||||
signal?: AbortSignal
|
||||
}
|
||||
|
||||
/**
|
||||
* Executes a sandbox task inside the shared isolated-vm pool and returns the
|
||||
* binary result buffer. Throws with a human-readable message if the task fails
|
||||
* so callers can propagate the error verbatim to UI.
|
||||
*/
|
||||
export async function runSandboxTask<TInput extends SandboxTaskInput>(
|
||||
taskId: SandboxTaskId,
|
||||
input: TInput,
|
||||
options: RunSandboxTaskOptions = {}
|
||||
): Promise<Buffer> {
|
||||
const task = getSandboxTask(taskId)
|
||||
const requestId = generateShortId(12)
|
||||
|
||||
const brokerContext: SandboxBrokerContext = {
|
||||
workspaceId: input.workspaceId,
|
||||
requestId,
|
||||
}
|
||||
const brokers: Record<string, IsolatedVMBrokerHandler> = {}
|
||||
for (const broker of task.brokers) {
|
||||
brokers[broker.name] = (args) => broker.handle(brokerContext, args)
|
||||
}
|
||||
|
||||
const request: IsolatedVMExecutionRequest = {
|
||||
code: input.code,
|
||||
params: {},
|
||||
envVars: {},
|
||||
contextVariables: {},
|
||||
timeoutMs: task.timeoutMs,
|
||||
requestId,
|
||||
ownerKey: options.ownerKey,
|
||||
ownerWeight: 1,
|
||||
task: {
|
||||
id: task.id,
|
||||
bundles: [...task.bundles],
|
||||
bootstrap: task.bootstrap,
|
||||
brokers: task.brokers.map((b) => b.name),
|
||||
finalize: task.finalize,
|
||||
},
|
||||
}
|
||||
|
||||
const start = Date.now()
|
||||
const result = await executeInIsolatedVM(request, { brokers, signal: options.signal })
|
||||
const elapsedMs = Date.now() - start
|
||||
|
||||
if (result.error) {
|
||||
logger.warn('Sandbox task failed', {
|
||||
taskId,
|
||||
requestId,
|
||||
elapsedMs,
|
||||
error: result.error.message,
|
||||
errorName: result.error.name,
|
||||
})
|
||||
const err = new Error(result.error.message)
|
||||
err.name = result.error.name || 'SandboxTaskError'
|
||||
if (result.error.stack) err.stack = result.error.stack
|
||||
throw err
|
||||
}
|
||||
|
||||
if (typeof result.bytesBase64 !== 'string' || result.bytesBase64.length === 0) {
|
||||
logger.error('Sandbox task returned no bytes', { taskId, requestId })
|
||||
throw new Error(`Sandbox task "${taskId}" finalize did not return any bytes`)
|
||||
}
|
||||
|
||||
const bytes = Buffer.from(result.bytesBase64, 'base64')
|
||||
logger.info('Sandbox task completed', {
|
||||
taskId,
|
||||
requestId,
|
||||
elapsedMs,
|
||||
bytes: bytes.length,
|
||||
})
|
||||
return task.toResult(new Uint8Array(bytes.buffer, bytes.byteOffset, bytes.byteLength), input)
|
||||
}
|
||||
@@ -0,0 +1,56 @@
|
||||
/**
|
||||
* Types for the sandbox task system.
|
||||
*
|
||||
* A `SandboxTask` is a recipe that tells the isolated-vm pool how to run a
|
||||
* particular kind of user code: which pre-built library bundles to install,
|
||||
* which host-side brokers to expose, and how to serialize the final result.
|
||||
*/
|
||||
|
||||
export type SandboxBundleName = 'pptxgenjs' | 'docx' | 'pdf-lib'
|
||||
|
||||
export interface SandboxBroker<TArgs = unknown, TResult = unknown> {
|
||||
/**
|
||||
* Name the isolate-side bootstrap references (e.g. `__brokers.workspaceFile`).
|
||||
* Must be a plain JS identifier segment.
|
||||
*/
|
||||
name: string
|
||||
/**
|
||||
* Host-side handler invoked when the isolate calls the broker.
|
||||
* `ctx` carries per-execution metadata (workspaceId, requestId, etc.).
|
||||
*/
|
||||
handle(ctx: SandboxBrokerContext, args: TArgs): Promise<TResult>
|
||||
}
|
||||
|
||||
export interface SandboxBrokerContext {
|
||||
workspaceId: string
|
||||
requestId: string
|
||||
}
|
||||
|
||||
export interface SandboxTaskInput {
|
||||
workspaceId: string
|
||||
code: string
|
||||
}
|
||||
|
||||
export interface SandboxTask<TInput extends SandboxTaskInput = SandboxTaskInput> {
|
||||
/** Kebab-case stable identifier, used for logging + lookups. */
|
||||
id: string
|
||||
/** Script execution timeout inside the isolate. */
|
||||
timeoutMs: number
|
||||
/** Library bundles to load as isolate globals before the bootstrap runs. */
|
||||
bundles: ReadonlyArray<SandboxBundleName>
|
||||
/** Host-side brokers this task is allowed to call from inside the isolate. */
|
||||
brokers: ReadonlyArray<SandboxBroker>
|
||||
/**
|
||||
* JS code run inside the isolate after bundles are installed and before
|
||||
* user code. Should hoist bundle globals to friendly names and install any
|
||||
* helper functions users expect (e.g. `getFileBase64`).
|
||||
*/
|
||||
bootstrap: string
|
||||
/**
|
||||
* JS source that, when evaluated inside an async IIFE after user code, must
|
||||
* return a `Uint8Array`. The bytes are transferred out via `ExternalCopy`.
|
||||
*/
|
||||
finalize: string
|
||||
/** Host-side transform from raw isolate bytes to the caller's return type. */
|
||||
toResult(bytes: Uint8Array, input: TInput): Buffer
|
||||
}
|
||||
@@ -94,8 +94,7 @@ const nextConfig: NextConfig = {
|
||||
'/*': [
|
||||
'./node_modules/sharp/**/*',
|
||||
'./node_modules/@img/**/*',
|
||||
'./dist/pptx-worker.cjs',
|
||||
'./dist/doc-worker.cjs',
|
||||
'./lib/execution/sandbox/bundles/*.cjs',
|
||||
],
|
||||
},
|
||||
experimental: {
|
||||
|
||||
@@ -16,9 +16,8 @@
|
||||
"load:workflow:baseline": "BASE_URL=${BASE_URL:-http://localhost:3000} WARMUP_DURATION=${WARMUP_DURATION:-10} WARMUP_RATE=${WARMUP_RATE:-2} PEAK_RATE=${PEAK_RATE:-8} HOLD_DURATION=${HOLD_DURATION:-20} bunx artillery run scripts/load/workflow-concurrency.yml",
|
||||
"load:workflow:waves": "BASE_URL=${BASE_URL:-http://localhost:3000} WAVE_ONE_DURATION=${WAVE_ONE_DURATION:-10} WAVE_ONE_RATE=${WAVE_ONE_RATE:-6} QUIET_DURATION=${QUIET_DURATION:-5} WAVE_TWO_DURATION=${WAVE_TWO_DURATION:-15} WAVE_TWO_RATE=${WAVE_TWO_RATE:-8} WAVE_THREE_DURATION=${WAVE_THREE_DURATION:-20} WAVE_THREE_RATE=${WAVE_THREE_RATE:-10} bunx artillery run scripts/load/workflow-waves.yml",
|
||||
"load:workflow:isolation": "BASE_URL=${BASE_URL:-http://localhost:3000} ISOLATION_DURATION=${ISOLATION_DURATION:-30} TOTAL_RATE=${TOTAL_RATE:-9} WORKSPACE_A_WEIGHT=${WORKSPACE_A_WEIGHT:-8} WORKSPACE_B_WEIGHT=${WORKSPACE_B_WEIGHT:-1} bunx artillery run scripts/load/workflow-isolation.yml",
|
||||
"build": "bun run build:pptx-worker && bun run build:doc-worker && next build",
|
||||
"build:pptx-worker": "bun build ./lib/execution/pptx-worker.cjs --target=node --format=cjs --outfile ./dist/pptx-worker.cjs",
|
||||
"build:doc-worker": "bun build ./lib/execution/doc-worker.cjs --target=node --format=cjs --outfile ./dist/doc-worker.cjs",
|
||||
"build": "bun run build:sandbox-bundles && next build",
|
||||
"build:sandbox-bundles": "bun run ./lib/execution/sandbox/bundles/build.ts",
|
||||
"start": "next start",
|
||||
"prepare": "cd ../.. && bun husky",
|
||||
"test": "vitest run",
|
||||
|
||||
@@ -0,0 +1,52 @@
|
||||
import { workspaceFileBroker } from '@/lib/execution/sandbox/brokers/workspace-file'
|
||||
import { defineSandboxTask } from '@/lib/execution/sandbox/define-task'
|
||||
import type { SandboxTaskInput } from '@/lib/execution/sandbox/types'
|
||||
|
||||
export const docxGenerateTask = defineSandboxTask<SandboxTaskInput>({
|
||||
id: 'docx-generate',
|
||||
timeoutMs: 60_000,
|
||||
bundles: ['docx'],
|
||||
brokers: [workspaceFileBroker],
|
||||
bootstrap: `
|
||||
const docx = globalThis.__bundles['docx'];
|
||||
if (!docx) throw new Error('docx bundle not loaded');
|
||||
globalThis.docx = docx;
|
||||
globalThis.__docxSections = [];
|
||||
globalThis.addSection = (section) => {
|
||||
globalThis.__docxSections.push(section);
|
||||
};
|
||||
globalThis.getFileBase64 = async (fileId) => {
|
||||
const res = await globalThis.__brokers.workspaceFile({ fileId });
|
||||
return res.dataUri;
|
||||
};
|
||||
`,
|
||||
// JSZip's browser build doesn't support nodebuffer output, so we go through
|
||||
// base64 and decode back to bytes inside the isolate (avoids DataURL / Blob).
|
||||
finalize: `
|
||||
let doc = globalThis.doc;
|
||||
if (!doc && globalThis.__docxSections.length > 0) {
|
||||
doc = new globalThis.docx.Document({ sections: globalThis.__docxSections });
|
||||
}
|
||||
if (!doc) {
|
||||
throw new Error('No document created. Use addSection({ children: [...] }) for chunked writes, or set doc = new docx.Document({...}) for a single write.');
|
||||
}
|
||||
const b64 = await globalThis.docx.Packer.toBase64String(doc);
|
||||
const alphabet = 'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/';
|
||||
const lookup = new Uint8Array(128);
|
||||
for (let i = 0; i < alphabet.length; i++) lookup[alphabet.charCodeAt(i)] = i;
|
||||
const clean = b64.replace(/=+$/, '');
|
||||
const out = new Uint8Array(Math.floor((clean.length * 3) / 4));
|
||||
let pos = 0;
|
||||
for (let i = 0; i < clean.length; i += 4) {
|
||||
const c0 = lookup[clean.charCodeAt(i)];
|
||||
const c1 = lookup[clean.charCodeAt(i + 1)];
|
||||
const c2 = lookup[clean.charCodeAt(i + 2)];
|
||||
const c3 = lookup[clean.charCodeAt(i + 3)];
|
||||
out[pos++] = (c0 << 2) | (c1 >> 4);
|
||||
if (i + 2 < clean.length) out[pos++] = ((c1 & 0x0f) << 4) | (c2 >> 2);
|
||||
if (i + 3 < clean.length) out[pos++] = ((c2 & 0x03) << 6) | c3;
|
||||
}
|
||||
return out.subarray(0, pos);
|
||||
`,
|
||||
toResult: (bytes) => Buffer.from(bytes),
|
||||
})
|
||||
@@ -0,0 +1,33 @@
|
||||
/**
|
||||
* Everything that runs inside the isolated-vm pool lives in one of two tiers.
|
||||
* This file is the single grep surface for "what user code can execute in the
|
||||
* sandbox" — if you are adding a new place that spawns into the isolate,
|
||||
* register it below and/or extend one of the two tiers.
|
||||
*
|
||||
* Tier 1 — Sandbox tasks (this folder).
|
||||
* Bytes-producing tasks with a fixed input shape (`{ workspaceId, code }`),
|
||||
* pre-loaded library bundles, and a host-side broker set. User code is
|
||||
* trusted to the extent that our bootstrap + finalize wrap every execution.
|
||||
* Invoked via `runSandboxTask(id, input, options?)`.
|
||||
* - `pptx-generate` → `apps/sim/sandbox-tasks/pptx-generate.ts`
|
||||
* - `docx-generate` → `apps/sim/sandbox-tasks/docx-generate.ts`
|
||||
* - `pdf-generate` → `apps/sim/sandbox-tasks/pdf-generate.ts`
|
||||
*
|
||||
* Tier 2 — Raw isolated-vm consumers.
|
||||
* Value-producing executions where the user supplies arbitrary JS and the
|
||||
* host consumes whatever the code returns. Different contract (no finalize,
|
||||
* no bundles, no broker allowlist — just the built-in fetch bridge) so they
|
||||
* call `executeInIsolatedVM` directly rather than going through
|
||||
* `runSandboxTask`. If you add a new Tier 2 caller, record it here so the
|
||||
* set of sandbox entry points stays grep-able from one place.
|
||||
* - `apps/sim/app/api/function/execute/route.ts` — user function blocks
|
||||
* - `apps/sim/executor/orchestrators/loop.ts` — loop-condition eval
|
||||
*
|
||||
* E2B-routed executions (untrusted workflow runs) are a separate runtime
|
||||
* entirely and are not part of this registry.
|
||||
*/
|
||||
|
||||
export { docxGenerateTask } from '@/sandbox-tasks/docx-generate'
|
||||
export { pdfGenerateTask } from '@/sandbox-tasks/pdf-generate'
|
||||
export { pptxGenerateTask } from '@/sandbox-tasks/pptx-generate'
|
||||
export { getSandboxTask, SANDBOX_TASKS, type SandboxTaskId } from '@/sandbox-tasks/registry'
|
||||
@@ -0,0 +1,39 @@
|
||||
import { workspaceFileBroker } from '@/lib/execution/sandbox/brokers/workspace-file'
|
||||
import { defineSandboxTask } from '@/lib/execution/sandbox/define-task'
|
||||
import type { SandboxTaskInput } from '@/lib/execution/sandbox/types'
|
||||
|
||||
export const pdfGenerateTask = defineSandboxTask<SandboxTaskInput>({
|
||||
id: 'pdf-generate',
|
||||
timeoutMs: 60_000,
|
||||
bundles: ['pdf-lib'],
|
||||
brokers: [workspaceFileBroker],
|
||||
bootstrap: `
|
||||
const PDFLib = globalThis.__bundles['pdf-lib'];
|
||||
if (!PDFLib) throw new Error('pdf-lib bundle not loaded');
|
||||
globalThis.PDFLib = PDFLib;
|
||||
globalThis.pdf = await PDFLib.PDFDocument.create();
|
||||
globalThis.embedImage = async (dataUri) => {
|
||||
const comma = dataUri.indexOf(',');
|
||||
const header = dataUri.slice(0, comma);
|
||||
const base64 = dataUri.slice(comma + 1);
|
||||
const binary = globalThis.Buffer ? globalThis.Buffer.from(base64, 'base64') : null;
|
||||
if (!binary) throw new Error('Buffer polyfill missing');
|
||||
const mime = header.split(';')[0].split(':')[1] || '';
|
||||
if (mime.includes('png')) return globalThis.pdf.embedPng(binary);
|
||||
return globalThis.pdf.embedJpg(binary);
|
||||
};
|
||||
globalThis.getFileBase64 = async (fileId) => {
|
||||
const res = await globalThis.__brokers.workspaceFile({ fileId });
|
||||
return res.dataUri;
|
||||
};
|
||||
`,
|
||||
finalize: `
|
||||
const pdf = globalThis.pdf;
|
||||
if (!pdf) {
|
||||
throw new Error('No PDF document. Use the injected pdf object or load one with PDFLib.PDFDocument.load().');
|
||||
}
|
||||
const bytes = await pdf.save();
|
||||
return bytes;
|
||||
`,
|
||||
toResult: (bytes) => Buffer.from(bytes),
|
||||
})
|
||||
@@ -0,0 +1,24 @@
|
||||
import { workspaceFileBroker } from '@/lib/execution/sandbox/brokers/workspace-file'
|
||||
import { defineSandboxTask } from '@/lib/execution/sandbox/define-task'
|
||||
import type { SandboxTaskInput } from '@/lib/execution/sandbox/types'
|
||||
|
||||
export const pptxGenerateTask = defineSandboxTask<SandboxTaskInput>({
|
||||
id: 'pptx-generate',
|
||||
timeoutMs: 60_000,
|
||||
bundles: ['pptxgenjs'],
|
||||
brokers: [workspaceFileBroker],
|
||||
bootstrap: `
|
||||
const PptxGenJS = globalThis.__bundles['pptxgenjs'];
|
||||
if (!PptxGenJS) throw new Error('pptxgenjs bundle not loaded');
|
||||
globalThis.pptx = new PptxGenJS();
|
||||
globalThis.getFileBase64 = async (fileId) => {
|
||||
const res = await globalThis.__brokers.workspaceFile({ fileId });
|
||||
return res.dataUri;
|
||||
};
|
||||
`,
|
||||
finalize: `
|
||||
const bytes = await globalThis.pptx.write({ outputType: 'uint8array' });
|
||||
return bytes;
|
||||
`,
|
||||
toResult: (bytes) => Buffer.from(bytes),
|
||||
})
|
||||
@@ -0,0 +1,25 @@
|
||||
import type { SandboxTask } from '@/lib/execution/sandbox/types'
|
||||
import { docxGenerateTask } from '@/sandbox-tasks/docx-generate'
|
||||
import { pdfGenerateTask } from '@/sandbox-tasks/pdf-generate'
|
||||
import { pptxGenerateTask } from '@/sandbox-tasks/pptx-generate'
|
||||
|
||||
/**
|
||||
* Every piece of user code that runs inside the isolated-vm sandbox is defined
|
||||
* here. Adding a new sandbox task = add one file under `apps/sim/sandbox-tasks/`
|
||||
* and register it below. Mirrors the `apps/sim/background/` pattern.
|
||||
*/
|
||||
export const SANDBOX_TASKS = {
|
||||
'pptx-generate': pptxGenerateTask,
|
||||
'docx-generate': docxGenerateTask,
|
||||
'pdf-generate': pdfGenerateTask,
|
||||
} as const satisfies Record<string, SandboxTask>
|
||||
|
||||
export type SandboxTaskId = keyof typeof SANDBOX_TASKS
|
||||
|
||||
export function getSandboxTask(id: SandboxTaskId): SandboxTask {
|
||||
const task = SANDBOX_TASKS[id]
|
||||
if (!task) {
|
||||
throw new Error(`Unknown sandbox task: "${id}"`)
|
||||
}
|
||||
return task
|
||||
}
|
||||
@@ -15,25 +15,18 @@ export default defineConfig({
|
||||
},
|
||||
dirs: ['./background'],
|
||||
build: {
|
||||
external: ['isolated-vm', 'pptxgenjs', 'docx', 'pdf-lib'],
|
||||
external: ['isolated-vm'],
|
||||
extensions: [
|
||||
additionalFiles({
|
||||
files: [
|
||||
'./lib/execution/isolated-vm-worker.cjs',
|
||||
'./lib/execution/pptx-worker.cjs',
|
||||
'./lib/execution/doc-worker.cjs',
|
||||
'./lib/execution/sandbox/bundles/pptxgenjs.cjs',
|
||||
'./lib/execution/sandbox/bundles/docx.cjs',
|
||||
'./lib/execution/sandbox/bundles/pdf-lib.cjs',
|
||||
],
|
||||
}),
|
||||
additionalPackages({
|
||||
packages: [
|
||||
'unpdf',
|
||||
'pdf-lib',
|
||||
'isolated-vm',
|
||||
'pptxgenjs',
|
||||
'docx',
|
||||
'react-dom',
|
||||
'@react-email/render',
|
||||
],
|
||||
packages: ['unpdf', 'isolated-vm', 'react-dom', '@react-email/render'],
|
||||
}),
|
||||
],
|
||||
},
|
||||
|
||||
@@ -113,9 +113,10 @@ COPY --from=deps --chown=nextjs:nodejs /app/node_modules/isolated-vm ./node_modu
|
||||
# Copy the isolated-vm worker script
|
||||
COPY --from=builder --chown=nextjs:nodejs /app/apps/sim/lib/execution/isolated-vm-worker.cjs ./apps/sim/lib/execution/isolated-vm-worker.cjs
|
||||
|
||||
# Copy the bundled worker artifacts
|
||||
COPY --from=builder --chown=nextjs:nodejs /app/apps/sim/dist/pptx-worker.cjs ./apps/sim/dist/pptx-worker.cjs
|
||||
COPY --from=builder --chown=nextjs:nodejs /app/apps/sim/dist/doc-worker.cjs ./apps/sim/dist/doc-worker.cjs
|
||||
# Copy the pre-built sandbox library bundles (pptxgenjs, docx, pdf-lib) that
|
||||
# run inside the V8 isolate. Committed into the repo; see
|
||||
# apps/sim/lib/execution/sandbox/bundles/build.ts to regenerate.
|
||||
COPY --from=builder --chown=nextjs:nodejs /app/apps/sim/lib/execution/sandbox/bundles ./apps/sim/lib/execution/sandbox/bundles
|
||||
|
||||
# Guardrails setup with pip caching
|
||||
COPY --from=builder --chown=nextjs:nodejs /app/apps/sim/lib/guardrails/requirements.txt ./apps/sim/lib/guardrails/requirements.txt
|
||||
|
||||
Reference in New Issue
Block a user