Most reliable pattern for blocking RabbitMQ subscribe in Ruby, with robust reconnect, logging, and error handling?

[kingpalethe]

I've been using RabbitMQ with the great "bunny" gem for a few years, and it has generally worked well, but I still have not found the 100% stable pattern for a blocking subscribe, which can do all of these things:

1. Reliably reconnect if the connection goes down
2. Reliably handle errors

After literally years of trying different patterns, and scouring the internet for a pattern which would achieve all of these things... below is my state-of-the-art pattern which I am currently using in production...

First -- you will notice am trying to do error handling in two different places... the final rescue => e statement is supposed to re-instantiate the connection and channel when either of those stop working... so far I think this sometimes works, and sometimes doesn't... after years of trying I still have not found a way to reliably recover from and re-instantiate these connections....

HERE IS THE BEST PATTERN I HAVE FOUND SO FAR:

def make_rabbit_queue(channel_name)
    rabbitmq_connection =  Bunny.new(ENV['RABBITMQ_URL'], {log_level: :error})
    rabbitmq_connection.start
    rabbitmq_channel = rabbitmq_connection.create_channel
    incoming_queue = rabbitmq_channel.queue(channel_name, durable: true)
    return [rabbitmq_connection, rabbitmq_channel, incoming_queue]
end


task poll: :environment do
  rabbitmq_connection, rabbitmq_channel, incoming_queue = make_rabbit_queue('my-queue-name')
  loop do
    begin
      incoming_queue.subscribe(manual_ack: true) do |delivery_info, properties, payload|
        begin
          body = JSON.parse(payload)
          do_something_with_body(body)
        rescue => e
          ErrorUtil.print_error(e)
        ensure
          rabbitmq_channel.acknowledge(delivery_info.delivery_tag, false)
        end
      end
      loop do
        sleep 1
      end
    rescue => e
      ErrorUtil.print_error(e)
      rabbitmq_connection, rabbitmq_channel, incoming_queue = make_rabbit_queue('my-queue-name')
      sleep 1
      next
    end
  end
end

[kingpalethe]

An example of a kind of error which is not successfully handled by this pattern:

E, [2023-10-30T16:14:12.452412 #9] ERROR -- #<Bunny::Session:0x9150 [email protected]:5671, vhost=asdflkasdjfl, addresses=[bunny-server-address.com:5671]>: Uncaught exception from consumer #<Bunny::Consumer:19020 @channel_id=1 @queue=queue-name @consumer_tag=bunny-1698674006000-410372492256>: #<Bunny::ChannelAlreadyClosed: cannot use a closed channel! Channel id: 1, closed due to a server-reported channel error: PRECONDITION_FAILED - delivery acknowledgement on channel 1 timed out. Timeout value used: 7200000 ms. This timeout value can be configured, see consumers doc guide to learn more> @ /usr/local/bundle/gems/bunny-2.22.0/lib/bunny/channel.rb:2075:in `raise_if_no_longer_open!'

So, a major weakness in my pattern is that it isn't able to re-open a closed channel.

[michaelklishin]

Moving the part that opens a channel into the loop would do (with an obvious risk of an infinite loop that leaks channels).

Detecting a server-initiated channel close in a consumer is difficult in general but if you are willing to use a loop like that, you can check the channel right before you add a consumer (subscribe).

Not really related to the conversation but this is a good example of how Erlang and Elixir make certain types of (concurrent) failure handling trivial. You wouldn't really think twice about how to achieve this in Elixir. But your version is reasonably close to somewhat similar in Ruby.

[kingpalethe]

Thanks for your reply. You wrote:

"Moving the part that opens a channel into the loop would do (with an obvious risk of an infinite loop that leaks channels)."

My code sample has two loops, and exterior loop and an interior loop... which loop do you suggest I move the channel opening into?

My guess is that you are saying to move the channel opening to the top of the first loop... then my code would look like this?

def make_rabbit_queue(channel_name)
    rabbitmq_connection =  Bunny.new(ENV['RABBITMQ_URL'], {log_level: :error})
    rabbitmq_connection.start
    incoming_queue = rabbitmq_channel.queue(channel_name, durable: true)
    return [rabbitmq_connection,  incoming_queue]
end


task poll: :environment do
  rabbitmq_connection,  incoming_queue = make_rabbit_queue('my-queue-name')
  loop do
    begin
      rabbitmq_channel = rabbitmq_connection.create_channel
      incoming_queue.subscribe(manual_ack: true) do |delivery_info, properties, payload|
        begin
          body = JSON.parse(payload)
          do_something_with_body(body)
        rescue => e
          ErrorUtil.print_error(e)
        ensure
          rabbitmq_channel.acknowledge(delivery_info.delivery_tag, false)
        end
      end
      loop do
        sleep 1
      end
    rescue => e
      ErrorUtil.print_error(e)
      rabbitmq_connection,  incoming_queue = make_rabbit_queue('my-queue-name')
      rabbitmq_channel = rabbitmq_connection.create_channel
      sleep 1
      next
    end
  end
end

thanks

[michaelklishin]

Something like this. Like I said, as long as you avoid a loop of channel.open frames and leaking channels, it should be a perfectly fine thing to do. Well, I assume that this loop won't actually run very often.