usockets(win): surface a dead connection behind a paused socket's deferred empty-buffer FIN

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.adopted = 0;
     s->flags.allow_half_open = (options & LIBUS_SOCKET_ALLOW_HALF_OPEN);
     s->unclassified_send_failures = 0;
+    s->fin_deferred = 0;
     s->next = 0;
     s->prev = 0;
     s->connect_state = NULL;
@@ -508,6 +509,7 @@ static inline void us_internal_init_connect_socket(struct us_socket_t *s,
     s->flags.adopted = 0;
     s->flags.last_write_failed = 0;
     s->unclassified_send_failures = 0;
+    s->fin_deferred = 0;
     s->connect_state = NULL;
     s->connect_next = NULL;
 }

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

@@ -103,14 +103,23 @@ static void poll_cb(uv_poll_t *p, int status, int events) {
        * until resume; pending data keeps the pause honored untouched. */
       char probe;
       ssize_t peeked = bsd_recv(us_poll_fd(wp), &probe, 1, MSG_PEEK);
+      struct us_socket_t *sock = us_internal_poll_cb_adopted_socket(wp);
       if (peeked == 0) {
+        /* Graceful FIN with nothing buffered: the shared dispatch defers the
+         * eof until resume (paused-EOF contract), which leaves this socket in
+         * the same consumed-DISCONNECT state as the data-deferred branch
+         * below - a LATER reset has no event left to ride. Hand it to the
+         * sweep as well. */
+        if (!sock->fin_deferred) {
+          sock->fin_deferred = 1;
+          sock->group->loop->data.fin_deferred_count++;
+        }
         eof = 1;
         events |= UV_READABLE;
       } else if (peeked < 0 && !bsd_would_block()) {
         error = 1;
         events |= UV_READABLE;
       } else if (peeked > 0) {
-        struct us_socket_t *sock = us_internal_poll_cb_adopted_socket(wp);
         if (us_socket_get_error(sock) != 0 || us_internal_libuv_peer_reset_probe(us_poll_fd(wp))) {
           /* Data is buffered ahead of whatever ended the connection. If the
            * peer ABORTED, the kernel already discarded the stream's tail and

packages/bun-usockets/src/loop.c

@@ -540,6 +540,7 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in
                         s->flags.adopted = 0;
                         s->flags.last_write_failed = 0;
                         s->unclassified_send_failures = 0;
+                        s->fin_deferred = 0;
 
                         /* We always use nodelay */
                         bsd_socket_nodelay(client_fd, 1);

packages/bun-usockets/src/socket.c

@@ -360,6 +360,15 @@ __attribute__((always_inline)) struct us_socket_t *us_socket_close(struct us_soc
 // - does not emit on_close event
 // - does not close
 struct us_socket_t *us_socket_detach(struct us_socket_t *s) {
+#ifdef LIBUS_USE_LIBUV
+    /* The fd leaves usockets' management, so the sweep must forget it
+     * (mirrors us_internal_socket_close_raw; a stale flag would leak
+     * fin_deferred_count and keep the sweep walking forever). */
+    if (s->fin_deferred) {
+        s->fin_deferred = 0;
+        s->group->loop->data.fin_deferred_count--;
+    }
+#endif
     if (!us_socket_is_closed(s)) {
         struct us_loop_t *loop = s->group->loop;
 
@@ -457,6 +466,7 @@ struct us_socket_t *us_socket_from_fd(struct us_socket_group_t *group, unsigned
     s->flags.adopted = 0;
     s->flags.last_write_failed = 0;
     s->unclassified_send_failures = 0;
+    s->fin_deferred = 0;
     s->connect_state = NULL;
 
     /* We always use nodelay */

test/js/bun/net/socket.test.ts

@@ -3667,3 +3667,89 @@ describe.concurrent("connect() failure promise settlement", () => {
     ).rejects.toBe(boom);
   });
 });
+
+// A paused socket polls without READABLE, so a peer FIN arriving while the
+// receive buffer is empty is only reported through AFD's one-shot DISCONNECT.
+// The eof is deferred until resume (pause contract), which consumes the only
+// event the poll had; when the connection later dies, nothing is subscribed
+// that could report it, so the 4s sweep has to probe the socket. Writing one
+// byte into the dead connection resets the victim's TCB the same way a remote
+// peer's RST segment would (Windows emits no RST from FIN_WAIT_2 on loopback,
+// so the peer's abort alone is invisible to an idle victim here). The sleeps
+// are structural: the sweep runs on a fixed 4s cadence, and the FIN window
+// asserts the absence of a close across one full sweep.
+it.concurrent.skipIf(!isWindows)(
+  "paused socket with a deferred empty-buffer FIN still closes when the connection later dies",
+  async () => {
+    const victimClosed = Promise.withResolvers<string>();
+    const victimOpen = Promise.withResolvers<Socket>();
+    let closedHow: string | null = null;
+    let endFired = false;
+
+    using server = Bun.listen({
+      hostname: "127.0.0.1",
+      port: 0,
+      socket: {
+        open(s) {
+          s.pause(); // receive backpressure; rx buffer stays empty
+          victimOpen.resolve(s);
+        },
+        data() {},
+        end() {
+          endFired = true;
+        },
+        error() {
+          closedHow ??= "error";
+          victimClosed.resolve("error");
+        },
+        close() {
+          closedHow ??= "close";
+          victimClosed.resolve("close");
+        },
+      },
+    });
+
+    const peerOpened = Promise.withResolvers<Socket>();
+    const peer = await Bun.connect({
+      hostname: "127.0.0.1",
+      port: server.port,
+      socket: {
+        open(s) {
+          peerOpened.resolve(s);
+        },
+        data() {},
+        end() {},
+        error() {},
+        close() {},
+      },
+    });
+    await peerOpened.promise;
+    const victim = await victimOpen.promise;
+
+    // Let the pause settle: the writable dispatch drops WRITABLE, leaving the
+    // victim's poll subscribed to DISCONNECT only.
+    await Bun.sleep(200);
+
+    // Clean FIN with the victim's receive buffer empty.
+    peer.shutdown();
+
+    // A full sweep period passes: the deferred FIN alone must not close the
+    // paused victim (its peer is alive and half-closed; the sweep's probe has
+    // to keep it deferred).
+    await Bun.sleep(4600);
+    expect(closedHow).toBeNull();
+
+    // Peer dies; the victim streams on, and the byte's RST reply resets its
+    // TCB. Without the sweep escalation nothing is ever delivered and the
+    // socket strands, so a bounded race is the condition check.
+    peer.terminate();
+    await Bun.sleep(100);
+    victim.write("x");
+
+    const result = await Promise.race([victimClosed.promise, Bun.sleep(12_000).then(() => "stranded")]);
+    expect(result).not.toBe("stranded");
+    // The deferred eof must not have been delivered as end; the socket died.
+    expect(endFired).toBe(false);
+  },
+  40_000,
+);