diff --git a/docs/runtime/workers.mdx b/docs/runtime/workers.mdx index 7b1b6734574f..c85df928ad1f 100644 --- a/docs/runtime/workers.mdx +++ b/docs/runtime/workers.mdx @@ -310,3 +310,19 @@ if (Bun.isMainThread) { console.log("I'm in a worker"); } ``` + +## `Worker.data` + +Pass a `data` option to `new Worker()` to send a value to the worker at startup. Inside the worker it is available as `Worker.data`, and also as `require("node:worker_threads").workerData`. + +```ts +// main.ts +const worker = new Worker("./worker.ts", { + data: { greeting: "hello" }, +}); + +// worker.ts +console.log(Worker.data); // => { greeting: "hello" } +``` + +The value is cloned with the HTML structured clone algorithm. Use `transferList` to transfer `MessagePort`, `ArrayBuffer`, and similar values instead of copying them. diff --git a/packages/bun-types/bun.d.ts b/packages/bun-types/bun.d.ts index 46096c814092..6c246878462e 100644 --- a/packages/bun-types/bun.d.ts +++ b/packages/bun-types/bun.d.ts @@ -431,6 +431,29 @@ declare module "bun" { * Equivalent to passing the `--preload` CLI argument, but only for this Worker. */ preload?: string[] | string | undefined; + + /** + * Any JavaScript value that is cloned and made available inside the worker + * as `Worker.data` and `require("node:worker_threads").workerData`. + * + * The cloning uses the HTML structured clone algorithm; use + * `transferList` to transfer values such as `MessagePort` instead of + * copying them. + */ + data?: any; + + /** + * Alias for {@link data}. If both are provided, `workerData` takes + * precedence. + */ + workerData?: any; + + /** + * Transferable values (such as `MessagePort` or `ArrayBuffer`) referenced + * from `data`/`workerData` that should be moved to the worker instead of + * cloned. + */ + transferList?: readonly import("node:worker_threads").TransferListItem[] | undefined; } interface Worker extends EventTarget, AbstractWorker { diff --git a/src/js/node/worker_threads.ts b/src/js/node/worker_threads.ts index 921df1a621d2..9da4630c3a8f 100644 --- a/src/js/node/worker_threads.ts +++ b/src/js/node/worker_threads.ts @@ -81,6 +81,7 @@ const { 8: _markAsUncloneable, 9: _setEntryEvaluatedHook, 10: _isNodeWorker, + 11: _setWorkerData, } = $cpp("Worker.cpp", "createNodeWorkerThreadsBinding") as [ unknown, number, @@ -93,6 +94,7 @@ const { (value: unknown) => void, (hook: () => void) => void, boolean, + (value: unknown) => void, ]; type NodeWorkerOptions = import("node:worker_threads").WorkerOptions; @@ -744,6 +746,8 @@ if ( if (stdioPorts) setupWorkerStdio(stdioPorts); if (controlPort) messaging.setupMainThreadPort(controlPort, _setEntryEvaluatedHook); } +// Keep the native Worker.data cache in sync with the unpacked/unwrapped value above. +if (workerData !== _workerData) _setWorkerData(workerData); function receiveMessageOnPort(port: MessagePort) { let res = _receiveMessageOnPort(port); if (!res) return undefined; diff --git a/src/jsc/bindings/ZigGlobalObject.cpp b/src/jsc/bindings/ZigGlobalObject.cpp index a45cf0016dfe..556558d17953 100644 --- a/src/jsc/bindings/ZigGlobalObject.cpp +++ b/src/jsc/bindings/ZigGlobalObject.cpp @@ -4140,6 +4140,7 @@ void GlobalObject::adoptNapiEnvsForTestIsolation(GlobalObject* oldGlobal) } void GlobalObject::setNodeWorkerEnvironmentData(JSMap* data) { m_nodeWorkerEnvironmentData.set(vm(), this, data); } +void GlobalObject::setNodeWorkerData(JSValue data) { m_nodeWorkerData.set(vm(), this, data); } void GlobalObject::setNodeWorkerEntryEvaluatedHook(JSObject* hook) { if (hook) diff --git a/src/jsc/bindings/ZigGlobalObject.h b/src/jsc/bindings/ZigGlobalObject.h index ca759a74da9e..fc8ae40cbd2b 100644 --- a/src/jsc/bindings/ZigGlobalObject.h +++ b/src/jsc/bindings/ZigGlobalObject.h @@ -511,6 +511,8 @@ class GlobalObject : public Bun::GlobalScope { /* Supports getEnvironmentData() and setEnvironmentData(), and is cloned into newly-created */ \ /* Workers. Initialized in createNodeWorkerThreadsBinding. */ \ V(private, WriteBarrier, m_nodeWorkerEnvironmentData) \ + /* workerData cached by createNodeWorkerThreadsBinding; backs the Worker.data getter. */ \ + V(private, WriteBarrier, m_nodeWorkerData) \ /* setupMainThreadPort's drain callback; run once by WebWorker__dispatchOnline */ \ /* after entry-module evaluation. Stored here (not on globalThis) so user code can't clobber it. */ \ V(private, WriteBarrier, m_nodeWorkerEntryEvaluatedHook) \ @@ -748,6 +750,8 @@ class GlobalObject : public Bun::GlobalScope { JSMap* nodeWorkerEnvironmentData() { return m_nodeWorkerEnvironmentData.get(); } void setNodeWorkerEnvironmentData(JSMap* data); + JSValue nodeWorkerData() { return m_nodeWorkerData.get(); } + void setNodeWorkerData(JSValue data); JSObject* nodeWorkerEntryEvaluatedHook() { return m_nodeWorkerEntryEvaluatedHook.get(); } void setNodeWorkerEntryEvaluatedHook(JSObject* hook); diff --git a/src/jsc/bindings/webcore/JSWorker.cpp b/src/jsc/bindings/webcore/JSWorker.cpp index 0e94c9adb507..49f66e4e97cf 100644 --- a/src/jsc/bindings/webcore/JSWorker.cpp +++ b/src/jsc/bindings/webcore/JSWorker.cpp @@ -406,6 +406,19 @@ template<> JSValue JSWorkerDOMConstructor::prototypeForStructure(JSC::VM& vm, co return JSEventTarget::getConstructor(vm, &globalObject); } +JSC_DEFINE_CUSTOM_GETTER(jsWorkerConstructor_data, (JSGlobalObject * lexicalGlobalObject, JSC::EncodedJSValue, PropertyName)) +{ + auto& vm = JSC::getVM(lexicalGlobalObject); + auto throwScope = DECLARE_THROW_SCOPE(vm); + auto* globalObject = defaultGlobalObject(lexicalGlobalObject); + // nodeWorkerData is seeded by createNodeWorkerThreadsBinding; force that once if node:worker_threads hasn't run. + if (!globalObject->nodeWorkerEnvironmentData()) { + createNodeWorkerThreadsBinding(globalObject); + RETURN_IF_EXCEPTION(throwScope, {}); + } + RELEASE_AND_RETURN(throwScope, JSValue::encode(globalObject->nodeWorkerData())); +} + template<> void JSWorkerDOMConstructor::initializeProperties(VM& vm, JSDOMGlobalObject& globalObject) { putDirect(vm, vm.propertyNames->length, jsNumber(1), JSC::PropertyAttribute::ReadOnly | JSC::PropertyAttribute::DontEnum); @@ -413,6 +426,7 @@ template<> void JSWorkerDOMConstructor::initializeProperties(VM& vm, JSDOMGlobal m_originalName.set(vm, this, nameString); putDirect(vm, vm.propertyNames->name, nameString, JSC::PropertyAttribute::ReadOnly | JSC::PropertyAttribute::DontEnum); putDirect(vm, vm.propertyNames->prototype, JSWorker::prototype(vm, globalObject), JSC::PropertyAttribute::ReadOnly | JSC::PropertyAttribute::DontEnum | JSC::PropertyAttribute::DontDelete); + putDirectCustomAccessor(vm, Identifier::fromString(vm, "data"_s), JSC::CustomGetterSetter::create(vm, jsWorkerConstructor_data, nullptr), JSC::PropertyAttribute::ReadOnly | JSC::PropertyAttribute::DontEnum | JSC::PropertyAttribute::CustomAccessor); } JSC_DEFINE_CUSTOM_GETTER(jsWorker_threadIdGetter, (JSGlobalObject * lexicalGlobalObject, JSC::EncodedJSValue thisValue, PropertyName)) diff --git a/src/jsc/bindings/webcore/Worker.cpp b/src/jsc/bindings/webcore/Worker.cpp index 1a97f67c2571..b25b5652fab1 100644 --- a/src/jsc/bindings/webcore/Worker.cpp +++ b/src/jsc/bindings/webcore/Worker.cpp @@ -825,6 +825,12 @@ JSC_DEFINE_HOST_FUNCTION(jsFunctionSetEntryEvaluatedHook, (JSC::JSGlobalObject * return JSC::JSValue::encode(jsUndefined()); } +JSC_DEFINE_HOST_FUNCTION(jsFunctionSetWorkerData, (JSC::JSGlobalObject * lexicalGlobalObject, JSC::CallFrame* callFrame)) +{ + defaultGlobalObject(lexicalGlobalObject)->setNodeWorkerData(callFrame->argument(0)); + return JSC::JSValue::encode(jsUndefined()); +} + JSValue createNodeWorkerThreadsBinding(Zig::GlobalObject* globalObject) { VM& vm = globalObject->vm(); @@ -833,7 +839,10 @@ JSValue createNodeWorkerThreadsBinding(Zig::GlobalObject* globalObject) JSValue workerData = jsNull(); JSValue threadId = jsNumber(0); JSValue threadName = jsEmptyString(vm); - JSMap* environmentData = nullptr; + // Re-entrant (Worker.data getter + node:worker_threads both call this): reuse the first run's cache. + JSMap* environmentData = globalObject->nodeWorkerEnvironmentData(); + if (JSValue cached = globalObject->nodeWorkerData()) + workerData = cached; if (auto* worker = WebWorker__getParentWorker(globalObject->bunVM())) { auto& options = worker->options(); @@ -877,12 +886,13 @@ JSValue createNodeWorkerThreadsBinding(Zig::GlobalObject* globalObject) } ASSERT(environmentData); globalObject->setNodeWorkerEnvironmentData(environmentData); + globalObject->setNodeWorkerData(workerData); bool isNodeWorker = false; if (auto* worker = WebWorker__getParentWorker(globalObject->bunVM())) isNodeWorker = worker->options().kind == WorkerOptions::Kind::Node; - JSObject* array = constructEmptyArray(globalObject, nullptr, 11); + JSObject* array = constructEmptyArray(globalObject, nullptr, 12); RETURN_IF_EXCEPTION(scope, {}); array->putDirectIndex(globalObject, 0, workerData); array->putDirectIndex(globalObject, 1, threadId); @@ -895,6 +905,7 @@ JSValue createNodeWorkerThreadsBinding(Zig::GlobalObject* globalObject) array->putDirectIndex(globalObject, 8, JSFunction::create(vm, globalObject, 1, "markAsUncloneable"_s, jsFunctionMarkAsUncloneable, ImplementationVisibility::Public, NoIntrinsic)); array->putDirectIndex(globalObject, 9, JSFunction::create(vm, globalObject, 1, "setEntryEvaluatedHook"_s, jsFunctionSetEntryEvaluatedHook, ImplementationVisibility::Public, NoIntrinsic)); array->putDirectIndex(globalObject, 10, jsBoolean(isNodeWorker)); + array->putDirectIndex(globalObject, 11, JSFunction::create(vm, globalObject, 1, "setWorkerData"_s, jsFunctionSetWorkerData, ImplementationVisibility::Public, NoIntrinsic)); return array; } diff --git a/test/js/web/workers/worker.test.ts b/test/js/web/workers/worker.test.ts index 9e20bd0d8b4d..0ba4cceeef9e 100644 --- a/test/js/web/workers/worker.test.ts +++ b/test/js/web/workers/worker.test.ts @@ -458,4 +458,110 @@ describe("worker_threads", () => { await p; expect(message).toEqual("hello"); }); + + // https://github.com/oven-sh/bun/issues/9330 + // Spawned so that the worker body can observe `Worker.data` before the + // `node:worker_threads` module has been imported (the deserialization is + // shared between the two, so ordering matters). + describe("Worker.data", () => { + test.concurrent("is the cloned value of the `data` option", async () => { + const body = ` + const before = Worker.data; + const { workerData } = require("node:worker_threads"); + postMessage({ before, after: Worker.data, workerData, + same: before === workerData && Worker.data === workerData });`; + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + `const w = new Worker(${JSON.stringify("data:text/javascript," + encodeURIComponent(body))}, { + data: { greeting: "hello" }, + }); + w.onerror = e => { console.error(e.message); process.exit(1); }; + w.onmessage = e => { console.log(JSON.stringify(e.data)); w.terminate(); };`, + ], + env: bunEnv, + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(JSON.parse(stdout)).toEqual({ + before: { greeting: "hello" }, + after: { greeting: "hello" }, + workerData: { greeting: "hello" }, + same: true, + }); + expect(exitCode).toBe(0); + }); + + test.concurrent("agrees with workerData when node:worker_threads is imported first", async () => { + const body = ` + const { workerData } = require("node:worker_threads"); + postMessage({ workerData, data: Worker.data, same: Worker.data === workerData });`; + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + `const w = new Worker(${JSON.stringify("data:text/javascript," + encodeURIComponent(body))}, { + data: { v: 1 }, + }); + w.onerror = e => { console.error(e.message); process.exit(1); }; + w.onmessage = e => { console.log(JSON.stringify(e.data)); w.terminate(); };`, + ], + env: bunEnv, + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(JSON.parse(stdout)).toEqual({ workerData: { v: 1 }, data: { v: 1 }, same: true }); + expect(exitCode).toBe(0); + }); + + test.concurrent("is the unwrapped workerData inside a node:worker_threads Worker", async () => { + // The node wt.Worker constructor wraps workerData to carry internal + // stdio/messaging ports; Worker.data must be the unwrapped user value, + // identical to `workerData`, not the transport wrapper. + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + `const { Worker } = require("node:worker_threads"); + const src = 'const wt = require("node:worker_threads");' + + 'console.log(JSON.stringify({ data: Worker.data, same: Worker.data === wt.workerData,' + + ' keys: typeof Worker.data === "object" ? Object.keys(Worker.data) : null }));'; + const w = new Worker(src, { eval: true, workerData: { greeting: "hello" } }); + w.on("error", e => { console.error(e.message); process.exit(1); }); + w.on("exit", code => process.exit(code));`, + ], + env: bunEnv, + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(JSON.parse(stdout)).toEqual({ data: { greeting: "hello" }, same: true, keys: ["greeting"] }); + expect(exitCode).toBe(0); + }); + + test.concurrent("mirrors workerData when no data is passed", async () => { + // node:worker_threads' workerData is null on the main thread but undefined + // inside a worker that received no workerData; Worker.data should match both. + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + `console.log(Worker.data === require("node:worker_threads").workerData ? "main ok" : "main mismatch"); + const body = 'postMessage(Worker.data === require("node:worker_threads").workerData ? "worker ok" : "worker mismatch")'; + const w = new Worker("data:text/javascript," + encodeURIComponent(body)); + w.onerror = e => { console.error(e.message); process.exit(1); }; + w.onmessage = e => { console.log(e.data); w.terminate(); };`, + ], + env: bunEnv, + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(stdout).toBe("main ok\nworker ok\n"); + expect(exitCode).toBe(0); + }); + }); });