diff --git a/bun.lock b/bun.lock index 7f868a1..2ad3596 100644 --- a/bun.lock +++ b/bun.lock @@ -1,5 +1,6 @@ { "lockfileVersion": 1, + "configVersion": 0, "workspaces": { "": { "name": "@org/source", @@ -88,7 +89,7 @@ "@assemblyscript/loader": ["@assemblyscript/loader@0.25.2", "", {}, "sha512-q2PyrBmepQ0YxlWN99Vb1MmOSL982ePhGRxAscX2XlZZPlzhv5DgH5UAefTUFgBHfQE8FY1CWONBIvPxS7Mjjw=="], - "@assemblyscript/wasi-shim": ["@assemblyscript/wasi-shim@github:sedaprotocol/wasi-shim#8fda40c", {}, "sedaprotocol-wasi-shim-8fda40c"], + "@assemblyscript/wasi-shim": ["@assemblyscript/wasi-shim@github:sedaprotocol/wasi-shim#8fda40c", {}, "sedaprotocol-wasi-shim-8fda40c", "sha512-qbNo0kUd0Maj6Kopn4PxGAQUFxLXw8x2nMscCs4QViWSA1yRWqaewKdPZzvBrcCa3Ck8ypTxwWhj8cr3fg2GDA=="], "@babel/code-frame": ["@babel/code-frame@7.26.2", "", { "dependencies": { "@babel/helper-validator-identifier": "^7.25.9", "js-tokens": "^4.0.0", "picocolors": "^1.0.0" } }, "sha512-RJlIHRueQgwWitWgF8OdFYGZX328Ax5BCemNGlqHfplnRT9ESi8JkFlvaVYbS+UubVY6dpv87Fs2u5M29iNFVQ=="], diff --git a/libs/vm/package.json b/libs/vm/package.json index ed437aa..bb97b5b 100644 --- a/libs/vm/package.json +++ b/libs/vm/package.json @@ -1,7 +1,7 @@ { "name": "@seda-protocol/vm", "type": "module", - "version": "1.1.3", + "version": "1.2.0", "dependencies": { "@seda-protocol/wasm-metering-ts": "^2.0.1", "uwasi": "^1.4.1", diff --git a/libs/vm/src/call-vm.test.ts b/libs/vm/src/call-vm.test.ts new file mode 100644 index 0000000..c9d5475 --- /dev/null +++ b/libs/vm/src/call-vm.test.ts @@ -0,0 +1,320 @@ +import { describe, expect, it, setDefaultTimeout } from "bun:test"; +import { existsSync, readFileSync } from "node:fs"; +import { mkdtempSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join, resolve } from "node:path"; +import { pathToFileURL } from "node:url"; +import { Worker } from "node:worker_threads"; +import { Result } from "true-myth"; +import { TallyVmAdapter, callVm } from "./index.js"; +import { WasmModuleCache } from "./services/wasm-module-cache.js"; +import { createWasmModule } from "./services/wasm-module.js"; +import { + type HttpFetchAction, + HttpFetchResponse, + type ProxyHttpFetchAction, +} from "./types/vm-actions.js"; +import type { VmAdapter } from "./types/vm-adapter.js"; +import type { VmCallData } from "./types/vm-call-data.js"; +import { PromiseStatus } from "./types/vm-promise.js"; +import { + type VmActionExecuteMessage, + WorkerMessageType, +} from "./types/worker-messages.js"; +import { HostToWorker } from "./worker-host-communication.js"; + +setDefaultTimeout(30_000); + +// This file also runs from the compiled dist output (bun test matches both +// copies), where modules are .js and the repo layout sits one level deeper. +function firstExisting(candidates: string[]): string { + const found = candidates.find((candidate) => existsSync(candidate)); + if (!found) { + throw new Error( + `None of the candidate paths exist: ${candidates.join(", ")}`, + ); + } + return found; +} + +const WORKER_URL = pathToFileURL( + firstExisting([ + resolve(import.meta.dir, "worker.ts"), + resolve(import.meta.dir, "worker.js"), + ]), +); +const BRIDGE_WORKER_URL = pathToFileURL( + firstExisting([ + resolve(import.meta.dir, "test-helpers/bridge-test-worker.ts"), + resolve(import.meta.dir, "test-helpers/bridge-test-worker.js"), + ]), +); + +const tallyProgram = readFileSync( + firstExisting([ + resolve(import.meta.dir, "../../wasm-integration-tests/test-vm.wasm"), + resolve( + import.meta.dir, + "../../../../libs/wasm-integration-tests/test-vm.wasm", + ), + ]), +); + +const PRICE_FEED_REVEALS = + '[{"salt":[115,101,100,97,95,115,100,107],"exit_code":0,"gas_used":200,"reveal":[123,34,112,114,105,99,101,34,58,32,49,49,50,57,57,51,50,125]},{"salt":[115,101,100,97,95,115,100,107],"exit_code":0,"gas_used":198,"reveal":[123,34,112,114,105,99,101,34,58,32,49,49,50,57,57,52,53,125]},{"salt":[115,101,100,97,95,115,100,107],"exit_code":0,"gas_used":201,"reveal":[123,34,112,114,105,99,101,34,58,32,49,49,50,57,57,50,56,125]},{"salt":[115,101,100,97,95,115,100,107],"exit_code":0,"gas_used":199,"reveal":[123,34,112,114,105,99,101,34,58,32,49,49,50,57,57,51,55,125]},{"salt":[115,101,100,97,95,115,100,107],"exit_code":0,"gas_used":202,"reveal":[123,34,112,114,105,99,101,34,58,32,49,49,50,57,57,51,48,125]},{"salt":[115,101,100,97,95,115,100,107],"exit_code":0,"gas_used":197,"reveal":[123,34,112,114,105,99,101,34,58,32,49,49,50,57,57,52,49,125]},{"salt":[115,101,100,97,95,115,100,107],"exit_code":0,"gas_used":200,"reveal":[123,34,112,114,105,99,101,34,58,32,49,49,50,57,57,51,53,125]},{"salt":[115,101,100,97,95,115,100,107],"exit_code":0,"gas_used":203,"reveal":[123,34,112,114,105,99,101,34,58,32,49,49,50,57,57,51,57,125]},{"salt":[115,101,100,97,95,115,100,107],"exit_code":0,"gas_used":196,"reveal":[123,34,112,114,105,99,101,34,58,32,49,49,50,57,57,51,51,125]},{"salt":[115,101,100,97,95,115,100,107],"exit_code":0,"gas_used":201,"reveal":[123,34,112,114,105,99,101,34,58,32,49,49,50,57,57,51,54,125]}]'; +const PRICE_FEED_CONSENSUS = "[0,0,0,0,0,0,0,0,0,0]"; + +function priceFeedCallData( + binary: VmCallData["binary"] = tallyProgram, +): VmCallData { + return { + args: [ + Buffer.from("price_feed_tally").toString("hex"), + PRICE_FEED_REVEALS, + PRICE_FEED_CONSENSUS, + ], + binary, + envs: { + CONSENSUS: "true", + VM_MODE: "tally", + DR_TALLY_GAS_LIMIT: "50000000000000", + }, + vmMode: "tally", + gasLimit: 50000000000000n, + }; +} + +function stderrCallData(): VmCallData { + return { + args: [Buffer.from("call_result_write_0").toString("hex"), "[]", "[]"], + binary: tallyProgram, + envs: { + CONSENSUS: "true", + VM_MODE: "tally", + DR_TALLY_GAS_LIMIT: "150000000000000", + }, + vmMode: "tally", + }; +} + +describe("callVm worker reuse", () => { + it("reuses a persistent worker across sequential calls without listener growth or cross-talk", async () => { + const worker = new Worker(WORKER_URL); + + try { + for (let i = 0; i < 3; i++) { + const priceResult = await callVm( + priceFeedCallData(), + worker, + new TallyVmAdapter(), + ); + expect(priceResult.stdout).toBe("1129935\n"); + expect(priceResult.exitCode).toBe(0); + + const errResult = await callVm( + stderrCallData(), + worker, + new TallyVmAdapter(), + ); + expect(errResult.stderr).toInclude( + "call_result_write: result_data_ptr length does not match call_value length", + ); + + expect(worker.listenerCount("message")).toBe(0); + expect(worker.listenerCount("error")).toBe(0); + expect(worker.listenerCount("exit")).toBe(0); + } + } finally { + await worker.terminate(); + } + }); + + it("produces results identical to sync (in-process) mode", async () => { + const worker = new Worker(WORKER_URL); + + try { + const workerResult = await callVm( + priceFeedCallData(), + worker, + new TallyVmAdapter(), + ); + const syncResult = await callVm( + priceFeedCallData(), + undefined, + new TallyVmAdapter(), + true, + ); + + expect(workerResult.exitCode).toBe(syncResult.exitCode); + expect(workerResult.stdout).toBe(syncResult.stdout); + expect(workerResult.stderr).toBe(syncResult.stderr); + expect(workerResult.gasUsed).toBe(syncResult.gasUsed); + expect(workerResult.resultAsString).toBe(syncResult.resultAsString); + } finally { + await worker.terminate(); + } + }); + + it("strips the host-side WasmModuleCache and runs through the worker-local cache", async () => { + const worker = new Worker(WORKER_URL); + const cacheDir = mkdtempSync(join(tmpdir(), "vm-cache-")); + + try { + for (let i = 0; i < 2; i++) { + const result = await callVm( + { + ...priceFeedCallData(), + cache: { dir: cacheDir, id: "price-feed-test" }, + wasmModuleCache: new WasmModuleCache(5), + }, + worker, + new TallyVmAdapter(), + ); + + expect(result.exitCode).toBe(0); + expect(result.stdout).toBe("1129935\n"); + } + } finally { + await worker.terminate(); + } + }); + + it("accepts a precompiled WebAssembly.Module as binary", async () => { + const moduleResult = await createWasmModule(tallyProgram, "tally"); + expect(moduleResult.isOk).toBe(true); + if (moduleResult.isErr) return; + + const worker = new Worker(WORKER_URL); + + try { + const result = await callVm( + priceFeedCallData(moduleResult.value), + worker, + new TallyVmAdapter(), + ); + + expect(result.exitCode).toBe(0); + expect(result.stdout).toBe("1129935\n"); + } finally { + await worker.terminate(); + } + }); +}); + +class MockBridgeAdapter implements VmAdapter { + modifyVmCallData(input: VmCallData): VmCallData { + return input; + } + + setProcessId(_processId: string) {} + + async httpFetch( + action: HttpFetchAction, + ): Promise> { + const body = Array.from(new TextEncoder().encode("bridge-response")); + + return PromiseStatus.fulfilled( + new HttpFetchResponse({ + bytes: body, + content_length: body.length, + headers: {}, + status: 200, + url: action.url, + }), + ); + } + + async getProxyHttpFetchGasCost( + _action: ProxyHttpFetchAction, + ): Promise> { + return Result.ok(0n); + } + + async proxyHttpFetch( + _action: ProxyHttpFetchAction, + ): Promise> { + return HttpFetchResponse.createRejectedPromise("not used"); + } + + getClockTime(_mode: "monotonic" | "realtime"): number { + return 0; + } +} + +describe("WorkerToHost bridge", () => { + it("round-trips a host action with the processId attached to every message", async () => { + const processId = "bridge-test-1"; + const notifierBuffer = new SharedArrayBuffer(8); + const hostToWorker = new HostToWorker( + new MockBridgeAdapter(), + processId, + notifierBuffer, + ); + + const worker = new Worker(BRIDGE_WORKER_URL, { + workerData: { notifierBuffer, processId, hostCallTimeoutMs: 5_000 }, + }); + + const seenProcessIds: string[] = []; + + try { + const reply = await new Promise<{ ok: boolean; resultUtf8?: string }>( + (resolve, reject) => { + worker.on("message", async (message) => { + if (message.type === WorkerMessageType.VmActionExecute) { + seenProcessIds.push( + (message as VmActionExecuteMessage).processId, + ); + await hostToWorker.executeAction(message.action); + } else if ( + message.type === WorkerMessageType.VmActionResultBuffer + ) { + seenProcessIds.push(message.processId); + await hostToWorker.sendActionResultToWorker(message.buffer); + } else { + resolve(message); + } + }); + worker.on("error", reject); + }, + ); + + expect(reply.ok).toBe(true); + expect(reply.resultUtf8).toInclude("Fulfilled"); + expect(seenProcessIds).toEqual([processId, processId]); + } finally { + await worker.terminate(); + } + }); + + it("times out instead of wedging the worker when the host never answers", async () => { + const processId = "bridge-test-timeout"; + const notifierBuffer = new SharedArrayBuffer(8); + + const worker = new Worker(BRIDGE_WORKER_URL, { + workerData: { notifierBuffer, processId, hostCallTimeoutMs: 200 }, + }); + + try { + const reply = await new Promise<{ ok: boolean; error?: string }>( + (resolve, reject) => { + worker.on("message", (message) => { + // Ignore the action requests: this host deliberately never answers + if ( + message.type !== WorkerMessageType.VmActionExecute && + message.type !== WorkerMessageType.VmActionResultBuffer + ) { + resolve(message); + } + }); + worker.on("error", reject); + }, + ); + + expect(reply.ok).toBe(false); + expect(reply.error).toInclude("did not answer the VM action"); + } finally { + await worker.terminate(); + } + }); +}); diff --git a/libs/vm/src/errors.ts b/libs/vm/src/errors.ts index b0f5089..37012be 100644 --- a/libs/vm/src/errors.ts +++ b/libs/vm/src/errors.ts @@ -2,6 +2,7 @@ export enum VmErrorType { OutOfGas = "OutOfGas", HttpFetchTimeout = "HttpFetchTimeout", HttpFetchGlobalTimeout = "HttpFetchGlobalTimeout", + HostCallTimeout = "HostCallTimeout", InsufficientDataProxyFee = "InsufficientDataProxyFee", Unknown = "Unknown", } diff --git a/libs/vm/src/index.ts b/libs/vm/src/index.ts index d276ee4..30928f4 100644 --- a/libs/vm/src/index.ts +++ b/libs/vm/src/index.ts @@ -1,3 +1,4 @@ +import { randomBytes } from "node:crypto"; import { format, parse } from "node:path"; import { Worker } from "node:worker_threads"; import { tryAsync } from "@seda-protocol/utils"; @@ -44,8 +45,14 @@ export { /** * Executes the given WASM binary as if it were an Oracle Program * + * A persistent `Worker` may be passed in and reused across sequential calls: + * all message traffic is scoped by a per-call process id and the listeners are + * removed once the call settles. Only one call may be in flight per worker at + * a time (the Atomics notifier buffer is per-call state); callers reusing a + * worker must queue calls themselves. + * * @param callData The call data passed to the VM - * @param worker URL of the compiled worker.js + * @param worker URL of the compiled worker.js, or a (reusable) Worker instance * @param vmAdapter Option to insert a custom VM adapter, can be used to mock * @returns */ @@ -59,7 +66,11 @@ export function callVm( return new Promise(async (resolve) => { const finalCallData: VmCallData = vmAdapter.modifyVmCallData(callData); - const processId = createProcessId(finalCallData); + // The nonce makes the id unique per call so a reused worker's stale + // messages (same binary and args) can never be attributed to this call. + const processId = sync + ? createProcessId(finalCallData) + : `${createProcessId(finalCallData)}-${randomBytes(8).toString("hex")}`; vmAdapter.setProcessId(processId); // We run in a synchronous environment, no workers are used. @@ -72,38 +83,62 @@ export function callVm( } let vmWorker: Worker; + let ownsWorker = false; if (worker === undefined) { const CURRENT_FILE_PATH = parse(import.meta.url); CURRENT_FILE_PATH.base = "worker.js"; const DEFAULT_WORKER_PATH = format(CURRENT_FILE_PATH); vmWorker = new Worker(DEFAULT_WORKER_PATH); + ownsWorker = true; + } else if (typeof worker === "string") { + vmWorker = new Worker(new URL(worker)); + ownsWorker = true; } else { - vmWorker = - typeof worker === "string" ? new Worker(new URL(worker)) : worker; + vmWorker = worker; } const notifierBuffer = new SharedArrayBuffer(8); // 4 bytes for notifying, 4 bytes for storing i32 numbers const hostToWorker = new HostToWorker(vmAdapter, processId, notifierBuffer); + + // WasmModuleCache does not survive structured clone (it would arrive as + // a broken plain object); the worker keeps its own module cache. + const { wasmModuleCache: _hostCache, ...transferableCallData } = + finalCallData; const workerMessage: VmCallWorkerMessage = { processId, - callData: { - ...finalCallData, - binary: finalCallData.binary, - // binary: Array.from(finalCallData.binary), - }, + callData: transferableCallData, notifierBuffer, type: WorkerMessageType.VmCall, }; - vmWorker.on("message", async (message: WorkerMessage) => { + let settled = false; + const settle = (result: VmResult) => { + if (settled) return; + settled = true; + + vmWorker.off("message", onMessage); + vmWorker.off("error", onError); + vmWorker.off("exit", onExit); + + if (ownsWorker) { + void vmWorker.terminate(); + } + + resolve(result); + }; + + const onMessage = async (message: WorkerMessage) => { try { if (message.type === WorkerMessageType.VmResult) { - resolve(message.result); + if (message.processId !== processId) return; + settle(message.result); } else if (message.type === WorkerMessageType.VmActionExecute) { + if (message.processId !== processId) return; await hostToWorker.executeAction(message.action); } else if (message.type === WorkerMessageType.VmActionResultBuffer) { + if (message.processId !== processId) return; await hostToWorker.sendActionResultToWorker(message.buffer); } else { console.warn(`[${processId}] - Unknown message: ${message}`); @@ -111,25 +146,31 @@ export function callVm( } catch (error) { console.error(`[${processId}] - @callVm-onMessage: `, error); } - }); + }; - vmWorker.on("error", (error) => { - resolve({ + const onError = (error: Error) => { + settle({ exitCode: 1, stderr: `[${processId}] - Worker threw an uncaught error: ${error}`, stdout: "", gasUsed: 0n, + crashed: true, }); - }); + }; - vmWorker.on("exit", (exitCode) => { - resolve({ + const onExit = (exitCode: number) => { + settle({ exitCode, stderr: `[${processId}] - The worker has been terminated`, stdout: "", gasUsed: 0n, + crashed: true, }); - }); + }; + + vmWorker.on("message", onMessage); + vmWorker.on("error", onError); + vmWorker.on("exit", onExit); vmWorker.postMessage(workerMessage); }); diff --git a/libs/vm/src/test-helpers/bridge-test-worker.ts b/libs/vm/src/test-helpers/bridge-test-worker.ts new file mode 100644 index 0000000..13a57f4 --- /dev/null +++ b/libs/vm/src/test-helpers/bridge-test-worker.ts @@ -0,0 +1,31 @@ +import { parentPort, workerData } from "node:worker_threads"; +import { HttpFetchMethod } from "../types/vm-actions.js"; +import { WorkerToHost } from "../worker-host-communication.js"; + +const { notifierBuffer, processId, hostCallTimeoutMs } = workerData as { + notifierBuffer: SharedArrayBuffer; + processId: string; + hostCallTimeoutMs: number; +}; + +const workerToHost = new WorkerToHost( + notifierBuffer, + [], + processId, + hostCallTimeoutMs, +); + +try { + const result = workerToHost.callActionOnHost({ + type: "http-fetch-action", + url: "https://bridge.test/value", + options: { + method: HttpFetchMethod.Get, + headers: {}, + }, + }); + + parentPort?.postMessage({ ok: true, resultUtf8: result.toString("utf-8") }); +} catch (error) { + parentPort?.postMessage({ ok: false, error: String(error) }); +} diff --git a/libs/vm/src/types/vm-adapter.ts b/libs/vm/src/types/vm-adapter.ts index 13d04d9..467f62d 100644 --- a/libs/vm/src/types/vm-adapter.ts +++ b/libs/vm/src/types/vm-adapter.ts @@ -40,6 +40,17 @@ export interface VmAdapter { action: ProxyHttpFetchAction, ): Promise>; + /** + * Optional synchronous variant of getProxyHttpFetchGasCost for adapters + * with a flat fee model. When implemented and returning a value, the + * in-process (replay) mode answers the gas cost without a host action, + * which saves one full abort-and-replay execution of the wasm. Return + * undefined to fall back to the async host action. + */ + getProxyHttpFetchGasCostSync?( + action: ProxyHttpFetchAction, + ): bigint | undefined; + /** * Method to do a remote proxy http fetch call * diff --git a/libs/vm/src/types/vm-call-data.ts b/libs/vm/src/types/vm-call-data.ts index f2f6cab..c5c5b99 100644 --- a/libs/vm/src/types/vm-call-data.ts +++ b/libs/vm/src/types/vm-call-data.ts @@ -16,6 +16,12 @@ export interface VmCallData { stdoutLimit?: number; stderrLimit?: number; wasmModuleCache?: WasmModuleCache; + /** + * Maximum time (ms) a worker-mode VM blocks waiting for the host to answer + * a single host call before the call fails. Prevents a dead host from + * wedging the VM worker thread forever. Only used in worker (Atomics) mode. + */ + hostCallTimeoutMs?: number; } export function createCacheKey(processId: string, callData: VmCallData) { diff --git a/libs/vm/src/types/worker-messages.ts b/libs/vm/src/types/worker-messages.ts index 16149b9..8350240 100644 --- a/libs/vm/src/types/worker-messages.ts +++ b/libs/vm/src/types/worker-messages.ts @@ -18,16 +18,19 @@ export interface VmCallWorkerMessage { export interface VmResultWorkerMessage { result: VmResult; + processId: string; type: WorkerMessageType.VmResult; } export interface VmActionResultBufferMessage { buffer: SharedArrayBuffer; + processId: string; type: WorkerMessageType.VmActionResultBuffer; } export interface VmActionExecuteMessage { action: VmAction; + processId: string; type: WorkerMessageType.VmActionExecute; } diff --git a/libs/vm/src/vm-imports.ts b/libs/vm/src/vm-imports.ts index b3b029f..e016393 100644 --- a/libs/vm/src/vm-imports.ts +++ b/libs/vm/src/vm-imports.ts @@ -52,6 +52,8 @@ export default class VmImports { this.workerToHost = new WorkerToHost( notifierBufferOrAdapter, asyncRequests, + processId, + callData.hostCallTimeoutMs, ); this.processId = processId; this.gasMeter = gasMeter; @@ -85,25 +87,40 @@ export default class VmImports { ...JSON.parse(messageRaw), type: "proxy-http-fetch-action", }; - const gasCostMessage: ProxyHttpFetchGasCostAction = { - type: "proxy-http-fetch-gas-cost-action", - fetchAction: message, - }; + // Adapters with a flat fee model can answer the gas cost synchronously, + // skipping a host action that would otherwise abort and replay the VM + // in the in-process mode. Worker (Atomics) mode never aborts, so the + // fallback host action is cheap there either way. + const adapter = + this.notifierBufferOrAdapter instanceof SharedArrayBuffer + ? undefined + : this.notifierBufferOrAdapter; + const syncGasCost = adapter?.getProxyHttpFetchGasCostSync?.(message); + + let gasCost: bigint; + if (syncGasCost !== undefined) { + gasCost = syncGasCost; + } else { + const gasCostMessage: ProxyHttpFetchGasCostAction = { + type: "proxy-http-fetch-gas-cost-action", + fetchAction: message, + }; + + // First we try to fetch the price of the proxy call in gas units + const gasCostMessageResponse: ResultJSON = JSON.parse( + this.workerToHost.callActionOnHost(gasCostMessage).toString(), + ); - // First we try to fetch the price of the proxy call in gas units - const gasCostMessageResponse: ResultJSON = JSON.parse( - this.workerToHost.callActionOnHost(gasCostMessage).toString(), - ); + if (gasCostMessageResponse.variant === "Err") { + this.callResult = HttpFetchResponse.createRejectedPromise( + `Failed to get proxy gas cost: ${gasCostMessageResponse.error}`, + ).toBuffer(); - if (gasCostMessageResponse.variant === "Err") { - this.callResult = HttpFetchResponse.createRejectedPromise( - `Failed to get proxy gas cost: ${gasCostMessageResponse.error}`, - ).toBuffer(); + return this.callResult.length; + } - return this.callResult.length; + gasCost = BigInt(gasCostMessageResponse.value); } - - const gasCost = BigInt(gasCostMessageResponse.value); // Ensure we have enough gas to pay for the proxy call should it succeed if (this.gasMeter.pointsRemaining < gasCost) { const remainingPoints = this.gasMeter @@ -111,7 +128,7 @@ export default class VmImports { .mapOr("0", (t) => t.toString()); throw new VmError( - `Insufficient gas to pay for data proxy. ${gasCostMessageResponse.value} required, only ${remainingPoints} left`, + `Insufficient gas to pay for data proxy. ${gasCost} required, only ${remainingPoints} left`, { type: VmErrorType.InsufficientDataProxyFee, }, diff --git a/libs/vm/src/vm.ts b/libs/vm/src/vm.ts index 14a4bbf..3ed34db 100644 --- a/libs/vm/src/vm.ts +++ b/libs/vm/src/vm.ts @@ -20,6 +20,12 @@ export interface VmResult { gasUsed: bigint; result?: Uint8Array; resultAsString?: string; + /** + * Set when the result was synthesized because the worker thread died + * (error/exit) instead of completing the VM call. Lets pool implementations + * distinguish a crashed worker from a program that exited non-zero. + */ + crashed?: boolean; } type PropertyWithMessage = { diff --git a/libs/vm/src/worker-host-communication.ts b/libs/vm/src/worker-host-communication.ts index c60a8d6..a7a9ca9 100644 --- a/libs/vm/src/worker-host-communication.ts +++ b/libs/vm/src/worker-host-communication.ts @@ -21,6 +21,9 @@ import { WorkerMessageType } from "./types/worker-messages.js"; const MAX_I32_VALUE = 2_147_483_647; +/** Generous default so slow-but-alive hosts are never cut off; override via VmCallData.hostCallTimeoutMs */ +export const DEFAULT_HOST_CALL_TIMEOUT_MS = 300_000; + /** Location where the worker thread should listen for changes */ const NOTIFIER_INDEX = 0; @@ -44,6 +47,7 @@ function resetNotifierState(buffer: Int32Array) { function waitForNotifierStateChange( buffer: Int32Array, initialState: AtomicState, + timeoutMs: number, ) { const currentState = Atomics.load(buffer, NOTIFIER_INDEX); @@ -54,7 +58,19 @@ function waitForNotifierStateChange( } Atomics.store(buffer, NOTIFIER_INDEX, initialState); - Atomics.wait(buffer, NOTIFIER_INDEX, initialState); + const waitResult = Atomics.wait( + buffer, + NOTIFIER_INDEX, + initialState, + timeoutMs, + ); + + if (waitResult === "timed-out") { + throw new VmError( + `Host did not answer the VM action within ${timeoutMs}ms`, + { type: VmErrorType.HostCallTimeout }, + ); + } } /** @@ -157,6 +173,8 @@ export class WorkerToHost { constructor( private notifierBufferOrAdapter: SharedArrayBuffer | VmAdapter, private asyncRequests: VmActionRequest[] = [], + private processId = "", + private hostCallTimeoutMs: number = DEFAULT_HOST_CALL_TIMEOUT_MS, ) {} callActionSync(action: VmAction): Buffer { @@ -191,6 +209,7 @@ export class WorkerToHost { const actionMessage: VmActionExecuteMessage = { type: WorkerMessageType.VmActionExecute, action, + processId: this.processId, }; parentPort?.postMessage(actionMessage); @@ -198,6 +217,7 @@ export class WorkerToHost { waitForNotifierStateChange( notifierBufferi32, AtomicState.RequestResultLength, + this.hostCallTimeoutMs, ); const length = notifierBufferi32[1]; @@ -211,11 +231,16 @@ export class WorkerToHost { const message: VmActionResultBufferMessage = { buffer: valueBuffer, type: WorkerMessageType.VmActionResultBuffer, + processId: this.processId, }; parentPort?.postMessage(message); - waitForNotifierStateChange(notifierBufferi32, AtomicState.RequestResult); + waitForNotifierStateChange( + notifierBufferi32, + AtomicState.RequestResult, + this.hostCallTimeoutMs, + ); return Buffer.from(new Uint8Array(valueBuffer)); } diff --git a/libs/vm/src/worker.ts b/libs/vm/src/worker.ts index 1727e2f..ed5c353 100644 --- a/libs/vm/src/worker.ts +++ b/libs/vm/src/worker.ts @@ -1,4 +1,6 @@ import { isMainThread, parentPort } from "node:worker_threads"; +import { WasmModuleCache } from "./services/wasm-module-cache.js"; +import type { VmCallData } from "./types/vm-call-data.js"; import { type VmResultWorkerMessage, type WorkerMessage, @@ -6,6 +8,12 @@ import { } from "./types/worker-messages.js"; import { executeVm } from "./vm.js"; +const WASM_MODULE_CACHE_CAPACITY = 10; + +// A WasmModuleCache instance does not survive structured clone, so the host +// strips it from the posted callData and the worker maintains its own. +const wasmModuleCache = new WasmModuleCache(WASM_MODULE_CACHE_CAPACITY); + let isStarted = false; export function startWorker() { @@ -13,24 +21,46 @@ export function startWorker() { isStarted = true; parentPort?.on("message", async (event) => { + const message: WorkerMessage = event; + + if (message.type !== WorkerMessageType.VmCall) return; + try { - const message: WorkerMessage = event; - - if (message.type === WorkerMessageType.VmCall) { - const result = await executeVm( - message.callData, - message.processId, - message.notifierBuffer, - ); - const response: VmResultWorkerMessage = { - result, - type: WorkerMessageType.VmResult, - }; - - parentPort?.postMessage(response); - } + const callData: VmCallData = { + ...message.callData, + wasmModuleCache: message.callData.cache?.id + ? wasmModuleCache + : undefined, + }; + + const result = await executeVm( + callData, + message.processId, + message.notifierBuffer, + ); + const response: VmResultWorkerMessage = { + result, + processId: message.processId, + type: WorkerMessageType.VmResult, + }; + + parentPort?.postMessage(response); } catch (error) { console.error("@worker:message, error thrown: ", error); + + // Always answer, otherwise a persistent host waits forever on this call + const response: VmResultWorkerMessage = { + result: { + exitCode: 1, + stderr: `[${message.processId}] - Worker failed to execute VM call: ${error}`, + stdout: "", + gasUsed: 0n, + }, + processId: message.processId, + type: WorkerMessageType.VmResult, + }; + + parentPort?.postMessage(response); } }); }