Consumer channel replacement fixes

CHANGELOG.md

@@ -13,6 +13,10 @@ timeout is indistinguishable from a queue deletion. The consumer channel is
 replaced in the process, so the deliveries the cancelled consumer still held
 are requeued instead of dropped.
 
+Handlers that are still running get a graceful exit to finish before
+the old channel is closed, and their acknowledgements are dropped rather than
+sent on the new channel, where their delivery tags mean nothing.
+
 ### Hutch Now Requires Ruby 3.2 or JRuby 10
 
 Ruby 3.0 and 3.1 have reached EOL.

lib/hutch/acknowledgements/nack_on_all_failures.rb

@@ -10,7 +10,7 @@ def handle(delivery_info, properties, broker, ex)
         prefix = "message(#{properties.message_id || '-'}): "
         logger.debug "#{prefix} nacking message"
 
-        broker.nack(delivery_info.delivery_tag)
+        broker.nack(delivery_info.delivery_tag, channel: delivery_info.channel)
 
         true
       end

lib/hutch/adapters/bunny.rb

@@ -12,6 +12,7 @@ class BunnyAdapter
       DEFAULT_VHOST = Bunny::Session::DEFAULT_VHOST
 
       ConnectionRefused = Bunny::TCPConnectionFailed
+      ChannelAlreadyClosed = Bunny::ChannelAlreadyClosed
       PreconditionFailed = Bunny::PreconditionFailed
 
       def_delegators :@connection, :start, :disconnect, :close, :create_channel, :open?, :recover_channel_topology

lib/hutch/adapters/march_hare.rb

@@ -9,6 +9,7 @@ class MarchHareAdapter
       DEFAULT_VHOST = "/"
 
       ConnectionRefused = MarchHare::ConnectionRefused
+      ChannelAlreadyClosed = MarchHare::ChannelAlreadyClosed
       PreconditionFailed = MarchHare::PreconditionFailed
 
       def_delegators :@connection, :start, :disconnect, :close, :open?

lib/hutch/broker.rb

@@ -123,12 +123,13 @@ def open_channel!
       @channel = open_channel
     end
 
-    # Replaces the consumer channel. Closing the old one makes the broker forget
-    # the consumers registered on it and requeue the deliveries they still hold.
+    # Closing the old channel makes RabbitMQ forget the consumers on it
+    # and requeue their unacknowledged deliveries.
     def replace_channel!
-      @channel.close if @channel
+      close_consumer_channel
       open_channel!
       declare_exchange!
+      declare_publisher!
     end
 
     def declare_exchange(ch = channel)
@@ -243,29 +244,26 @@ def stop
       if defined?(JRUBY_VERSION)
         channel.close
       else
-        # Enqueue a failing job that kills the consumer loop
-        channel_work_pool.shutdown
-        # Give `timeout` seconds to jobs that are still being processed
-        channel_work_pool.join(@config[:graceful_exit_timeout])
-        # If after `timeout` they are still running, they are killed
-        channel_work_pool.kill
+        drain_consumer_work_pool
       end
     end
 
-    def requeue(delivery_tag)
-      channel.reject(delivery_tag, true)
+    # Delivery tags are scoped to their channel: a tag from a replaced one
+    # is dropped, the server has requeued that delivery anyway.
+    def requeue(delivery_tag, channel: self.channel)
+      channel.reject(delivery_tag, true) if current_channel?(channel)
     end
 
-    def reject(delivery_tag, requeue=false)
-      channel.reject(delivery_tag, requeue)
+    def reject(delivery_tag, requeue = false, channel: self.channel)
+      channel.reject(delivery_tag, requeue) if current_channel?(channel)
     end
 
-    def ack(delivery_tag)
-      channel.ack(delivery_tag, false)
+    def ack(delivery_tag, channel: self.channel)
+      channel.ack(delivery_tag, false) if current_channel?(channel)
     end
 
-    def nack(delivery_tag)
-      channel.nack(delivery_tag, false, false)
+    def nack(delivery_tag, channel: self.channel)
+      channel.nack(delivery_tag, false, false) if current_channel?(channel)
     end
 
     def publish(*args)
@@ -287,6 +285,27 @@ def using_publisher_confirmations?
 
     private
 
+    def current_channel?(ch)
+      ch.equal?(channel)
+    end
+
+    def close_consumer_channel
+      return unless @channel && @channel.open?
+
+      drain_consumer_work_pool unless defined?(JRUBY_VERSION)
+      @channel.close
+    rescue Hutch::Adapter::ChannelAlreadyClosed
+      # The server closed the channel first: same outcome.
+    end
+
+    # Bunny kills the consumer work pool when the channel closes. Draining
+    # first gives the running handlers `graceful_exit_timeout` to finish.
+    def drain_consumer_work_pool
+      channel_work_pool.shutdown
+      channel_work_pool.join(@config[:graceful_exit_timeout])
+      channel_work_pool.kill
+    end
+
     Config = Struct.new(:host, :port, :username, :password, :ssl, :protocol, :sanitized_uri)
     private_constant :Config
 

lib/hutch/consumer.rb

@@ -14,12 +14,12 @@ def self.included(base)
 
     def reject!
       @message_rejected = true
-      broker.reject(delivery_info.delivery_tag)
+      broker.reject(delivery_info.delivery_tag, channel: delivery_info.channel)
     end
 
     def requeue!
       @message_rejected = true
-      broker.requeue(delivery_info.delivery_tag)
+      broker.requeue(delivery_info.delivery_tag, channel: delivery_info.channel)
     end
 
     def message_rejected?

lib/hutch/serializers/json.rb

@@ -1,4 +1,4 @@
- require 'json'
+require 'json'
 require 'active_support/core_ext/hash/indifferent_access'
 
 module Hutch

lib/hutch/worker.rb

@@ -53,7 +53,7 @@ def setup_queue(consumer)
       queue = @broker.queue(queue_name, consumer.get_options)
       @broker.bind_queue(queue, consumer.routing_keys)
 
-      on_cancellation = proc { handle_cancellation(consumer, queue_name) }
+      on_cancellation = proc { handle_cancellation(queue_name, queue.channel) }
       queue.subscribe(consumer_tag: unique_consumer_tag, manual_ack: true,
                       on_cancellation: on_cancellation) do |*args|
         delivery_info, properties, payload = Hutch::Adapter.decode_message(*args)
@@ -63,20 +63,20 @@ def setup_queue(consumer)
 
     # A server-sent `basic.cancel` carries no reason, so a consumer timeout
     # is indistinguishable from a queue deletion. Only the former is recoverable.
-    def handle_cancellation(consumer, queue_name)
+    def handle_cancellation(queue_name, cancelled_channel)
       unless @broker.queue_exists?(queue_name)
         logger.error "consumer on queue #{queue_name} was cancelled by the server: the queue no longer exists"
         return
       end
 
       logger.warn "consumer on queue #{queue_name} was cancelled by the server, re-subscribing"
-      resubscribe_on_a_new_channel(@broker.channel)
+      resubscribe_on_a_new_channel(cancelled_channel)
     rescue => ex
       logger.error "consumer re-subscription failed: #{ex.class}: #{ex.message}"
     end
 
-    # Runs on its own thread: closing the channel shuts down the consumer work
-    # pool this callback is dispatched on.
+    # Runs on its own thread: closing the channel kills the consumer work
+    # pool this callback runs on. Returns the thread so that tests can join it.
     def resubscribe_on_a_new_channel(cancelled_channel)
       Thread.new do
         begin
@@ -108,7 +108,7 @@ def handle_message(consumer, delivery_info, properties, payload)
       message = Message.new(delivery_info, properties, payload, serializer)
       consumer_instance = consumer.new.tap { |c| c.broker, c.delivery_info = @broker, delivery_info }
       with_tracing(consumer_instance).handle(message)
-      @broker.ack(delivery_info.delivery_tag) unless consumer_instance.message_rejected?
+      @broker.ack(delivery_info.delivery_tag, channel: delivery_info.channel) unless consumer_instance.message_rejected?
     rescue => ex
       acknowledge_error(delivery_info, properties, @broker, ex)
       handle_error(properties, payload, consumer, ex, delivery_info)

spec/hutch/broker_spec.rb

@@ -269,6 +269,100 @@
     end
   end
 
+  describe '#replace_channel!', adapter: :bunny do
+    let(:old_channel) { double('Bunny::Channel', open?: true, close: nil) }
+    let(:new_channel) { double('Bunny::Channel').as_null_object }
+    let(:work_pool) { double('Bunny::ConsumerWorkPool', shutdown: nil, join: nil, kill: nil) }
+
+    before do
+      broker.channel = old_channel
+      allow(broker).to receive(:channel_work_pool).and_return(work_pool)
+      allow(broker).to receive(:open_channel).and_return(new_channel)
+      allow(broker).to receive(:declare_exchange!)
+      allow(broker).to receive(:declare_publisher!)
+    end
+
+    it 'drains the work pool before closing the old channel' do
+      expect(work_pool).to receive(:shutdown).ordered
+      expect(work_pool).to receive(:join).with(config[:graceful_exit_timeout]).ordered
+      expect(work_pool).to receive(:kill).ordered
+      expect(old_channel).to receive(:close).ordered
+
+      broker.replace_channel!
+    end
+
+    it 'redeclares the exchange and the publisher on the new channel' do
+      expect(broker).to receive(:declare_exchange!)
+      expect(broker).to receive(:declare_publisher!)
+
+      broker.replace_channel!
+
+      expect(broker.channel).to eq(new_channel)
+    end
+
+    context 'when the old channel is already closed' do
+      let(:old_channel) { double('Bunny::Channel', open?: false) }
+
+      it 'opens a new channel anyway' do
+        expect(old_channel).not_to receive(:close)
+        expect(work_pool).not_to receive(:shutdown)
+
+        broker.replace_channel!
+
+        expect(broker.channel).to eq(new_channel)
+      end
+    end
+
+    context 'when the server closes the old channel concurrently' do
+      it 'opens a new channel anyway' do
+        allow(old_channel).to receive(:close).
+          and_raise(Hutch::Adapter::ChannelAlreadyClosed.new('already closed', old_channel))
+
+        broker.replace_channel!
+
+        expect(broker.channel).to eq(new_channel)
+      end
+    end
+  end
+
+  describe 'acknowledgements' do
+    let(:current_channel) { double('Channel').as_null_object }
+    let(:superseded_channel) { double('Channel') }
+
+    before { broker.channel = current_channel }
+
+    it 'sends them on the channel the delivery arrived on' do
+      expect(current_channel).to receive(:ack).with('dt', false)
+      broker.ack('dt', channel: current_channel)
+
+      expect(current_channel).to receive(:nack).with('dt', false, false)
+      broker.nack('dt', channel: current_channel)
+
+      expect(current_channel).to receive(:reject).with('dt', false)
+      broker.reject('dt', channel: current_channel)
+
+      expect(current_channel).to receive(:reject).with('dt', true)
+      broker.requeue('dt', channel: current_channel)
+    end
+
+    it 'drops them when the channel has been replaced' do
+      expect(superseded_channel).not_to receive(:ack)
+      expect(superseded_channel).not_to receive(:nack)
+      expect(superseded_channel).not_to receive(:reject)
+
+      broker.ack('dt', channel: superseded_channel)
+      broker.nack('dt', channel: superseded_channel)
+      broker.reject('dt', channel: superseded_channel)
+      broker.requeue('dt', channel: superseded_channel)
+    end
+
+    it 'defaults to the current channel' do
+      expect(current_channel).to receive(:ack).with('dt', false)
+
+      broker.ack('dt')
+    end
+  end
+
   describe '#declare_exchange' do
     before do
       broker.open_connection!

spec/hutch/worker_spec.rb

@@ -84,11 +84,65 @@
     end
   end
 
+  describe '#handle_cancellation' do
+    let(:cancelled_channel) { double('Channel') }
+    let(:new_channel) { double('Channel') }
+    let(:log) { StringIO.new }
+
+    before do
+      allow(Hutch::Logging).to receive(:logger).and_return(Logger.new(log))
+      allow(broker).to receive(:channel).and_return(cancelled_channel)
+      allow(broker).to receive(:replace_channel!) do
+        allow(broker).to receive(:channel).and_return(new_channel)
+      end
+      allow(worker).to receive(:setup_queues)
+    end
+
+    context 'when the queue still exists' do
+      before { allow(broker).to receive(:queue_exists?).with('consumer').and_return(true) }
+
+      it 'replaces the channel and re-subscribes' do
+        expect(broker).to receive(:replace_channel!)
+        expect(worker).to receive(:setup_queues)
+
+        worker.handle_cancellation('consumer', cancelled_channel).join
+      end
+
+      it 'replaces the channel once for all the consumers that shared it' do
+        expect(broker).to receive(:replace_channel!).once
+
+        [
+          worker.handle_cancellation('consumer', cancelled_channel),
+          worker.handle_cancellation('consumer', cancelled_channel)
+        ].each(&:join)
+      end
+
+      it 'does not replace a channel that has already been replaced' do
+        expect(broker).not_to receive(:replace_channel!)
+
+        worker.handle_cancellation('consumer', double('An older channel')).join
+      end
+    end
+
+    context 'when the queue is gone' do
+      before { allow(broker).to receive(:queue_exists?).with('consumer').and_return(false) }
+
+      it 'does not re-subscribe' do
+        expect(broker).not_to receive(:replace_channel!)
+
+        worker.handle_cancellation('consumer', cancelled_channel)
+
+        log.rewind
+        expect(log.read).to match(/the queue no longer exists/)
+      end
+    end
+  end
+
   describe '#handle_message' do
     subject { worker.handle_message(consumer, delivery_info, properties, payload) }
     let(:payload) { '{}' }
     let(:consumer_instance) { double('Consumer instance') }
-    let(:delivery_info) { double('Delivery Info', routing_key: '',
+    let(:delivery_info) { double('Delivery Info', routing_key: '', channel: nil,
                                  delivery_tag: 'dt') }
     let(:properties) { double('Properties', message_id: nil, content_type: "application/json") }
     let(:log) { StringIO.new }
@@ -108,7 +162,7 @@
 
     it 'acknowledges the message' do
       allow(consumer_instance).to receive(:process)
-      expect(broker).to receive(:ack).with(delivery_info.delivery_tag)
+      expect(broker).to receive(:ack).with(delivery_info.delivery_tag, channel: delivery_info.channel)
       expect(consumer_instance).to receive(:message_rejected?).and_return(false)
       subject
     end
@@ -141,7 +195,7 @@
       end
 
       it 'rejects the message' do
-        expect(broker).to receive(:nack).with(delivery_info.delivery_tag)
+        expect(broker).to receive(:nack).with(delivery_info.delivery_tag, channel: delivery_info.channel)
         subject
       end
 
@@ -198,15 +252,15 @@ def handle(_properties, _payload, _consumer, _ex)
       end
 
       it 'rejects the message' do
-        expect(broker).to receive(:nack).with(delivery_info.delivery_tag)
+        expect(broker).to receive(:nack).with(delivery_info.delivery_tag, channel: delivery_info.channel)
         subject
       end
     end
   end
 
 
   describe '#acknowledge_error' do
-    let(:delivery_info) { double('Delivery Info', routing_key: '',
+    let(:delivery_info) { double('Delivery Info', routing_key: '', channel: nil,
                                  delivery_tag: 'dt') }
     let(:properties) { double('Properties', message_id: 'abc123') }