diff --git a/Makefile b/Makefile index 9ac8d24df..ff049c556 100644 --- a/Makefile +++ b/Makefile @@ -145,37 +145,14 @@ WASIX_SKIP_UNIX_SOCKET_TESTS := \ parallel/test-tls-connect-pipe.js \ parallel/test-tls-net-connect-prefer-path.js \ parallel/test-tls-wrap-econnreset-pipe.js \ - parallel/test-http-client-response-domain.js -# The known_issues/ entry is a later addition (2026-08-11). Cluster shares a -# dgram socket by sending the bound descriptor to the worker over IPC, and -# primary.js always picks SharedHandle for udp4/udp6 regardless of scheduling -# policy. WASI has no msghdr/SCM_RIGHTS, so that descriptor can never be -# delivered and the worker's bind callback never fires -- the run times out -# rather than failing, which a [negative] test needs it to do. Its non-negative -# sibling sequential/test-dgram-bind-shared-ports.js is skipped just above for -# the same reason; this one only escaped because the list carried no -# known_issues/ paths. See wasix-org/libuv#14, which makes the underlying send -# fail with ENOSYS instead of stalling silently -- necessary for diagnosing -# this, but not sufficient to make the test pass. -WASIX_SKIP_CLUSTER_FORK_TESTS := \ - parallel/test-dgram-bind-socket-close-before-cluster-reply.js \ - parallel/test-dgram-cluster-close-during-bind.js \ - parallel/test-dgram-cluster-close-in-listening.js \ - parallel/test-dgram-unref-in-cluster.js \ - parallel/test-http-server-drop-connections-in-cluster.js \ - parallel/test-tls-ticket-cluster.js \ - parallel/test-diagnostics-channel-process.js \ - parallel/test-http-chunk-problem.js \ - parallel/test-http-client-with-create-connection.js \ - parallel/test-http-full-response.js \ - parallel/test-http-server-stale-close.js \ - parallel/test-dgram-deprecation-error.js \ - parallel/test-https-agent-unref-socket.js \ - parallel/test-crypto-secure-heap.js \ - parallel/test-domain-top-level-error-handler-throw.js \ - parallel/test-domain-uncaught-exception.js \ - sequential/test-dgram-bind-shared-ports.js \ - known_issues/test-dgram-bind-shared-ports-after-port-0.js + parallel/test-http-client-response-domain.js \ + parallel/test-http-client-with-create-connection.js +# Emptied 2026-07-07: with fork IPC (libuv-wasix plain read) and the cluster +# reuseport scheduling strategy (TCP and UDP) in place, every cluster/fork +# test in the wasix lanes passes. test-http-client-with-create-connection +# moved to the unix-socket group and test-crypto-secure-heap to the crypto +# group (misfiled here; their failures are unrelated to cluster/fork). +WASIX_SKIP_CLUSTER_FORK_TESTS := WASIX_SKIP_SUBPROCESS_SHELL_TESTS := \ parallel/test-stream-pipeline-process.js \ parallel/test-domain-abort-on-uncaught.js \ @@ -197,7 +174,8 @@ WASIX_SKIP_CRYPTO_UNSUPPORTED_TESTS := \ parallel/test-crypto-argon2.js \ parallel/test-crypto-no-algorithm.js \ parallel/test-webcrypto-derivebits-argon2.js \ - parallel/test-crypto-pqc-keygen-slh-dsa.js + parallel/test-crypto-pqc-keygen-slh-dsa.js \ + parallel/test-crypto-secure-heap.js WASIX_SKIP_TLS_SUBPROCESS_ENV_TESTS := \ parallel/test-tls-enable-keylog-cli.js \ parallel/test-tls-env-bad-extra-ca.js \ @@ -268,9 +246,15 @@ WASIX_SLOW_WEBCRYPTO_TESTS := \ parallel/test-webcrypto-webidl.js \ parallel/test-webcrypto-wrap-unwrap.js # CI-only harness timeouts under parallel WASIX load (default harness timeout is 10s). +# test-http-chunk-problem (spawns cat) and test-http-full-response (execs ab +# through a shell) run external guest binaries; the first such exec +# cold-downloads and compiles wasmer/bash + wasmer/coreutils on runners with +# an empty wasmer cache, so they need the scaled timeout rather than a skip. WASIX_SLOW_TESTS := \ parallel/test-crypto-oneshot-hash-xof.js \ parallel/test-fastutf8stream-flush-sync.js \ + parallel/test-http-chunk-problem.js \ + parallel/test-http-full-response.js \ parallel/test-http2-respond-file-with-pipe.js \ parallel/test-stringbytes-external.js \ parallel/test-url-parse-invalid-input.js \ diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index 63d9eed4a..63a2efbaf 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -9,6 +9,7 @@ set(EDGE_RUNTIME_CORE_SOURCES "${CMAKE_CURRENT_LIST_DIR}/edge_runtime.cc" "${CMAKE_CURRENT_LIST_DIR}/edge_runtime_platform_v8.cc" "${CMAKE_CURRENT_LIST_DIR}/edge_node_compat.cc" + "${CMAKE_CURRENT_LIST_DIR}/edge_cluster_wasix.cc" "${CMAKE_CURRENT_LIST_DIR}/edge_process.cc" "${CMAKE_CURRENT_LIST_DIR}/edge_active_resource.cc" "${CMAKE_CURRENT_LIST_DIR}/edge_worker_env.cc" diff --git a/src/edge_cluster_wasix.cc b/src/edge_cluster_wasix.cc new file mode 100644 index 000000000..b0acc51b7 --- /dev/null +++ b/src/edge_cluster_wasix.cc @@ -0,0 +1,414 @@ +#include "edge_cluster_wasix.h" + +#if defined(__wasi__) + +#include +#include + +#include + +#include "edge_module_loader.h" + +namespace { + +// Message acts exchanged between the worker strategy and the primary broker +// below. Defined once and passed into both scripts so they cannot drift. +constexpr const char kReserveAct[] = "edge:wasix-reserve-port"; +constexpr const char kReportAct[] = "edge:wasix-report-port"; + +// Invoked with (cluster, net, dgram, sendHelper, UV_TCP_REUSEPORT, +// UV_UDP_REUSEPORT, RESERVE_ACT, REPORT_ACT). Kept as an embedded script +// instead of a lib/ module because the Node lib/ tree stays byte-identical to +// upstream; the behavior is edge/WASIX-specific. +// +// TCP and UDP port listens replace queryServer with a round trip to the broker +// in the primary (see the broker script below); the worker then binds its own +// SO_REUSEPORT socket rather than receiving a descriptor, and reports the +// 'listening' act for the primary's bookkeeping (worker state, cluster +// 'listening' event). fd and pipe listens keep the upstream path. +// +// The round trip settles the two things a worker cannot decide alone: +// +// * the port, when the caller asked for 0. "Any port" only holds a cluster +// together if every worker lands on the same one, which upstream gets for +// free by binding once in the primary and sharing the descriptor. WASI has +// no SCM_RIGHTS, so instead the broker arbitrates the *number*: one worker +// is elected to bind ephemerally and report what it got, and the rest are +// told to bind that. +// * the server data (obj._getServerData/_setServerData, i.e. TLS session +// ticket keys), which has to come from one place for all workers or a +// session only resumes on the worker that happened to issue it. +// +// Keeping the round trip even for a fixed port costs nothing relative to +// upstream, which always makes one; only the descriptor passing is dropped. +// +// Remaining deviation vs upstream: UDP datagrams are distributed by the +// kernel's reuseport source hash (flows pin to a worker) instead of +// shared-socket delivery. +constexpr const char kWorkerInstallScript[] = R"JS( +(function installWasixClusterReusePort(cluster, net, dgram, sendHelper, + UV_TCP_REUSEPORT, UV_UDP_REUSEPORT, + RESERVE_ACT, REPORT_ACT) { + 'use strict'; + + if (!cluster.isWorker || typeof cluster._getServer !== 'function') + return; + + const UV_EINVAL = -22; + const originalGetServer = cluster._getServer; + const ownHandles = new Set(); + // Per-worker sequence per listen target, mirroring the `index` that + // lib/internal/cluster/child.js attaches to queryServer. + const indexes = new Map(); + let disconnectHookInstalled = false; + + cluster._getServer = function(obj, options, cb) { + const isTcp = options.addressType === 4 || options.addressType === 6; + const isUdp = options.addressType === 'udp4' || + options.addressType === 'udp6'; + // dgram passes the raw bind() arguments through: port can be null, + // undefined, or even the bind callback function (socket.bind(cb)); + // per the bind([port][, address][, callback]) signature all of those + // mean an ephemeral-port listen, not an fd/pipe listen. + const port = (isUdp && typeof options.port !== 'number') ? 0 : options.port; + const isPortListen = + typeof port === 'number' && port >= 0 && + (options.fd == null || options.fd < 0); + + if ((!isTcp && !isUdp) || !isPortListen) + return originalGetServer.call(this, obj, options, cb); + + function createHandle(boundPort) { + return isTcp ? + net._createServerHandle(options.address, boundPort, + options.addressType, options.fd, + (options.flags | UV_TCP_REUSEPORT) >>> 0) : + dgram._createSocketHandle(options.address, boundPort, + options.addressType, options.fd, + (options.flags | UV_UDP_REUSEPORT) >>> 0); + } + + function adopt(rval) { + ownHandles.add(rval); + const originalClose = rval.close; + rval.close = function() { + ownHandles.delete(rval); + return originalClose.apply(rval, arguments); + }; + + if (!disconnectHookInstalled && cluster.worker) { + disconnectHookInstalled = true; + // Mirror lib/internal/cluster/child.js: close listeners when the + // worker disconnects so the process can drain and exit. + cluster.worker.once('disconnect', () => { + for (const handle of ownHandles) + handle.close(); + ownHandles.clear(); + }); + } + + obj.once('listening', () => { + if (cluster.worker) + cluster.worker.state = 'listening'; + const address = obj.address(); + sendHelper(process, { + act: 'listening', + address: options.address, + port: (address && address.port) || options.port, + addressType: options.addressType, + fd: options.fd, + }, null); + }); + } + + // Keyed exactly like queryServer() in lib/internal/cluster/primary.js so + // that two distinct ephemeral servers in one worker stay distinct, and the + // Nth ephemeral server of every worker agrees with the Nth of the others. + const indexesKey = + `${options.address}:${port}:${options.addressType}:${options.fd}`; + const index = indexes.get(indexesKey) || 0; + indexes.set(indexesKey, index + 1); + const key = `${indexesKey}:${index}`; + + // Every listen takes the round trip, exactly as upstream does, even when + // the port is fixed and needs no arbitration. The reply also carries the + // server data (TLS session ticket keys), which has to come from one place + // for all workers or a session only resumes on the worker that issued it. + // Binding locally with SO_REUSEPORT is the only part that diverges from + // upstream; the round trip itself is not extra cost. + sendHelper(process, { + act: RESERVE_ACT, + key, + port, + data: typeof obj._getServerData === 'function' ? obj._getServerData() : null, + }, null, (reply) => { + if (!reply || (!reply.assign && !reply.port)) { + cb(UV_EINVAL, null); + return; + } + + if (typeof obj._setServerData === 'function') + obj._setServerData(reply.data); + + const rval = createHandle(reply.assign ? 0 : reply.port); + if (typeof rval === 'number') { + // Report the failed election so a waiter can take over; otherwise + // every other worker blocks forever on a port nobody owns. + if (reply.assign) { + sendHelper(process, + { act: REPORT_ACT, key, port: 0, errno: rval }, null); + } + cb(rval, null); + return; + } + + if (reply.assign) { + const sockname = {}; + const err = rval.getsockname(sockname); + const chosen = err === 0 ? sockname.port : 0; + sendHelper(process, + { act: REPORT_ACT, key, port: chosen, errno: err || 0 }, + null); + if (!chosen) { + // Bound, but to a port we cannot name, so no peer can join it. + // Fail this listen rather than run a one-worker "cluster". + rval.close(); + cb(err || UV_EINVAL, null); + return; + } + } + + adopt(rval); + cb(0, rval); + }); + }; +}) +)JS"; + +// Invoked with (cluster, sendHelper, RESERVE_ACT, REPORT_ACT) in the primary. +// +// Arbitrates the ephemeral port for a listen key. The first worker to ask is +// elected to pick one and report it back; anyone who asks in the meantime is +// parked until it does. Only a number crosses the channel, never a descriptor, +// so this works on WASIX where the upstream shared-handle path cannot. +// +// lib/internal/cluster/primary.js ignores acts it does not know +// (methodMessageMapping lookup), so these ride alongside the upstream protocol +// on a second 'internalMessage' listener without disturbing it. +constexpr const char kPrimaryInstallScript[] = R"JS( +(function installWasixClusterPortBroker(cluster, sendHelper, + RESERVE_ACT, REPORT_ACT) { + 'use strict'; + + if (!cluster.isPrimary) return; + // Spawn calls this on every cluster.fork(); only the first should install. + if (cluster.__edgeWasixPortBroker) return; + Object.defineProperty(cluster, '__edgeWasixPortBroker', { value: true }); + + // key -> { port, elector, waiters: [{ worker, seq }] } + const reservations = new Map(); + + function reply(worker, seq, payload) { + if (!worker || !worker.process || !worker.process.connected) return; + payload.ack = seq; + sendHelper(worker.process, payload, null); + } + + function elect(entry, worker, seq) { + entry.elector = worker; + reply(worker, seq, { assign: true, data: entry.data }); + } + + function promoteNextWaiter(key, entry) { + const next = entry.waiters.shift(); + if (next) elect(entry, next.worker, next.seq); + else reservations.delete(key); + } + + function onMessage(worker, message) { + if (!message || message.cmd !== 'NODE_CLUSTER') return; + + if (message.act === RESERVE_ACT) { + let entry = reservations.get(message.key); + if (entry === undefined) { + // A fixed port is already agreed by definition; only port 0 needs an + // election. Either way the first caller's data becomes the shared + // copy, matching `handle.data ||= message.data` in primary.js. + entry = { + port: message.port > 0 ? message.port : 0, + data: message.data, + elector: null, + waiters: [], + }; + reservations.set(message.key, entry); + } else if (entry.data == null) { + entry.data = message.data; + } + + if (entry.port) + reply(worker, message.seq, { port: entry.port, data: entry.data }); + else if (entry.elector === null) + elect(entry, worker, message.seq); + else + entry.waiters.push({ worker, seq: message.seq }); + return; + } + + if (message.act === REPORT_ACT) { + const entry = reservations.get(message.key); + if (entry === undefined || entry.elector !== worker) return; + entry.elector = null; + if (message.port > 0) { + entry.port = message.port; + for (const waiter of entry.waiters) + reply(waiter.worker, waiter.seq, { port: entry.port, data: entry.data }); + entry.waiters.length = 0; + } else { + promoteNextWaiter(message.key, entry); + } + } + } + + cluster.on('fork', (worker) => { + worker.process.on('internalMessage', + (message) => onMessage(worker, message)); + worker.once('exit', () => { + // A worker that dies mid-election would otherwise strand every waiter. + for (const { 0: key, 1: entry } of reservations) { + entry.waiters = entry.waiters.filter((w) => w.worker !== worker); + if (entry.elector === worker && !entry.port) { + entry.elector = null; + promoteNextWaiter(key, entry); + } + } + }); + }); +}) +)JS"; + +void ClearPendingException(napi_env env) { + bool pending = false; + if (napi_is_exception_pending(env, &pending) == napi_ok && pending) { + napi_value ignored = nullptr; + (void)napi_get_and_clear_last_exception(env, &ignored); + } +} + +// Compiles `script` and calls it with `argv`. Returns false and swallows the +// exception if anything along the way fails; a missing strategy degrades to +// upstream behavior rather than taking the process down. +bool RunInstallScript(napi_env env, + const char* source, + napi_value* argv, + size_t argc) { + napi_value script = nullptr; + napi_value install_fn = nullptr; + if (napi_create_string_utf8(env, source, NAPI_AUTO_LENGTH, &script) != napi_ok || + napi_run_script(env, script, &install_fn) != napi_ok || install_fn == nullptr) { + ClearPendingException(env); + return false; + } + + napi_value global = nullptr; + if (napi_get_global(env, &global) != napi_ok) { + ClearPendingException(env); + return false; + } + + napi_value result = nullptr; + if (napi_call_function(env, global, install_fn, argc, argv, &result) != napi_ok) { + ClearPendingException(env); + return false; + } + return true; +} + +// Resolves internal/cluster/utils#sendHelper, the ack-matching send both +// halves of the protocol ride on. +bool GetSendHelper(napi_env env, napi_value* out) { + napi_value utils = nullptr; + if (!EdgeRequireBuiltin(env, "internal/cluster/utils", &utils) || utils == nullptr) { + ClearPendingException(env); + return false; + } + if (napi_get_named_property(env, utils, "sendHelper", out) != napi_ok || *out == nullptr) { + ClearPendingException(env); + return false; + } + return true; +} + +bool GetActNames(napi_env env, napi_value* reserve, napi_value* report) { + if (napi_create_string_utf8(env, kReserveAct, NAPI_AUTO_LENGTH, reserve) != napi_ok || + napi_create_string_utf8(env, kReportAct, NAPI_AUTO_LENGTH, report) != napi_ok) { + ClearPendingException(env); + return false; + } + return true; +} + +} // namespace + +void EdgeMaybeInstallWasixClusterReusePort(napi_env env) { + // pre_execution deletes NODE_UNIQUE_ID from process.env, but this runs + // before the main builtin, while the variable is still present. + if (std::getenv("NODE_UNIQUE_ID") == nullptr) return; + + napi_value cluster = nullptr; + napi_value net = nullptr; + napi_value dgram = nullptr; + if (!EdgeRequireBuiltin(env, "cluster", &cluster) || cluster == nullptr || + !EdgeRequireBuiltin(env, "net", &net) || net == nullptr || + !EdgeRequireBuiltin(env, "internal/dgram", &dgram) || dgram == nullptr) { + ClearPendingException(env); + return; + } + + napi_value send_helper = nullptr; + if (!GetSendHelper(env, &send_helper)) return; + + napi_value tcp_reuseport_flag = nullptr; + napi_value udp_reuseport_flag = nullptr; + napi_value reserve_act = nullptr; + napi_value report_act = nullptr; + if (napi_create_uint32(env, static_cast(UV_TCP_REUSEPORT), &tcp_reuseport_flag) != napi_ok || + napi_create_uint32(env, static_cast(UV_UDP_REUSEPORT), &udp_reuseport_flag) != napi_ok) { + ClearPendingException(env); + return; + } + if (!GetActNames(env, &reserve_act, &report_act)) return; + + napi_value argv[] = {cluster, net, dgram, send_helper, + tcp_reuseport_flag, udp_reuseport_flag, + reserve_act, report_act}; + (void)RunInstallScript(env, kWorkerInstallScript, argv, 8); +} + +void EdgeMaybeInstallWasixClusterPrimary(napi_env env) { + // Only reached from a cluster.fork() spawn, so `cluster` is already resolved + // and this require is a cache hit. Requiring it eagerly at startup instead + // would cost every process ~21 ms for a module most never touch. + napi_value cluster = nullptr; + if (!EdgeRequireBuiltin(env, "cluster", &cluster) || cluster == nullptr) { + ClearPendingException(env); + return; + } + + napi_value send_helper = nullptr; + if (!GetSendHelper(env, &send_helper)) return; + + napi_value reserve_act = nullptr; + napi_value report_act = nullptr; + if (!GetActNames(env, &reserve_act, &report_act)) return; + + napi_value argv[] = {cluster, send_helper, reserve_act, report_act}; + (void)RunInstallScript(env, kPrimaryInstallScript, argv, 4); +} + +#else // !defined(__wasi__) + +void EdgeMaybeInstallWasixClusterReusePort(napi_env /*env*/) {} + +void EdgeMaybeInstallWasixClusterPrimary(napi_env /*env*/) {} + +#endif // defined(__wasi__) diff --git a/src/edge_cluster_wasix.h b/src/edge_cluster_wasix.h new file mode 100644 index 000000000..3a004bef2 --- /dev/null +++ b/src/edge_cluster_wasix.h @@ -0,0 +1,40 @@ +#ifndef EDGE_CLUSTER_WASIX_H_ +#define EDGE_CLUSTER_WASIX_H_ + +#include "unofficial_napi.h" + +// Installs the WASIX cluster reuseport scheduling strategy in cluster worker +// processes. No-op on native targets and outside cluster workers. +// +// WASIX cannot pass listen handles between processes (no SCM_RIGHTS over the +// IPC channel), which breaks both of Node cluster's scheduling strategies: +// round robin passes every accepted connection to a worker, and shared-handle +// mode passes the listen handle itself. SO_REUSEPORT works end to end, so TCP +// and UDP port listens in cluster workers bind their own socket instead and +// the host kernel distributes traffic between the workers. +// +// The strategy is implemented by replacing the worker-side cluster._getServer +// (an exported, documented-as-replaceable property) from an embedded script. +// It lives here rather than in lib/ because the Node lib/ tree is kept +// byte-identical to upstream. +void EdgeMaybeInstallWasixClusterReusePort(napi_env env); + +// Installs the primary-side half of the strategy: a broker that arbitrates the +// ephemeral port behind a port-0 listen. No-op on native targets, outside the +// primary, and after the first call. +// +// A port-0 listen means "any port", but a cluster only stays one server if +// every worker lands on the *same* one. Upstream gets that for free by binding +// once in the primary and sharing the descriptor; without SCM_RIGHTS the +// workers must instead be told which number to bind, which is what the broker +// answers. Only a port number crosses the channel, never a descriptor. +// +// Call this when a cluster.fork() spawn is observed rather than at startup: +// that is the first point where the process is known to be a cluster primary, +// and `cluster` is already loaded by then, so it costs nothing. Requiring +// `cluster` eagerly would add ~21 ms to the startup of every process. +// createWorkerProcess() runs before cluster.emit('fork'), so a broker +// installed during the spawn still observes the worker that triggered it. +void EdgeMaybeInstallWasixClusterPrimary(napi_env env); + +#endif // EDGE_CLUSTER_WASIX_H_ diff --git a/src/edge_process_wrap.cc b/src/edge_process_wrap.cc index 397fe17a8..1207550cc 100644 --- a/src/edge_process_wrap.cc +++ b/src/edge_process_wrap.cc @@ -22,6 +22,7 @@ #include "edge_handle_scope.h" #include "edge_handle_wrap.h" #include "edge_runtime.h" +#include "edge_cluster_wasix.h" #include "edge_stream_base.h" namespace { @@ -554,6 +555,17 @@ napi_value ProcessSpawn(napi_env env, napi_callback_info info) { UpdateEnvPairInt(&env_storage, &envp, "NODE_CHANNEL_FD", kWasixIpcFd); } } + // cluster.fork() is the only spawn that sets NODE_UNIQUE_ID, so this is the + // first point at which the process is known to be a cluster primary. Install + // the port broker now: `cluster` is already loaded (it is what called us), so + // this is a cache hit, whereas requiring it at startup would cost every + // process ~21 ms. The installer is idempotent. + for (const std::string& pair : env_storage) { + if (pair.rfind("NODE_UNIQUE_ID=", 0) == 0) { + EdgeMaybeInstallWasixClusterPrimary(env); + break; + } + } #endif uv_process_options_t options{}; diff --git a/src/edge_runtime.cc b/src/edge_runtime.cc index e9e47af50..a59363e15 100644 --- a/src/edge_runtime.cc +++ b/src/edge_runtime.cc @@ -50,6 +50,7 @@ #include "edge_crypto.h" #include "edge_encoding.h" #include "edge_http_parser.h" +#include "edge_cluster_wasix.h" #include "edge_module_loader.h" #include "edge_os.h" #include "edge_option_helpers.h" @@ -3043,6 +3044,11 @@ int RunScriptWithGlobals(napi_env env, delete_global_named("__dirname"); } + // Under WASIX, cluster workers get the reuseport scheduling strategy + // installed before the main builtin runs pre-execution (which consumes + // NODE_UNIQUE_ID). No-op elsewhere. + EdgeMaybeInstallWasixClusterReusePort(env); + napi_value result = nullptr; if (selected_main_builtin_id != nullptr && selected_main_builtin_id[0] != '\0') { if (EdgeExecuteBuiltin(env, selected_main_builtin_id, &result)) {