RUBY-1894 Clear connection pools when monitor ismaster times out (#1475)

* Clear server pool when a server is marked unknown

* Comment out tests that are no longer applicable

* Integration test for pool being cleared on network errors and timeouts in monitor

* improve test reliability

* try lower timeouts

* Retry all monitoring tests

Author: p-mongo

lib/mongo/cluster.rb

@@ -502,18 +502,42 @@ def scan!(sync=true)
       true
     end
 
+    # Runs SDAM flow on the cluster.
+    #
+    # This method can be invoked to process a new server description returned
+    # by the server on a monitoring or non-monitoring connection, and also
+    # by the driver when it marks a server unknown as a result of a (network)
+    # error.
+    #
     # @param [ Server::Description ] previous_desc Previous server description.
     # @param [ Server::Description ] updated_desc The changed description.
+    # @param [ Hash ] options Options.
+    #
+    # @option options [ true | false ] :keep_connection_pool Usually when the
+    #   new server description is unknown, the connection pool on the
+    #   respective server is cleared. Set this option to true to keep the
+    #   existing connection pool (required when handling not master errors
+    #   on 4.2+ servers).
     #
     # @api private
-    def run_sdam_flow(previous_desc, updated_desc)
+    def run_sdam_flow(previous_desc, updated_desc, options = {})
       @sdam_flow_lock.synchronize do
         flow = SdamFlow.new(self, previous_desc, updated_desc)
         flow.server_description_changed
 
         # SDAM flow may alter the updated description - grab the final
         # version for the purposes of broadcasting if a server is available
         updated_desc = flow.updated_desc
+
+        unless options[:keep_connection_pool]
+          if flow.became_unknown?
+            servers_list.each do |server|
+              if server.address == updated_desc.address
+                server.clear_connection_pool
+              end
+            end
+          end
+        end
       end
 
       # Some updated descriptions, e.g. a mismatched me one, result in the

lib/mongo/cluster/sdam_flow.rb

@@ -28,7 +28,7 @@ class SdamFlow
     def initialize(cluster, previous_desc, updated_desc)
       @cluster = cluster
       @topology = cluster.topology
-      @previous_desc = previous_desc
+      @original_desc = @previous_desc = previous_desc
       @updated_desc = updated_desc
     end
 
@@ -48,6 +48,7 @@ def initialize(cluster, previous_desc, updated_desc)
 
     attr_reader :previous_desc
     attr_reader :updated_desc
+    attr_reader :original_desc
 
     def_delegators :topology, :replica_set_name
 
@@ -490,5 +491,12 @@ def stale_primary?
       end
       false
     end
+
+    # Returns whether the server whose description this flow processed
+    # was not previously unknown, and is now. Used to decide, in particular,
+    # whether to clear the server's connection pool.
+    def became_unknown?
+      updated_desc.unknown? && !original_desc.unknown?
+    end
   end
 end

lib/mongo/operation/shared/executable.rb

@@ -62,18 +62,13 @@ def process_result(result, server)
 
         if result.not_master? || result.node_recovering?
           if result.node_shutting_down?
-            disconnect_pool = true
+            keep_pool = false
           else
-            # Max wire version needs to be checked prior to marking the
-            # server unknown
-            disconnect_pool = !server.description.server_version_gte?('4.2')
+            # Max wire version needs to be examined while the server is known
+            keep_pool = server.description.server_version_gte?('4.2')
           end
 
-          server.unknown!
-
-          if disconnect_pool
-            server.pool.disconnect!
-          end
+          server.unknown!(keep_connection_pool: keep_pool)
 
           server.scan_semaphore.signal
         end

lib/mongo/server.rb

@@ -432,7 +432,6 @@ def handle_auth_failure!
     rescue Mongo::Error::SocketError
       # non-timeout network error
       unknown!
-      pool.disconnect!
       raise
     rescue Auth::Unauthorized
       # auth error, keep server description and topology as they are
@@ -465,18 +464,35 @@ def retry_writes?
     # Marks server unknown and publishes the associated SDAM event
     # (server description changed).
     #
+    # @param [ Hash ] options Options.
+    #
+    # @option options [ true | false ] :keep_connection_pool Usually when the
+    #   new server description is unknown, the connection pool on the
+    #   respective server is cleared. Set this option to true to keep the
+    #   existing connection pool (required when handling not master errors
+    #   on 4.2+ servers).
+    #
     # @since 2.4.0, SDAM events are sent as of version 2.7.0
-    def unknown!
+    def unknown!(options = {})
       # SDAM flow will update description on the server without in-place
       # mutations and invoke SDAM transitions as needed.
-      cluster.run_sdam_flow(description, Description.new(address))
+      cluster.run_sdam_flow(description, Description.new(address), options)
     end
 
     # @api private
     def update_description(description)
       @description = description
     end
 
+    # @api private
+    def clear_connection_pool
+      @pool_lock.synchronize do
+        if @pool
+          @pool.disconnect!
+        end
+      end
+    end
+
     # @api private
     def next_connection_id
       @connection_id_gen.next_id

lib/mongo/server/connection.rb

@@ -384,7 +384,6 @@ def deliver(message)
         # Important: timeout errors are not handled here
         rescue Error::SocketError
           @server.unknown!
-          @server.pool.disconnect!
           raise
         end
       end

spec/integration/sdam_error_handling_spec.rb

@@ -5,13 +5,63 @@
     ClientRegistry.instance.close_all_clients
   end
 
-  describe 'when there is an error during an operation' do
+  # These tests operate on specific servers, and don't work in a multi
+  # shard cluster where multiple servers are equally eligible
+  require_no_multi_shard
+
+  let(:client) { authorized_client_without_any_retries }
 
-    # These tests operate on specific servers, and don't work in a multi
-    # shard cluster where multiple servers are equally eligible
-    require_no_multi_shard
+  let(:server) { client.cluster.next_primary }
+
+  shared_examples_for 'marks server unknown' do
+    it 'marks server unknown' do
+      expect(server).not_to be_unknown
+      operation
+      expect(server).to be_unknown
+    end
+  end
 
-    let(:client) { authorized_client_without_any_retries }
+  shared_examples_for 'does not mark server unknown' do
+    it 'does not mark server unknown' do
+      expect(server).not_to be_unknown
+      operation
+      expect(server).not_to be_unknown
+    end
+  end
+
+  shared_examples_for 'requests server scan' do
+    it 'requests server scan' do
+      expect(server.scan_semaphore).to receive(:signal)
+      operation
+    end
+  end
+
+  shared_examples_for 'does not request server scan' do
+    it 'does not request server scan' do
+      expect(server.scan_semaphore).not_to receive(:signal)
+      operation
+    end
+  end
+
+  shared_examples_for 'clears connection pool' do
+    it 'clears connection pool' do
+      generation = server.pool.generation
+      operation
+      new_generation = server.pool.generation
+      expect(new_generation).to eq(generation + 1)
+    end
+  end
+
+  shared_examples_for 'does not clear connection pool' do
+    it 'does not clear connection pool' do
+      generation = server.pool.generation
+      operation
+      new_generation = server.pool.generation
+      expect(new_generation).to eq(generation)
+    end
+  end
+
+  describe 'when there is an error during an operation' do
 
     before do
       wait_for_all_servers(client.cluster)
@@ -24,63 +74,13 @@
       end
     end
 
-    let(:server) { client.cluster.next_primary }
-
     let(:operation) do
       expect_any_instance_of(Mongo::Server::Connection).to receive(:deliver).and_return(reply)
       expect do
         client.database.command(ping: 1)
       end.to raise_error(Mongo::Error::OperationFailure, exception_message)
     end
 
-    shared_examples_for 'marks server unknown' do
-      it 'marks server unknown' do
-        expect(server).not_to be_unknown
-        operation
-        expect(server).to be_unknown
-      end
-    end
-
-    shared_examples_for 'does not mark server unknown' do
-      it 'does not mark server unknown' do
-        expect(server).not_to be_unknown
-        operation
-        expect(server).not_to be_unknown
-      end
-    end
-
-    shared_examples_for 'requests server scan' do
-      it 'requests server scan' do
-        expect(server.scan_semaphore).to receive(:signal)
-        operation
-      end
-    end
-
-    shared_examples_for 'does not request server scan' do
-      it 'does not request server scan' do
-        expect(server.scan_semaphore).not_to receive(:signal)
-        operation
-      end
-    end
-
-    shared_examples_for 'clears connection pool' do
-      it 'clears connection pool' do
-        generation = server.pool.generation
-        operation
-        new_generation = server.pool.generation
-        expect(new_generation).to eq(generation + 1)
-      end
-    end
-
-    shared_examples_for 'does not clear connection pool' do
-      it 'does not clear connection pool' do
-        generation = server.pool.generation
-        operation
-        new_generation = server.pool.generation
-        expect(new_generation).to eq(generation)
-      end
-    end
-
     shared_examples_for 'not master or node recovering' do
       it_behaves_like 'marks server unknown'
       it_behaves_like 'requests server scan'
@@ -170,4 +170,39 @@
       end
     end
   end
+
+  # These tests fail intermittently in Evergreen
+  describe 'when there is an error on monitoring connection', retry: 3 do
+    let(:client) do
+      authorized_client_without_any_retries.with(
+        connect_timeout: 1, socket_timeout: 1)
+    end
+
+    let(:operation) do
+      expect(server.monitor.connection).not_to be nil
+      expect(server.monitor.connection).to receive(:ismaster).at_least(:once).and_raise(exception)
+      server.scan_semaphore.broadcast
+      6.times do
+        sleep 0.5
+        if server.unknown?
+          break
+        end
+      end
+      expect(server).to be_unknown
+    end
+
+    context 'non-timeout network error' do
+      let(:exception) { Mongo::Error::SocketError }
+
+      it_behaves_like 'marks server unknown'
+      it_behaves_like 'clears connection pool'
+    end
+
+    context 'network timeout' do
+      let(:exception) { Mongo::Error::SocketTimeoutError }
+
+      it_behaves_like 'marks server unknown'
+      it_behaves_like 'clears connection pool'
+    end
+  end
 end

spec/mongo/server/connection_spec.rb

@@ -262,6 +262,7 @@ class ConnectionSpecTestException < Exception; end
       end
     end
 
+=begin These assertions require a working cluster with working SDAM flow, which the tests do not configure
     shared_examples_for 'does not disconnect connection pool' do
       it 'does not disconnect non-monitoring sockets' do
         allow(server).to receive(:pool).and_return(pool)
@@ -277,6 +278,7 @@ class ConnectionSpecTestException < Exception; end
         error
       end
     end
+=end
 
     let(:auth_mechanism) do
       if ClusterConfig.instance.server_version >= '3'
@@ -323,14 +325,14 @@ class ConnectionSpecTestException < Exception; end
             expect(error).to be_a(Mongo::Auth::Unauthorized)
           end
 
-          it_behaves_like 'disconnects connection pool'
+          #it_behaves_like 'disconnects connection pool'
           it_behaves_like 'keeps server type and topology'
         end
 
         # need a separate context here, otherwise disconnect expectation
         # is ignored due to allowing disconnects in the other context
         context 'checking pool disconnection' do
-          it_behaves_like 'disconnects connection pool'
+          #it_behaves_like 'disconnects connection pool'
         end
       end
 
@@ -360,7 +362,7 @@ class ConnectionSpecTestException < Exception; end
           expect(error).to be_a(Mongo::Error::SocketTimeoutError)
         end
 
-        it_behaves_like 'does not disconnect connection pool'
+        #it_behaves_like 'does not disconnect connection pool'
         it_behaves_like 'keeps server type and topology'
       end
 
@@ -390,7 +392,7 @@ class ConnectionSpecTestException < Exception; end
           expect(error).to be_a(Mongo::Error::SocketError)
         end
 
-        it_behaves_like 'disconnects connection pool'
+        #it_behaves_like 'disconnects connection pool'
         it_behaves_like 'marks server unknown'
       end
 
@@ -731,10 +733,12 @@ class ConnectionSpecTestException < Exception; end
           expect(connection).to_not be_connected
         end
 
+=begin These assertions require a working cluster with working SDAM flow, which the tests do not configure
         it 'does not disconnect connection pool' do
           expect(server.pool).not_to receive(:disconnect!)
           result
         end
+=end
 
         it 'does not mark server unknown' do
           expect(server).not_to be_unknown