node:http: emit 'data' on the server 'connection' socket with the raw request bytes

packages/bun-uws/src/App.h

@@ -750,6 +750,9 @@ struct TemplatedApp {
     void setOnSocketData(HttpContextData<SSL>::OnSocketDataCallback onData) {
         httpContext->getSocketContextData()->onSocketData = onData;
     }
+    void setOnSocketRawData(HttpContextData<SSL>::OnSocketDataCallback onRawData) {
+        httpContext->getSocketContextData()->onSocketRawData = onRawData;
+    }
 
     void setOnClientError(HttpContextData<SSL>::OnClientErrorCallback onClientError) {
         httpContext->getSocketContextData()->onClientError = std::move(onClientError);

packages/bun-uws/src/HttpContext.h

@@ -306,6 +306,23 @@
             }
         }
 
+        /* node:http compat: Node's server socket is a net.Socket whose 'data'
+         * event carries the raw TCP payload (the parser is just another 'data'
+         * listener). Forward the raw bytes to the JS socket wrapper before
+         * parsing so a user 'data' listener on the 'connection' socket sees
+         * them. Tunnel mode routes through onSocketData instead. The receiver
+         * early-returns when no JS callback is installed, so the common path
+         * pays only the indirect call. */
+        if constexpr (IsNodeHttp) {
+            if (!httpResponseData->isConnectRequest && httpResponseData->socketData && httpContextData->onSocketRawData) {
+                httpContextData->onSocketRawData(httpResponseData->socketData, SSL, s, data, length, false);
+                if (us_socket_is_closed(s)) {
+                    us_socket_unref(s);
+                    return s;
+                }
+            }
+        }
+
         /* Cork this socket */
         ((AsyncSocket<SSL> *) s)->cork();
 

packages/bun-uws/src/HttpContextData.h

@@ -76,6 +76,11 @@ struct alignas(16) HttpContextData {
     OnSocketClosedCallback onSocketClosed = nullptr;
     OnSocketDrainCallback onSocketDrain = nullptr;
     OnSocketDataCallback onSocketData = nullptr;
+    /* node:http compat: tap of every TCP payload before HTTP parsing (Node's
+     * server socket emits 'data' with the raw request bytes to every listener,
+     * its parser included). The receiver gates on its own callback slot so this
+     * is a null-check when no user 'data' listener is attached. */
+    OnSocketDataCallback onSocketRawData = nullptr;
     OnSocketUpgradedCallback onSocketUpgraded = nullptr;
     OnClientErrorCallback onClientError = nullptr;
 

src/js/node/_http_server.ts

@@ -898,6 +898,7 @@ Server.prototype[kRealListen] = function (tls, port, host, socketPath, reusePort
 
         if (isSocketNew && !reachedRequestsLimit) {
           server.emit("connection", socket);
+          maybeEnableRawSocketData(socket);
         }
 
         if (!isPipelined) {
@@ -1223,6 +1224,7 @@ function onServerConnection(this: Server, socketHandle) {
   // emitting 'connection'; expose the shim here so listeners see it populated.
   socket.parser = createServerParserShim(socket);
   this.emit("connection", socket);
+  maybeEnableRawSocketData(socket);
   if (isTLS && socketHandle.secureEstablished) {
     this.emit("secureConnection", socket);
   }
@@ -1393,6 +1395,19 @@ function detachSocketListenersForHandoff(socket) {
 }
 const kSocketTimeoutTimer = Symbol("socketTimeoutTimer");
 const kStreamingEnabled = Symbol("kStreamingEnabled");
+const kEnableRawData = Symbol("kEnableRawData");
+// Node's server socket is the net.Socket the parser reads from, so every
+// 'data' listener sees the raw request bytes. Here parsing runs natively; when
+// a user reads the socket, have the native side forward the raw TCP payload
+// into this Duplex so 'data'/'readable' behave the same.
+function maybeEnableRawSocketData(socket) {
+  if (
+    !socket[kStreamingEnabled] &&
+    (socket.listenerCount("data") > 0 || socket.listenerCount("readable") > 0)
+  ) {
+    socket[kEnableRawData]();
+  }
+}
 // Scratch options object for the builtin ServerResponse (see the dispatcher).
 const scratchResponseOptions = {
   [kHandle]: undefined,
@@ -1609,6 +1624,10 @@ const NodeHTTPServerSocket = class Socket extends NetSocket {
       if (enable) {
         handle.ondata = this.#onData.bind(this);
         handle.ondrain = this.#onDrain.bind(this);
+        // Tunnel mode owns the readable side from here; the raw-bytes tap would
+        // deliver the same bytes a second time. The already-queued chunk for
+        // this packet is dropped by the task's own null-check.
+        handle.onrawdata = undefined;
       } else {
         handle.ondata = undefined;
         handle.ondrain = undefined;
@@ -1625,6 +1644,23 @@ const NodeHTTPServerSocket = class Socket extends NetSocket {
     }
     this.emit("drain");
   }
+  #onRawData(chunk) {
+    this._unrefTimer();
+    this.bytesRead += chunk.length;
+    if (!this.push(chunk)) {
+      // Readable buffer filled: drop the tap so the native side stops
+      // allocating a Buffer per TCP read; _read() re-arms on next demand.
+      // HTTP parsing continues natively regardless.
+      const handle = this[kHandle];
+      if (handle) handle.onrawdata = undefined;
+    }
+  }
+  [kEnableRawData]() {
+    const handle = this[kHandle];
+    if (handle && !handle.onrawdata) {
+      handle.onrawdata = this.#onRawData.bind(this);
+    }
+  }
   #onData(chunk, last) {
     this._unrefTimer();
     if (chunk) {
@@ -1879,6 +1915,10 @@ const NodeHTTPServerSocket = class Socket extends NetSocket {
 
   _read(_size) {
     // https://github.com/nodejs/node/blob/13e3aef053776be9be262f210dc438ecec4a3c8d/lib/net.js#L725-L737
+    // (Re-)arm the raw-bytes tap: a listener added after 'connection' reaches
+    // here via the Readable's resume() → read(0), and a push()==false disarm
+    // re-arms on the next demand.
+    maybeEnableRawSocketData(this);
     this.#resumeSocket();
   }
 

src/jsc/bindings/NodeHTTP.cpp

@@ -613,6 +613,11 @@ static void assignOnNodeJSCompat(uWS::TemplatedApp<isSSL>* app)
         ASSERT(rawSocket == socket->socket || socket->socket == nullptr);
         socket->onData(data, length, last);
     });
+    app->setOnSocketRawData([](void* socketData, int is_ssl, struct us_socket_t* rawSocket, const char* data, int length, bool last) -> void {
+        auto* socket = reinterpret_cast<JSNodeHTTPServerSocket*>(socketData);
+        ASSERT(rawSocket == socket->socket || socket->socket == nullptr);
+        socket->onRawData(data, length);
+    });
     app->setOnSocketUpgraded([](void* socketData, int is_ssl, struct us_socket_t* rawSocket) -> void {
         auto* socket = reinterpret_cast<JSNodeHTTPServerSocket*>(socketData);
         // the socket is adopted and might not be the same as the rawSocket

src/jsc/bindings/node/JSNodeHTTPServerSocket.cpp

@@ -649,6 +649,56 @@ void JSNodeHTTPServerSocket::onData(const char* data, int length, bool last)
     }
 }
 
+void JSNodeHTTPServerSocket::onRawData(const char* data, int length)
+{
+    // Called for every TCP payload on a node:http connection before HTTP
+    // parsing. When no JS 'data' listener is attached the callback slot is
+    // null and this is the only work done.
+    if (!functionToCallOnRawData) {
+        return;
+    }
+    Zig::GlobalObject* globalObject = static_cast<Zig::GlobalObject*>(this->globalObject());
+    WebCore::ScriptExecutionContext* scriptExecutionContext = globalObject->scriptExecutionContext();
+    if (!scriptExecutionContext) {
+        return;
+    }
+
+    auto scope = DECLARE_TOP_EXCEPTION_SCOPE(globalObject->vm());
+    JSC::JSUint8Array* buffer = WebCore::createBuffer(globalObject, std::span<const uint8_t>(reinterpret_cast<const uint8_t*>(data), length));
+    auto chunk = JSC::JSValue(buffer);
+    if (auto* exception = scope.exception()) {
+        (void)scope.tryClearException();
+        globalObject->reportUncaughtExceptionAtEventLoop(globalObject, exception);
+        return;
+    }
+    gcProtect(chunk);
+    scriptExecutionContext->postTask([self = this, chunk = chunk](ScriptExecutionContext& context) {
+        WTF::NakedPtr<JSC::Exception> exception;
+        auto* globalObject = defaultGlobalObject(context.globalObject());
+        auto* thisObject = self;
+        auto* callbackObject = thisObject->functionToCallOnRawData.get();
+        EnsureStillAliveScope ensureChunkStillAlive(chunk);
+        gcUnprotect(chunk);
+        if (!callbackObject) {
+            return;
+        }
+
+        auto callData = JSC::getCallData(callbackObject);
+        MarkedArgumentBuffer args;
+        args.append(chunk);
+        EnsureStillAliveScope ensureStillAlive(self);
+
+        if (globalObject->scriptExecutionStatus(globalObject, thisObject) == ScriptExecutionStatus::Running) {
+            profiledCall(globalObject, JSC::ProfilingReason::API, callbackObject, callData, thisObject, args, exception);
+
+            if (auto* ptr = exception.get()) {
+                exception.clear();
+                globalObject->reportUncaughtExceptionAtEventLoop(globalObject, ptr);
+            }
+        }
+    });
+}
+
 JSC::Structure* JSNodeHTTPServerSocket::createStructure(JSC::VM& vm, JSC::JSGlobalObject* globalObject)
 {
     auto* structure = JSC::Structure::create(vm, globalObject, globalObject->objectPrototype(), JSC::TypeInfo(JSC::ObjectType, StructureFlags), JSNodeHTTPServerSocketPrototype::info());
@@ -672,6 +722,7 @@ void JSNodeHTTPServerSocket::visitChildrenImpl(JSCell* cell, Visitor& visitor)
     visitor.append(fn->functionToCallOnClose);
     visitor.append(fn->functionToCallOnDrain);
     visitor.append(fn->functionToCallOnData);
+    visitor.append(fn->functionToCallOnRawData);
     visitor.append(fn->m_remoteAddress);
     visitor.append(fn->m_localAddress);
     visitor.append(fn->m_duplex);

src/jsc/bindings/node/JSNodeHTTPServerSocket.h

@@ -128,6 +128,7 @@ class JSNodeHTTPServerSocket : public JSC::JSDestructibleObject {
     mutable JSC::WriteBarrier<JSC::JSObject> functionToCallOnClose;
     mutable JSC::WriteBarrier<JSC::JSObject> functionToCallOnDrain;
     mutable JSC::WriteBarrier<JSC::JSObject> functionToCallOnData;
+    mutable JSC::WriteBarrier<JSC::JSObject> functionToCallOnRawData;
     mutable JSC::WriteBarrier<WebCore::JSNodeHTTPResponse> currentResponseObject;
     mutable JSC::WriteBarrier<JSC::JSObject> m_remoteAddress;
     mutable JSC::WriteBarrier<JSC::JSObject> m_localAddress;
@@ -160,6 +161,7 @@ class JSNodeHTTPServerSocket : public JSC::JSDestructibleObject {
     void onClose();
     void onDrain();
     void onData(const char* data, int length, bool last);
+    void onRawData(const char* data, int length);
 
     static JSC::Structure* createStructure(JSC::VM& vm, JSC::JSGlobalObject* globalObject);
     void finishCreation(JSC::VM& vm);

src/jsc/bindings/node/JSNodeHTTPServerSocketPrototype.cpp

@@ -27,6 +27,8 @@ JSC_DECLARE_CUSTOM_SETTER(jsNodeHttpServerSocketSetterOnClose);
 JSC_DECLARE_CUSTOM_SETTER(jsNodeHttpServerSocketSetterOnDrain);
 JSC_DECLARE_CUSTOM_SETTER(jsNodeHttpServerSocketSetterOnData);
 JSC_DECLARE_CUSTOM_GETTER(jsNodeHttpServerSocketGetterOnData);
+JSC_DECLARE_CUSTOM_SETTER(jsNodeHttpServerSocketSetterOnRawData);
+JSC_DECLARE_CUSTOM_GETTER(jsNodeHttpServerSocketGetterOnRawData);
 JSC_DECLARE_CUSTOM_GETTER(jsNodeHttpServerSocketGetterBytesWritten);
 JSC_DECLARE_HOST_FUNCTION(jsFunctionNodeHTTPServerSocketClose);
 JSC_DECLARE_HOST_FUNCTION(jsFunctionNodeHTTPServerSocketWrite);
@@ -57,6 +59,7 @@ static const JSC::HashTableValue JSNodeHTTPServerSocketPrototypeTableValues[] =
     { "onclose"_s, static_cast<unsigned>(JSC::PropertyAttribute::CustomAccessor), JSC::NoIntrinsic, { JSC::HashTableValue::GetterSetterType, jsNodeHttpServerSocketGetterOnClose, jsNodeHttpServerSocketSetterOnClose } },
     { "ondrain"_s, static_cast<unsigned>(JSC::PropertyAttribute::CustomAccessor), JSC::NoIntrinsic, { JSC::HashTableValue::GetterSetterType, jsNodeHttpServerSocketGetterOnDrain, jsNodeHttpServerSocketSetterOnDrain } },
     { "ondata"_s, static_cast<unsigned>(JSC::PropertyAttribute::CustomAccessor), JSC::NoIntrinsic, { JSC::HashTableValue::GetterSetterType, jsNodeHttpServerSocketGetterOnData, jsNodeHttpServerSocketSetterOnData } },
+    { "onrawdata"_s, static_cast<unsigned>(JSC::PropertyAttribute::CustomAccessor), JSC::NoIntrinsic, { JSC::HashTableValue::GetterSetterType, jsNodeHttpServerSocketGetterOnRawData, jsNodeHttpServerSocketSetterOnRawData } },
     { "bytesWritten"_s, static_cast<unsigned>(JSC::PropertyAttribute::CustomAccessor), JSC::NoIntrinsic, { JSC::HashTableValue::GetterSetterType, jsNodeHttpServerSocketGetterBytesWritten, noOpSetter } },
     { "closed"_s, static_cast<unsigned>(JSC::PropertyAttribute::CustomAccessor | JSC::PropertyAttribute::ReadOnly), JSC::NoIntrinsic, { JSC::HashTableValue::GetterSetterType, jsNodeHttpServerSocketGetterClosed, noOpSetter } },
     { "response"_s, static_cast<unsigned>(JSC::PropertyAttribute::CustomAccessor | JSC::PropertyAttribute::ReadOnly), JSC::NoIntrinsic, { JSC::HashTableValue::GetterSetterType, jsNodeHttpServerSocketGetterResponse, noOpSetter } },
@@ -469,6 +472,44 @@ JSC_DEFINE_CUSTOM_SETTER(jsNodeHttpServerSocketSetterOnData, (JSC::JSGlobalObjec
     return true;
 }
 
+JSC_DEFINE_CUSTOM_GETTER(jsNodeHttpServerSocketGetterOnRawData, (JSC::JSGlobalObject * globalObject, JSC::EncodedJSValue thisValue, JSC::PropertyName))
+{
+    auto* thisObject = dynamicDowncast<JSNodeHTTPServerSocket>(JSC::JSValue::decode(thisValue));
+    if (!thisObject) [[unlikely]] {
+        return JSValue::encode(JSC::jsUndefined());
+    }
+
+    if (thisObject->functionToCallOnRawData) {
+        return JSValue::encode(thisObject->functionToCallOnRawData.get());
+    }
+
+    return JSValue::encode(JSC::jsUndefined());
+}
+
+JSC_DEFINE_CUSTOM_SETTER(jsNodeHttpServerSocketSetterOnRawData, (JSC::JSGlobalObject * globalObject, JSC::EncodedJSValue thisValue, JSC::EncodedJSValue encodedValue, JSC::PropertyName propertyName))
+{
+    auto& vm = globalObject->vm();
+    auto scope = DECLARE_THROW_SCOPE(vm);
+
+    auto* thisObject = dynamicDowncast<JSNodeHTTPServerSocket>(JSC::JSValue::decode(thisValue));
+    if (!thisObject) [[unlikely]] {
+        return false;
+    }
+    JSValue value = JSC::JSValue::decode(encodedValue);
+
+    if (value.isUndefined() || value.isNull()) {
+        thisObject->functionToCallOnRawData.clear();
+        return true;
+    }
+
+    if (!value.isCallable()) {
+        return false;
+    }
+
+    thisObject->functionToCallOnRawData.set(vm, thisObject, value.getObject());
+    return true;
+}
+
 JSC_DEFINE_CUSTOM_SETTER(jsNodeHttpServerSocketSetterOnClose, (JSC::JSGlobalObject * globalObject, JSC::EncodedJSValue thisValue, JSC::EncodedJSValue encodedValue, JSC::PropertyName propertyName))
 {
     auto& vm = globalObject->vm();