From d08277bab6b73f82ab81ab2ce95c52f7aaa018f8 Mon Sep 17 00:00:00 2001 From: John Bolliger Date: Thu, 6 Aug 2026 10:11:05 -0600 Subject: [PATCH 01/15] Add Appraisal matrix and split CI by Rails version Introduce an Appraisal matrix covering Rails 6.1 through 8.1, with the default-gem shims (logger, mutex_m, bigdecimal, drb, base64, benchmark) that ActiveSupport < 7.1 needs on Ruby >= 3.4. CircleCI now runs build_and_test as a parameterized job across the Ruby and JRuby images and each appraisal gemfile. Rails 8.0/8.1 require Ruby >= 3.2, so they get their own matrix (cimg/ruby:3.4, jruby:10.0) rather than exclude entries against the Ruby 3.1-class images. Coverage is unchanged at 20 jobs. Also add a spec helper that waits for RabbitMQ before the integration suite starts, so a cold broker fails with one clear message instead of a flurry of reconnect warnings. Co-Authored-By: Claude Opus 5 (1M context) --- .circleci/config.yml | 57 ++++++++++++++++++++++++++++---------- .gitignore | 4 +++ Appraisals | 51 ++++++++++++++++++++++++++++++++++ Rakefile | 9 ++++++ action_subscriber.gemspec | 1 + gemfiles/rails_6.1.gemfile | 14 ++++++++++ gemfiles/rails_7.0.gemfile | 14 ++++++++++ gemfiles/rails_7.1.gemfile | 8 ++++++ gemfiles/rails_7.2.gemfile | 8 ++++++ gemfiles/rails_8.0.gemfile | 8 ++++++ gemfiles/rails_8.1.gemfile | 8 ++++++ spec/spec_helper.rb | 7 +++++ spec/support/rabbitmq.rb | 52 ++++++++++++++++++++++++++++++++++ 13 files changed, 227 insertions(+), 14 deletions(-) create mode 100644 Appraisals create mode 100644 gemfiles/rails_6.1.gemfile create mode 100644 gemfiles/rails_7.0.gemfile create mode 100644 gemfiles/rails_7.1.gemfile create mode 100644 gemfiles/rails_7.2.gemfile create mode 100644 gemfiles/rails_8.0.gemfile create mode 100644 gemfiles/rails_8.1.gemfile create mode 100644 spec/support/rabbitmq.rb diff --git a/.circleci/config.yml b/.circleci/config.yml index 24ef51e..a232b12 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -6,6 +6,9 @@ jobs: docker_image: type: string description: "The Ruby or JRuby Docker image to test against" + gemfile: + type: string + description: "The Appraisal gemfile (Rails version) to test against" docker: # 1. The Primary Container (where your code actually runs) @@ -13,15 +16,19 @@ jobs: environment: JRUBY_OPTS: "-J-Xmx1024m" RAILS_ENV: test - # Tell your app where to find RabbitMQ (if your app uses this ENV var) + # Select the Rails version under test via the Appraisal gemfile. + BUNDLE_GEMFILE: << parameters.gemfile >> + # Tell the suite where to find RabbitMQ. RABBITMQ_URL: "amqp://guest:guest@localhost:5672" + RABBITMQ_HOST: "localhost" + RABBITMQ_PORT: "5672" - # 2. The Service Container (runs in the background) + # 2. The Service Container (runs in the background). + # NOTE: action_subscriber declares transient (non-durable) queues by default, + # which RabbitMQ 4.x denies out of the box. rabbitmq:3.12 still permits them. + # If/when moving to a 4.x image, permit the deprecated feature via config: + # deprecated_features.permit.transient_nonexcl_queues = true - image: rabbitmq:3.12-management - # If you need the management UI for debugging, use `rabbitmq:3-management` instead - # environment: - # RABBITMQ_DEFAULT_USER: guest - # RABBITMQ_DEFAULT_PASS: guest working_directory: ~/project @@ -35,12 +42,13 @@ jobs: sudo apt-get update && sudo apt-get install -y build-essential git fi - checkout - # Note: We added the docker_image parameter to the cache key - # so MRI and JRuby gems don't conflict. + # Cache key includes the Ruby image + the specific appraisal gemfile + the gemspec, + # so MRI/JRuby and each Rails version get independent caches. (Gemfile.lock and the + # generated gemfiles/*.lock are gitignored, so we key on committed files instead.) - restore_cache: keys: - - v1-gems-<< parameters.docker_image >>-{{ checksum "Gemfile.lock" }} - - v1-gems-<< parameters.docker_image >>- + - v2-gems-<< parameters.docker_image >>-<< parameters.gemfile >>-{{ checksum "action_subscriber.gemspec" }} + - v2-gems-<< parameters.docker_image >>-<< parameters.gemfile >>- - run: name: Install Ruby Dependencies @@ -52,7 +60,7 @@ jobs: - save_cache: paths: - ./vendor/bundle - key: v1-gems-<< parameters.docker_image >>-{{ checksum "Gemfile.lock" }} + key: v2-gems-<< parameters.docker_image >>-<< parameters.gemfile >>-{{ checksum "action_subscriber.gemspec" }} # Wait for RabbitMQ to be ready before running tests. # Service containers can sometimes take a few seconds to boot up. @@ -77,14 +85,35 @@ jobs: workflows: version: 2 - ruby_compatibility_matrix: + ruby_rails_compatibility_matrix: jobs: + # Rails 6.1 - 7.2 run on every supported Ruby. Rails 6.1/7.0/7.1 need the + # default-gem shims on Ruby >= 3.4 (see Appraisals); 7.2 needs Ruby >= 3.1. - build_and_test: - name: test-<< matrix.docker_image >> + name: test-<< matrix.docker_image >>-<< matrix.gemfile >> matrix: parameters: docker_image: - "cimg/ruby:3.1" - "cimg/ruby:3.4" - "jruby:9.4" - - "jruby:10.0" \ No newline at end of file + - "jruby:10.0" + gemfile: + - "gemfiles/rails_6.1.gemfile" + - "gemfiles/rails_7.0.gemfile" + - "gemfiles/rails_7.1.gemfile" + - "gemfiles/rails_7.2.gemfile" + + # Rails 8.0/8.1 require Ruby >= 3.2, so they get their own matrix rather + # than excludes against the Ruby 3.1-class images. jruby:9.4 targets Ruby + # 3.1 compatibility and is therefore not eligible here either. + - build_and_test: + name: test-<< matrix.docker_image >>-<< matrix.gemfile >> + matrix: + parameters: + docker_image: + - "cimg/ruby:3.4" + - "jruby:10.0" + gemfile: + - "gemfiles/rails_8.0.gemfile" + - "gemfiles/rails_8.1.gemfile" diff --git a/.gitignore b/.gitignore index 88db647..776bb87 100644 --- a/.gitignore +++ b/.gitignore @@ -15,6 +15,10 @@ tmp Gemfile.lock +# Appraisal-generated lock files (the .gemfile stubs are committed, the locks are not) +gemfiles/*.gemfile.lock +gemfiles/.bundle + # YARD artifacts .yardoc _yardoc diff --git a/Appraisals b/Appraisals new file mode 100644 index 0000000..a9f4123 --- /dev/null +++ b/Appraisals @@ -0,0 +1,51 @@ +# Appraisal matrix for the Rails/ActiveSupport versions action_subscriber supports. +# +# Notes on Ruby compatibility when running this matrix in CI: +# * rails 6.1 / 7.0 / 7.1 run on Ruby >= 2.7 (and on Ruby 3.4 with the +# default-gem shims added below, since logger/mutex_m/bigdecimal/drb/base64 +# were removed from the default gem set). +# * rails 7.2 requires Ruby >= 3.1. +# * rails 8.0 / 8.1 require Ruby >= 3.2. +# Pair each gemfile with a compatible Ruby in the CI matrix. + +# Shims required by ActiveSupport < 7.1 on Ruby >= 3.4 (default gems removed). +older_rails_shims = proc do + gem "logger" + gem "mutex_m" + gem "bigdecimal" + gem "drb" + gem "base64" + gem "benchmark" +end + +appraise "rails-6.1" do + instance_exec(&older_rails_shims) + gem "activesupport", "~> 6.1.0" + gem "activerecord", "~> 6.1.0" +end + +appraise "rails-7.0" do + instance_exec(&older_rails_shims) + gem "activesupport", "~> 7.0.0" + gem "activerecord", "~> 7.0.0" +end + +appraise "rails-7.1" do + gem "activesupport", "~> 7.1.0" + gem "activerecord", "~> 7.1.0" +end + +appraise "rails-7.2" do + gem "activesupport", "~> 7.2.0" + gem "activerecord", "~> 7.2.0" +end + +appraise "rails-8.0" do + gem "activesupport", "~> 8.0.0" + gem "activerecord", "~> 8.0.0" +end + +appraise "rails-8.1" do + gem "activesupport", "~> 8.1.0" + gem "activerecord", "~> 8.1.0" +end diff --git a/Rakefile b/Rakefile index 950a17e..9c3a189 100644 --- a/Rakefile +++ b/Rakefile @@ -4,5 +4,14 @@ require "rspec/core/rake_task" desc "Run specs" RSpec::Core::RakeTask.new(:spec) +# Appraisal wires up per-Rails-version tasks (rake appraisal:rails-8.1 spec, etc.) +# when the appraisal gem is available. It is only a development dependency, so we +# guard the require to keep the Rakefile usable without it (e.g. from an installed gem). +begin + require "appraisal" +rescue LoadError + # appraisal not installed; per-version tasks are unavailable +end + desc "Run specs (default)" task :default => :spec diff --git a/action_subscriber.gemspec b/action_subscriber.gemspec index b2cf28c..520952b 100644 --- a/action_subscriber.gemspec +++ b/action_subscriber.gemspec @@ -34,6 +34,7 @@ Gem::Specification.new do |spec| spec.add_development_dependency "active_publisher", "1.6.0.pre1" spec.add_development_dependency "activerecord", ">= 6.0" + spec.add_development_dependency "appraisal", "~> 2.5" spec.add_development_dependency "bundler" spec.add_development_dependency "pry-nav" spec.add_development_dependency "rabbitmq_http_api_client", "~> 1.15.0" diff --git a/gemfiles/rails_6.1.gemfile b/gemfiles/rails_6.1.gemfile new file mode 100644 index 0000000..cd10b75 --- /dev/null +++ b/gemfiles/rails_6.1.gemfile @@ -0,0 +1,14 @@ +# This file was generated by Appraisal + +source "https://rubygems.org" + +gem "logger" +gem "mutex_m" +gem "bigdecimal" +gem "drb" +gem "base64" +gem "benchmark" +gem "activesupport", "~> 6.1.0" +gem "activerecord", "~> 6.1.0" + +gemspec path: "../" diff --git a/gemfiles/rails_7.0.gemfile b/gemfiles/rails_7.0.gemfile new file mode 100644 index 0000000..0b24599 --- /dev/null +++ b/gemfiles/rails_7.0.gemfile @@ -0,0 +1,14 @@ +# This file was generated by Appraisal + +source "https://rubygems.org" + +gem "logger" +gem "mutex_m" +gem "bigdecimal" +gem "drb" +gem "base64" +gem "benchmark" +gem "activesupport", "~> 7.0.0" +gem "activerecord", "~> 7.0.0" + +gemspec path: "../" diff --git a/gemfiles/rails_7.1.gemfile b/gemfiles/rails_7.1.gemfile new file mode 100644 index 0000000..6728e20 --- /dev/null +++ b/gemfiles/rails_7.1.gemfile @@ -0,0 +1,8 @@ +# This file was generated by Appraisal + +source "https://rubygems.org" + +gem "activesupport", "~> 7.1.0" +gem "activerecord", "~> 7.1.0" + +gemspec path: "../" diff --git a/gemfiles/rails_7.2.gemfile b/gemfiles/rails_7.2.gemfile new file mode 100644 index 0000000..21a60d9 --- /dev/null +++ b/gemfiles/rails_7.2.gemfile @@ -0,0 +1,8 @@ +# This file was generated by Appraisal + +source "https://rubygems.org" + +gem "activesupport", "~> 7.2.0" +gem "activerecord", "~> 7.2.0" + +gemspec path: "../" diff --git a/gemfiles/rails_8.0.gemfile b/gemfiles/rails_8.0.gemfile new file mode 100644 index 0000000..d7504cc --- /dev/null +++ b/gemfiles/rails_8.0.gemfile @@ -0,0 +1,8 @@ +# This file was generated by Appraisal + +source "https://rubygems.org" + +gem "activesupport", "~> 8.0.0" +gem "activerecord", "~> 8.0.0" + +gemspec path: "../" diff --git a/gemfiles/rails_8.1.gemfile b/gemfiles/rails_8.1.gemfile new file mode 100644 index 0000000..97d936a --- /dev/null +++ b/gemfiles/rails_8.1.gemfile @@ -0,0 +1,8 @@ +# This file was generated by Appraisal + +source "https://rubygems.org" + +gem "activesupport", "~> 8.1.0" +gem "activerecord", "~> 8.1.0" + +gemspec path: "../" diff --git a/spec/spec_helper.rb b/spec/spec_helper.rb index cd6b9a1..5e66089 100644 --- a/spec/spec_helper.rb +++ b/spec/spec_helper.rb @@ -18,6 +18,7 @@ # Require spec support files require 'support/user_subscriber' +require 'support/rabbitmq' require 'action_subscriber/rspec' # Silence the Logger @@ -30,6 +31,12 @@ mocks.verify_partial_doubles = true end + # Fail fast with a clear message (rather than a flurry of Bunny reconnect warnings) + # if the broker isn't up yet when the integration suite starts. + config.before(:suite) do + RabbitMQTestHelper.wait_for_rabbitmq! if RSpec.world.filtered_examples.values.flatten.any? { |ex| ex.metadata[:integration] } + end + config.before(:each, :integration => true) do $messages = Set.new draw_routes diff --git a/spec/support/rabbitmq.rb b/spec/support/rabbitmq.rb new file mode 100644 index 0000000..98353a0 --- /dev/null +++ b/spec/support/rabbitmq.rb @@ -0,0 +1,52 @@ +require "socket" + +# Helpers for running the suite against a real RabbitMQ broker. +# +# The integration specs talk to a live broker (the same approach used in CI, where a +# `rabbitmq` service container is started alongside the test job). Locally you can point +# at any running broker via RABBITMQ_URL / the standard host+port; by default we assume +# localhost:5672. +# +# NOTE: action_subscriber defaults to non-durable ("transient") queues. RabbitMQ 4.x +# denies transient non-exclusive queues by default, so a 4.x broker used for the suite +# must permit the deprecated feature: +# +# # rabbitmq.conf +# deprecated_features.permit.transient_nonexcl_queues = true +# +# The rabbitmq:3.12 image used in CI still allows them out of the box. See the phantom +# queue triage doc for why the production recommendation is to move to durable topology. +module RabbitMQTestHelper + module_function + + def host + ENV.fetch("RABBITMQ_HOST", "localhost") + end + + def port + Integer(ENV.fetch("RABBITMQ_PORT", "5672")) + end + + # Block until the broker's AMQP port accepts a TCP connection, or raise after `timeout` + # seconds. Keeps the suite from failing with confusing connection errors when the broker + # is still booting (common in CI service containers). + def wait_for_rabbitmq!(timeout: Integer(ENV.fetch("RABBITMQ_WAIT_TIMEOUT", "30"))) + deadline = ::Time.now + timeout + last_error = nil + loop do + begin + ::Socket.tcp(host, port, connect_timeout: 1) { |sock| sock.close } + return true + rescue ::StandardError => e + last_error = e + end + + if ::Time.now >= deadline + raise "RabbitMQ was not reachable at #{host}:#{port} within #{timeout}s " \ + "(last error: #{last_error.class}: #{last_error.message}). " \ + "Start a broker (see spec/support/rabbitmq.rb) before running the integration suite." + end + sleep 0.5 + end + end +end From 1a348ecbfb5827380380ca9fab5b45058204d5b4 Mon Sep 17 00:00:00 2001 From: John Bolliger Date: Thu, 6 Aug 2026 10:11:21 -0600 Subject: [PATCH 02/15] Add first party queue_type setting, defaulting to the broker default Adds a queue_type setting, configurable globally (config.queue_type) or per route (:queue_type => ...). Values are nil (default), :classic, :quorum and :stream, with :broker_default accepted as a readable alias for nil. nil leaves x-queue-type off the wire so RabbitMQ applies its own default_queue_type. :quorum and :stream force the route to be durable, since RabbitMQ only supports those as durable queues. Values are normalized on assignment, so an invalid value raises where it was set rather than at route-draw time or inside MessageRetry at runtime. BREAKING on JRuby. The two drivers disagreed: march_hare reads its :type option with fetch(:type, ... Types::CLASSIC), and fetch only falls back when the key is absent, so omitting :type injected x-queue-type: classic on every declare. bunny reads @options[:type] and sent no argument at all. Both drivers are now passed :type explicitly, so neither sends x-queue-type by default. MRI behavior is unchanged. On JRuby, newly declared queues change from classic to whatever the broker defaults to. Set config.queue_type to :classic to retain the previous behavior. Because queue type is fixed at declaration, redeclaring an existing queue against a vhost whose default_queue_type is not classic will fail with PRECONDITION_FAILED -- audit vhosts before upgrading a JRuby deployment. Known limitation, documented in the README: MessageRetry declares retry queues from the global config.queue_type rather than the originating route's, so a route-level type is not propagated to its retry queue. Co-Authored-By: Claude Opus 5 (1M context) --- README.md | 81 ++++++++++++++++ changelog.md | 20 ++++ lib/action_subscriber/bunny/subscriber.rb | 2 +- lib/action_subscriber/configuration.rb | 10 ++ .../march_hare/subscriber.rb | 4 +- lib/action_subscriber/message_retry.rb | 3 +- lib/action_subscriber/queue_type.rb | 46 +++++++++ lib/action_subscriber/route.rb | 13 ++- spec/lib/action_subscriber/queue_type_spec.rb | 96 +++++++++++++++++++ spec/lib/action_subscriber/router_spec.rb | 47 +++++++++ 10 files changed, 318 insertions(+), 4 deletions(-) create mode 100644 lib/action_subscriber/queue_type.rb create mode 100644 spec/lib/action_subscriber/queue_type_spec.rb diff --git a/README.md b/README.md index c9b35d5..93b5021 100644 --- a/README.md +++ b/README.md @@ -45,6 +45,86 @@ end That will give you a similar behavior to the old `--mode=pop` where messages polled from the server, but with reduced latency. +Queue Types +----------- + +Set the queue type globally, or per route: + +```ruby +::ActionSubscriber.configure do |config| + config.queue_type = :quorum +end + +::ActionSubscriber.draw_routes do + route UserSubscriber, :created, :queue_type => :quorum + route AuditSubscriber, :created, :queue_type => :broker_default +end +``` + +| Value | `x-queue-type` sent | +| --- | --- | +| `nil` (default), or `:broker_default` | *not sent* — the broker applies its own `default_queue_type` | +| `:classic` | `classic` | +| `:quorum` | `quorum` | +| `:stream` | `stream` | + +The default declares a queue without expressing an opinion, which lets an +operator move a vhost onto quorum queues with a broker policy instead of a code +change. `:broker_default` is accepted as a more readable spelling of `nil`; both +normalize to `nil`, and `config.queue_type` always reads back as `nil` or one of +the three type symbols. + +`:quorum` and `:stream` queues only exist as durable queues, so those two values +force `:durable => true` on the route regardless of what you pass. + +Invalid values raise an `ArgumentError` at the point they are assigned, rather +than later when routes are drawn or a queue is declared. + +> Note: a queue's type is fixed at declaration. Changing this setting will not +> convert an existing queue — the queue has to be deleted and redeclared, and +> redeclaring an existing queue with a conflicting type fails with +> `PRECONDITION_FAILED`. + +### Breaking change on JRuby + +Prior to this setting the two drivers disagreed. `march_hare` defaults its +`:type` option to `classic` and so injected `x-queue-type: classic` on every +queue it declared, while `bunny` sent no argument at all. ActionSubscriber now +passes `:type` explicitly on both drivers and defaults to `nil`, so neither +platform sends `x-queue-type`. + +**MRI behavior is unchanged. On JRuby, newly declared queues change from +`classic` to whatever the broker defaults to.** Set `config.queue_type = :classic` +to keep the previous JRuby behavior. + +The reason this matters beyond new queues: because queue type is fixed at +declaration, an existing `classic` queue is now *redeclared* without +`x-queue-type`. That is harmless on a vhost whose `default_queue_type` is +classic, since the broker resolves to the same type. It fails with +`PRECONDITION_FAILED` on a vhost whose default is `quorum` or `stream`. Before +upgrading a JRuby deployment, audit the `default_queue_type` of every vhost it +connects to: + +``` +rabbitmqctl list_vhosts name default_queue_type +``` + +If any are non-classic, set `config.queue_type = :classic` before rolling out. + +### Known limitation: retry queues + +`ActionSubscriber::MessageRetry` declares its `*.retry_*` queues using the +**global** `config.queue_type`, not the type of the route that produced the +message. A route that opts into `:quorum` while the global setting is left at +the default will dead-letter into a retry queue of a different type. + +If you rely on per-route queue types and on retries, set `config.queue_type` to +match rather than setting it per route. + +Note also that retry queues carry `x-message-ttl` and `x-dead-letter-exchange`, +which streams do not support — so a global `config.queue_type = :stream` will +make every retry declaration fail. + Supported Message Types ----------------- ActionSubscriber support JSON and plain text out of the box, but you can easily @@ -135,6 +215,7 @@ Other configuration options include : * config.network_recovery_interval - reconnection interval for TCP connection failures (default 1) * config.password - RabbitMQ password (default "guest") * config.prefetch - number of messages to hold in the local queue in subscriber mode +* config.queue_type - default queue type for all routes: `nil` (default, defers to the broker), `:classic`, `:quorum` or `:stream` * config.resubscribe_on_consumer_cancellation - resubscribe when the consumer is cancelled (queue deleted or cluster fails, default true) * config.seconds_to_wait_for_graceful_shutdown - time to wait before force stopping server after shutdown signal * config.threadpool_size - set the number of threads available to action_subscriber diff --git a/changelog.md b/changelog.md index 108c0b3..8a0e73c 100644 --- a/changelog.md +++ b/changelog.md @@ -1,5 +1,25 @@ ### Changelog +### Unreleased + +**Breaking change on JRuby.** `march_hare` defaults its `:type` option to +`classic` and so was injecting `x-queue-type: classic` on every queue it +declared, while `bunny` sent no argument at all. Both drivers are now passed +`:type` explicitly, defaulting to `nil`, so neither sends `x-queue-type` and the +broker's own `default_queue_type` applies. + +MRI behavior is unchanged. On JRuby, newly declared queues change from `classic` +to whatever the broker defaults to. Set `config.queue_type = :classic` to retain +the previous JRuby behavior. Note that queue type is fixed at declaration: +redeclaring an existing queue with a conflicting type fails with +`PRECONDITION_FAILED`, so audit any vhost whose `default_queue_type` is not +classic before upgrading. + +Added a first party `queue_type` setting behind that change, configurable +globally (`config.queue_type`) or per route (`:queue_type => ...`). Values are +`nil` (default), `:classic`, `:quorum` and `:stream`; `:broker_default` is +accepted as a readable alias for `nil`. `:quorum` and `:stream` force the route +to be durable. Invalid values raise where they are assigned. ### 5.4.0 - April 10, 2026 Added Ruby 3.4 / JRuby 10 support. \ No newline at end of file diff --git a/lib/action_subscriber/bunny/subscriber.rb b/lib/action_subscriber/bunny/subscriber.rb index 1957a1a..4ff037f 100644 --- a/lib/action_subscriber/bunny/subscriber.rb +++ b/lib/action_subscriber/bunny/subscriber.rb @@ -76,7 +76,7 @@ def start_subscriber_for_subscription(subscription) def setup_queue(route) channel = ::ActionSubscriber::RabbitConnection.with_connection{|connection| connection.create_channel(nil, 1) } exchange = channel.topic(route.exchange) - queue = channel.queue(route.queue, :durable => route.durable) + queue = channel.queue(route.queue, :durable => route.durable, :type => route.driver_queue_type) queue.bind(exchange, :routing_key => route.routing_key) queue end diff --git a/lib/action_subscriber/configuration.rb b/lib/action_subscriber/configuration.rb index 9f9a8d2..0ab7ea0 100644 --- a/lib/action_subscriber/configuration.rb +++ b/lib/action_subscriber/configuration.rb @@ -1,4 +1,5 @@ require "yaml" +require "action_subscriber/queue_type" require "action_subscriber/uri" module ActionSubscriber @@ -28,6 +29,10 @@ class Configuration :verify_peer, :virtual_host + # Written through QueueType.normalize so a typo raises where it was set + # rather than at route-draw time. nil means "defer to the broker". + attr_reader :queue_type + CONFIGURATION_MUTEX = ::Mutex.new NETWORK_RECOVERY_INTERVAL = 1.freeze @@ -43,6 +48,7 @@ class Configuration :password => "guest", :port => 5672, :prefetch => 2, + :queue_type => nil, :resubscribe_on_consumer_cancellation => true, :seconds_to_wait_for_graceful_shutdown => 30, :threadpool_size => 8, @@ -144,6 +150,10 @@ def middleware @middleware ||= Middleware.initialize_stack end + def queue_type=(value) + @queue_type = ::ActionSubscriber::QueueType.normalize(value) + end + def inspect inspection_string = <<-INSPECT.strip_heredoc Rabbit Hosts: #{hosts} diff --git a/lib/action_subscriber/march_hare/subscriber.rb b/lib/action_subscriber/march_hare/subscriber.rb index 06a89f4..45038d4 100644 --- a/lib/action_subscriber/march_hare/subscriber.rb +++ b/lib/action_subscriber/march_hare/subscriber.rb @@ -75,7 +75,9 @@ def start_subscriber_for_subscription(subscription) def setup_queue(route) channel = ::ActionSubscriber::RabbitConnection.with_connection{|connection| connection.create_channel } exchange = channel.topic(route.exchange) - queue = channel.queue(route.queue, :durable => route.durable) + # :type must be passed even when nil -- omitting it declares a classic + # queue here, unlike bunny. See ActionSubscriber::QueueType. + queue = channel.queue(route.queue, :durable => route.durable, :type => route.driver_queue_type) queue.bind(exchange, :routing_key => route.routing_key) queue end diff --git a/lib/action_subscriber/message_retry.rb b/lib/action_subscriber/message_retry.rb index ab95914..6a81fed 100644 --- a/lib/action_subscriber/message_retry.rb +++ b/lib/action_subscriber/message_retry.rb @@ -51,7 +51,8 @@ def self.with_exchange(env, ttl, retry_queue_name) channel.confirm_select # an empty string is the default exchange [see bunny docs](http://rubybunny.info/articles/exchanges.html#default_exchange) exchange = channel.topic("") - queue = channel.queue(retry_queue_name, :arguments => {"x-dead-letter-exchange" => "", "x-message-ttl" => ttl, "x-dead-letter-routing-key" => env.queue}) + queue_type = ::ActionSubscriber::QueueType.driver_option(::ActionSubscriber.config.queue_type) + queue = channel.queue(retry_queue_name, :type => queue_type, :arguments => {"x-dead-letter-exchange" => "", "x-message-ttl" => ttl, "x-dead-letter-routing-key" => env.queue}) yield(exchange) channel.wait_for_confirms end diff --git a/lib/action_subscriber/queue_type.rb b/lib/action_subscriber/queue_type.rb new file mode 100644 index 0000000..e9c3872 --- /dev/null +++ b/lib/action_subscriber/queue_type.rb @@ -0,0 +1,46 @@ +module ActionSubscriber + # Normalizes the `queue_type` setting into the value the underlying driver + # expects for its `:type` option. + # + # nil is the canonical "let the broker decide" value: it leaves `x-queue-type` + # off the wire so RabbitMQ applies its own `default_queue_type`. + # + # march_hare reads the driver option with + # `@options.fetch(:type, ... Types::CLASSIC)`, and `fetch` only falls back when + # the key is *absent*. So omitting `:type` silently declares a classic queue, + # while passing an explicit nil is what actually suppresses the argument. + module QueueType + # An explicit, readable alias for nil on input. Normalizes away to nil. + BROKER_DEFAULT = :broker_default + + SUPPORTED = [:classic, :quorum, :stream].freeze + + # Queue types that RabbitMQ only supports as durable queues. + ALWAYS_DURABLE = [:quorum, :stream].freeze + + def self.normalize(value) + queue_type = value.to_s.strip.downcase + return nil if queue_type.empty? || queue_type == BROKER_DEFAULT.to_s + + queue_type = queue_type.to_sym + unless SUPPORTED.include?(queue_type) + raise ::ArgumentError, + "unsupported queue_type #{value.inspect}, supported types are: #{SUPPORTED.join(', ')} " \ + "(or nil / :#{BROKER_DEFAULT} to defer to the broker)" + end + + queue_type + end + + # The value to hand the driver's `:type` option. Expects a normalized type. + def self.driver_option(queue_type) + return nil if queue_type.nil? + queue_type.to_s + end + + # Expects a normalized type. + def self.always_durable?(queue_type) + ALWAYS_DURABLE.include?(queue_type) + end + end +end diff --git a/lib/action_subscriber/route.rb b/lib/action_subscriber/route.rb index 9d25078..0a92598 100644 --- a/lib/action_subscriber/route.rb +++ b/lib/action_subscriber/route.rb @@ -2,10 +2,12 @@ module ActionSubscriber class Route attr_reader :acknowledgements, :action, + :driver_queue_type, :durable, :exchange, :prefetch, :queue, + :queue_type, :routing_key, :subscriber, :threadpool_name @@ -13,7 +15,16 @@ class Route def initialize(attributes) @acknowledgements = attributes.fetch(:acknowledgements) @action = attributes.fetch(:action) - @durable = attributes.fetch(:durable) + durable = attributes.fetch(:durable) + # Falls back to the global setting when a route does not name a type, the + # same way :prefetch does. nil means "defer to the broker". + @queue_type = ::ActionSubscriber::QueueType.normalize( + attributes.fetch(:queue_type) { ::ActionSubscriber.config.queue_type } + ) + @driver_queue_type = ::ActionSubscriber::QueueType.driver_option(@queue_type) + # Quorum and stream queues only exist as durable queues, so the broker + # rejects them otherwise. march_hare already forces this internally. + @durable = ::ActionSubscriber::QueueType.always_durable?(@queue_type) || durable @exchange = attributes.fetch(:exchange).to_s @prefetch = attributes.fetch(:prefetch) { ::ActionSubscriber.config.prefetch } @queue = attributes.fetch(:queue) diff --git a/spec/lib/action_subscriber/queue_type_spec.rb b/spec/lib/action_subscriber/queue_type_spec.rb new file mode 100644 index 0000000..ab43500 --- /dev/null +++ b/spec/lib/action_subscriber/queue_type_spec.rb @@ -0,0 +1,96 @@ +describe ActionSubscriber::QueueType do + describe ".normalize" do + it "treats nil and blank strings as the broker default" do + expect(described_class.normalize(nil)).to be_nil + expect(described_class.normalize("")).to be_nil + expect(described_class.normalize(" ")).to be_nil + end + + it "treats :broker_default as an alias for nil" do + expect(described_class.normalize(:broker_default)).to be_nil + expect(described_class.normalize("broker_default")).to be_nil + end + + it "accepts strings and symbols for the supported types" do + expect(described_class.normalize("classic")).to eq(:classic) + expect(described_class.normalize(:quorum)).to eq(:quorum) + expect(described_class.normalize("STREAM")).to eq(:stream) + end + + it "raises on an unsupported type" do + expect { described_class.normalize(:mirrored) }.to raise_error(ArgumentError, /unsupported queue_type/) + end + end + + describe ".driver_option" do + # nil is what keeps x-queue-type off the wire in both bunny and march_hare. + it "is nil for the broker default" do + expect(described_class.driver_option(nil)).to be_nil + end + + it "is the type name for an explicit type" do + expect(described_class.driver_option(:quorum)).to eq("quorum") + expect(described_class.driver_option(:classic)).to eq("classic") + end + end + + describe ".always_durable?" do + it "is true for quorum and stream queues" do + expect(described_class.always_durable?(:quorum)).to eq(true) + expect(described_class.always_durable?(:stream)).to eq(true) + end + + it "is false for classic and broker default queues" do + expect(described_class.always_durable?(:classic)).to eq(false) + expect(described_class.always_durable?(nil)).to eq(false) + end + end + + describe "configuration" do + around do |example| + original = ActionSubscriber.config.queue_type + example.run + ActionSubscriber.config.queue_type = original + end + + it "defaults to nil" do + expect(ActionSubscriber.config.queue_type).to be_nil + end + + it "normalizes on assignment so readers always see a symbol or nil" do + ActionSubscriber.config.queue_type = "quorum" + expect(ActionSubscriber.config.queue_type).to eq(:quorum) + + ActionSubscriber.config.queue_type = :broker_default + expect(ActionSubscriber.config.queue_type).to be_nil + end + + it "raises at the point the bad value is set" do + expect { ActionSubscriber.config.queue_type = "qourum" }.to raise_error(ArgumentError, /unsupported queue_type/) + end + end + + # The behavior this whole module exists for: march_hare reads its :type option + # with fetch(:type, ... CLASSIC), so omitting the key declares a classic queue + # while passing an explicit nil leaves x-queue-type off the wire. + if ::RUBY_PLATFORM == "java" + describe "march_hare integration" do + def arguments_for(type) + ::MarchHare::Queue.new(nil, "test.queue", :durable => false, :type => type).arguments + end + + it "sends no x-queue-type when the driver option is nil" do + expect(arguments_for(described_class.driver_option(nil))).to eq({}) + end + + it "sends x-queue-type when a type is named" do + expect(arguments_for(described_class.driver_option(:quorum))).to eq("x-queue-type" => "quorum") + end + + it "would send classic if the :type key were omitted entirely" do + omitted = ::MarchHare::Queue.new(nil, "test.queue", :durable => false).arguments + expect(omitted).to eq("x-queue-type" => "classic") + end + end + end +end diff --git a/spec/lib/action_subscriber/router_spec.rb b/spec/lib/action_subscriber/router_spec.rb index 24eeee8..ac35676 100644 --- a/spec/lib/action_subscriber/router_spec.rb +++ b/spec/lib/action_subscriber/router_spec.rb @@ -57,6 +57,53 @@ class FakeSubscriber; end expect(routes.first.queue).to eq("alice.fake.foo") end + it "defers to the broker by default" do + routes = described_class.draw_routes do + route FakeSubscriber, :foo + end + + expect(routes.first.queue_type).to be_nil + expect(routes.first.driver_queue_type).to be_nil + end + + it "accepts :broker_default as an explicit alias for nil" do + routes = described_class.draw_routes do + route FakeSubscriber, :foo, :queue_type => :broker_default + end + + expect(routes.first.queue_type).to be_nil + expect(routes.first.driver_queue_type).to be_nil + end + + it "can specify a queue type" do + routes = described_class.draw_routes do + route FakeSubscriber, :foo, :queue_type => :classic + end + + expect(routes.first.queue_type).to eq(:classic) + expect(routes.first.driver_queue_type).to eq("classic") + expect(routes.first.durable).to eq(false) + end + + it "forces quorum queues to be durable" do + routes = described_class.draw_routes do + route FakeSubscriber, :foo, :queue_type => :quorum + end + + expect(routes.first.queue_type).to eq(:quorum) + expect(routes.first.durable).to eq(true) + end + + it "inherits the queue type from the global configuration" do + allow(ActionSubscriber.config).to receive(:queue_type).and_return(:quorum) + + routes = described_class.draw_routes do + route FakeSubscriber, :foo + end + + expect(routes.first.queue_type).to eq(:quorum) + end + it "can specify a queue is durable" do routes = described_class.draw_routes do route FakeSubscriber, :foo, :durable => true From 6c1727f9ee3687d016e82524df07f121febf789c Mon Sep 17 00:00:00 2001 From: John Bolliger Date: Thu, 6 Aug 2026 10:14:39 -0600 Subject: [PATCH 03/15] Release 7.5.0 Bump VERSION to 7.5.0 and date the changelog entry for the queue_type setting and the JRuby x-queue-type behavior change. Co-Authored-By: Claude Opus 5 (1M context) --- changelog.md | 2 +- lib/action_subscriber/version.rb | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/changelog.md b/changelog.md index 8a0e73c..ac8c00a 100644 --- a/changelog.md +++ b/changelog.md @@ -1,6 +1,6 @@ ### Changelog -### Unreleased +### 7.5.0 - August 6, 2026 **Breaking change on JRuby.** `march_hare` defaults its `:type` option to `classic` and so was injecting `x-queue-type: classic` on every queue it diff --git a/lib/action_subscriber/version.rb b/lib/action_subscriber/version.rb index cdfa208..99d8174 100644 --- a/lib/action_subscriber/version.rb +++ b/lib/action_subscriber/version.rb @@ -1,3 +1,3 @@ module ActionSubscriber - VERSION = "5.4.0" + VERSION = "7.5.0" end From 74bf1846fab25f166a9271442a76a1caf3b8956a Mon Sep 17 00:00:00 2001 From: John Bolliger Date: Thu, 6 Aug 2026 10:17:33 -0600 Subject: [PATCH 04/15] Correct release version to 6.0.0 The previous commit bumped to 7.5.0, which skipped the 6.x line and read as a minor bump. Use 6.0.0 instead: a major bump is warranted because the x-queue-type change is breaking for JRuby consumers, where queues that march_hare previously declared as classic are now declared with whatever the broker defaults to. Co-Authored-By: Claude Opus 5 (1M context) --- changelog.md | 2 +- lib/action_subscriber/version.rb | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/changelog.md b/changelog.md index ac8c00a..8612ad5 100644 --- a/changelog.md +++ b/changelog.md @@ -1,6 +1,6 @@ ### Changelog -### 7.5.0 - August 6, 2026 +### 6.0.0 - August 6, 2026 **Breaking change on JRuby.** `march_hare` defaults its `:type` option to `classic` and so was injecting `x-queue-type: classic` on every queue it diff --git a/lib/action_subscriber/version.rb b/lib/action_subscriber/version.rb index 99d8174..4b0a00c 100644 --- a/lib/action_subscriber/version.rb +++ b/lib/action_subscriber/version.rb @@ -1,3 +1,3 @@ module ActionSubscriber - VERSION = "7.5.0" + VERSION = "6.0.0" end From 5ca2d3f4bb2469491e35e57522fdeb09049eb249 Mon Sep 17 00:00:00 2001 From: John Bolliger Date: Thu, 6 Aug 2026 10:25:10 -0600 Subject: [PATCH 05/15] Generate appraisal gemfiles in CI instead of committing them The gemfiles/ directory is fully derived from the Appraisals file, so committing the stubs meant keeping generated output in sync by hand. Gitignore the whole directory and regenerate it in CI instead. The generate step has to override BUNDLE_GEMFILE to the root Gemfile: the job sets it to the target appraisal gemfile, which does not exist until this step writes it. Verified that `appraisal generate` runs from a clean checkout with no prior bundle install, and that its output is byte-identical to the stubs being removed here. Cache keys now also checksum Appraisals, so editing it busts the bundle cache, and are bumped to v3 since the previous caches predate this. Document the local workflow in the README, since gemfiles/ no longer exists after a fresh clone. Co-Authored-By: Claude Opus 5 (1M context) --- .circleci/config.yml | 25 ++++++++++++++++++------- .gitignore | 7 ++++--- README.md | 22 ++++++++++++++++++++++ gemfiles/rails_6.1.gemfile | 14 -------------- gemfiles/rails_7.0.gemfile | 14 -------------- gemfiles/rails_7.1.gemfile | 8 -------- gemfiles/rails_7.2.gemfile | 8 -------- gemfiles/rails_8.0.gemfile | 8 -------- gemfiles/rails_8.1.gemfile | 8 -------- 9 files changed, 44 insertions(+), 70 deletions(-) delete mode 100644 gemfiles/rails_6.1.gemfile delete mode 100644 gemfiles/rails_7.0.gemfile delete mode 100644 gemfiles/rails_7.1.gemfile delete mode 100644 gemfiles/rails_7.2.gemfile delete mode 100644 gemfiles/rails_8.0.gemfile delete mode 100644 gemfiles/rails_8.1.gemfile diff --git a/.circleci/config.yml b/.circleci/config.yml index a232b12..1221e41 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -42,25 +42,36 @@ jobs: sudo apt-get update && sudo apt-get install -y build-essential git fi - checkout - # Cache key includes the Ruby image + the specific appraisal gemfile + the gemspec, - # so MRI/JRuby and each Rails version get independent caches. (Gemfile.lock and the - # generated gemfiles/*.lock are gitignored, so we key on committed files instead.) + # The gemfiles/ directory is gitignored and regenerated below, so it does not + # exist at checkout. BUNDLE_GEMFILE points at the target appraisal gemfile for + # the whole job, which means it has to be overridden to the root Gemfile here -- + # otherwise appraisal would try to read the very file it is about to write. + - run: + name: Generate Appraisal Gemfiles + command: | + gem install bundler appraisal + BUNDLE_GEMFILE=Gemfile appraisal generate + ls -1 gemfiles/ + + # Cache key includes the Ruby image + the specific appraisal gemfile + the gemspec + # + the Appraisals file, so MRI/JRuby and each Rails version get independent caches + # and a dependency change in either file busts them. (All lockfiles are gitignored, + # so we key on committed sources instead.) - restore_cache: keys: - - v2-gems-<< parameters.docker_image >>-<< parameters.gemfile >>-{{ checksum "action_subscriber.gemspec" }} - - v2-gems-<< parameters.docker_image >>-<< parameters.gemfile >>- + - v3-gems-<< parameters.docker_image >>-<< parameters.gemfile >>-{{ checksum "action_subscriber.gemspec" }}-{{ checksum "Appraisals" }} + - v3-gems-<< parameters.docker_image >>-<< parameters.gemfile >>- - run: name: Install Ruby Dependencies command: | - gem install bundler bundle config set --local path 'vendor/bundle' bundle install --jobs=4 --retry=3 - save_cache: paths: - ./vendor/bundle - key: v2-gems-<< parameters.docker_image >>-<< parameters.gemfile >>-{{ checksum "action_subscriber.gemspec" }} + key: v3-gems-<< parameters.docker_image >>-<< parameters.gemfile >>-{{ checksum "action_subscriber.gemspec" }}-{{ checksum "Appraisals" }} # Wait for RabbitMQ to be ready before running tests. # Service containers can sometimes take a few seconds to boot up. diff --git a/.gitignore b/.gitignore index 776bb87..f0f8b8f 100644 --- a/.gitignore +++ b/.gitignore @@ -15,9 +15,10 @@ tmp Gemfile.lock -# Appraisal-generated lock files (the .gemfile stubs are committed, the locks are not) -gemfiles/*.gemfile.lock -gemfiles/.bundle +# Appraisal output. Both the .gemfile stubs and their locks are generated from +# the Appraisals file -- CI regenerates them, and locally you run: +# bundle exec appraisal generate +gemfiles/ # YARD artifacts .yardoc diff --git a/README.md b/README.md index 93b5021..09ea620 100644 --- a/README.md +++ b/README.md @@ -322,3 +322,25 @@ $ cd action_subscriber $ bundle install $ bundle exec rspec ``` + +### Testing against multiple Rails versions + +The supported Rails versions are declared in `Appraisals`. The `gemfiles/` +directory is **generated, not committed** — it is gitignored, and CI regenerates +it on every run. To create it locally: + +``` +$ bundle exec appraisal generate # writes gemfiles/*.gemfile +$ bundle exec appraisal install # resolves a lockfile for each +``` + +Then run the suite against one version, or all of them: + +``` +$ BUNDLE_GEMFILE=gemfiles/rails_8.0.gemfile bundle exec rspec +$ bundle exec appraisal rspec +``` + +Re-run `appraisal generate` after editing `Appraisals`. Note that Rails 7.2 +requires Ruby >= 3.1 and Rails 8.0/8.1 require Ruby >= 3.2, so those gemfiles +will not resolve on older interpreters. diff --git a/gemfiles/rails_6.1.gemfile b/gemfiles/rails_6.1.gemfile deleted file mode 100644 index cd10b75..0000000 --- a/gemfiles/rails_6.1.gemfile +++ /dev/null @@ -1,14 +0,0 @@ -# This file was generated by Appraisal - -source "https://rubygems.org" - -gem "logger" -gem "mutex_m" -gem "bigdecimal" -gem "drb" -gem "base64" -gem "benchmark" -gem "activesupport", "~> 6.1.0" -gem "activerecord", "~> 6.1.0" - -gemspec path: "../" diff --git a/gemfiles/rails_7.0.gemfile b/gemfiles/rails_7.0.gemfile deleted file mode 100644 index 0b24599..0000000 --- a/gemfiles/rails_7.0.gemfile +++ /dev/null @@ -1,14 +0,0 @@ -# This file was generated by Appraisal - -source "https://rubygems.org" - -gem "logger" -gem "mutex_m" -gem "bigdecimal" -gem "drb" -gem "base64" -gem "benchmark" -gem "activesupport", "~> 7.0.0" -gem "activerecord", "~> 7.0.0" - -gemspec path: "../" diff --git a/gemfiles/rails_7.1.gemfile b/gemfiles/rails_7.1.gemfile deleted file mode 100644 index 6728e20..0000000 --- a/gemfiles/rails_7.1.gemfile +++ /dev/null @@ -1,8 +0,0 @@ -# This file was generated by Appraisal - -source "https://rubygems.org" - -gem "activesupport", "~> 7.1.0" -gem "activerecord", "~> 7.1.0" - -gemspec path: "../" diff --git a/gemfiles/rails_7.2.gemfile b/gemfiles/rails_7.2.gemfile deleted file mode 100644 index 21a60d9..0000000 --- a/gemfiles/rails_7.2.gemfile +++ /dev/null @@ -1,8 +0,0 @@ -# This file was generated by Appraisal - -source "https://rubygems.org" - -gem "activesupport", "~> 7.2.0" -gem "activerecord", "~> 7.2.0" - -gemspec path: "../" diff --git a/gemfiles/rails_8.0.gemfile b/gemfiles/rails_8.0.gemfile deleted file mode 100644 index d7504cc..0000000 --- a/gemfiles/rails_8.0.gemfile +++ /dev/null @@ -1,8 +0,0 @@ -# This file was generated by Appraisal - -source "https://rubygems.org" - -gem "activesupport", "~> 8.0.0" -gem "activerecord", "~> 8.0.0" - -gemspec path: "../" diff --git a/gemfiles/rails_8.1.gemfile b/gemfiles/rails_8.1.gemfile deleted file mode 100644 index 97d936a..0000000 --- a/gemfiles/rails_8.1.gemfile +++ /dev/null @@ -1,8 +0,0 @@ -# This file was generated by Appraisal - -source "https://rubygems.org" - -gem "activesupport", "~> 8.1.0" -gem "activerecord", "~> 8.1.0" - -gemspec path: "../" From a38220f574598778784a39c3d596fd8ccae41581 Mon Sep 17 00:00:00 2001 From: John Bolliger Date: Thu, 6 Aug 2026 10:31:48 -0600 Subject: [PATCH 06/15] change active_publisher version --- action_subscriber.gemspec | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/action_subscriber.gemspec b/action_subscriber.gemspec index 520952b..fcdf0b5 100644 --- a/action_subscriber.gemspec +++ b/action_subscriber.gemspec @@ -32,7 +32,7 @@ Gem::Specification.new do |spec| spec.add_dependency "middleware" spec.add_dependency "thor" - spec.add_development_dependency "active_publisher", "1.6.0.pre1" + spec.add_development_dependency "active_publisher", "1.6.0" spec.add_development_dependency "activerecord", ">= 6.0" spec.add_development_dependency "appraisal", "~> 2.5" spec.add_development_dependency "bundler" From 10a616fa959a403de51e9dbbea9395321f348072 Mon Sep 17 00:00:00 2001 From: John Bolliger Date: Thu, 6 Aug 2026 10:41:59 -0600 Subject: [PATCH 07/15] Fix appraisal gemfile generation in CI The generate step ran `appraisal generate` without installing the root bundle first. appraisal runs under bundler, so it aborted with: Could not find gem 'active_publisher (= 1.6.0)' in locally installed gems. Run `bundle install --gemfile Gemfile` to install missing gems. Install the root Gemfile before generating. The two bundles differ only in their Rails pins and share vendor/bundle, so the follow-up install is mostly a no-op. Also cache ./gemfiles/vendor/bundle alongside ./vendor/bundle. Because `bundle config --local` resolves relative to the directory holding BUNDLE_GEMFILE, the appraisal bundle installs under gemfiles/, so caching only ./vendor/bundle reinstalled the gems the tests actually use on every run. Cache keys bumped to v4. Verified the full sequence from a clean clone with an isolated GEM_HOME under JRuby 10: root install, generate, and the rails_8.0 target install all exit 0 and resolve activesupport 8.0.5.1. Co-Authored-By: Claude Opus 5 (1M context) --- .circleci/config.yml | 41 +++++++++++++++++++++++++++-------------- 1 file changed, 27 insertions(+), 14 deletions(-) diff --git a/.circleci/config.yml b/.circleci/config.yml index 1221e41..50ba9dd 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -42,25 +42,33 @@ jobs: sudo apt-get update && sudo apt-get install -y build-essential git fi - checkout - # The gemfiles/ directory is gitignored and regenerated below, so it does not - # exist at checkout. BUNDLE_GEMFILE points at the target appraisal gemfile for - # the whole job, which means it has to be overridden to the root Gemfile here -- - # otherwise appraisal would try to read the very file it is about to write. - - run: - name: Generate Appraisal Gemfiles - command: | - gem install bundler appraisal - BUNDLE_GEMFILE=Gemfile appraisal generate - ls -1 gemfiles/ - # Cache key includes the Ruby image + the specific appraisal gemfile + the gemspec # + the Appraisals file, so MRI/JRuby and each Rails version get independent caches # and a dependency change in either file busts them. (All lockfiles are gitignored, # so we key on committed sources instead.) - restore_cache: keys: - - v3-gems-<< parameters.docker_image >>-<< parameters.gemfile >>-{{ checksum "action_subscriber.gemspec" }}-{{ checksum "Appraisals" }} - - v3-gems-<< parameters.docker_image >>-<< parameters.gemfile >>- + - v4-gems-<< parameters.docker_image >>-<< parameters.gemfile >>-{{ checksum "action_subscriber.gemspec" }}-{{ checksum "Appraisals" }} + - v4-gems-<< parameters.docker_image >>-<< parameters.gemfile >>- + + # gemfiles/ is gitignored and generated here, so it does not exist at checkout. + # + # Two things this step has to get right: + # 1. BUNDLE_GEMFILE is set job-wide to the target appraisal gemfile, which does + # not exist yet. Override it to the root Gemfile or appraisal tries to read + # the very file it is about to write. + # 2. appraisal runs under bundler, so the root Gemfile must be *installed* + # first -- otherwise it aborts with "Could not find gem ... in locally + # installed gems". The two bundles differ only in their Rails pins and share + # vendor/bundle, so the second install below is mostly a no-op. + - run: + name: Generate Appraisal Gemfiles + command: | + gem install bundler appraisal + bundle config set --local path 'vendor/bundle' + BUNDLE_GEMFILE=Gemfile bundle install --jobs=4 --retry=3 + BUNDLE_GEMFILE=Gemfile bundle exec appraisal generate + ls -1 gemfiles/ - run: name: Install Ruby Dependencies @@ -68,10 +76,15 @@ jobs: bundle config set --local path 'vendor/bundle' bundle install --jobs=4 --retry=3 + # Two paths: `bundle config --local` is relative to the directory holding + # BUNDLE_GEMFILE, so the root bundle lands in ./vendor/bundle while the + # appraisal bundle lands in ./gemfiles/vendor/bundle. Caching only the first + # would silently reinstall the gems the tests actually run against. - save_cache: paths: - ./vendor/bundle - key: v3-gems-<< parameters.docker_image >>-<< parameters.gemfile >>-{{ checksum "action_subscriber.gemspec" }}-{{ checksum "Appraisals" }} + - ./gemfiles/vendor/bundle + key: v4-gems-<< parameters.docker_image >>-<< parameters.gemfile >>-{{ checksum "action_subscriber.gemspec" }}-{{ checksum "Appraisals" }} # Wait for RabbitMQ to be ready before running tests. # Service containers can sometimes take a few seconds to boot up. From 41939687d20be89f6c2d9f2febb8c3718c9da71a Mon Sep 17 00:00:00 2001 From: John Bolliger Date: Wed, 12 Aug 2026 13:53:55 -0700 Subject: [PATCH 08/15] Add a first party durable setting Durability could only be set per route. This adds it as a configuration setting, so it can be turned on globally -- including from config/action_subscriber.yml, which means an operator can do it without a code change -- or declared per subscriber: production: durable: true class UserSubscriber < ::ActionSubscriber::Base durable true end Precedence is route option > subscriber declaration > config.durable. The default is unchanged at false. This is the smallest way onto RabbitMQ 4.x, which denies the transient queues every default route declares. config.queue_type = :quorum also works but changes the queue type as well; config.durable leaves it alone. Two notes on the implementation: * :durable had to come out of Router::DEFAULT_SETTINGS and DefaultRouting. Baking in a value makes an unspecified route indistinguishable from one that explicitly asked for false, so the fallback could never fire. :prefetch was already the precedent for leaving a key out and letting Route resolve it. * the resolution lives in Route rather than Router so that both `route` and `default_routes_for` honor the subscriber's declaration. Note that :acknowledgements does not do this -- manual_acknowledgement! only takes effect through default_routes_for. QueueType.durable? gives the "quorum and stream are always durable" rule a single owner, since the retry path needs it too. Co-Authored-By: Claude Opus 5 (1M context) --- lib/action_subscriber/configuration.rb | 5 ++ lib/action_subscriber/default_routing.rb | 3 +- lib/action_subscriber/dsl.rb | 23 ++++++ lib/action_subscriber/queue_type.rb | 8 ++ lib/action_subscriber/route.rb | 20 ++++- lib/action_subscriber/router.rb | 9 ++- .../action_subscriber/configuration_spec.rb | 20 ++++- spec/lib/action_subscriber/dsl_spec.rb | 26 +++++++ spec/lib/action_subscriber/queue_type_spec.rb | 22 ++++-- spec/lib/action_subscriber/router_spec.rb | 75 +++++++++++++++++++ spec/support/sample_config.yml | 5 ++ 11 files changed, 204 insertions(+), 12 deletions(-) diff --git a/lib/action_subscriber/configuration.rb b/lib/action_subscriber/configuration.rb index 0ab7ea0..720052e 100644 --- a/lib/action_subscriber/configuration.rb +++ b/lib/action_subscriber/configuration.rb @@ -9,6 +9,7 @@ class Configuration :connection_reaping_timeout_interval, :decoder, :default_exchange, + :durable, :error_handler, :heartbeat, :host, @@ -41,6 +42,10 @@ class Configuration :connection_reaping_interval => 6, :connection_reaping_timeout_interval => 5, :default_exchange => 'events', + # Default durability for every route that does not name one. Kept at false for + # backwards compatibility, but note that a transient queue cannot be declared at + # all on a stock RabbitMQ 4.x broker -- see the README. + :durable => false, :heartbeat => 5, :host => 'localhost', :hosts => [], diff --git a/lib/action_subscriber/default_routing.rb b/lib/action_subscriber/default_routing.rb index ab150b8..8610d5a 100644 --- a/lib/action_subscriber/default_routing.rb +++ b/lib/action_subscriber/default_routing.rb @@ -5,10 +5,11 @@ def routes(route_settings) routes = [] exchange_names.each do |exchange_name| subscribable_methods.each do |method_name| + # No :durable key -- Route falls back to config.durable when it is absent, + # and passing false here would override the global setting. settings = { acknowledgements: acknowledge_messages?, action: method_name, - durable: false, exchange: exchange_name, routing_key: routing_key_name_for_method(method_name), subscriber: self, diff --git a/lib/action_subscriber/dsl.rb b/lib/action_subscriber/dsl.rb index dc25f87..5712545 100644 --- a/lib/action_subscriber/dsl.rb +++ b/lib/action_subscriber/dsl.rb @@ -55,6 +55,29 @@ def acknowledge_messages? !!@_acknowledge_messages end + # Set durability for every queue drawn from this subscriber, overriding + # config.durable. A route's own :durable option still wins over this. + # + # class UserSubscriber < ::ActionSubscriber::Base + # exchange :events + # durable true + # end + # + # Pass false to pin a subscriber transient when config.durable is on. Note that + # RabbitMQ 4.x refuses transient non-exclusive queues outright. + # + # Durability is fixed when a queue is created -- the broker rejects a redeclaration + # that disagrees with the existing queue -- so changing this for a subscriber whose + # queues already exist means deleting those queues first. + # + # Reads back nil when never set, which is how a route knows to fall back to + # config.durable. Deliberately not `durable?` -- every predicate in this file + # returns a strict boolean, and this cannot. + def durable(value = nil) + @_durable = value unless value.nil? + @_durable + end + def around_filter(callback_method, options = nil) filter = Filter.new(callback_method, options) conditionally_add_filter!(filter) diff --git a/lib/action_subscriber/queue_type.rb b/lib/action_subscriber/queue_type.rb index e9c3872..80a31fb 100644 --- a/lib/action_subscriber/queue_type.rb +++ b/lib/action_subscriber/queue_type.rb @@ -42,5 +42,13 @@ def self.driver_option(queue_type) def self.always_durable?(queue_type) ALWAYS_DURABLE.include?(queue_type) end + + # The durability a declaration should actually use. Quorum and stream queues only + # exist as durable queues, so those two override whatever was requested. Every + # declaration site goes through here -- march_hare forces this internally and bunny + # does not, so open-coding it once per site is how the drivers drift apart. + def self.durable?(queue_type, requested) + always_durable?(queue_type) || !!requested + end end end diff --git a/lib/action_subscriber/route.rb b/lib/action_subscriber/route.rb index 0a92598..b460f77 100644 --- a/lib/action_subscriber/route.rb +++ b/lib/action_subscriber/route.rb @@ -15,16 +15,17 @@ class Route def initialize(attributes) @acknowledgements = attributes.fetch(:acknowledgements) @action = attributes.fetch(:action) - durable = attributes.fetch(:durable) + # Precedence: the route's own :durable option, then the subscriber's `durable` + # declaration, then config.durable. Resolved here rather than in Router so that + # both `route` and `default_routes_for` honor the subscriber's declaration. + durable = attributes.fetch(:durable) { default_durability(attributes.fetch(:subscriber)) } # Falls back to the global setting when a route does not name a type, the # same way :prefetch does. nil means "defer to the broker". @queue_type = ::ActionSubscriber::QueueType.normalize( attributes.fetch(:queue_type) { ::ActionSubscriber.config.queue_type } ) @driver_queue_type = ::ActionSubscriber::QueueType.driver_option(@queue_type) - # Quorum and stream queues only exist as durable queues, so the broker - # rejects them otherwise. march_hare already forces this internally. - @durable = ::ActionSubscriber::QueueType.always_durable?(@queue_type) || durable + @durable = ::ActionSubscriber::QueueType.durable?(@queue_type, durable) @exchange = attributes.fetch(:exchange).to_s @prefetch = attributes.fetch(:prefetch) { ::ActionSubscriber.config.prefetch } @queue = attributes.fetch(:queue) @@ -45,5 +46,16 @@ def acknowledgements? def queue_subscription_options { :manual_ack => acknowledgements? } end + + private + + # nil from the subscriber means it did not express an opinion. Guarded by + # respond_to? because a route can name any object as its subscriber -- only + # ActionSubscriber::Base descendants carry the DSL. + def default_durability(subscriber) + declared = subscriber.durable if subscriber.respond_to?(:durable) + return declared unless declared.nil? + ::ActionSubscriber.config.durable + end end end diff --git a/lib/action_subscriber/router.rb b/lib/action_subscriber/router.rb index 4b3714d..224b229 100644 --- a/lib/action_subscriber/router.rb +++ b/lib/action_subscriber/router.rb @@ -6,9 +6,16 @@ def self.draw_routes(&block) router.routes end + # :durable is deliberately absent -- baking it in here would make an unspecified + # route indistinguishable from one that explicitly asked for false, and Route needs + # to tell them apart to fall back to config.durable. Same reason :prefetch is absent. + # + # The two keys that remain predate that rule and do not follow it: :exchange shadows + # config.default_exchange, and :acknowledgements shadows the subscriber's + # at_least_once! / manual_acknowledgement! declaration, for `route` but not for + # `default_routes_for`. Moving either into Route is a behavior change, not a cleanup. DEFAULT_SETTINGS = { :acknowledgements => false, - :durable => false, :exchange => "events", }.freeze diff --git a/spec/lib/action_subscriber/configuration_spec.rb b/spec/lib/action_subscriber/configuration_spec.rb index d117c59..d13ffb4 100644 --- a/spec/lib/action_subscriber/configuration_spec.rb +++ b/spec/lib/action_subscriber/configuration_spec.rb @@ -2,6 +2,7 @@ describe "default values" do specify { expect(subject.allow_low_priority_methods).to eq(false) } specify { expect(subject.default_exchange).to eq("events") } + specify { expect(subject.durable).to eq(false) } specify { expect(subject.heartbeat).to eq(5) } specify { expect(subject.host).to eq("localhost") } specify { expect(subject.network_recovery_interval).to eq(1) } @@ -14,7 +15,10 @@ end describe ".configure_from_yaml_and_cli" do - context "when using a yaml file" do + # These examples really load the fixture, so whatever it sets sticks. :as_config + # names durable so it gets snapshotted and put back -- every route drawn by the rest + # of the suite reads it. + context "when using a yaml file", :as_config => { :durable => false } do let!(:sample_yaml_location) { ::File.expand_path(::File.join("spec", "support", "sample_config.yml")) } before { allow(::File).to receive(:expand_path) { sample_yaml_location } } @@ -23,6 +27,20 @@ expect(::ActionSubscriber.configuration).to receive(:password=).with("WAT").and_return(true) ::ActionSubscriber::Configuration.configure_from_yaml_and_cli({}, true) end + + # durable has no bespoke wiring -- being in DEFAULTS is the whole implementation. + # Worth pinning, since the point of the setting is that an operator can turn + # durability on from the config file without a code change. + it "loads durable" do + # The fixture sets password too, and this example -- unlike the one above -- really + # applies what it loads. A leaked password fails every later integration example + # with Bunny::AuthenticationFailureError, so keep that one from landing. + expect(::ActionSubscriber.configuration).to receive(:password=).with("WAT") + + ::ActionSubscriber::Configuration.configure_from_yaml_and_cli({}, true) + + expect(::ActionSubscriber.configuration.durable).to eq(true) + end end it "can override a true value with a false value" do diff --git a/spec/lib/action_subscriber/dsl_spec.rb b/spec/lib/action_subscriber/dsl_spec.rb index 59fe0f3..664f7d3 100644 --- a/spec/lib/action_subscriber/dsl_spec.rb +++ b/spec/lib/action_subscriber/dsl_spec.rb @@ -42,6 +42,32 @@ end end + describe "durable" do + context "when set to true" do + before { subscriber.durable true } + + it "reads back as true" do + expect(subscriber.durable).to eq(true) + end + end + + context "when set to false" do + before { subscriber.durable false } + + # Distinct from "unset" -- this pins the subscriber transient even when + # config.durable is on. + it "reads back as false" do + expect(subscriber.durable).to eq(false) + end + end + + context "when not set" do + it "is nil, so routes fall back to config.durable" do + expect(subscriber.durable).to be_nil + end + end + end + describe "exchange_names" do context "when exchange names are set" do before { subscriber.exchange_names :foo, :bar } diff --git a/spec/lib/action_subscriber/queue_type_spec.rb b/spec/lib/action_subscriber/queue_type_spec.rb index ab43500..7dcf6a6 100644 --- a/spec/lib/action_subscriber/queue_type_spec.rb +++ b/spec/lib/action_subscriber/queue_type_spec.rb @@ -46,13 +46,25 @@ end end - describe "configuration" do - around do |example| - original = ActionSubscriber.config.queue_type - example.run - ActionSubscriber.config.queue_type = original + describe ".durable?" do + it "honors the request for types that can be either" do + expect(described_class.durable?(:classic, true)).to eq(true) + expect(described_class.durable?(:classic, false)).to eq(false) + expect(described_class.durable?(nil, false)).to eq(false) end + it "forces durability for types that only exist as durable queues" do + expect(described_class.durable?(:quorum, false)).to eq(true) + expect(described_class.durable?(:stream, false)).to eq(true) + end + + it "always answers with a boolean" do + expect(described_class.durable?(nil, nil)).to eq(false) + end + end + + # These examples assign the global setting; :as_config puts it back. + describe "configuration", :as_config => { :queue_type => nil } do it "defaults to nil" do expect(ActionSubscriber.config.queue_type).to be_nil end diff --git a/spec/lib/action_subscriber/router_spec.rb b/spec/lib/action_subscriber/router_spec.rb index ac35676..a2fa84d 100644 --- a/spec/lib/action_subscriber/router_spec.rb +++ b/spec/lib/action_subscriber/router_spec.rb @@ -118,6 +118,81 @@ class FakeSubscriber; end expect(routes.first.queue).to eq("alice.fake.foo") end + describe "durability" do + context "with a global durable setting", :as_config => { :durable => true } do + it "is inherited by a route that does not name durability" do + routes = described_class.draw_routes do + route FakeSubscriber, :foo + end + + expect(routes.first.durable).to eq(true) + end + + # The distinction Router::DEFAULT_SETTINGS used to erase: an unspecified route has + # to be told apart from one that explicitly asked for false. + it "can be opted out of per route" do + routes = described_class.draw_routes do + route FakeSubscriber, :foo, :durable => false + end + + expect(routes.first.durable).to eq(false) + end + end + + context "with a subscriber that declares durability" do + class DurableDeclaringSubscriber < ::ActionSubscriber::Base + durable true + def foo; end + end + + class TransientDeclaringSubscriber < ::ActionSubscriber::Base + durable false + def foo; end + end + + it "beats the global configuration" do + routes = described_class.draw_routes do + route DurableDeclaringSubscriber, :foo + end + + expect(routes.first.durable).to eq(true) + end + + it "applies to default_routes_for as well" do + routes = described_class.draw_routes do + default_routes_for DurableDeclaringSubscriber + end + + expect(routes.first.durable).to eq(true) + end + + it "loses to an explicit route option" do + routes = described_class.draw_routes do + route DurableDeclaringSubscriber, :foo, :durable => false + end + + expect(routes.first.durable).to eq(false) + end + + it "can pin a subscriber transient against a global durable setting", :as_config => { :durable => true } do + routes = described_class.draw_routes do + route TransientDeclaringSubscriber, :foo + end + + expect(routes.first.durable).to eq(false) + end + + # Quorum queues only exist as durable queues, so this is not overridable. + it "cannot make a quorum route transient" do + routes = described_class.draw_routes do + route TransientDeclaringSubscriber, :foo, :queue_type => :quorum + end + + expect(routes.first.durable).to eq(true) + end + end + end + it "can specify a prefetch value" do routes = described_class.draw_routes do route FakeSubscriber, :foo, :acknowledgements => true, :prefetch => 10 diff --git a/spec/support/sample_config.yml b/spec/support/sample_config.yml index b156fc4..7e9d4d3 100644 --- a/spec/support/sample_config.yml +++ b/spec/support/sample_config.yml @@ -1,8 +1,13 @@ +# Every environment carries the same values: the suite does not set RAILS_ENV locally +# (so it reads "development") while CI does set it to "test". development: password: <%= "WAT" %> + durable: true test: password: <%= "WAT" %> + durable: true production: password: <%= "WAT" %> + durable: true From 8e507cdadadc6bf2c17c9e6b2d78a71605e8ac15 Mon Sep 17 00:00:00 2001 From: John Bolliger Date: Wed, 12 Aug 2026 13:54:04 -0700 Subject: [PATCH 09/15] Fix retry queue declaration under quorum and durable settings MessageRetry declared its retry queues without passing :durable. On JRuby that was harmless, because march_hare forces quorum and stream queues durable internally -- but bunny does not, so on MRI a config.queue_type of :quorum made every retry declaration fail with PRECONDITION_FAILED - invalid property 'non-durable' for queue Retries were therefore broken on MRI for any deployment using quorum queues, while JRuby worked, because march_hare was silently compensating. Retry queues now derive durability through QueueType.durable?, so they follow config.durable as well. Without that, a deployment that set config.durable to get onto RabbitMQ 4.x would still fall over the first time a message was retried -- and because this path reuses env.channel rather than opening its own the way setup_queue does, the refused declaration takes the connection down and surfaces as a cascade of unrelated failures. Co-Authored-By: Claude Opus 5 (1M context) --- lib/action_subscriber/message_retry.rb | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/lib/action_subscriber/message_retry.rb b/lib/action_subscriber/message_retry.rb index 6a81fed..ac5890e 100644 --- a/lib/action_subscriber/message_retry.rb +++ b/lib/action_subscriber/message_retry.rb @@ -51,8 +51,12 @@ def self.with_exchange(env, ttl, retry_queue_name) channel.confirm_select # an empty string is the default exchange [see bunny docs](http://rubybunny.info/articles/exchanges.html#default_exchange) exchange = channel.topic("") - queue_type = ::ActionSubscriber::QueueType.driver_option(::ActionSubscriber.config.queue_type) - queue = channel.queue(retry_queue_name, :type => queue_type, :arguments => {"x-dead-letter-exchange" => "", "x-message-ttl" => ttl, "x-dead-letter-routing-key" => env.queue}) + # Retry queues follow the global settings, since they are declared here rather + # than drawn as routes. Both matter: a non-durable quorum queue is refused + # outright, and a transient retry queue cannot be declared at all on RabbitMQ 4.x. + queue_type = ::ActionSubscriber.config.queue_type + durable = ::ActionSubscriber::QueueType.durable?(queue_type, ::ActionSubscriber.config.durable) + channel.queue(retry_queue_name, :durable => durable, :type => ::ActionSubscriber::QueueType.driver_option(queue_type), :arguments => {"x-dead-letter-exchange" => "", "x-message-ttl" => ttl, "x-dead-letter-routing-key" => env.queue}) yield(exchange) channel.wait_for_confirms end From 76fcf7c947d1d9fff668538dcaeed226e31bf2ad Mon Sep 17 00:00:00 2001 From: John Bolliger Date: Wed, 12 Aug 2026 13:54:18 -0700 Subject: [PATCH 10/15] Let the integration suite target any broker and configuration Groundwork for testing against more than one RabbitMQ series. The suite previously assumed localhost:5672 and the library defaults. * RabbitMQTestHelper reads host, ports, credentials and vhost from the environment, so two brokers can run side by side and the suite can be pointed at either. spec_helper configures both ActionSubscriber and ActivePublisher from it. The port has to travel inside the host entry: march_hare builds its address list from :hosts alone, where a bare hostname means 5672 no matter what :port says. * ACTION_SUBSCRIBER_QUEUE_TYPE and ACTION_SUBSCRIBER_DURABLE override the configuration for integration examples only, so the unit specs still assert the real defaults. * with_action_subscriber_config and the :as_config metadata replace the save-mutate-restore block that would otherwise be written out at every site. It restores in an ensure -- a leaked setting reappears later as a PRECONDITION_FAILED from an unrelated spec, because queue type and durability are fixed when a queue is declared. * the helper reads queues back over the management API, which is the only way to see what a declaration actually produced when x-queue-type was left off the wire, and it declares out of band on its own connection so a refusal cannot poison the subscribers'. * the suite clears the vhost at startup, for the same fixed-at-declaration reason. consumer_cancellation already did this mid-suite; it now calls the shared helper. The helper deliberately opens a connection per call. Memoizing one was tried and reverted: several specs provoke a connection-level refusal on 4.x, bunny does not reliably recover from that, and reuse produced order-dependent failures and hangs. Co-Authored-By: Claude Opus 5 (1M context) --- spec/integration/automatic_reconnect_spec.rb | 3 +- .../integration/consumer_cancellation_spec.rb | 10 +- spec/spec_helper.rb | 64 ++++- spec/support/rabbitmq.rb | 243 ++++++++++++++++-- 4 files changed, 292 insertions(+), 28 deletions(-) diff --git a/spec/integration/automatic_reconnect_spec.rb b/spec/integration/automatic_reconnect_spec.rb index f19cb9b..db3a77d 100644 --- a/spec/integration/automatic_reconnect_spec.rb +++ b/spec/integration/automatic_reconnect_spec.rb @@ -1,4 +1,3 @@ -require "rabbitmq/http/client" class GusSubscriber < ActionSubscriber::Base def spoke @@ -12,7 +11,7 @@ def spoke default_routes_for GusSubscriber end end - let(:http_client) { RabbitMQ::HTTP::Client.new("http://127.0.0.1:15672") } + let(:http_client) { RabbitMQTestHelper.http_client } let(:subscriber) { GusSubscriber } it "reconnects when a connection drops" do diff --git a/spec/integration/consumer_cancellation_spec.rb b/spec/integration/consumer_cancellation_spec.rb index eb25aa3..104f437 100644 --- a/spec/integration/consumer_cancellation_spec.rb +++ b/spec/integration/consumer_cancellation_spec.rb @@ -1,6 +1,3 @@ -require "spec_helper" -require "rabbitmq/http/client" - class YoloSubscriber < ActionSubscriber::Base def created $messages << payload @@ -13,7 +10,6 @@ def created default_routes_for ::YoloSubscriber end end - let(:http_client) { ::RabbitMQ::HTTP::Client.new("http://127.0.0.1:15672") } let(:subscriber) { ::YoloSubscriber } it "resubscribes on cancellation" do @@ -115,9 +111,9 @@ def rabbit_consumers route_set.try(:bunny_consumers) || route_set.try(:march_hare_consumers) end + # Deleting the queues out from under the consumers is how this spec triggers the + # cancellation it is testing. def delete_all_queues! - http_client.list_queues.each do |queue| - http_client.delete_queue(queue.vhost, queue.name) - end + RabbitMQTestHelper.delete_all_queues! end end diff --git a/spec/spec_helper.rb b/spec/spec_helper.rb index 5e66089..8c24ead 100644 --- a/spec/spec_helper.rb +++ b/spec/spec_helper.rb @@ -26,6 +26,35 @@ ::ActionSubscriber::Logging.initialize_logger(nil) ::ActionSubscriber.setup_default_threadpool! +# Point the publisher and the subscriber at the same broker the helper talks to, so a +# local run can target something other than localhost:5672 -- e.g. two containers on +# 5673/5674 when checking a change against both RabbitMQ series at once. +# +# Both gems pass :hosts through to the driver, and march_hare builds its address list +# from :hosts alone -- a bare hostname there means port 5672 no matter what :port says. +# So the port has to travel in the host entry itself. +SPEC_RABBITMQ_ADDRESS = "#{RabbitMQTestHelper.host}:#{RabbitMQTestHelper.port}".freeze + +[::ActionSubscriber, ::ActivePublisher].each do |gem_module| + gem_module.configure do |config| + config.host = RabbitMQTestHelper.host + config.port = RabbitMQTestHelper.port + config.hosts = [SPEC_RABBITMQ_ADDRESS] + end +end + +# Lets CI run the whole integration suite against settings other than the defaults. The +# 4.x jobs need one of these, because RabbitMQ 4.x denies the transient queues a default +# route declares -- `quorum` forces durability via the queue type, `durable` sets it +# directly. See spec/support/rabbitmq.rb. +# +# Applied per integration example rather than globally, so they cannot reach the unit +# specs that assert what the *defaults* are. +SPEC_CONFIG_OVERRIDES = { + :queue_type => RabbitMQTestHelper.env("ACTION_SUBSCRIBER_QUEUE_TYPE"), + :durable => RabbitMQTestHelper.env("ACTION_SUBSCRIBER_DURABLE") { |value| value == "true" }, +}.reject { |_setting, value| value.nil? }.freeze + RSpec.configure do |config| config.mock_with :rspec do |mocks| mocks.verify_partial_doubles = true @@ -34,7 +63,26 @@ # Fail fast with a clear message (rather than a flurry of Bunny reconnect warnings) # if the broker isn't up yet when the integration suite starts. config.before(:suite) do - RabbitMQTestHelper.wait_for_rabbitmq! if RSpec.world.filtered_examples.values.flatten.any? { |ex| ex.metadata[:integration] } + next unless RSpec.world.filtered_examples.values.flatten.any? { |ex| ex.metadata[:integration] } + + RabbitMQTestHelper.wait_for_rabbitmq! + # Start from a clean slate: a queue left behind by a previous run under a different + # ACTION_SUBSCRIBER_QUEUE_TYPE would fail every redeclaration with + # PRECONDITION_FAILED, because type and durability are fixed at declaration. + RabbitMQTestHelper.delete_all_queues! + end + + # An around hook so it wraps the before hook below -- routes read the config when they + # are drawn. Examples that need particular settings use :as_config, which nests inside + # this one and therefore wins. + config.around(:each, :integration => true) do |example| + with_action_subscriber_config(SPEC_CONFIG_OVERRIDES) { example.run } + end + + # Opt in from any example or group with, e.g.: + # describe "...", :as_config => { :queue_type => nil, :durable => true } do + config.around(:each) do |example| + with_action_subscriber_config(example.metadata[:as_config] || {}) { example.run } end config.before(:each, :integration => true) do @@ -53,6 +101,20 @@ end end +# Set ActionSubscriber configuration for the duration of the block and put it back +# afterwards, even if the block raises -- a leaked setting reappears as a +# PRECONDITION_FAILED from some unrelated spec, since queue type and durability are +# fixed when a queue is declared. +def with_action_subscriber_config(settings) + return yield if settings.empty? + + original = settings.keys.map { |setting| [setting, ::ActionSubscriber.config.public_send(setting)] } + settings.each { |setting, value| ::ActionSubscriber.config.public_send("#{setting}=", value) } + yield +ensure + original.each { |setting, value| ::ActionSubscriber.config.public_send("#{setting}=", value) } if original +end + def verify_expectation_within(number_of_seconds, check_every = 0.02) waiting_since = ::Time.now begin diff --git a/spec/support/rabbitmq.rb b/spec/support/rabbitmq.rb index 98353a0..a791f2b 100644 --- a/spec/support/rabbitmq.rb +++ b/spec/support/rabbitmq.rb @@ -1,52 +1,259 @@ require "socket" +require "rabbitmq/http/client" # Helpers for running the suite against a real RabbitMQ broker. # # The integration specs talk to a live broker (the same approach used in CI, where a # `rabbitmq` service container is started alongside the test job). Locally you can point -# at any running broker via RABBITMQ_URL / the standard host+port; by default we assume -# localhost:5672. +# at any running broker via the standard host+port; by default we assume localhost:5672 +# with the management plugin on 15672. # -# NOTE: action_subscriber defaults to non-durable ("transient") queues. RabbitMQ 4.x -# denies transient non-exclusive queues by default, so a 4.x broker used for the suite -# must permit the deprecated feature: +# CI runs the suite against more than one broker series (see .circleci/config.yml). The +# two series do not accept the same queue declarations, so specs that care branch on the +# helpers here rather than on a hardcoded version: # -# # rabbitmq.conf -# deprecated_features.permit.transient_nonexcl_queues = true +# * RabbitMQ 3.x has `transient_nonexcl_queues` in the `permitted_by_default` +# deprecation phase, so action_subscriber's default (non-durable) route declares +# fine. +# * RabbitMQ 4.x moved it to `denied_by_default`. A non-durable, non-exclusive queue +# declaration is refused with a *connection*-level 541 INTERNAL_ERROR, which tears +# down the whole connection rather than just the channel. To run the default route +# shape against 4.x a broker has to opt back in: # -# The rabbitmq:3.12 image used in CI still allows them out of the box. See the phantom -# queue triage doc for why the production recommendation is to move to durable topology. +# # rabbitmq.conf +# deprecated_features.permit.transient_nonexcl_queues = true +# +# CI instead runs the 4.x job with ACTION_SUBSCRIBER_QUEUE_TYPE=quorum, since quorum +# queues are always durable and so sidestep the deprecated feature entirely. module RabbitMQTestHelper module_function + # Every environment variable the suite honors goes through here. Blank is treated as + # unset: a CircleCI job parameter that defaults to "" still reaches the environment as + # an empty string. Pass a block to coerce a value that is actually present. + def env(name, default = nil) + value = ENV[name].to_s.strip + return default if value.empty? + block_given? ? yield(value) : value + end + def host - ENV.fetch("RABBITMQ_HOST", "localhost") + env("RABBITMQ_HOST", "127.0.0.1") end def port - Integer(ENV.fetch("RABBITMQ_PORT", "5672")) + env("RABBITMQ_PORT", 5672) { |value| Integer(value) } + end + + def management_port + env("RABBITMQ_MANAGEMENT_PORT", 15672) { |value| Integer(value) } + end + + def username + env("RABBITMQ_USERNAME", "guest") + end + + def password + env("RABBITMQ_PASSWORD", "guest") + end + + def vhost + env("RABBITMQ_VHOST", "/") + end + + def http_client + @http_client ||= ::RabbitMQ::HTTP::Client.new( + "http://#{host}:#{management_port}", + :username => username, + :password => password + ) + end + + # e.g. "3.13.7" or "4.3.4" + def broker_version + @broker_version ||= http_client.overview.rabbitmq_version.to_s + end + + def broker_major + @broker_major ||= Integer(broker_version.split(".").first) + end + + # The vhost's default_queue_type, which is what the broker resolves an absent + # x-queue-type to. 3.x reports it as the string "undefined" when unset and 4.x reports + # it explicitly; unset behaves as "classic" either way. + def default_queue_type + @default_queue_type ||= begin + record = http_client.vhost_info(vhost) + type = record.respond_to?(:default_queue_type) ? record.default_queue_type.to_s : "" + type.empty? || type == "undefined" ? "classic" : type + end + end + + # Probed rather than inferred from the version, because a 4.x broker can permit the + # feature back on and a 3.x broker can deny it. Memoized -- the probe costs a + # connection, and on 4.x it costs a *failed* one. + def transient_nonexcl_queues_permitted? + return @transient_nonexcl_queues_permitted if defined?(@transient_nonexcl_queues_permitted) + + name = "action_subscriber.spec.transient_probe" + @transient_nonexcl_queues_permitted = + begin + declare_queue!(name, :durable => false, :type => "classic") + delete_queue!(name) + true + rescue ::StandardError + false + end + end + + ## + # Out-of-band queue management + # + # These deliberately use their own connection rather than + # ActionSubscriber::RabbitConnection, so that a declaration the broker refuses cannot + # poison the connection the subscribers are using. On 4.x a denied transient + # declaration closes the whole connection, not just the channel. + # + + # One short-lived connection per call. Reusing a memoized one is tempting -- the suite + # makes ~40 of them per run -- but it was tried and reverted: several specs here + # deliberately provoke a connection-level refusal (4.x answers a denied transient queue + # with a 541), and bunny does not reliably recover a connection in that state. Reuse + # produced order-dependent failures and, twice, a suite that hung past 700s. The + # handshakes are cheaper than the flakiness. + def with_raw_channel + connection = build_raw_connection + yield(connection.create_channel) + ensure + begin + connection.close if connection + rescue ::StandardError + nil + end end - # Block until the broker's AMQP port accepts a TCP connection, or raise after `timeout` - # seconds. Keeps the suite from failing with confusing connection errors when the broker - # is still booting (common in CI service containers). - def wait_for_rabbitmq!(timeout: Integer(ENV.fetch("RABBITMQ_WAIT_TIMEOUT", "30"))) + def declare_queue!(name, options = {}) + with_raw_channel { |channel| declare(channel, name, options) } + end + + # Declares the queue a route describes, using the same option mapping the drivers' + # setup_queue uses. Keeping that mapping in one place stops specs from asserting + # against a stale copy of it. Out-of-band on purpose -- unlike RouteSet#setup_queue + # this cannot take the subscribers' connection down with it. + # + # `name` is overridable because a spec often wants a route's *settings* without + # touching the queue its own subscription already declared. + def declare_route_queue!(route, name = route.queue) + declare_queue!(name, :durable => route.durable, :type => route.driver_queue_type) + end + + def delete_queue!(name) + with_raw_channel { |channel| channel.queue_delete(name) } + rescue ::StandardError + nil + end + + # Deletes every queue in the vhost. The suite needs this because a queue's type and + # durability are fixed at declaration: a queue left behind by a previous run under a + # different queue type fails every later redeclaration with PRECONDITION_FAILED. Not + # all of the suite's queues are named after APP_NAME (some specs name their own), so + # there is nothing narrower to key on -- and spec/integration/consumer_cancellation + # already clears the whole vhost mid-suite. Point the suite at a broker you own. + def delete_all_queues! + http_client.list_queues(vhost).each do |queue| + http_client.delete_queue(queue.vhost, queue.name) + end + rescue ::StandardError + nil + end + + # The broker's view of a queue -- notably `type` and `durable`, which is the only way + # to tell what a declaration actually produced when x-queue-type was left off the wire. + def queue_info(name) + http_client.queue_info(vhost, name) + end + + def queue_type_of(name) + queue_info(name).type.to_s + end + + def queue_durable?(name) + queue_info(name).durable + end + + # Block until the broker is reachable, or raise after `timeout` seconds. Keeps the + # suite from failing with confusing connection errors when the broker is still booting + # (common in CI service containers). + # + # Waits on the management API as well as AMQP: the suite reads queue types back over + # HTTP, and the management plugin finishes starting after the AMQP listener does. No + # separate wait on the management *port* -- a closed one surfaces as ECONNREFUSED from + # the overview call, which retries against the same deadline. + def wait_for_rabbitmq!(timeout: env("RABBITMQ_WAIT_TIMEOUT", 60) { |value| Integer(value) }) deadline = ::Time.now + timeout + + wait_until!(deadline, "AMQP port #{host}:#{port} to accept a connection") do + ::Socket.tcp(host, port, connect_timeout: 1, &:close) + true + end + + wait_until!(deadline, "management API at #{host}:#{management_port} to answer") do + # overview is only served once the management plugin is fully up. + @broker_version = nil + !broker_version.empty? + end + end + + # Retry the block until it returns truthy or the deadline passes, then raise naming + # what we were waiting for and why the last attempt failed. + def wait_until!(deadline, description) last_error = nil loop do begin - ::Socket.tcp(host, port, connect_timeout: 1) { |sock| sock.close } - return true + return true if yield rescue ::StandardError => e last_error = e end if ::Time.now >= deadline - raise "RabbitMQ was not reachable at #{host}:#{port} within #{timeout}s " \ + raise "Timed out waiting for #{description} " \ "(last error: #{last_error.class}: #{last_error.message}). " \ "Start a broker (see spec/support/rabbitmq.rb) before running the integration suite." end sleep 0.5 end end + + # Several specs here declare queues the broker is expected to refuse, and a refusal at + # the connection level (4.x answers a denied transient queue with a 541) makes the + # driver log the resulting socket teardown with a full Java backtrace. Expected noise, + # so keep it out of the CI log. + def quiet_logger + @quiet_logger ||= ::Logger.new(::File::NULL) + end + + def build_raw_connection + if ::RUBY_PLATFORM == "java" + ::MarchHare.connect(:host => host, :port => port, :username => username, + :password => password, :vhost => vhost, + :logger => quiet_logger) + else + connection = ::Bunny.new(:host => host, :port => port, :username => username, + :password => password, :vhost => vhost, + :log_level => :fatal, :automatically_recover => false, + # Bunny does not decode the 541 a 4.x broker sends for a + # denied transient queue, it just waits out this timeout. + # Three specs deliberately trigger that, so keep it tight. + :continuation_timeout => 2_000) + connection.start + connection + end + end + + # bunny and march_hare disagree about an omitted :type -- march_hare fills in + # "classic", bunny sends nothing -- so always pass it explicitly here, the same way + # ActionSubscriber::QueueType makes the drivers agree. + def declare(channel, name, options) + channel.queue(name, { :type => nil }.merge(options)) + end end From a9169090edb7bd4add8c052266bc8b0f73b2e805 Mon Sep 17 00:00:00 2001 From: John Bolliger Date: Wed, 12 Aug 2026 13:54:31 -0700 Subject: [PATCH 11/15] Add queue type and broker compatibility integration specs Both assert what the broker actually created, read back over the management API. The driver option and the wire arguments were already covered by unit specs, and neither tells you whether the broker agreed. queue_type_spec covers the queue each route produces, and pins the regression that motivated ActionSubscriber::QueueType: subscribing to a queue somebody else declared as a durable quorum queue. Master could not do it, in three independent ways -- * on JRuby, march_hare filled in :type => "classic" for the omitted option, so the broker rejected the redeclaration on x-queue-type; * on MRI, bunny sent no x-queue-type, which the broker resolves against the vhost default and rejects the same way; * and both sent durable => false, which a quorum queue rejects on its own. broker_compatibility_spec pins the differences between the two supported series. RabbitMQ 3.x has transient_nonexcl_queues in the permitted_by_default deprecation phase; 4.x moved it to denied_by_default. Since every default route is a transient queue, the library cannot declare one on a stock 4.x broker at all -- the spec asserts that limitation rather than papering over it, along with the two settings that work around it. The examples branch on what the broker actually permits rather than on its version, so a 3.x broker with the feature denied, or a 4.x broker with it permitted, still gets a true answer. Co-Authored-By: Claude Opus 5 (1M context) --- spec/integration/broker_compatibility_spec.rb | 178 ++++++++++++++++++ spec/integration/queue_type_spec.rb | 171 +++++++++++++++++ 2 files changed, 349 insertions(+) create mode 100644 spec/integration/broker_compatibility_spec.rb create mode 100644 spec/integration/queue_type_spec.rb diff --git a/spec/integration/broker_compatibility_spec.rb b/spec/integration/broker_compatibility_spec.rb new file mode 100644 index 0000000..d4da084 --- /dev/null +++ b/spec/integration/broker_compatibility_spec.rb @@ -0,0 +1,178 @@ +class BrokerCompatibilitySubscriber < ActionSubscriber::Base + def created + $messages << payload + end +end + +class DurableCompatibilitySubscriber < ActionSubscriber::Base + durable true + + def created + $messages << payload + end +end + +# Pins the differences between the RabbitMQ series action_subscriber claims to support. +# CI runs this file against both (see .circleci/config.yml); the examples branch on what +# the broker actually permits rather than on its version, so a 3.x broker with the +# deprecated feature denied, or a 4.x broker with it permitted, still gets a true answer. +describe "Broker compatibility", :integration => true do + let(:helper) { RabbitMQTestHelper } + let(:draw_routes) do + ::ActionSubscriber.draw_routes do + route ::BrokerCompatibilitySubscriber, :created, :queue_type => :quorum + end + end + + # Guards against a CI job silently testing the wrong broker -- e.g. an image tag that + # stopped resolving to the series the job name claims. Only enforced when CI says what + # it expects. + expected_major = RabbitMQTestHelper.env("EXPECTED_RABBITMQ_MAJOR") { |value| Integer(value) } + if expected_major + it "is running against the broker series CI selected" do + expect(helper.broker_major).to eq(expected_major) + end + end + + describe "durable declarations" do + %w[classic quorum].each do |type| + it "works for a durable #{type} queue on every supported broker" do + name = "alice.compat.durable.#{type}" + helper.declare_queue!(name, :durable => true, :type => type) + info = helper.queue_info(name) + + expect(info.type).to eq(type) + expect(info.durable).to eq(true) + + helper.delete_queue!(name) + end + end + end + + # action_subscriber's routes default to :durable => false, which makes every default + # route a transient non-exclusive queue. That is `permitted_by_default` on RabbitMQ 3.x + # and `denied_by_default` on 4.x, where the broker answers with a connection-level 541 + # INTERNAL_ERROR rather than a channel-level error -- so the failure takes the whole + # connection down with it. + describe "transient non-exclusive queues (action_subscriber's default route shape)" do + let(:queue_name) { "alice.compat.transient" } + + after { helper.delete_queue!(queue_name) } + + # Declares through the same option mapping the drivers use, so these cannot pass + # against a stale copy of it. Into `queue_name` rather than the route's own queue: + # this file's subscription already declared that one, as quorum. + def declare_route!(route) + helper.declare_route_queue!(route, queue_name) + helper.queue_info(queue_name) + end + + it "declares if and only if the broker permits the deprecated feature" do + if helper.transient_nonexcl_queues_permitted? + helper.declare_queue!(queue_name, :durable => false, :type => "classic") + + expect(helper.queue_durable?(queue_name)).to eq(false) + else + expect { + helper.declare_queue!(queue_name, :durable => false, :type => "classic") + }.to raise_error(::StandardError) + end + end + + # The direct fix: config.durable, which an operator can set from the yaml file + # without touching code. Unlike :quorum this leaves the queue type alone, so it is + # the smaller change for an existing classic-queue deployment moving to 4.x. + context "with config.durable on", :as_config => { :durable => true } do + it "is sidestepped, without changing the queue type" do + durable_route = ::ActionSubscriber::Router.draw_routes do + route ::BrokerCompatibilitySubscriber, :created, :queue_type => :broker_default + end.first + + expect(durable_route.durable).to eq(true) + + info = declare_route!(durable_route) + expect(info.type).to eq(helper.default_queue_type) + expect(info.durable).to eq(true) + end + end + + # And the same via the subscriber DSL rather than the global setting. + it "is sidestepped by a subscriber declaring `durable true`" do + declared_route = ::ActionSubscriber::Router.draw_routes do + route ::DurableCompatibilitySubscriber, :created, :queue_type => :broker_default + end.first + + expect(declared_route.durable).to eq(true) + expect(declare_route!(declared_route).durable).to eq(true) + end + + # The other first-party option: :quorum forces :durable => true on the route, so it + # sidesteps the deprecated feature too. This is what the 4.x quorum CI job relies on. + it "is sidestepped by :quorum, which forces the route durable" do + # Not named `route`: a local by that name would shadow the DSL method inside the + # draw_routes block. + quorum_route = ::ActionSubscriber::Router.draw_routes do + route ::BrokerCompatibilitySubscriber, :created, :queue_type => :quorum, :durable => false + end.first + + expect(quorum_route.durable).to eq(true) + expect(declare_route!(quorum_route).type).to eq("quorum") + end + end + + # MessageRetry declares its own queues, and does it with the global config.queue_type + # rather than the route's. It has to produce a declaration the broker accepts on both + # series and on both drivers -- bunny does not force durability for quorum the way + # march_hare does, so this is the shape most likely to regress. + describe "retry queues" do + let(:queue_name) { "alice.compat.retry_target.retry_100" } + + after { helper.delete_queue!(queue_name) } + + def declare_retry_queue! + helper.with_raw_channel do |channel| + env = double(:channel => channel, :queue => "alice.compat.retry_target") + ::ActionSubscriber::MessageRetry.with_exchange(env, 100, queue_name) { |_exchange| nil } + end + end + + # Each context pins both settings, not just the one it is about: CI runs this whole + # file with one or the other turned on. + context "with quorum configured", :as_config => { :queue_type => :quorum, :durable => false } do + it "declares a durable quorum retry queue" do + declare_retry_queue! + info = helper.queue_info(queue_name) + + expect(info.type).to eq("quorum") + expect(info.durable).to eq(true) + end + end + + # Retry queues are declared here rather than drawn as routes, so they have to pick + # up config.durable on their own. Missed at first, and it only showed up as a + # cascade of unrelated failures: the refused declaration takes down the connection. + context "with config.durable on", :as_config => { :queue_type => nil, :durable => true } do + it "declares a durable retry queue of the broker's default type" do + declare_retry_queue! + info = helper.queue_info(queue_name) + + expect(info.durable).to eq(true) + expect(info.type).to eq(helper.default_queue_type) + end + end + + context "with nothing configured", :as_config => { :queue_type => nil, :durable => false } do + # Retry queues are transient by default, so they inherit the same 4.x limitation + # the default route shape has. Asserting it keeps the constraint visible instead of + # surfacing as a mystery failure the first time somebody retries a message on 4.x. + it "declares a transient retry queue only where transient queues are permitted" do + if helper.transient_nonexcl_queues_permitted? + declare_retry_queue! + expect(helper.queue_durable?(queue_name)).to eq(false) + else + expect { declare_retry_queue! }.to raise_error(::StandardError) + end + end + end + end +end diff --git a/spec/integration/queue_type_spec.rb b/spec/integration/queue_type_spec.rb new file mode 100644 index 0000000..8536780 --- /dev/null +++ b/spec/integration/queue_type_spec.rb @@ -0,0 +1,171 @@ +# What actually landed on the broker is the only thing worth asserting here: the driver +# option and the wire arguments are already covered by the unit specs, and neither tells +# you whether the broker agreed. These read the queue back over the management API. + +class QuorumQueueSubscriber < ActionSubscriber::Base + def created + $messages << payload + end +end + +class ClassicQueueSubscriber < ActionSubscriber::Base + def created + $messages << payload + end +end + +class BrokerDefaultQueueSubscriber < ActionSubscriber::Base + def created + $messages << payload + end +end + +class ConflictingTypeSubscriber < ActionSubscriber::Base + def created + $messages << payload + end +end + +describe "Queue types", :integration => true do + let(:helper) { RabbitMQTestHelper } + + # A queue of the right type is only half the claim -- it also has to carry messages. + shared_examples "a working subscription" do |routing_key, body| + it "delivers messages" do + ::ActionSubscriber.start_subscribers! + ::ActivePublisher.publish(routing_key, body, "events") + + verify_expectation_within(5.0) do + expect($messages).to eq(Set.new([body])) + end + end + end + + describe "a :quorum route" do + let(:draw_routes) do + ::ActionSubscriber.draw_routes do + route ::QuorumQueueSubscriber, :created, :queue_type => :quorum + end + end + let(:queue_name) { "alice.quorum_queue.created" } + + it "declares a durable quorum queue" do + info = helper.queue_info(queue_name) + + expect(info.type).to eq("quorum") + expect(info.durable).to eq(true) + end + + it_behaves_like "a working subscription", "quorum_queue.created", "Ohai Quorum" + end + + describe "a :quorum route declared with :durable => false" do + let(:draw_routes) do + ::ActionSubscriber.draw_routes do + route ::QuorumQueueSubscriber, :created, :queue_type => :quorum, :durable => false + end + end + + # Route forces this rather than letting the broker refuse the declaration. + it "is still durable" do + expect(helper.queue_durable?("alice.quorum_queue.created")).to eq(true) + end + end + + describe "a :classic route" do + let(:draw_routes) do + ::ActionSubscriber.draw_routes do + route ::ClassicQueueSubscriber, :created, :queue_type => :classic, :durable => true + end + end + let(:queue_name) { "alice.classic_queue.created" } + + it "declares a classic queue" do + expect(helper.queue_type_of(queue_name)).to eq("classic") + end + + it_behaves_like "a working subscription", "classic_queue.created", "Ohai Classic" + end + + describe "a :broker_default route" do + # Named explicitly rather than left to the global setting, because CI runs the whole + # suite with ACTION_SUBSCRIBER_QUEUE_TYPE set on some jobs. Durable so the + # declaration is legal on a broker that denies transient non-exclusive queues. + let(:draw_routes) do + ::ActionSubscriber.draw_routes do + route ::BrokerDefaultQueueSubscriber, :created, :queue_type => :broker_default, :durable => true + end + end + let(:queue_name) { "alice.broker_default_queue.created" } + + # Sending no x-queue-type is what lets an operator move a vhost onto quorum queues + # with a broker setting instead of a code change. On JRuby this is the whole point of + # the :type => nil option -- march_hare would otherwise pin the queue to classic. + it "gets whatever the vhost's default_queue_type is" do + expect(helper.queue_type_of(queue_name)).to eq(helper.default_queue_type) + end + + it_behaves_like "a working subscription", "broker_default_queue.created", "Ohai Default" + end + + # The regression that motivated ActionSubscriber::QueueType. Master had no way to say + # "quorum", so subscribing to a queue somebody else had declared as a durable quorum + # queue could not work: + # + # * on JRuby, march_hare filled in :type => "classic" for the omitted option and the + # broker rejected the redeclaration on x-queue-type; + # * on MRI, bunny sent no x-queue-type, which the broker resolves against the vhost's + # default_queue_type -- classic on a default vhost -- and rejected the same way; + # * and both sent durable => false, which a quorum queue rejects on its own. + describe "subscribing to a queue that already exists as a durable quorum queue" do + # A harmless route, so the suite-wide integration hook has something to set up. The + # examples below drive setup_queue directly against the conflicting queue. + let(:draw_routes) do + ::ActionSubscriber.draw_routes do + route ::QuorumQueueSubscriber, :created, :queue_type => :quorum + end + end + let(:queue_name) { "alice.conflicting_type.created" } + + before do + helper.delete_queue!(queue_name) + helper.declare_queue!(queue_name, :durable => true, :type => "quorum") + end + + after { helper.delete_queue!(queue_name) } + + def setup_queue_for(route_options) + routes = ::ActionSubscriber::Router.draw_routes do + route ::ConflictingTypeSubscriber, :created, route_options + end + ::ActionSubscriber::RouteSet.new(routes).send(:setup_queue, routes.first) + end + + it "fails when the route does not name the type (what master always did)" do + expect { + setup_queue_for(:queue_type => :broker_default, :durable => true) + }.to raise_error(/PRECONDITION_FAILED/) + end + + it "fails when the route names a conflicting type" do + expect { + setup_queue_for(:queue_type => :classic, :durable => true) + }.to raise_error(/PRECONDITION_FAILED/) + end + + it "fails on durability alone when the route is transient" do + # Reaches the broker as a durable mismatch on 3.x. On 4.x the transient + # declaration is refused before that, which is its own kind of failure. + expect { + setup_queue_for(:queue_type => :broker_default, :durable => false) + }.to raise_error(::StandardError) + end + + it "succeeds when the route names :quorum" do + queue = setup_queue_for(:queue_type => :quorum) + + expect(queue.name).to eq(queue_name) + expect(helper.queue_type_of(queue_name)).to eq("quorum") + end + end +end From c0332426d0d1b9a1aebe94b2c09245c4fba8dbe0 Mon Sep 17 00:00:00 2001 From: John Bolliger Date: Wed, 12 Aug 2026 13:54:40 -0700 Subject: [PATCH 12/15] Test against the latest RabbitMQ 3.x and 4.x The Rails matrix pinned one broker (3.12). This parameterises the broker image, the queue type and the durability setting, bumps the Rails matrix to 3.13, and adds a broker_compatibility_matrix that pins the broker axis against one Rails version so the two do not multiply out. Both drivers are covered deliberately: bunny and march_hare disagree about queue declaration in ways only a live broker shows. march_hare fills in x-queue-type: classic for an omitted :type where bunny sends nothing, and march_hare forces quorum queues durable where bunny does not. The 4.x jobs have to configure something, because a default route is a transient queue and 4.x denies those. Both documented ways out get a job: durable classic queues, which is the smaller change for an existing deployment, and quorum queues, which also exercise forced durability and quorum retry queues. expected_rabbitmq_major is asserted by the suite, so an image tag that stops resolving to the series a job name claims fails loudly instead of passing as a duplicate of another job. Co-Authored-By: Claude Opus 5 (1M context) --- .circleci/config.yml | 107 +++++++++++++++++++++++++++++++++++++++---- 1 file changed, 98 insertions(+), 9 deletions(-) diff --git a/.circleci/config.yml b/.circleci/config.yml index 50ba9dd..64bae00 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -9,6 +9,34 @@ jobs: gemfile: type: string description: "The Appraisal gemfile (Rails version) to test against" + rabbitmq_image: + type: string + default: "rabbitmq:3.13-management" + description: "The RabbitMQ Docker image to run the integration suite against" + expected_rabbitmq_major: + type: string + default: "3" + description: > + The RabbitMQ major version this job believes it selected. The suite asserts it, + so an image tag that quietly stops resolving to that series fails loudly + instead of passing as a duplicate of another job. + queue_type: + type: string + default: "" + description: > + Sets config.queue_type for the whole integration suite. Empty means the broker + default. + durable: + type: string + default: "" + description: > + Sets config.durable for the whole integration suite. Empty leaves the default + (false). One of this or queue_type=quorum is required on RabbitMQ 4.x: + action_subscriber's routes default to :durable => false, and 4.x denies + transient non-exclusive queues, so the default route shape cannot be declared + there at all. Both are covered below, since they are the two fixes the README + offers and they exercise different code paths -- durable keeps the broker's + queue type, quorum changes it. docker: # 1. The Primary Container (where your code actually runs) @@ -19,16 +47,17 @@ jobs: # Select the Rails version under test via the Appraisal gemfile. BUNDLE_GEMFILE: << parameters.gemfile >> # Tell the suite where to find RabbitMQ. - RABBITMQ_URL: "amqp://guest:guest@localhost:5672" RABBITMQ_HOST: "localhost" RABBITMQ_PORT: "5672" + RABBITMQ_MANAGEMENT_PORT: "15672" + ACTION_SUBSCRIBER_QUEUE_TYPE: << parameters.queue_type >> + ACTION_SUBSCRIBER_DURABLE: << parameters.durable >> + EXPECTED_RABBITMQ_MAJOR: << parameters.expected_rabbitmq_major >> - # 2. The Service Container (runs in the background). - # NOTE: action_subscriber declares transient (non-durable) queues by default, - # which RabbitMQ 4.x denies out of the box. rabbitmq:3.12 still permits them. - # If/when moving to a 4.x image, permit the deprecated feature via config: - # deprecated_features.permit.transient_nonexcl_queues = true - - image: rabbitmq:3.12-management + # 2. The Service Container (runs in the background). Management plugin included -- + # the suite reads queue types and durability back over the HTTP API, since that is + # the only way to see what a declaration actually produced. + - image: << parameters.rabbitmq_image >> working_directory: ~/project @@ -87,7 +116,9 @@ jobs: key: v4-gems-<< parameters.docker_image >>-<< parameters.gemfile >>-{{ checksum "action_subscriber.gemspec" }}-{{ checksum "Appraisals" }} # Wait for RabbitMQ to be ready before running tests. - # Service containers can sometimes take a few seconds to boot up. + # Service containers can sometimes take a few seconds to boot up. The management + # plugin comes up after the AMQP listener, and the suite reads queue types back + # over the HTTP API, so wait for both. - run: name: Wait for RabbitMQ command: | @@ -101,7 +132,10 @@ jobs: while ! nc -z localhost 5672; do sleep 1 done - echo "RabbitMQ is ready!" + while ! nc -z localhost 15672; do + sleep 1 + done + echo "RabbitMQ is ready (<< parameters.rabbitmq_image >>)!" - run: name: Run Tests @@ -141,3 +175,58 @@ workflows: gemfile: - "gemfiles/rails_8.0.gemfile" - "gemfiles/rails_8.1.gemfile" + + # Broker compatibility. The matrix above pins the Rails axis against one broker; this + # one pins the broker axis against one Rails version, so the two do not multiply out. + # + # Both drivers are covered deliberately: bunny and march_hare disagree about queue + # declaration in ways only a live broker shows. march_hare fills in + # x-queue-type: classic for an omitted :type where bunny sends nothing, and march_hare + # forces quorum queues durable where bunny does not. + broker_compatibility_matrix: + jobs: + # RabbitMQ 3.x is already covered with broker-default queues by the Rails matrix. + # This adds the quorum path on 3.x, so a quorum deployment is tested on both series + # rather than only on the one that requires it. + # Every value goes through matrix.parameters, including the ones that only ever + # take a single value, so the whole invocation uses one mechanism. + - build_and_test: + name: broker-3.13-quorum-<< matrix.docker_image >> + matrix: + parameters: + docker_image: + - "cimg/ruby:3.4" + - "jruby:10.0" + gemfile: ["gemfiles/rails_8.1.gemfile"] + rabbitmq_image: ["rabbitmq:3.13-management"] + expected_rabbitmq_major: ["3"] + queue_type: ["quorum"] + + # RabbitMQ 4.x, durable classic queues. This is the smaller of the two upgrade + # paths for an existing deployment -- the queue type is untouched, only durability + # changes -- so it is the one most users will take. + - build_and_test: + name: broker-4.x-durable-<< matrix.docker_image >> + matrix: + parameters: + docker_image: + - "cimg/ruby:3.4" + - "jruby:10.0" + gemfile: ["gemfiles/rails_8.1.gemfile"] + rabbitmq_image: ["rabbitmq:4-management"] + expected_rabbitmq_major: ["4"] + durable: ["true"] + + # RabbitMQ 4.x, quorum queues. The other path, and the one that also exercises + # quorum-specific behavior (forced durability, quorum retry queues). + - build_and_test: + name: broker-4.x-quorum-<< matrix.docker_image >> + matrix: + parameters: + docker_image: + - "cimg/ruby:3.4" + - "jruby:10.0" + gemfile: ["gemfiles/rails_8.1.gemfile"] + rabbitmq_image: ["rabbitmq:4-management"] + expected_rabbitmq_major: ["4"] + queue_type: ["quorum"] From 7017a6b7f24ab54ddc2035b31b1e9e5aee81dffb Mon Sep 17 00:00:00 2001 From: John Bolliger Date: Wed, 12 Aug 2026 13:54:50 -0700 Subject: [PATCH 13/15] Document durability and RabbitMQ version support Adds a Durability section covering the new setting and its precedence, and a Supported RabbitMQ Versions section covering the 3.x/4.x split. The 4.x limitation is worth stating plainly: routes default to :durable => false, RabbitMQ 4.x moved transient_nonexcl_queues to denied_by_default, so a stock 4.x broker refuses every default route. The failure is hard to read on both drivers -- it is a connection-level 541 that neither decodes, so march_hare reports "Unknown reply code: 541" and bunny simply blocks until continuation_timeout. Also documents that durability, unlike queue type, cannot be deferred to the broker: it is a field in the declaration frame rather than an optional argument, so a client always states a value. Co-Authored-By: Claude Opus 5 (1M context) --- README.md | 141 ++++++++++++++++++++++++++++++++++++++++++++++++--- changelog.md | 51 +++++++++++++++++++ 2 files changed, 186 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index 09ea620..c05a7d0 100644 --- a/README.md +++ b/README.md @@ -45,6 +45,55 @@ end That will give you a similar behavior to the old `--mode=pop` where messages polled from the server, but with reduced latency. +Durability +---------- + +Queues default to transient. Set durability globally, per subscriber, or per +route: + +```yaml +# config/action_subscriber.yml +production: + durable: true +``` + +```ruby +::ActionSubscriber.configure do |config| + config.durable = true +end + +class UserSubscriber < ::ActionSubscriber::Base + exchange :events + durable true # every queue drawn from this subscriber +end + +::ActionSubscriber.draw_routes do + route AuditSubscriber, :created, :durable => true # just this route +end +``` + +The most specific setting wins: a route's `:durable` option beats the +subscriber's `durable` declaration, which beats `config.durable`. `durable false` +on a subscriber or route pins it transient even when the global setting is on. +`:quorum` and `:stream` queues override all of it — those types only exist as +durable queues. + +> Note: durability is fixed when a queue is created. Unlike most settings it is +> not something the client can defer to the broker — `durable` is a field in the +> declaration frame, so a client always states a value, and RabbitMQ rejects a +> redeclaration that disagrees with the existing queue: +> +> ``` +> PRECONDITION_FAILED - inequivalent arg 'durable' for queue 'x' in vhost '/': +> received 'false' but current is 'true' +> ``` +> +> Turning this on for queues that already exist means deleting them first. + +Note that retry queues follow `config.durable` — the global setting only, since +they are declared by `ActionSubscriber::MessageRetry` rather than drawn as +routes. + Queue Types ----------- @@ -114,17 +163,63 @@ If any are non-classic, set `config.queue_type = :classic` before rolling out. ### Known limitation: retry queues `ActionSubscriber::MessageRetry` declares its `*.retry_*` queues using the -**global** `config.queue_type`, not the type of the route that produced the -message. A route that opts into `:quorum` while the global setting is left at -the default will dead-letter into a retry queue of a different type. +**global** `config.queue_type` and `config.durable`, not the settings of the +route that produced the message. A route that opts into `:quorum` while the +global setting is left at the default will dead-letter into a retry queue of a +different type, and the same goes for a route that sets `:durable => true` on +its own. -If you rely on per-route queue types and on retries, set `config.queue_type` to -match rather than setting it per route. +If you rely on per-route queue types or durability and on retries, set the +global settings to match rather than setting them per route. Note also that retry queues carry `x-message-ttl` and `x-dead-letter-exchange`, which streams do not support — so a global `config.queue_type = :stream` will make every retry declaration fail. +Retry queues are transient unless the configured type forces otherwise, which +means they carry the same RabbitMQ 4.x limitation as the default route shape — +see below. + +Supported RabbitMQ Versions +--------------------------- + +ActionSubscriber is tested against the latest RabbitMQ 3.x and 4.x on both +drivers. The two series do not accept the same queue declarations. + +**Routes default to `:durable => false`, which makes every default route a +transient non-exclusive queue.** RabbitMQ moved that from +`permitted_by_default` to `denied_by_default` in 4.0: + +| | RabbitMQ 3.x | RabbitMQ 4.x | +| --- | --- | --- | +| `transient_nonexcl_queues` | `permitted_by_default` | `denied_by_default` | + +So on a stock 4.x broker a default route cannot be declared at all. The broker +answers with a *connection*-level `541 INTERNAL_ERROR`, which takes down the +whole connection rather than just the channel — and neither driver decodes it +into something readable. march_hare reports `Unknown reply code: 541` and bunny +simply blocks until `continuation_timeout` and raises `Timeout::Error`. + +There are two ways to run on 4.x: + +1. **Declare durable queues** (recommended) — `config.durable = true`, which + can be set from the YAML config with no code change, or + `config.queue_type = :quorum`, which forces durability as a side effect of + changing the queue type. CI runs both. See "Durability" below. +2. **Permit the deprecated feature on the broker**, which keeps the current + transient topology working for now but not past its removal: + + ``` + # rabbitmq.conf + deprecated_features.permit.transient_nonexcl_queues = true + ``` + +Check where a broker currently stands with: + +``` +rabbitmqctl list_deprecated_features +``` + Supported Message Types ----------------- ActionSubscriber support JSON and plain text out of the box, but you can easily @@ -209,6 +304,7 @@ Other configuration options include : * config.connection_reaping_interval - Connection reaping interval when using a project ActiveRecord * config.connection_reaping_timeout_interval - Connection reaping timeout interval when using a project ActiveRecord * config.default_exchange - set the default exchange that your queues will use, using the default RabbitMQ exchange is not recommended +* config.durable - default durability for all routes (default false). Required on RabbitMQ 4.x, which refuses transient queues * config.error_handler - handle error like you want to handle them! * config.heartbeat - number of seconds between hearbeats (default 5) [see bunny documentation for more details](http://rubybunny.info/articles/connecting.html) * config.hosts - an array of hostnames in your cluster (ie `["rabbit1.myapp.com", "rabbit2.myapp.com"]`) @@ -312,7 +408,7 @@ Development If you want to work on `action_subscriber` you will need to have a rabbitmq instance running locally on port 5672 with a management plugin enabled on port 15672. Usually the easiest way to accomplish this is to use docker and run the command: ``` -$ docker run --net=host --rm=true --hostname diagon --name rabbit rabbitmq:3.6.6-management +$ docker run -d --rm --name rabbit -p 5672:5672 -p 15672:15672 rabbitmq:3.13-management ``` Now that rabbitmq is running you can clone this project and run: @@ -344,3 +440,36 @@ $ bundle exec appraisal rspec Re-run `appraisal generate` after editing `Appraisals`. Note that Rails 7.2 requires Ruby >= 3.1 and Rails 8.0/8.1 require Ruby >= 3.2, so those gemfiles will not resolve on older interpreters. + +### Testing against multiple RabbitMQ versions + +The suite reads its broker location from the environment, so you can run two +brokers side by side and point it at either: + +``` +$ docker run -d --rm --name rabbit3 -p 5673:5672 -p 15673:15672 rabbitmq:3.13-management +$ docker run -d --rm --name rabbit4 -p 5674:5672 -p 15674:15672 rabbitmq:4-management +``` + +``` +$ RABBITMQ_PORT=5673 RABBITMQ_MANAGEMENT_PORT=15673 bundle exec rspec + +# 4.x denies transient non-exclusive queues, so the suite needs durable queues +# there -- either way works. See "Supported RabbitMQ Versions" above. +$ RABBITMQ_PORT=5674 RABBITMQ_MANAGEMENT_PORT=15674 \ + ACTION_SUBSCRIBER_DURABLE=true bundle exec rspec +$ RABBITMQ_PORT=5674 RABBITMQ_MANAGEMENT_PORT=15674 \ + ACTION_SUBSCRIBER_QUEUE_TYPE=quorum bundle exec rspec +``` + +`ACTION_SUBSCRIBER_QUEUE_TYPE` and `ACTION_SUBSCRIBER_DURABLE` set +`config.queue_type` and `config.durable` for the integration examples only, so +the unit specs still assert the real defaults. +`EXPECTED_RABBITMQ_MAJOR` makes the suite verify it reached the broker series +you meant. The other knobs are `RABBITMQ_HOST`, `RABBITMQ_USERNAME`, +`RABBITMQ_PASSWORD`, `RABBITMQ_VHOST` and `RABBITMQ_WAIT_TIMEOUT`. + +**The suite deletes every queue in the vhost when it starts** — a run under one +queue type would otherwise collide with queues left by a run under another, +since type and durability are fixed at declaration. Point it at a broker you +own. diff --git a/changelog.md b/changelog.md index 8612ad5..3ff3960 100644 --- a/changelog.md +++ b/changelog.md @@ -1,5 +1,56 @@ ### Changelog +### Unreleased + +**Added:** a first party `durable` setting, so durability can be configured +rather than only passed per route. Configurable globally — including from +`config/action_subscriber.yml`, which means an operator can turn it on without a +code change — or per subscriber: + +```yaml +production: + durable: true +``` + +```ruby +class UserSubscriber < ::ActionSubscriber::Base + durable true +end +``` + +Precedence is route option > subscriber declaration > `config.durable`, with +`:quorum` and `:stream` still forcing durability regardless. The default is +unchanged at `false`. + +This is the smallest way onto RabbitMQ 4.x, which refuses the transient queues +every default route declares. `config.queue_type = :quorum` also works, but +changes the queue type as well; `config.durable` leaves it alone. + +**Fixed:** `ActionSubscriber::MessageRetry` declared its retry queues without +passing `:durable`. On JRuby that was harmless, because march_hare forces quorum +and stream queues durable internally — but bunny does not, so on MRI a +`config.queue_type` of `:quorum` made every retry declaration fail with +`PRECONDITION_FAILED - invalid property 'non-durable' for queue`. Retry queues +are now declared durable whenever the configured queue type requires it, and +they follow `config.durable` as well — otherwise a deployment that set +`config.durable` to get onto RabbitMQ 4.x would still fall over the first time a +message was retried. + +**Testing:** CI now runs the integration suite against the latest RabbitMQ 3.x +and 4.x on both drivers, and two new integration specs assert what the broker +actually created rather than what was put on the wire — +`spec/integration/queue_type_spec.rb` covers the queue type each route produces +(including the redeclaration conflicts that made a durable quorum queue +unusable before the `queue_type` setting existed), and +`spec/integration/broker_compatibility_spec.rb` pins the differences between the +two broker series. + +That second file documents a limitation rather than fixing it: routes still +default to `:durable => false`, and RabbitMQ 4.x denies transient non-exclusive +queues, so the default route shape cannot be declared on a stock 4.x broker. +Set `config.queue_type = :quorum` (or `:durable => true`) to run on 4.x. See +"Supported RabbitMQ Versions" in the README. + ### 6.0.0 - August 6, 2026 **Breaking change on JRuby.** `march_hare` defaults its `:type` option to From 3668e585c5f98b276ce39008f90a2e31b109f992 Mon Sep 17 00:00:00 2001 From: John Bolliger Date: Wed, 12 Aug 2026 14:49:54 -0700 Subject: [PATCH 14/15] Stop integration specs interfering through the shared connection Two defects in specs added on this branch, both of which make an unrelated example fail at an unrelated line. Bunny raises asynchronous connection errors on Thread.main, so anything that disturbs the shared subscriber connection surfaces wherever the current example happens to be -- in CI, as a Bunny::NetworkFailure pointing at a `sleep` in consumer_cancellation_spec. * queue_type_spec drives RouteSet#setup_queue directly, which opens a channel on the shared subscriber connection and never closes it. Four examples leaked a channel each, along with its consumer work pool. They are closed now. * consumer_cancellation triggers the cancellation it tests by deleting queues out from under live consumers, and was doing that for every queue in the vhost. It only needs its own. The wider version also deleted queues that other examples still held channels against on the same connection. Both reduce cross-example interference rather than change what is tested. Note that the CI failure this addresses did not reproduce locally -- 16 runs against the same broker and driver under full CPU saturation, including at the seed CI used -- so this is a plausible cause, not a confirmed one. Co-Authored-By: Claude Opus 5 (1M context) --- .../integration/consumer_cancellation_spec.rb | 16 +++++++++++---- spec/integration/queue_type_spec.rb | 20 +++++++++++++++++-- 2 files changed, 30 insertions(+), 6 deletions(-) diff --git a/spec/integration/consumer_cancellation_spec.rb b/spec/integration/consumer_cancellation_spec.rb index 104f437..6fd6170 100644 --- a/spec/integration/consumer_cancellation_spec.rb +++ b/spec/integration/consumer_cancellation_spec.rb @@ -22,7 +22,7 @@ def created consumers = rabbit_consumers.dup # Signal a cancellation event to all subscribers. - delete_all_queues! + delete_subscriber_queues! # Give consumers a chance to restart. sleep 2.0 @@ -50,7 +50,7 @@ def created consumers = rabbit_consumers.dup # Signal a cancellation event to all subscribers. - delete_all_queues! + delete_subscriber_queues! # Give consumers a chance to restart. sleep 2.0 @@ -113,7 +113,15 @@ def rabbit_consumers # Deleting the queues out from under the consumers is how this spec triggers the # cancellation it is testing. - def delete_all_queues! - RabbitMQTestHelper.delete_all_queues! + # + # Only this spec's own queues, not every queue in the vhost. Both do trigger the + # cancellation, but the wider version also deletes queues that other examples left + # channels open against on the same shared connection, which turns an unrelated blip + # into a Bunny::NetworkFailure raised on Thread.main -- i.e. against whatever line the + # example happens to be on. + def delete_subscriber_queues! + ::ActionSubscriber.send(:route_set).routes.each do |route| + RabbitMQTestHelper.delete_queue!(route.queue) + end end end diff --git a/spec/integration/queue_type_spec.rb b/spec/integration/queue_type_spec.rb index 8536780..77129e8 100644 --- a/spec/integration/queue_type_spec.rb +++ b/spec/integration/queue_type_spec.rb @@ -128,17 +128,33 @@ def created let(:queue_name) { "alice.conflicting_type.created" } before do + @opened_channels = [] helper.delete_queue!(queue_name) helper.declare_queue!(queue_name, :durable => true, :type => "quorum") end - after { helper.delete_queue!(queue_name) } + after do + # setup_queue opens a channel on the *shared* subscriber connection and never + # closes it. Leaving them open leaks a consumer work pool per example and gives + # later specs -- consumer_cancellation deletes every queue in the vhost -- more + # channels on that connection to disturb. + @opened_channels.each do |channel| + begin + channel.close + rescue ::StandardError + nil + end + end + helper.delete_queue!(queue_name) + end def setup_queue_for(route_options) routes = ::ActionSubscriber::Router.draw_routes do route ::ConflictingTypeSubscriber, :created, route_options end - ::ActionSubscriber::RouteSet.new(routes).send(:setup_queue, routes.first) + queue = ::ActionSubscriber::RouteSet.new(routes).send(:setup_queue, routes.first) + @opened_channels << queue.channel + queue end it "fails when the route does not name the type (what master always did)" do From 560da4b8a8912722c7eb12f09467703f32076423 Mon Sep 17 00:00:00 2001 From: John Bolliger Date: Wed, 12 Aug 2026 15:37:04 -0700 Subject: [PATCH 15/15] Print how to reproduce a failed run CI failures on this suite are ordering- and configuration-dependent, and the seed alone no longer reproduces one: behaviour also depends on which broker the job talked to and which settings it ran under, and the matrix varies all three. On failure the suite now prints a complete command. Reproduce this run against RabbitMQ 3.13.7 with: ACTION_SUBSCRIBER_QUEUE_TYPE=quorum \ BUNDLE_GEMFILE=gemfiles/rails_8.1.gemfile \ EXPECTED_RABBITMQ_MAJOR=3 \ RABBITMQ_PORT=5673 \ bundle exec rspec --seed 40843 The variables are recorded by RabbitMQTestHelper.env as the suite reads them, rather than kept in a list somebody has to remember to update. A hand-kept list had already drifted while writing this: it omitted EXPECTED_RABBITMQ_MAJOR, which decides whether an example is defined at all -- so the command would have produced a different example count, and the same seed would have shuffled a different list. Deriving it from what the run actually consumed also means only variables that were really set get printed. Registered as a :close listener rather than an after(:suite) hook. Suite hooks run inside Reporter#report, so their output lands above the failure dump and above RSpec's own seed line -- scrolled off the bottom, which is where anyone opening a failed job starts reading. :close fires last. Two related changes: * the management client now sets explicit timeouts. It passed its options straight to Faraday, which sets none of its own, so the Net::HTTP defaults applied: a host that drops packets rather than refusing would wedge the suite for up to two minutes with no output. * example_status_persistence_file_path, so --only-failures and --next-failure work locally. No CI artifact for the status file: --only-failures reads the local copy, so using CI's would mean downloading it and putting it in place, which nobody will do. If flake detection is wanted, store_test_results with a JUnit formatter is the mechanism, and that is worth doing separately. Co-Authored-By: Claude Opus 5 (1M context) --- .circleci/config.yml | 3 ++ .gitignore | 3 ++ spec/spec_helper.rb | 7 ++++ spec/support/rabbitmq.rb | 25 +++++++++++- spec/support/reproduction_reporter.rb | 56 +++++++++++++++++++++++++++ 5 files changed, 93 insertions(+), 1 deletion(-) create mode 100644 spec/support/reproduction_reporter.rb diff --git a/.circleci/config.yml b/.circleci/config.yml index 64bae00..d5dba17 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -137,6 +137,9 @@ jobs: done echo "RabbitMQ is ready (<< parameters.rabbitmq_image >>)!" + # On failure the suite prints the full reproduction command -- seed, broker and + # every environment variable the run actually consumed -- as the last thing in the + # log. See spec/support/reproduction_reporter.rb. - run: name: Run Tests command: bundle exec rspec diff --git a/.gitignore b/.gitignore index f0f8b8f..e85e8cd 100644 --- a/.gitignore +++ b/.gitignore @@ -9,6 +9,9 @@ lib/bundler/man pkg rdoc spec/reports + +# rspec --only-failures state, rewritten as specs run +spec/examples.txt test/tmp test/version_tmp tmp diff --git a/spec/spec_helper.rb b/spec/spec_helper.rb index 8c24ead..5cb7d9b 100644 --- a/spec/spec_helper.rb +++ b/spec/spec_helper.rb @@ -19,6 +19,7 @@ # Require spec support files require 'support/user_subscriber' require 'support/rabbitmq' +require 'support/reproduction_reporter' require 'action_subscriber/rspec' # Silence the Logger @@ -60,6 +61,11 @@ mocks.verify_partial_doubles = true end + # Lets `rspec --only-failures` and `--next-failure` work locally. + config.example_status_persistence_file_path = "spec/examples.txt" + + ReproductionReporter.register!(config) + # Fail fast with a clear message (rather than a flurry of Bunny reconnect warnings) # if the broker isn't up yet when the integration suite starts. config.before(:suite) do @@ -99,6 +105,7 @@ ::ActionSubscriber.stop_subscribers!(0.1) ::ActionSubscriber::RabbitConnection.subscriber_disconnect! end + end # Set ActionSubscriber configuration for the duration of the block and put it back diff --git a/spec/support/rabbitmq.rb b/spec/support/rabbitmq.rb index a791f2b..4c2542f 100644 --- a/spec/support/rabbitmq.rb +++ b/spec/support/rabbitmq.rb @@ -31,12 +31,30 @@ module RabbitMQTestHelper # Every environment variable the suite honors goes through here. Blank is treated as # unset: a CircleCI job parameter that defaults to "" still reaches the environment as # an empty string. Pass a block to coerce a value that is actually present. + # + # Reads are recorded so the suite can print back exactly what a run was configured + # with. Deriving that from what was actually consumed, rather than from a hand-kept + # list, is the only way it stays correct as knobs are added. def env(name, default = nil) value = ENV[name].to_s.strip return default if value.empty? + observed_env[name] = value block_given? ? yield(value) : value end + # Raw strings, so they can be pasted back into a shell. Only variables that were + # actually set appear -- defaults are not worth restating. + def observed_env + @observed_env ||= {} + end + + # The broker version if some earlier call already fetched it, otherwise nil. Callers + # that only want it for a diagnostic must not trigger the fetch: the management client + # has generous timeouts, and a unit-only run has no broker to ask. + def known_broker_version + @broker_version + end + def host env("RABBITMQ_HOST", "127.0.0.1") end @@ -61,11 +79,16 @@ def vhost env("RABBITMQ_VHOST", "/") end + # Timeouts are set explicitly: the client passes its options straight to Faraday, which + # sets none of its own, so the Net::HTTP defaults (60s connect, 60s read) apply. A host + # that drops packets rather than refusing -- a killed CI service container, say -- would + # otherwise wedge the suite for two minutes with no output. def http_client @http_client ||= ::RabbitMQ::HTTP::Client.new( "http://#{host}:#{management_port}", :username => username, - :password => password + :password => password, + :request => { :open_timeout => 5, :timeout => 5 } ) end diff --git a/spec/support/reproduction_reporter.rb b/spec/support/reproduction_reporter.rb new file mode 100644 index 0000000..7b4a2ba --- /dev/null +++ b/spec/support/reproduction_reporter.rb @@ -0,0 +1,56 @@ +# Prints how to reproduce a failed run. +# +# The seed on its own is not enough. Ordering depends on it, but behaviour depends on +# which broker the run talked to and which settings it ran under -- and CI varies all +# three across the matrix. Worse, EXPECTED_RABBITMQ_MAJOR decides whether an example is +# defined at all, so omitting it changes the example count and the same seed shuffles a +# different list. +# +# Registered as a :close listener rather than an after(:suite) hook. Suite hooks run +# inside Reporter#report, so their output lands above the failure dump and above RSpec's +# own seed line -- i.e. scrolled off the bottom of a CI log, which is where anyone +# reading a failed job starts. :close fires last. +class ReproductionReporter + def self.register!(configuration) + configuration.reporter.register_listener(new(configuration.output_stream), :close) + end + + def initialize(output) + @output = output + end + + def close(_notification) + failures = ::RSpec.configuration.reporter.failed_examples + return if failures.empty? + + @output.puts(message) + end + + private + + def message + <<~REPRO + + Reproduce this run#{broker_description} with: + + #{command} + REPRO + end + + def command + settings = RabbitMQTestHelper.observed_env.dup + # Bundler reads BUNDLE_GEMFILE itself, so it never passes through the helper. It + # names the Rails half of the CI matrix, so it belongs in the command. Relative, so + # the line pastes cleanly. + gemfile = ENV["BUNDLE_GEMFILE"].to_s.strip + settings["BUNDLE_GEMFILE"] = gemfile.sub("#{::Dir.pwd}/", "") unless gemfile.empty? + + assignments = settings.sort.map { |name, value| "#{name}=#{value}" } + (assignments + ["bundle exec rspec --seed #{::RSpec.configuration.seed}"]).join(" \\\n ") + end + + def broker_description + version = RabbitMQTestHelper.known_broker_version + version ? " against RabbitMQ #{version}" : "" + end +end