Bound TCP resolution and connection waits with real deadlines

.github/workflows/ci.yml

@@ -155,6 +155,10 @@ jobs:
         run: python3 unittest/io_timeout_check.py
       - name: Check addressed UDP with bounded children
         run: python3 unittest/udp_datagram_check.py
+      - name: Check TCP resolver lifecycle and deadlines
+        run: |
+          python3 unittest/socket_glue_check.py
+          python3 unittest/tcp_connect_check.py
       - name: Compare every corpus program
         run: sh run_corpus.sh
       - name: Check YAML store round trips

.gitignore

@@ -3,6 +3,7 @@
 !vendor/prism/src/*.c
 !vendor/prism/src/util/*.c
 !tools/prism_glue.c
+!tools/socket_glue.c
 vendor/prism/obj/
 mere_extern.h
 *.o

KNOWN_GAPS.md

@@ -1070,11 +1070,15 @@ on a chosen address and answers getsockname / getpeername itself:
   whatever the kernel named -- for pipes and files too, since the layer is
   shared. The messages are the class's own text; ruby's sometimes add the
   call site (`@ io_fillbuf - fd:11`), which is not reproduced.
-- **TCP, either family.** `TCPSocket`/`Socket#connect` go through
-  `tcp_connect`. `TCPSocket.new` resolves the host itself and tries each
-  address in turn, as ruby does ("localhost" is ::1, then 127.0.0.1); a name
-  the files half cannot resolve is left to `tcp_connect`'s own resolver,
-  which tries only its first address. `TCPServer.new` is the runtime's
+- **TCP, either family.** `TCPSocket.new` resolves IPv4 and IPv6 on native
+  workers and polls nonblocking connections while yielding to Ruby threads.
+  It enforces open, resolution, and connection deadlines, binds a requested
+  local address/port before connecting, and supports the resolver delay/error
+  controls used by Ruby's socket tests. Fast fallback prefers IPv6 within a
+  50 ms resolution window and starts another attempt after 250 ms; disabling
+  it serializes connection attempts. Resolver workers own libc data only and
+  release it when a cancelled resolution eventually returns. `Socket#connect`
+  still uses the older blocking `tcp_connect` path. `TCPServer.new` uses
   tcp_listen_at, which is init_inetsock_internal's loop: every AI_PASSIVE
   answer in turn gets socket(2), SO_REUSEADDR and bind(2), the first that
   binds is the listener, and a refusal is the Errno the kernel named

corpus/503_tcp_connect_deadlines.rb

@@ -0,0 +1,66 @@
+require 'socket'
+
+# A listening socket can queue a connection without a Ruby accept thread.
+# Failures therefore leave no thread blocked in an ensure clause.
+def exchange(host, **options)
+  server = TCPServer.new(host, 0)
+  client = TCPSocket.new('localhost', server.addr[1], **options)
+  accepted = server.accept
+  accepted.write("a\0b")
+  p [client.remote_address.ip_address == host, client.recvfrom(3)]
+  accepted.close
+  accepted = nil
+  p client.recvfrom(1) # EOF is nil, not a two-element packet.
+ensure
+  accepted&.close
+  client&.close
+  server&.close
+end
+
+exchange('127.0.0.1', test_mode_settings: {delay: {ipv6: 1000}})
+exchange('127.0.0.1', test_mode_settings: {error: {ipv6: Socket::EAI_FAIL}})
+# A deadline for the unresolved family must preserve the IPv4 answer and
+# release the preference wait, even before that answer has been attempted.
+exchange('127.0.0.1', resolv_timeout: 0.01,
+         test_mode_settings: {delay: {ipv6: 1000}})
+
+# A failed connection does not hide the unresolved family's timeout when
+# there is no usable address or active attempt left.
+closed = TCPServer.new('127.0.0.1', 0)
+port = closed.addr[1]
+closed.close
+begin
+  TCPSocket.new('localhost', port, resolv_timeout: 0.01,
+               test_mode_settings: {delay: {ipv4: 1000}})
+rescue IO::TimeoutError
+  p :resolution_timeout_after_connection_failure
+end
+
+[:open_timeout, :resolv_timeout].each do |kind|
+  begin
+    TCPSocket.new('localhost', 1, **{kind => 0.02},
+                  test_mode_settings: {delay: {ipv4: 1000, ipv6: 1000}})
+  rescue IO::TimeoutError
+    p kind
+  end
+end
+
+begin
+  TCPSocket.new('localhost', 1, open_timeout: 1, connect_timeout: 1)
+rescue ArgumentError
+  p :conflicting_timeouts
+end
+
+server = TCPServer.new('127.0.0.1', 0)
+client = TCPSocket.new('127.0.0.1', server.addr[1], '127.0.0.1', 0,
+                       connect_timeout: 1, fast_fallback: false)
+accepted = server.accept
+p client.local_address.ip_address == '127.0.0.1'
+client.close
+accepted.close
+server.close
+
+# Each address comes from getifaddrs; IPv6 scope IDs must survive conversion.
+addresses = Socket.ip_address_list
+p [addresses.any?(&:ipv4_loopback?), addresses.all?(&:ip?),
+   addresses.all? { |a| a.ip_port == 0 && a.socktype == 0 }]

m_eval_leaf.mere

@@ -6432,6 +6432,10 @@ and stdlib_src = fn (name: str) ->
     "end\n" ++
     "\n" ++
     "class TCPSocket < IPSocket\n" ++
+    "  def recvfrom(maxlen, flags = 0)\n" ++
+    "    data = recv(maxlen, flags)\n" ++
+    "    data.nil? ? nil : [data, nil]\n" ++
+    "  end\n" ++
     "  def self.gethostbyname(host)\n" ++
     "    list = Addrinfo.__getaddrinfo(host, nil, 0, Socket::SOCK_STREAM, 0, Socket::AI_CANONNAME, false)\n" ++
     "    names = [list[0].canonname || host, []]\n" ++
@@ -6456,23 +6460,117 @@ and stdlib_src = fn (name: str) ->
     "    end\n" ++
     "    -1\n" ++
     "  end\n" ++
-    "  # A connect here is the runtime's blocking connect(2): it cannot be given\n" ++
-    "  # a deadline. A deadline of zero is the one that is known in advance -- no\n" ++
-    "  # connect completes in no time, and ruby's own nonblocking connect times out\n" ++
-    "  # on it even over loopback -- so it is answered without connecting. A\n" ++
-    "  # positive one is not enforced (KNOWN_GAPS).\n" ++
-    "  def initialize(host, port, local_host = nil, local_port = nil, connect_timeout: nil, open_timeout: nil, resolv_timeout: nil, fast_fallback: true)\n" ++
+    "  # Resolution workers never run Ruby. This loop owns their lifetime and yields\n" ++
+    "  # while DNS or connect(2) is pending, including during timeout cleanup.\n" ++
+    "  def self.__connect_with_deadlines(host, port, local_host, local_port, connect_timeout, open_timeout, resolv_timeout, fast_fallback, settings)\n" ++
+    "    start = Process.clock_gettime(Process::CLOCK_MONOTONIC)\n" ++
+    "    jobs = \{}\n" ++
+    "    ready = \{6 => [], 4 => []}\n" ++
+    "    active = []\n" ++
+    "    last_error = nil\n" ++
+    "    resolution_error = nil\n" ++
+    "    resolution_timeout_error = nil\n" ++
+    "    first_v4 = nil\n" ++
+    "    next_attempt = start\n" ++
+    "    last_family = 4\n" ++
+    "    local = local_host.nil? ? '' : Addrinfo.__getaddrinfo(local_host, nil, 0, 0, 0, 0, false)[0].ip_address\n" ++
+    "    families = host.include?(':') ? [6] : (host.match?(/\\A[0-9.]+\\z/) ? [4] : [6, 4])\n" ++
+    "    settings = \{} unless settings.is_a?(Hash)\n" ++
+    "    delays = settings[:delay] || \{}\n" ++
+    "    errors = settings[:error] || \{}\n" ++
+    "    families.each do |family|\n" ++
+    "      key = family == 6 ? :ipv6 : :ipv4\n" ++
+    "      job = Socket.__resolve_start(host, port, family, Integer(delays[key] || 0), Integer(errors[key] || 0))\n" ++
+    "      raise SystemCallError.new('getaddrinfo worker', -job) if job < 0\n" ++
+    "      jobs[family] = job\n" ++
+    "    end\n" ++
+    "    loop do\n" ++
+    "      now = Process.clock_gettime(Process::CLOCK_MONOTONIC)\n" ++
+    "      raise IO::TimeoutError, 'user specified open timeout' if open_timeout && now - start >= open_timeout\n" ++
+    "      jobs.keys.each do |family|\n" ++
+    "        job = jobs[family]\n" ++
+    "        next if Socket.__resolve_done(job) == 0\n" ++
+    "        error = Socket.__resolve_error(job)\n" ++
+    "        if error != 0\n" ++
+    "          resolution_error = Socket::ResolutionError.new('getaddrinfo failed')\n" ++
+    "        else\n" ++
+    "          index = 0\n" ++
+    "          loop do\n" ++
+    "            address = Socket.__resolve_address(job, index)\n" ++
+    "            break if address.empty?\n" ++
+    "            ready[family] << address unless ready[family].include?(address)\n" ++
+    "            index += 1\n" ++
+    "          end\n" ++
+    "          first_v4 ||= now if family == 4\n" ++
+    "        end\n" ++
+    "        Socket.__resolve_release(job)\n" ++
+    "        jobs.delete(family)\n" ++
+    "      end\n" ++
+    "      active.dup.each do |entry|\n" ++
+    "        fd, since = entry\n" ++
+    "        if (IO.__poll([fd], [2], 0)[0] & 6) != 0\n" ++
+    "          error = Socket.__connect_error(fd)\n" ++
+    "          active.delete(entry)\n" ++
+    "          return fd if error == 0\n" ++
+    "          IO.__fdclose(fd)\n" ++
+    "          last_error = SystemCallError.new(\"connect(2) for #\{host.inspect} port #\{port}\", error)\n" ++
+    "          next_attempt = now\n" ++
+    "        elsif connect_timeout && now - since >= connect_timeout\n" ++
+    "          active.delete(entry)\n" ++
+    "          IO.__fdclose(fd)\n" ++
+    "          last_error = IO::TimeoutError.new('user specified connect timeout')\n" ++
+    "          next_attempt = now\n" ++
+    "        end\n" ++
+    "      end\n" ++
+    "      if resolv_timeout && !jobs.empty? && now - start >= resolv_timeout\n" ++
+    "        jobs.each_value \{ |job| Socket.__resolve_release(job) }\n" ++
+    "        jobs.clear\n" ++
+    "        resolution_timeout_error = IO::TimeoutError.new('user specified resolution timeout')\n" ++
+    "        # Resolved addresses and active attempts can still connect. Removing\n" ++
+    "        # the pending jobs also ends the IPv6 preference wait.\n" ++
+    "      end\n" ++
+    "      if now >= next_attempt && (fast_fallback || active.empty?)\n" ++
+    "        family = last_family == 6 ? 4 : 6\n" ++
+    "        family = 10 - family if ready[family].empty?\n" ++
+    "        wait_v6 = family == 4 && jobs.key?(6) && first_v4 && now - first_v4 < 0.05\n" ++
+    "        unless ready[family].empty? || wait_v6\n" ++
+    "          address = ready[family].shift\n" ++
+    "          last_family = family\n" ++
+    "          fd = Socket.__connect_start(address, port, local, local_port || 0)\n" ++
+    "          if fd < 0\n" ++
+    "            context = local_host || local_port ? \"bind/connect(2) for #\{local_host.inspect} port #\{local_port}\" : \"connect(2) for #\{host.inspect} port #\{port}\"\n" ++
+    "            last_error = SystemCallError.new(context, -fd)\n" ++
+    "            next_attempt = now\n" ++
+    "          else\n" ++
+    "            active << [fd, now]\n" ++
+    "            next_attempt = now + 0.25\n" ++
+    "          end\n" ++
+    "          next\n" ++
+    "        end\n" ++
+    "      end\n" ++
+    "      if jobs.empty? && active.empty? && ready[4].empty? && ready[6].empty?\n" ++
+    "        raise(resolution_timeout_error || last_error || resolution_error || Socket::ResolutionError.new('getaddrinfo returned no addresses'))\n" ++
+    "      end\n" ++
+    "      sleep 0.001\n" ++
+    "    end\n" ++
+    "  ensure\n" ++
+    "    jobs&.each_value \{ |job| Socket.__resolve_release(job) }\n" ++
+    "    active&.each \{ |entry| IO.__fdclose(entry[0]) }\n" ++
+    "  end\n" ++
+    "  def initialize(host, port, local_host = nil, local_port = nil, connect_timeout: nil, open_timeout: nil, resolv_timeout: nil, fast_fallback: true, test_mode_settings: nil)\n" ++
     "    __new_block_warn if block_given?\n" ++
+    "    if open_timeout && (connect_timeout || resolv_timeout)\n" ++
+    "      raise ArgumentError, 'open_timeout cannot be combined with connect_timeout or resolv_timeout'\n" ++
+    "    end\n" ++
+    "    raise Socket::ResolutionError, 'getaddrinfo: no service specified' if port.nil?\n" ++
     "    @host = host\n" ++
     "    @port = port\n" ++
-    "    t = open_timeout || connect_timeout\n" ++
-    "    raise IO::TimeoutError, \"user specified timeout for #\{host}:#\{port}\" if t && t <= 0\n" ++
-    "    fd = TCPSocket.__connect_any(host, Addrinfo.__port_of(port, \"tcp\"))\n" ++
-    "    if fd < 0\n" ++
-    "      raise Errno::ECONNREFUSED, \"connect(2) for #\{host.inspect} port #\{port}\"\n" ++
+    "    [open_timeout, connect_timeout, resolv_timeout].each do |timeout|\n" ++
+    "      raise IO::TimeoutError, 'user specified timeout' if timeout && timeout <= 0\n" ++
     "    end\n" ++
+    "    host = host.nil? ? '' : host.to_str\n" ++
+    "    fd = TCPSocket.__connect_with_deadlines(host, Addrinfo.__port_of(port, 'tcp'), local_host, local_port && Addrinfo.__port_of(local_port, 'tcp'), connect_timeout, open_timeout, resolv_timeout, fast_fallback, test_mode_settings)\n" ++
     "    __setup(fd)\n" ++
-    "    Socket::Endpoint__.check_local(self, local_host, local_port) if local_host || local_port\n" ++
     "  end\n" ++
     "end\n" ++
     "\n" ++
@@ -6732,6 +6830,9 @@ and stdlib_src = fn (name: str) ->
     "end\n" ++
     "\n" ++
     "class Socket < BasicSocket\n" ++
+    "  def self.ip_address_list\n" ++
+    "    Socket.__ip_addresses.split.map \{ |address| Addrinfo.new([address].pack('H*'), nil, 0, 0) }\n" ++
+    "  end\n" ++
     "  module Constants\n" ++
     "    def self.__table\n" ++
     "      @__table ||= begin\n" ++

m_state.mere

@@ -17,6 +17,15 @@
 // writes cross through the shared byte arena (`mem_*`), so a payload with a
 // zero byte survives. The Wasm host has no sockets and stubs these to -1.
 extern fn tcp_connect: str -> int -> int;
+// tools/socket_glue.c: resolver workers and nonblocking numeric connects.
+extern fn mr_resolve_start: str -> int -> int -> int -> int -> int;
+extern fn mr_resolve_done: int -> int;
+extern fn mr_resolve_error: int -> int;
+extern fn mr_resolve_address: int -> int -> str;
+extern fn mr_resolve_release: int -> int;
+extern fn mr_connect_start: str -> int -> str -> int -> int;
+extern fn mr_connect_error: int -> int;
+extern fn mr_ip_addresses: unit -> str;
 extern fn tcp_listen: int -> int;
 extern fn tcp_accept: int -> int;
 extern fn tcp_close: int -> unit;

main.mere

@@ -20850,6 +20850,25 @@ and invoke_val_c0 = fn env -> fn world -> fn blk -> fn (recv0: Val) -> fn (name0
          (match args with
           | Cons (VStr h, Cons (VInt port, _)) -> VInt (tcp_connect (sget h) port)
           | _ -> VInt (0 - 1))
+       else if str_eq name "__ip_addresses" then mkstr (mr_ip_addresses ())
+       else if str_eq name "__resolve_start" then
+         (match args with
+          | Cons (VStr h, Cons (VInt p, Cons (VInt f, Cons (VInt d, Cons (VInt e, _))))) -> VInt (mr_resolve_start (sget h) p f d e)
+          | _ -> VInt (0 - 22))
+       else if str_eq name "__resolve_address" then
+         (match args with Cons (VInt j, Cons (VInt i, _)) -> mkstr (mr_resolve_address j i) | _ -> mkstr "")
+       else if str_eq name "__resolve_done" then
+         (match args with Cons (VInt j, _) -> VInt (mr_resolve_done j) | _ -> VInt 1)
+       else if str_eq name "__resolve_error" then
+         (match args with Cons (VInt j, _) -> VInt (mr_resolve_error j) | _ -> VInt 1)
+       else if str_eq name "__resolve_release" then
+         (match args with Cons (VInt j, _) -> VInt (mr_resolve_release j) | _ -> VInt 0)
+       else if str_eq name "__connect_start" then
+         (match args with
+          | Cons (VStr h, Cons (VInt p, Cons (VStr l, Cons (VInt lp, _)))) -> VInt (mr_connect_start (sget h) p (sget l) lp)
+          | _ -> VInt (0 - 22))
+       else if str_eq name "__connect_error" then
+         (match args with Cons (VInt fd, _) -> VInt (mr_connect_error fd) | _ -> VInt 22)
        else if (str_len name == 8 && str_eq name "__listen") then
          (match args with Cons (VInt port, _) -> VInt (tcp_listen port) | _ -> VInt (0 - 1))
        // compiler v0.1.555: a listener on the host asked for (TCPServer.new's

tools/build.sh

@@ -113,12 +113,12 @@ if [ "$pneed" = 1 ]; then
   ar rcs "$plib" "$pobj"/*.o || { rm -rf "$pobj"; exit 1; }
 fi
 
-printf 'cc     clang %s %s mr.c tools/prism_glue.c libprism.a -o mere-ruby\n' "$opt" "$plat"
+printf 'cc     clang %s %s mr.c tools/prism_glue.c tools/socket_glue.c libprism.a -o mere-ruby\n' "$opt" "$plat"
 # ⚠ NEVER `cp` OVER A RUNNING BINARY -- the text pages of a process that is
 # mid-run are backed by that file, and overwriting them SIGKILLs it or, worse,
 # gives it another build's instructions. Build beside it and rename.
 # shellcheck disable=SC2086
-clang $opt -w $plat -I "$root/vendor/prism/include" mr.c "$root/tools/prism_glue.c" "$plib" -o mere-ruby.new || { rm -f mere-ruby.new; exit 1; }
+clang $opt -w $plat -I "$root/vendor/prism/include" mr.c "$root/tools/prism_glue.c" "$root/tools/socket_glue.c" -pthread "$plib" -o mere-ruby.new || { rm -f mere-ruby.new; exit 1; }
 /bin/mv -f mere-ruby.new mere-ruby
 
 # ...and say which build this is, where something that must not be misled can