RUBY-3602 Include server address in OperationFailure and BulkWriteError messages (#3027)

Author: comandeo-mongo

lib/mongo.rb

@@ -101,6 +101,7 @@ def self.delegate_option(obj, opt)
     delegate_option Config, :broken_view_options
     delegate_option Config, :validate_update_replace
     delegate_option Config, :csfle_convert_to_ruby_types
+    delegate_option Config, :include_server_address_in_errors
   end
 
   # Clears the driver's OCSP response cache.

lib/mongo/bulk_write/result.rb

@@ -25,6 +25,10 @@ def acknowledged?
         @acknowledged
       end
 
+      # @return [ Array<String> ] Deduplicated list of "host:port" addresses of
+      #   the servers that produced this bulk write's operations.
+      attr_reader :server_addresses
+
       # Constant for number removed.
       #
       # @since 2.1.0
@@ -97,13 +101,17 @@ def deleted_count
       #
       # @param [ BSON::Document, Hash ] results The results document.
       # @param [ Boolean ] acknowledged Is the result acknowledged?
+      # @param [ Array<String> ] server_addresses Deduplicated "host:port"
+      #   addresses of the servers that produced the underlying operation
+      #   results.
       #
       # @since 2.1.0
       #
       # @api private
-      def initialize(results, acknowledged)
+      def initialize(results, acknowledged, server_addresses = [])
         @results = results
         @acknowledged = acknowledged
+        @server_addresses = Array(server_addresses).compact.uniq
       end
 
       # Returns the number of documents inserted.
@@ -189,7 +197,7 @@ def upserted_ids
       #
       # @since 2.1.0
       def validate!
-        raise Error::BulkWriteError.new(@results) if @results['writeErrors'] || @results['writeConcernErrors']
+        raise Error::BulkWriteError.new(@results, server_addresses: @server_addresses) if @results['writeErrors'] || @results['writeConcernErrors']
 
         self
       end

lib/mongo/bulk_write/result_combiner.rb

@@ -28,6 +28,10 @@ class ResultCombiner
       # @return [ Hash ] results The results hash.
       attr_reader :results
 
+      # @return [ Array<String> ] Deduplicated list of "host:port" addresses of
+      #   the servers that produced the combined operation results.
+      attr_reader :server_addresses
+
       # Create the new result combiner.
       #
       # @api private
@@ -39,6 +43,7 @@ class ResultCombiner
       def initialize
         @results = {}
         @count = 0
+        @server_addresses = []
       end
 
       # Adds a result to the overall results.
@@ -68,6 +73,8 @@ def combine!(result, count)
         combine_errors!(result)
         @count += count
         @acknowledged = result.acknowledged?
+        seed = result.connection_description&.address&.seed
+        @server_addresses << seed if seed && !@server_addresses.include?(seed)
       end
 
       # Get the final result.
@@ -78,7 +85,7 @@ def combine!(result, count)
       #
       # @since 2.1.0
       def result
-        BulkWrite::Result.new(results, @acknowledged).validate!
+        BulkWrite::Result.new(results, @acknowledged, @server_addresses).validate!
       end
 
       private

lib/mongo/config.rb

@@ -29,6 +29,11 @@ module Config
     # decryption instead of BSON types.
     option :csfle_convert_to_ruby_types, default: false
 
+    # When this flag is set to true, the (host:port) of the server that produced
+    # the error is appended to error messages for OperationFailure and
+    # BulkWriteError. See RUBY-3602.
+    option :include_server_address_in_errors, default: false
+
     # Set the configuration options.
     #
     # @example Set the options.

lib/mongo/error/bulk_write_error.rb

@@ -34,17 +34,26 @@ class BulkWriteError < Error
       # @return [ BSON::Document ] result The error result.
       attr_reader :result
 
+      # @return [ Array<String> ] Deduplicated list of "host:port" addresses of
+      #   the servers that produced this bulk write error. Empty when no
+      #   addresses were supplied.
+      attr_reader :server_addresses
+
       # Instantiate the new exception.
       #
       # @example Instantiate the exception.
       #   Mongo::Error::BulkWriteError.new(response)
       #
       # @param [ Hash ] result A processed response from the server
       #   reporting results of the operation.
+      # @param [ Array<String | Mongo::Address | Mongo::Server::Description> ]
+      #   server_addresses Addresses of the servers that produced this error.
+      #   Entries are normalized to "host:port" strings.
       #
       # @since 2.0.0
-      def initialize(result)
+      def initialize(result, server_addresses: nil)
         @result = result
+        @server_addresses = normalize_server_addresses(server_addresses)
 
         # Exception constructor behaves differently for a nil argument and
         # for no argument. Avoid passing nil explicitly.
@@ -93,8 +102,27 @@ def build_message
 
         fragment = "Multiple errors: #{fragment}" if errors.length > 1
 
+        if Mongo.include_server_address_in_errors && @server_addresses.any?
+          fragment = "#{fragment} (on #{@server_addresses.join(', ')})"
+        end
+
         fragment
       end
+
+      def normalize_server_addresses(value)
+        return [] if value.nil?
+
+        Array(value).filter_map do |entry|
+          case entry
+          when String then entry
+          when Mongo::Address then entry.seed
+          when Mongo::Server::Description then entry.address&.seed
+          else
+            raise ArgumentError,
+                  "server_addresses entries must be String, Mongo::Address, or Mongo::Server::Description; got #{entry.class}"
+          end
+        end.uniq
+      end
     end
   end
 end

lib/mongo/error/operation_failure.rb

@@ -53,6 +53,10 @@ module OperationFailure::Family
         # @api experimental
         attr_reader :server_message
 
+        # @return [ String | nil ] The address ("host:port") of the server
+        #   that produced this error, if known.
+        attr_reader :server_address
+
         # Error codes and code names that should result in a failing getMore
         # command on a change stream NOT being resumed.
         #
@@ -172,6 +176,8 @@ def write_concern_error?
         #   error document.
         # @option options [ String ] server_message The server-returned
         #   error message parsed from the response.
+        # @option options [ nil | String | Mongo::Address | Mongo::Server::Description ]
+        #   :server_address The address of the server that produced the error.
         # @option options [ Hash ] :write_concern_error_document The
         #   server-supplied write concern error document, if any.
         # @option options [ Integer ] :write_concern_error_code Error code for
@@ -185,7 +191,8 @@ def write_concern_error?
         # @option options [ true | false ] :wtimeout Whether the error is a wtimeout.
         def initialize(message = nil, result = nil, options = {})
           @details = retrieve_details(options[:document])
-          super(append_details(message, @details))
+          @server_address = normalize_server_address(options[:server_address])
+          super(append_server_address(append_details(message, @details)))
 
           @result = result
           @code = options[:code]
@@ -241,6 +248,38 @@ def append_details(message, details)
 
           message + " -- #{details.to_json}"
         end
+
+        # Append the server address suffix to the message when the
+        # Mongo.include_server_address_in_errors flag is enabled and
+        # a server address is known.
+        #
+        # @return [ String | nil ] the message with the suffix appended,
+        #   or the original message unchanged.
+        def append_server_address(message)
+          return message unless Mongo.include_server_address_in_errors
+          return message if @server_address.nil?
+          return "(on #{@server_address})" if message.nil? || message.empty?
+
+          "#{message} (on #{@server_address})"
+        end
+
+        # Normalize a server_address option into a String "host:port" form.
+        #
+        # @param [ nil | String | Mongo::Address | Mongo::Server::Description ] value
+        #
+        # @return [ String | nil ] The normalized address, or nil.
+        def normalize_server_address(value)
+          case value
+          when nil then nil
+          when String then value
+          when Mongo::Address then value.seed
+          when Mongo::Server::Description
+            value.address.is_a?(Mongo::Address) ? value.address.seed : nil
+          else
+            raise ArgumentError,
+                  "server_address must be nil, String, Mongo::Address, or Mongo::Server::Description; got #{value.class}"
+          end
+        end
       end
 
       # OperationFailure is the canonical implementor of the

lib/mongo/operation/result.rb

@@ -355,7 +355,8 @@ def error
           wtimeout: parser.wtimeout,
           connection_description: connection_description,
           document: parser.document,
-          server_message: parser.server_message
+          server_message: parser.server_message,
+          server_address: connection_description
         )
       end
 

spec/integration/bulk_write_error_message_spec.rb

@@ -63,4 +63,26 @@
       e.message.scan(' -- ').length.should be <= 1
     end
   end
+
+  context 'when include_server_address_in_errors is enabled' do
+    around do |example|
+      original = Mongo.include_server_address_in_errors
+      Mongo.include_server_address_in_errors = true
+      example.run
+    ensure
+      Mongo.include_server_address_in_errors = original
+    end
+
+    it 'includes the server address suffix in the bulk error message' do
+      collection.insert_one(_id: 1)
+      begin
+        collection.insert_many([ { _id: 1 }, { _id: 2 }, { _id: 2 } ])
+      rescue Mongo::Error::BulkWriteError => e
+        expect(e.message).to match(/\(on [^)]+:\d+(?:, [^)]+:\d+)*\)\z/)
+        expect(e.server_addresses).not_to be_empty
+      else
+        raise 'expected BulkWriteError'
+      end
+    end
+  end
 end

spec/integration/operation_failure_message_spec.rb

@@ -52,4 +52,22 @@
                              /User bogus.*is not authorized.*\[18:AuthenticationFailed\]: Authentication failed/)
     end
   end
+
+  context 'when include_server_address_in_errors is enabled' do
+    around do |example|
+      original = Mongo.include_server_address_in_errors
+      Mongo.include_server_address_in_errors = true
+      example.run
+    ensure
+      Mongo.include_server_address_in_errors = original
+    end
+
+    it 'includes the server address in the message' do
+      client.command(bogus_command: nil)
+      raise('Should have raised')
+    rescue Mongo::Error::OperationFailure => e
+      expect(e.message).to match(/\(on [^)]+:\d+\)\z/)
+      expect(e.server_address).to be_a(String)
+    end
+  end
 end

spec/mongo/bulk_write/result_combiner_spec.rb

@@ -0,0 +1,38 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+describe Mongo::BulkWrite::ResultCombiner do
+  describe 'server_addresses accumulation' do
+    let(:combiner) { described_class.new }
+
+    def stub_op_result(seed)
+      description = instance_double(
+        Mongo::Server::Description,
+        address: Mongo::Address.new(seed)
+      )
+      result = double('op_result')
+      allow(result).to receive(:write_concern_error?).and_return(false)
+      allow(result).to receive(:acknowledged?).and_return(true)
+      allow(result).to receive(:aggregate_write_errors).and_return(nil)
+      allow(result).to receive(:aggregate_write_concern_errors).and_return(nil)
+      allow(result).to receive(:validate!).and_return(true)
+      allow(result).to receive(:respond_to?).and_return(false)
+      allow(result).to receive(:connection_description).and_return(description)
+      result
+    end
+
+    it 'collects unique seeds from combined results' do
+      combiner.combine!(stub_op_result('h1:27017'), 1)
+      combiner.combine!(stub_op_result('h2:27017'), 1)
+      combiner.combine!(stub_op_result('h1:27017'), 1)
+
+      final = combiner.result
+      expect(final.server_addresses).to contain_exactly('h1:27017', 'h2:27017')
+    end
+
+    it 'defaults to empty when no results combined' do
+      expect(combiner.server_addresses).to eq([])
+    end
+  end
+end