fixed sig for callbacks_for

Closes #353

doc/release_notes/1_5_1.md

@@ -0,0 +1,6 @@
+# 1.5.1
+
+## Bugfixes
+
+* connection errors on persistent connections which have just been checked out from the pool no longer account for retries bookkeeping; the assumption should be that, if a connection has been checked into the pool in an open state, chances are, when it eventually gets checked out, it may be corrupt. This issue was more exacerbated in `:persistent` plugin connections, which by design have a retry of 1, thus failing often immediately after check out without a legitimate request try.
+* native resolver: fix issue with process interrupts during DNS request, which caused a busy loop when closing the selector.

docker-compose-ruby-2.7.yml

@@ -9,7 +9,7 @@ services:
       - doh
 
   doh:
-    image: registry.gitlab.com/os85/httpx/nghttp2:1
+    image: registry.gitlab.com/os85/httpx/nghttp2:3
     depends_on:
       - doh-proxy
     entrypoint:

docker-compose-ruby-3.0.yml

@@ -9,7 +9,7 @@ services:
       - doh
 
   doh:
-    image: registry.gitlab.com/os85/httpx/nghttp2:1
+    image: registry.gitlab.com/os85/httpx/nghttp2:3
     depends_on:
       - doh-proxy
     entrypoint:

docker-compose-ruby-3.1.yml

@@ -9,7 +9,7 @@ services:
       - doh
 
   doh:
-    image: registry.gitlab.com/os85/httpx/nghttp2:1
+    image: registry.gitlab.com/os85/httpx/nghttp2:3
     depends_on:
       - doh-proxy
     entrypoint:

docker-compose-ruby-3.2.yml

@@ -9,7 +9,7 @@ services:
       - doh
 
   doh:
-    image: registry.gitlab.com/os85/httpx/nghttp2:1
+    image: registry.gitlab.com/os85/httpx/nghttp2:3
     depends_on:
       - doh-proxy
     entrypoint:

docker-compose-ruby-3.3.yml

@@ -9,7 +9,7 @@ services:
       - doh
 
   doh:
-    image: registry.gitlab.com/os85/httpx/nghttp2:1
+    image: registry.gitlab.com/os85/httpx/nghttp2:3
     depends_on:
       - doh-proxy
     entrypoint:

docker-compose-ruby-3.4.yml

@@ -9,7 +9,7 @@ services:
       - doh
 
   doh:
-    image: registry.gitlab.com/os85/httpx/nghttp2:1
+    image: registry.gitlab.com/os85/httpx/nghttp2:3
     depends_on:
       - doh-proxy
     entrypoint: /usr/local/bin/nghttpx

docker-compose.yml

@@ -69,7 +69,7 @@ services:
     command: -d 3
 
   http2proxy:
-    image: registry.gitlab.com/os85/httpx/nghttp2:1
+    image: registry.gitlab.com/os85/httpx/nghttp2:3
     ports:
       - 3300:80
     depends_on:
@@ -78,7 +78,7 @@ services:
     command: --no-ocsp --frontend '*,80;no-tls' --backend 'httpproxy,3128' --http2-proxy
 
   nghttp2:
-    image: registry.gitlab.com/os85/httpx/nghttp2:1
+    image: registry.gitlab.com/os85/httpx/nghttp2:3
     ports:
       - 80:80
       - 443:443
@@ -94,7 +94,7 @@ services:
           - another
 
   altsvc-nghttp2:
-    image: registry.gitlab.com/os85/httpx/nghttp2:1
+    image: registry.gitlab.com/os85/httpx/nghttp2:3
     ports:
       - 81:80
       - 444:443

integration_tests/sentry_test.rb

@@ -133,7 +133,7 @@ class SentryTest < Minitest::Test
 
     Sentry.init do |config|
       config.traces_sample_rate = 1.0
-      config.logger = mock_logger
+      config.sdk_logger = mock_logger
       config.dsn = DUMMY_DSN
       config.transport.transport_class = Sentry::DummyTransport
       config.background_worker_threads = 0

lib/httpx/adapters/datadog.rb

@@ -13,7 +13,11 @@ module Datadog::Tracing
 
       TYPE_OUTBOUND = Datadog::Tracing::Metadata::Ext::HTTP::TYPE_OUTBOUND
 
-      TAG_BASE_SERVICE = Datadog::Tracing::Contrib::Ext::Metadata::TAG_BASE_SERVICE
+      TAG_BASE_SERVICE = if Gem::Version.new(DATADOG_VERSION::STRING) < Gem::Version.new("1.15.0")
+        "_dd.base_service"
+      else
+        Datadog::Tracing::Contrib::Ext::Metadata::TAG_BASE_SERVICE
+      end
       TAG_PEER_HOSTNAME = Datadog::Tracing::Metadata::Ext::TAG_PEER_HOSTNAME
 
       TAG_KIND = Datadog::Tracing::Metadata::Ext::TAG_KIND
@@ -89,7 +93,7 @@ module Datadog::Tracing
 
               span.set_tags(
                 Datadog.configuration.tracing.header_tags.response_tags(response.headers.to_h)
-              )
+              ) if Datadog.configuration.tracing.respond_to?(:header_tags)
             end
 
             span.finish
@@ -139,7 +143,7 @@ module Datadog::Tracing
 
             span.set_tags(
               Datadog.configuration.tracing.header_tags.request_tags(request.headers.to_h)
-            )
+            ) if Datadog.configuration.tracing.respond_to?(:header_tags)
 
             span
           rescue StandardError => e

lib/httpx/callbacks.rb

@@ -20,7 +20,7 @@ module HTTPX
     end
 
     def callbacks_for?(type)
-      @callbacks.key?(type) && @callbacks[type].any?
+      @callbacks && @callbacks.key?(type) && @callbacks[type].any?
     end
 
     protected

lib/httpx/connection.rb

@@ -50,7 +50,11 @@ module HTTPX
     protected :sibling
 
     def initialize(uri, options)
-      @current_session = @current_selector = @sibling = @coalesced_connection = nil
+      @current_session = @current_selector =
+        @parser = @sibling = @coalesced_connection =
+                    @io = @ssl_session = @timeout =
+                            @connected_at = @response_received_at = nil
+
       @exhausted = @cloned = @main_sibling = false
 
       @options = Options.new(options)
@@ -61,6 +65,8 @@ module HTTPX
       @read_buffer = Buffer.new(@options.buffer_size)
       @write_buffer = Buffer.new(@options.buffer_size)
       @pending = []
+      @inflight = 0
+      @keep_alive_timeout = @options.timeout[:keep_alive_timeout]
 
       on(:error, &method(:on_error))
       if @options.io
@@ -98,9 +104,6 @@ module HTTPX
         build_altsvc_connection(alt_origin, origin, alt_params)
       end
 
-      @inflight = 0
-      @keep_alive_timeout = @options.timeout[:keep_alive_timeout]
-
       self.addresses = @options.addresses if @options.addresses
     end
 
@@ -369,8 +372,6 @@ module HTTPX
     end
 
     def handle_connect_error(error)
-      @connect_error = error
-
       return handle_error(error) unless @sibling && @sibling.connecting?
 
       @sibling.merge(self)
@@ -553,6 +554,10 @@ module HTTPX
       return unless @state == :inactive
 
       transition(:active)
+      # mark request as ping, as this inactive connection may have been
+      # closed by the server, and we don't want that to influence retry
+      # bookkeeping.
+      request.ping!
     end
 
     def build_parser(protocol = @io.protocol)
@@ -744,6 +749,7 @@ module HTTPX
         # activate
         @current_session.select_connection(self, @current_selector)
       end
+      log(level: 3) { "#{@state} -> #{nextstate}" }
       @state = nextstate
     end
 

lib/httpx/connection/http2.rb

@@ -58,6 +58,8 @@ module HTTPX
       if @connection.state == :closed
         return unless @handshake_completed
 
+        return if @buffer.empty?
+
         return :w
       end
 
@@ -229,7 +231,7 @@ module HTTPX
       end
 
       log(level: 1, color: :yellow) do
-        request.headers.merge(extra_headers).each.map { |k, v| "#{stream.id}: -> HEADER: #{k}: #{log_redact(v)}" }.join("\n")
+        "\n#{request.headers.merge(extra_headers).each.map { |k, v| "#{stream.id}: -> HEADER: #{k}: #{log_redact(v)}" }.join("\n")}"
       end
       stream.headers(request.headers.each(extra_headers), end_stream: request.body.empty?)
     end

lib/httpx/io/tcp.rb

@@ -75,9 +75,18 @@ module HTTPX
         @io = build_socket
       end
       try_connect
+    rescue Errno::EHOSTUNREACH,
+           Errno::ENETUNREACH => e
+      raise e if @ip_index <= 0
+
+      log { "failed connecting to #{@ip} (#{e.message}), evict from cache and trying next..." }
+      Resolver.cached_lookup_evict(@hostname, @ip)
+
+      @ip_index -= 1
+      @io = build_socket
+      retry
     rescue Errno::ECONNREFUSED,
            Errno::EADDRNOTAVAIL,
-           Errno::EHOSTUNREACH,
            SocketError,
            IOError => e
       raise e if @ip_index <= 0

lib/httpx/io/unix.rb

@@ -48,7 +48,7 @@ module HTTPX
       transition(:connected)
     rescue Errno::EINPROGRESS,
            Errno::EALREADY,
-           ::IO::WaitReadable
+           IO::WaitReadable
     end
 
     def expired?

lib/httpx/loggable.rb

@@ -34,7 +34,10 @@ module HTTPX
         klass = klass.superclass
       end
 
-      message = +"(pid:#{Process.pid} tid:#{Thread.current.object_id}, self:#{class_name}##{object_id}) "
+      message = +"(pid:#{Process.pid}, " \
+                 "tid:#{Thread.current.object_id}, " \
+                 "fid:#{Fiber.current.object_id}, " \
+                 "self:#{class_name}##{object_id}) "
       message << msg.call << "\n"
       message = "\e[#{COLORS[color]}m#{message}\e[0m" if color && debug_stream.respond_to?(:isatty) && debug_stream.isatty
       debug_stream << message

lib/httpx/options.rb

@@ -18,7 +18,7 @@ module HTTPX
     # https://github.com/ruby/resolv/blob/095f1c003f6073730500f02acbdbc55f83d70987/lib/resolv.rb#L408
     ip_address_families = begin
       list = Socket.ip_address_list
-      if list.any? { |a| a.ipv6? && !a.ipv6_loopback? && !a.ipv6_linklocal? && !a.ipv6_unique_local? }
+      if list.any? { |a| a.ipv6? && !a.ipv6_loopback? && !a.ipv6_linklocal? }
         [Socket::AF_INET6, Socket::AF_INET]
       else
         [Socket::AF_INET]
@@ -103,7 +103,7 @@ module HTTPX
     # :debug :: an object which log messages are written to (must respond to <tt><<</tt>)
     # :debug_level :: the log level of messages (can be 1, 2, or 3).
     # :debug_redact :: whether header/body payload should be redacted (defaults to <tt>false</tt>).
-    # :ssl :: a hash of options which can be set as params of OpenSSL::SSL::SSLContext (see HTTPX::IO::SSL)
+    # :ssl :: a hash of options which can be set as params of OpenSSL::SSL::SSLContext (see HTTPX::SSL)
     # :http2_settings :: a hash of options to be passed to a HTTP2::Connection (ex: <tt>{ max_concurrent_streams: 2 }</tt>)
     # :fallback_protocol :: version of HTTP protocol to use by default in the absence of protocol negotiation
     #                       like ALPN (defaults to <tt>"http/1.1"</tt>)
@@ -144,7 +144,16 @@ module HTTPX
     #
     # This list of options are enhanced with each loaded plugin, see the plugin docs for details.
     def initialize(options = {})
-      do_initialize(options)
+      defaults = DEFAULT_OPTIONS.merge(options)
+      defaults.each do |k, v|
+        next if v.nil?
+
+        option_method_name = :"option_#{k}"
+        raise Error, "unknown option: #{k}" unless respond_to?(option_method_name)
+
+        value = __send__(option_method_name, v)
+        instance_variable_set(:"@#{k}", value)
+      end
       freeze
     end
 
@@ -227,7 +236,7 @@ module HTTPX
     %i[
       request_class response_class headers_class request_body_class
       response_body_class connection_class options_class
-      pool_class pool_options
+      pool_class
       io fallback_protocol debug debug_redact resolver_class
       compress_request_body decompress_response_body
       persistent close_on_fork
@@ -350,19 +359,6 @@ module HTTPX
 
     private
 
-    def do_initialize(options = {})
-      defaults = DEFAULT_OPTIONS.merge(options)
-      defaults.each do |k, v|
-        next if v.nil?
-
-        option_method_name = :"option_#{k}"
-        raise Error, "unknown option: #{k}" unless respond_to?(option_method_name)
-
-        value = __send__(option_method_name, v)
-        instance_variable_set(:"@#{k}", value)
-      end
-    end
-
     def access_option(obj, k, ivar_map)
       case obj
       when Hash

lib/httpx/plugins/internal_telemetry.rb

@@ -50,15 +50,6 @@ module HTTPX
         end
       end
 
-      module NativeResolverMethods
-        def transition(nextstate)
-          state = @state
-          val = super
-          meter_elapsed_time("Resolver::Native: #{state} -> #{nextstate}")
-          val
-        end
-      end
-
       module InstanceMethods
         def self.included(klass)
           klass.prepend TrackTimeMethods
@@ -69,13 +60,6 @@ module HTTPX
           meter_elapsed_time("Session: initializing...")
           super
           meter_elapsed_time("Session: initialized!!!")
-          resolver_type = @options.resolver_class
-          resolver_type = Resolver.resolver_for(resolver_type)
-          return unless resolver_type <= Resolver::Native
-
-          resolver_type.prepend TrackTimeMethods
-          resolver_type.prepend NativeResolverMethods
-          @options = @options.merge(resolver_class: resolver_type)
         end
 
         def close(*)
@@ -104,33 +88,6 @@ module HTTPX
         end
       end
 
-      module RequestMethods
-        def self.included(klass)
-          klass.prepend Loggable
-          klass.prepend TrackTimeMethods
-          super
-        end
-
-        def transition(nextstate)
-          prev_state = @state
-          super
-          meter_elapsed_time("Request##{object_id}[#{@verb} #{@uri}: #{prev_state}] -> #{@state}") if prev_state != @state
-        end
-      end
-
-      module ConnectionMethods
-        def self.included(klass)
-          klass.prepend TrackTimeMethods
-          super
-        end
-
-        def handle_transition(nextstate)
-          state = @state
-          super
-          meter_elapsed_time("Connection##{object_id}[#{@origin}]: #{state} -> #{nextstate}") if nextstate == @state
-        end
-      end
-
       module PoolMethods
         def self.included(klass)
           klass.prepend Loggable
@@ -138,12 +95,6 @@ module HTTPX
           super
         end
 
-        def checkout_connection(request_uri, options)
-          super.tap do |connection|
-            meter_elapsed_time("Pool##{object_id}: checked out connection for Connection##{connection.object_id}[#{connection.origin}]}")
-          end
-        end
-
         def checkin_connection(connection)
           super.tap do
             meter_elapsed_time("Pool##{object_id}: checked in connection for Connection##{connection.object_id}[#{connection.origin}]}")

lib/httpx/plugins/proxy.rb

@@ -1,7 +1,7 @@
 # frozen_string_literal: true
 
 module HTTPX
-  class HTTPProxyError < ConnectionError; end
+  class ProxyError < ConnectionError; end
 
   module Plugins
     #
@@ -15,7 +15,8 @@ module HTTPX
     # https://gitlab.com/os85/httpx/wikis/Proxy
     #
     module Proxy
-      Error = HTTPProxyError
+      class ProxyConnectionError < ProxyError; end
+
       PROXY_ERRORS = [TimeoutError, IOError, SystemCallError, Error].freeze
 
       class << self
@@ -28,6 +29,12 @@ module HTTPX
         def extra_options(options)
           options.merge(supported_proxy_protocols: [])
         end
+
+        def subplugins
+          {
+            retries: ProxyRetries,
+          }
+        end
       end
 
       class Parameters
@@ -160,9 +167,9 @@ module HTTPX
 
           next_proxy = proxy.uri
 
-          raise Error, "Failed to connect to proxy" unless next_proxy
+          raise ProxyError, "Failed to connect to proxy" unless next_proxy
 
-          raise Error,
+          raise ProxyError,
                 "#{next_proxy.scheme}: unsupported proxy protocol" unless options.supported_proxy_protocols.include?(next_proxy.scheme)
 
           if (no_proxy = proxy.no_proxy)
@@ -179,20 +186,28 @@ module HTTPX
         private
 
         def fetch_response(request, selector, options)
-          response = super
+          response = request.response # in case it goes wrong later
 
-          if response.is_a?(ErrorResponse) && proxy_error?(request, response, options)
-            options.proxy.shift
+          begin
+            response = super
 
-            # return last error response if no more proxies to try
-            return response if options.proxy.uri.nil?
+            if response.is_a?(ErrorResponse) && proxy_error?(request, response, options)
+              options.proxy.shift
 
-            log { "failed connecting to proxy, trying next..." }
-            request.transition(:idle)
-            send_request(request, selector, options)
-            return
+              # return last error response if no more proxies to try
+              return response if options.proxy.uri.nil?
+
+              log { "failed connecting to proxy, trying next..." }
+              request.transition(:idle)
+              send_request(request, selector, options)
+              return
+            end
+            response
+          rescue ProxyError
+            # may happen if coupled with retries, and there are no more proxies to try, in which case
+            # it'll end up here
+            response
           end
-          response
         end
 
         def proxy_error?(_request, response, options)
@@ -211,7 +226,7 @@ module HTTPX
             proxy_uri = URI(options.proxy.uri)
 
             error.message.end_with?(proxy_uri.to_s)
-          when *PROXY_ERRORS
+          when ProxyConnectionError
             # timeout errors connecting to proxy
             true
           else
@@ -251,6 +266,14 @@ module HTTPX
           when :connecting
             consume
           end
+        rescue *PROXY_ERRORS => e
+          if connecting?
+            error = ProxyConnectionError.new(e.message)
+            error.set_backtrace(e.backtrace)
+            raise error
+          end
+
+          raise e
         end
 
         def reset
@@ -292,13 +315,29 @@ module HTTPX
           end
           super
         end
+
+        def purge_after_closed
+          super
+          @io = @io.proxy_io if @io.respond_to?(:proxy_io)
+        end
+      end
+
+      module ProxyRetries
+        module InstanceMethods
+          def retryable_error?(ex)
+            super || ex.is_a?(ProxyConnectionError)
+          end
+        end
       end
     end
     register_plugin :proxy, Proxy
   end
 
   class ProxySSL < SSL
+    attr_reader :proxy_io
+
     def initialize(tcp, request_uri, options)
+      @proxy_io = tcp
       @io = tcp.to_io
       super(request_uri, tcp.addresses, options)
       @hostname = request_uri.host

lib/httpx/plugins/proxy/socks4.rb

@@ -4,7 +4,7 @@ require "resolv"
 require "ipaddr"
 
 module HTTPX
-  class Socks4Error < HTTPProxyError; end
+  class Socks4Error < ProxyError; end
 
   module Plugins
     module Proxy

lib/httpx/plugins/proxy/socks5.rb

@@ -1,7 +1,7 @@
 # frozen_string_literal: true
 
 module HTTPX
-  class Socks5Error < HTTPProxyError; end
+  class Socks5Error < ProxyError; end
 
   module Plugins
     module Proxy

lib/httpx/plugins/response_cache/file_store.rb

@@ -81,7 +81,7 @@ module HTTPX::Plugins
 
           response.body.rewind
 
-          ::IO.copy_stream(response.body, f)
+          IO.copy_stream(response.body, f)
         end
       end
 
@@ -131,7 +131,7 @@ module HTTPX::Plugins
         response.original_request = original_request
         response.finish!
 
-        ::IO.copy_stream(f, response.body)
+        IO.copy_stream(f, response.body)
 
         response
       end

lib/httpx/plugins/stream_bidi.rb

@@ -119,7 +119,7 @@ module HTTPX
       class Signal
         def initialize
           @closed = false
-          @pipe_read, @pipe_write = ::IO.pipe
+          @pipe_read, @pipe_write = IO.pipe
         end
 
         def state
@@ -127,7 +127,7 @@ module HTTPX
         end
 
         # noop
-        def log(**); end
+        def log(**, &_); end
 
         def to_io
           @pipe_read.to_io

lib/httpx/plugins/upgrade.rb

@@ -65,6 +65,12 @@ module HTTPX
       module ConnectionMethods
         attr_reader :upgrade_protocol, :hijacked
 
+        def initialize(*)
+          super
+
+          @upgrade_protocol = nil
+        end
+
         def hijack_io
           @hijacked = true
 

lib/httpx/request.rb

@@ -8,6 +8,7 @@ module HTTPX
   # as well as maintaining the state machine which manages streaming the request onto the wire.
   class Request
     extend Forwardable
+    include Loggable
     include Callbacks
     using URIExtensions
 
@@ -293,6 +294,7 @@ module HTTPX
         return if @state == :expect
 
       end
+      log(level: 3) { "#{@state}] -> #{nextstate}" }
       @state = nextstate
       emit(@state, self)
       nil

lib/httpx/request/body.rb

@@ -56,7 +56,7 @@ module HTTPX
           block.call(chunk)
         end
         # TODO: use copy_stream once bug is resolved: https://bugs.ruby-lang.org/issues/21131
-        # ::IO.copy_stream(body, ProcIO.new(block))
+        # IO.copy_stream(body, ProcIO.new(block))
       elsif body.respond_to?(:each)
         body.each(&block)
       else

lib/httpx/resolver.rb

@@ -83,6 +83,18 @@ module HTTPX
       end
     end
 
+    def cached_lookup_evict(hostname, ip)
+      ip = ip.to_s
+
+      lookup_synchronize do |lookups|
+        entries = lookups[hostname]
+
+        return unless entries
+
+        lookups.delete_if { |entry| entry["data"] == ip }
+      end
+    end
+
     # do not use directly!
     def lookup(hostname, lookups, ttl)
       return unless lookups.key?(hostname)

lib/httpx/resolver/multi.rb

@@ -35,6 +35,10 @@ module HTTPX
       @resolvers.each { |r| r.__send__(__method__, s) }
     end
 
+    def log(*args, **kwargs, &blk)
+      @resolvers.each { |r| r.__send__(__method__, *args, **kwargs, &blk) }
+    end
+
     def closed?
       @resolvers.all?(&:closed?)
     end

lib/httpx/resolver/native.rb

@@ -480,6 +480,7 @@ module HTTPX
         @write_buffer.clear
         @read_buffer.clear
       end
+      log(level: 3) { "#{@state} -> #{nextstate}" }
       @state = nextstate
     rescue Errno::ECONNREFUSED,
            Errno::EADDRNOTAVAIL,

lib/httpx/resolver/system.rb

@@ -127,7 +127,7 @@ module HTTPX
       when :open
         return unless @state == :idle
 
-        @pipe_read, @pipe_write = ::IO.pipe
+        @pipe_read, @pipe_write = IO.pipe
       when :closed
         return unless @state == :open
 

lib/httpx/response/body.rb

@@ -136,7 +136,7 @@ module HTTPX
       if dest.respond_to?(:path) && @buffer.respond_to?(:path)
         FileUtils.mv(@buffer.path, dest.path)
       else
-        ::IO.copy_stream(@buffer, dest)
+        IO.copy_stream(@buffer, dest)
       end
     end
 

lib/httpx/response/buffer.rb

@@ -110,7 +110,7 @@ module HTTPX
 
       if aux
         aux.rewind
-        ::IO.copy_stream(aux, @buffer)
+        IO.copy_stream(aux, @buffer)
         aux.close
       end
 

lib/httpx/selector.rb

@@ -63,7 +63,10 @@ module HTTPX
       # array may change during iteration
       selectables = @selectables.reject(&:inflight?)
 
-      selectables.each(&:terminate)
+      selectables.delete_if do |sel|
+        sel.terminate
+        sel.state == :closed
+      end
 
       until selectables.empty?
         next_tick

lib/httpx/session.rb

@@ -136,6 +136,9 @@ module HTTPX
     alias_method :select_resolver, :select_connection
 
     def deselect_connection(connection, selector, cloned = false)
+      connection.log(level: 2) do
+        "deregistering connection##{connection.object_id}(#{connection.state}) from selector##{selector.object_id}"
+      end
       selector.deregister(connection)
 
       # when connections coalesce
@@ -145,14 +148,19 @@ module HTTPX
 
       return if @closing && connection.state == :closed
 
+      connection.log(level: 2) { "check-in connection##{connection.object_id}(#{connection.state}) in pool##{@pool.object_id}" }
       @pool.checkin_connection(connection)
     end
 
     def deselect_resolver(resolver, selector)
+      resolver.log(level: 2) do
+        "deregistering resolver##{resolver.object_id}(#{resolver.state}) from selector##{selector.object_id}"
+      end
       selector.deregister(resolver)
 
       return if @closing && resolver.closed?
 
+      resolver.log(level: 2) { "check-in resolver##{resolver.object_id}(#{resolver.state}) in pool##{@pool.object_id}" }
       @pool.checkin_resolver(resolver)
     end
 
@@ -221,7 +229,11 @@ module HTTPX
     def fetch_response(request, _selector, _options)
       response = request.response
 
-      response if response && response.finished?
+      return unless response && response.finished?
+
+      log(level: 2) { "response fetched" }
+
+      response
     end
 
     # sends the +request+ to the corresponding HTTPX::Connection
@@ -303,8 +315,7 @@ module HTTPX
 
     # returns the array of HTTPX::Response objects corresponding to the array of HTTPX::Request +requests+.
     def receive_requests(requests, selector)
-      # @type var responses: Array[response]
-      responses = []
+      responses = [] # : Array[response]
 
       # guarantee ordered responses
       loop do
@@ -326,12 +337,30 @@ module HTTPX
         # handshake error, and the error responses have already been emitted, but there was no
         # opportunity to traverse the requests, hence we're returning only a fraction of the errors
         # we were supposed to. This effectively fetches the existing responses and return them.
-        while (request = requests.shift)
-          response = fetch_response(request, selector, request.options)
-          request.emit(:complete, response) if response
-          responses << response
+        exit_from_loop = true
+
+        requests_to_remove = [] # : Array[Request]
+
+        requests.each do |req|
+          response = fetch_response(req, selector, request.options)
+
+          if exit_from_loop && response
+            req.emit(:complete, response)
+            responses << response
+            requests_to_remove << req
+          else
+            # fetch_response may resend requests. when that happens, we need to go back to the initial
+            # loop and process the selector. we still do a pass-through on the remainder of requests, so
+            # that every request that need to be resent, is resent.
+            exit_from_loop = false
+
+            raise Error, "something went wrong, responses not found and requests not resent" if selector.empty?
+          end
         end
-        break
+
+        break if exit_from_loop
+
+        requests -= requests_to_remove
       end
       responses
     end
@@ -380,14 +409,16 @@ module HTTPX
     end
 
     def find_resolver_for(connection, selector)
-      resolver = selector.find_resolver(connection.options)
-
-      unless resolver
-        resolver = @pool.checkout_resolver(connection.options)
-        resolver.current_session = self
-        resolver.current_selector = selector
+      if (resolver = selector.find_resolver(connection.options))
+        resolver.log(level: 2) { "found resolver##{connection.object_id}(#{connection.state}) in selector##{selector.object_id}" }
+        return resolver
       end
 
+      resolver = @pool.checkout_resolver(connection.options)
+      resolver.log(level: 2) { "found resolver##{connection.object_id}(#{connection.state}) in pool##{@pool.object_id}" }
+      resolver.current_session = self
+      resolver.current_selector = selector
+
       resolver
     end
 
@@ -397,7 +428,10 @@ module HTTPX
       unless conn1.coalescable?(conn2)
         conn2.log(level: 2) { "not coalescing with conn##{conn1.object_id}[#{conn1.origin}])" }
         select_connection(conn2, selector)
-        @pool.checkin_connection(conn1) if from_pool
+        if from_pool
+          conn1.log(level: 2) { "check-in connection##{conn1.object_id}(#{conn1.state}) in pool##{@pool.object_id}" }
+          @pool.checkin_connection(conn1)
+        end
         return false
       end
 

lib/httpx/transcoder/multipart/mime_type_detector.rb

@@ -44,7 +44,7 @@ module HTTPX
 
           Open3.popen3(*%w[file --mime-type --brief -]) do |stdin, stdout, stderr, thread|
             begin
-              ::IO.copy_stream(file, stdin.binmode)
+              IO.copy_stream(file, stdin.binmode)
             rescue Errno::EPIPE
             end
             file.rewind

lib/httpx/transcoder/utils/deflater.rb

@@ -63,7 +63,7 @@ module HTTPX
         buffer = Response::Buffer.new(
           threshold_size: Options::MAX_BODY_THRESHOLD_SIZE
         )
-        ::IO.copy_stream(self, buffer)
+        IO.copy_stream(self, buffer)
 
         buffer.rewind if buffer.respond_to?(:rewind)
 

lib/httpx/version.rb

@@ -1,5 +1,5 @@
 # frozen_string_literal: true
 
 module HTTPX
-  VERSION = "1.5.0"
+  VERSION = "1.5.1"
 end

regression_tests/bug_0_22_2_test.rb

@@ -36,4 +36,5 @@ class Bug_0_22_2_Test < Minitest::Test
     assert connection_ipv4.family == Socket::AF_INET
     assert connection_ipv6.family == Socket::AF_INET6
   end
-end if HTTPX::Session.default_options.ip_families.size > 1
+  # TODO: remove this once gitlab docker allows TCP connectivity alongside DNS
+end unless ENV.key?("CI")

regression_tests/bug_1_1_0_test.rb

@@ -11,7 +11,7 @@ class Bug_1_1_0_Test < Minitest::Test
   include HTTPHelpers
 
   def test_read_timeout_firing_too_soon_before_select
-    timeout = { read_timeout: 1 }
+    timeout = { read_timeout: 2 }
 
     uri = build_uri("/get")
 

regression_tests/bug_1_1_1_test.rb

@@ -6,7 +6,7 @@ require "support/http_helpers"
 class Bug_1_1_1_Test < Minitest::Test
   include HTTPHelpers
 
-  def test_conection_callbacks_fire_setup_once
+  def test_connection_callbacks_fire_setup_once
     uri = build_uri("/get")
 
     connected = 0

regression_tests/bug_1_5_0_test.rb

@@ -0,0 +1,80 @@
+# frozen_string_literal: true
+
+require "test_helper"
+require "support/http_helpers"
+require "webmock/minitest"
+require "httpx/adapters/webmock"
+
+class Bug_1_5_0_Test < Minitest::Test
+  include HTTPHelpers
+
+  def test_persistent_do_not_exhaust_retry_on_eof_error
+    start_test_servlet(KeepAlivePongThenGoawayServer) do |server|
+      persistent_session = HTTPX.plugin(SessionWithPool)
+                                .plugin(:persistent)
+                                .with(ssl: { verify_mode: OpenSSL::SSL::VERIFY_NONE })
+      uri = "#{server.origin}/"
+      # artificially create two connections
+      responses = 2.times.map do
+        Thread.new do
+          Thread.current.abort_on_exception = true
+          Thread.current.report_on_exception = true
+
+          persistent_session.get(uri)
+        end
+      end.map(&:value)
+
+      responses.each do |response|
+        verify_status(response, 200)
+      end
+
+      conns1 = persistent_session.connections
+      assert conns1.size == 2, "should have started two different connections to the same origin"
+      assert conns1.none? { |c| c.state == :closed }, "all connections should have been open"
+
+      sleep(2)
+      response = persistent_session.get(uri)
+      verify_status(response, 200) # should not raise GoAwayError
+      conns2 = persistent_session.connections
+      assert conns2.size == 2
+      assert conns2.count { |c| c.state == :closed } == 1, "one of them should have been closed"
+    ensure
+      persistent_session.close
+    end
+  end
+end
+
+class OnPingDisconnectServer < TestHTTP2Server
+  module GoAwayOnFirstPing
+    attr_accessor :num_requests
+
+    def activate_stream(*, **)
+      super.tap do
+        @num_requests += 1
+      end
+    end
+
+    def ping_management(*)
+      if @num_requests == 1
+        @num_requests = 0
+        goaway
+      else
+        super
+      end
+    end
+  end
+
+  def initialize(*)
+    super
+    @num_requests = Hash.new(0)
+  end
+
+  private
+
+  def handle_connection(conn, _)
+    super
+
+    conn.extend(GoAwayOnFirstPing)
+    conn.num_requests = 0
+  end
+end

sig/callbacks.rbs

@@ -8,7 +8,7 @@ module HTTPX
     def once: (Symbol) { (*untyped) -> void } -> ^(*untyped) -> void
     def emit: (Symbol, *untyped) -> void
 
-    def callbacks_for?: (Symbol) -> bool
+    def callbacks_for?: (Symbol) -> boolish
     def callbacks: () -> Hash[Symbol, Array[_Callable]]
                   | (Symbol) -> Array[_Callable]
   end

sig/connection.rbs

@@ -82,7 +82,7 @@ module HTTPX
 
     def interests: () -> io_interests?
 
-    def to_io: () -> ::IO
+    def to_io: () -> IO
 
     def call: () -> void
 

sig/io.rbs

@@ -1,6 +1,3 @@
 module HTTPX
   type io_type = "udp" | "tcp" | "ssl" | "unix"
-
-  module IO
-  end
 end

sig/io/tcp.rbs

@@ -14,18 +14,24 @@ module HTTPX
 
     alias host ip
 
+    @io: Socket
+
     @hostname: String
 
     @options: Options
 
     @fallback_protocol: String
 
+    @keep_open: bool
+
+    @ip_index: Integer
+
     # TODO: lift when https://github.com/ruby/rbs/issues/1497 fixed
     def initialize: (URI::Generic origin, Array[ipaddr]? addresses, Options options) ?{ (instance) -> void } -> void
 
     def add_addresses: (Array[ipaddr] addrs) -> void
 
-    def to_io: () -> ::IO
+    def to_io: () -> IO
 
     def protocol: () -> String
 

sig/io/udp.rbs

@@ -4,7 +4,7 @@ module HTTPX
 
     def initialize: (String ip, Integer port, Options options) -> void
 
-    def to_io: () -> ::IO
+    def to_io: () -> IO
 
     def connect: () -> void
 

sig/options.rbs

@@ -140,8 +140,6 @@ module HTTPX
 
     def initialize: (?options options) -> void
 
-    def do_initialize: (?options options) -> void
-
     def access_option: (Hash[Symbol, untyped] | Object | nil obj, Symbol k, Hash[Symbol, Symbol]? ivar_map) -> untyped
   end
 

sig/plugins/proxy.rbs

@@ -1,8 +1,9 @@
 module HTTPX
-  class HTTPProxyError < ConnectionError
+  class ProxyError < ConnectionError
   end
 
   class ProxySSL < SSL
+    attr_reader proxy_io: TCP | SSL
   end
 
   module Plugins
@@ -11,7 +12,9 @@ module HTTPX
     end
 
     module Proxy
-      Error: singleton(HTTPProxyError)
+      class ProxyConnectionError < ProxyError
+      end
+
       PROXY_ERRORS: Array[singleton(StandardError)]
 
       class Parameters

sig/plugins/proxy/socks4.rbs

@@ -1,5 +1,5 @@
 module HTTPX
-  class Socks4Error < HTTPProxyError
+  class Socks4Error < ProxyError
   end
 
   module Plugins

sig/plugins/proxy/socks5.rbs

@@ -1,5 +1,5 @@
 module HTTPX
-  class Socks5Error < HTTPProxyError
+  class Socks5Error < ProxyError
   end
 
   module Plugins

sig/plugins/stream_bidi.rbs

@@ -25,8 +25,8 @@ module HTTPX
 
       class Signal
         @closed: bool
-        @pipe_read: ::IO
-        @pipe_write: ::IO
+        @pipe_read: IO
+        @pipe_write: IO
 
         include _Selectable
 

sig/request.rbs

@@ -1,6 +1,7 @@
 module HTTPX
   class Request
     extend Forwardable
+    include Loggable
     include Callbacks
 
     METHODS: Array[Symbol]

sig/resolver.rbs

@@ -32,6 +32,8 @@ module HTTPX
 
     def self?.cached_lookup_set: (String hostname, ip_family family, Array[dns_result] addresses) -> void
 
+    def self?.cached_lookup_evict: (String hostname, ipaddr ip) -> void
+
     def self?.lookup: (String hostname, Hash[String, Array[dns_result]] lookups, Numeric ttl) -> Array[IPAddr]?
 
     def self?.generate_id: () -> Integer

sig/resolver/system.rbs

@@ -11,8 +11,8 @@ module HTTPX
       @queries: Array[[ip_family, Connection]]
       @ips: Array[[ip_family, Connection, (Array[Addrinfo] | StandardError)]]
       @pipe_mutex: Thread::Mutex
-      @pipe_read: ::IO
-      @pipe_write: ::IO
+      @pipe_read: IO
+      @pipe_write: IO
 
       attr_reader state: Symbol
 

sig/selector.rbs

@@ -2,7 +2,7 @@ module HTTPX
   interface _Selectable
     def state: () -> Symbol
 
-    def to_io: () -> ::IO
+    def to_io: () -> IO
 
     def call: () -> void
 

sig/transcoder/multipart.rbs

@@ -90,7 +90,7 @@ module HTTPX
       module MimeTypeDetector
         DEFAULT_MIMETYPE: String
 
-        def self?.call: (::IO | Tempfile file, String filename) -> String?
+        def self?.call: (IO | Tempfile file, String filename) -> String?
       end
     end
   end

test/https_test.rb

@@ -115,11 +115,11 @@ class HTTPSTest < Minitest::Test
   # HTTP/2-specific tests
 
   {
-    http1: { uri: "https://httpbin.org/get", ssl: { alpn_protocols: %w[http/1.1] } },
+    http1: { uri: "https://aws:4566", ssl: { verify_mode: OpenSSL::SSL::VERIFY_NONE, alpn_protocols: %w[http/1.1] } },
     http2: {},
   }.each do |proto, proto_options|
     define_method :"test_multiple_get_max_requests_#{proto}" do
-      uri = proto_options.delete(:uri) || URI(build_uri("/get"))
+      uri = proto_options.delete(:uri) || URI(build_uri("/"))
       options = { max_requests: 2, **proto_options }
 
       HTTPX.plugin(SessionWithPool).with(options).wrap do |http|
@@ -132,7 +132,9 @@ class HTTPSTest < Minitest::Test
         verify_body_length(response3)
         connection_count = http.connection_count
         assert connection_count == 2, "expected to have 2 connections, instead have #{connection_count}"
-        assert http.connections.size == 1, "expected connection to have been reused on exhaustion"
+        http.connections.tally(&:family).each_value do |count|
+          assert count == 1, "expected connection to have been reused on exhaustion"
+        end
 
         # ssl session ought to be reused
         conn = http.connections.first

test/proxy_test.rb

@@ -31,7 +31,7 @@ class ProxyTest < Minitest::Test
 
   def test_proxy_unsupported_scheme
     res = HTTPX.plugin(:proxy).with_proxy(uri: "https://proxy:123").get("http://smth.com")
-    verify_error_response(res, HTTPX::HTTPProxyError)
+    verify_error_response(res, HTTPX::ProxyError)
     verify_error_response(res, "https: unsupported proxy protocol")
   end
 

test/support/ci/Dockerfile.nghttp2

@@ -1,8 +1,9 @@
-FROM alpine:3.10.2
+FROM alpine:3.22.1
+
 RUN \
  NGHTTP2_VERSION='1.46.0' \
  BUILD_DEPS='build-base wget' \
- RUN_DEPS='ca-certificates libstdc++ openssl-dev libev-dev zlib-dev jansson-dev libxml2-dev c-ares-dev' \
+ RUN_DEPS='ca-certificates libstdc++ openssl-dev libev-dev zlib-dev jansson-dev libxml2-dev c-ares-dev brotli-dev' \
  && apk --no-cache add $BUILD_DEPS $RUN_DEPS \
  && cd /tmp \
  && wget -qO- "https://github.com/tatsuhiro-t/nghttp2/releases/download/v${NGHTTP2_VERSION}/nghttp2-${NGHTTP2_VERSION}.tar.gz" | tar -xz \

test/support/requests/callbacks.rb

@@ -136,5 +136,10 @@ module Requests
         end
       end
     end
+
+    def test_callbacks_can_compose_with
+      http = HTTPX.plugin(:callbacks).with(persistent: true)
+      assert http.instance_variable_get(:@persistent)
+    end
   end
 end

test/support/requests/plugins/brotli.rb

@@ -5,7 +5,7 @@ module Requests
     module Brotli
       def test_brotli
         session = HTTPX.plugin(:brotli)
-        response = session.get("http://httpbin.org/brotli")
+        response = session.get("http://nghttp2.org/httpbin/brotli")
         verify_status(response, 200)
         body = json_body(response)
         assert body["brotli"], "response should be deflated"

test/support/requests/plugins/persistent.rb

@@ -99,6 +99,30 @@ module Requests
           http.close
         end
       end unless RUBY_ENGINE == "jruby"
+
+      def test_persistent_proxy_retry_http2_goaway
+        return unless origin.start_with?("https")
+
+        start_test_servlet(KeepAlivePongThenGoawayServer) do |server|
+          start_test_servlet(ProxyServer) do |proxy|
+            http = HTTPX.plugin(SessionWithPool)
+                        .plugin(RequestInspector)
+                        .plugin(:persistent) # implicit max_retries == 1
+                        .plugin(:proxy)
+                        .with(
+                          proxy: { uri: proxy.origin },
+                          ssl: { verify_mode: OpenSSL::SSL::VERIFY_NONE }
+                        )
+            uri = "#{server.origin}/"
+            response = http.get(uri)
+            verify_status(response, 200)
+            response = http.get(uri)
+            verify_status(response, 200)
+            assert http.calls == 2, "expect request to be built 2 times (was #{http.calls})"
+            http.close
+          end
+        end
+      end unless RUBY_ENGINE == "jruby"
     end
   end
 end

test/support/requests/plugins/proxy.rb

@@ -14,7 +14,7 @@ module Requests
         http = HTTPX.plugin(:proxy)
         uri = build_uri("/get")
         res = http.with_proxy(uri: []).get(uri)
-        verify_error_response(res, HTTPX::HTTPProxyError)
+        verify_error_response(res, HTTPX::ProxyError)
       end
 
       def test_plugin_http_http_proxy

test/support/requests/plugins/ssrf_filter.rb

@@ -4,7 +4,7 @@ module Requests
   module Plugins
     module SsrfFilter
       def test_plugin_ssrf_filter_allows
-        uri = "#{scheme}httpbin.org"
+        uri = "#{scheme}nghttp2.org"
 
         session = HTTPX.plugin(:ssrf_filter)
         response = session.get(uri)

test/support/requests/plugins/upgrade.rb

@@ -6,28 +6,30 @@ module Requests
       def test_plugin_upgrade_h2
         return unless origin.start_with?("https://")
 
-        http = HTTPX.plugin(SessionWithPool)
+        start_test_servlet(H2Upgrade, alpn_protocols: %w[http/1.1 h2]) do |server|
+          http = HTTPX.plugin(SessionWithPool)
 
-        http = http.with(ssl: { alpn_protocols: %w[http/1.1] }) # disable alpn negotiation
+          http = http.with(ssl: { verify_mode: OpenSSL::SSL::VERIFY_NONE, alpn_protocols: %w[http/1.1] }) # disable alpn negotiation
 
-        http.plugin(:upgrade).wrap do |session|
-          uri = build_uri("/", "https://stadtschreiber.ruhr")
+          http.plugin(:upgrade).wrap do |session|
+            uri = "#{server.origin}/"
 
-          request = session.build_request("GET", uri)
-          request2 = session.build_request("GET", uri)
+            request = session.build_request("GET", uri)
+            request2 = session.build_request("GET", uri)
 
-          response = session.request(request)
-          verify_status(response, 200)
-          assert response.version == "1.1", "first request should be in HTTP/1.1"
-          response.close
-          # verifies that first request was used to upgrade the connection
-          verify_header(response.headers, "upgrade", "h2,h2c")
-          response2 = session.request(request2)
-          verify_status(response2, 200)
-          assert response2.version == "2.0", "second request should already be in HTTP/2"
-          response2.close
+            response = session.request(request)
+            verify_status(response, 200)
+            assert response.version == "1.1", "first request should be in HTTP/1.1"
+            response.close
+            # verifies that first request was used to upgrade the connection
+            verify_header(response.headers, "upgrade", "h2")
+            response2 = session.request(request2)
+            verify_status(response2, 200)
+            assert response2.version == "2.0", "second request should already be in HTTP/2"
+            response2.close
+          end
         end
-      end
+      end unless RUBY_ENGINE == "jruby"
 
       def test_plugin_upgrade_websockets
         return unless origin.start_with?("http://")

test/support/servlets/keep_alive.rb

@@ -33,20 +33,20 @@ class KeepAlivePongThenGoawayServer < TestHTTP2Server
   attr_reader :pings, :pongs
 
   def initialize(**)
-    @sent = false
-    super
+    @sent = Hash.new(false)
+    super()
   end
 
   private
 
   def handle_stream(conn, stream)
     # responds once, then closes the connection
-    if @sent
+    if @sent[conn]
       conn.goaway
-      @sent = false
+      @sent[conn] = false
     else
       super
-      @sent = true
+      @sent[conn] = true
     end
   end
 end

test/support/servlets/test.rb

@@ -23,7 +23,7 @@ end
 class TestHTTP2Server
   attr_reader :origin
 
-  def initialize(tls: true)
+  def initialize(tls: true, alpn_protocols: %w[h2])
     @port = 0
     @host = "localhost"
 
@@ -44,12 +44,12 @@ class TestHTTP2Server
       ctx.key = OpenSSL::PKey.read(File.read(File.join(certs_dir, "server.key")))
 
       ctx.ssl_version = :TLSv1_2
-      ctx.alpn_protocols = ["h2"]
+      ctx.alpn_protocols = alpn_protocols
 
       ctx.alpn_select_cb = lambda do |protocols|
-        raise "Protocol h2 is required" unless protocols.include?("h2")
+        raise "Protocol h2 is required" if (alpn_protocols & protocols).empty?
 
-        "h2"
+        protocols.first
       end
 
       @server = OpenSSL::SSL::SSLServer.new(@server, ctx)
@@ -125,7 +125,7 @@ class TestHTTP2Server
         close_socket(sock)
         return
       else
-        @conns[sock] << data
+        buffer_to_socket(sock, data)
 
         if sock.closed?
           purge_socket(sock)
@@ -139,6 +139,10 @@ class TestHTTP2Server
     close_socket(sock)
   end
 
+  def buffer_to_socket(sock, data)
+    @conns[sock] << data
+  end
+
   def handle_stream(_conn, stream)
     stream.on(:half_close) do
       response = "OK"

test/support/servlets/upgrade.rb

@@ -0,0 +1,17 @@
+# frozen_string_literal: true
+
+class H2Upgrade < TestHTTP2Server
+  def buffer_to_socket(sock, data)
+    return super unless @conns[sock].state == :waiting_magic &&
+                        !data.start_with?("PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n")
+
+    # assume HTTP/1.1
+    sock << "HTTP/1.1 200 OK\r\n" \
+            "Content-Length: 2\r\n" \
+            "Content-Type: text/plain\r\n" \
+            "Connection: Upgrade\r\n" \
+            "Upgrade: h2\r\n\r\n" \
+            "OK"
+    sock.close
+  end
+end