Consider a call flow like:

0. @read_buffer in our underlying async/io/stream contains exactly 9 bytes.
1. read_frame (takes 9 bytes off the underlying @read_buffer in consume_read_buffer) and this completely drains @read_buffer
2. successfully gets the header
3. reads payload, times out, @read_buffer is still empty, we do not parse the frame, and exit the call flow.
4. retry read_frame with a higher timeout
5. we enter read_header again, which will call fill_read_buffer (fills buffer with ~thousands of bytes)
6. @read_buffer in the underlying stream now contains the payload of the previous frame, instead of a valid frame header, and we get a protocol error.

I think in this case the "right" thing to do is put the 9 bytes back in the read buffer, or hold the frame header and retry reading the payload, instead of trying to read the header out of what is certainly payload.

def read_frame(maximum_frame_size = MAXIMUM_ALLOWED_FRAME_SIZE)
  # Read the header:
  length, type, flags, stream_id = read_header <- second time we come here, we're reading payload bytes, not header bytes
				
  # Async.logger.debug(self) {"read_frame: length=#{length} type=#{type} flags=#{flags} stream_id=#{stream_id} -> klass=#{@frames[type].inspect}"}
				
  # Allocate the frame:
  klass = @frames[type] || Frame
  frame = klass.new(stream_id, flags, type, length)
				
  # Read the payload:
  frame.read(@stream, maximum_frame_size) <- timeout occurs here
				
  # Async.logger.debug(self, name: "read") {frame.inspect}
				
  return frame
end