fix: Recover Hutch worker when consumer channel is closed due to delivery_ack_timeout?

Author: eglitobias

Gemfile

@@ -4,6 +4,8 @@ ruby '>= 2.7.0'
 
 gemspec
 
+gem 'bunny', github: 'ruby-amqp/bunny', branch: 'main'
+
 group :development do
   gem "rake"
   gem "guard", "~> 2.14", platform: :mri

lib/hutch/adapters/bunny.rb

@@ -11,7 +11,7 @@ class BunnyAdapter
       ConnectionRefused = Bunny::TCPConnectionFailed
       PreconditionFailed = Bunny::PreconditionFailed
 
-      def_delegators :@connection, :start, :disconnect, :close, :create_channel, :open?
+      def_delegators :@connection, :start, :disconnect, :close, :create_channel, :open?, :recover_channel_topology
 
       def initialize(opts={})
         @connection = Bunny.new(opts)

lib/hutch/broker.rb

@@ -4,6 +4,7 @@
 require 'hutch/logging'
 require 'hutch/exceptions'
 require 'hutch/publisher'
+require 'thread'
 
 module Hutch
   class Broker
@@ -113,13 +114,33 @@ def open_channel
           logger.info 'enabling publisher confirms'
           ch.confirm_select
         end
+
+        install_channel_recovery!(ch)
       end
     end
 
     def open_channel!
       @channel = open_channel
     end
 
+    def install_channel_recovery!(ch)
+      ch.on_error do |channel, close|
+        next unless close.delivery_ack_timeout?
+
+        # Reopen performs blocking protocol operations, so run it outside Bunny's
+        # channel error callback thread to avoid timing out waiting for OpenOk.
+        Thread.new do
+          begin
+            channel.reopen
+            connection.recover_channel_topology(channel)
+            logger.info 'channel recovery succeeded'
+          rescue => ex
+            logger.error "channel recovery failed: #{ex.class}: #{ex.message}"
+          end
+        end
+      end
+    end
+
     def declare_exchange(ch = channel)
       exchange_name = @config[:mq_exchange]
       exchange_type = @config[:mq_exchange_type]

spec/integration/channel_recovery_spec.rb

@@ -0,0 +1,139 @@
+require 'spec_helper'
+require 'hutch/broker'
+require 'hutch/worker'
+require 'hutch/consumer'
+require 'bunny'
+require 'json'
+require 'securerandom'
+require 'timeout'
+
+describe 'channel recovery after delivery acknowledgement timeout', rabbitmq: true, adapter: :bunny do
+  let(:log) { StringIO.new }
+  let(:logger) { Logger.new(log) }
+  let(:exchange_name) { "hutch.integration.exchange.#{SecureRandom.hex(4)}" }
+  let(:queue_name) { "hutch.integration.queue.#{SecureRandom.hex(4)}" }
+  let(:routing_key) { "hutch.integration.key.#{SecureRandom.hex(4)}" }
+
+  let(:processed) { [] }
+  let(:processed_lock) { Mutex.new }
+  let(:timed_out_once) { [false] }
+
+  let(:consumer_class) do
+    msgs = processed
+    lock = processed_lock
+    rk = routing_key
+    qn = queue_name
+    timed_out = timed_out_once
+
+    Class.new do
+      include Hutch::Consumer
+
+      consume rk
+      queue_name qn
+      arguments(
+        'x-queue-type' => 'quorum',
+        'x-consumer-timeout' => 60_000
+      )
+
+      define_method(:process) do |message|
+        if message['id'] == 'trigger-timeout' && !timed_out[0]
+          timed_out[0] = true
+          sleep 210
+        end
+
+        lock.synchronize { msgs << message['id'] }
+      end
+    end
+  end
+
+  let(:broker) { Hutch::Broker.new }
+  let(:worker) { Hutch::Worker.new(broker, [consumer_class], []) }
+
+  let(:publisher) do
+    Bunny.new(
+      host: Hutch::Config[:mq_host],
+      port: Hutch::Config[:mq_port],
+      username: Hutch::Config[:mq_username],
+      password: Hutch::Config[:mq_password],
+      vhost: Hutch::Config[:mq_vhost]
+    ).tap(&:start)
+  end
+
+  let(:publisher_channel) { publisher.create_channel }
+  let(:exchange) { publisher_channel.topic(exchange_name, durable: true) }
+
+  before do
+    Hutch::Logging.logger = logger
+    Hutch::Config.set(:mq_exchange, exchange_name)
+    Hutch::Config.set(:force_publisher_confirms, false)
+    Hutch::Config.set(:client_logger, logger)
+  end
+
+  after do
+    publisher_channel.close rescue nil
+    publisher.close rescue nil
+    broker.disconnect rescue nil
+    Hutch::Logging.logger = Logger.new(File::NULL)
+  end
+
+  def wait_for(timeout, label)
+    Timeout.timeout(timeout) do
+      loop do
+        return true if yield
+        sleep 0.25
+      end
+    end
+  rescue Timeout::Error
+    raise <<~MSG
+      Timed out waiting for: #{label}
+
+      processed_messages=#{processed_messages.inspect}
+      channel_open=#{broker.channel.open? rescue 'unknown'}
+      channel_closed=#{broker.channel.closed? rescue 'unknown'}
+
+      log output:
+      #{log_output}
+    MSG
+  end
+
+  def processed_messages
+    processed_lock.synchronize { processed.dup }
+  end
+
+  def log_output
+    log.rewind
+    log.read
+  end
+
+  def publish_message(id)
+    exchange.publish(
+      JSON.dump('id' => id),
+      routing_key: routing_key,
+      content_type: 'application/json',
+      persistent: true
+    )
+  end
+
+  # This spec is intentionally slow because RabbitMQ enforces delivery
+  # acknowledgement timeouts on a periodic sweep, not immediately at the deadline.
+  it 're-subscribes and consumes later messages after RabbitMQ closes the channel for ack timeout' do
+    broker.connect
+    worker.setup_queues
+
+    publish_message('trigger-timeout')
+
+    wait_for(240, 'delivery acknowledgement timeout') do
+      log_output.match?(/delivery acknowledgement on channel \d+ timed out/i)
+    end
+
+    publish_message('after-recovery')
+
+    wait_for(90, 'after-recovery message consumption') do
+      processed_messages.include?('after-recovery')
+    end
+
+    expect(log_output).to match(/delivery acknowledgement on channel \d+ timed out/i)
+    expect(log_output).to match(/channel recovery succeeded/i)
+    expect(processed_messages).to include('after-recovery')
+  end
+end