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;
@@ -3389,6 +3406,10 @@ Server.prototype.listen = function listen(port, hostname, onListen) {
     throw $ERR_SERVER_ALREADY_LISTEN();
   }
 
+  if (netServerListen.asyncStart.hasSubscribers) {
+    netServerListen.asyncStart.publish({ server: this, options: normalizeArgs(arguments)[0] });
+  }
+
   if (onListen != null) {
     this.once("listening", onListen);
   }
@@ -3433,7 +3454,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 +3556,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,108 @@ 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);
+  });
+});

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();