diff --git a/.github/workflows/benchmark.yml b/.github/workflows/benchmark.yml index b989dd1faaef45..48b3f1d6125975 100644 --- a/.github/workflows/benchmark.yml +++ b/.github/workflows/benchmark.yml @@ -25,11 +25,43 @@ on: type: number default: 30 description: How many times to repeat each benchmark + post-comment: + type: boolean + description: Post a comment linking to the run, with the aggregated result once known. + token: + type: string + description: A GitHub token to post comments cross repository (not recommended) permissions: contents: read jobs: + post-comment: + if: inputs.post-comment + outputs: + body: ${{ steps.comment.outputs.COMMENT_BODY }} + url: ${{ steps.comment.outputs.COMMENT_URL }} + runs-on: ubuntu-slim + permissions: + pull-requests: write + steps: + - name: Mark token input as sensitive + if: inputs.token != '' + run: echo "::add-mask::${{ inputs.token }}" + - name: Add link to the current run + id: comment + run: | + FILTER_IF_SET= + [ -z "$FILTER" ] || FILTER_IF_SET=" / $FILTER" + COMMENT_BODY="Benchmark GHA (${CATEGORIES}${FILTER_IF_SET}): ${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY}/actions/runs/${GITHUB_RUN_ID}" + echo "COMMENT_BODY=$COMMENT_BODY" >> "$GITHUB_OUTPUT" + echo "COMMENT_URL=$(gh pr comment -R "${REPO:-$GITHUB_REPOSITORY}" "$PR_ID" --body "$COMMENT_BODY")" >> "$GITHUB_OUTPUT" + env: + CATEGORIES: ${{ inputs.category }} + FILTER: ${{ inputs.filter }} + GH_TOKEN: ${{ inputs.token || github.token }} + REPO: ${{ inputs.repo }} + PR_ID: ${{ inputs.pr_id }} build: strategy: fail-fast: true @@ -155,9 +187,13 @@ jobs: path: ${{ matrix.system }}.csv aggregate-results: - needs: build + needs: [build, post-comment] + if: ${{ always() && needs.build.result == 'success' && needs.post-comment.result == (inputs.post-comment && 'success' || 'skipped') }} name: Aggregate benchmark results runs-on: ubuntu-slim + permissions: + contents: read + pull-requests: write steps: - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 with: @@ -200,4 +236,30 @@ jobs: echo "> [!WARNING] " echo "> Do not take GHA benchmark results as face value, always confirm them" echo "> using a dedicated machine, e.g. Jenkins CI." - ' | tee /dev/stderr >> "$GITHUB_STEP_SUMMARY" + ' | tee /dev/stderr ${{ inputs.post-comment && 'body.txt' || '' }} >> "$GITHUB_STEP_SUMMARY" + - name: Mark token input as sensitive + if: inputs.token != '' + run: echo "::add-mask::${{ inputs.token }}" + - name: Edit comment + if: inputs.post-comment + run: | + { + echo "$COMMENT_BODY" + echo + echo '
Results' + echo + cat body.txt + echo + echo '
' + } | jq -Rrcs '{ body: . }' | gh api \ + --method PATCH \ + -H "Accept: application/vnd.github+json" \ + -H "X-GitHub-Api-Version: 2026-03-10" \ + "/repos/${REPO:-$GITHUB_REPOSITORY}/issues/comments/${COMMENT_URL##*-}" \ + --input - + env: + GH_TOKEN: ${{ inputs.token || github.token }} + REPO: ${{ inputs.repo }} + COMMENT_BODY: ${{ needs.post-comment.outputs.body }} + COMMENT_URL: ${{ needs.post-comment.outputs.url }} + PR_ID: ${{ inputs.pr_id }} diff --git a/doc/api/errors.md b/doc/api/errors.md index db3eb81eff6b90..29cd9e8a4937d1 100644 --- a/doc/api/errors.md +++ b/doc/api/errors.md @@ -3582,6 +3582,22 @@ added: v18.1.0 The `Response` that has been passed to `WebAssembly.compileStreaming` or to `WebAssembly.instantiateStreaming` is not a valid WebAssembly response. + + +### `ERR_WORKER_HANDLE_NOT_TRANSFERABLE` + +An attempt was made to transfer a `net.Socket` or `net.Server` to another thread +via a `worker_threads` `postMessage()` call while it was not in a transferable +state, for example because it had already started reading or had buffered data. + + + +### `ERR_WORKER_HANDLE_TRANSFER_UNSUPPORTED` + +An attempt was made to transfer a `net.Socket` or `net.Server` to another thread +on a platform where moving the underlying handle between event loops is not +supported (currently Windows). + ### `ERR_WORKER_INIT_FAILED` diff --git a/doc/api/net.md b/doc/api/net.md index e2f06776409898..81863d9b0a13c5 100644 --- a/doc/api/net.md +++ b/doc/api/net.md @@ -305,6 +305,11 @@ added: v0.1.90 This class is used to create a TCP or [IPC][] server. +A listening TCP `net.Server` can be transferred to a worker thread by listing it +in the `transferList` of a [`worker_threads`][] `postMessage()` call. This moves +the underlying listening socket to the receiving thread, where it resumes +accepting connections. See [Transferring TCP handles to other threads][]. + ### `new net.Server([options][, connectionListener])` * `options` {Object} See @@ -751,6 +756,41 @@ is received. For example, it is passed to the listeners of a [`'connection'`][] event emitted on a [`net.Server`][], so the user can use it to interact with the client. +### Transferring TCP handles to other threads + +A connected TCP `net.Socket` can be moved to another thread by listing it in the +`transferList` of a [`worker_threads`][] `postMessage()` call. After the +transfer, the source socket is destroyed on the sending thread (further use +fails with `ERR_STREAM_DESTROYED` rather than silently dropping data), and the +socket continues to work on the receiving thread. This makes it possible to +accept connections on one thread and distribute them across a pool of worker +threads, for example to build a `node:cluster`-like model on top of worker +threads. + +The socket must be a freshly accepted or created TCP connection: it must still +be attached to a live handle, must not be connecting or destroyed, and must not +have started reading or have buffered data. Otherwise `postMessage()` throws +`ERR_WORKER_HANDLE_NOT_TRANSFERABLE`. Only TCP sockets are supported, and only +on Unix-like platforms; on Windows `postMessage()` throws +`ERR_WORKER_HANDLE_TRANSFER_UNSUPPORTED`. + +```cjs +const net = require('node:net'); +const { Worker } = require('node:worker_threads'); + +// worker.js receives `{ socket }` messages and handles each connection. +const worker = new Worker('./worker.js'); + +const server = net.createServer((socket) => { + // Hand the freshly accepted connection off to the worker thread. + worker.postMessage({ socket }, [socket]); +}); +server.listen(8000); +``` + +A listening [`net.Server`][] can be transferred the same way, which moves the +listening socket itself (and its pending accept queue) to the receiving thread. + ### `new net.Socket([options])` + +* `data` {Object} + * `column` {number|undefined} The column number where the test is defined, or + `undefined` if the test was run through the REPL. + * `data` {any} The structured payload passed to [`context.log`][], or + `undefined` if none was provided. The test runner does not interpret this + value. + * `entryFile` {string|undefined} The path of the test file that was + executed as the entry point of the child process that emitted this event. + Only present when tests run with process isolation. May differ from + `file` when the test is defined in a module imported by the entry file. + * `file` {string|undefined} The path of the test file, + `undefined` if test was run through the REPL. + * `line` {number|undefined} The line number where the test is defined, or + `undefined` if the test was run through the REPL. + * `message` {string} The log message. + * `name` {string} The test name. + * `nesting` {number} The nesting level of the test. + * `parentId` {number|undefined} The `testId` of the enclosing test, or + `undefined` for top-level tests. + * `testId` {number} A numeric identifier for the test instance that emitted + the log message. + +Emitted when [`context.log`][] is called. Unlike [`'test:diagnostic'`][], +this event is emitted immediately, in the order that the tests execute, +making it suitable for reporters that render test output unbuffered. + ### Event: `'test:pass'` * `data` {Object} @@ -4272,6 +4308,29 @@ test('top level test', (t) => { }); ``` +### `context.log(message[, data])` + + + +* `message` {string} Message to be reported. +* `data` {any} Optional structured payload attached to the message. The test + runner passes it through untouched. When tests run with process isolation, + this value must be compatible with the [HTML structured clone algorithm][]. + +This function is used to write a log message to the output. Unlike +[`context.diagnostic`][], the resulting [`'test:log'`][] event is emitted +immediately, in the order that the tests execute, rather than being buffered +until the test reports its results. This function does not return a value. + +```js +test('top level test', (t) => { + t.log('fetched user', { userId: 42 }); + t.log('retrying flaky endpoint', { attempt: 3 }); +}); +``` + ### `context.filePath` + +* `message` {string} Message to be reported. +* `data` {any} Optional structured payload attached to the message. The test + runner passes it through untouched. + +Write a log message to the output. The resulting [`'test:log'`][] event is +emitted immediately, in the order that the tests execute. + +```js +test.describe('my suite', (suite) => { + suite.log('Suite log message'); +}); +``` + +[HTML structured clone algorithm]: https://developer.mozilla.org/en-US/docs/Web/API/Web_Workers_API/Structured_clone_algorithm [TAP]: https://testanything.org/ [Test tags]: #test-tags [`'test:complete'`]: #event-testcomplete @@ -4770,6 +4849,7 @@ test.describe('my suite', (suite) => { [`'test:enqueue'`]: #event-testenqueue [`'test:fail'`]: #event-testfail [`'test:interrupted'`]: #event-testinterrupted +[`'test:log'`]: #event-testlog [`'test:pass'`]: #event-testpass [`'test:plan'`]: #event-testplan [`'test:start'`]: #event-teststart @@ -4805,6 +4885,7 @@ test.describe('my suite', (suite) => { [`TracingChannel`]: diagnostics_channel.md#class-tracingchannel [`assert.throws`]: assert.md#assertthrowsfn-error-message [`context.diagnostic`]: #contextdiagnosticmessage +[`context.log`]: #contextlogmessage-data [`context.skip`]: #contextskipmessage [`context.tags`]: #contexttags [`context.todo`]: #contexttodomessage diff --git a/doc/api/worker_threads.md b/doc/api/worker_threads.md index 525825ade82621..d0b4f0a830b03e 100644 --- a/doc/api/worker_threads.md +++ b/doc/api/worker_threads.md @@ -1208,6 +1208,8 @@ In particular, the significant differences to `JSON` are: * {KeyObject}s, * {MessagePort}s, * {net.BlockList}s, + * {net.Server}s (TCP only, when listed in `transferList`), + * {net.Socket}s (TCP only, when listed in `transferList`), * {net.SocketAddress}es, * {X509Certificate}s. @@ -1235,12 +1237,20 @@ circularData.foo = circularData; port2.postMessage(circularData); ``` -`transferList` may be a list of {ArrayBuffer}, [`MessagePort`][], and -[`FileHandle`][] objects. +`transferList` may be a list of {ArrayBuffer}, [`MessagePort`][], +[`FileHandle`][], {net.Server}, and {net.Socket} objects. After transferring, they are not usable on the sending side of the channel -anymore (even if they are not contained in `value`). Unlike with -[child processes][], transferring handles such as network sockets is currently -not supported. +anymore (even if they are not contained in `value`). + +Transferring a {net.Server} moves its listening socket — together with any +pending connections in the accept queue — to the receiving thread's event loop. +Transferring a {net.Socket} moves a single connection; the socket must be a +freshly accepted or created TCP connection that has not yet started reading and +has no buffered data, otherwise `postMessage()` throws +`ERR_WORKER_HANDLE_NOT_TRANSFERABLE`. This makes it possible to accept +connections on one thread and distribute them across a pool of worker threads. +Only TCP handles are supported, and only on Unix-like platforms; on Windows +`postMessage()` throws `ERR_WORKER_HANDLE_TRANSFER_UNSUPPORTED`. If `value` contains {SharedArrayBuffer} instances, those are accessible from either thread. They cannot be listed in `transferList`. @@ -2276,7 +2286,6 @@ thread spawned will spawn another until the application crashes. [async-resource-worker-pool]: async_context.md#using-asyncresource-for-a-worker-thread-pool [browser `LockManager`]: https://developer.mozilla.org/en-US/docs/Web/API/LockManager [browser `MessagePort`]: https://developer.mozilla.org/en-US/docs/Web/API/MessagePort -[child processes]: child_process.md [contextified]: vm.md#what-does-it-mean-to-contextify-an-object [locks.request()]: #locksrequestname-options-callback [v8.serdes]: v8.md#serialization-api diff --git a/lib/_http_client.js b/lib/_http_client.js index 891da4e3f9a984..d202a28f03502f 100644 --- a/lib/_http_client.js +++ b/lib/_http_client.js @@ -121,6 +121,8 @@ let debug = require('internal/util/debuglog').debuglog('http', (fn) => { const INVALID_PATH_REGEX = /[^\u0021-\u00ff]/; const kError = Symbol('kError'); const kPath = Symbol('kPath'); +const kAuthority = Symbol('kAuthority'); +const kProxyRewrittenToAbsolute = Symbol('kProxyRewrittenToAbsolute'); const HTTP_CLIENT_TRACE_EVENT_NAME = 'http.client.request'; @@ -318,6 +320,7 @@ function rewriteForProxiedHttp(req, reqOptions, proxyAuthority, userHostHeader, if (requestURL !== undefined) { // Convert the path to absolute-form. The authority is built from options. req.path = requestURL.href; + req[kProxyRewrittenToAbsolute] = true; } debug(`updated request for HTTP proxy ${href} with ${req.path} `, req[kOutHeaders]); return true; @@ -479,6 +482,7 @@ function ClientRequest(input, options, cb) { this.reusedSocket = false; this.host = host; this.protocol = protocol; + this[kProxyRewrittenToAbsolute] = false; if (this.agent) { // If there is an agent we should default to Connection:keep-alive, @@ -509,6 +513,9 @@ function ClientRequest(input, options, cb) { if (port && +port !== defaultPort) { hostHeaderFromOptions += ':' + port; } + // Preserve the request authority (with the port when non-default) so that + // the perf_hooks entry can report a faithful URL. + this[kAuthority] = hostHeaderFromOptions; const headersArray = ArrayIsArray(options.headers); if (!headersArray) { if (options.headers) { @@ -627,7 +634,11 @@ ClientRequest.prototype._finish = function _finish() { detail: { req: { method: this.method, - url: `${this.protocol}//${this.host}${this.path}`, + // If the path has been rewritten to absolute-form for proxying, + // it is already a full URL. + url: this[kProxyRewrittenToAbsolute] ? + this.path : + `${this.protocol}//${this[kAuthority]}${this.path}`, headers: typeof this.getHeaders === 'function' ? this.getHeaders() : {}, }, }, @@ -741,10 +752,24 @@ function socketErrorListener(err) { debug('SOCKET ERROR:', err.message, err.stack); if (req) { + const res = req.res; + const isUserDestroyError = err === req[kError] || err === res?.errored; + // For Safety. Some additional errors might fire later on // and we need to make sure we don't double-fire the error event. socket._hadError = true; - emitErrorEvent(req, err); + // Before a response exists, the request itself failed. Once a response + // exists, socket teardown belongs to the IncomingMessage and is finalized + // by socketCloseListener. Preserve errors explicitly used to destroy the + // request or response, which have historically been emitted on the request. + if (!res || isUserDestroyError) { + emitErrorEvent(req, err); + } + + if (res) { + socket.destroy(); + return; + } } const parser = socket.parser; diff --git a/lib/internal/errors.js b/lib/internal/errors.js index 03d27332c894d8..5f4cba63a64499 100644 --- a/lib/internal/errors.js +++ b/lib/internal/errors.js @@ -1976,6 +1976,12 @@ E('ERR_WEBASSEMBLY_NOT_SUPPORTED', 'WebAssembly is not supported in this environment, but is required for %s', Error); E('ERR_WEBASSEMBLY_RESPONSE', 'WebAssembly response %s', TypeError); +E('ERR_WORKER_HANDLE_NOT_TRANSFERABLE', + '%s cannot be transferred in its current state; it must be a freshly ' + + 'created or accepted handle that has not started reading and has no ' + + 'pending writes', Error); +E('ERR_WORKER_HANDLE_TRANSFER_UNSUPPORTED', + 'Transferring a %s to another thread is not supported on this platform', Error); E('ERR_WORKER_INIT_FAILED', 'Worker initialization failure: %s', Error); E('ERR_WORKER_INVALID_EXEC_ARGV', (errors, msg = 'invalid execArgv flags') => `Initiated Worker with ${msg}: ${ArrayPrototypeJoin(errors, ', ')}`, diff --git a/lib/internal/test_runner/reporter/junit.js b/lib/internal/test_runner/reporter/junit.js index 130ff60af405ad..ed25a4bd5fbd7c 100644 --- a/lib/internal/test_runner/reporter/junit.js +++ b/lib/internal/test_runner/reporter/junit.js @@ -154,7 +154,8 @@ module.exports = async function* junitReporter(source) { } break; } - case 'test:diagnostic': { + case 'test:diagnostic': + case 'test:log': { const parent = currentSuite?.children ?? roots; ArrayPrototypePush(parent, { __proto__: null, nesting: event.data.nesting, comment: event.data.message, diff --git a/lib/internal/test_runner/reporter/spec.js b/lib/internal/test_runner/reporter/spec.js index fce0754e25061a..7f5920c7b38a30 100644 --- a/lib/internal/test_runner/reporter/spec.js +++ b/lib/internal/test_runner/reporter/spec.js @@ -91,8 +91,9 @@ class SpecReporter extends Transform { case 'test:stderr': case 'test:stdout': return data.message; - case 'test:diagnostic':{ - const diagnosticColor = reporterColorMap[data.level] || reporterColorMap['test:diagnostic']; + case 'test:diagnostic': + case 'test:log': { + const diagnosticColor = reporterColorMap[data.level] || reporterColorMap[type]; return `${diagnosticColor}${indent(data.nesting)}${reporterUnicodeSymbolMap[type]}${data.message}${colors.white}\n`; } case 'test:coverage': diff --git a/lib/internal/test_runner/reporter/tap.js b/lib/internal/test_runner/reporter/tap.js index c3bbd1d144eb35..2bacba167570a8 100644 --- a/lib/internal/test_runner/reporter/tap.js +++ b/lib/internal/test_runner/reporter/tap.js @@ -56,6 +56,7 @@ async function * tapReporter(source) { } break; } case 'test:diagnostic': + case 'test:log': yield `${indent(data.nesting)}# ${tapEscape(data.message)}\n`; break; case 'test:coverage': diff --git a/lib/internal/test_runner/reporter/utils.js b/lib/internal/test_runner/reporter/utils.js index 67030680019838..995155c533c51d 100644 --- a/lib/internal/test_runner/reporter/utils.js +++ b/lib/internal/test_runner/reporter/utils.js @@ -21,6 +21,7 @@ const reporterUnicodeSymbolMap = { 'test:fail': '\u2716 ', 'test:pass': '\u2714 ', 'test:diagnostic': '\u2139 ', + 'test:log': '\u2139 ', 'test:coverage': '\u2139 ', 'arrow:right': '\u25B6 ', 'hyphen:minus': '\uFE63 ', @@ -38,6 +39,9 @@ const reporterColorMap = { get 'test:diagnostic'() { return colors.blue; }, + get 'test:log'() { + return colors.blue; + }, get 'info'() { return colors.blue; }, diff --git a/lib/internal/test_runner/runner.js b/lib/internal/test_runner/runner.js index 7f50cc8521bcda..dc6529c220539a 100644 --- a/lib/internal/test_runner/runner.js +++ b/lib/internal/test_runner/runner.js @@ -130,7 +130,7 @@ const kCanceledTests = new SafeSet() // Execution-ordered events are forwarded immediately, bypassing the // per-file declaration-order buffer. const kExecutionOrderedEvents = new SafeSet() - .add('test:enqueue').add('test:dequeue').add('test:complete'); + .add('test:enqueue').add('test:dequeue').add('test:complete').add('test:log'); let kResistStopPropagation; diff --git a/lib/internal/test_runner/test.js b/lib/internal/test_runner/test.js index d19d6349f104bf..1b9faad58a68b5 100644 --- a/lib/internal/test_runner/test.js +++ b/lib/internal/test_runner/test.js @@ -80,6 +80,7 @@ const { validateNumber, validateObject, validateOneOf, + validateString, validateUint32, } = require('internal/validators'); const { @@ -315,6 +316,10 @@ class TestContext { this.#test.diagnostic(message); } + log(message, data) { + this.#test.log(message, data); + } + plan(count, options = kEmptyObject) { if (this.#test.plan !== null) { throw new ERR_TEST_FAILURE( @@ -523,6 +528,10 @@ class SuiteContext { diagnostic(message) { this.#suite.diagnostic(message); } + + log(message, data) { + this.#suite.log(message, data); + } } function parseExpectFailure(expectFailure) { @@ -1198,6 +1207,12 @@ class Test extends AsyncResource { ArrayPrototypePush(this.diagnostics, message); } + log(message, data) { + validateString(message, 'message'); + this.reporter.log(this.nesting, this.loc, message, data, + this.name, this.testId, this.parent?.testId); + } + start() { this.applyFilters(); diff --git a/lib/internal/test_runner/tests_stream.js b/lib/internal/test_runner/tests_stream.js index 8be33f90dfa0f2..7fb514fb99e2f5 100644 --- a/lib/internal/test_runner/tests_stream.js +++ b/lib/internal/test_runner/tests_stream.js @@ -139,6 +139,19 @@ class TestsStream extends Readable { }); } + log(nesting, loc, message, data, name, testId, parentId) { + this[kEmitMessage]('test:log', { + __proto__: null, + name, + nesting, + testId, + parentId, + message, + data, + ...loc, + }); + } + diagnostic(nesting, loc, message, level = 'info') { this[kEmitMessage]('test:diagnostic', { __proto__: null, diff --git a/lib/net.js b/lib/net.js index fa685769422186..d0620449c31e7b 100644 --- a/lib/net.js +++ b/lib/net.js @@ -119,9 +119,17 @@ const { ERR_SOCKET_CLOSED_BEFORE_CONNECTION, ERR_SOCKET_CONNECTION_TIMEOUT, ERR_SOCKET_HANDLE_ADOPTED, + ERR_WORKER_HANDLE_NOT_TRANSFERABLE, + ERR_WORKER_HANDLE_TRANSFER_UNSUPPORTED, }, genericNodeError, } = require('internal/errors'); +const { + markTransferMode, + kDeserialize, + kTransfer, + kTransferList, +} = require('internal/worker/js_transferable'); const { isUint8Array } = require('internal/util/types'); const { queueMicrotask } = require('internal/process/task_queues'); const { @@ -484,6 +492,9 @@ class BoundSocket { function Socket(options) { if (!(this instanceof Socket)) return new Socket(options); + // A connected TCP Socket can be moved to another thread by listing it in the + // transferList of a worker_threads postMessage() call. See [kTransfer](). + markTransferMode(this, false, true); if (options?.objectMode) { throw new ERR_INVALID_ARG_VALUE( 'options.objectMode', @@ -1501,6 +1512,58 @@ Socket.prototype[kReinitializeHandle] = function reinitializeHandle(handle) { initSocketHandle(this); }; +// A Socket can be transferred to another thread only while it is a freshly +// accepted/created TCP connection: still attached to a live handle, not +// connecting or destroyed, and with no data already buffered in either +// direction (which would otherwise be lost on the sending side). +function assertTransferableSocket(socket) { + if (isWindows) { + throw new ERR_WORKER_HANDLE_TRANSFER_UNSUPPORTED('net.Socket'); + } + const handle = socket._handle; + if (handle == null || !(handle instanceof TCP) || + socket.destroyed || socket.connecting || + socket.bytesRead > 0 || socket.bytesWritten > 0 || + socket.readableLength > 0 || socket.writableLength > 0 || + socket.readableEncoding != null) { + throw new ERR_WORKER_HANDLE_NOT_TRANSFERABLE('net.Socket'); + } +} + +Socket.prototype[kTransferList] = function() { + assertTransferableSocket(this); + return [this._handle]; +}; + +Socket.prototype[kTransfer] = function() { + assertTransferableSocket(this); + const handle = this._handle; + const data = { + handle, + allowHalfOpen: this.allowHalfOpen, + }; + // Detach the handle from this source socket; the messaging layer takes + // ownership of it via TCPWrap::TransferForMessaging(). Destroy the source so + // any further use on the sending side fails cleanly (the socket is now owned + // by the receiving thread) instead of silently dropping data. + this._handle = null; + this.destroy(); + return { + data, + deserializeInfo: 'net:Socket', + }; +}; + +Socket.prototype[kDeserialize] = function(data) { + const handle = data?.handle; + if (handle == null || !(handle instanceof TCP)) { + throw new ERR_WORKER_HANDLE_NOT_TRANSFERABLE('net.Socket'); + } + this.allowHalfOpen = Boolean(data.allowHalfOpen); + this[kReinitializeHandle](handle); + this.readable = this.writable = true; +}; + function socketToDnsFamily(family) { switch (family) { case 'IPv4': @@ -1993,6 +2056,10 @@ function Server(options, connectionListener) { EventEmitter.call(this); + // A listening TCP Server can be moved to another thread by listing it in the + // transferList of a worker_threads postMessage() call. See [kTransfer](). + markTransferMode(this, false, true); + if (typeof options === 'function') { connectionListener = options; options = kEmptyObject; @@ -2199,6 +2266,61 @@ function setupListenHandle(address, port, addressType, backlog, fd, flags) { Server.prototype._listen2 = setupListenHandle; // legacy alias +// A listening TCP Server can be transferred to another thread, which moves the +// underlying listening socket (and its pending accept queue) to that thread's +// event loop. Only a server bound to a live TCP handle can be transferred. +function assertTransferableServer(server) { + if (isWindows) { + throw new ERR_WORKER_HANDLE_TRANSFER_UNSUPPORTED('net.Server'); + } + if (server._handle == null || !(server._handle instanceof TCP)) { + throw new ERR_WORKER_HANDLE_NOT_TRANSFERABLE('net.Server'); + } +} + +Server.prototype[kTransferList] = function() { + assertTransferableServer(this); + return [this._handle]; +}; + +Server.prototype[kTransfer] = function() { + assertTransferableServer(this); + const handle = this._handle; + const data = { + handle, + // Construction-time options that govern accepted sockets, so the receiving + // server reproduces the same behaviour. + allowHalfOpen: this.allowHalfOpen, + pauseOnConnect: this.pauseOnConnect, + noDelay: this.noDelay, + keepAlive: this.keepAlive, + keepAliveInitialDelay: this.keepAliveInitialDelay * 1000, + highWaterMark: this.highWaterMark, + }; + // Detach so the source server no longer references the handle being moved. + this._handle = null; + return { + data, + deserializeInfo: 'net:Server', + }; +}; + +Server.prototype[kDeserialize] = function(data) { + const { handle, ...options } = data ?? {}; + if (handle == null || !(handle instanceof TCP)) { + throw new ERR_WORKER_HANDLE_NOT_TRANSFERABLE('net.Server'); + } + this.allowHalfOpen = Boolean(options.allowHalfOpen); + this.pauseOnConnect = Boolean(options.pauseOnConnect); + this.noDelay = Boolean(options.noDelay); + this.keepAlive = Boolean(options.keepAlive); + this.keepAliveInitialDelay = ~~(options.keepAliveInitialDelay / 1000); + this.highWaterMark = options.highWaterMark ?? getDefaultHighWaterMark(); + // Adopt the transferred listening handle (mirrors the child_process server + // hand-off path), which re-arms accept() on this thread's event loop. + this.listen(handle); +}; + function emitErrorNT(self, err) { self.emit('error', err); } diff --git a/src/tcp_wrap.cc b/src/tcp_wrap.cc index 1abf0d6d83e349..dc3cd6553d4f9b 100644 --- a/src/tcp_wrap.cc +++ b/src/tcp_wrap.cc @@ -43,6 +43,7 @@ #include #include #include +#include // dup(), close() #endif namespace node { @@ -379,6 +380,87 @@ void TCPWrap::Open(const FunctionCallbackInfo& args) { args.GetReturnValue().Set(err); } +BaseObject::TransferMode TCPWrap::GetTransferMode() const { +#ifdef _WIN32 + // Re-adopting a socket into another thread's event loop requires + // re-associating it with that loop's IOCP, which needs same-process + // WSADuplicateSocket support that is not wired up yet. The JS net layer + // throws a clearer error before reaching here; this is the backstop for the + // low-level `socket._handle` transfer path. + return TransferMode::kDisallowCloneAndTransfer; +#else + // Only a live handle that is not already being torn down can be transferred. + // Higher-level guards (no buffered reads, no pending writes) are enforced by + // the JS net.Socket/net.Server layer before a handle reaches here. + if (!HandleWrap::IsAlive(this) || IsHandleClosing()) + return TransferMode::kDisallowCloneAndTransfer; + return TransferMode::kTransferable; +#endif +} + +std::unique_ptr TCPWrap::TransferForMessaging() { +#ifdef _WIN32 + return {}; +#else + CHECK_NE(GetTransferMode(), TransferMode::kDisallowCloneAndTransfer); + + uv_os_fd_t fd; + if (uv_fileno(reinterpret_cast(&handle_), &fd) != 0) return {}; + + // dup() the descriptor so the receiving event loop owns an independent + // reference to the same socket. We then close the source handle, which + // renders it unusable on this side (true transfer semantics) while the dup + // keeps the underlying socket alive for the destination thread. + int dup_fd = dup(fd); + if (dup_fd < 0) return {}; + + SocketType type = + provider_type() == ProviderType::PROVIDER_TCPSERVERWRAP ? SERVER : SOCKET; + + // Stop watching the fd and tear down the source handle. + Close(); + + return std::make_unique(dup_fd, type); +#endif +} + +TCPWrap::TransferData::~TransferData() { + // Only reached if the message was never delivered (e.g. the destination port + // closed in flight); close the dup'd fd so it is not leaked. + if (fd_ >= 0) { + uv_fs_t req; + CHECK_EQ(0, uv_fs_close(nullptr, &req, fd_, nullptr)); + uv_fs_req_cleanup(&req); + } +} + +BaseObjectPtr TCPWrap::TransferData::Deserialize( + Environment* env, + Local context, + std::unique_ptr self) { + // Construct a fresh TCPWrap in the receiving Environment. We cannot use + // TCPWrap::Instantiate() here because it requires a parent AsyncWrap to + // establish the async_hooks trigger id, and a deserialized handle has none. + if (env->tcp_constructor_template().IsEmpty()) return {}; + Local constructor; + if (!env->tcp_constructor_template()->GetFunction(context).ToLocal( + &constructor)) { + return {}; + } + Local type_arg = Int32::New(env->isolate(), type_); + Local obj; + if (!constructor->NewInstance(context, 1, &type_arg).ToLocal(&obj)) return {}; + + TCPWrap* wrap = BaseObject::Unwrap(obj); + if (wrap == nullptr) return {}; + + if (uv_tcp_open(&wrap->handle_, fd_) != 0) return {}; + + wrap->set_fd(fd_); + fd_ = -1; // Ownership has been handed to the new handle. + return BaseObjectPtr(wrap); +} + template void TCPWrap::Bind(const FunctionCallbackInfo& args, int family, diff --git a/src/tcp_wrap.h b/src/tcp_wrap.h index 11332f2bf40a4d..d1bcb528157479 100644 --- a/src/tcp_wrap.h +++ b/src/tcp_wrap.h @@ -26,6 +26,7 @@ #include "async_wrap.h" #include "connection_wrap.h" +#include "node_messaging.h" namespace node { @@ -61,9 +62,36 @@ class TCPWrap : public ConnectionWrap { } } + // Transfer the underlying socket to another thread via .postMessage(). Within + // a single process all threads share the same file descriptor table, so the + // transfer dup()s the fd and re-adopts it (uv_tcp_open) in the receiving + // event loop. This is the building block for distributing listening sockets + // and accepted connections across worker_threads. + BaseObject::TransferMode GetTransferMode() const override; + std::unique_ptr TransferForMessaging() override; + private: typedef uv_tcp_t HandleType; + class TransferData : public worker::TransferData { + public: + explicit TransferData(int fd, SocketType type) : fd_(fd), type_(type) {} + ~TransferData() override; + + BaseObjectPtr Deserialize( + Environment* env, + v8::Local context, + std::unique_ptr self) override; + + SET_NO_MEMORY_INFO() + SET_MEMORY_INFO_NAME(TCPWrapTransferData) + SET_SELF_SIZE(TransferData) + + private: + int fd_; + SocketType type_; + }; + template friend void GetSockOrPeerName(const v8::FunctionCallbackInfo&); diff --git a/test/client-proxy/test-http-proxy-perf-hooks.mjs b/test/client-proxy/test-http-proxy-perf-hooks.mjs new file mode 100644 index 00000000000000..4469bbf24831ce --- /dev/null +++ b/test/client-proxy/test-http-proxy-perf-hooks.mjs @@ -0,0 +1,63 @@ +// This tests that when a request path is rewritten to absolute-form for +// proxying, the perf_hooks HTTP entries report it as the URL as-is, instead +// of appending it to the protocol and authority again. +// Refs: https://github.com/nodejs/node/issues/59625 +import * as common from '../common/index.mjs'; +import assert from 'node:assert'; +import { once } from 'events'; +import http from 'node:http'; +import { PerformanceObserver } from 'node:perf_hooks'; +import { createProxyServer } from '../common/proxy-server.js'; + +const entries = []; +const obs = new PerformanceObserver(common.mustCallAtLeast((items) => { + entries.push(...items.getEntries()); +})); +obs.observe({ type: 'http' }); + +// Start a server to process the final request. +const server = http.createServer(common.mustCall((req, res) => { + res.end('Hello world'); +})); +server.on('error', common.mustNotCall((err) => { console.error('Server error', err); })); +server.listen(0); +await once(server, 'listening'); + +// Start a minimal proxy server. +const { proxy, logs } = createProxyServer(); +proxy.listen(0); +await once(proxy, 'listening'); + +const requestUrl = `http://localhost:${server.address().port}/test`; +const agent = new http.Agent({ + proxyEnv: { + HTTP_PROXY: `http://localhost:${proxy.address().port}`, + }, +}); + +const res = await new Promise((resolve, reject) => { + http.request(requestUrl, { agent }, resolve).on('error', reject).end(); +}); +res.resume(); +await once(res, 'end'); + +// Verify that the request went through the proxy. +assert.strictEqual(logs.length, 1); +assert.strictEqual(logs[0].url, requestUrl); + +proxy.close(); +server.close(); + +process.on('exit', () => { + // Two HttpClient entries are expected: one for the proxied request, one for + // the request the proxy makes to forward it. Both should report the full + // URL, including the port. + const clientUrls = entries.filter((entry) => entry.name === 'HttpClient') + .map((entry) => entry.detail.req.url); + assert.deepStrictEqual(clientUrls, [requestUrl, requestUrl]); + // Two HttpRequest entries are expected: the proxy server receives the + // request target in absolute-form, the final server in origin-form. + const requestUrls = entries.filter((entry) => entry.name === 'HttpRequest') + .map((entry) => entry.detail.req.url).sort(); + assert.deepStrictEqual(requestUrls, ['/test', requestUrl].sort()); +}); diff --git a/test/fixtures/test-runner/execution-ordered-bypass/fast-fail.mjs b/test/fixtures/test-runner/execution-ordered-bypass/fast-fail.mjs index 74b77682b6821d..6c99569e11533c 100644 --- a/test/fixtures/test-runner/execution-ordered-bypass/fast-fail.mjs +++ b/test/fixtures/test-runner/execution-ordered-bypass/fast-fail.mjs @@ -1,6 +1,7 @@ import { test } from 'node:test'; import assert from 'node:assert'; -test('fast-fail', () => { +test('fast-fail', (t) => { + t.log('live'); assert.fail('fast'); }); diff --git a/test/fixtures/test-runner/log-events/basic.mjs b/test/fixtures/test-runner/log-events/basic.mjs new file mode 100644 index 00000000000000..f19a951b982980 --- /dev/null +++ b/test/fixtures/test-runner/log-events/basic.mjs @@ -0,0 +1,19 @@ +import { suite, test } from 'node:test'; +import { setTimeout } from 'node:timers/promises'; + +suite('my suite', (s) => { + s.log('suite message'); + test('in suite', () => {}); +}); + +test('parent', { concurrency: 2 }, async (t) => { + await Promise.all([ + t.test('slow', async () => { + await setTimeout(200); + }), + t.test('logger', (t) => { + t.log('hello', { foo: 1 }); + t.log('warned', { level: 'warn', attempt: 2 }); + }), + ]); +}); diff --git a/test/parallel/test-blob.js b/test/parallel/test-blob.js index 9ec2a3c04079a7..cbe7a42ada467f 100644 --- a/test/parallel/test-blob.js +++ b/test/parallel/test-blob.js @@ -384,7 +384,9 @@ assert.throws(() => new Blob({}), { assert.strictEqual(value.byteLength, 5); assert(!done); setTimeout(common.mustCall(() => { - assert.strictEqual(stream[kState].controller.desiredSize, 0); + // Same setImmediate() pull vs. timer race as the non-BYOB case above: + // the stream may already have pulled the next 'hello' (5 bytes), hence -5. + assert([0, -5].includes(stream[kState].controller.desiredSize)); }), 0); })().then(common.mustCall()); diff --git a/test/parallel/test-http-client-complete-response-reset.js b/test/parallel/test-http-client-complete-response-reset.js new file mode 100644 index 00000000000000..aa9337dbcc5206 --- /dev/null +++ b/test/parallel/test-http-client-complete-response-reset.js @@ -0,0 +1,55 @@ +'use strict'; + +const common = require('../common'); +const assert = require('assert'); +const http = require('http'); + +const BODY = Buffer.alloc(2 * 1024 * 1024); + +const server = http.createServer(common.mustCall((request, response) => { + let received = 0; + + request.on('data', function onData(chunk) { + received += chunk.length; + request.off('data', onData); + request.destroy(); + }); + + response.writeHead(413, { 'content-length': 0 }); + response.end(); + + request.on('end', common.mustNotCall()); + request.on('close', common.mustCall(() => { + assert.strictEqual(request.complete, false); + assert.ok(received > 0); + })); +})); + +server.listen(0, common.mustCall(() => { + let response; + const req = http.request({ + method: 'POST', + port: server.address().port, + headers: { + 'content-type': 'application/json', + 'content-length': BODY.length, + }, + }, common.mustCall((res) => { + response = res; + assert.strictEqual(res.statusCode, 413); + // Deliberately do not consume the response body. A response that has + // already been received should not be followed by a late ClientRequest + // socket error. + req.write(BODY); + req.end(); + })); + + req.on('error', common.mustNotCall()); + req.on('close', common.mustCall(() => { + assert(response); + assert.strictEqual(response.complete, true); + server.close(); + })); + + req.flushHeaders(); +})); diff --git a/test/parallel/test-http-perf_hooks.js b/test/parallel/test-http-perf_hooks.js index 9acf54c5617bd8..674bb591f0cbf2 100644 --- a/test/parallel/test-http-perf_hooks.js +++ b/test/parallel/test-http-perf_hooks.js @@ -36,16 +36,18 @@ const server = http.Server(common.mustCall((req, res) => { })); }, 2)); +let port; server.listen(0, common.mustCall(async () => { + port = server.address().port; await Promise.all([ makeRequest({ - port: server.address().port, + port, path: '/', method: 'POST', data: expected }), makeRequest({ - port: server.address().port, + port, path: '/', method: 'POST', data: expected @@ -63,14 +65,17 @@ process.on('exit', () => { assert.strictEqual(typeof entry.duration, 'number'); if (entry.name === 'HttpClient') { numberOfHttpClients++; + // The reported URL must include the port when it is non-default. + // Refs: https://github.com/nodejs/node/issues/59625 + assert.strictEqual(entry.detail.req.url, `http://localhost:${port}/`); } else if (entry.name === 'HttpRequest') { numberOfHttpRequests++; + assert.strictEqual(entry.detail.req.url, '/'); } - assert.strictEqual(typeof entry.detail.req.method, 'string'); - assert.strictEqual(typeof entry.detail.req.url, 'string'); + assert.strictEqual(entry.detail.req.method, 'POST'); assert.strictEqual(typeof entry.detail.req.headers, 'object'); - assert.strictEqual(typeof entry.detail.res.statusCode, 'number'); - assert.strictEqual(typeof entry.detail.res.statusMessage, 'string'); + assert.strictEqual(entry.detail.res.statusCode, 200); + assert.strictEqual(entry.detail.res.statusMessage, 'OK'); assert.strictEqual(typeof entry.detail.res.headers, 'object'); } assert.strictEqual(numberOfHttpClients, 2); diff --git a/test/parallel/test-net-server-transfer-worker.js b/test/parallel/test-net-server-transfer-worker.js new file mode 100644 index 00000000000000..6a8fc6a770b5c6 --- /dev/null +++ b/test/parallel/test-net-server-transfer-worker.js @@ -0,0 +1,60 @@ +'use strict'; + +// This test verifies that a listening net.Server can be transferred to a worker +// thread via worker_threads postMessage()'s transferList. The parent thread +// binds and listens, then hands the server off; the worker accepts connections and +// responds, proving the listening socket (and its accept queue) moved loops. + +const common = require('../common'); + +if (common.isWindows) { + common.skip('transferring TCP handles between threads is not supported on ' + + 'Windows yet'); +} + +const assert = require('assert'); +const net = require('net'); +const { + Worker, + parentPort, + threadId, + workerData, +} = require('worker_threads'); + +if (workerData?.role === 'server') { + parentPort.on('message', common.mustCall(({ server }) => { + assert.ok(server instanceof net.Server); + server.on('connection', (socket) => { + socket.end(`served-by:${threadId}`); + }); + })); + return; +} + +const worker = new Worker(__filename, { workerData: { role: 'server' } }); + +const server = net.createServer(); + +server.listen(0, common.mustCall(() => { + const { port } = server.address(); + + // Move the listening server to the worker thread. + worker.postMessage({ server }, [server]); + assert.strictEqual(server._handle, null); + + const N = 3; + let done = 0; + for (let i = 0; i < N; i++) { + const client = net.connect(port); + client.setEncoding('utf8'); + let response = ''; + client.on('data', (chunk) => { response += chunk; }); + client.on('end', common.mustCall(() => { + assert.match(response, /^served-by:\d+$/); + assert.notStrictEqual(response, `served-by:${threadId}`); + if (++done === N) { + worker.terminate(); + } + })); + } +})); diff --git a/test/parallel/test-net-socket-transfer-worker-http.js b/test/parallel/test-net-socket-transfer-worker-http.js new file mode 100644 index 00000000000000..23a3ec1790c808 --- /dev/null +++ b/test/parallel/test-net-socket-transfer-worker-http.js @@ -0,0 +1,80 @@ +'use strict'; + +// This test verifies HTTP load balancing on top of transferred TCP sockets: +// the parent thread accepts connections and distributes each one round-robin to +// a pool of worker threads, where an http.Server handles the request. It exercises +// the intended use case for transferring net.Socket handles across threads. + +const common = require('../common'); + +if (common.isWindows) { + common.skip('transferring TCP handles between threads is not supported on ' + + 'Windows yet'); +} + +const assert = require('assert'); +const net = require('net'); +const http = require('http'); +const { + Worker, + parentPort, + threadId, + workerData, +} = require('worker_threads'); + +const POOL = 4; +const REQ_PER_WORKER = 5; +const TOTAL = POOL * REQ_PER_WORKER; + +if (workerData?.role === 'http-server') { + // Each worker serves HTTP over connections transferred from the parent thread. + const server = http.createServer((req, res) => { + res.end(`thread:${threadId}`); + }); + parentPort.on('message', common.mustCall(({ socket }) => { + assert.ok(socket instanceof net.Socket); + server.emit('connection', socket); + }, REQ_PER_WORKER)); + return; +} + +const workers = []; +for (let i = 0; i < POOL; i++) + workers.push(new Worker(__filename, { workerData: { role: 'http-server' } })); + +let next = 0; +const front = net.createServer((socket) => { + // Distribute each freshly accepted connection round-robin to a worker. + workers[next++ % POOL].postMessage({ socket }, [socket]); +}); + +front.listen(0, common.mustCall(() => { + const { port } = front.address(); + const counts = { __proto__: null }; + let done = 0; + + for (let i = 0; i < TOTAL; i++) { + // agent: false forces a fresh connection per request, so each request maps + // to exactly one round-robin hand-off. + http.get({ port, agent: false }, common.mustCall((res) => { + let body = ''; + res.setEncoding('utf8'); + res.on('data', (chunk) => { body += chunk; }); + res.on('end', common.mustCall(() => { + assert.match(body, /^thread:\d+$/); + counts[body] = (counts[body] || 0) + 1; + + if (++done === TOTAL) { + // Every worker handled the same number of requests. + assert.strictEqual(Object.keys(counts).length, POOL); + for (const key of Object.keys(counts)) + assert.strictEqual(counts[key], REQ_PER_WORKER); + + front.close(); + for (const worker of workers) + worker.terminate(); + } + })); + })); + } +})); diff --git a/test/parallel/test-net-socket-transfer-worker.js b/test/parallel/test-net-socket-transfer-worker.js new file mode 100644 index 00000000000000..fe512ee3723453 --- /dev/null +++ b/test/parallel/test-net-socket-transfer-worker.js @@ -0,0 +1,65 @@ +'use strict'; + +// This test verifies that an accepted net.Socket connection can be transferred +// to a worker thread via worker_threads postMessage()'s transferList, and that +// the worker can read from and write to it. The parent thread accepts the +// connection and hands it off; the worker echoes data back. + +const common = require('../common'); + +if (common.isWindows) { + common.skip('transferring TCP handles between threads is not supported on ' + + 'Windows yet'); +} + +const assert = require('assert'); +const net = require('net'); +const { + Worker, + parentPort, + workerData, +} = require('worker_threads'); + +if (workerData?.role === 'socket') { + // Worker side: receive the transferred connection and echo everything back. + parentPort.on('message', common.mustCall(({ socket }) => { + assert.ok(socket instanceof net.Socket); + socket.setEncoding('utf8'); + socket.on('data', common.mustCall((chunk) => { + socket.end(`echo:${chunk}`); + })); + })); + return; +} + +const worker = new Worker(__filename, { workerData: { role: 'socket' } }); + +const server = net.createServer(common.mustCall((socket) => { + // Hand the freshly accepted connection off to the worker. It must not be + // read from in this thread before transferring. + worker.postMessage({ socket }, [socket]); + + // The source socket is now owned by the worker: it is detached and destroyed + // on this side, so further use fails cleanly instead of dropping data. + assert.strictEqual(socket._handle, null); + assert.strictEqual(socket.destroyed, true); + assert.strictEqual(socket.writable, false); + socket.write('lost', common.mustCall((err) => { + assert.strictEqual(err.code, 'ERR_STREAM_DESTROYED'); + })); + + server.close(); +})); + +server.listen(0, common.mustCall(() => { + const client = net.connect(server.address().port, common.mustCall(() => { + client.write('hello'); + })); + client.setEncoding('utf8'); + let response = ''; + client.on('data', (chunk) => { response += chunk; }); + client.on('end', common.mustCall(() => { + assert.strictEqual(response, 'echo:hello'); + worker.terminate(); + })); +})); diff --git a/test/parallel/test-net-transfer-guards.js b/test/parallel/test-net-transfer-guards.js new file mode 100644 index 00000000000000..4724913b9f3352 --- /dev/null +++ b/test/parallel/test-net-transfer-guards.js @@ -0,0 +1,52 @@ +'use strict'; + +// This test verifies the guards that reject transferring a net.Socket or +// net.Server that is not in a clean, movable state. Serialization is triggered +// with a MessageChannel, so no worker thread is required. + +const common = require('../common'); + +if (common.isWindows) { + common.skip('transferring TCP handles between threads is not supported on ' + + 'Windows yet'); +} + +const assert = require('assert'); +const net = require('net'); +const { MessageChannel } = require('worker_threads'); + +const { port1 } = new MessageChannel(); + +// A Server that is not listening (no handle) cannot be transferred. +{ + const server = net.createServer(); + assert.throws(() => port1.postMessage({ server }, [server]), { + code: 'ERR_WORKER_HANDLE_NOT_TRANSFERABLE', + }); +} + +// A Socket that has already consumed data cannot be transferred, because that +// buffered data would be lost on the sending side. +{ + const server = net.createServer(common.mustCall((socket) => { + socket.on('data', common.mustCall(() => { + assert.throws(() => port1.postMessage({ socket }, [socket]), { + code: 'ERR_WORKER_HANDLE_NOT_TRANSFERABLE', + }); + // The socket is still usable after a rejected transfer. + assert.ok(socket._handle != null); + socket.destroy(); + server.close(); + port1.close(); + })); + })); + + server.listen(0, common.mustCall(() => { + const client = net.connect(server.address().port, common.mustCall(() => { + client.end('data'); + })); + // The server destroys the socket after the rejected transfer, which may + // surface as a reset on the client; swallow it either way. + client.on('error', () => {}); + })); +} diff --git a/test/parallel/test-runner-execution-ordered-bypass.mjs b/test/parallel/test-runner-execution-ordered-bypass.mjs index 85b468cae2a7e1..ac1c97ee007d68 100644 --- a/test/parallel/test-runner-execution-ordered-bypass.mjs +++ b/test/parallel/test-runner-execution-ordered-bypass.mjs @@ -35,6 +35,10 @@ test('execution-ordered events bypass FileTest declaration-order buffer', async } }); + stream.on('test:log', (data) => { + events.push(`log:${data.message}`); + }); + // eslint-disable-next-line no-unused-vars for await (const _ of stream); @@ -56,4 +60,12 @@ test('execution-ordered events bypass FileTest declaration-order buffer', async failFast > completeSlow, `test:fail for fast-fail should arrive after test:complete for slow; events=${events.join(', ')}`, ); + + // test:log is execution-ordered, so it must bypass the buffer too. + const logFast = events.indexOf('log:live'); + assert.notStrictEqual(logFast, -1); + assert.ok( + logFast < completeSlow, + `test:log for fast-fail should arrive before slow's test:complete; events=${events.join(', ')}`, + ); }); diff --git a/test/parallel/test-runner-log.mjs b/test/parallel/test-runner-log.mjs new file mode 100644 index 00000000000000..8e767e0f903c8e --- /dev/null +++ b/test/parallel/test-runner-log.mjs @@ -0,0 +1,80 @@ +// Flags: --no-warnings +import '../common/index.mjs'; +import * as fixtures from '../common/fixtures.mjs'; +import assert from 'node:assert'; +import { spawnSync } from 'node:child_process'; +import { test, run } from 'node:test'; + +const fixture = fixtures.path('test-runner', 'log-events', 'basic.mjs'); + +test('t.log() validates its arguments', (t) => { + assert.throws(() => t.log(123), { code: 'ERR_INVALID_ARG_TYPE' }); +}); + +test('built-in reporters render test:log like diagnostics', () => { + const tap = spawnSync(process.execPath, ['--test', '--test-reporter=tap', fixture]); + assert.match(tap.stdout.toString(), /# hello/); + assert.match(tap.stdout.toString(), /# suite message/); + + const spec = spawnSync(process.execPath, ['--test', '--test-reporter=spec', fixture]); + assert.match(spec.stdout.toString(), /ℹ hello/); + assert.match(spec.stdout.toString(), /ℹ warned/); + + const junit = spawnSync(process.execPath, ['--test', '--test-reporter=junit', fixture]); + assert.match(junit.stdout.toString(), //); +}); + +test('test:log is emitted immediately with the expected payload', async () => { + const stream = run({ + files: [fixture], + isolation: 'none', + }); + + const logs = []; + const events = []; + + stream.on('test:log', (data) => { + logs.push(data); + events.push(`log:${data.message}`); + }); + + stream.on('test:pass', (data) => { + events.push(`pass:${data.name}`); + }); + + // eslint-disable-next-line no-unused-vars + for await (const _ of stream); + + assert.strictEqual(logs.length, 3); + + const suiteLog = logs.find((log) => log.message === 'suite message'); + assert.strictEqual(suiteLog.name, 'my suite'); + assert.strictEqual(suiteLog.data, undefined); + + const hello = logs.find((log) => log.message === 'hello'); + assert.strictEqual(hello.name, 'logger'); + assert.ok(!('level' in hello)); + assert.deepStrictEqual(hello.data, { foo: 1 }); + assert.strictEqual(typeof hello.nesting, 'number'); + assert.strictEqual(typeof hello.testId, 'number'); + assert.strictEqual(typeof hello.parentId, 'number'); + assert.strictEqual(typeof hello.file, 'string'); + assert.strictEqual(typeof hello.line, 'number'); + assert.strictEqual(typeof hello.column, 'number'); + + // The runner does not interpret data; conventions like a level live there. + const warned = logs.find((log) => log.message === 'warned'); + assert.ok(!('level' in warned)); + assert.deepStrictEqual(warned.data, { level: 'warn', attempt: 2 }); + + // 'logger' logs immediately while its earlier-declared sibling 'slow' is + // still running, so the log must arrive before slow's buffered test:pass. + const logIndex = events.indexOf('log:hello'); + const slowPassIndex = events.indexOf('pass:slow'); + assert.notStrictEqual(logIndex, -1); + assert.notStrictEqual(slowPassIndex, -1); + assert.ok( + logIndex < slowPassIndex, + `test:log should arrive before slow's buffered test:pass; events=${events.join(', ')}`, + ); +}); diff --git a/test/parallel/test-worker-message-port-transfer-fake-js-transferable-internal.js b/test/parallel/test-worker-message-port-transfer-fake-js-transferable-internal.js index 46895871e0a179..3d33fff65d9d52 100644 --- a/test/parallel/test-worker-message-port-transfer-fake-js-transferable-internal.js +++ b/test/parallel/test-worker-message-port-transfer-fake-js-transferable-internal.js @@ -15,10 +15,14 @@ const { once } = require('events'); const kTransfer = Object.getOwnPropertySymbols(Object.getPrototypeOf(fh)) .find((symbol) => symbol.description === 'messaging_transfer_symbol'); assert.strictEqual(typeof kTransfer, 'symbol'); + // Use a module:export pair that is not a registered transferable. net.Socket + // and net.Server are real transferables now, so they would be constructed (and + // then safely reject the bogus data); a nonexistent export keeps exercising + // the "Unknown deserialize spec" allowlist rejection. fh[kTransfer] = () => { return { data: '✨', - deserializeInfo: 'net:Socket' + deserializeInfo: 'net:NotARealTransferable' }; }; @@ -28,6 +32,7 @@ const { once } = require('events'); const [ exception ] = await once(port2, 'messageerror'); - assert.strictEqual(exception.message, 'Unknown deserialize spec net:Socket'); + assert.strictEqual(exception.message, + 'Unknown deserialize spec net:NotARealTransferable'); port2.close(); })().then(common.mustCall());