From d330b0d1349099fca41b6910d4028947bc6f091b Mon Sep 17 00:00:00 2001
From: Michael Klishin <michaelklishin@icloud.com>
Date: Fri, 21 Aug 2026 19:56:51 -0700
Subject: [PATCH 1/2] Consumer channel replacement fixes

 * Publishing no longer breaks after a replacement: the publisher held on
   to the old channel and exchange, now it is redeclared with them
 * Closing a channel the server has already closed is now idempotent
 * Give running handlers a graceul timeout before the work pool is killed
 * Acknowledgements go to the channel the deliveryy has arrived on
 * Handle stale delivery tags better
 * The cancellation callback captures the channel the consumer subscribed on

In addition, this ports a few eventual matchers (such as `await_condition/2`)
from RabbitMQ's CT helpers and adopts them to reduce test flakiness.

Plus fixes timeouts that are at risk of being too short.
---
 CHANGELOG.md                              |   4 +
 lib/hutch/adapters/bunny.rb               |   1 +
 lib/hutch/adapters/march_hare.rb          |   1 +
 lib/hutch/broker.rb                       |  80 ++++++++++++----
 lib/hutch/serializers/json.rb             |   2 +-
 lib/hutch/worker.rb                       |  17 ++--
 spec/hutch/broker_spec.rb                 | 110 ++++++++++++++++++++++
 spec/hutch/worker_spec.rb                 |  64 ++++++++++++-
 spec/integration/channel_recovery_spec.rb |  50 ++++++----
 spec/integration/publish_consume_spec.rb  |   3 +-
 spec/spec_helper.rb                       |   2 +
 spec/support/await_helpers.rb             |  51 ++++++++++
 12 files changed, 335 insertions(+), 50 deletions(-)
 create mode 100644 spec/support/await_helpers.rb

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 7b5e23a..fac2506 100644
--- a/CHANGELOG.md
+++ b/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 `graceful_exit_timeout` 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.
diff --git a/lib/hutch/adapters/bunny.rb b/lib/hutch/adapters/bunny.rb
index 7e81edc..808e151 100644
--- a/lib/hutch/adapters/bunny.rb
+++ b/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
diff --git a/lib/hutch/adapters/march_hare.rb b/lib/hutch/adapters/march_hare.rb
index 6fa6e48..62a0ab8 100644
--- a/lib/hutch/adapters/march_hare.rb
+++ b/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?
diff --git a/lib/hutch/broker.rb b/lib/hutch/broker.rb
index 3964131..07b7edd 100644
--- a/lib/hutch/broker.rb
+++ b/lib/hutch/broker.rb
@@ -5,6 +5,7 @@
 require 'hutch/logging'
 require 'hutch/exceptions'
 require 'hutch/publisher'
+require 'delegate'
 
 module Hutch
   class Broker
@@ -123,12 +124,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 +245,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, ch = channel)
+      ch.reject(delivery_tag, true) if current_channel?(ch)
     end
 
-    def reject(delivery_tag, requeue=false)
-      channel.reject(delivery_tag, requeue)
+    def reject(delivery_tag, requeue = false, ch = channel)
+      ch.reject(delivery_tag, requeue) if current_channel?(ch)
     end
 
-    def ack(delivery_tag)
-      channel.ack(delivery_tag, false)
+    def ack(delivery_tag, ch = channel)
+      ch.ack(delivery_tag, false) if current_channel?(ch)
     end
 
-    def nack(delivery_tag)
-      channel.nack(delivery_tag, false, false)
+    def nack(delivery_tag, ch = channel)
+      ch.nack(delivery_tag, false, false) if current_channel?(ch)
     end
 
     def publish(*args)
@@ -287,6 +286,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
 
@@ -419,4 +439,30 @@ def consumer_pool_abort_on_exception
       @config[:consumer_pool_abort_on_exception]
     end
   end
+
+  # The Broker a consumer sees while it handles one delivery. Acknowledgements
+  # go to the channel the delivery arrived on, so acknowledgement strategies
+  # do not need to know about channel replacement.
+  class ChannelBoundBroker < SimpleDelegator
+    def initialize(broker, channel)
+      super(broker)
+      @channel = channel
+    end
+
+    def ack(delivery_tag)
+      __getobj__.ack(delivery_tag, @channel)
+    end
+
+    def nack(delivery_tag)
+      __getobj__.nack(delivery_tag, @channel)
+    end
+
+    def reject(delivery_tag, requeue = false)
+      __getobj__.reject(delivery_tag, requeue, @channel)
+    end
+
+    def requeue(delivery_tag)
+      __getobj__.requeue(delivery_tag, @channel)
+    end
+  end
 end
diff --git a/lib/hutch/serializers/json.rb b/lib/hutch/serializers/json.rb
index 7d3516c..67dc01f 100644
--- a/lib/hutch/serializers/json.rb
+++ b/lib/hutch/serializers/json.rb
@@ -1,4 +1,4 @@
- require 'json'
+require 'json'
 require 'active_support/core_ext/hash/indifferent_access'
 
 module Hutch
diff --git a/lib/hutch/worker.rb b/lib/hutch/worker.rb
index d14dff9..d6ec0f5 100644
--- a/lib/hutch/worker.rb
+++ b/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
@@ -96,6 +96,7 @@ def resubscribe_on_a_new_channel(cancelled_channel)
     # Called internally when a new messages comes in from RabbitMQ. Responsible
     # for wrapping up the message and passing it to the consumer.
     def handle_message(consumer, delivery_info, properties, payload)
+      broker = ChannelBoundBroker.new(@broker, delivery_info.channel)
       serializer = consumer.get_serializer || Hutch::Config[:serializer]
       logger.debug {
         spec   = serializer.binary? ? "#{payload.bytesize} bytes" : "#{payload}"
@@ -106,11 +107,11 @@ 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 }
+      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) unless consumer_instance.message_rejected?
     rescue => ex
-      acknowledge_error(delivery_info, properties, @broker, ex)
+      acknowledge_error(delivery_info, properties, broker, ex)
       handle_error(properties, payload, consumer, ex, delivery_info)
     end
 
diff --git a/spec/hutch/broker_spec.rb b/spec/hutch/broker_spec.rb
index 3fe0cb5..27cd81a 100644
--- a/spec/hutch/broker_spec.rb
+++ b/spec/hutch/broker_spec.rb
@@ -269,6 +269,116 @@
     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', current_channel)
+
+      expect(current_channel).to receive(:nack).with('dt', false, false)
+      broker.nack('dt', current_channel)
+
+      expect(current_channel).to receive(:reject).with('dt', false)
+      broker.reject('dt', false, current_channel)
+
+      expect(current_channel).to receive(:reject).with('dt', true)
+      broker.requeue('dt', 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', superseded_channel)
+      broker.nack('dt', superseded_channel)
+      broker.reject('dt', false, superseded_channel)
+      broker.requeue('dt', superseded_channel)
+    end
+
+    it 'defaults to the current channel' do
+      expect(current_channel).to receive(:ack).with('dt', false)
+
+      broker.ack('dt')
+    end
+
+    describe Hutch::ChannelBoundBroker do
+      it 'pins acknowledgements to the channel it was built with' do
+        bound = Hutch::ChannelBoundBroker.new(broker, current_channel)
+
+        expect(current_channel).to receive(:ack).with('dt', false)
+        bound.ack('dt')
+      end
+
+      it 'drops acknowledgements once its channel is replaced' do
+        bound = Hutch::ChannelBoundBroker.new(broker, superseded_channel)
+
+        expect(superseded_channel).not_to receive(:nack)
+        bound.nack('dt')
+      end
+    end
+  end
+
   describe '#declare_exchange' do
     before do
       broker.open_connection!
diff --git a/spec/hutch/worker_spec.rb b/spec/hutch/worker_spec.rb
index f753bb5..75012d3 100644
--- a/spec/hutch/worker_spec.rb
+++ b/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, 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, delivery_info.channel)
         subject
       end
 
@@ -198,7 +252,7 @@ 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, delivery_info.channel)
         subject
       end
     end
@@ -206,7 +260,7 @@ def handle(_properties, _payload, _consumer, _ex)
 
 
   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') }
 
diff --git a/spec/integration/channel_recovery_spec.rb b/spec/integration/channel_recovery_spec.rb
index 1131145..60a3748 100644
--- a/spec/integration/channel_recovery_spec.rb
+++ b/spec/integration/channel_recovery_spec.rb
@@ -8,7 +8,6 @@
 require 'bunny'
 require 'json'
 require 'securerandom'
-require 'timeout'
 
 describe 'channel recovery after delivery acknowledgement timeout', rabbitmq: true, adapter: :bunny do
   CONSUMER_TIMEOUT_MS = 5_000
@@ -92,25 +91,25 @@
     Hutch::Config.set(:mq_exchange, exchange_name)
     Hutch::Config.set(:force_publisher_confirms, false)
     Hutch::Config.set(:client_logger, logger)
+    # Channel replacement waits this long for the deliberately blocked
+    # handler: the default of 11s would use up most of the consumption window.
+    @graceful_exit_timeout = Hutch::Config[:graceful_exit_timeout]
+    Hutch::Config.set(:graceful_exit_timeout, 1)
   end
 
   after do
     publisher_channel.close rescue nil
     publisher.close rescue nil
     broker.disconnect rescue nil
+    Hutch::Config.set(:graceful_exit_timeout, @graceful_exit_timeout)
     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
+  def wait_for(timeout, label, &condition)
+    await_condition(timeout, label, &condition)
+  rescue AwaitHelpers::ConditionTimeout => e
     raise <<~MSG
-      Timed out waiting for: #{label}
+      #{e.message}
 
       processed_messages=#{processed_messages.inspect}
       channel_open=#{broker.channel.open? rescue 'unknown'}
@@ -125,9 +124,10 @@ def processed_messages
     processed_lock.synchronize { processed.dup }
   end
 
+  # `StringIO#string` does not touch the stream position: a rewind here would
+  # race with the worker threads that log concurrently and corrupt the buffer.
   def log_output
-    log.rewind
-    log.read
+    log.string
   end
 
   def publish_message(id)
@@ -141,10 +141,21 @@ def publish_message(id)
 
   # RabbitMQ up to 4.2 closes the channel on a delivery acknowledgement
   # timeout; 4.3+ quorum queues instead cancel only the timed out consumer.
+  let(:cancels_the_consumer) { server_version >= Gem::Version.new('4.3') }
+
   let(:recovery_log_pattern) do
-    /delivery acknowledgement on channel \d+ timed out|cancelled by the server, re-subscribing/i
+    if cancels_the_consumer
+      /cancelled by the server, re-subscribing/i
+    else
+      /delivery acknowledgement on channel \d+ timed out/i
+    end
   end
 
+  # Replacing the channel requeues what the cancelled consumer still held, so
+  # consumption resumes at once. Reopening one does not, so it resumes at the
+  # next consumer timeout tick.
+  let(:recovery_seconds) { cancels_the_consumer ? 15 : 90 }
+
   # This spec is intentionally slow because RabbitMQ enforces delivery
   # acknowledgement timeouts periodically on a timer, not immediately at the deadline.
   it 're-subscribes and consumes later messages after RabbitMQ times out an unacknowledged delivery' do
@@ -159,13 +170,18 @@ def publish_message(id)
 
     publish_message('after-recovery')
 
-    # Replacing the channel requeues what the cancelled consumer still held, so
-    # consumption resumes at once rather than at the next consumer timeout tick.
-    wait_for(15, 'after-recovery message consumption') do
+    wait_for(recovery_seconds, 'after-recovery message consumption') do
       processed_messages.include?('after-recovery')
     end
 
     expect(log_output).to match(recovery_log_pattern)
-    expect(processed_messages).to include('after-recovery')
+
+    # A consumed message proves the recovery finished, the publisher rebuild
+    # included.
+    broker.publish(routing_key, 'id' => 'published-after-recovery')
+
+    wait_for(15, 'published-after-recovery message consumption') do
+      processed_messages.include?('published-after-recovery')
+    end
   end
 end
diff --git a/spec/integration/publish_consume_spec.rb b/spec/integration/publish_consume_spec.rb
index cb7d71a..1f1db5e 100644
--- a/spec/integration/publish_consume_spec.rb
+++ b/spec/integration/publish_consume_spec.rb
@@ -3,7 +3,6 @@
 require 'hutch/worker'
 require 'hutch/consumer'
 require 'securerandom'
-require 'timeout'
 
 describe 'publishing and consuming messages', rabbitmq: true, adapter: :bunny do
   let(:exchange_name) { "hutch.test.#{SecureRandom.hex(4)}" }
@@ -41,7 +40,7 @@
 
     broker.publish(routing_key, { test: 'data' })
 
-    Timeout.timeout(5) { sleep 0.1 until received.any? }
+    await_condition(15, 'message consumption') { received.any? }
 
     expect(received.first).to eq('test' => 'data')
   end
diff --git a/spec/spec_helper.rb b/spec/spec_helper.rb
index 8215513..9dd949e 100644
--- a/spec/spec_helper.rb
+++ b/spec/spec_helper.rb
@@ -7,12 +7,14 @@
 
 require 'hutch'
 require 'logger'
+require_relative 'support/await_helpers'
 
 # set logger to be a null logger
 Hutch::Logging.logger = Logger.new(File::NULL)
 
 RSpec.configure do |config|
   config.raise_errors_for_deprecations!
+  config.include AwaitHelpers
 
   if defined?(JRUBY_VERSION)
     config.filter_run_excluding adapter: :bunny
diff --git a/spec/support/await_helpers.rb b/spec/support/await_helpers.rb
new file mode 100644
index 0000000..c60decf
--- /dev/null
+++ b/spec/support/await_helpers.rb
@@ -0,0 +1,51 @@
+# Ruby ports of `await_condition/2` and friends from `rabbit_ct_helpers`.
+module AwaitHelpers
+  class ConditionTimeout < StandardError; end
+
+  POLL_INTERVAL = 0.05
+
+  def await_condition(timeout = 15, label = 'condition')
+    deadline = monotonic_now + timeout
+
+    until yield
+      raise ConditionTimeout, "#{label} did not materialize in #{timeout}s" if monotonic_now >= deadline
+
+      sleep POLL_INTERVAL
+    end
+  end
+
+  def await_condition_ignoring_exceptions(timeout = 15, label = 'condition', &condition)
+    await_condition(timeout, label) do
+      begin
+        condition.call
+      rescue StandardError
+        false
+      end
+    end
+  end
+
+  # Retries a failed expectation, e.g. eventually { expect(queue.message_count).to eq(3) }
+  def eventually(poll_interval = 0.2, poll_count = 5)
+    yield
+  rescue RSpec::Expectations::ExpectationNotMetError, StandardError
+    poll_count -= 1
+    raise if poll_count.zero?
+
+    sleep poll_interval
+    retry
+  end
+
+  # Asserts that an expectation keeps holding, e.g. consistently { expect(queue.message_count).to eq(0) }
+  def consistently(poll_interval = 0.2, poll_count = 5)
+    poll_count.times do
+      yield
+      sleep poll_interval
+    end
+  end
+
+  private
+
+  def monotonic_now
+    Process.clock_gettime(Process::CLOCK_MONOTONIC)
+  end
+end

From 448c46190d21ae20fbc8b50e6236d5b2380b09b1 Mon Sep 17 00:00:00 2001
From: Michael Klishin <michaelklishin@icloud.com>
Date: Fri, 21 Aug 2026 20:06:19 -0700
Subject: [PATCH 2/2] Get rid of delegation

It's arguably uglier than this alternative and
is known not to be type checking-friendly.
---
 CHANGELOG.md                                  |  2 +-
 .../acknowledgements/nack_on_all_failures.rb  |  2 +-
 lib/hutch/broker.rb                           | 43 ++++---------------
 lib/hutch/consumer.rb                         |  4 +-
 lib/hutch/worker.rb                           |  7 ++-
 spec/hutch/broker_spec.rb                     | 32 ++++----------
 spec/hutch/worker_spec.rb                     |  6 +--
 7 files changed, 26 insertions(+), 70 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index fac2506..bb1507f 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -13,7 +13,7 @@ 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 `graceful_exit_timeout` to finish before
+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.
 
diff --git a/lib/hutch/acknowledgements/nack_on_all_failures.rb b/lib/hutch/acknowledgements/nack_on_all_failures.rb
index 274fa39..0313fb4 100644
--- a/lib/hutch/acknowledgements/nack_on_all_failures.rb
+++ b/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
diff --git a/lib/hutch/broker.rb b/lib/hutch/broker.rb
index 07b7edd..50834dc 100644
--- a/lib/hutch/broker.rb
+++ b/lib/hutch/broker.rb
@@ -5,7 +5,6 @@
 require 'hutch/logging'
 require 'hutch/exceptions'
 require 'hutch/publisher'
-require 'delegate'
 
 module Hutch
   class Broker
@@ -251,20 +250,20 @@ def stop
 
     # 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, ch = channel)
-      ch.reject(delivery_tag, true) if current_channel?(ch)
+    def requeue(delivery_tag, channel: self.channel)
+      channel.reject(delivery_tag, true) if current_channel?(channel)
     end
 
-    def reject(delivery_tag, requeue = false, ch = channel)
-      ch.reject(delivery_tag, requeue) if current_channel?(ch)
+    def reject(delivery_tag, requeue = false, channel: self.channel)
+      channel.reject(delivery_tag, requeue) if current_channel?(channel)
     end
 
-    def ack(delivery_tag, ch = channel)
-      ch.ack(delivery_tag, false) if current_channel?(ch)
+    def ack(delivery_tag, channel: self.channel)
+      channel.ack(delivery_tag, false) if current_channel?(channel)
     end
 
-    def nack(delivery_tag, ch = channel)
-      ch.nack(delivery_tag, false, false) if current_channel?(ch)
+    def nack(delivery_tag, channel: self.channel)
+      channel.nack(delivery_tag, false, false) if current_channel?(channel)
     end
 
     def publish(*args)
@@ -439,30 +438,4 @@ def consumer_pool_abort_on_exception
       @config[:consumer_pool_abort_on_exception]
     end
   end
-
-  # The Broker a consumer sees while it handles one delivery. Acknowledgements
-  # go to the channel the delivery arrived on, so acknowledgement strategies
-  # do not need to know about channel replacement.
-  class ChannelBoundBroker < SimpleDelegator
-    def initialize(broker, channel)
-      super(broker)
-      @channel = channel
-    end
-
-    def ack(delivery_tag)
-      __getobj__.ack(delivery_tag, @channel)
-    end
-
-    def nack(delivery_tag)
-      __getobj__.nack(delivery_tag, @channel)
-    end
-
-    def reject(delivery_tag, requeue = false)
-      __getobj__.reject(delivery_tag, requeue, @channel)
-    end
-
-    def requeue(delivery_tag)
-      __getobj__.requeue(delivery_tag, @channel)
-    end
-  end
 end
diff --git a/lib/hutch/consumer.rb b/lib/hutch/consumer.rb
index b4c48ab..4909a3e 100644
--- a/lib/hutch/consumer.rb
+++ b/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?
diff --git a/lib/hutch/worker.rb b/lib/hutch/worker.rb
index d6ec0f5..40a4445 100644
--- a/lib/hutch/worker.rb
+++ b/lib/hutch/worker.rb
@@ -96,7 +96,6 @@ def resubscribe_on_a_new_channel(cancelled_channel)
     # Called internally when a new messages comes in from RabbitMQ. Responsible
     # for wrapping up the message and passing it to the consumer.
     def handle_message(consumer, delivery_info, properties, payload)
-      broker = ChannelBoundBroker.new(@broker, delivery_info.channel)
       serializer = consumer.get_serializer || Hutch::Config[:serializer]
       logger.debug {
         spec   = serializer.binary? ? "#{payload.bytesize} bytes" : "#{payload}"
@@ -107,11 +106,11 @@ 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 }
+      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)
+      acknowledge_error(delivery_info, properties, @broker, ex)
       handle_error(properties, payload, consumer, ex, delivery_info)
     end
 
diff --git a/spec/hutch/broker_spec.rb b/spec/hutch/broker_spec.rb
index 27cd81a..e416c48 100644
--- a/spec/hutch/broker_spec.rb
+++ b/spec/hutch/broker_spec.rb
@@ -333,16 +333,16 @@
 
     it 'sends them on the channel the delivery arrived on' do
       expect(current_channel).to receive(:ack).with('dt', false)
-      broker.ack('dt', current_channel)
+      broker.ack('dt', channel: current_channel)
 
       expect(current_channel).to receive(:nack).with('dt', false, false)
-      broker.nack('dt', current_channel)
+      broker.nack('dt', channel: current_channel)
 
       expect(current_channel).to receive(:reject).with('dt', false)
-      broker.reject('dt', false, current_channel)
+      broker.reject('dt', channel: current_channel)
 
       expect(current_channel).to receive(:reject).with('dt', true)
-      broker.requeue('dt', current_channel)
+      broker.requeue('dt', channel: current_channel)
     end
 
     it 'drops them when the channel has been replaced' do
@@ -350,10 +350,10 @@
       expect(superseded_channel).not_to receive(:nack)
       expect(superseded_channel).not_to receive(:reject)
 
-      broker.ack('dt', superseded_channel)
-      broker.nack('dt', superseded_channel)
-      broker.reject('dt', false, superseded_channel)
-      broker.requeue('dt', superseded_channel)
+      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
@@ -361,22 +361,6 @@
 
       broker.ack('dt')
     end
-
-    describe Hutch::ChannelBoundBroker do
-      it 'pins acknowledgements to the channel it was built with' do
-        bound = Hutch::ChannelBoundBroker.new(broker, current_channel)
-
-        expect(current_channel).to receive(:ack).with('dt', false)
-        bound.ack('dt')
-      end
-
-      it 'drops acknowledgements once its channel is replaced' do
-        bound = Hutch::ChannelBoundBroker.new(broker, superseded_channel)
-
-        expect(superseded_channel).not_to receive(:nack)
-        bound.nack('dt')
-      end
-    end
   end
 
   describe '#declare_exchange' do
diff --git a/spec/hutch/worker_spec.rb b/spec/hutch/worker_spec.rb
index 75012d3..4ca27be 100644
--- a/spec/hutch/worker_spec.rb
+++ b/spec/hutch/worker_spec.rb
@@ -162,7 +162,7 @@
 
     it 'acknowledges the message' do
       allow(consumer_instance).to receive(:process)
-      expect(broker).to receive(:ack).with(delivery_info.delivery_tag, delivery_info.channel)
+      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
@@ -195,7 +195,7 @@
       end
 
       it 'rejects the message' do
-        expect(broker).to receive(:nack).with(delivery_info.delivery_tag, delivery_info.channel)
+        expect(broker).to receive(:nack).with(delivery_info.delivery_tag, channel: delivery_info.channel)
         subject
       end
 
@@ -252,7 +252,7 @@ def handle(_properties, _payload, _consumer, _ex)
       end
 
       it 'rejects the message' do
-        expect(broker).to receive(:nack).with(delivery_info.delivery_tag, delivery_info.channel)
+        expect(broker).to receive(:nack).with(delivery_info.delivery_tag, channel: delivery_info.channel)
         subject
       end
     end