Fix RUBY-1647 Reset connection pool on non-timeout network errors (#1258)

Author: p-mongo

lib/mongo/cluster.rb

@@ -340,7 +340,9 @@ def summary
     end
 
     # @api private
-    attr_reader :server_selection_semaphore
+    def server_selection_semaphore
+      options[:server_selection_semaphore]
+    end
 
     # Finalize the cluster for garbage collection.
     #

lib/mongo/error.rb

@@ -134,6 +134,7 @@ def add_label(label)
   end
 end
 
+require 'mongo/error/sdam_error_detection'
 require 'mongo/error/parser'
 require 'mongo/error/write_retryable'
 require 'mongo/error/change_stream_resumable'

lib/mongo/error/operation_failure.rb

@@ -20,6 +20,7 @@ class Error
     # @since 2.0.0
     class OperationFailure < Error
       extend Forwardable
+      include SdamErrorDetection
 
       # Error codes and code names that should result in a failing write
       # being retried.
@@ -28,7 +29,7 @@ class OperationFailure < Error
       # @api private
       WRITE_RETRY_ERRORS = [
         {:code_name => 'InterruptedAtShutdown', :code => 11600},
-        {:code_name => 'InterruptedDueToReplStateChange', :code => 11602},
+        {:code_name => 'InterruptedDueToStepDown', :code => 11602},
         {:code_name => 'NotMaster', :code => 10107},
         {:code_name => 'NotMasterNoSlaveOk', :code => 13435},
         {:code_name => 'NotMasterOrSecondary', :code => 13436},
@@ -175,6 +176,9 @@ def change_stream_resumable_code?
       #
       # @option options [ Integer ] :code Error code
       # @option options [ String ] :code_name Error code name
+      # @option options [ Array<String> ] :labels The set of labels associated
+      #   with the error
+      # @option options [ true | false ] :wtimeout Whether the error is a wtimeout
       #
       # @since 2.5.0, options added in 2.6.0
       def initialize(message = nil, result = nil, options = {})
@@ -186,9 +190,9 @@ def initialize(message = nil, result = nil, options = {})
         super(message)
       end
 
-      # Whether the error was a write concern timeout.
+      # Whether the error is a write concern timeout.
       #
-      # @return [ true | false ] Whether the error was a write concern timeout.
+      # @return [ true | false ] Whether the error is a write concern timeout.
       #
       # @since 2.7.1
       def wtimeout?

lib/mongo/error/parser.rb

@@ -44,6 +44,7 @@ class Error
     #
     # @since 2.0.0
     class Parser
+      include SdamErrorDetection
 
       # @return [ BSON::Document ] document The returned document.
       attr_reader :document
@@ -62,7 +63,7 @@ class Parser
       # @since 2.6.0
       attr_reader :code_name
 
-      # @return [ Array ] labels The set of labels associated with the error.
+      # @return [ Array<String> ] labels The set of labels associated with the error.
       # @since 2.7.0
       attr_reader :labels
 

lib/mongo/error/sdam_error_detection.rb

@@ -0,0 +1,54 @@
+module Mongo
+  class Error
+    # @note Although not_master? and node_recovering? methods of this module
+    #   are part of the public API, the fact that these methods are defined on
+    #   this module and not on the classes which include this module is not
+    #   part of the public API.
+    #
+    # @api semipublic
+    module SdamErrorDetection
+
+      # @api private
+      NOT_MASTER_CODES = [10107, 13435].freeze
+
+      # @api private
+      NODE_RECOVERING_CODES = [11600, 11602, 13436, 189, 91].freeze
+
+      # Whether the error is a "not master" error, or one of its variants.
+      #
+      # See https://github.com/mongodb/specifications/blob/master/source/server-discovery-and-monitoring/server-discovery-and-monitoring.rst#not-master-and-node-is-recovering.
+      #
+      # @return [ true | false ] Whether the error is a not master.
+      #
+      # @since 2.8.0
+      def not_master?
+        if node_recovering?
+          false
+        elsif code && NOT_MASTER_CODES.include?(code)
+          true
+        elsif message
+          message.include?('not master')
+        else
+          false
+        end
+      end
+
+      # Whether the error is a "node is recovering" error, or one of its variants.
+      #
+      # See https://github.com/mongodb/specifications/blob/master/source/server-discovery-and-monitoring/server-discovery-and-monitoring.rst#not-master-and-node-is-recovering.
+      #
+      # @return [ true | false ] Whether the error is a node is recovering.
+      #
+      # @since 2.8.0
+      def node_recovering?
+        if code && NODE_RECOVERING_CODES.include?(code)
+          true
+        elsif message
+          message.include?('node is recovering') || message.include?('not master or secondary')
+        else
+          false
+        end
+      end
+    end
+  end
+end

lib/mongo/operation/result.rb

@@ -79,6 +79,9 @@ def initialize(replies)
       # @return [ Array<Protocol::Reply> ] replies The wrapped wire protocol replies.
       attr_reader :replies
 
+      def_delegators :parser,
+        :not_master?, :node_recovering?
+
       # Is the result acknowledged?
       #
       # @note On MongoDB 2.6 and higher all writes are acknowledged since the

lib/mongo/operation/shared/executable.rb

@@ -45,6 +45,10 @@ def dispatch_message(server)
 
       def process_result(result, server)
         server.update_cluster_time(result)
+        if result.not_master? || result.node_recovering?
+          server.unknown!
+          server.monitor.scan_semaphore.signal
+        end
         session.process(result) if session
         result
       end

lib/mongo/retryable.rb

@@ -42,18 +42,22 @@ def read_with_retry(session = nil)
         attempt += 1
         yield
       rescue Error::SocketError, Error::SocketTimeoutError => e
-        raise(e) if attempt > cluster.max_read_retries || (session && session.in_transaction?)
+        if attempt > cluster.max_read_retries || (session && session.in_transaction?)
+          raise
+        end
         log_retry(e)
         cluster.scan!(false)
         retry
       rescue Error::OperationFailure => e
         if cluster.sharded? && e.retryable? && !(session && session.in_transaction?)
-          raise(e) if attempt > cluster.max_read_retries
+          if attempt > cluster.max_read_retries
+            raise
+          end
           log_retry(e)
           sleep(cluster.read_retry_interval)
           retry
         else
-          raise e
+          raise
         end
       end
     end
@@ -141,10 +145,14 @@ def write_with_retry(session, write_concern, ending_transaction = false, &block)
         txn_num = session.in_transaction? ? session.txn_num : session.next_txn_num
         yield(server, txn_num, false)
       rescue Error::SocketError, Error::SocketTimeoutError => e
-        raise e if session.in_transaction? && !ending_transaction
+        if session.in_transaction? && !ending_transaction
+          raise
+        end
         retry_write(e, txn_num, &block)
       rescue Error::OperationFailure => e
-        raise e if (session.in_transaction? && !ending_transaction) || !e.write_retryable?
+        if (session.in_transaction? && !ending_transaction) || !e.write_retryable?
+          raise
+        end
         retry_write(e, txn_num, &block)
       end
     end
@@ -167,17 +175,19 @@ def retry_write_allowed?(session, write_concern)
     end
 
     def retry_write(original_error, txn_num, &block)
-      cluster.scan!(false)
+      # We do not request a scan of the cluster here, because error handling
+      # for the error which triggered the retry should have updated the
+      # server description and/or topology as necessary (specifically,
+      # a socket error or a not master error should have marked the respective
+      # server unknown). Here we just need to wait for server selection.
       server = cluster.next_primary
       raise original_error unless (server.retry_writes? && txn_num)
       log_retry(original_error)
       yield(server, txn_num, true)
     rescue Error::SocketError, Error::SocketTimeoutError => e
-      cluster.scan!(false)
       raise e
     rescue Error::OperationFailure => e
       raise original_error unless e.write_retryable?
-      cluster.scan!(false)
       raise e
     rescue
       raise original_error
@@ -193,13 +203,15 @@ def legacy_write_with_retry(server = nil, session = nil)
         yield(server || cluster.next_primary)
       rescue Error::OperationFailure => e
         server = nil
-        raise(e) if attempt > Cluster::MAX_WRITE_RETRIES
+        if attempt > Cluster::MAX_WRITE_RETRIES
+          raise
+        end
         if e.write_retryable? && !(session && session.in_transaction?)
           log_retry(e)
           cluster.scan!(false)
           retry
         else
-          raise(e)
+          raise
         end
       end
     end

lib/mongo/server/connection.rb

@@ -293,6 +293,17 @@ def authenticate!(pending_connection)
       def default_mechanism
         @auth_mechanism || (@server.features.scram_sha_1_enabled? ? :scram : :mongodb_cr)
       end
+
+      def deliver(message)
+        begin
+          super
+        # Important: timeout errors are not handled here
+        rescue Error::SocketError
+          @server.unknown!
+          @server.pool.disconnect!
+          raise
+        end
+      end
     end
   end
 end

lib/mongo/server/connection_base.rb

@@ -16,6 +16,12 @@ module Mongo
   class Server
 
     # This class encapsulates common connection functionality.
+    #
+    # @note Although methods of this module are part of the public API,
+    #   the fact that these methods are defined on this module and not on
+    #   the classes which include this module is not part of the public API.
+    #
+    # @api semipublic
     class ConnectionBase
       extend Forwardable
       include Monitoring::Publishable