diff --git a/src/bun.js/bindings/ProcessIdentifier.cpp b/src/bun.js/bindings/ProcessIdentifier.cpp index 03e73426430ca1..f87cbe3363f574 100644 --- a/src/bun.js/bindings/ProcessIdentifier.cpp +++ b/src/bun.js/bindings/ProcessIdentifier.cpp @@ -31,23 +31,12 @@ namespace WebCore { namespace Process { -static std::optional globalIdentifier; - -void setIdentifier(ProcessIdentifier processIdentifier) -{ - ASSERT(isUIThread()); - globalIdentifier = processIdentifier; -} +// Bun only has 1 process +static ProcessIdentifier globalIdentifier { 1 }; ProcessIdentifier identifier() { - static std::once_flag onceFlag; - std::call_once(onceFlag, [] { - if (!globalIdentifier) - globalIdentifier = ProcessIdentifier::generate(); - }); - - return *globalIdentifier; + return globalIdentifier; } } // namespace ProcessIdent diff --git a/src/bun.js/bindings/ProcessIdentifier.h b/src/bun.js/bindings/ProcessIdentifier.h index f96e26c5aee2d5..c290d2b794e7f0 100644 --- a/src/bun.js/bindings/ProcessIdentifier.h +++ b/src/bun.js/bindings/ProcessIdentifier.h @@ -34,7 +34,6 @@ using ProcessIdentifier = ObjectIdentifier; namespace Process { -WEBCORE_EXPORT void setIdentifier(ProcessIdentifier); WEBCORE_EXPORT ProcessIdentifier identifier(); } // namespace Process diff --git a/src/bun.js/bindings/ZigGlobalObject.cpp b/src/bun.js/bindings/ZigGlobalObject.cpp index 5e534e5e107168..b7a577dade9e0d 100644 --- a/src/bun.js/bindings/ZigGlobalObject.cpp +++ b/src/bun.js/bindings/ZigGlobalObject.cpp @@ -143,7 +143,6 @@ #include "Performance.h" #include "ProcessBindingConstants.h" #include "ProcessBindingTTYWrap.h" -#include "ProcessIdentifier.h" #include "ReadableStream.h" #include "SerializedScriptValue.h" #include "StructuredClone.h" diff --git a/src/bun.js/bindings/webcore/JSWorker.cpp b/src/bun.js/bindings/webcore/JSWorker.cpp index 0250768c45c695..d72a58a81c4818 100644 --- a/src/bun.js/bindings/webcore/JSWorker.cpp +++ b/src/bun.js/bindings/webcore/JSWorker.cpp @@ -138,13 +138,13 @@ template<> JSC::EncodedJSValue JSC_HOST_CALL_ATTRIBUTES JSWorkerDOMConstructor:: } RETURN_IF_EXCEPTION(throwScope, {}); EnsureStillAliveScope argument1 = callFrame->argument(1); + + WorkerOptions options {}; JSValue nodeWorkerObject {}; if (callFrame->argumentCount() == 3) { nodeWorkerObject = callFrame->argument(2); + options.kind = WorkerOptions::Kind::Node; } - RETURN_IF_EXCEPTION(throwScope, {}); - - auto options = WorkerOptions {}; JSValue workerData = jsUndefined(); Vector> transferList; diff --git a/src/bun.js/bindings/webcore/Worker.cpp b/src/bun.js/bindings/webcore/Worker.cpp index e08cb923d51256..3c2cfb998eac85 100644 --- a/src/bun.js/bindings/webcore/Worker.cpp +++ b/src/bun.js/bindings/webcore/Worker.cpp @@ -389,12 +389,10 @@ void Worker::fireEarlyMessages(Zig::GlobalObject* workerGlobalObject) } } -void Worker::dispatchError(WTF::String message) +void Worker::dispatchErrorWithMessage(WTF::String message) { - auto* ctx = scriptExecutionContext(); - if (!ctx) - return; + if (!ctx) return; ScriptExecutionContext::postTaskTo(ctx->identifier(), [protectedThis = Ref { *this }, message = message.isolatedCopy()](ScriptExecutionContext& context) -> void { ErrorEvent::Init init; @@ -404,6 +402,27 @@ void Worker::dispatchError(WTF::String message) protectedThis->dispatchEvent(event); }); } + +bool Worker::dispatchErrorWithValue(Zig::GlobalObject* workerGlobalObject, JSValue value) +{ + auto* ctx = scriptExecutionContext(); + if (!ctx) return false; + auto serialized = SerializedScriptValue::create(*workerGlobalObject, value, SerializationForStorage::No, SerializationErrorMode::NonThrowing); + if (!serialized) return false; + + ScriptExecutionContext::postTaskTo(ctx->identifier(), [protectedThis = Ref { *this }, serialized](ScriptExecutionContext& context) -> void { + auto* globalObject = context.globalObject(); + ErrorEvent::Init init; + JSValue deserialized = serialized->deserialize(*globalObject, globalObject, SerializationErrorMode::NonThrowing); + if (!deserialized) return; + init.error = deserialized; + + auto event = ErrorEvent::create(eventNames().errorEvent, init, EventIsTrusted::Yes); + protectedThis->dispatchEvent(event); + }); + return true; +} + void Worker::dispatchExit(int32_t exitCode) { auto* ctx = scriptExecutionContext(); @@ -483,7 +502,16 @@ extern "C" void WebWorker__dispatchError(Zig::GlobalObject* globalObject, Worker init.bubbles = false; globalObject->globalEventScope->dispatchEvent(ErrorEvent::create(eventNames().errorEvent, init, EventIsTrusted::Yes)); - worker->dispatchError(message.toWTFString(BunString::ZeroCopy)); + switch (worker->options().kind) { + case WorkerOptions::Kind::Web: + return worker->dispatchErrorWithMessage(message.toWTFString(BunString::ZeroCopy)); + case WorkerOptions::Kind::Node: + if (!worker->dispatchErrorWithValue(globalObject, error)) { + // If serialization threw an error, use the string instead + worker->dispatchErrorWithMessage(message.toWTFString(BunString::ZeroCopy)); + } + return; + } } extern "C" WebCore::Worker* WebWorker__getParentWorker(void* bunVM); diff --git a/src/bun.js/bindings/webcore/Worker.h b/src/bun.js/bindings/webcore/Worker.h index 1aec9a7c95417d..0b473d20295368 100644 --- a/src/bun.js/bindings/webcore/Worker.h +++ b/src/bun.js/bindings/webcore/Worker.h @@ -27,10 +27,7 @@ #include "ActiveDOMObject.h" #include "EventTarget.h" -// #include "MessagePort.h" #include "WorkerOptions.h" -// #include "WorkerScriptLoaderClient.h" -// #include "WorkerType.h" #include #include #include @@ -50,7 +47,6 @@ class RTCRtpScriptTransform; class RTCRtpScriptTransformer; class ScriptExecutionContext; class WorkerGlobalScopeProxy; -// class WorkerScriptLoader; struct StructuredSerializeOptions; struct WorkerOptions; @@ -80,15 +76,6 @@ class Worker final : public ThreadSafeRefCounted, public EventTargetWith void dispatchCloseEvent(Event&); void setKeepAlive(bool); -#if ENABLE(WEB_RTC) - void createRTCRtpScriptTransformer(RTCRtpScriptTransform&, MessageWithMessagePorts&&); -#endif - - // WorkerType type() const - // { - // return m_options.type; - // } - void postTaskToWorkerGlobalScope(Function&&); static void forEachWorker(const Function()>&); @@ -97,7 +84,9 @@ class Worker final : public ThreadSafeRefCounted, public EventTargetWith void dispatchOnline(Zig::GlobalObject* workerGlobalObject); // Fire a 'message' event in the Worker for messages that were sent before the Worker started running void fireEarlyMessages(Zig::GlobalObject* workerGlobalObject); - void dispatchError(WTF::String message); + void dispatchErrorWithMessage(WTF::String message); + // true if successful + bool dispatchErrorWithValue(Zig::GlobalObject* workerGlobalObject, JSValue value); void dispatchExit(int32_t exitCode); ScriptExecutionContext* scriptExecutionContext() const final { return ContextDestructionObserver::scriptExecutionContext(); } ScriptExecutionContextIdentifier clientIdentifier() const { return m_clientIdentifier; } @@ -111,16 +100,6 @@ class Worker final : public ThreadSafeRefCounted, public EventTargetWith void derefEventTarget() final { deref(); } void eventListenersDidChange() final {}; - // void didReceiveResponse(ResourceLoaderIdentifier, const ResourceResponse&) final; - // void notifyFinished() final; - - // ActiveDOMObject. - // void stop() final; - // void suspend(ReasonForSuspension) final; - // void resume() final; - // const char* activeDOMObjectName() const final; - // bool virtualHasPendingActivity() const final; - static void networkStateChanged(bool isOnLine); static constexpr uint8_t OnlineFlag = 1 << 0; @@ -128,15 +107,9 @@ class Worker final : public ThreadSafeRefCounted, public EventTargetWith static constexpr uint8_t TerminateRequestedFlag = 1 << 0; static constexpr uint8_t TerminatedFlag = 1 << 1; - // RefPtr m_scriptLoader; WorkerOptions m_options; String m_identifier; - // WorkerGlobalScopeProxy& m_contextProxy; // The proxy outlives the worker to perform thread shutdown. - // std::optional m_contentSecurityPolicyResponseHeaders; MonotonicTime m_workerCreationTime; - // bool m_shouldBypassMainWorldContentSecurityPolicy { false }; - // bool m_isSuspendedForBackForwardCache { false }; - // JSC::RuntimeFlags m_runtimeFlags; Deque> m_pendingEvents; Lock m_pendingTasksMutex; Deque> m_pendingTasks; diff --git a/src/bun.js/bindings/webcore/WorkerOptions.h b/src/bun.js/bindings/webcore/WorkerOptions.h index f01d1bf5704e9a..3feed8512c4efe 100644 --- a/src/bun.js/bindings/webcore/WorkerOptions.h +++ b/src/bun.js/bindings/webcore/WorkerOptions.h @@ -8,6 +8,13 @@ namespace WebCore { struct WorkerOptions { + enum class Kind : uint8_t { + // Created by the global Worker constructor + Web, + // Created by the `require("node:worker_threads").Worker` constructor + Node, + }; + String name; bool mini { false }; bool unref { false }; @@ -16,6 +23,7 @@ struct WorkerOptions { // true, then we need to make sure that `process.argv` contains "[worker eval]" instead of the // Blob URL. bool evalMode { false }; + Kind kind { Kind::Web }; // Serialized array containing [workerData, environmentData] // (environmentData is always a Map) RefPtr workerDataAndEnvironmentData; diff --git a/src/bun.js/web_worker.zig b/src/bun.js/web_worker.zig index 32736484993b49..b8fea6e31ba033 100644 --- a/src/bun.js/web_worker.zig +++ b/src/bun.js/web_worker.zig @@ -1,624 +1,623 @@ +//! Shared implementation of Web and Node `Worker` const bun = @import("bun"); -const JSC = bun.JSC; +const jsc = bun.jsc; const Output = bun.Output; const log = Output.scoped(.Worker, true); const std = @import("std"); -const JSValue = JSC.JSValue; +const JSValue = jsc.JSValue; const Async = bun.Async; const WTFStringImpl = @import("../string.zig").WTFStringImpl; +const WebWorker = @This(); + +/// null when haven't started yet +vm: ?*jsc.VirtualMachine = null, +status: std.atomic.Value(Status) = .init(.start), +/// To prevent UAF, the `spin` function (aka the worker's event loop) will call deinit once this is set and properly exit the loop. +requested_terminate: std.atomic.Value(bool) = .init(false), +execution_context_id: u32 = 0, +parent_context_id: u32 = 0, +parent: *jsc.VirtualMachine, + +/// To be resolved on the Worker thread at startup, in spin(). +unresolved_specifier: []const u8, +preloads: [][]const u8 = &.{}, +store_fd: bool = false, +arena: ?bun.MimallocArena = null, +name: [:0]const u8 = "Worker", +cpp_worker: *anyopaque, +mini: bool, +// Most of our code doesn't care whether `eval` was passed, because worker_threads.ts +// automatically passes a Blob URL instead of a file path if `eval` is true. But, if `eval` is +// true, then we need to make sure that `process.argv` contains "[worker eval]" instead of the +// Blob URL. +eval_mode: bool, + +/// `user_keep_alive` is the state of the user's .ref()/.unref() calls +/// if false, then the parent poll will always be unref, otherwise the worker's event loop will keep the poll alive. +user_keep_alive: bool = false, +worker_event_loop_running: bool = true, +parent_poll_ref: Async.KeepAlive = .{}, + +// kept alive by C++ Worker object +argv: []const WTFStringImpl, +execArgv: ?[]const WTFStringImpl, + +/// Used to distinguish between terminate() called by exit(), and terminate() called for other reasons +exit_called: bool = false, + +pub const Status = enum(u8) { + start, + starting, + running, + terminated, +}; -/// Shared implementation of Web and Node `Worker` -pub const WebWorker = struct { - /// null when haven't started yet - vm: ?*JSC.VirtualMachine = null, - status: std.atomic.Value(Status) = .init(.start), - /// To prevent UAF, the `spin` function (aka the worker's event loop) will call deinit once this is set and properly exit the loop. - requested_terminate: std.atomic.Value(bool) = .init(false), - execution_context_id: u32 = 0, - parent_context_id: u32 = 0, - parent: *JSC.VirtualMachine, - - /// To be resolved on the Worker thread at startup, in spin(). - unresolved_specifier: []const u8, - preloads: [][]const u8 = &.{}, - store_fd: bool = false, - arena: ?bun.MimallocArena = null, - name: [:0]const u8 = "Worker", - cpp_worker: *anyopaque, - mini: bool, - // Most of our code doesn't care whether `eval` was passed, because worker_threads.ts - // automatically passes a Blob URL instead of a file path if `eval` is true. But, if `eval` is - // true, then we need to make sure that `process.argv` contains "[worker eval]" instead of the - // Blob URL. - eval_mode: bool, - - /// `user_keep_alive` is the state of the user's .ref()/.unref() calls - /// if false, then the parent poll will always be unref, otherwise the worker's event loop will keep the poll alive. - user_keep_alive: bool = false, - worker_event_loop_running: bool = true, - parent_poll_ref: Async.KeepAlive = .{}, - - // kept alive by C++ Worker object - argv: []const WTFStringImpl, - execArgv: ?[]const WTFStringImpl, - - /// Used to distinguish between terminate() called by exit(), and terminate() called for other reasons - exit_called: bool = false, - - pub const Status = enum(u8) { - start, - starting, - running, - terminated, +extern fn WebWorker__dispatchExit(?*jsc.JSGlobalObject, *anyopaque, i32) void; +extern fn WebWorker__dispatchOnline(cpp_worker: *anyopaque, *jsc.JSGlobalObject) void; +extern fn WebWorker__fireEarlyMessages(cpp_worker: *anyopaque, *jsc.JSGlobalObject) void; +extern fn WebWorker__dispatchError(*jsc.JSGlobalObject, *anyopaque, bun.String, JSValue) void; + +export fn WebWorker__getParentWorker(vm: *jsc.VirtualMachine) ?*anyopaque { + const worker = vm.worker orelse return null; + return worker.cpp_worker; +} + +pub fn hasRequestedTerminate(this: *const WebWorker) bool { + return this.requested_terminate.load(.monotonic); +} + +pub fn setRequestedTerminate(this: *WebWorker) bool { + return this.requested_terminate.swap(true, .release); +} + +export fn WebWorker__updatePtr(worker: *WebWorker, ptr: *anyopaque) bool { + worker.cpp_worker = ptr; + + var thread = std.Thread.spawn( + .{ .stack_size = bun.default_thread_stack_size }, + startWithErrorHandling, + .{worker}, + ) catch { + worker.deinit(); + return false; }; + thread.detach(); + return true; +} + +fn resolveEntryPointSpecifier( + parent: *jsc.VirtualMachine, + str: []const u8, + error_message: *bun.String, + logger: *bun.logger.Log, +) ?[]const u8 { + if (parent.standalone_module_graph) |graph| { + if (graph.find(str) != null) { + return str; + } - extern fn WebWorker__dispatchExit(?*JSC.JSGlobalObject, *anyopaque, i32) void; - extern fn WebWorker__dispatchOnline(cpp_worker: *anyopaque, *JSC.JSGlobalObject) void; - extern fn WebWorker__fireEarlyMessages(cpp_worker: *anyopaque, *JSC.JSGlobalObject) void; - extern fn WebWorker__dispatchError(*JSC.JSGlobalObject, *anyopaque, bun.String, JSValue) void; - - export fn WebWorker__getParentWorker(vm: *JSC.VirtualMachine) ?*anyopaque { - const worker = vm.worker orelse return null; - return worker.cpp_worker; - } - - pub fn hasRequestedTerminate(this: *const WebWorker) bool { - return this.requested_terminate.load(.monotonic); - } - - pub fn setRequestedTerminate(this: *WebWorker) bool { - return this.requested_terminate.swap(true, .release); - } - - export fn WebWorker__updatePtr(worker: *WebWorker, ptr: *anyopaque) bool { - worker.cpp_worker = ptr; - - var thread = std.Thread.spawn( - .{ .stack_size = bun.default_thread_stack_size }, - startWithErrorHandling, - .{worker}, - ) catch { - worker.deinit(); - return false; - }; - thread.detach(); - return true; - } + // Since `bun build --compile` renames files to `.js` by + // default, we need to do the reverse of our file extension + // mapping. + // + // new Worker("./foo") -> new Worker("./foo.js") + // new Worker("./foo.ts") -> new Worker("./foo.js") + // new Worker("./foo.jsx") -> new Worker("./foo.js") + // new Worker("./foo.mjs") -> new Worker("./foo.js") + // new Worker("./foo.mts") -> new Worker("./foo.js") + // new Worker("./foo.cjs") -> new Worker("./foo.js") + // new Worker("./foo.cts") -> new Worker("./foo.js") + // new Worker("./foo.tsx") -> new Worker("./foo.js") + // + if (bun.strings.hasPrefixComptime(str, "./") or bun.strings.hasPrefixComptime(str, "../")) try_from_extension: { + var pathbuf: bun.PathBuffer = undefined; + var base = str; + + base = bun.path.joinAbsStringBuf(bun.StandaloneModuleGraph.base_public_path_with_default_suffix, &pathbuf, &.{str}, .loose); + const extname = std.fs.path.extension(base); + + // ./foo -> ./foo.js + if (extname.len == 0) { + pathbuf[base.len..][0..3].* = ".js".*; + if (graph.find(pathbuf[0 .. base.len + 3])) |js_file| { + return js_file.name; + } - fn resolveEntryPointSpecifier( - parent: *JSC.VirtualMachine, - str: []const u8, - error_message: *bun.String, - logger: *bun.logger.Log, - ) ?[]const u8 { - if (parent.standalone_module_graph) |graph| { - if (graph.find(str) != null) { - return str; + break :try_from_extension; } - // Since `bun build --compile` renames files to `.js` by - // default, we need to do the reverse of our file extension - // mapping. - // - // new Worker("./foo") -> new Worker("./foo.js") - // new Worker("./foo.ts") -> new Worker("./foo.js") - // new Worker("./foo.jsx") -> new Worker("./foo.js") - // new Worker("./foo.mjs") -> new Worker("./foo.js") - // new Worker("./foo.mts") -> new Worker("./foo.js") - // new Worker("./foo.cjs") -> new Worker("./foo.js") - // new Worker("./foo.cts") -> new Worker("./foo.js") - // new Worker("./foo.tsx") -> new Worker("./foo.js") - // - if (bun.strings.hasPrefixComptime(str, "./") or bun.strings.hasPrefixComptime(str, "../")) try_from_extension: { - var pathbuf: bun.PathBuffer = undefined; - var base = str; - - base = bun.path.joinAbsStringBuf(bun.StandaloneModuleGraph.base_public_path_with_default_suffix, &pathbuf, &.{str}, .loose); - const extname = std.fs.path.extension(base); - - // ./foo -> ./foo.js - if (extname.len == 0) { - pathbuf[base.len..][0..3].* = ".js".*; - if (graph.find(pathbuf[0 .. base.len + 3])) |js_file| { - return js_file.name; - } - - break :try_from_extension; + // ./foo.ts -> ./foo.js + if (bun.strings.eqlComptime(extname, ".ts")) { + pathbuf[base.len - 3 .. base.len][0..3].* = ".js".*; + if (graph.find(pathbuf[0..base.len])) |js_file| { + return js_file.name; } - // ./foo.ts -> ./foo.js - if (bun.strings.eqlComptime(extname, ".ts")) { - pathbuf[base.len - 3 .. base.len][0..3].* = ".js".*; - if (graph.find(pathbuf[0..base.len])) |js_file| { - return js_file.name; - } - - break :try_from_extension; - } + break :try_from_extension; + } - if (extname.len == 4) { - inline for (.{ ".tsx", ".jsx", ".mjs", ".mts", ".cts", ".cjs" }) |ext| { - if (bun.strings.eqlComptime(extname, ext)) { - pathbuf[base.len - ext.len ..][0..".js".len].* = ".js".*; - const as_js = pathbuf[0 .. base.len - ext.len + ".js".len]; - if (graph.find(as_js)) |js_file| { - return js_file.name; - } - break :try_from_extension; + if (extname.len == 4) { + inline for (.{ ".tsx", ".jsx", ".mjs", ".mts", ".cts", ".cjs" }) |ext| { + if (bun.strings.eqlComptime(extname, ext)) { + pathbuf[base.len - ext.len ..][0..".js".len].* = ".js".*; + const as_js = pathbuf[0 .. base.len - ext.len + ".js".len]; + if (graph.find(as_js)) |js_file| { + return js_file.name; } + break :try_from_extension; } } } } + } - if (JSC.WebCore.ObjectURLRegistry.isBlobURL(str)) { - if (JSC.WebCore.ObjectURLRegistry.singleton().has(str["blob:".len..])) { - return str; - } else { - error_message.* = bun.String.static("Blob URL is missing"); - return null; - } + if (bun.webcore.ObjectURLRegistry.isBlobURL(str)) { + if (bun.webcore.ObjectURLRegistry.singleton().has(str["blob:".len..])) { + return str; + } else { + error_message.* = bun.String.static("Blob URL is missing"); + return null; } + } - var resolved_entry_point: bun.resolver.Result = parent.transpiler.resolveEntryPoint(str) catch { - const out = (logger.toJS(parent.global, bun.default_allocator, "Error resolving Worker entry point") catch bun.outOfMemory()).toBunString(parent.global) catch { - error_message.* = bun.String.static("unexpected exception"); - return null; - }; - error_message.* = out; + var resolved_entry_point: bun.resolver.Result = parent.transpiler.resolveEntryPoint(str) catch { + const out = (logger.toJS(parent.global, bun.default_allocator, "Error resolving Worker entry point") catch bun.outOfMemory()).toBunString(parent.global) catch { + error_message.* = bun.String.static("unexpected exception"); return null; }; + error_message.* = out; + return null; + }; - const entry_path: *bun.fs.Path = resolved_entry_point.path() orelse { - error_message.* = bun.String.static("Worker entry point is missing"); - return null; - }; - return entry_path.text; - } + const entry_path: *bun.fs.Path = resolved_entry_point.path() orelse { + error_message.* = bun.String.static("Worker entry point is missing"); + return null; + }; + return entry_path.text; +} + +pub fn create( + cpp_worker: *void, + parent: *jsc.VirtualMachine, + name_str: bun.String, + specifier_str: bun.String, + error_message: *bun.String, + parent_context_id: u32, + this_context_id: u32, + mini: bool, + default_unref: bool, + eval_mode: bool, + argv_ptr: ?[*]WTFStringImpl, + argv_len: usize, + inherit_execArgv: bool, + execArgv_ptr: ?[*]WTFStringImpl, + execArgv_len: usize, + preload_modules_ptr: ?[*]bun.String, + preload_modules_len: usize, +) callconv(.c) ?*WebWorker { + jsc.markBinding(@src()); + log("[{d}] WebWorker.create", .{this_context_id}); + var spec_slice = specifier_str.toUTF8(bun.default_allocator); + defer spec_slice.deinit(); + const prev_log = parent.transpiler.log; + var temp_log = bun.logger.Log.init(bun.default_allocator); + parent.transpiler.setLog(&temp_log); + defer parent.transpiler.setLog(prev_log); + defer temp_log.deinit(); + + const preload_modules = if (preload_modules_ptr) |ptr| ptr[0..preload_modules_len] else &.{}; + + var preloads = std.ArrayList([]const u8).initCapacity(bun.default_allocator, preload_modules_len) catch bun.outOfMemory(); + for (preload_modules) |module| { + const utf8_slice = module.toUTF8(bun.default_allocator); + defer utf8_slice.deinit(); + if (resolveEntryPointSpecifier(parent, utf8_slice.slice(), error_message, &temp_log)) |preload| { + preloads.append(bun.default_allocator.dupe(u8, preload) catch bun.outOfMemory()) catch bun.outOfMemory(); + } - pub fn create( - cpp_worker: *void, - parent: *JSC.VirtualMachine, - name_str: bun.String, - specifier_str: bun.String, - error_message: *bun.String, - parent_context_id: u32, - this_context_id: u32, - mini: bool, - default_unref: bool, - eval_mode: bool, - argv_ptr: ?[*]WTFStringImpl, - argv_len: usize, - inherit_execArgv: bool, - execArgv_ptr: ?[*]WTFStringImpl, - execArgv_len: usize, - preload_modules_ptr: ?[*]bun.String, - preload_modules_len: usize, - ) callconv(.C) ?*WebWorker { - JSC.markBinding(@src()); - log("[{d}] WebWorker.create", .{this_context_id}); - var spec_slice = specifier_str.toUTF8(bun.default_allocator); - defer spec_slice.deinit(); - const prev_log = parent.transpiler.log; - var temp_log = bun.logger.Log.init(bun.default_allocator); - parent.transpiler.setLog(&temp_log); - defer parent.transpiler.setLog(prev_log); - defer temp_log.deinit(); - - const preload_modules = if (preload_modules_ptr) |ptr| ptr[0..preload_modules_len] else &.{}; - - var preloads = std.ArrayList([]const u8).initCapacity(bun.default_allocator, preload_modules_len) catch bun.outOfMemory(); - for (preload_modules) |module| { - const utf8_slice = module.toUTF8(bun.default_allocator); - defer utf8_slice.deinit(); - if (resolveEntryPointSpecifier(parent, utf8_slice.slice(), error_message, &temp_log)) |preload| { - preloads.append(bun.default_allocator.dupe(u8, preload) catch bun.outOfMemory()) catch bun.outOfMemory(); + if (!error_message.isEmpty()) { + for (preloads.items) |preload| { + bun.default_allocator.free(preload); } + preloads.deinit(); + return null; + } + } - if (!error_message.isEmpty()) { - for (preloads.items) |preload| { - bun.default_allocator.free(preload); - } - preloads.deinit(); - return null; + var worker = bun.default_allocator.create(WebWorker) catch bun.outOfMemory(); + worker.* = WebWorker{ + .cpp_worker = cpp_worker, + .parent = parent, + .parent_context_id = parent_context_id, + .execution_context_id = this_context_id, + .mini = mini, + .eval_mode = eval_mode, + .unresolved_specifier = (spec_slice.toOwned(bun.default_allocator) catch bun.outOfMemory()).slice(), + .store_fd = parent.transpiler.resolver.store_fd, + .name = brk: { + if (!name_str.isEmpty()) { + break :brk std.fmt.allocPrintZ(bun.default_allocator, "{}", .{name_str}) catch bun.outOfMemory(); } - } + break :brk ""; + }, + .user_keep_alive = !default_unref, + .worker_event_loop_running = true, + .argv = if (argv_ptr) |ptr| ptr[0..argv_len] else &.{}, + .execArgv = if (inherit_execArgv) null else (if (execArgv_ptr) |ptr| ptr[0..execArgv_len] else &.{}), + .preloads = preloads.items, + }; - var worker = bun.default_allocator.create(WebWorker) catch bun.outOfMemory(); - worker.* = WebWorker{ - .cpp_worker = cpp_worker, - .parent = parent, - .parent_context_id = parent_context_id, - .execution_context_id = this_context_id, - .mini = mini, - .eval_mode = eval_mode, - .unresolved_specifier = (spec_slice.toOwned(bun.default_allocator) catch bun.outOfMemory()).slice(), - .store_fd = parent.transpiler.resolver.store_fd, - .name = brk: { - if (!name_str.isEmpty()) { - break :brk std.fmt.allocPrintZ(bun.default_allocator, "{}", .{name_str}) catch bun.outOfMemory(); - } - break :brk ""; - }, - .user_keep_alive = !default_unref, - .worker_event_loop_running = true, - .argv = if (argv_ptr) |ptr| ptr[0..argv_len] else &.{}, - .execArgv = if (inherit_execArgv) null else (if (execArgv_ptr) |ptr| ptr[0..execArgv_len] else &.{}), - .preloads = preloads.items, - }; + worker.parent_poll_ref.ref(parent); - worker.parent_poll_ref.ref(parent); + return worker; +} - return worker; +pub fn startWithErrorHandling( + this: *WebWorker, +) void { + bun.Analytics.Features.workers_spawned += 1; + start(this) catch |err| { + Output.panic("An unhandled error occurred while starting a worker: {s}\n", .{@errorName(err)}); + }; +} + +pub fn start( + this: *WebWorker, +) anyerror!void { + if (this.name.len > 0) { + Output.Source.configureNamedThread(this.name); + } else { + Output.Source.configureNamedThread("Worker"); } - pub fn startWithErrorHandling( - this: *WebWorker, - ) void { - bun.Analytics.Features.workers_spawned += 1; - start(this) catch |err| { - Output.panic("An unhandled error occurred while starting a worker: {s}\n", .{@errorName(err)}); - }; + if (this.hasRequestedTerminate()) { + this.exitAndDeinit(); + return; } - pub fn start( - this: *WebWorker, - ) anyerror!void { - if (this.name.len > 0) { - Output.Source.configureNamedThread(this.name); - } else { - Output.Source.configureNamedThread("Worker"); - } - - if (this.hasRequestedTerminate()) { - this.exitAndDeinit(); - return; - } + assert(this.status.load(.acquire) == .start); + assert(this.vm == null); - assert(this.status.load(.acquire) == .start); - assert(this.vm == null); + var transform_options = this.parent.transpiler.options.transform_options; - var transform_options = this.parent.transpiler.options.transform_options; - - if (this.execArgv) |exec_argv| parse_new_args: { - var new_args: std.ArrayList([]const u8) = try .initCapacity(bun.default_allocator, exec_argv.len); - defer { - for (new_args.items) |arg| { - bun.default_allocator.free(arg); - } - new_args.deinit(); - } - - for (exec_argv) |arg| { - try new_args.append(arg.toOwnedSliceZ(bun.default_allocator)); + if (this.execArgv) |exec_argv| parse_new_args: { + var new_args: std.ArrayList([]const u8) = try .initCapacity(bun.default_allocator, exec_argv.len); + defer { + for (new_args.items) |arg| { + bun.default_allocator.free(arg); } + new_args.deinit(); + } - var diag: bun.clap.Diagnostic = .{}; - var iter: bun.clap.args.SliceIterator = .init(new_args.items); + for (exec_argv) |arg| { + try new_args.append(arg.toOwnedSliceZ(bun.default_allocator)); + } - var args = bun.clap.parseEx(bun.clap.Help, bun.CLI.Command.Tag.RunCommand.params(), &iter, .{ - .diagnostic = &diag, - .allocator = bun.default_allocator, + var diag: bun.clap.Diagnostic = .{}; + var iter: bun.clap.args.SliceIterator = .init(new_args.items); - // just one for executable - .stop_after_positional_at = 1, - }) catch { - // ignore param parsing errors - break :parse_new_args; - }; - defer args.deinit(); + var args = bun.clap.parseEx(bun.clap.Help, bun.CLI.Command.Tag.RunCommand.params(), &iter, .{ + .diagnostic = &diag, + .allocator = bun.default_allocator, - // override the existing even if it was set - transform_options.allow_addons = !args.flag("--no-addons"); + // just one for executable + .stop_after_positional_at = 1, + }) catch { + // ignore param parsing errors + break :parse_new_args; + }; + defer args.deinit(); - // TODO: currently this only checks for --no-addons. I think - // this should go through most flags and update the options. - } + // override the existing even if it was set + transform_options.allow_addons = !args.flag("--no-addons"); - this.arena = try bun.MimallocArena.init(); - var vm = try JSC.VirtualMachine.initWorker(this, .{ - .allocator = this.arena.?.allocator(), - .args = transform_options, - .store_fd = this.store_fd, - .graph = this.parent.standalone_module_graph, - }); - vm.allocator = this.arena.?.allocator(); - vm.arena = &this.arena.?; + // TODO: currently this only checks for --no-addons. I think + // this should go through most flags and update the options. + } - var b = &vm.transpiler; + this.arena = try bun.MimallocArena.init(); + var vm = try jsc.VirtualMachine.initWorker(this, .{ + .allocator = this.arena.?.allocator(), + .args = transform_options, + .store_fd = this.store_fd, + .graph = this.parent.standalone_module_graph, + }); + vm.allocator = this.arena.?.allocator(); + vm.arena = &this.arena.?; - b.configureDefines() catch { - this.flushLogs(); - this.exitAndDeinit(); - return; - }; + var b = &vm.transpiler; - // TODO: we may have to clone other parts of vm state. this will be more - // important when implementing vm.deinit() - const map = try vm.allocator.create(bun.DotEnv.Map); - map.* = try vm.transpiler.env.map.cloneWithAllocator(vm.allocator); + b.configureDefines() catch { + this.flushLogs(); + this.exitAndDeinit(); + return; + }; - const loader = try vm.allocator.create(bun.DotEnv.Loader); - loader.* = bun.DotEnv.Loader.init(map, vm.allocator); + // TODO: we may have to clone other parts of vm state. this will be more + // important when implementing vm.deinit() + const map = try vm.allocator.create(bun.DotEnv.Map); + map.* = try vm.transpiler.env.map.cloneWithAllocator(vm.allocator); - vm.transpiler.env = loader; + const loader = try vm.allocator.create(bun.DotEnv.Loader); + loader.* = bun.DotEnv.Loader.init(map, vm.allocator); - vm.loadExtraEnvAndSourceCodePrinter(); - vm.is_main_thread = false; - JSC.VirtualMachine.is_main_thread_vm = false; - vm.onUnhandledRejection = onUnhandledRejection; - const callback = JSC.OpaqueWrap(WebWorker, WebWorker.spin); + vm.transpiler.env = loader; - this.vm = vm; + vm.loadExtraEnvAndSourceCodePrinter(); + vm.is_main_thread = false; + jsc.VirtualMachine.is_main_thread_vm = false; + vm.onUnhandledRejection = onUnhandledRejection; + const callback = jsc.OpaqueWrap(WebWorker, WebWorker.spin); - vm.global.vm().holdAPILock(this, callback); - } + this.vm = vm; - /// Deinit will clean up vm and everything. - /// Early deinit may be called from caller thread, but full vm deinit will only be called within worker's thread. - fn deinit(this: *WebWorker) void { - log("[{d}] deinit", .{this.execution_context_id}); - this.parent_poll_ref.unrefConcurrently(this.parent); - bun.default_allocator.free(this.unresolved_specifier); - for (this.preloads) |preload| { - bun.default_allocator.free(preload); - } - bun.default_allocator.free(this.preloads); - bun.default_allocator.destroy(this); - } + vm.global.vm().holdAPILock(this, callback); +} - fn flushLogs(this: *WebWorker) void { - JSC.markBinding(@src()); - var vm = this.vm orelse return; - if (vm.log.msgs.items.len == 0) return; - const err = vm.log.toJS(vm.global, bun.default_allocator, "Error in worker") catch bun.outOfMemory(); - const str = err.toBunString(vm.global) catch @panic("unexpected exception"); - defer str.deref(); - WebWorker__dispatchError(vm.global, this.cpp_worker, str, err); +/// Deinit will clean up vm and everything. +/// Early deinit may be called from caller thread, but full vm deinit will only be called within worker's thread. +fn deinit(this: *WebWorker) void { + log("[{d}] deinit", .{this.execution_context_id}); + this.parent_poll_ref.unrefConcurrently(this.parent); + bun.default_allocator.free(this.unresolved_specifier); + for (this.preloads) |preload| { + bun.default_allocator.free(preload); } - - fn onUnhandledRejection(vm: *JSC.VirtualMachine, globalObject: *JSC.JSGlobalObject, error_instance_or_exception: JSC.JSValue) void { - // Prevent recursion - vm.onUnhandledRejection = &JSC.VirtualMachine.onQuietUnhandledRejectionHandlerCaptureValue; - - var error_instance = error_instance_or_exception.toError() orelse error_instance_or_exception; - - var array = bun.MutableString.init(bun.default_allocator, 0) catch unreachable; - defer array.deinit(); - - var buffered_writer_ = bun.MutableString.BufferedWriter{ .context = &array }; - var buffered_writer = &buffered_writer_; - var worker = vm.worker orelse @panic("Assertion failure: no worker"); - - const writer = buffered_writer.writer(); - const Writer = @TypeOf(writer); - // we buffer this because it'll almost always be < 4096 - // when it's under 4096, we want to avoid the dynamic allocation - bun.JSC.ConsoleObject.format2( - .Debug, - globalObject, - &[_]JSC.JSValue{error_instance}, - 1, - Writer, - Writer, - writer, - .{ - .enable_colors = false, - .add_newline = false, - .flush = false, - .max_depth = 32, - }, - ) catch |err| { - switch (err) { - error.JSError => {}, - error.OutOfMemory => globalObject.throwOutOfMemory() catch {}, - } - error_instance = globalObject.tryTakeException().?; - }; - buffered_writer.flush() catch { - bun.outOfMemory(); - }; - JSC.markBinding(@src()); - WebWorker__dispatchError(globalObject, worker.cpp_worker, bun.String.createUTF8(array.slice()), error_instance); - if (vm.worker) |worker_| { - _ = worker.setRequestedTerminate(); - worker.parent_poll_ref.unrefConcurrently(worker.parent); - worker_.exitAndDeinit(); + bun.default_allocator.free(this.preloads); + bun.default_allocator.destroy(this); +} + +fn flushLogs(this: *WebWorker) void { + jsc.markBinding(@src()); + var vm = this.vm orelse return; + if (vm.log.msgs.items.len == 0) return; + const err = vm.log.toJS(vm.global, bun.default_allocator, "Error in worker") catch bun.outOfMemory(); + const str = err.toBunString(vm.global) catch @panic("unexpected exception"); + defer str.deref(); + WebWorker__dispatchError(vm.global, this.cpp_worker, str, err); +} + +fn onUnhandledRejection(vm: *jsc.VirtualMachine, globalObject: *jsc.JSGlobalObject, error_instance_or_exception: jsc.JSValue) void { + // Prevent recursion + vm.onUnhandledRejection = &jsc.VirtualMachine.onQuietUnhandledRejectionHandlerCaptureValue; + + var error_instance = error_instance_or_exception.toError() orelse error_instance_or_exception; + + var array = bun.MutableString.init(bun.default_allocator, 0) catch unreachable; + defer array.deinit(); + + var buffered_writer_ = bun.MutableString.BufferedWriter{ .context = &array }; + var buffered_writer = &buffered_writer_; + var worker = vm.worker orelse @panic("Assertion failure: no worker"); + + const writer = buffered_writer.writer(); + const Writer = @TypeOf(writer); + // we buffer this because it'll almost always be < 4096 + // when it's under 4096, we want to avoid the dynamic allocation + jsc.ConsoleObject.format2( + .Debug, + globalObject, + &[_]jsc.JSValue{error_instance}, + 1, + Writer, + Writer, + writer, + .{ + .enable_colors = false, + .add_newline = false, + .flush = false, + .max_depth = 32, + }, + ) catch |err| { + switch (err) { + error.JSError => {}, + error.OutOfMemory => globalObject.throwOutOfMemory() catch {}, } + error_instance = globalObject.tryTakeException().?; + }; + buffered_writer.flush() catch { + bun.outOfMemory(); + }; + jsc.markBinding(@src()); + WebWorker__dispatchError(globalObject, worker.cpp_worker, bun.String.createUTF8(array.slice()), error_instance); + if (vm.worker) |worker_| { + _ = worker.setRequestedTerminate(); + worker.parent_poll_ref.unrefConcurrently(worker.parent); + worker_.exitAndDeinit(); } +} + +fn setStatus(this: *WebWorker, status: Status) void { + log("[{d}] status: {s}", .{ this.execution_context_id, @tagName(status) }); + + this.status.store(status, .release); +} + +fn unhandledError(this: *WebWorker, _: anyerror) void { + this.flushLogs(); +} + +fn spin(this: *WebWorker) void { + log("[{d}] spin start", .{this.execution_context_id}); + + var vm = this.vm.?; + assert(this.status.load(.acquire) == .start); + this.setStatus(.starting); + vm.preload = this.preloads; + // resolve entrypoint + var resolve_error = bun.String.empty; + defer resolve_error.deref(); + const path = resolveEntryPointSpecifier(vm, this.unresolved_specifier, &resolve_error, vm.log) orelse { + vm.exit_handler.exit_code = 1; + if (vm.log.errors == 0 and !resolve_error.isEmpty()) { + const err = resolve_error.toUTF8(bun.default_allocator); + defer err.deinit(); + vm.log.addError(null, .Empty, err.slice()) catch bun.outOfMemory(); + } + this.flushLogs(); + this.exitAndDeinit(); + return; + }; + defer bun.default_allocator.free(path); - fn setStatus(this: *WebWorker, status: Status) void { - log("[{d}] status: {s}", .{ this.execution_context_id, @tagName(status) }); - - this.status.store(status, .release); + // If the worker is terminated before we even try to run any code, the exit code should be 0 + if (this.hasRequestedTerminate()) { + this.flushLogs(); + this.exitAndDeinit(); + return; } - fn unhandledError(this: *WebWorker, _: anyerror) void { + var promise = vm.loadEntryPointForWebWorker(path) catch { + // If we called process.exit(), don't override the exit code + if (!this.exit_called) vm.exit_handler.exit_code = 1; this.flushLogs(); - } + this.exitAndDeinit(); + return; + }; - fn spin(this: *WebWorker) void { - log("[{d}] spin start", .{this.execution_context_id}); - - var vm = this.vm.?; - assert(this.status.load(.acquire) == .start); - this.setStatus(.starting); - vm.preload = this.preloads; - // resolve entrypoint - var resolve_error = bun.String.empty; - defer resolve_error.deref(); - const path = resolveEntryPointSpecifier(vm, this.unresolved_specifier, &resolve_error, vm.log) orelse { - vm.exit_handler.exit_code = 1; - if (vm.log.errors == 0 and !resolve_error.isEmpty()) { - const err = resolve_error.toUTF8(bun.default_allocator); - defer err.deinit(); - vm.log.addError(null, .Empty, err.slice()) catch bun.outOfMemory(); - } - this.flushLogs(); - this.exitAndDeinit(); - return; - }; - defer bun.default_allocator.free(path); + if (promise.status(vm.global.vm()) == .rejected) { + const handled = vm.uncaughtException(vm.global, promise.result(vm.global.vm()), true); - // If the worker is terminated before we even try to run any code, the exit code should be 0 - if (this.hasRequestedTerminate()) { - this.flushLogs(); + if (!handled) { + vm.exit_handler.exit_code = 1; this.exitAndDeinit(); return; } + } else { + _ = promise.result(vm.global.vm()); + } - var promise = vm.loadEntryPointForWebWorker(path) catch { - // If we called process.exit(), don't override the exit code - if (!this.exit_called) vm.exit_handler.exit_code = 1; - this.flushLogs(); - this.exitAndDeinit(); - return; - }; - - if (promise.status(vm.global.vm()) == .rejected) { - const handled = vm.uncaughtException(vm.global, promise.result(vm.global.vm()), true); - - if (!handled) { - vm.exit_handler.exit_code = 1; - this.exitAndDeinit(); - return; - } - } else { - _ = promise.result(vm.global.vm()); - } + this.flushLogs(); + log("[{d}] event loop start", .{this.execution_context_id}); + // TODO(@190n) call dispatchOnline earlier (basically as soon as spin() starts, before + // we start running JS) + WebWorker__dispatchOnline(this.cpp_worker, vm.global); + WebWorker__fireEarlyMessages(this.cpp_worker, vm.global); + this.setStatus(.running); + + // don't run the GC if we don't actually need to + if (vm.isEventLoopAlive() or + vm.eventLoop().tickConcurrentWithCount() > 0) + { + vm.global.vm().releaseWeakRefs(); + _ = vm.arena.gc(); + _ = vm.global.vm().runGC(false); + } - this.flushLogs(); - log("[{d}] event loop start", .{this.execution_context_id}); - // TODO(@190n) call dispatchOnline earlier (basically as soon as spin() starts, before - // we start running JS) - WebWorker__dispatchOnline(this.cpp_worker, vm.global); - WebWorker__fireEarlyMessages(this.cpp_worker, vm.global); - this.setStatus(.running); - - // don't run the GC if we don't actually need to - if (vm.isEventLoopAlive() or - vm.eventLoop().tickConcurrentWithCount() > 0) - { - vm.global.vm().releaseWeakRefs(); - _ = vm.arena.gc(); - _ = vm.global.vm().runGC(false); - } + // always doing a first tick so we call CppTask without delay after dispatchOnline + vm.tick(); - // always doing a first tick so we call CppTask without delay after dispatchOnline + while (vm.isEventLoopAlive()) { vm.tick(); + if (this.hasRequestedTerminate()) break; + vm.eventLoop().autoTickActive(); + if (this.hasRequestedTerminate()) break; + } - while (vm.isEventLoopAlive()) { - vm.tick(); - if (this.hasRequestedTerminate()) break; - vm.eventLoop().autoTickActive(); - if (this.hasRequestedTerminate()) break; - } + log("[{d}] before exit {s}", .{ this.execution_context_id, if (this.hasRequestedTerminate()) "(terminated)" else "(event loop dead)" }); - log("[{d}] before exit {s}", .{ this.execution_context_id, if (this.hasRequestedTerminate()) "(terminated)" else "(event loop dead)" }); + // Only call "beforeExit" if we weren't from a .terminate + if (!this.hasRequestedTerminate()) { + // TODO: is this able to allow the event loop to continue? + vm.onBeforeExit(); + } - // Only call "beforeExit" if we weren't from a .terminate - if (!this.hasRequestedTerminate()) { - // TODO: is this able to allow the event loop to continue? - vm.onBeforeExit(); - } + this.flushLogs(); + this.exitAndDeinit(); + log("[{d}] spin done", .{this.execution_context_id}); +} - this.flushLogs(); - this.exitAndDeinit(); - log("[{d}] spin done", .{this.execution_context_id}); +/// This is worker.ref()/.unref() from JS (Caller thread) +pub fn setRef(this: *WebWorker, value: bool) callconv(.c) void { + if (this.hasRequestedTerminate()) { + return; } - /// This is worker.ref()/.unref() from JS (Caller thread) - pub fn setRef(this: *WebWorker, value: bool) callconv(.C) void { - if (this.hasRequestedTerminate()) { - return; - } + this.setRefInternal(value); +} - this.setRefInternal(value); +pub fn setRefInternal(this: *WebWorker, value: bool) void { + if (value) { + this.parent_poll_ref.ref(this.parent); + } else { + this.parent_poll_ref.unref(this.parent); } - - pub fn setRefInternal(this: *WebWorker, value: bool) void { - if (value) { - this.parent_poll_ref.ref(this.parent); - } else { - this.parent_poll_ref.unref(this.parent); - } +} + +/// Implement process.exit(). May only be called from the Worker thread. +pub fn exit(this: *WebWorker) void { + this.exit_called = true; + this.notifyNeedTermination(); +} + +/// Request a terminate from any thread. +pub fn notifyNeedTermination(this: *WebWorker) callconv(.c) void { + if (this.status.load(.acquire) == .terminated) { + return; } - - /// Implement process.exit(). May only be called from the Worker thread. - pub fn exit(this: *WebWorker) void { - this.exit_called = true; - this.notifyNeedTermination(); + if (this.setRequestedTerminate()) { + return; } + log("[{d}] notifyNeedTermination", .{this.execution_context_id}); - /// Request a terminate from any thread. - pub fn notifyNeedTermination(this: *WebWorker) callconv(.C) void { - if (this.status.load(.acquire) == .terminated) { - return; - } - if (this.setRequestedTerminate()) { - return; - } - log("[{d}] notifyNeedTermination", .{this.execution_context_id}); - - if (this.vm) |vm| { - vm.eventLoop().wakeup(); - // TODO(@190n) notifyNeedTermination - } - - // TODO(@190n) delete - this.setRefInternal(false); + if (this.vm) |vm| { + vm.eventLoop().wakeup(); + // TODO(@190n) notifyNeedTermination } - /// This handles cleanup, emitting the "close" event, and deinit. - /// Only call after the VM is initialized AND on the same thread as the worker. - /// Otherwise, call `notifyNeedTermination` to cause the event loop to safely terminate. - pub fn exitAndDeinit(this: *WebWorker) noreturn { - JSC.markBinding(@src()); - this.setStatus(.terminated); - bun.Analytics.Features.workers_terminated += 1; - - log("[{d}] exitAndDeinit", .{this.execution_context_id}); - const cpp_worker = this.cpp_worker; - var exit_code: i32 = 0; - var globalObject: ?*JSC.JSGlobalObject = null; - var vm_to_deinit: ?*JSC.VirtualMachine = null; - var loop: ?*bun.uws.Loop = null; - if (this.vm) |vm| { - loop = vm.uwsLoop(); - this.vm = null; - vm.is_shutting_down = true; - vm.onExit(); - exit_code = vm.exit_handler.exit_code; - globalObject = vm.global; - vm_to_deinit = vm; - } - var arena = this.arena; - - WebWorker__dispatchExit(globalObject, cpp_worker, exit_code); - if (loop) |loop_| { - loop_.internal_loop_data.jsc_vm = null; - } + // TODO(@190n) delete + this.setRefInternal(false); +} + +/// This handles cleanup, emitting the "close" event, and deinit. +/// Only call after the VM is initialized AND on the same thread as the worker. +/// Otherwise, call `notifyNeedTermination` to cause the event loop to safely terminate. +pub fn exitAndDeinit(this: *WebWorker) noreturn { + jsc.markBinding(@src()); + this.setStatus(.terminated); + bun.Analytics.Features.workers_terminated += 1; + + log("[{d}] exitAndDeinit", .{this.execution_context_id}); + const cpp_worker = this.cpp_worker; + var exit_code: i32 = 0; + var globalObject: ?*jsc.JSGlobalObject = null; + var vm_to_deinit: ?*jsc.VirtualMachine = null; + var loop: ?*bun.uws.Loop = null; + if (this.vm) |vm| { + loop = vm.uwsLoop(); + this.vm = null; + vm.is_shutting_down = true; + vm.onExit(); + exit_code = vm.exit_handler.exit_code; + globalObject = vm.global; + vm_to_deinit = vm; + } + var arena = this.arena; - bun.uws.onThreadExit(); - this.deinit(); + WebWorker__dispatchExit(globalObject, cpp_worker, exit_code); + if (loop) |loop_| { + loop_.internal_loop_data.jsc_vm = null; + } - if (vm_to_deinit) |vm| { - vm.deinit(); // NOTE: deinit here isn't implemented, so freeing workers will leak the vm. - } - bun.deleteAllPoolsForThreadExit(); - if (arena) |*arena_| { - arena_.deinit(); - } + bun.uws.onThreadExit(); + this.deinit(); - bun.exitThread(); + if (vm_to_deinit) |vm| { + vm.deinit(); // NOTE: deinit here isn't implemented, so freeing workers will leak the vm. } - - comptime { - @export(&create, .{ .name = "WebWorker__create" }); - @export(¬ifyNeedTermination, .{ .name = "WebWorker__notifyNeedTermination" }); - @export(&setRef, .{ .name = "WebWorker__setRef" }); - _ = WebWorker__updatePtr; + bun.deleteAllPoolsForThreadExit(); + if (arena) |*arena_| { + arena_.deinit(); } -}; + + bun.exitThread(); +} + +comptime { + @export(&create, .{ .name = "WebWorker__create" }); + @export(¬ifyNeedTermination, .{ .name = "WebWorker__notifyNeedTermination" }); + @export(&setRef, .{ .name = "WebWorker__setRef" }); + _ = WebWorker__updatePtr; +} const assert = bun.assert; diff --git a/src/bun.js/webcore.zig b/src/bun.js/webcore.zig index 2bb0b85d0d40c5..930957ef5a06bc 100644 --- a/src/bun.js/webcore.zig +++ b/src/bun.js/webcore.zig @@ -14,7 +14,7 @@ pub const ByteListPool = bun.ObjectPool(bun.ByteList, null, true, 8); pub const Crypto = @import("webcore/Crypto.zig"); pub const AbortSignal = @import("bindings/AbortSignal.zig").AbortSignal; -pub const WebWorker = @import("web_worker.zig").WebWorker; +pub const WebWorker = @import("web_worker.zig"); pub const AutoFlusher = @import("webcore/AutoFlusher.zig"); pub const EncodingLabel = @import("webcore/EncodingLabel.zig").EncodingLabel; pub const Fetch = @import("webcore/fetch.zig"); diff --git a/src/js/node/worker_threads.ts b/src/js/node/worker_threads.ts index 02faea4130a88e..d24344a883ac31 100644 --- a/src/js/node/worker_threads.ts +++ b/src/js/node/worker_threads.ts @@ -362,7 +362,9 @@ class Worker extends EventEmitter { #onError(event: ErrorEvent) { this.#isRunning = false; let error = event?.error; - if (!error) { + // if the thrown value serialized successfully, the message will be empty + // if not the message is the actual error + if (event.message !== "") { error = new Error(event.message, { cause: event }); const stack = event?.stack; if (stack) { diff --git a/test/js/node/test/parallel/test-worker-nested-uncaught.js b/test/js/node/test/parallel/test-worker-nested-uncaught.js new file mode 100644 index 00000000000000..00bb6832203442 --- /dev/null +++ b/test/js/node/test/parallel/test-worker-nested-uncaught.js @@ -0,0 +1,14 @@ +'use strict'; +const common = require('../common'); +const { Worker } = require('worker_threads'); + +// Regression test for https://github.com/nodejs/node/issues/34309 + +const w = new Worker( + `const { Worker } = require('worker_threads'); + new Worker("throw new Error('uncaught')", { eval:true })`, + { eval: true }); +w.on('error', common.expectsError({ + name: 'Error', + message: 'uncaught' +})); diff --git a/test/js/node/worker_threads/worker_threads.test.ts b/test/js/node/worker_threads/worker_threads.test.ts index ab0a1a63cbaaec..4a3f4932b461ad 100644 --- a/test/js/node/worker_threads/worker_threads.test.ts +++ b/test/js/node/worker_threads/worker_threads.test.ts @@ -399,3 +399,25 @@ describe("environmentData", () => { expect(proc.exitCode).toBe(0); }); }); + +describe("error event", () => { + test("is fired with a copy of the error value", async () => { + const worker = new Worker("throw new TypeError('oh no')", { eval: true }); + const [err] = await once(worker, "error"); + expect(err).toBeInstanceOf(TypeError); + expect(err.message).toBe("oh no"); + }); + + test("falls back to string when the error cannot be serialized", async () => { + const worker = new Worker( + /* js */ ` + import { MessageChannel } from "node:worker_threads"; + const { port1 } = new MessageChannel(); + throw port1;`, + { eval: true }, + ); + const [err] = await once(worker, "error"); + expect(err).toBeInstanceOf(Error); + expect(err.message).toMatch(/MessagePort \{.*\}/s); + }); +}); diff --git a/test/js/web/workers/worker.test.ts b/test/js/web/workers/worker.test.ts index 47925159708980..403205bde46f71 100644 --- a/test/js/web/workers/worker.test.ts +++ b/test/js/web/workers/worker.test.ts @@ -286,6 +286,16 @@ describe("web worker", () => { return promise; }); }); + + describe("error event", () => { + test("is fired with a string of the error", async () => { + const worker = new Worker("data:text/javascript,throw 5"); + const [err] = await once(worker, "error"); + expect(err.type).toBe("error"); + expect(err.message).toBe("5"); + expect(err.error).toBe(null); + }); + }); }); // TODO: move to node:worker_threads tests directory