Merge branch 'gh-100' into 'master'

fixes to socket addresses expiration rebalancing

See merge request os85/httpx!407

lib/httpx/connection.rb

@@ -44,7 +44,7 @@ module HTTPX
 
     attr_accessor :current_session, :family
 
-    protected :sibling
+    protected :ssl_session, :sibling
 
     def initialize(uri, options)
       @current_session = @current_selector =
@@ -177,7 +177,7 @@ module HTTPX
 
     def merge(connection)
       @origins |= connection.instance_variable_get(:@origins)
-      if connection.ssl_session
+      if @ssl_session.nil? && connection.ssl_session
         @ssl_session = connection.ssl_session
         @io.session_new_cb do |sess|
           @ssl_session = sess

lib/httpx/connection/http2.rb

@@ -23,8 +23,8 @@ module HTTPX
     end
 
     class GoawayError < Error
-      def initialize
-        super(0, :no_error)
+      def initialize(code = :no_error)
+        super(0, code)
       end
     end
 
@@ -385,12 +385,10 @@ module HTTPX
           while (request = @pending.shift)
             emit(:error, request, error)
           end
-        when :no_error
-          ex = GoawayError.new
+        else
+          ex = GoawayError.new(error)
           @pending.unshift(*@streams.keys)
           teardown
-        else
-          ex = Error.new(0, error)
         end
 
         if ex

lib/httpx/io/tcp.rb

@@ -14,7 +14,10 @@ module HTTPX
 
     def initialize(origin, addresses, options)
       @state = :idle
+      @keep_open = false
       @addresses = []
+      @ip_index = -1
+      @ip = nil
       @hostname = origin.host
       @options = options
       @fallback_protocol = @options.fallback_protocol
@@ -53,8 +56,8 @@ module HTTPX
         @addresses = [*@addresses[0, ip_index], *addrs, *@addresses[ip_index..-1]]
       else
         @addresses.unshift(*addrs)
-        @ip_index += addrs.size if @ip_index
       end
+      @ip_index += addrs.size
     end
 
     # eliminates expired entries and returns whether there are still any left.
@@ -63,9 +66,7 @@ module HTTPX
 
       @addresses.delete_if(&:expired?)
 
-      unless (decr = prev_addr_size - @addresses.size).zero?
-        @ip_index = @addresses.size - decr
-      end
+      @ip_index = @addresses.size - 1 if prev_addr_size != @addresses.size
 
       @addresses.any?
     end
@@ -81,6 +82,17 @@ module HTTPX
     def connect
       return unless closed?
 
+      if @addresses.empty?
+        # an idle connection trying to connect with no available addresses is a connection
+        # out of the initial context which is back to the DNS resolution loop. This may
+        # happen in a fiber-aware context where a connection reconnects with expired addresses,
+        # and context is passed back to a fiber on the same connection while waiting for the
+        # DNS answer.
+        log { "tried connecting while resolving, skipping..." }
+
+        return
+      end
+
       if !@io || @io.closed?
         transition(:idle)
         @io = build_socket
@@ -88,29 +100,33 @@ module HTTPX
       try_connect
     rescue Errno::EHOSTUNREACH,
            Errno::ENETUNREACH => e
-      raise e if @ip_index <= 0
+      @ip_index -= 1
+
+      raise e if @ip_index.negative?
 
       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,
            SocketError,
            IOError => e
-      raise e if @ip_index <= 0
+      @ip_index -= 1
+
+      raise e if @ip_index.negative?
 
       log { "failed connecting to #{@ip} (#{e.message}), trying next..." }
-      @ip_index -= 1
       @io = build_socket
       retry
     rescue Errno::ETIMEDOUT => e
-      raise ConnectTimeoutError.new(@options.timeout[:connect_timeout], e.message) if @ip_index <= 0
+      @ip_index -= 1
+
+      raise ConnectTimeoutError.new(@options.timeout[:connect_timeout], e.message) if @ip_index.negative?
 
       log { "failed connecting to #{@ip} (#{e.message}), trying next..." }
-      @ip_index -= 1
+
       @io = build_socket
       retry
     end

regression_tests/bug_1_6_0_test.rb

@@ -0,0 +1,42 @@
+# frozen_string_literal: true
+
+require "test_helper"
+require "support/http_helpers"
+require "webmock/minitest"
+
+class Bug_1_6_1_Test < Minitest::Test
+  include HTTPHelpers
+
+  def test_retries_should_retry_on_goaway_cancel
+    start_test_servlet(GoawayCancelErrorServer) do |server|
+      http = HTTPX.plugin(SessionWithPool)
+                  .plugin(RequestInspector)
+                  .plugin(:retries)
+                  .with(ssl: { verify_mode: OpenSSL::SSL::VERIFY_NONE })
+
+      uri = "#{server.origin}/"
+      response = http.get(uri)
+      verify_status(response, 200)
+      assert http.calls == 1, "expect request to be built 1 more time (was #{http.calls})"
+      http.close
+    end
+  end
+
+  class GoawayCancelErrorServer < TestHTTP2Server
+    def initialize(**)
+      @sent = Hash.new(false)
+      super
+    end
+
+    private
+
+    def handle_stream(conn, stream)
+      if @cancelled
+        super
+      else
+        conn.goaway(:cancel)
+        @cancelled = true
+      end
+    end
+  end
+end

sig/connection/http2.rbs

@@ -106,7 +106,7 @@ module HTTPX
     end
 
     class GoawayError < Error
-      def initialize: () -> void
+      def initialize: (?Symbol code) -> void
     end
 
     class PingError < Error

test/io/tcp_test.rb

@@ -0,0 +1,63 @@
+# frozen_string_literal: true
+
+require "tempfile"
+require_relative "../test_helper"
+
+class TCPTest < Minitest::Test
+  include HTTPX
+
+  def test_tcp_ip_index_rebalance_on_new_addresses
+    origin = URI("http://example.com")
+    options = Options.new
+
+    tcp_class = Class.new(TCP) do
+      attr_accessor :ip_index
+    end
+
+    # initialize with no addresses, ip index points nowhere
+    tcp = tcp_class.new(origin, [], options)
+    assert tcp.ip_index == -1
+
+    # initialize with addresses, ip index points to the last element
+    tcp1 = tcp_class.new(origin, [Resolver::Entry.new("127.0.0.1")], options)
+    assert tcp1.addresses == ["127.0.0.1"]
+    assert tcp1.ip_index.zero?
+    tcp2 = tcp_class.new(origin, [Resolver::Entry.new("127.0.0.1"), Resolver::Entry.new("127.0.0.2")], options)
+    assert tcp2.addresses == ["127.0.0.1", "127.0.0.2"]
+    assert tcp2.ip_index == 1
+    tcp3 = tcp_class.new(origin, [Resolver::Entry.new("::1")], options)
+    assert tcp3.addresses == ["::1"]
+    assert tcp3.ip_index.zero?
+
+    # add addresses, ip index must point to previous ip after address expansion
+    tcp.add_addresses([Resolver::Entry.new("::1")])
+    assert tcp.addresses == ["::1"]
+    assert tcp.ip_index.zero?
+    tcp1.add_addresses([Resolver::Entry.new("::1")])
+    assert tcp1.addresses == ["::1", "127.0.0.1"]
+    assert tcp1.ip_index == 1
+    # makes the ipv6 address the next address to try
+    tcp2.add_addresses([Resolver::Entry.new("::1")])
+    assert tcp2.addresses == ["127.0.0.1", "::1", "127.0.0.2"]
+    assert tcp2.ip_index == 2
+    tcp3.add_addresses([Resolver::Entry.new("127.0.0.1")])
+    assert tcp3.addresses == ["127.0.0.1", "::1"]
+    assert tcp3.ip_index == 1
+    tcp3.add_addresses([Resolver::Entry.new("::2")])
+    assert tcp3.addresses == ["127.0.0.1", "::2", "::1"]
+    assert tcp3.ip_index == 2
+
+    # expiring entries should recalculate the pointer
+    now = Utils.now
+    tcp4 = tcp_class.new(origin, [Resolver::Entry.new("127.0.0.1", now + 1), Resolver::Entry.new("127.0.0.2", now + 4)], options)
+    assert tcp4.addresses == ["127.0.0.1", "127.0.0.2"]
+    assert tcp4.ip_index == 1
+    sleep(2)
+    assert tcp4.addresses?
+    assert tcp4.addresses == ["127.0.0.2"]
+    assert tcp4.ip_index.zero?
+    sleep(2)
+    assert !tcp4.addresses?
+    assert tcp4.ip_index == -1
+  end
+end