node:net: publish diagnostics_channel events (net.client.socket, net.server.socket, net.server.listen)

src/js/node/net.ts

@@ -42,6 +42,11 @@ const { kTimeout, getTimerDuration } = require("internal/timers");
 const { validateFunction, validateNumber, validateAbortSignal, validatePort, validateBoolean, validateInt32, validateString } = require("internal/validators"); // prettier-ignore
 const { isIPv4, isIPv6, isIP } = require("internal/net/isIP");
 
+const dc = require("node:diagnostics_channel");
+const netClientSocketChannel = dc.channel("net.client.socket");
+const netServerSocketChannel = dc.channel("net.server.socket");
+const netServerListen = dc.tracingChannel("net.server.listen");
+
 const ArrayPrototypeIncludes = Array.prototype.includes;
 const ArrayPrototypeJoin = Array.prototype.join;
 const ArrayPrototypePush = Array.prototype.push;
@@ -982,6 +987,11 @@ function onconnection(err, clientHandle) {
   }
 
   self.emit("connection", _socket);
+  if (netServerSocketChannel.hasSubscribers) {
+    netServerSocketChannel.publish({
+      socket: _socket,
+    });
+  }
   // the duplex implementation start paused, so we resume when pauseOnConnect is falsy
   if (!pauseOnConnect && !isTLS) {
     _socket.resume();
@@ -1593,6 +1603,13 @@ Socket.prototype.connect = function connect(...args) {
   {
     const [options, connectListener] =
       $isArray(args[0]) && args[0][normalizedArgsSymbol] ? args[0] : normalizeArgs(args);
+
+    if (netClientSocketChannel.hasSubscribers) {
+      netClientSocketChannel.publish({
+        socket: this,
+      });
+    }
+
     let connection = this[ksocket];
     let upgradeDuplex = false;
     let { port, host, path, socket, rejectUnauthorized, checkServerIdentity, session, fd, pauseOnConnect } = options;
@@ -3237,6 +3254,15 @@ Server.prototype.getConnections = function getConnections(callback) {
 
 Server.prototype.listen = function listen(port, hostname, onListen) {
   const argsLength = arguments.length;
+
+  if (this._handle) {
+    throw $ERR_SERVER_ALREADY_LISTEN();
+  }
+
+  if (netServerListen.asyncStart.hasSubscribers) {
+    netServerListen.asyncStart.publish({ server: this, options: normalizeArgs(arguments)[0] });
+  }
+
   if (typeof port === "string") {
     const numPort = Number(port);
     if (!Number.isNaN(numPort)) port = numPort;
@@ -3385,10 +3411,6 @@ Server.prototype.listen = function listen(port, hostname, onListen) {
     throw $ERR_SOCKET_BAD_PORT(`options.port should be >= 0 and < 65536. Received type number: (${port})`);
   }
 
-  if (this._handle) {
-    throw $ERR_SERVER_ALREADY_LISTEN();
-  }
-
   if (onListen != null) {
     this.once("listening", onListen);
   }
@@ -3433,7 +3455,11 @@ Server.prototype.listen = function listen(port, hostname, onListen) {
     );
   } catch (err) {
     const isUnix = path != null;
-    setTimeout(emitErrorNextTick, 1, this, formatListenError(err, isUnix ? path : hostname, isUnix ? undefined : port));
+    const error = formatListenError(err, isUnix ? path : hostname, isUnix ? undefined : port);
+    if (netServerListen.error.hasSubscribers) {
+      netServerListen.error.publish({ server: this, error });
+    }
+    setTimeout(emitErrorNextTick, 1, this, error);
   }
   return this;
 };
@@ -3531,6 +3557,10 @@ Server.prototype[kRealListen] = function (
     }
   }
 
+  if (netServerListen.asyncEnd.hasSubscribers) {
+    netServerListen.asyncEnd.publish({ server: this });
+  }
+
   // Unref the handle if the server was unref'ed prior to listening
   if (this._unref) this.unref();
 

test/js/node/net/node-net.test.ts

@@ -1,7 +1,16 @@
 import { Socket as _BunSocket, TCPSocketListener } from "bun";
 import { heapStats } from "bun:jsc";
 import { describe, expect, it } from "bun:test";
-import { bunEnv, bunExe, expectMaxObjectTypeCount, isASAN, isDebug, isWindows, tmpdirSync } from "harness";
+import {
+  bunEnv,
+  bunExe,
+  expectMaxObjectTypeCount,
+  isASAN,
+  isDebug,
+  isWindows,
+  normalizeBunSnapshot,
+  tmpdirSync,
+} from "harness";
 import { randomUUID } from "node:crypto";
 import fs from "node:fs";
 import {
@@ -1064,3 +1073,140 @@ it.skipIf(isWindows)("connect({ localPort }) succeeds when the local port has TI
     target.close();
   }
 });
+
+describe("diagnostics_channel", () => {
+  // https://nodejs.org/api/diagnostics_channel.html#built-in-channels
+  it("publishes net.client.socket, net.server.socket and net.server.listen tracing channels", async () => {
+    const fixture = `
+      const dc = require("node:diagnostics_channel");
+      const net = require("node:net");
+      const events = [];
+      dc.subscribe("net.client.socket", ({ socket }) => {
+        events.push("net.client.socket: " + (socket instanceof net.Socket));
+      });
+      dc.subscribe("net.server.socket", ({ socket }) => {
+        events.push("net.server.socket: " + (socket instanceof net.Socket));
+      });
+      dc.tracingChannel("net.server.listen").subscribe({
+        asyncStart({ server, options }) {
+          events.push("listen asyncStart: " + (server instanceof net.Server) + " " + JSON.stringify(options));
+        },
+        asyncEnd({ server }) {
+          events.push("listen asyncEnd: " + (server instanceof net.Server));
+        },
+        error({ server, error }) {
+          events.push("listen error: " + (server instanceof net.Server) + " " + error?.code);
+        },
+      });
+
+      const server = net.createServer(s => s.end());
+      server.listen({ port: 0, host: "127.0.0.1", customOption: true }, () => {
+        const port = server.address().port;
+        let closed = 0;
+        const done = () => {
+          if (++closed === 3) {
+            server.close(() => {
+              for (const e of events) console.log(e);
+            });
+          }
+        };
+        // All documented entry points that publish net.client.socket.
+        net.connect(port, "127.0.0.1").on("close", done).on("error", done);
+        net.createConnection(port, "127.0.0.1").on("close", done).on("error", done);
+        new net.Socket().connect(port, "127.0.0.1").on("close", done).on("error", done);
+      });
+      server.on("error", err => { throw err; });
+    `;
+    await using proc = Bun.spawn({
+      cmd: [bunExe(), "-e", fixture],
+      env: bunEnv,
+      stderr: "pipe",
+    });
+    const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+    expect(stderr).toBe("");
+    expect(normalizeBunSnapshot(stdout)).toMatchInlineSnapshot(`
+      "listen asyncStart: true {"port":0,"host":"127.0.0.1","customOption":true}
+      listen asyncEnd: true
+      net.client.socket: true
+      net.client.socket: true
+      net.client.socket: true
+      net.server.socket: true
+      net.server.socket: true
+      net.server.socket: true"
+    `);
+    expect(exitCode).toBe(0);
+  });
+
+  it("publishes tracing:net.server.listen:error when listen fails", async () => {
+    const fixture = `
+      const dc = require("node:diagnostics_channel");
+      const net = require("node:net");
+      dc.tracingChannel("net.server.listen").subscribe({
+        asyncStart({ server, options }) {
+          console.log("asyncStart " + (server instanceof net.Server) + " " + (typeof options.port === "number"));
+        },
+        asyncEnd() {
+          console.log("asyncEnd");
+        },
+        error({ server, error }) {
+          console.log("error " + (server instanceof net.Server) + " " + error?.code);
+        },
+      });
+      const first = net.createServer();
+      first.listen(0, "127.0.0.1", () => {
+        const second = net.createServer();
+        second.on("error", () => {
+          first.close();
+          second.close();
+        });
+        second.listen({ port: first.address().port, host: "127.0.0.1" });
+      });
+    `;
+    await using proc = Bun.spawn({
+      cmd: [bunExe(), "-e", fixture],
+      env: bunEnv,
+      stderr: "pipe",
+    });
+    const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+    expect(stderr).toBe("");
+    expect(normalizeBunSnapshot(stdout)).toMatchInlineSnapshot(`
+      "asyncStart true true
+      asyncEnd
+      asyncStart true true
+      error true EADDRINUSE"
+    `);
+    expect(exitCode).toBe(0);
+  });
+
+  it("publishes asyncStart before option validation throws, but not on ERR_SERVER_ALREADY_LISTEN", async () => {
+    const fixture = `
+      const dc = require("node:diagnostics_channel");
+      const net = require("node:net");
+      let starts = 0;
+      dc.subscribe("tracing:net.server.listen:asyncStart", ({ options }) => {
+        starts++;
+        console.log("asyncStart", JSON.stringify(options));
+      });
+      try { net.createServer().listen({ port: -1 }); } catch (e) { console.log("threw", e.code, "starts=" + starts); }
+      const s = net.createServer();
+      s.listen(0, "127.0.0.1", () => {
+        try { s.listen(0); } catch (e) { console.log("threw", e.code, "starts=" + starts); }
+        s.close();
+      });
+    `;
+    await using proc = Bun.spawn({
+      cmd: [bunExe(), "-e", fixture],
+      env: bunEnv,
+      stderr: "pipe",
+    });
+    const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+    expect(stderr).toBe("");
+    expect(normalizeBunSnapshot(stdout)).toMatchInlineSnapshot(`
+      "asyncStart {"port":-1}
+      threw ERR_SOCKET_BAD_PORT starts=1
+      asyncStart {"port":0,"host":"127.0.0.1"}
+      threw ERR_SERVER_ALREADY_LISTEN starts=2"
+    `);
+    expect(exitCode).toBe(0);
+  });
+});

test/js/node/test/parallel/test-diagnostics-channel-net.js

@@ -0,0 +1,101 @@
+'use strict';
+const common = require('../common');
+const Countdown = require('../common/countdown');
+const assert = require('assert');
+const net = require('net');
+const dc = require('diagnostics_channel');
+
+const isNetSocket = (socket) => socket instanceof net.Socket;
+const isNetServer = (server) => server instanceof net.Server;
+
+function testDiagnosticChannel(subscribers, test, after) {
+  dc.tracingChannel('net.server.listen').subscribe(subscribers);
+
+  test(common.mustCall(() => {
+    dc.tracingChannel('net.server.listen').unsubscribe(subscribers);
+    after?.();
+  }));
+}
+
+const testSuccessfulListen = common.mustCall(() => {
+  let cb;
+  const netClientSocketCount = 3;
+  const countdown = new Countdown(netClientSocketCount, () => {
+    server.close();
+    cb();
+  });
+  const server = net.createServer(common.mustCall((socket) => {
+    socket.destroy();
+    countdown.dec();
+  }, netClientSocketCount));
+
+  dc.subscribe('net.client.socket', common.mustCall(({ socket }) => {
+    assert.strictEqual(isNetSocket(socket), true);
+  }, netClientSocketCount));
+
+  dc.subscribe('net.server.socket', common.mustCall(({ socket }) => {
+    assert.strictEqual(isNetSocket(socket), true);
+  }, netClientSocketCount));
+
+  testDiagnosticChannel(
+    {
+      asyncStart: common.mustCall(({ server: currentServer, options }) => {
+        assert.strictEqual(isNetServer(server), true);
+        assert.strictEqual(currentServer, server);
+        assert.strictEqual(options.customOption, true);
+      }),
+      asyncEnd: common.mustCall(({ server: currentServer }) => {
+        assert.strictEqual(isNetServer(server), true);
+        assert.strictEqual(currentServer, server);
+      }),
+      error: common.mustNotCall()
+    },
+    common.mustCall((callback) => {
+      cb = callback;
+      server.listen({ port: 0, customOption: true }, () => {
+        // All supported ways of creating a net client socket connection.
+        const { port } = server.address();
+        net.connect(port);
+
+        net.createConnection(port);
+
+        new net.Socket().connect(port);
+      });
+    }),
+    testFailingListen
+  );
+});
+
+const testFailingListen = common.mustCall(() => {
+  const originalServer = net.createServer(common.mustNotCall());
+
+  originalServer.listen(common.mustCall(() => {
+    const server = net.createServer(common.mustNotCall());
+
+    testDiagnosticChannel(
+      {
+        asyncStart: common.mustCall(({ server: currentServer, options }) => {
+          assert.strictEqual(isNetServer(server), true);
+          assert.strictEqual(currentServer, server);
+          assert.strictEqual(options.customOption, true);
+        }),
+        asyncEnd: common.mustNotCall(),
+        error: common.mustCall(({ server: currentServer }) => {
+          assert.strictEqual(isNetServer(server), true);
+          assert.strictEqual(currentServer, server);
+        }),
+      },
+      common.mustCall((callback) => {
+        server.on('error', () => {});
+        server.listen({ port: originalServer.address().port, customOption: true });
+        callback();
+      }),
+      common.mustCall(() => {
+        originalServer.close();
+        server.close();
+      })
+    );
+  }));
+});
+
+testSuccessfulListen();

test/napi/node-napi-tests/test/js-native-api/test_function/test.js

@@ -31,10 +31,15 @@ assert.strictEqual(test_function.TestCall(func4, 1), 2);
 assert.strictEqual(test_function.TestName.name, 'Name');
 assert.strictEqual(test_function.TestNameShort.name, 'Name_');
 
-let tracked_function = test_function.MakeTrackedFunction(common.mustCall());
-assert(!!tracked_function);
-tracked_function = null;
-global.gc();
+// We use IIFE for the tracked_function scope instead of a block to be
+// compatible with non-V8 JS engines whose conservative stack scan may keep
+// the object alive while the creating frame is still on the stack.
+(() => {
+  let tracked_function = test_function.MakeTrackedFunction(common.mustCall());
+  assert(!!tracked_function);
+  tracked_function = null;
+})();
+for (let i = 0; i < 10; ++i) global.gc();
 
 assert.deepStrictEqual(test_function.TestCreateFunctionParameters(), {
   envIsNull: 'Invalid argument',

test/napi/node-napi-tests/test/js-native-api/test_instance_data/test.js

@@ -17,8 +17,13 @@ if (module !== require.main) {
   assert.strictEqual(test_instance_data.increment(), 42);
 
   // Test that the instance data can be accessed from a finalizer.
-  test_instance_data.objectWithFinalizer(common.mustCall());
-  global.gc();
+  // We use IIFE for the object's scope to be compatible with non-V8 JS
+  // engines whose conservative stack scan may keep the object alive while
+  // the creating frame is still on the stack.
+  (() => {
+    test_instance_data.objectWithFinalizer(common.mustCall());
+  })();
+  for (let i = 0; i < 10; ++i) global.gc();
 } else {
   // When launched as a script, run tests in either a child process or in a
   // worker thread.