Commit 1faaeff4 authored by Tiago's avatar Tiago
Browse files

http2: resend unprocessed requests on GOAWAY 

a GOAWAY frame will inform of the last stream id where data may have
been processed. anything beyond that is considered unprocessed, and may
as well be sent on a new connection.

previously, httpx was bubbling up the goaway error to requests that were
send but never processed, and even to pending requests. This is now
corrected, and requests which were sent with a stream id above the
GOAWAY last frame id will be rerouted to a new connection, as well as
pending requests from the parser.
parent dc080710

lib/httpx/connection/http2.rb

+33 −6
Changes for lib/httpx/connection/http2.rb: 33 added lines, 6 removed lines.
Original line number Diff line number Diff line
@@ -23,7 +23,10 @@ module HTTPX 
    end



    class GoawayError < Error

      def initialize(code = :no_error)

      attr_reader :last_stream_id



      def initialize(code, last_stream_id)

        @last_stream_id = last_stream_id

        super(0, code)

      end

    end
@@ -155,18 +158,43 @@ module HTTPX 
    end



    def handle_error(ex, request = nil)

      if ex.is_a?(OperationTimeoutError) && !@handshake_completed && @connection.state != :closed

      last_stream_id = 0

      case ex

      when OperationTimeoutError

        if !@handshake_completed && @connection.state != :closed

          @connection.goaway(:settings_timeout, "closing due to settings timeout")

          emit(:close_handshake)

          settings_ex = SettingsTimeoutError.new(ex.timeout, ex.message)

          settings_ex.set_backtrace(ex.backtrace)

          ex = settings_ex

        end

      while (req, _ = @streams.shift)

      when GoawayError

        last_stream_id = ex.last_stream_id

      end



      inflight_unprocessed_requests = [] #: Array[Request]



      while (req, stream = @streams.shift)

        next if request && request == req



        if stream.id > last_stream_id

          req.transition(:idle)

          # unprocessed request

          inflight_unprocessed_requests << req



          next

        end



        emit(:error, req, ex)

      end



      if ex.is_a?(GoawayError)

        # resend unprocessed requests on a different connection

        @pending.unshift(*inflight_unprocessed_requests) if inflight_unprocessed_requests.any?

        emit(:exhausted) if @pending.any?

        return

      end



      while (req = @pending.shift)

        next if request && request == req
@@ -445,7 +473,7 @@ module HTTPX 
      send_pending

    end



    def on_close(_last_frame, error, _payload)

    def on_close(last_stream_id, error, _payload)

      is_connection_closed = @connection.closed?

      if error

        @buffer.clear if is_connection_closed
@@ -455,12 +483,11 @@ module HTTPX 
            emit(:error, request, error)

          end

        else

          ex = GoawayError.new(error)

          ex = GoawayError.new(error, last_stream_id)

          ex.set_backtrace(caller)



          handle_error(ex)

          teardown



        end

      end

      return unless is_connection_closed && @streams.empty?

sig/connection/http2.rbs

+3 −1
Changes for sig/connection/http2.rbs: 3 added lines, 1 removed line.
Original line number Diff line number Diff line
@@ -116,7 +116,9 @@ module HTTPX 
    end



    class GoawayError < Error

      def initialize: (?Symbol code) -> void

      attr_reader last_stream_id: Integer



      def initialize: (Symbol code, Integer last_stream_id) -> void

    end



    class PingError < Error

test/https_test.rb

+13 −0
Changes for test/https_test.rb: 13 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -137,6 +137,19 @@ class HTTPSTest < Minitest::Test 
    end

  end



  def test_http2_retry_unprocessed_requests_on_goaway

    start_test_servlet(CloseAfterXRequests) do |server|

      HTTPX.plugin(SessionWithPool).with(ssl: { verify_mode: OpenSSL::SSL::VERIFY_NONE }).wrap do |http|

        uri = "#{server.origin}/"

        responses = http.get(uri, uri, uri)

        assert responses.size == 3

        connection_count = http.connection_count

        # the test server does not support pipelining

        assert connection_count == 3, "expected to have 3 connections, instead have #{connection_count}"

      end

    end

  end



  def test_http2_uncoalesce_on_misdirected

    uri = build_uri("/status/421")

    HTTPX.plugin(SessionWithPool).wrap do |http|

test/support/servlets/close_after_x_requests.rb

0 → 100644
+33 −0
Changes for test/support/servlets/close_after_x_requests.rb: 33 added lines, 0 removed lines.
Original line number Diff line number Diff line 
# frozen_string_literal: true



class CloseAfterXRequests < TestHTTP2Server

  def initialize(requests_to_process: 1, **kw)

    super(**kw)

    @num_requests = @requests_to_process = requests_to_process

  end



  private



  def handle_stream(conn, stream)

    response = "".b



    stream.on(:data) do |data|

      response << data

    end



    stream.on(:half_close) do

      stream.headers({

                       ":status" => "200",

                       "content-length" => response.bytesize.to_s,

                       "content-type" => "text/plain",

                     }, end_stream: false)

      stream.data(response, end_stream: true)

      @num_requests -= 1



      if @num_requests.zero?

        conn.goaway

        @num_requests = @requests_to_process

      end

    end

  end

end