x-dead-letter-exchange in Bunny Channel using default value -retry and not configuration value set.

[anuonifade]

When creating a Bunny channel with default options set, If I try to connect to the queue with default x-dead-letter-exchange set by sneaker, it throws an error.

Here is my sneaker configuration for Bunny

require 'connection_pool'

module Sneakers
  class Bunny
    include Singleton

    cattr_accessor :options do
      {
        # The number of channels to make available in the pool.
        # Usually this should be equal the number of threads working in parallel
        channel_pool_size: 20,
        # Time to wait for a channel to become available,
        # before raising a Timeout error
        channel_pool_timeout: 2,
        # The exchange type declared by default.
        # One of :direct, :fanout, :topic, :headers
        exchange_type: :topic,
        # The exchange name use by default.
        exchange: :events,
        # Default options passed to the exchange declaration.
        exchange_options: { durable: true },
        # Preload bunny connection when true.
        preload: false
      }
    end

    class << self
      delegate :with, to: :instance

      def configure(options)
        options.merge!(options)

        instance.send(:ensure_connection!) if options[:preload]
      end
    end

    # For more information on concurrency visit:
    # http://rubybunny.info/articles/concurrency.html
    def initialize
      @mutex = Mutex.new
      @bunny = ::Bunny.new(Settings.amqp_url)
      @channel_pool = ConnectionPool.new(
        size: options[:channel_pool_size],
        timeout: options[:channel_pool_timeout]
      ) { create_channel }

      # gracefully shutdown Bunny connection
      at_exit { stop }
    end

    def stop
      @channel_pool.shutdown(&:close)
      @bunny.stop
    end

    def with(&block)
      ensure_connection!

      @channel_pool.with(&block)
    end

    private

      def ensure_connection!
        @mutex.synchronize do
          @bunny.start
        end
      end

      def create_channel
        @mutex.synchronize do
          @bunny.create_channel
        end
      end
  end
end

The Queue causing error is defined below.

class IamProfileUpdateWorker < ApplicationWorker
  from_queue 'lo_iam_profile_update',
             ack: true,
             exchange: 'iam_profile_update',
            # retry_exchange: '-retry',  # It works if I pass this to the queue.
             exchange_options: {
               type: :fanout,
               durable: true
             }

  def process_message(message, _delivery_info, _properties)
    Iam::ProfileUpdate.call(decode(message))
    ack!
  end
end

This is the way I am using it

require 'sneakers/bunny'

module Sneakers
  module Purge
    extend ActiveSupport::Concern

    module ClassMethods
      def purge
        Rails.logger.debug { "[CLEANER] Purging #{self}!" }
        new.purge
      end
    end

    # Purge all messages from the queue.
    def purge
      Sneakers::Bunny.with do |channel|
        handler_klass = opts[:handler] || Sneakers::CONFIG.fetch(:handler)
        # Configure options if needed
        if handler_klass.respond_to?(:configure_queue)
          @opts[:queue_options] = handler_klass.configure_queue(
            @queue.name,
            @opts
          )
        end

        channel.queue(@queue.name, @opts[:queue_options]).purge # Line generating error.
      end
    end
  end
end

If I don't add the retry_exchange option to the worker IamProfileUpdateWorker, I get the following error

PRECONDITION_FAILED - inequivalent arg 'x-dead-letter-exchange' for queue 'lo_iam_profile_update' in vhost '/': received 'activejob-retry' but current is '-retry'

activejob-retry is the default name set for the MaxRetry queue during config but it seems Bunny isn't using it when creating the connection.

Is there something I am not doing right?

[michaelklishin]

@anuonifade channels do not control what optional queue arguments like DLX keys will be. In other words, you can use a custom channel or not, the question is what will be passed on to Bunny::Channel#queue or equivalent by Sneakers.

Setting Bunny log level to debug or taking a traffic capture would show you what exactly is sent.

It's weird to see exchange_options set for a queue, if you need to define a topology like that, perhaps using a definition import to seed the topology would make more sense.

Another thing that immediately stands out: Bunny explicitly prohibits sharing channels between threads and it is very likely that this will happen in your code.

[anuonifade]

@michaelklishin thanks for the quick response. I am still going to refactor appropriately. The issue I am currently having is that I set the retry exchange in my sneaker initializer but I realised it is not being used by bunny. Below is my sneaker setup

require 'rufus-scheduler'

SNEAKERS_BUNNY_CONNECTION = Bunny.new(Settings.amqp_url)

# Defining default options. All Options can be overriden on the worker level.
Sneakers.configure(
  # Rabbit MQ Options
  # #################
  connection: SNEAKERS_BUNNY_CONNECTION,
  handler: Sneakers::Handlers::Maxretry,
  # AMQP exchange
  # exchange: 'sneakers',

  # Handler Options
  # ##############
  # retry options
  retry_exchange: 'activejob-retry',
  retry_error_exchange: 'activejob-error',
  retry_requeue_exchange: 'activejob-retry-requeue',
  # delay options
  delay_exchange: 'activejob-delay',
  delay_requeue_exchange: 'activejob-delay-requeue',
  delay_periods: [1, 60, 600, 3600, 86_400], # seconds

  # Deamon Options
  # ##############
  # A configuration file (see below)
  # runner_config_file: nil,
  # A metrics provider implementation
  # metrics: nil,
  # Send to background
  # daemonize: false,
  # When workers do frenzy-die, randomize to avoid resource starvation
  # Delay between thread startup
  # start_worker_delay: 1,
  # Number of per-cpu processes to run
  # workers: 1,
  # Log file
  log: Settings.sneakers.log == 'STDOUT' ? $stdout : Settings.sneakers.log,
  # Pid file
  pid_path: 'tmp/pids/sneakers.pid',

  # Workers Options
  # ###############
  # Maximal seconds to wait for job
  # job timeouts will be eliminated with the next sneakers version
  timeout_job_after: 10.minutes,
  # Grab 10 jobs together. Better speed.
  prefetch: Settings.sneakers.threads,
  # Threadpool size (good to match prefetch)
  threads: Settings.sneakers.threads,
  # Shared thread pool between workers within a worker group process
  # Should be set to true to avoid all kinds of undesired effects with a large
  # number of workers
  share_threads: true,
  # Environment
  # env: ENV['RACK_ENV'],
  # Is queue durable?
  # durable: true,
  # Must we acknowledge?
  # ack: true,
  # Keep a good connection with broker
  heartbeat: 10,
  # prefork/postfork hooks
  hooks: {
    # For ActiveRecord external network connections are closed before forking
    # and reestablished later to avoid concurrency errors related to shared sockets.
    before_fork: proc {
      Sneakers.logger.info('Worker: Disconnect from the database before fork')

      ActiveRecord::Base.connection_pool.disconnect!
    },
    after_fork: proc {
      self.class_eval do
        class RufusSingleton
          Scheduler = Rufus::Scheduler.singleton
        end
      end
      worker_classes = Sneakers::Utils.parse_workers(ENV.fetch('WORKERS')).first
      # max number of paralell threads => db connections
      threads = (3 + (worker_classes.count * 3)) + Settings.sneakers.threads

      ActiveSupport.on_load(:active_record) do
        config = Rails.application.config.database_configuration[Rails.env]
        config['pool'] = threads
        ActiveRecord::Base.establish_connection(config)
        Sneakers.logger.info("Worker: Establish database pool (size: #{ActiveRecord::Base.connection_pool.size})")
      end

      Sneakers.logger.info('Worker: Configure Bunny')
      Sneakers::Bunny.configure({
        # Increase channel_pool_size for sneakers processes to 10
        # to max number of threads in the process.
        channel_pool_size: threads,
        # Increase channel_pool_timeout for sneakers processes to 10 seconds
        # (defaults to 2 sec).
        channel_pool_timeout: 10,
        preload: ENV.fetch('CYPRESS', nil)
      })

      if ENV.fetch('WORKERS', nil).include?('AutomationWorker')
        Rails.logger.info('Worker: Schedule automations')
        Rails.application.config.after_initialize do
          Automation.where.not(scheduled_behaviors: nil, activated_at: nil).pluck(:id).each do |id|
            Automation::Scheduler.call(id)
          end
        end
      end
    }
  }
)

# Set the log level
Sneakers.logger.level = Logger.const_get(
  (Settings.log_level || 'error').upcase
)

at_exit do
  SNEAKERS_BUNNY_CONNECTION.close
end

Am I setting it correctly?