fix(net): hold a connect to a full accept queue until accept makes room, as Linux does

src/net-node.ts

@@ -34,6 +34,9 @@ function parseIp(s: string | undefined): Uint8Array {
 
 const ipString = (a: Uint8Array) => (E.isWildcard(a) ? "0.0.0.0" : `${a[0]}.${a[1]}.${a[2]}.${a[3]}`);
 const peerString = (a: Uint8Array) => (E.isWildcard(a) ? "127.0.0.1" : ipString(a));
+// whether a is an IPv4 address of one of this host's interfaces
+const hostHas = (a: Uint8Array) =>
+  Object.values(os.networkInterfaces()).some((list) => list?.some((i) => i.family === "IPv4" && i.address === ipString(a)));
 
 function sysfs(name: string, file: string): number | null {
   try {
@@ -54,6 +57,22 @@ interface TcpImpl {
   srv?: net.Server;
   werr?: number;
   resume?: () => void;
+  // listener: connections the host accepted beyond the backlog, and the clients in this VM connecting to it by local port
+  overflow?: { c: Sock; pairing?: Pairing }[];
+  clients?: Map<number, Pairing>;
+  // client: its connect to a listener of this VM
+  pairing?: Pairing;
+}
+
+// A connect from this VM to a listener of this VM finishes once the listener's accept queue
+// admits the connection, as a kernel that drops the SYN of a full queue makes it.
+interface Pairing {
+  client: Sock;
+  listener: Sock;
+  port: number;
+  connected: boolean;
+  admitted: boolean;
+  finish?: () => void;
 }
 
 interface UdpImpl {
@@ -192,14 +211,21 @@ export class NodeNet implements NetBackend {
           c.peerInVm = client !== undefined;
         }
         this.attach(c, ns);
-        s.backlog.push(c);
+        const im = s.impl as TcpImpl;
+        const pairing = im.clients?.get(ns.remotePort ?? -1);
+        // the queue holds maxBacklog + 1, as Linux's does
+        if (s.backlog.length > s.maxBacklog) im.overflow!.push({ c, pairing });
+        else {
+          s.backlog.push(c);
+          if (pairing) this.admit(pairing);
+        }
         s.notify();
       });
       srv.listen(options, () => {
         srv.removeAllListeners("error");
         srv.on("error", () => {});
         bound(srv);
-        s.impl = { srv } satisfies TcpImpl;
+        s.impl = { srv, overflow: [], clients: new Map() } satisfies TcpImpl;
         this.live.add(s);
         resolve(0);
       });
@@ -233,11 +259,38 @@ export class NodeNet implements NetBackend {
     });
   }
 
-  listen(s: Sock) {
+  listen(s: Sock, backlog: number) {
     s.listening = true;
+    s.maxBacklog = E.backlogLimit(backlog);
+    this.accepted(s);
     return 0;
   }
 
+  // The accept queue lost its head: connections waiting beyond the backlog move in, and
+  // the clients of this VM among them finish connecting.
+  accepted(l: Sock) {
+    const q = (l.impl as TcpImpl | null)?.overflow;
+    while (q && q.length > 0 && l.backlog.length <= l.maxBacklog) {
+      const { c, pairing } = q.shift()!;
+      l.backlog.push(c);
+      l.notify();
+      if (pairing) this.admit(pairing);
+    }
+  }
+
+  private admit(p: Pairing) {
+    p.admitted = true;
+    this.finishPairing(p);
+  }
+
+  private finishPairing(p: Pairing) {
+    if (!p.admitted || !p.connected) return;
+    (p.listener.impl as TcpImpl | null)?.clients?.delete(p.port);
+    const im = p.client.impl as TcpImpl | null;
+    if (im) im.pairing = undefined;
+    p.finish!();
+  }
+
   // Wires a connected host socket to s.
   private attach(s: Sock, ns: net.Socket) {
     const im: TcpImpl = { ns, resume: () => ns.resume() };
@@ -285,22 +338,56 @@ export class NodeNet implements NetBackend {
     }, (ns) => {
       s.local = { addr: parseIp(ns.localAddress), port: ns.localPort ?? 0 };
       s.remote = { addr: ep.addr.slice(), port: ep.port };
-    });
+    }, undefined, this.listenerAt(ep));
   }
 
-  private connectTo(s: Sock, options: net.NetConnectOpts, connected: (ns: net.Socket) => void, failed?: () => void): number {
+  private listenerAt(ep: Endpoint): Sock | undefined {
+    const to = E.isWildcard(ep.addr) ? Uint8Array.of(127, 0, 0, 1) : ep.addr;
+    let found: Sock | undefined;
+    for (const o of this.live) {
+      if (!o.listening || o.unix || o.local?.port !== ep.port || !E.overlap(o.local.addr, to)) continue;
+      // listeners sharing the port (SO_REUSEPORT): the host picks one, so the connect cannot be held
+      if (found) return undefined;
+      found = o;
+    }
+    // a wildcard listener matches any address: the connect must go to this host to reach it
+    if (found && E.isWildcard(found.local!.addr) && to[0] !== 127 && !hostHas(to)) return undefined;
+    return found;
+  }
+
+  // listener is a listening socket of this VM that the connect goes to
+  private connectTo(
+    s: Sock, options: net.NetConnectOpts, connected: (ns: net.Socket) => void, failed?: () => void, listener?: Sock,
+  ): number {
     s.connecting = true;
     const start = () => {
       const ns = net.connect({ ...options, allowHalfOpen: true });
       this.attach(s, ns);
+      const im = s.impl as TcpImpl;
+      let pairing: Pairing | undefined;
+      // Registered on the next tick: net.connect issues the connect in a tick queued before this one,
+      // so the local port is known, and ticks run before any I/O callback, so before the listener's
+      // connection event. That event can still come before this socket's own connect event.
+      if (listener) process.nextTick(() => {
+        const clients = (listener.impl as TcpImpl | null)?.clients;
+        if (!clients || ns.localPort === undefined || s.impl !== im) return;
+        pairing = im.pairing = { client: s, listener, port: ns.localPort, connected: false, admitted: false };
+        clients.set(ns.localPort, pairing);
+      });
       let made = false;
       if (failed) ns.once("close", () => { if (!made) failed(); });
       ns.once("connect", () => {
         made = true;
-        s.connecting = false;
-        s.connected = true;
-        connected(ns);
-        s.notify();
+        const finish = () => {
+          s.connecting = false;
+          s.connected = true;
+          connected(ns);
+          s.notify();
+        };
+        if (!pairing) return finish();
+        pairing.finish = finish;
+        pairing.connected = true;
+        this.finishPairing(pairing);
       });
     };
     // a socket bound before connecting holds its port in a listening server
@@ -442,16 +529,35 @@ export class NodeNet implements NetBackend {
   close(s: Sock) {
     this.live.delete(s);
     // a listener resets the connections nobody accepted
-    for (const b of s.backlog) {
-      this.live.delete(b);
-      const ns = (b.impl as TcpImpl | null)?.ns;
-      b.impl = null;
-      if (ns) reset(ns);
-    }
+    for (const b of s.backlog) this.dropQueued(b);
     s.backlog = [];
     const im = s.impl as TcpImpl & UdpImpl | null;
     s.impl = null;
     if (!im) return;
+    // a connect that timed out or was abandoned leaves the listener's queues
+    const p = im.pairing;
+    if (p) {
+      const l = p.listener.impl as TcpImpl | null;
+      l?.clients?.delete(p.port);
+      const at = l?.overflow?.findIndex((e) => e.pairing === p) ?? -1;
+      if (at >= 0) this.dropQueued(l!.overflow!.splice(at, 1)[0].c);
+    }
+    // the listener resets what nobody accepted; like a kernel that dropped the SYN, a held connect then gets refused
+    for (const { c, pairing } of im.overflow ?? []) {
+      this.dropQueued(c);
+      const client = pairing?.client;
+      const ci = client?.impl as TcpImpl | undefined;
+      if (!client || !ci) continue;
+      ci.pairing = undefined;
+      ci.ns?.removeAllListeners();
+      ci.ns?.on("error", () => {});
+      ci.ns?.destroy();
+      client.impl = null;
+      client.connecting = false;
+      client.fail(E.ECONNREFUSED);
+    }
+    im.overflow = [];
+    im.clients?.clear();
     if (im.srv) {
       // libuv removes a closed server's socket file, a kernel leaves it
       const path = im.srv.address();
@@ -469,6 +575,13 @@ export class NodeNet implements NetBackend {
     else ns.destroy();
   }
 
+  private dropQueued(c: Sock) {
+    this.live.delete(c);
+    const ns = (c.impl as TcpImpl | null)?.ns;
+    c.impl = null;
+    if (ns) reset(ns);
+  }
+
   closeAll() {
     for (const s of [...this.live]) this.close(s);
   }

src/sockets.ts

@@ -303,7 +303,8 @@ export const isWildcard = (a: Uint8Array) => a[0] === 0 && a[1] === 0 && a[2] ==
 export const sameAddr = (a: Uint8Array, b: Uint8Array) => a[0] === b[0] && a[1] === b[1] && a[2] === b[2] && a[3] === b[3];
 const WILDCARD = new Uint8Array(4);
 const SOMAXCONN = 4096;
-const overlap = (a: Uint8Array, b: Uint8Array) => isWildcard(a) || isWildcard(b) || sameAddr(a, b);
+export const backlogLimit = (backlog: number) => (backlog < 0 ? SOMAXCONN : Math.min(backlog, SOMAXCONN));
+export const overlap = (a: Uint8Array, b: Uint8Array) => isWildcard(a) || isWildcard(b) || sameAddr(a, b);
 
 export function ipv4Literal(s: string | undefined): Uint8Array | null {
   const m = s ? /^(\d{1,3})\.(\d{1,3})\.(\d{1,3})\.(\d{1,3})$/.exec(s) : null;
@@ -390,7 +391,7 @@ export class LoopbackNet implements NetBackend {
 
   listen(s: Sock, backlog: number) {
     s.listening = true;
-    s.maxBacklog = backlog < 0 ? SOMAXCONN : Math.min(backlog, SOMAXCONN);
+    s.maxBacklog = backlogLimit(backlog);
     return 0;
   }
 

test/java/ListenBacklog.java

@@ -0,0 +1,113 @@
+import java.io.IOException;
+import java.net.*;
+import java.nio.channels.*;
+import java.util.*;
+import java.util.concurrent.*;
+
+public class ListenBacklog {
+    static final InetAddress LO = InetAddress.getLoopbackAddress();
+
+    public static void main(String[] args) throws Exception {
+        fullQueueTimesOut();
+        nonBlockingBurst();
+        acceptMakesRoom();
+        listenerCloseRefuses();
+    }
+
+    static String connect(int port, int timeoutMs, List<Socket> keep) {
+        Socket s = new Socket();
+        try {
+            s.connect(new InetSocketAddress(LO, port), timeoutMs);
+            keep.add(s);
+            return "ok";
+        } catch (SocketTimeoutException e) {
+            close(s);
+            return "timeout";
+        } catch (IOException e) {
+            close(s);
+            return e.getClass().getSimpleName();
+        }
+    }
+
+    static void close(Socket s) {
+        try { s.close(); } catch (IOException e) { }
+    }
+
+    // a listener that never accepts holds backlog + 1 connections, the rest time out
+    static void fullQueueTimesOut() throws Exception {
+        List<Socket> clients = new ArrayList<>();
+        try (ServerSocket ss = new ServerSocket(0, 2, LO)) {
+            StringBuilder sb = new StringBuilder();
+            for (int i = 0; i < 8; i++) sb.append(connect(ss.getLocalPort(), 300, clients)).append(' ');
+            System.out.println("backlog 2: " + sb.toString().trim());
+        }
+        clients.forEach(ListenBacklog::close);
+    }
+
+    // non-blocking connects made at once: backlog + 1 finish, the rest stay pending
+    static void nonBlockingBurst() throws Exception {
+        try (ServerSocket ss = new ServerSocket(0, 10, LO); Selector sel = Selector.open()) {
+            List<SocketChannel> channels = new ArrayList<>();
+            for (int i = 0; i < 40; i++) {
+                SocketChannel sc = SocketChannel.open();
+                sc.configureBlocking(false);
+                if (!sc.connect(new InetSocketAddress(LO, ss.getLocalPort()))) sc.register(sel, SelectionKey.OP_CONNECT);
+                channels.add(sc);
+            }
+            int finished = 0;
+            long end = System.currentTimeMillis() + 1500;
+            while (System.currentTimeMillis() < end) {
+                sel.select(100);
+                for (SelectionKey k : sel.selectedKeys()) {
+                    if (((SocketChannel) k.channel()).finishConnect()) {
+                        finished++;
+                        k.cancel();
+                    }
+                }
+                sel.selectedKeys().clear();
+            }
+            System.out.println("burst finished " + finished);
+            for (SocketChannel sc : channels) sc.close();
+        }
+    }
+
+    // a blocked connect finishes once accept removes a connection from the queue
+    static void acceptMakesRoom() throws Exception {
+        List<Socket> clients = new ArrayList<>();
+        try (ServerSocket ss = new ServerSocket(0, 1, LO)) {
+            int port = ss.getLocalPort();
+            int ok = 0;
+            for (int i = 0; i < 2; i++) if (connect(port, 1000, clients).equals("ok")) ok++;
+            System.out.println("queued " + ok);
+            System.out.println("full " + connect(port, 300, clients));
+            ExecutorService pool = Executors.newSingleThreadExecutor();
+            Future<String> waiting = pool.submit(() -> connect(port, 10000, clients));
+            Thread.sleep(300);
+            System.out.println("still waiting " + !waiting.isDone());
+            try (Socket a = ss.accept()) {
+                System.out.println("accepted " + a.getInetAddress().isLoopbackAddress());
+            }
+            System.out.println("waiting connect " + waiting.get(10, TimeUnit.SECONDS));
+            try (Socket a = ss.accept()) {
+                System.out.println("accepted again " + a.getInetAddress().isLoopbackAddress());
+            }
+            pool.shutdown();
+        }
+        clients.forEach(ListenBacklog::close);
+    }
+
+    // a listener that closes while a connect waits refuses it
+    static void listenerCloseRefuses() throws Exception {
+        List<Socket> clients = new ArrayList<>();
+        ServerSocket ss = new ServerSocket(0, 1, LO);
+        int port = ss.getLocalPort();
+        for (int i = 0; i < 2; i++) connect(port, 1000, clients);
+        ExecutorService pool = Executors.newSingleThreadExecutor();
+        Future<String> waiting = pool.submit(() -> connect(port, 10000, clients));
+        Thread.sleep(300);
+        ss.close();
+        System.out.println("after close " + waiting.get(10, TimeUnit.SECONDS));
+        pool.shutdown();
+        clients.forEach(ListenBacklog::close);
+    }
+}

test/jtreg/passing.txt

@@ -578,6 +578,7 @@ java/net/httpclient/HeadersTest.java
 java/net/httpclient/HttpClientAuthRetryLimitTest.java
 java/net/httpclient/HttpClientBuilderTest.java
 java/net/httpclient/HttpClientSNITest.java
+java/net/httpclient/HttpGetInCancelledFuture.java
 java/net/httpclient/HttpSlowServerTest.java
 java/net/httpclient/HttpsTunnelAuthTest.java
 java/net/httpclient/HttpsTunnelTest.java

test/jtreg/times.tsv

@@ -433,6 +433,7 @@
 16965	java/util/Date/DateGregorianCalendarTest.java
 16815	javax/crypto/spec/PBEKeySpec/PBEKeySpecRacing.java
 16776	java/net/DatagramSocket/DatagramSocketMulticasting.java
+16774	java/net/httpclient/HttpGetInCancelledFuture.java
 16760	java/lang/module/ModuleDescriptorHashCodeTest.java
 16687	java/io/Serializable/GetField/ReadFieldsCNF.java
 16658	java/text/Format/common/Bug6215962.java