Ensure wire always contains a full H/2 frame

[maruth-stripe]

Follow-up to socketry/async-io#72 to fix #14
Updates read_header to use Stream#peek instead of `Stream#read. We maintain the invariant that the read buffer / wire is always frame-aligned. Instead of reading data off the wire when reading the header for an H/2 frame, we just peek the header and then read the full frame.

Types of Changes

• Bug fix.

Contribution

• I added tests for my changes.
• I tested my changes locally.
• I agree to the Developer's Certificate of Origin 1.1.

[maruth-stripe]

Verified the fix works with the following script (modification to the repro in #14)

# requires
require "stringio"
require "async/reactor"


class FunkyIO
  def initialize
    @f = Protocol::HTTP2::DataFrame.new(401, 0, Protocol::HTTP2::DataFrame::TYPE, 13, "a" * 13)
    @sio = StringIO.new
    @f.write_header(@sio)
    @sio.rewind

    @state = :read_header
  end

  def peek(size, buf = nil)
    if @state == :read_header || @state == :try_header_again
      res = @sio.read(size, buf)

      case @state
      when :read_header 
        @state = :now_timeout
      when :try_header_again
        @state = :read_frame
      end

      @sio = StringIO.new
      @f.write_header(@sio)
      @f.write_payload(@sio)
      @sio.rewind
      
      return res
    end
  end

  def read(size, buf = nil)
    case @state
    when :now_timeout
      @state = :try_header_again
      raise Async::TimeoutError
    when :read_frame
      res = @sio.read(size, buf)
      res
    end
  end
end

f = Protocol::HTTP2::Framer.new(FunkyIO.new)
Async::Reactor.run do
  begin
    p(f.read_frame)
  rescue Async::TimeoutError
    # try again
  end

  p(f.read_frame)
end

outputs
#<Protocol::HTTP2::DataFrame stream_id=401 flags=0 13b>

[ioquatix]

I released async-io v1.37.0 which I assume we can use here?

[maruth-stripe]

Hi @ioquatix really sorry for disappearing on this PR for a bit, got caught up with other stuff!

I've updated it a bit, and tested it with the follwing script:

# requires
require "stringio"
require "async"
require "async/reactor"
require "protocol/http2/framer"
require "protocol/http2/data_frame"



class FunkyIO
  def initialize
    @f = Protocol::HTTP2::DataFrame.new(401, 0, Protocol::HTTP2::DataFrame::TYPE, 13, "a" * 13)
    @sio = StringIO.new
    @f.write_header(@sio)
    @sio.rewind

    @state = :read_header
  end

	def sync=(x)
	end

  def read_nonblock(size, buf = nil, exception)
		puts "read_nonblock(#{size}) state: #{@state}"
    if @state == :read_header || @state == :try_header_again
      res = @sio.read(size, buf)

      case @state
      when :read_header 
        @state = :now_timeout
      when :try_header_again
        @state = :read_frame
      end

      @sio = StringIO.new
      @f.write_payload(@sio)
      @sio.rewind
      
      return res
		elsif @state == :now_timeout
      @state = :try_header_again
			raise Async::TimeoutError
		elsif @state == :read_frame
			return @sio.read(size, buf)
    end
  end
end

f = Protocol::HTTP2::Framer.new(FunkyIO.new)
Async::Reactor.run do
  begin
    p(f.read_frame)
  rescue Async::TimeoutError
    # try again
  end

  p(f.read_frame)
end

which outputs

read_nonblock(65536) state: read_header
read_nonblock(65536) state: now_timeout
read_nonblock(65536) state: try_header_again
#<Protocol::HTTP2::DataFrame stream_id=401 flags=0 13b>

I'm not sure why but bundle exec bake test seems to be taking quite a while. Unsure if it's something messed up with my local setup.

[maruth-stripe]

One note: this PR now implicitly adds a requirement that the stream passed into Framer must implement sync= and read_nonblock (or be an `Async::IO::Stream)

I believe StringIO, Socket do implement this.

[maruth-stripe]

@ioquatix do you know why the test suite might be timing out here?

[ioquatix]

The next step for this code is to work with standard IO instances. However, I don't think a method like peek exists for IO. Do you have any ideas?

[ioquatix]

After considering this further, I believe we should adopt the peek(n) model in this code. We will have to use an appropriate wrapper.

[ioquatix]

I'm going to merge this into another branch so I can finish it off.