node:http: a 'connect' or 'upgrade' socket keeps its unread bytes when the connection closes

packages/bun-usockets/src/context.c

@@ -346,6 +346,7 @@ static void us_internal_init_listen_socket(struct us_listen_socket_t *ls,
     s->flags.allow_half_open = (options & LIBUS_SOCKET_ALLOW_HALF_OPEN);
     s->unclassified_send_failures = 0;
     s->read_eof = 0;
+    s->read_fin = 0;
     s->hangup_closes_unsent = 0;
     s->next = 0;
     s->prev = 0;
@@ -526,6 +527,7 @@ static inline void us_internal_init_connect_socket(struct us_socket_t *s,
     s->flags.last_write_failed = 0;
     s->unclassified_send_failures = 0;
     s->read_eof = 0;
+    s->read_fin = 0;
     s->hangup_closes_unsent = 0;
     s->connect_state = NULL;
     s->connect_next = NULL;

packages/bun-usockets/src/internal/internal.h

@@ -369,6 +369,8 @@ struct us_socket_t {
   unsigned char ssl_has_pending_events : 1;
   /* Peer FIN was dispatched as on_end on a half-open socket; readable interest is never re-added and on_end never re-fires. */
   unsigned char read_eof : 1;
+  /* A read returned 0: the peer ended its stream with a FIN. on_end is not always dispatched for it (see us_socket_read_fin). */
+  unsigned char read_fin : 1;
   /* A hangup that leaves bytes unsent closes the socket even while it is paused (loop.c defers it otherwise).
    * For an owner whose pause can wait for those bytes to drain: node:http's pipelining. */
   unsigned char hangup_closes_unsent : 1;

packages/bun-usockets/src/libusockets.h

@@ -714,6 +714,10 @@ int us_socket_ssl_handshake_callback_has_fired(us_socket_r s) nonnull_fn_decl;
 unsigned int us_socket_ssl_spill_pending(us_socket_r s) nonnull_fn_decl;
 /* Whether some of that ciphertext was thrown away before the kernel took it, which happens when the connection ends. */
 int us_socket_ssl_spill_dropped(us_socket_r s) nonnull_fn_decl;
+/* Whether a read of this socket returned 0, which is the peer's FIN. The close code does not tell: a close that this
+ * side starts and a failed TLS session close with code 0 too. When the FIN follows this side's own, the socket closes
+ * with no on_end, so this is the one record of it. */
+int us_socket_read_fin(us_socket_r s) nonnull_fn_decl;
 
 struct us_socket_t *us_socket_close(us_socket_r s, int code, void *reason) __attribute__((nonnull(1)));
 /* Closes before it returns. A TLS socket first sends what the kernel takes now: its unsent ciphertext, then close_notify. */

packages/bun-usockets/src/loop.c

@@ -555,6 +555,7 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in
                         s->flags.last_write_failed = 0;
                         s->unclassified_send_failures = 0;
                         s->read_eof = 0;
+                        s->read_fin = 0;
                         s->hangup_closes_unsent = 0;
 
                         /* We always use nodelay */
@@ -845,6 +846,7 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in
                     } else if (!length) {
                         eof = LIBUS_POLL_EOF; // lets handle EOF in the same place
                         read_fin = 1;
+                        s->read_fin = 1;
                         break;
                     } else if (length == LIBUS_SOCKET_ERROR && !bsd_would_block()) {
                         /* A read error closes with its errno even when the drain above

packages/bun-usockets/src/socket.c

@@ -168,6 +168,10 @@ int us_socket_ssl_spill_dropped(struct us_socket_t *s) {
     return s->ssl && s->ssl_spill_dropped;
 }
 
+int us_socket_read_fin(struct us_socket_t *s) {
+    return s->read_fin;
+}
+
 int us_connecting_socket_is_closed(struct us_connecting_socket_t *c) {
     return c->closed;
 }
@@ -452,6 +456,7 @@ struct us_socket_t *us_socket_from_fd(struct us_socket_group_t *group, unsigned
     s->flags.last_write_failed = 0;
     s->unclassified_send_failures = 0;
     s->read_eof = 0;
+    s->read_fin = 0;
     s->hangup_closes_unsent = 0;
     s->connect_state = NULL;
 

packages/bun-uws/src/HttpContext.h

@@ -263,7 +263,10 @@ struct HttpContext {
         bool nodeHttpTunnelAfterBody = false;
         if constexpr (IsNodeHttp) nodeHttpTunnelAfterBody = (httpResponseData->state & HttpResponseData<SSL>::HTTP_NODE_TUNNEL_AFTER_BODY) != 0;
         if(httpResponseData->isConnectRequest || nodeHttpTunnelAfterBody) {
-            if (httpResponseData->socketData && httpContextData->onSocketData) {
+            /* node:http: only the peer's FIN ends its stream. A reset, a failed TLS session and a close that this side starts do not. */
+            bool streamEnded = true;
+            if constexpr (IsNodeHttp) streamEnded = us_socket_read_fin(s);
+            if (streamEnded && httpResponseData->socketData && httpContextData->onSocketData) {
                 httpContextData->onSocketData(httpResponseData->socketData, SSL, s, "", 0, true);
             }
             /* node:http: the body of a tunnel that started has had its last chunk from the parser. This close can come from inside that call. */

src/js/node/_http_server.ts

@@ -89,9 +89,24 @@ const kPendingCloseGenerations = Symbol("http.server.pendingCloseGenerations");
 const kListenerGeneration = Symbol("http.server.listenerGeneration");
 // Set on the 'connect'/'upgrade' handoff; closeAll/closeIdleConnections() skip these, as in Node.
 const kHandedOff = Symbol("http.server.socketHandedOff");
+// Its values. Node hands its listener the net.Socket itself when the request is complete at the emit, and otherwise an
+// UpgradeStream, which the close of the connection destroys:
+// https://github.com/nodejs/node/blob/v26.3.0/lib/_http_server.js#L1027-L1050
+// https://github.com/nodejs/node/blob/v26.3.0/lib/_http_server.js#L1110-L1143
+// Bun emits 'upgrade' before it parses the body, so an Upgrade request with a body is never complete there.
+const kHandedOffSocket = 1;
+const kHandedOffUpgradeStream = 2;
+// #closeHandled of a kHandedOffSocket that has the end of its stream. Like Node's net.Socket, it closes after its own
+// 'end' and 'finish', so the close of the connection leaves it the bytes that its reader has not taken.
+const kClosesAfterEnd = 2;
+// The connection has closed, and the socket waits for its reader.
+const kOutlivesConnection = 3;
 function closeHandle(handle) {
   if (!handle.closed) handle.close();
 }
+function destroySocket(socket) {
+  socket.destroy();
+}
 const kHttpAllowHalfOpen = Symbol("http.server.httpAllowHalfOpen");
 const kMaxHeadersCount = Symbol("http.server.maxHeadersCount");
 
@@ -819,7 +834,7 @@ Server.prototype[kRealListen] = function (tls, port, host, socketPath, reusePort
             socket[kHandoffResponse] = handle;
             // The parser is detached: the socket is handed over with only
             // net.Socket's 'end' listener left, like Node.js.
-            detachSocketListenersForHandoff(socket);
+            detachSocketListenersForHandoff(socket, kHandedOffSocket);
             // Pass the pipelined data (head buffer) if any was received with the CONNECT request
             const head = connectHead ? connectHead : kEmptyBuffer;
             socket.bytesRead += head.length;
@@ -1052,7 +1067,7 @@ Server.prototype[kRealListen] = function (tls, port, host, socketPath, reusePort
           socketHandle.upgradeToTunnel(hasBody, handle);
           socket[kHandoffResponse] = handle;
           socket[kEnableStreaming](true);
-          detachSocketListenersForHandoff(socket);
+          detachSocketListenersForHandoff(socket, hasBody ? kHandedOffUpgradeStream : kHandedOffSocket);
           // Node frees the parser before emitting 'upgrade' (socket.parser === null there).
           releaseServerParserShim(socket, http_req);
           if (hasBody) {
@@ -1313,7 +1328,8 @@ function onServerRequestEvent(this: NodeHTTPServerSocket, event: NodeHTTPRespons
   const socket: NodeHTTPServerSocket = this;
   switch (event) {
     case NodeHTTPResponseAbortEvent.abort: {
-      if (!socket.destroyed) {
+      // A handed-off socket closes in #onClose, which the same native close calls next.
+      if (!socket.destroyed && !socket[kHandedOff]) {
         socket.destroy();
       }
       break;
@@ -1482,8 +1498,8 @@ function resumePausedUpgradeIncoming(socket) {
 // connection-listener set removed (onParserExecuteCommon removes its data/end/
 // close/drain/error/timeout listeners) and only net.Socket's own 'end' listener
 // left in place.
-function detachSocketListenersForHandoff(socket) {
-  socket[kHandedOff] = true;
+function detachSocketListenersForHandoff(socket, as: typeof kHandedOffSocket | typeof kHandedOffUpgradeStream) {
+  socket[kHandedOff] = as;
   socket.removeListener("error", socketOnError);
   socket.removeListener("timeout", onNodeHTTPServerSocketTimeout);
   socket.on("end", onReadableStreamEnd);
@@ -1670,7 +1686,7 @@ function getNodeHTTPServerSocket() {
     [kOnReadParsed] = undefined;
     [kHandoffResponse] = undefined;
     [kDestroySoon] = false;
-    [kHandedOff] = false;
+    [kHandedOff]: false | typeof kHandedOffSocket | typeof kHandedOffUpgradeStream = false;
     server: Server;
     _httpMessage;
     _secureEstablished = false;
@@ -1679,15 +1695,17 @@ function getNodeHTTPServerSocket() {
     _paused = false;
     #pendingCallback = null;
     #pendingAbortMessage;
-    #closeHandled = false;
+    #closeHandled: boolean | typeof kClosesAfterEnd | typeof kOutlivesConnection = false;
     #peer;
     #closingHandle;
     #resetSupported;
     #closeError: Error | undefined = undefined;
     declare encrypted: boolean;
+    // A slot of net.Socket.
+    declare _server: Server;
     declare resetAndClosing: boolean;
     declare _writableState: { emitClose: boolean; decodeStrings: boolean };
-    declare _readableState: { emitClose: boolean };
+    declare _readableState: { emitClose: boolean; ended: boolean };
     declare writableBuffer: { chunk: Buffer }[];
     constructor(server: Server, handle, encrypted, listenerGeneration) {
       // allowHalfOpen: node's connectionListener sockets never auto-end the
@@ -1707,6 +1725,8 @@ function getNodeHTTPServerSocket() {
       this._readableState.emitClose = true;
       this._writableState.decodeStrings = true;
       this.server = server;
+      // The server that accepted the connection tracks it. A 'connection' listener of any server reassigns `server`.
+      this._server = server;
       this.#resetSupported = !encrypted && !listenerGeneration?.isUnix;
       this[kHandle] = handle;
       this._secureEstablished = !!handle?.secureEstablished;
@@ -1805,11 +1825,19 @@ function getNodeHTTPServerSocket() {
           this[kHandoffResponse]?.unref();
         }
 
-        this.push(null);
-        // Like Node's net.Socket (and its UpgradeStream): EOF on a socket with no
-        // pending data emits 'end' even when nothing is reading the socket.
-        if (this.readableLength === 0) {
-          this.resume();
+        if (this[kHandedOff] === kHandedOffSocket) {
+          // Before push(null), which runs 'readable' listeners.
+          this.#closeHandled = kClosesAfterEnd;
+          // https://github.com/nodejs/node/blob/v26.3.0/lib/internal/stream_base_commons.js#L231-L232
+          this.push(null);
+          // 'end' with nothing to read and nobody reading. A paused socket stays paused. A flowing one has its 'end' from push(null).
+          if (this.readableFlowing !== true) this.read(0);
+        } else {
+          this.push(null);
+          // https://github.com/nodejs/node/blob/v26.3.0/lib/_http_server.js#L1042-L1050
+          if (this.readableLength === 0) {
+            this.resume();
+          }
         }
       }
     }
@@ -1831,9 +1859,39 @@ function getNodeHTTPServerSocket() {
         handle.close();
       }
     }
+    // Whether a handed-off socket whose connection closed with no FIN of the peer gets 'end' when its reader has
+    // taken every byte, as it always has. That is the close after a reset of the peer, and after a failed TLS
+    // session. Node emits 'error' there. pipe() needs the 'end' to end its destination.
+    #endsWithItsConnection() {
+      // This side closed the connection. Node's net.Socket emits no 'end' after destroySoon().
+      if (this.destroyed || (this[kDestroySoon] && this[kHandedOff] === kHandedOffSocket)) return false;
+      // A WebSocket took the connection over, and reads it in native code.
+      return !this[kHandoffResponse]?.upgraded;
+    }
     #onClose(closedHandle = this[kHandle]) {
-      // Once: a queued response's abort can destroy the socket, and run this, before the native close calls it.
-      if (this.#closeHandled) return;
+      if (this.#closeHandled) {
+        // Once: a queued response's abort can destroy the socket, and run this, before the native close calls it.
+        if (this.#closeHandled === true) return;
+        if (!this.destroyed && !this.readableEnded && !this[kDestroySoon] && !this.#pendingCallback) {
+          // The peer's FIN ended the stream of a net.Socket, and its reader has not taken all of it. Like Node's, the
+          // socket closes after its own 'end' and 'finish': _destroy() runs the rest of this function. A write that
+          // waits for a drain has failed, and destroySoon() asked for the close, so neither waits for the reader.
+          this.#closeHandled = kOutlivesConnection;
+          // The handle went with the connection.
+          this[kBytesWritten] = handleBytesWritten(closedHandle);
+          this[kClosedWithUnsentData] = closedHandle.closedWithUnsentData;
+          this[kHandle] = null;
+          // So did the server's part: it tracks connections, and keeps nothing for a socket that its owner can drop.
+          // Node, whose socket keeps its handle, counts the socket until it closes, and server.close() waits for it.
+          const server = this._server;
+          const tracked = server[kTrackedConnections];
+          tracked.delete(this);
+          if (tracked.size === 0 && server[kPendingDrainClose]) process.nextTick(emitCloseServer, server);
+          // The keep-alive timer that the dispatcher leaves armed with no timeout would refresh itself for ever.
+          if (this.timeout === 0) this.setTimeout(0);
+          return;
+        }
+      }
       this.#closeHandled = true;
       const errored = this.errored;
       // freeParser equivalent: runs before 'close' listeners so they observe the
@@ -1844,7 +1902,10 @@ function getNodeHTTPServerSocket() {
         this[kBytesWritten] = handleBytesWritten(closedHandle);
         this[kClosedWithUnsentData] = closedHandle.closedWithUnsentData;
         // Peer FIN: 'end' before 'close', like Node's net.Socket. read(0) emits it when nothing reads the socket.
-        if (closedHandle.peerEnded) {
+        if (
+          closedHandle.peerEnded ||
+          (this[kHandedOff] && !this._readableState.ended && this.#endsWithItsConnection())
+        ) {
           this.push(null);
           this.read(0);
         }
@@ -1856,7 +1917,7 @@ function getNodeHTTPServerSocket() {
           this.#closeError = er.code === closeError.code ? er : closeError;
         }
       }
-      const server = this.server;
+      const server = this._server;
       const tracked = server?.[kTrackedConnections];
       if (tracked) {
         tracked.delete(this);
@@ -2000,6 +2061,8 @@ function getNodeHTTPServerSocket() {
       }
       const handle = this[kHandle];
       if (!handle) {
+        // The connection closed before, and the socket waited for its reader: the rest of its close runs now.
+        if (this.#closeHandled === kOutlivesConnection) this.#onClose(null);
         if ($isCallable(callback)) callback(err);
         return;
       }
@@ -2152,6 +2215,12 @@ function getNodeHTTPServerSocket() {
         }
         this.resetAndClosing = true;
         this.destroy();
+      } else if (this.#closeHandled === kOutlivesConnection) {
+        // Nothing is left to reset. Node, which still has the connection, resets it and emits no error.
+        if (!this.#resetSupported) {
+          throw $ERR_INVALID_HANDLE_TYPE();
+        }
+        this.destroy();
       } else {
         this.destroy($ERR_SOCKET_CLOSED());
       }
@@ -2230,6 +2299,13 @@ function getNodeHTTPServerSocket() {
             this.#pendingCallback = _callback;
             return false;
           }
+        } else if (this.#closeHandled === kOutlivesConnection) {
+          // The connection has closed, and the socket waits for its reader. The write fails like one that waited for a
+          // drain at the close: ERR_STREAM_DESTROYED to its callback, and no 'error'. Node fails it with EPIPE, and
+          // emits 'error'.
+          this.#pendingCallback = _callback;
+          process.nextTick(destroySocket, this);
+          return false;
         }
       } catch (e) {
         err = e;