node:net: count bytesRead on the native socket handle

packages/bun-types/bun.d.ts

@@ -6914,6 +6914,12 @@ declare module "bun" {
      */
     readonly bytesWritten: number;
 
+    /**
+     * The total number of bytes received on the socket since it was established.
+     * On a TLS socket this counts the decrypted bytes.
+     */
+    readonly bytesRead: number;
+
     /**
      * Alias for `socket.end()`. Allows the socket to be used with `using` declarations
      * for automatic resource management.

src/js/node/net.ts

@@ -138,6 +138,7 @@ const { owner_symbol } = require("internal/async_hooks").symbols;
 // native Listener reachable via accepted socket handles (see a93d2fa48e).
 const kServerSocket = Symbol("kServerSocket");
 const kBytesWritten = Symbol("kBytesWritten");
+const kBytesRead = Symbol("kBytesRead");
 const bunTLSConnectOptions = Symbol.for("::buntlsconnectoptions::");
 // tls.Server exposes its native SecureContext constructor through this key so
 // the SNI dispatch (below) can recognize a raw native context the way Node's
@@ -188,6 +189,10 @@ function onUpgradeWriteClose(callback) {
 }
 const kUpgradeAttached = Symbol("kUpgradeAttached");
 const kOnreadTail = Symbol("kOnreadTail");
+// Bytes the handle read that the onread callback has not received yet: the
+// tail plus the rest of the buffer deliver() is slicing. bytesRead leaves them
+// out, because node's handle reads at most one onread buffer per callback.
+const kOnreadUndelivered = Symbol("kOnreadUndelivered");
 const kOnreadDraining = Symbol("kOnreadDraining");
 const kOnreadBuffer = Symbol("kOnreadBuffer");
 const kOnreadPendingEnd = Symbol("kOnreadPendingEnd");
@@ -262,8 +267,13 @@ function closeAdoptedTLSRawNowNT(handle, self, isException) {
   handle.close(onSocketHandleClosed);
   setImmediate(emitCloseNT, self, isException);
 }
+function handleBytesRead(self, handle) {
+  return handle.bytesRead - (self[kOnreadUndelivered] || 0);
+}
 function detachSocket(self) {
   if (!self) self = this;
+  const handle = self._handle;
+  if (handle) self[kBytesRead] = handleBytesRead(self, handle);
   self._handle = null;
 }
 function destroyNT(self, err) {
@@ -479,7 +489,6 @@ const SocketHandlers: SocketHandler = {
     if (!self) return;
 
     self._unrefTimer();
-    self.bytesRead += buffer.length;
     pushDataToSocket(self, socket, buffer);
   },
   drain(socket) {
@@ -799,7 +808,6 @@ const ServerHandlers: SocketHandler<NetSocket> = {
     if (!self) return;
 
     self._unrefTimer();
-    self.bytesRead += buffer.length;
     pushDataToSocket(self, socket, buffer);
   },
   keylog(socket, line) {
@@ -1313,7 +1321,6 @@ const SocketHandlers2: SocketHandler<NonNullable<import("node:net").Socket["_han
     $debug("Bun.Socket data");
     const { self } = socket.data;
     self._unrefTimer();
-    self.bytesRead += buffer.length;
     pushDataToSocket(self, socket, buffer);
   },
   drain(socket) {
@@ -1608,7 +1615,7 @@ function Socket(options?) {
 
   this[khandlers] = SocketHandlers2;
   this[kDestroyOnRead] = false;
-  this.bytesRead = 0;
+  this[kBytesRead] = 0;
   this[kBytesWritten] = undefined;
   this[kclosed] = false;
   this[kended] = false;
@@ -1696,6 +1703,7 @@ function Socket(options?) {
     const self = this;
     this[kOnreadTail] = undefined;
     this[kOnreadDraining] = false;
+    this[kOnreadUndelivered] = 0;
     // Node calls the factory once at initSocketHandle time, then once after
     // every callback (stream_base_commons onStreamRead): the first delivery
     // already has a real buffer, and a non-Uint8Array result leaves the prior
@@ -1718,6 +1726,7 @@ function Socket(options?) {
       while (offset < total) {
         const dest = self[kOnreadBuffer];
         if (dest === true) {
+          self[kOnreadUndelivered] -= total - offset;
           let ret;
           try {
             ret = onreadCallback(total - offset, true);
@@ -1749,6 +1758,7 @@ function Socket(options?) {
         const n = MathMin(dest.length, total - offset);
         dest.set(buffer.subarray(offset, offset + n));
         offset += n;
+        self[kOnreadUndelivered] -= n;
         let ret;
         try {
           ret = onreadCallback(n, dest);
@@ -1777,6 +1787,7 @@ function Socket(options?) {
         const { self } = socket.data;
         if (!self) return;
         self._unrefTimer();
+        self[kOnreadUndelivered] += buffer.length;
         const tail = self[kOnreadTail];
         if (tail !== undefined) {
           self[kOnreadTail] = Buffer.concat([tail, buffer]);
@@ -1846,6 +1857,16 @@ Object.defineProperty(Socket.prototype, "_bytesDispatched", {
   },
 });
 
+// https://github.com/nodejs/node/blob/v26.3.0/lib/net.js#L896-L900: the handle
+// counts every byte it reads, so a native consumer of the socket (an h2
+// session) is counted too. The count moves to kBytesRead when the handle goes.
+Object.defineProperty(Socket.prototype, "bytesRead", {
+  get: function () {
+    const handle = this._handle;
+    return (handle ? handleBytesRead(this, handle) : this[kBytesRead]) || 0;
+  },
+});
+
 Object.defineProperty(Socket.prototype, "bytesWritten", {
   get: function () {
     let bytes = this[kBytesWritten] || 0;
@@ -1906,7 +1927,7 @@ Socket.prototype[kCloseRawConnection] = function () {
   // Only a destroy the connection's owner started counts as it closing under this socket.
   if (!connection.destroyed) connection.removeListener("close", this[kOnUpgradedClose]);
   connection.connecting = false;
-  connection._handle = null;
+  detachSocket(connection);
   connection.unref();
   connection.destroy();
 };
@@ -2223,7 +2244,7 @@ Socket.prototype._destroy = function _destroy(err, callback) {
     $debug("close handle");
     const isException = err ? true : false;
     // `bytesRead` and `kBytesWritten` should be accessible after `.destroy()`
-    // this[kBytesRead] = this._handle.bytesRead;
+    this[kBytesRead] = handleBytesRead(this, this._handle);
     this[kBytesWritten] = this._handle.bytesWritten;
 
     const currentHandle = this._handle;

src/runtime/node/node_net_binding.rs

@@ -152,6 +152,7 @@ pub(crate) fn new_detached_socket(global: &JSGlobalObject, frame: &CallFrame) ->
             server_name: JsCell::new(None),
             buffered_data_for_node_net: Default::default(),
             bytes_written: Cell::new(0),
+            bytes_read: Cell::new(0),
             native_callback: JsCell::new(NativeCallbacks::None),
             twin: JsCell::new(None),
             verify_error: JsCell::new(None),

src/runtime/socket/Listener.rs

@@ -645,6 +645,7 @@ impl Listener {
             server_name: JsCell::new(None),
             buffered_data_for_node_net: Default::default(),
             bytes_written: Cell::new(0),
+            bytes_read: Cell::new(0),
             native_callback: JsCell::new(crate::socket::NativeCallbacks::None),
             twin: JsCell::new(None),
             verify_error: JsCell::new(None),
@@ -691,6 +692,7 @@ impl Listener {
             server_name: JsCell::new(None),
             buffered_data_for_node_net: Default::default(),
             bytes_written: Cell::new(0),
+            bytes_read: Cell::new(0),
             native_callback: JsCell::new(crate::socket::NativeCallbacks::None),
             twin: JsCell::new(None),
             verify_error: JsCell::new(None),
@@ -1257,6 +1259,7 @@ impl Listener {
                             ref_pollref_on_connect: Cell::new(true),
                             buffered_data_for_node_net: Default::default(),
                             bytes_written: Cell::new(0),
+                            bytes_read: Cell::new(0),
                             native_callback: JsCell::new(crate::socket::NativeCallbacks::None),
                             twin: JsCell::new(None),
                             verify_error: JsCell::new(None),
@@ -1344,6 +1347,7 @@ impl Listener {
                             ref_pollref_on_connect: Cell::new(true),
                             buffered_data_for_node_net: Default::default(),
                             bytes_written: Cell::new(0),
+                            bytes_read: Cell::new(0),
                             native_callback: JsCell::new(crate::socket::NativeCallbacks::None),
                             twin: JsCell::new(None),
                             verify_error: JsCell::new(None),
@@ -1581,6 +1585,7 @@ fn connect_finish<const IS_SSL: bool>(
             ref_pollref_on_connect: Cell::new(true),
             buffered_data_for_node_net: Default::default(),
             bytes_written: Cell::new(0),
+            bytes_read: Cell::new(0),
             native_callback: JsCell::new(crate::socket::NativeCallbacks::None),
             twin: JsCell::new(None),
             verify_error: JsCell::new(None),

src/runtime/socket/socket_body.rs

@@ -326,6 +326,9 @@ pub struct NewSocket<const SSL: bool> {
     pub(crate) server_name: JsCell<Option<Box<[u8]>>>,
     pub(crate) buffered_data_for_node_net: JsCell<Vec<u8>>,
     pub(crate) bytes_written: Cell<u64>,
+    /// Payload bytes this wrapper delivered through `on_data` (plaintext on a
+    /// TLS view). Node's `socket.bytesRead` reads the handle's counter.
+    pub(crate) bytes_read: Cell<u64>,
 
     pub(crate) native_callback: JsCell<NativeCallbacks>,
     /// `upgradeTLS` produces two `TLSSocket` wrappers over one
@@ -2203,6 +2206,10 @@ impl<const SSL: bool> NewSocket<SSL> {
         if this.socket.get().is_detached() {
             return Ok(());
         }
+        // Counted before any consumer sees the bytes, so `bytesRead` is
+        // right whether a native consumer (an h2 session) or the JS `data`
+        // handler takes them.
+        this.bytes_read.set(this.bytes_read.get() + data.len() as u64);
         if this.native_callback.get().on_data(data)? {
             return Ok(());
         }
@@ -3342,6 +3349,11 @@ impl<const SSL: bool> NewSocket<SSL> {
         )
     }
 
+    #[bun_jsc::host_fn(getter)]
+    pub(crate) fn get_bytes_read(this: &Self, _global: &JSGlobalObject) -> JSValue {
+        JSValue::js_number(this.bytes_read.get() as f64)
+    }
+
     /// In-place TCP→TLS upgrade. The underlying `us_socket_t` is
     /// `adoptTLS`'d into the per-VM TLS group with a fresh (or
     /// SecureContext-shared) `SSL_CTX*`. Returns `[raw, tls]` — two
@@ -3550,6 +3562,7 @@ impl<const SSL: bool> NewSocket<SSL> {
             ref_pollref_on_connect: Cell::new(true),
             buffered_data_for_node_net: JsCell::new(Vec::new()),
             bytes_written: Cell::new(0),
+            bytes_read: Cell::new(0),
             native_callback: JsCell::new(NativeCallbacks::None),
             twin: JsCell::new(None),
             verify_error: JsCell::new(None),
@@ -3654,6 +3667,10 @@ impl<const SSL: bool> NewSocket<SSL> {
             ref_pollref_on_connect: Cell::new(true),
             buffered_data_for_node_net: JsCell::new(Vec::new()),
             bytes_written: Cell::new(0),
+            // Same fd as the retired TCP wrapper: the bytes read before the
+            // upgrade stay with the raw view, like node's TCP handle under a
+            // TLSWrap.
+            bytes_read: Cell::new(this.bytes_read.get()),
             native_callback: JsCell::new(NativeCallbacks::None),
             twin: JsCell::new(None),
             verify_error: JsCell::new(None),
@@ -4688,6 +4705,7 @@ pub fn js_upgrade_duplex_to_tls(
         ref_pollref_on_connect: Cell::new(true),
         buffered_data_for_node_net: JsCell::new(Vec::new()),
         bytes_written: Cell::new(0),
+        bytes_read: Cell::new(0),
         native_callback: JsCell::new(NativeCallbacks::None),
         twin: JsCell::new(None),
         verify_error: JsCell::new(None),

src/runtime/socket/sockets.classes.ts

@@ -107,6 +107,9 @@ function generate(ssl) {
       bytesWritten: {
         getter: "getBytesWritten",
       },
+      bytesRead: {
+        getter: "getBytesRead",
+      },
       fd: {
         getter: "getFD",
       },

test/js/node/http2/node-http2.test.js

@@ -6329,3 +6329,51 @@ describe("frames issued from inside a user-supplied Duplex transport's _write",
     },
   );
 });
+
+// The h2 session consumes the socket natively, so the JS 'data' handler of
+// net.Socket never runs. bytesRead has to come from the handle, like node.
+describe.each(["h2c", "h2"])("%s session.socket.bytesRead", transport => {
+  it("counts the bytes the peer wrote, on the client and on the server", async () => {
+    const secure = transport === "h2";
+    const server = secure ? http2.createSecureServer({ ...TLS_CERT, allowHTTP1: false }) : http2.createServer();
+    let serverSocket;
+    // The client's SETTINGS ack is its last frame. The server has read every byte the client
+    // wrote once it acked the server's settings ('localSettings').
+    const serverAcked = new Promise(resolve => {
+      server.on("session", session => {
+        serverSocket = session.socket;
+        session.once("localSettings", resolve);
+      });
+    });
+    server.on("stream", stream => {
+      stream.respond({ ":status": 200 });
+      stream.end("hello world from h2");
+    });
+    await new Promise(resolve => server.listen(0, "127.0.0.1", resolve));
+    const url = `${secure ? "https" : "http"}://127.0.0.1:${server.address().port}`;
+    const client = http2.connect(url, secure ? TLS_OPTIONS : undefined);
+    try {
+      const req = client.request({ ":path": "/" });
+      req.resume();
+      await new Promise((resolve, reject) => {
+        req.on("end", resolve);
+        req.on("error", reject);
+      });
+      await serverAcked;
+      // The response reached 'end', so the client read every byte the server wrote. Under TLS
+      // both counts are of decrypted bytes, which is what node's TLSWrap reports too.
+      expect({
+        clientRead: client.socket.bytesRead,
+        serverRead: serverSocket.bytesRead,
+      }).toEqual({
+        clientRead: serverSocket.bytesWritten,
+        serverRead: client.socket.bytesWritten,
+      });
+      expect(client.socket.bytesRead).toBeGreaterThan(0);
+      expect(serverSocket.bytesRead).toBeGreaterThan(0);
+    } finally {
+      client.close();
+      server.close();
+    }
+  });
+});