[BUG] Request & response compression (Gzip) not behaving as advertised

What is the bug?

The doc comments for the initialize method in the client class state:

# ... :compression Whether to compress requests. Gzip compression will be used. 
#   The default is false. Responses will automatically be inflated if they are compressed.
#   If a custom transport object is used, it must handle the request compression and response inflation.

There are two separate claims:

1. That when compression: true Gzip compression will be applied to the request bodies. AFAICT, there is no compression applied at any point in this gem. It appears that in the transport layer the Accept-Encoding is conditionally set, however no other transformation is applied to the request on condition of this flag's state.

2. That responses received that have been compressed will be automatically inflated, even if the compression flag is false - the default state. To me, this is implied in the note that the flag only controls request behavior in the initial description and that this sentence follows the call-out of the default state. However, the guard that is currently in place will only inflate responses if the compression flag has been passed to the client during initialization.

How can one reproduce the bug?

Reproduction of point 1:

require 'opensearch'
require 'webrick'
require 'stringio'

port = 19200
seen = {}
server = WEBrick::HTTPServer.new(Port: port, Logger: WEBrick::Log.new(File::NULL), AccessLog: [])
server.mount_proc('/') do |req, res|
  seen[:content_encoding]   = req['Content-Encoding']
  seen[:accept_encoding]    = req['Accept-Encoding']
  seen[:gzip_bytes_present] = req.body&.b&.byteslice(0, 2) == "\x1f\x8b".b # same as GZIP_FIRST_TWO_BYTES in transport base
  seen[:body]               = req.body

  res['Content-Type'] = 'application/json'
  res.body = '{}'

  # attempt inflation
  begin
    io = StringIO.new(req.body)
    reader = Zlib::GzipReader.new(io)
    reader.read
  rescue => e
    seen[:inflation_failed] = e
  end
end

Thread.new { server.start }
sleep 0.2

transport = OpenSearch::Transport::Transport::HTTP::Faraday.new(
  hosts: [{ host: '127.0.0.1', port: port, protocol: 'http' }],
  options: { compression: true }
)
transport.perform_request('POST', '/', {}, { hello: 'world' })
server.shutdown
puts "Accept-Encoding:  #{seen[:accept_encoding].inspect}"  # => "gzip"
puts "Content-Encoding: #{seen[:content_encoding].inspect}" # => nil
puts "Body gzipped?:    #{seen[:gzip_bytes_present]}"       # => false
puts "Inflation failed: #{seen[:inflation_failed]}"         # => not in gzip format
puts "Body:             #{seen[:body].inspect}"             # => '{"hello":"world"}'

Reproduction of point 2, exercising the divergent behavior in the base transport layer:

require 'opensearch'
require 'zlib'
require 'stringio'

def gzip(str)
  io = StringIO.new
  Zlib::GzipWriter.wrap(io) { |gz| gz.write(str) }
  io.string
end

def gzipped?(str)
  str.b.byteslice(0, 2) == "\x1f\x8b".b # same as GZIP_FIRST_TWO_BYTES in transport base
end

compressed = gzip('{"ok":true}')
def transport(compression:)
  opts = compression.nil? ? {} : { compression: compression }
  OpenSearch::Transport::Transport::HTTP::Faraday.new(
    hosts: [{ host: 'localhost', port: 9200 }],
    options: opts
  )
end
raw = transport(compression: nil).send(:decompress_response, compressed)
puts "without compression still gzip?: #{gzipped?(raw)}"      # => true
puts "without compression equal input?: #{raw == compressed}" # => true
plain = transport(compression: true).send(:decompress_response, compressed)
puts "with compression still gzip?: #{gzipped?(plain)}" # => false
puts "with compression body: #{plain.inspect}"          # => "{\"ok\":true}"

What is the expected behavior?

1. That in addition to the correctly set Accept-Encoding header, the body is compressed if not already in a compressed state and that Content-Encoding is set correctly to gzip.
2. That all Gzip compressed requests received by the client are inflated, regardless of the compression flag state.

Do you have any additional context?

It appears this was fixed in the elasticsearch-ruby gem back in August of 2021, which lines up with around when this project was forked based on what I'm seeing in early issues (e.g. Milestone 1 tasks). I would be happy to handle pulling that MR through this repository if that would work for everyone.

EDIT: Advised during the triage meeting that we'll need to avoid pulling in any direct ElasticSearch changes for licensing reasons. I'm happy to roll an independent solution.