node:http: the server follows Node (request body, framing, response finish, lifecycle)

packages/bun-usockets/src/bsd.c

@@ -23,6 +23,7 @@
 #include "internal/internal.h"
 #include "internal/fault_inject.h"
 
+#include <limits.h>
 #include <stdio.h>
 #include <stdlib.h>
 
@@ -941,48 +942,57 @@ ssize_t bsd_recvmsg(LIBUS_SOCKET_DESCRIPTOR fd, struct msghdr *msg, int flags) {
 }
 #endif
 
+#ifndef MSG_NOSIGNAL
+#define MSG_NOSIGNAL 0
+#endif
+/* Every write to a stream socket uses these, vectored or not. */
+#define BSD_SEND_FLAGS (MSG_NOSIGNAL | MSG_DONTWAIT)
+
 #if !defined(_WIN32)
 #include <sys/uio.h>
 
+/* sendmsg, not writev: writev takes no flags, so a reset peer would raise SIGPIPE on Linux. */
 ssize_t bsd_writev(LIBUS_SOCKET_DESCRIPTOR fd, const struct us_iovec_t *iov, int count) {
-    ssize_t injected = 0; int unused = 0;
-    if (US_FAULT_CHECK(US_FAULT_WRITEV, fd, injected, unused)) return injected;
-    (void)unused;
-    /* POSIX writev fails with EINVAL above IOV_MAX (1024 on Linux/macOS); cap and
+    /* sendmsg fails with EMSGSIZE above IOV_MAX (1024 on Linux/macOS); cap and
      * let the caller's partial-write handling carry the remainder. */
     if (count > 1024) {
         count = 1024;
     }
-    while (1) {
-        ssize_t written = writev(fd, (const struct iovec *)iov, count);
-        if (UNLIKELY(IS_EINTR(written))) {
-            continue;
+    struct msghdr msg = {0};
+    msg.msg_iov = (struct iovec *)iov;
+    msg.msg_iovlen = count;
+
+#if defined(LIBUS_SOCKET_FAULT_INJECTION) && LIBUS_SOCKET_FAULT_INJECTION
+    /* A "short" rule keeps the first `clamp` bytes of the list. */
+    struct iovec clamped[1024];
+    ssize_t injected = 0; int clamp = INT_MAX;
+    if (US_FAULT_CHECK(US_FAULT_WRITEV, fd, injected, clamp)) return injected;
+    if (clamp != INT_MAX) {
+        size_t left = (size_t)clamp;
+        int kept = 0;
+        for (; kept < count && left; kept++) {
+            clamped[kept].iov_base = iov[kept].iov_base;
+            clamped[kept].iov_len = iov[kept].iov_len < left ? iov[kept].iov_len : left;
+            left -= clamped[kept].iov_len;
         }
-        return written;
+        msg.msg_iov = clamped;
+        msg.msg_iovlen = kept;
     }
-}
-
-ssize_t bsd_write2(LIBUS_SOCKET_DESCRIPTOR fd, const char *header, int header_length, const char *payload, int payload_length) {
-    ssize_t injected = 0; int unused = 0;
-    if (US_FAULT_CHECK(US_FAULT_WRITEV, fd, injected, unused)) return injected;
-    (void)unused;
-    struct iovec chunks[2];
-
-    chunks[0].iov_base = (char *)header;
-    chunks[0].iov_len = header_length;
-    chunks[1].iov_base = (char *)payload;
-    chunks[1].iov_len = payload_length;
+#endif
 
     while (1) {
-        ssize_t written = writev(fd, chunks, 2);
-
+        ssize_t written = sendmsg(fd, &msg, BSD_SEND_FLAGS);
         if (UNLIKELY(IS_EINTR(written))) {
             continue;
         }
-
         return written;
     }
 }
+
+ssize_t bsd_write2(LIBUS_SOCKET_DESCRIPTOR fd, const char *header, int header_length, const char *payload, int payload_length) {
+    struct us_iovec_t chunks[2] = {{(void *)header, (size_t)header_length}, {(void *)payload, (size_t)payload_length}};
+    return bsd_writev(fd, chunks, 2);
+}
 #else
 ssize_t bsd_writev(LIBUS_SOCKET_DESCRIPTOR fd, const struct us_iovec_t *iov, int count) {
     ssize_t total = 0;
@@ -1010,14 +1020,7 @@ ssize_t bsd_send(LIBUS_SOCKET_DESCRIPTOR fd, const char *buf, int length) {
     ssize_t injected = 0;
     if (US_FAULT_CHECK(US_FAULT_SEND, fd, injected, length)) return injected;
     while (1) {
-    // MSG_MORE (Linux), MSG_PARTIAL (Windows), TCP_NOPUSH (BSD)
-
-#ifndef MSG_NOSIGNAL
-#define MSG_NOSIGNAL 0
-#endif
-
-        // use TCP_NOPUSH
-        ssize_t rc = send(fd, buf, length, MSG_NOSIGNAL | MSG_DONTWAIT);
+        ssize_t rc = send(fd, buf, length, BSD_SEND_FLAGS);
 
         if (UNLIKELY(IS_EINTR(rc))) {
             continue;

packages/bun-usockets/src/context.c

@@ -364,6 +364,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->hangup_closes_unsent = 0;
     s->next = 0;
     s->prev = 0;
     s->connect_state = NULL;
@@ -544,6 +545,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->hangup_closes_unsent = 0;
     s->connect_state = NULL;
     s->connect_next = NULL;
 }

packages/bun-usockets/src/crypto/openssl.c

@@ -2696,7 +2696,14 @@ unsigned int us_internal_ssl_spill_pending(struct us_socket_t *s) {
 }
 
 int us_internal_ssl_write(struct us_socket_t *s, const char *data, int length) {
-  if (us_socket_is_closed(s) || us_internal_ssl_is_shut_down(s) || length == 0) return 0;
+  struct us_iovec_t iov = {(void *)data, (size_t)length};
+  return us_internal_ssl_writev(s, &iov, 1);
+}
+
+/* The records of every part share one batch, so they reach the kernel in one write. Returns the plaintext bytes taken, in order. */
+int us_internal_ssl_writev(struct us_socket_t *s, const struct us_iovec_t *iov, int count) {
+  while (count && iov->iov_len == 0) iov++, count--;
+  if (us_socket_is_closed(s) || us_internal_ssl_is_shut_down(s) || count == 0) return 0;
 
   /* Fast-path connect attaches SSL eagerly on a SEMI_SOCKET (see
    * us_socket_group_connect_resolved_dns); on_open hasn't fired yet so
@@ -2743,12 +2750,17 @@ int us_internal_ssl_write(struct us_socket_t *s, const char *data, int length) {
 
   int total = 0;
   int last_ssl_written = 1;
-  while (total < length) {
-    int chunk = length - total;
-    if (chunk > 16384) chunk = 16384;
+  size_t part_offset = 0;
+  while (count) {
+    if (part_offset == iov->iov_len) {
+      iov++, count--, part_offset = 0;
+      continue;
+    }
+    size_t part_left = iov->iov_len - part_offset;
+    int chunk = part_left > 16384 ? 16384 : (int)part_left;
     /* Same deferred-close protocol as the SSL_do_handshake/SSL_read drivers. */
     s->ssl_in_use = 1;
-    last_ssl_written = SSL_write(s_ssl(s), data + total, chunk);
+    last_ssl_written = SSL_write(s_ssl(s), (const char *)iov->iov_base + part_offset, chunk);
     s->ssl_in_use = 0;
     if (s->ssl_pending_detach) {
       /* Closed from inside the call: drop this write's records (and any held
@@ -2761,6 +2773,7 @@ int us_internal_ssl_write(struct us_socket_t *s, const char *data, int length) {
     }
     if (last_ssl_written <= 0) break;
     total += last_ssl_written;
+    part_offset += (size_t)last_ssl_written;
     /* A batching allocation failure marks the socket fatal from inside the BIO;
      * stop sealing records for a connection that is being torn down. */
     if (s->ssl_fatal_error) break;

packages/bun-usockets/src/eventing/epoll_kqueue.c

@@ -374,12 +374,14 @@ static void us_internal_dispatch_ready_polls(struct us_loop_t *loop) {
                  * connect refused), or a read-filter EV_EOF carrying the
                  * socket error in fflags: both are epoll's EPOLLERR. */
                 if (bits.send_eof && !bits.send_eof_err && !bits.eof_err && type == POLL_TYPE_SOCKET &&
-                    ((struct us_socket_t *) poll)->flags.is_paused) {
+                    ((struct us_socket_t *) poll)->flags.is_paused &&
+                    !(((struct us_socket_t *) poll)->hangup_closes_unsent && ((struct us_socket_t *) poll)->flags.last_write_failed)) {
                     /* fflags==0 with reads paused is AF_UNIX's graceful peer
                      * close (TCP only gets SS_CANTSENDMORE from RST, which
                      * carries the error): the receive buffer survives, so
                      * defer like epoll's EPOLLHUP - resume() delivers the
-                     * tail + end instead of an error close discarding it. */
+                     * tail + end instead of an error close discarding it.
+                     * Not when a hangup_closes_unsent socket has bytes unsent. */
                     eof = 1;
                 } else {
                     error = 1;

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

@@ -61,7 +61,7 @@ enum us_fault_action {
     US_FAULT_NONE,
     /* return -1 and set errno = errno_value */
     US_FAULT_ERRNO,
-    /* recv/send: clamp the length to clamp_bytes, then run the real syscall.
+    /* recv/send/writev: clamp the length to clamp_bytes, then run the real syscall.
      * Other syscalls have no length to clamp; the JS setter rejects them. */
     US_FAULT_SHORT,
     /* recv/recvmsg: return 0 (peer closed); send/sendmsg/writev: return 0

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

@@ -257,6 +257,7 @@ int us_internal_ssl_handshake_callback_has_fired(us_socket_r s);
 int us_internal_ssl_is_shut_down(us_socket_r s);
 void us_internal_ssl_shutdown(us_socket_r s);
 int us_internal_ssl_write(us_socket_r s, const char *data, int length);
+int us_internal_ssl_writev(us_socket_r s, const struct us_iovec_t *iov, int count);
 unsigned int us_internal_ssl_spill_pending(us_socket_r s);
 void *us_internal_ssl_get_native_handle(us_socket_r s);
 struct us_bun_verify_error_t us_internal_ssl_verify_error(us_socket_r s);
@@ -348,6 +349,9 @@ 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 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;
   /* The close code passed to the deferred close (e.g. a reset requested from
    * inside a handshake callback must still RST, not FIN, when it is finally
    * performed). */

packages/bun-usockets/src/libusockets.h

@@ -648,6 +648,8 @@ struct us_iovec_t {
  * partial-write poll handling, one writev for all chunks (sequential sends
  * on platforms without writev). Returns total bytes written. */
 int us_socket_raw_writev(us_socket_r s, const struct us_iovec_t *iov, int count) nonnull_fn_decl;
+/* Vectored us_socket_write: through TLS if `s->ssl` is set, with the records of all chunks in one write. */
+int us_socket_writev(us_socket_r s, const struct us_iovec_t *iov, int count) nonnull_fn_decl;
 
 int us_socket_raw_write(us_socket_r s, const char *data, int length);
 /* Like us_socket_write, but additionally reports a fatal (non-would-block)

packages/bun-usockets/src/loop.c

@@ -548,6 +548,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->hangup_closes_unsent = 0;
 
                         /* We always use nodelay */
                         bsd_socket_nodelay(client_fd, 1);
@@ -872,9 +873,12 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in
              * left behind it. This includes sockets we already shut down (a client
              * that end()ed before reading the reply), the case that truncated; once
              * read_eof is set there is nothing left to drain and deferring would only
-             * lose the close. Error-flagged events keep the error path. */
+             * lose the close. Error-flagged events keep the error path. A hangup
+             * takes the write side down too, so a hangup_closes_unsent socket does
+             * not wait when the write this event retried failed again. */
             const int eof_deferrable = eof && s && !error && !us_socket_is_closed(s) && !s->read_eof;
-            if (eof_deferrable && s->flags.is_paused) {
+            const int unsent_is_lost = hangup && s && s->hangup_closes_unsent && s->flags.last_write_failed;
+            if (eof_deferrable && s->flags.is_paused && !unsent_is_lost) {
 #ifdef LIBUS_USE_EPOLL
                 /* EPOLLHUP is unmaskable: leave epoll while paused so it cannot re-fire; the unread tail stays in
                  * the kernel until resume() re-adds the fd via us_poll_change (end() while paused keeps it parked). */

packages/bun-usockets/src/socket.c

@@ -470,6 +470,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->hangup_closes_unsent = 0;
     s->connect_state = NULL;
 
     /* We always use nodelay */
@@ -620,6 +621,13 @@ int us_socket_write_check_error(struct us_socket_t *s, const char *data, int len
     return written;
 }
 
+int us_socket_writev(struct us_socket_t *s, const struct us_iovec_t *iov, int count) {
+    if (s->ssl) {
+        return us_internal_ssl_writev(s, iov, count);
+    }
+    return us_socket_raw_writev(s, iov, count);
+}
+
 int us_socket_raw_writev(struct us_socket_t *s, const struct us_iovec_t *iov, int count) {
     if (us_socket_is_closed(s) ||
         us_internal_poll_type(&s->p) == POLL_TYPE_SOCKET_SHUT_DOWN) {

packages/bun-uws/src/App.h

@@ -230,7 +230,7 @@ struct TemplatedApp {
      * then BACKPRESSURE beats SUCCESS. */
     PublishStatus publish(std::string_view topic, std::string_view message, OpCode opCode, bool compress = false) {
         /* Anything big bypasses corking efforts */
-        if (message.length() >= LoopData::CORK_BUFFER_SIZE) {
+        if (message.length() >= LoopData::CORK_COPY_MAX) {
             PublishStatus worst = PublishStatus::SUCCESS;
             bool hasReceivers = false;
             topicTree->publishBig(nullptr, topic, {message, opCode, compress}, [&worst, &hasReceivers](Subscriber *s, TopicTreeBigMessage &message) {
@@ -427,13 +427,7 @@ struct TemplatedApp {
              * adopted into its own group), so the ext block is an HttpResponseData. */
             auto *data = (HttpResponseData<SSL> *) ((AsyncSocket<SSL> *) s)->getAsyncSocketData();
             struct us_socket_t *next = s->next;
-            if (data->isIdle) {
-                /* A socket is idle from the moment its response completes. When
-                 * that happens inside onData's parse loop the response can still
-                 * sit in the cork buffer, and JS that runs before the loop ends
-                 * (a graceful stop() from a microtask) gets here. close() sends
-                 * it first. */
-                ((AsyncSocket<SSL> *) s)->close();
+            if (((HttpResponse<SSL> *) s)->closeIfIdle()) {
                 closed++;
             } else if (closeWhenIdle) {
                 data->state |= HttpResponseData<SSL>::HTTP_CLOSE_WHEN_IDLE;
@@ -809,6 +803,11 @@ struct TemplatedApp {
         return std::move(*this);
     }
 
+    TemplatedApp &&setMaxHeadersCount(uint32_t maxHeadersCount) {
+        httpContext->getSocketContextData()->maxHeadersCount = maxHeadersCount;
+        return std::move(*this);
+    }
+
 };
 
 typedef TemplatedApp<false> App;

packages/bun-uws/src/AsyncSocket.h

@@ -348,7 +348,7 @@ struct AsyncSocket {
             if (slot != LoopData::INVALID_CORK_SLOT) {
                 /* We are corked */
                 auto *s = loopData->getCorkSlot(slot);
-                if (LoopData::CORK_BUFFER_SIZE - s->offset >= (unsigned int) length) {
+                if ((unsigned int) length <= LoopData::CORK_COPY_MAX && LoopData::CORK_BUFFER_SIZE - s->offset >= (unsigned int) length) {
                     /* If the entire chunk fits in cork buffer */
                     memcpy(s->buffer + s->offset, src, (unsigned int) length);
                     s->offset += (unsigned int) length;
@@ -357,7 +357,7 @@ struct AsyncSocket {
                     /* Fall through to default return */
                 } else {
                     /* Chunk doesn't fit; flush cork + write the rest. */
-                    return uncork(src, length, optionally);
+                    return writeFramed({}, src, length, {}, optionally);
                 }
             } else {
                 /* We are not corked */
@@ -389,6 +389,65 @@ struct AsyncSocket {
         return {length, false};
     }
 
+    /* write(head), write(src), write(tail) in one writev when they do not fit the cork buffer. The framing is never optional, and tail follows only a complete src. Returns like write(src). */
+    std::pair<int, bool> writeFramed(std::string_view head, const char *src, int length, std::string_view tail, bool optionally = false) {
+        LoopData *loopData = getLoopData();
+        BackPressure &backpressure = getAsyncSocketData()->buffer;
+        int slot = loopData->findCorkSlot(this);
+        unsigned int corked = slot != LoopData::INVALID_CORK_SLOT ? loopData->getCorkSlot(slot)->offset : 0;
+        const bool fitsCork = slot != LoopData::INVALID_CORK_SLOT && (unsigned int) length <= LoopData::CORK_COPY_MAX
+            && LoopData::CORK_BUFFER_SIZE - corked >= head.length() + (size_t) length + tail.length();
+
+        /* macOS refuses a writev of more than INT_MAX bytes. */
+        const bool tooLong = (size_t) length > (size_t) INT_MAX - LoopData::CORK_BUFFER_SIZE - head.length() - tail.length();
+
+        if (fitsCork || tooLong || backpressure.length() || us_socket_is_closed((us_socket_t *) this)) {
+            if (head.length()) {
+                write(head.data(), (int) head.length());
+            }
+            /* write() sends a src that does not fit the cork buffer back here without framing. */
+            const bool fromWrite = head.empty() && tail.empty() && slot != LoopData::INVALID_CORK_SLOT && !fitsCork;
+            auto result = fromWrite ? uncork(src, length, optionally) : write(src, length, optionally);
+            if (tail.length() && result.first == length) {
+                result.second |= write(tail.data(), (int) tail.length()).second;
+            }
+            return result;
+        }
+
+        const char *corkBuffer = nullptr;
+        if (slot != LoopData::INVALID_CORK_SLOT) {
+            corkBuffer = loopData->getCorkSlot(slot)->buffer;
+            loopData->releaseCorkSlot(slot);
+        }
+
+        const std::string_view parts[4] = {{corkBuffer, corked}, head, {src, (size_t) length}, tail};
+        struct us_iovec_t iov[4];
+        int count = 0;
+        for (auto part : parts) {
+            if (part.length()) {
+                iov[count++] = {(void *) part.data(), part.length()};
+            }
+        }
+        size_t written = (size_t) us_socket_writev((us_socket_t *) this, iov, count);
+
+        bool failed = false;
+        for (int i = 0; i < 4; i++) {
+            std::string_view part = parts[i];
+            size_t sent = std::min(written, part.length());
+            written -= sent;
+            if (sent == part.length()) {
+                continue;
+            }
+            failed = true;
+            if (optionally && i == 2) {
+                /* The caller keeps the rest of src, and sends tail behind it. */
+                return {(int) sent, true};
+            }
+            backpressure.append(part.data() + sent, part.length() - sent);
+        }
+        return {length, failed};
+    }
+
     /* Uncork this socket and flush or buffer any corked and/or passed data. It is essential to remember doing this. */
     /* It does NOT count bytes written from cork buffer (they are already accounted for in the write call responsible for its corking)! */
     std::pair<int, bool> uncork(const char *src = nullptr, int length = 0, bool optionally = false) {