diff --git a/.circleci/config.yml b/.circleci/config.yml index 24ef51e..d5dba17 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -6,6 +6,37 @@ 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" + 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) @@ -13,15 +44,20 @@ jobs: environment: JRUBY_OPTS: "-J-Xmx1024m" RAILS_ENV: test - # Tell your app where to find RabbitMQ (if your app uses this ENV var) - RABBITMQ_URL: "amqp://guest:guest@localhost:5672" + # Select the Rails version under test via the Appraisal gemfile. + BUNDLE_GEMFILE: << parameters.gemfile >> + # Tell the suite where to find RabbitMQ. + 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) - - 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 + # 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 @@ -35,27 +71,54 @@ 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 + # + 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: - - v1-gems-<< parameters.docker_image >>-{{ checksum "Gemfile.lock" }} - - v1-gems-<< parameters.docker_image >>- + - 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 command: | - gem install bundler 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: v1-gems-<< parameters.docker_image >>-{{ checksum "Gemfile.lock" }} + - ./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. + # 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: | @@ -69,22 +132,104 @@ 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 >>)!" + # 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 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" + + # 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"] diff --git a/.gitignore b/.gitignore index 88db647..e85e8cd 100644 --- a/.gitignore +++ b/.gitignore @@ -9,12 +9,20 @@ lib/bundler/man pkg rdoc spec/reports + +# rspec --only-failures state, rewritten as specs run +spec/examples.txt test/tmp test/version_tmp tmp Gemfile.lock +# 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 _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/README.md b/README.md index c9b35d5..c05a7d0 100644 --- a/README.md +++ b/README.md @@ -45,6 +45,181 @@ 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 +----------- + +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` 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 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 @@ -129,12 +304,14 @@ 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"]`) * 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 @@ -231,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: @@ -241,3 +418,58 @@ $ 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. + +### 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/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..fcdf0b5 100644 --- a/action_subscriber.gemspec +++ b/action_subscriber.gemspec @@ -32,8 +32,9 @@ 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" spec.add_development_dependency "pry-nav" spec.add_development_dependency "rabbitmq_http_api_client", "~> 1.15.0" diff --git a/changelog.md b/changelog.md index 108c0b3..3ff3960 100644 --- a/changelog.md +++ b/changelog.md @@ -1,5 +1,76 @@ ### 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 +`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..720052e 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 @@ -8,6 +9,7 @@ class Configuration :connection_reaping_timeout_interval, :decoder, :default_exchange, + :durable, :error_handler, :heartbeat, :host, @@ -28,6 +30,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 @@ -36,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 => [], @@ -43,6 +53,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 +155,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/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/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..ac5890e 100644 --- a/lib/action_subscriber/message_retry.rb +++ b/lib/action_subscriber/message_retry.rb @@ -51,7 +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 = channel.queue(retry_queue_name, :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 diff --git a/lib/action_subscriber/queue_type.rb b/lib/action_subscriber/queue_type.rb new file mode 100644 index 0000000..80a31fb --- /dev/null +++ b/lib/action_subscriber/queue_type.rb @@ -0,0 +1,54 @@ +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 + + # 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 9d25078..b460f77 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,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) + @durable = ::ActionSubscriber::QueueType.durable?(@queue_type, durable) @exchange = attributes.fetch(:exchange).to_s @prefetch = attributes.fetch(:prefetch) { ::ActionSubscriber.config.prefetch } @queue = attributes.fetch(:queue) @@ -34,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/lib/action_subscriber/version.rb b/lib/action_subscriber/version.rb index cdfa208..4b0a00c 100644 --- a/lib/action_subscriber/version.rb +++ b/lib/action_subscriber/version.rb @@ -1,3 +1,3 @@ module ActionSubscriber - VERSION = "5.4.0" + VERSION = "6.0.0" end 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/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/consumer_cancellation_spec.rb b/spec/integration/consumer_cancellation_spec.rb index eb25aa3..6fd6170 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 @@ -26,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 @@ -54,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 @@ -115,9 +111,17 @@ def rabbit_consumers route_set.try(:bunny_consumers) || route_set.try(:march_hare_consumers) end - def delete_all_queues! - http_client.list_queues.each do |queue| - http_client.delete_queue(queue.vhost, queue.name) + # Deleting the queues out from under the consumers is how this spec triggers the + # cancellation it is testing. + # + # 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 new file mode 100644 index 0000000..77129e8 --- /dev/null +++ b/spec/integration/queue_type_spec.rb @@ -0,0 +1,187 @@ +# 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 + @opened_channels = [] + helper.delete_queue!(queue_name) + helper.declare_queue!(queue_name, :durable => true, :type => "quorum") + end + + 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 + 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 + 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 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 new file mode 100644 index 0000000..7dcf6a6 --- /dev/null +++ b/spec/lib/action_subscriber/queue_type_spec.rb @@ -0,0 +1,108 @@ +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 ".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 + + 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..a2fa84d 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 @@ -71,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/spec_helper.rb b/spec/spec_helper.rb index cd6b9a1..5cb7d9b 100644 --- a/spec/spec_helper.rb +++ b/spec/spec_helper.rb @@ -18,6 +18,8 @@ # Require spec support files require 'support/user_subscriber' +require 'support/rabbitmq' +require 'support/reproduction_reporter' require 'action_subscriber/rspec' # Silence the Logger @@ -25,11 +27,70 @@ ::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 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 + 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 $messages = Set.new draw_routes @@ -44,6 +105,21 @@ ::ActionSubscriber.stop_subscribers!(0.1) ::ActionSubscriber::RabbitConnection.subscriber_disconnect! 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) diff --git a/spec/support/rabbitmq.rb b/spec/support/rabbitmq.rb new file mode 100644 index 0000000..4c2542f --- /dev/null +++ b/spec/support/rabbitmq.rb @@ -0,0 +1,282 @@ +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 the standard host+port; by default we assume localhost:5672 +# with the management plugin on 15672. +# +# 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 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: +# +# # 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. + # + # 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 + + def port + 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 + + # 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, + :request => { :open_timeout => 5, :timeout => 5 } + ) + 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 + + 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 + return true if yield + rescue ::StandardError => e + last_error = e + end + + if ::Time.now >= deadline + 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 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 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