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 of the chunk deliver() is slicing that the onread callback has not received yet.
+const kOnreadSlicing = Symbol("kOnreadSlicing");
+// Bumped by dropOnreadTail(), so deliver() stops a chunk whose connection was replaced under it.
+const kOnreadEpoch = Symbol("kOnreadEpoch");
 const kOnreadDraining = Symbol("kOnreadDraining");
 const kOnreadBuffer = Symbol("kOnreadBuffer");
 const kOnreadPendingEnd = Symbol("kOnreadPendingEnd");
@@ -262,8 +267,14 @@ function closeAdoptedTLSRawNowNT(handle, self, isException) {
   handle.close(onSocketHandleClosed);
   setImmediate(emitCloseNT, self, isException);
 }
+// Bytes the onread callback has not received do not count: node reads one onread buffer per callback.
+function handleBytesRead(self, handle) {
+  return handle.bytesRead - (self[kOnreadTail]?.length ?? 0) - (self[kOnreadSlicing] ?? 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) {
@@ -481,7 +492,6 @@ const SocketHandlers: SocketHandler = {
     if (!self) return;
 
     self._unrefTimer();
-    self.bytesRead += buffer.length;
     pushDataToSocket(self, socket, buffer);
   },
   drain(socket) {
@@ -653,6 +663,16 @@ function finishSocketEnd(self) {
   }
 }
 
+// The declined bytes belong to the handle that read them: a new connection starts without them.
+function dropOnreadTail(self) {
+  if (self[kOnreadBuffer] === undefined) return;
+  self[kOnreadTail] = undefined;
+  self[kOnreadSlicing] = 0;
+  self[kOnreadEpoch]++;
+  self[kOnreadPendingEnd] = false;
+  self[kOnreadReadRequested] = false;
+}
+
 function deferEndForOnreadTail(self) {
   if (self[kOnreadTail] === undefined || self.destroyed) return false;
   self[kOnreadPendingEnd] = true;
@@ -801,7 +821,6 @@ const ServerHandlers: SocketHandler<NetSocket> = {
     if (!self) return;
 
     self._unrefTimer();
-    self.bytesRead += buffer.length;
     pushDataToSocket(self, socket, buffer);
   },
   keylog(socket, line) {
@@ -1315,7 +1334,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) {
@@ -1537,6 +1555,8 @@ function kConnectPipe(self, req, address) {
 }
 
 function kConnectDispatch(self, req, opts) {
+  // doConnect swaps the native socket under a live handle: bytes declined since connect() stay with the old one.
+  dropOnreadTail(self);
   // Node's TCPWrap returns errno for sync uv_*_connect failure and defers
   // oncomplete; doConnect instead fires connectError inside this call. Bracket
   // it so connectError hands the errno back here instead of re-entering.
@@ -1610,7 +1630,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;
@@ -1698,6 +1718,8 @@ function Socket(options?) {
     const self = this;
     this[kOnreadTail] = undefined;
     this[kOnreadDraining] = false;
+    this[kOnreadSlicing] = 0;
+    this[kOnreadEpoch] = 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
@@ -1715,11 +1737,21 @@ function Socket(options?) {
     // native read in this loop, so the catch is per-slice.
     // https://github.com/nodejs/node/blob/v26.3.0/lib/internal/stream_base_commons.js#L179
     this[kOnreadDeliver] = function deliver(buffer) {
+      self[kOnreadSlicing] = buffer.length;
+      try {
+        deliverSlices(buffer);
+      } finally {
+        self[kOnreadSlicing] = 0;
+      }
+    };
+    function deliverSlices(buffer) {
       let offset = 0;
       const total = buffer.length;
+      const epoch = self[kOnreadEpoch];
       while (offset < total) {
         const dest = self[kOnreadBuffer];
         if (dest === true) {
+          self[kOnreadSlicing] = 0;
           let ret;
           try {
             ret = onreadCallback(total - offset, true);
@@ -1733,7 +1765,7 @@ function Socket(options?) {
             // synchronously the way node's bare call does.
             reportError(e);
           }
-          if (self.destroyed) return;
+          if (self.destroyed || self[kOnreadEpoch] !== epoch) return;
           if (ret === false || self.isPaused()) {
             self[kOnreadTail] = kOnreadEmptyTail;
             readStop(self, self._handle);
@@ -1751,6 +1783,7 @@ function Socket(options?) {
         const n = MathMin(dest.length, total - offset);
         dest.set(buffer.subarray(offset, offset + n));
         offset += n;
+        self[kOnreadSlicing] = total - offset;
         let ret;
         try {
           ret = onreadCallback(n, dest);
@@ -1763,15 +1796,16 @@ function Socket(options?) {
           // delivered, matching node's per-onStreamRead behavior.
           reportError(e);
         }
-        if (self.destroyed) return;
+        // A callback that called connect() replaced the connection: the rest of this chunk is the old one's.
+        if (self.destroyed || self[kOnreadEpoch] !== epoch) return;
         if (ret === false || self.isPaused()) {
           const rest = buffer.subarray(offset);
           self[kOnreadTail] = rest.length !== 0 ? rest : kOnreadEmptyTail;
           readStop(self, self._handle);
           return;
         }
       }
-    };
+    }
     // when the onread option is specified we use a different handlers object
     this[khandlers] = {
       ...SocketHandlers2,
@@ -1849,6 +1883,15 @@ Object.defineProperty(Socket.prototype, "_bytesDispatched", {
   },
 });
 
+// https://github.com/nodejs/node/blob/v26.3.0/lib/net.js#L939-L950
+Object.defineProperty(Socket.prototype, "bytesRead", {
+  enumerable: true,
+  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;
@@ -1909,7 +1952,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();
 };
@@ -2152,6 +2195,7 @@ Socket.prototype.connect = function connect(...args) {
   }
 
   this.connecting = true;
+  dropOnreadTail(this);
 
   const { path } = options;
   const pipe = !!path;
@@ -2226,7 +2270,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,8 @@ 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>,
+    /// Bytes delivered through `on_data` (plaintext on a TLS view).
+    pub(crate) bytes_read: Cell<u64>,
 
     pub(crate) native_callback: JsCell<NativeCallbacks>,
     /// `upgradeTLS` produces two `TLSSocket` wrappers over one
@@ -1354,6 +1356,7 @@ impl<const SSL: bool> NewSocket<SSL> {
         self.socket.set(SocketHandler::<SSL>::DETACHED);
         self.buffered_data_for_node_net
             .with_mut(|b| b.clear_and_free());
+        self.bytes_read.set(0);
         self.detach_native_callback();
         old.close(uws::CloseCode::Failure);
         self.poll_ref.with_mut(|p| p.unref(js_loop_ctx()));
@@ -2203,6 +2206,8 @@ impl<const SSL: bool> NewSocket<SSL> {
         if this.socket.get().is_detached() {
             return Ok(());
         }
+        this.bytes_read
+            .set(this.bytes_read.get() + data.len() as u64);
         if this.native_callback.get().on_data(data)? {
             return Ok(());
         }
@@ -3342,6 +3347,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 +3560,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 +3665,8 @@ 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: its count moves to the raw view.
+            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 +4701,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();
+    }
+  });
+});