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

@@ -37,10 +37,26 @@ int us_internal_libuv_peer_reset_probe(LIBUS_SOCKET_DESCRIPTOR fd) {
   if (send(fd, "", 0, 0) != SOCKET_ERROR) {
     return 0;
   }
-  int err = WSAGetLastError();
-  /* WSAESHUTDOWN means our own shutdown(SD_SEND) ran; that is not a peer
-   * reset. The fin_deferred sweep probes sockets after local shutdown. */
-  return err != WSAEWOULDBLOCK && err != WSAESHUTDOWN;
+  /* Allowlist of definitely-peer-gone codes, the WSA mirror of
+   * us_internal_send_errno_is_peer_gone (socket.c). A transient failure must
+   * not count - WSAENOBUFS is documented transient-on-healthy
+   * (bsd_send_is_transient_error) - because the sweep re-probes every 4s, so
+   * a missed detection self-heals, while a false positive reset-closes a
+   * healthy paused connection. WSAESHUTDOWN is our own shutdown(SD_SEND),
+   * not the peer; the sweep probes sockets after local shutdown. */
+  switch (WSAGetLastError()) {
+  case WSAECONNRESET:
+  case WSAECONNABORTED:
+  case WSAENETRESET:
+  case WSAENOTCONN:
+  case WSAETIMEDOUT:
+  case WSAENETDOWN:
+  case WSAENETUNREACH:
+  case WSAEHOSTUNREACH:
+    return 1;
+  default:
+    return 0;
+  }
 }
 
 static struct us_socket_t *us_internal_poll_cb_adopted_socket(struct us_poll_t *wp) {
@@ -103,14 +119,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,179 @@ describe.concurrent("connect() failure promise settlement", () => {
     ).rejects.toBe(boom);
   });
 });
+
+// A paused socket polls without READABLE, so a peer FIN 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 in these tests are structural: the
+// sweep runs on a fixed 4s cadence, and the deferred-FIN window asserts the
+// absence of a close across one full sweep.
+function pausedVictimPair() {
+  const state = {
+    victimClosed: Promise.withResolvers<string>(),
+    closedHow: null as string | null,
+    closeError: undefined as unknown,
+    endFired: false,
+    dataReceived: "",
+  };
+  const victimOpen = Promise.withResolvers<Socket>();
+  const server = Bun.listen({
+    hostname: "127.0.0.1",
+    port: 0,
+    socket: {
+      open(s) {
+        s.pause(); // receive backpressure before any bytes flow
+        victimOpen.resolve(s);
+      },
+      data(_s, buf) {
+        state.dataReceived += buf.toString();
+      },
+      end() {
+        state.endFired = true;
+      },
+      error(_s, e) {
+        state.closedHow ??= "error";
+        state.closeError = e;
+        state.victimClosed.resolve("error");
+      },
+      close(_s, e) {
+        state.closedHow ??= "close";
+        state.closeError ??= e;
+        state.victimClosed.resolve("close");
+      },
+    },
+  });
+  const peer = (async () => {
+    const peerOpened = Promise.withResolvers<Socket>();
+    await Bun.connect({
+      hostname: "127.0.0.1",
+      port: server.port,
+      socket: {
+        open(s) {
+          peerOpened.resolve(s);
+        },
+        data() {},
+        end() {},
+        error() {},
+        close() {},
+      },
+    });
+    return peerOpened.promise;
+  })();
+  return { state, server, victimOpen: victimOpen.promise, peer };
+}
+
+// Exercises the peeked == 0 path in poll_cb: the FIN lands on an empty
+// receive buffer.
+it.concurrent.skipIf(!isWindows)(
+  "paused socket with a deferred empty-buffer FIN still closes when the connection later dies",
+  async () => {
+    const { state, server, victimOpen, peer: peerP } = pausedVictimPair();
+    using _server = server;
+    const peer = await peerP;
+    const victim = await victimOpen;
+
+    // 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(state.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([state.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(state.endFired).toBe(false);
+  },
+  40_000,
+);
+
+// Exercises the pre-existing peeked > 0 latch in poll_cb (FIN deferred behind
+// buffered data), which the fin_deferred init fix first makes reachable: the
+// garbage-broken count kept the sweep gate closed, so this path had never
+// actually run.
+it.concurrent.skipIf(!isWindows)(
+  "paused socket with a FIN deferred behind buffered data still closes when the connection later dies",
+  async () => {
+    const { state, server, victimOpen, peer: peerP } = pausedVictimPair();
+    using _server = server;
+    const peer = await peerP;
+    const victim = await victimOpen;
+
+    await Bun.sleep(200);
+
+    // Data, then a clean FIN: the victim is paused, so the bytes sit in its
+    // kernel receive buffer and MSG_PEEK sees them at the DISCONNECT.
+    peer.write("data-behind-fin");
+    peer.flush();
+    await Bun.sleep(50);
+    peer.shutdown();
+
+    await Bun.sleep(4600);
+    expect(state.closedHow).toBeNull();
+
+    peer.terminate();
+    await Bun.sleep(100);
+    victim.write("x");
+
+    const result = await Promise.race([state.victimClosed.promise, Bun.sleep(12_000).then(() => "stranded")]);
+    expect(result).not.toBe("stranded");
+    // Node parity for a paused socket whose peer died: the reset wins; the
+    // buffered bytes and the end are never delivered.
+    expect(state.dataReceived).toBe("");
+    expect(state.endFired).toBe(false);
+  },
+  40_000,
+);
+
+// The resume side of the contract: a deferred FIN (and the data in front of
+// it) is delivered once the socket resumes, and the close is clean. Resume
+// also hands the socket back from the sweep (the fin_deferred count returns
+// to zero), so a wrong count here would silently re-break the sweep gate.
+it.concurrent.skipIf(!isWindows)(
+  "resuming a paused socket delivers the data and FIN that were deferred while paused",
+  async () => {
+    const { state, server, victimOpen, peer: peerP } = pausedVictimPair();
+    using _server = server;
+    const peer = await peerP;
+    const victim = await victimOpen;
+
+    await Bun.sleep(200);
+
+    peer.write("deferred-data");
+    peer.flush();
+    await Bun.sleep(50);
+    peer.shutdown();
+
+    // Let the FIN's DISCONNECT report arrive and defer while paused.
+    await Bun.sleep(300);
+    expect(state.closedHow).toBeNull();
+
+    victim.resume();
+
+    const result = await Promise.race([state.victimClosed.promise, Bun.sleep(12_000).then(() => "stranded")]);
+    expect(result).not.toBe("stranded");
+    expect(state.dataReceived).toBe("deferred-data");
+    expect(state.endFired).toBe(true);
+    // Clean shutdown, not a reset.
+    expect(state.closeError).toBeFalsy();
+  },
+  30_000,
+);