Bun.serve: do not idle-time-out a response the peer is still reading

packages/bun-usockets/src/bsd.c

@@ -680,6 +680,62 @@ int bsd_socket_buffer_size(LIBUS_SOCKET_DESCRIPTOR fd, int is_recv, int size, in
     return 0;
 }
 
+int bsd_socket_send_progress_mark(LIBUS_SOCKET_DESCRIPTOR fd, uint64_t *mark) {
+    /* A monotonic count of bytes this socket has delivered toward the peer.
+     * A queue-depth reading (SIOCOUTQ) is wrong here: a producer that refills
+     * the send buffer as fast as the peer drains it holds the depth constant,
+     * so the delta reads as no progress even while the peer reads steadily.
+     * The cumulative counter always grows with real throughput. */
+#if defined(_WIN32)
+    TCP_INFO_v0 info;
+    DWORD version = 0, bytes = 0;
+    memset(&info, 0, sizeof(info));
+    if (WSAIoctl(fd, SIO_TCP_INFO, &version, sizeof(version), &info, sizeof(info), &bytes, NULL, NULL) != 0) {
+        return -1;
+    }
+    *mark = info.BytesOut;
+    return 0;
+#elif defined(__APPLE__)
+    struct tcp_connection_info info;
+    socklen_t len = sizeof(info);
+    memset(&info, 0, sizeof(info));
+    if (getsockopt(fd, IPPROTO_TCP, TCP_CONNECTION_INFO, &info, &len) != 0) {
+        return -1;
+    }
+    *mark = info.tcpi_txbytes;
+    return 0;
+#elif defined(__linux__)
+    /* The kernel's struct tcp_info is append-only (uapi/linux/tcp.h). Some libc
+     * headers predate tcpi_bytes_acked, so match the ABI prefix up to it by
+     * hand rather than depend on the header: 8 bytes of __u8 fields, then 24
+     * __u32, then the __u64 block whose third member is tcpi_bytes_acked
+     * (RFC4898 tcpEStatsAppHCThruOctetsAcked, Linux 4.6+). */
+    struct {
+        uint8_t u8[8];
+        uint32_t u32[24];
+        uint64_t pacing_rate;
+        uint64_t max_pacing_rate;
+        uint64_t bytes_acked;
+    } info;
+    socklen_t len = sizeof(info);
+    memset(&info, 0, sizeof(info));
+    if (getsockopt(fd, IPPROTO_TCP, TCP_INFO, &info, &len) != 0) {
+        return -1;
+    }
+    /* An older kernel returns a shorter struct and leaves bytes_acked at 0,
+     * which reads as no progress (the old close-on-timeout behaviour). */
+    if (len < sizeof(info)) {
+        return -1;
+    }
+    *mark = info.bytes_acked;
+    return 0;
+#else
+    (void) fd;
+    (void) mark;
+    return -1;
+#endif
+}
+
 void bsd_socket_flush(LIBUS_SOCKET_DESCRIPTOR fd) {
     // Linux TCP_CORK has the same underlying corking mechanism as with MSG_MORE
 #ifdef TCP_CORK

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

@@ -190,6 +190,11 @@ int bsd_socket_get_tos(LIBUS_SOCKET_DESCRIPTOR fd);
  * resulting value to *out; on failure returns the setsockopt/getsockopt result
  * with the error left in errno (WSAGetLastError on Windows). */
 int bsd_socket_buffer_size(LIBUS_SOCKET_DESCRIPTOR fd, int is_recv, int size, int *out);
+/* See us_socket_send_progress_mark. A monotonic delivered-bytes counter: Linux
+ * TCP_INFO tcpi_bytes_acked, macOS tcpi_txbytes, Windows SIO_TCP_INFO BytesOut.
+ * Returns 0 and writes the mark, or -1 when the platform or socket kind cannot
+ * answer. */
+int bsd_socket_send_progress_mark(LIBUS_SOCKET_DESCRIPTOR fd, uint64_t *mark);
 int bsd_prepare_adopted_udp_socket(LIBUS_SOCKET_DESCRIPTOR fd);
 int bsd_set_reuseaddr(LIBUS_SOCKET_DESCRIPTOR fd);
 int bsd_bind_udp_fd(LIBUS_SOCKET_DESCRIPTOR fd, const struct sockaddr *addr, int addrlen, int flags);

packages/bun-usockets/src/libusockets.h

@@ -677,6 +677,9 @@ void us_socket_shutdown(us_socket_r s) nonnull_fn_decl;
 void us_socket_shutdown_read(us_socket_r s) nonnull_fn_decl;
 int us_socket_is_shut_down(us_socket_r s) nonnull_fn_decl;
 int us_socket_is_closed(us_socket_r s) nonnull_fn_decl;
+/* The last write was short (the kernel send buffer is full) and the socket
+ * polls for a writable event. Cleared when that event is dispatched. */
+int us_socket_is_awaiting_writable(us_socket_r s) nonnull_fn_decl;
 int us_socket_is_ssl_handshake_finished(us_socket_r s) nonnull_fn_decl;
 int us_socket_ssl_handshake_callback_has_fired(us_socket_r s) nonnull_fn_decl;
 /* TLS ciphertext bytes already sealed for this socket and reported as
@@ -724,6 +727,16 @@ int us_socket_get_error(us_socket_r s);
  * making a stall there routine backpressure, so it asks the kernel
  * (SO_ERROR, then a zero-byte send probe). */
 int us_socket_stalled_write_means_peer_gone(us_socket_r s);
+/* A monotonic count of bytes this socket has delivered toward the peer (Linux
+ * TCP_INFO tcpi_bytes_acked, macOS tcpi_txbytes, Windows SIO_TCP_INFO BytesOut).
+ * Returns 0 and writes the mark, or -1 where the platform or socket kind cannot
+ * answer. Take it when an idle timer is armed with outgoing bytes pending, so
+ * that the timeout can tell a slow reader from a stalled one with
+ * us_socket_send_progressed. */
+int us_socket_send_progress_mark(us_socket_r s, uint64_t *mark);
+/* Did the kernel move at least min_bytes toward the peer since *mark was
+ * taken, with no write in between? Advances *mark when it did. */
+int us_socket_send_progressed(us_socket_r s, uint64_t *mark, uint64_t min_bytes);
 
 void us_socket_ref(us_socket_r s);
 void us_socket_unref(us_socket_r s);

packages/bun-usockets/src/socket.c

@@ -422,6 +422,28 @@ int us_socket_stalled_write_means_peer_gone(struct us_socket_t *s) {
 #endif
 }
 
+int us_socket_send_progress_mark(struct us_socket_t *s, uint64_t *mark) {
+    if (us_socket_is_closed(s)) {
+        return -1;
+    }
+    return bsd_socket_send_progress_mark(us_poll_fd(&s->p), mark);
+}
+
+int us_socket_send_progressed(struct us_socket_t *s, uint64_t *mark, uint64_t min_bytes) {
+    uint64_t now;
+    if (us_socket_send_progress_mark(s, &now) != 0) {
+        return 0;
+    }
+    /* The mark is a monotonic delivered-bytes counter (see
+     * bsd_socket_send_progress_mark), so progress is how much it grew. */
+    uint64_t delivered = now > *mark ? now - *mark : 0;
+    if (delivered < min_bytes) {
+        return 0;
+    }
+    *mark = now;
+    return 1;
+}
+
 int us_socket_write2(struct us_socket_t *s, const char *header, int header_length, const char *payload, int payload_length) {
     if (us_socket_is_closed(s) || us_socket_is_shut_down(s)) {
         return 0;
@@ -710,6 +732,10 @@ __attribute__((always_inline)) int us_socket_is_shut_down(struct us_socket_t *s)
     return us_internal_poll_type(&s->p) == POLL_TYPE_SOCKET_SHUT_DOWN;
 }
 
+int us_socket_is_awaiting_writable(struct us_socket_t *s) {
+    return s->flags.last_write_failed;
+}
+
 int us_connecting_socket_is_shut_down(struct us_connecting_socket_t *c) {
     return c->shutdown;
 }

packages/bun-uws/src/Http2Context.h

@@ -427,6 +427,10 @@ struct Http2Connection {
     /* Seconds of silence before onTimeout; recomputed when a stream sets a
      * timeout or retires (see Http2Response::timeoutS). */
     unsigned idleTimeoutS = 0;
+    /* The idle timer was armed (touch) with outgoing bytes pending, and
+     * sendProgressMark holds the kernel's delivery mark from that moment. */
+    bool sendProgressMarked = false;
+    uint64_t sendProgressMark = 0;
     int drainDepth = 0;
 
     Http2Connection(us_socket_t *socket, Http2Context *context) : s(socket), ctx(context) {
@@ -959,6 +963,14 @@ struct Http2Context {
      * connection): streams in flight are aborted through onClose. */
     static us_socket_t *onTimeout(us_socket_t *s) {
         Http2Connection *conn = connection(s);
+        /* Same rule as HttpContext::onTimeout: a peer that took the pending
+         * response bytes at the minimum receive rate is slow, not idle. */
+        if (conn->sendProgressMarked
+            && us_socket_send_progressed(s, &conn->sendProgressMark,
+                                         (uint64_t) HttpContext<false>::HTTP_RECEIVE_THROUGHPUT_BYTES * conn->idleTimeoutS)) {
+            us_socket_timeout(s, conn->idleTimeoutS);
+            return s;
+        }
         conn->busy++;
         std::vector<Http2Response *> open = conn->streams;
         for (Http2Response *stream : open) {
@@ -1034,6 +1046,9 @@ inline void Http2Connection::writeSettings() {
 
 inline void Http2Connection::touch() {
     us_socket_timeout(s, idleTimeoutS);
+    /* See HttpResponse::resetTimeout. */
+    sendProgressMarked = (out->length() > 0 || us_socket_is_awaiting_writable(s))
+        && us_socket_send_progress_mark(s, &sendProgressMark) == 0;
 }
 
 inline void Http2Connection::recomputeIdleTimeout() {

packages/bun-uws/src/HttpContext.h

@@ -108,15 +108,17 @@ struct HttpContext {
     us_socket_group_t group{};
     HttpContextData<SSL> data;
 
-    /* Minimum allowed receive throughput per second (clients uploading less than 16kB/sec get dropped) */
-    static constexpr int HTTP_RECEIVE_THROUGHPUT_BYTES = 16 * 1024;
-
     /* Not constexpr — the ordinals are linked from `src/uws_sys/SocketKind.rs`
      * so a reorder there can't silently mis-route us. Only ever read
      * at runtime (listen/adopt). */
     static unsigned char socketKind() { return SSL ? US_SOCKET_KIND_UWS_HTTP_TLS : US_SOCKET_KIND_UWS_HTTP; }
 
 public:
+    /* Minimum allowed throughput per second in either direction: a client that
+     * uploads, or takes a response, slower than 16kB/sec is idle. Shared with
+     * Http2Context::onTimeout. */
+    static constexpr int HTTP_RECEIVE_THROUGHPUT_BYTES = 16 * 1024;
+
     /* node:http flood prevention: re-feed parked request bytes through the same
      * parse path fresh socket data takes. The caller guarantees the buffer has
      * LIBUS_RECV_BUFFER_PADDING of writable slack past `length`. */
@@ -945,6 +947,17 @@ struct HttpContext {
         // Node.js by default closes the connection but they emit the timeout event before that
         HttpResponseData<SSL> *httpResponseData = reinterpret_cast<HttpResponseData<SSL> *>(asyncSocket->getAsyncSocketData());
 
+        /* The timer was armed with outgoing bytes pending (see
+         * HttpResponse::resetTimeout). A peer that took them at the same
+         * minimum rate the receive side demands (onData) is a slow reader,
+         * not an idle one: give it another period. */
+        if ((httpResponseData->state & HttpResponseData<SSL>::HTTP_SEND_PROGRESS_MARKED)
+            && us_socket_send_progressed(s, &httpResponseData->sendProgressMark,
+                                         (uint64_t) HTTP_RECEIVE_THROUGHPUT_BYTES * httpResponseData->idleTimeout)) {
+            asyncSocket->timeout(httpResponseData->idleTimeout);
+            return s;
+        }
+
         if (httpResponseData->onTimeout) {
             httpResponseData->onTimeout((HttpResponse<SSL> *)s, httpResponseData->userData);
         }

packages/bun-uws/src/HttpResponse.h

@@ -54,15 +54,28 @@ struct HttpResponse : public AsyncSocket<SSL> {
     }
 
     void setTimeout(uint8_t seconds) {
-        auto* data = getHttpResponseData();
-        data->idleTimeout = seconds;
-        Super::timeout(data->idleTimeout);
+        getHttpResponseData()->idleTimeout = seconds;
+        resetTimeout();
     }
 
     void resetTimeout() {
         auto* data = getHttpResponseData();
-
         Super::timeout(data->idleTimeout);
+
+        /* Outgoing bytes the peer has not taken yet (the kernel refused part of
+         * the last write, or uWS holds a remainder) drain at the peer's pace,
+         * with no write on our side to re-arm the timer and a writable event
+         * only once a good part of the kernel send buffer is free. Remember how
+         * far the kernel got, so that HttpContext::onTimeout can tell a slow
+         * reader from a stalled one. */
+        uint64_t mark;
+        if ((Super::getBufferedAmount() > 0 || us_socket_is_awaiting_writable((us_socket_t *) this))
+            && us_socket_send_progress_mark((us_socket_t *) this, &mark) == 0) {
+            data->sendProgressMark = mark;
+            data->state |= HttpResponseData<SSL>::HTTP_SEND_PROGRESS_MARKED;
+        } else {
+            data->state &= ~HttpResponseData<SSL>::HTTP_SEND_PROGRESS_MARKED;
+        }
     }
     /* Write an unsigned 32-bit integer in hex */
     void writeUnsignedHex(unsigned int value) {

packages/bun-uws/src/HttpResponseData.h

@@ -150,6 +150,10 @@ struct HttpResponseData : AsyncSocketData<SSL>, HttpParser {
          * shutdown sweep; the shouldCloseConnection() gates act on it once the
          * in-flight work completes. */
         HTTP_CLOSE_WHEN_IDLE = 1 << 17,
+        /* The idle timer was armed while outgoing bytes were pending, and
+         * sendProgressMark holds the kernel's delivery mark from that moment
+         * (HttpResponse::resetTimeout). onTimeout compares against it. */
+        HTTP_SEND_PROGRESS_MARKED = 1 << 18,
 
         /* Bits that describe the connection rather than the response in flight.
          * There is one HttpResponseData per socket, reused by every request on a
@@ -197,6 +201,9 @@ struct HttpResponseData : AsyncSocketData<SSL>, HttpParser {
     OnTimeoutCallback onTimeout = nullptr;
     /* Outgoing offset */
     uint64_t offset = 0;
+    /* us_socket_send_progress_mark when the idle timer was last armed or
+     * extended, valid while HTTP_SEND_PROGRESS_MARKED is set. */
+    uint64_t sendProgressMark = 0;
 
     /* Let's track number of bytes since last timeout reset in data handler */
     unsigned int received_bytes_per_timeout = 0;

src/uws_sys/libuwsockets.cpp

@@ -1513,15 +1513,20 @@ size_t uws_req_get_header(uws_req_t *res, const char *lower_case_header,
 
   void uws_res_uncork(int ssl, uws_res_r res)
   {
+    /* resetTimeout on a short write records the pending bytes for onTimeout. */
     if (ssl)
     {
       uWS::HttpResponse<true> *uwsRes = (uWS::HttpResponse<true> *)res;
-      uwsRes->uncork();
+      if (uwsRes->uncork().second && !us_socket_is_closed((struct us_socket_t *)res)) {
+        uwsRes->resetTimeout();
+      }
     }
     else
     {
       uWS::HttpResponse<false> *uwsRes = (uWS::HttpResponse<false> *)res;
-      uwsRes->uncork();
+      if (uwsRes->uncork().second && !us_socket_is_closed((struct us_socket_t *)res)) {
+        uwsRes->resetTimeout();
+      }
     }
   }
 
@@ -1535,6 +1540,8 @@ size_t uws_req_get_header(uws_req_t *res, const char *lower_case_header,
      * half-open socket. */
     us_poll_change(&s->p, s->group->loop,
                    LIBUS_SOCKET_WRITABLE | ((s->flags.is_paused || s->read_eof) ? 0 : LIBUS_SOCKET_READABLE));
+    /* resetTimeout records the sendfile remainder for onTimeout. */
+    ((uWS::HttpResponse<false> *)res)->resetTimeout();
   }
 
 __attribute__((callback (corker, ctx)))