Undocumented breaking changes + better support for mocking client connections and message streams to protect against them

[chrisbloom7]

First, thank you for providing this library and the other async tools for Ruby. I know how much work goes into managing an open source library, never mind a suite of them plus your other projects. I bring this up not as a criticism but in the hopes of making the project better and let you know of a particular pain point I'm experiencing.

I've got gem "async-websocket", "~> 0.19" in my Gem file and recently ran bundle update --strict to update my dependencies. Somewhere in the dependency tree of async-websocket, between

    async (1.30.1)
      console (~> 1.10)
      nio4r (~> 2.3)
      timers (~> 4.1)
    async-http (0.56.5)
      async (>= 1.25)
      async-io (>= 1.28)
      async-pool (>= 0.2)
      protocol-http (~> 0.22.0)
      protocol-http1 (~> 0.14.0)
      protocol-http2 (~> 0.14.0)
    async-io (1.32.2)
      async
    async-pool (0.3.9)
      async (>= 1.25)
    async-websocket (0.19.0)
      async-http (~> 0.54)
      async-io (~> 1.23)
      protocol-websocket (~> 0.7.0)

and gem "async-websocket", "~> 0.19"

    async (2.12.1)
      console (~> 1.25, >= 1.25.2)
      fiber-annotation
      io-event (~> 1.6, >= 1.6.5)
    async-http (0.69.0)
      async (>= 2.10.2)
      async-pool (~> 0.7)
      io-endpoint (~> 0.11)
      io-stream (~> 0.4)
      protocol-http (~> 0.26)
      protocol-http1 (~> 0.19)
      protocol-http2 (~> 0.18)
      traces (>= 0.10)
    async-pool (0.7.0)
      async (>= 1.25)
    async-websocket (0.26.2)
      async-http (~> 0.54)
      protocol-rack (~> 0.5)
      protocol-websocket (~> 0.14)

the format of the message returned from

Async::WebSocket::Client.connect(endpoint) do |connection|
  while (message = connection.read)

changed from a Hash-like object that allowed direct access to message keys like message[:envelope_id] to a Protocol::WebSocket::Message object that requires calling message.parse (using a custom JSON parser proxy object for some reason) before being able to access message keys. That's a pretty big change to slip in there in the absence of a Changelog, major version bump, or at least a deprecation process with warning messages. It completely broke our websocket client without warning and led to a lot of scrambling to try to figure out what changed.

We still haven't got it back up and running fully, but troubleshooting it locally is difficult as there are no complete guides or built-in methods to mock an Async::Websocket connection and a message IO stream in tests, and stubbing them manually involves jumping through numerous dependent libraries to try to figure out what other async objects and methods need to be stubbed in order to get a mock connection working. (To date, I still haven't had any success.) This also leaves me without a way to protect against breaking changes after updates.

I'd like to ask two things of this project:

1. Our client was working in production for several years without issue until we upgraded. If this is intended to be production ready, please add a Changelog to the project that is updated per release, and/or commit to strict SemVer versioning where a breaking change like that requires a major version bump. And if it's not intended to be production ready/is still experimental, please indicate that more strongly in the README.
2. Please provide a guide or helper methods for mocking async websocket connections and sending messages across the IO stream. The testing example from the async-http project does not offer any tips for connecting that through to a mocked async-websocket client connection, nor how to send messages across it. Additionally, the requirement of standing up and running a real HTTP server in order to receive mock responses in tests deserves more information, along with how to properly route requests in the test suite.

[ioquatix]

That's fair, it was a huge breaking change.

I don't think there will be any other major changes to the interface going forward.

I agree with your suggestions.

Regarding mocking, would it be sufficient just to replay a set of messages?

[chrisbloom7]

Following up on this: I've spent the last two days trying to setup either a mock connection or a mock server that's just used for my tests, and I've had no luck. I'd prefer the latter so I can test passing real messages back and forth and which will also help protect against future breaking changes in the Async library since it would be more of an integration test than unit testing with a bunch of expected behaviour that may not be true outside of the test env.

I'm using Rspec. I've tried recreating the active server connection from https://github.com/socketry/sus-fixtures-async-http/blob/main/lib/sus/fixtures/async/http/server_context.rb and https://github.com/socketry/async-websocket/blob/main/test/async/websocket/client.rb but have had no luck.

  let(:middleware) do
    Protocol::HTTP::Middleware.for do |request|
      Async::WebSocket::Adapters::HTTP.open(request) do |connection|
        while message = connection.read
          connection.write(message)
        end

        connection.shutdown
      rescue Protocol::WebSocket::ClosedError
        # Ignore this error.
      ensure
        connection.close
      end or Protocol::HTTP::Response[404, {}, []]
    end
  end
  let(:protocol) { Async::HTTP::Protocol::HTTP1 }
  let(:url) { "http://0.0.0.0:999/" } # http://localhost:0 is not available inside my codespace
  let(:endpoint) { Async::HTTP::Endpoint.parse(url, timeout: 0.8, reuse_port: true, protocol:) }

  before do
    WebMock::HttpLibAdapters::AsyncHttpClientAdapter.disable!

    # We need an endpoint that is already bound to an address so we know it's
    # ready to accept connections, but it needs to quack like a regular endpoint
    # for the sake of the server.
    @bound_endpoint = endpoint.bound
    @bound_endpoint.extend(SingleForwardable)
    @bound_endpoint.def_delegators(:endpoint, :authority, :path, :protocol, :scheme)

    @server = Async::HTTP::Server.new(middleware, @bound_endpoint)
    @server_task = Async do
        @server.run
    end

    # Hangs here...

    @client_endpoint = @bound_endpoint.local_address_endpoint(timeout: endpoint.timeout)
    def @client_endpoint.endpoint=(endpoint)
      @endpoint = endpoint
    end
    def @client_endpoint.endpoint
      @endpoint
    end
    @client_endpoint.endpoint = endpoint
    @client_endpoint.extend(SingleForwardable)
    @client_endpoint.def_delegators(:endpoint, :authority, :path, :protocol, :scheme)
    allow(Async::WebSocket::Client).to receive(:connect).and_wrap_original do |original_method| #, *arguments, &block|
      original_method.call(@client_endpoint) #, *arguments, &block)
    end
  end

  after do
    Async::Task.current.with_timeout(1) do
      @server_task&.stop
    end
  rescue RuntimeError # In case there is no Async::Task yet
    nil
  ensure
    @bound_endpoint&.close
    WebMock::HttpLibAdapters::AsyncHttpClientAdapter.enable!
  end

The latest is that it just hangs after the Async block... not sure why the asynch is being treated as synchronous here.

Minor annoyance: With webmock installed, Async::HTTP::Client get's converted to a Async::HTTP::WebMockClientWrapper automatically, so you have to manually call WebMock::HttpLibAdapters::AsyncHttpClientAdapter.disable! to get around that. This took me a long time to figure out where the WebMockClientWrapper was coming from. Putting it here for the next person that runs up against this.

[chrisbloom7]

It looks like there's a sample Rack app here that might be helpful, but not sure where to go about adding it.

[chrisbloom7]

Last night I discovered that https://github.com/socketry/async https://github.com/socketry/async-rspec is a thing, and that got me past the blocking Async call - I'm guessing that without it I was missing a reactor/scheduler to allow the async task to suspend itself around each test. Once that was in place I finally got a working end-to-end test with a mocked server.

  include_context Async::RSpec::Reactor

  let(:protocol) { Async::HTTP::Protocol::HTTP1 }
  let(:url) { "http://0.0.0.0:999/" }
  let(:endpoint) { Async::HTTP::Endpoint.parse(url, timeout: 0.8, reuse_port: true, protocol:) }

  let(:hello_message) { { type: :hello } }
  let(:events_api_message) { { type: "events_api", envelope_id: generate(:envelope_id) } }
  let(:interactive_message) { { type: "interactive", envelope_id: generate(:envelope_id) } }
  let(:disconnect_message) { { type: "disconnect", envelope_id: generate(:envelope_id) } }

  let(:payload_proc) do
    ->(connection) do
      connection.send_text(hello_message.to_json)
      connection.send_text(events_api_message.to_json)
      connection.send_text(interactive_message.to_json)
      connection.send_text(disconnect_message.to_json)
    end
  end

  let(:middleware) do
    Protocol::HTTP::Middleware.for do |request|
      Async::WebSocket::Adapters::HTTP.open(request) do |connection|
        payload_proc.call(connection)
        connection.close
      end or Protocol::HTTP::Response[404, {}, []]
    end
  end

  before do
    WebMock::HttpLibAdapters::AsyncHttpClientAdapter.disable!

    # We need an endpoint that is already bound to an address so we know it's
    # ready to accept connections, but it needs to quack like a regular endpoint
    # for the sake of the server.
    @bound_endpoint = endpoint.bound
    @bound_endpoint.extend(SingleForwardable)
    @bound_endpoint.def_delegators(:endpoint, :authority, :path, :protocol, :scheme)

    @server = Async::HTTP::Server.new(middleware, @bound_endpoint)
    @server_task = Async do
      @server.run
    end

    # As with the server endpoint, we need a bound client endpoint that quacks like an unbound endpoint
    @client_endpoint = @bound_endpoint.local_address_endpoint(timeout: endpoint.timeout)
    @client_endpoint.instance_variable_set(:@endpoint, endpoint)
    @client_endpoint.extend(SingleForwardable)
    @client_endpoint.def_delegators(:@endpoint, :authority, :path, :protocol, :scheme)

    allow(Async::WebSocket::Client).to receive(:connect).and_wrap_original do |original_method, *arguments, &block|
      original_method.call(@client_endpoint, *arguments[1..], &block)
    end
  end

  after do
    Async::Task.current.with_timeout(1) do
      @server_task&.stop
    end
  rescue RuntimeError
    # raised when the task is not running
    nil
  ensure
    @bound_endpoint&.close
    WebMock::HttpLibAdapters::AsyncHttpClientAdapter.enable!
  end

[chrisbloom7]

Spoke too soon 😞 For some reason I'm getting alternating failures of Errno::EPIPE: Pipe broken or, if I ignore those, then my tests fail because the client can only read 3 of the 4 messages the server sends.

[chrisbloom7]

If I add a connection.shutdown to the server block after sending all of the messages then the tests no longer fail but I get this warning on all of them

  0.0s     warn: Async::Task: Reading HTTP/1.1 requests for Async::HTTP::Protocol::HTTP1::Server. [oid=0x3944] [ec=0x3958] [pid=9583] [2024-09-20 07:02:24 +0000]
               | Task may have ended with unhandled exception.
               |   Protocol::WebSocket::ProtocolError: Received text, but expecting continuation!

If I skip the connection.shutdown and only have the server deliver 3 messages instead of 4, it works fine 🤷🏻

[chrisbloom7]

Another breaking change I just discovered

where this used to work:

Async::WebSocket::Client.connect(endpoint) do |connection|
  while (message = connection.read)
    if message[:envelope_id]
      connection.write({ envelope_id: message[:envelope_id] })

it now needs to be

Async::WebSocket::Client.connect(endpoint) do |connection|
  while (payload = connection.read)
    message = payload.parse(Protocol::WebSocket::Coder::JSON.new(symbolize_names: true))
    if message[:envelope_id]
      Protocol::WebSocket::TextMessage.generate({ envelope_id: message[:envelope_id] }).send(connection)

That took me forever to figure out

[ioquatix]

I think this should be good enough - you don't need to specify a custom coder unless you have very specific requirements:

Async::WebSocket::Client.connect(endpoint) do |connection|
  while (payload = connection.read)
    message = payload.parse
    if message[:envelope_id]
      Protocol::WebSocket::TextMessage.generate({envelope_id: message[:envelope_id]}).send(connection)

I don't believe there will be be any further changes to this interface.

[chrisbloom7]

Thanks for the clarification on the encoder. I included that line for context, but the breaking change I wanted to highlight was that connection.write({ envelope_id: message[:envelope_id] }) no longer worked.

[ioquatix]

Yes, unfortunately there was no good way to keep compatibility there, and it probably wasn't a good interface (for example, assuming JSON encoding).

[chrisbloom7]

Makes sense

[chrisbloom7]

Another breaking change. Not sure if this was in async-websockets or one its dependencies, but with roughly this combination of gems...

async (2.12.1)
async-http (0.69.0)
async-pool (0.7.0)
async-websocket (0.27.0)
console (1.25.2)
fiber-storage (0.1.2)
io-endpoint (0.11.0)
io-stream (0.4.0)
protocol-hpack (1.4.3)
protocol-http (0.26.8)
protocol-http1 (0.19.1)
protocol-rack (0.6.0)
protocol-websocket (0.15.0)
traces (0.11.1)

...you could call Async::WebSocket::Client.connect(endpoint) without a block and get back the connection. After upgrading async-websockets to the most recent (which bumped all the dependency gems too) this now results in a LocalJumpError: unexpected return error originating from

async-websocket/lib/async/websocket/client.rb

Line 61 in aec454a

return ClientCloseDecorator.new(client, connection) unless block_given?

. I couldn't spot any obvious reason for this change in the diff between async-websocket 0.27 and 0.30 so it's likely from one of the other gems, but this change required me to refactor a bunch of code I had just finished decoupling. (This is on top of the code I had to refactor for the other two changes.)

Just to reiterate, I would not expect minor version changes to introduce breaking changes to working apps, and if for some reason they had to, I'd expect to see some documentation about what changed.

[chrisbloom7]

Thanks for the response and for tracking these issues, in addition to your work on this suite of libraries in general. I seem to have both a working client and a working set of tests now, though I'd still love to understand what I might have gotten wrong in my test setup that would prevent the client from being able to receive 4 messages but OK if I only sent 3.

[ioquatix]

Thanks for the links to the example, it is helpful to have test cases to review. I'll see if I can answer your question.

[ioquatix]

Okay, I checked the general code flow, and the reason why you are getting LocalJumpError is because you are expecting to return a WebSocket connection, without using an Async block:

# Okay:
Sync do
  connection = Async::WebSocket::Client.connect(endpoint)
  # ...
end

# Okay:
Async::WebSocket::Client.connect(endpoint) do |connection|
  # ...
end

# Not okay:
connection = Async::WebSocket::Client.connect(endpoint)
# ...

So probably you changed some code that was invoking the #connect method within an event loop already. Would that have been possible with the refactoring you did?

[chrisbloom7]

Ah, interesting! Thanks for the explanation! That should help me clean the code up a bit more.