RUBY-1563 Poll SRV records in unknown topologies (#1524)

* RUBY-1563 Monitor unknown topologies as the spec requires

* RUBY-1563 Add a docstring

* RUBY-1563 Test unknown to sharded transition

* RUBY-1563 Stop SRV monitor in RS and single topologies

* RUBY-1563 Test unknown topologies in reconnect test

* Wait for server check to complete to ensure topology is changed

* Give up on jruby here

Author: p-mongo

lib/mongo/cluster.rb

@@ -220,6 +220,8 @@ def initialize(seeds, monitoring, options = Options::Redacted.new)
           server_selection_semaphore.wait([time_remaining, 1].min)
         end
       end
+
+      start_stop_srv_monitor
     end
 
     # Create a cluster for the provided client, for use when we don't want the
@@ -562,6 +564,8 @@ def run_sdam_flow(previous_desc, updated_desc, options = {})
             end
           end
         end
+
+        start_stop_srv_monitor
       end
 
       # Some updated descriptions, e.g. a mismatched me one, result in the
@@ -578,8 +582,6 @@ def run_sdam_flow(previous_desc, updated_desc, options = {})
       unless updated_desc.unknown?
         server_selection_semaphore.broadcast
       end
-
-      check_and_start_srv_monitor
     end
 
     # Sets the list of servers to the addresses in the provided list of address
@@ -824,19 +826,35 @@ def sessions_supported?
     end
 
     # @api private
-    def check_and_start_srv_monitor
-      return unless topology.is_a?(Topology::Sharded) && options[:srv_uri]
-      @srv_monitor_lock.synchronize do
-        unless @srv_monitor
-          monitor_options = options.merge(
-            timeout: options[:connect_timeout] || Server::CONNECT_TIMEOUT)
-          @srv_monitor = _srv_monitor = SrvMonitor.new(self, monitor_options)
-          finalizer = lambda do
-            _srv_monitor.stop!
+    def start_stop_srv_monitor
+      # SRV URI is either always given or not for a given cluster, if one
+      # wasn't given we shouldn't ever have an SRV monitor to manage.
+      return unless options[:srv_uri]
+
+      if topology.is_a?(Topology::Sharded) || topology.is_a?(Topology::Unknown)
+        # Start SRV monitor
+        @srv_monitor_lock.synchronize do
+          unless @srv_monitor
+            monitor_options = options.merge(
+              timeout: options[:connect_timeout] || Server::CONNECT_TIMEOUT)
+            @srv_monitor = _srv_monitor = SrvMonitor.new(self, monitor_options)
+            finalizer = lambda do
+              _srv_monitor.stop!
+            end
+            ObjectSpace.define_finalizer(self, finalizer)
+          end
+          @srv_monitor.run!
+        end
+      else
+        # Stop SRV monitor if running. This path is taken when the client
+        # is given an SRV URI to a standalone/replica set; when the topology
+        # is discovered, since it's not a sharded cluster, the SRV monitor
+        # needs to be stopped.
+        @srv_monitor_lock.synchronize do
+          if @srv_monitor
+            @srv_monitor.stop!
           end
-          ObjectSpace.define_finalizer(self, finalizer)
         end
-        @srv_monitor.run!
       end
     end
   end

lib/mongo/srv/monitor.rb

@@ -15,6 +15,10 @@
 module Mongo
   module SRV
 
+    # Polls SRV records for the URI that a cluster was created for and
+    # updates the list of servers in the cluster when records change.
+    #
+    # @api private
     class Monitor
       include Loggable
 

spec/integration/reconnect_spec.rb

@@ -51,10 +51,7 @@
     end
   end
 
-  context 'in sharded topology' do
-    require_topology :sharded
-    require_default_port_deployment
-    require_multi_shard
+  context 'SRV monitor thread' do
 
     let(:uri) do
       "mongodb+srv://test1.test.build.10gen.cc/?tls=#{SpecConfig.instance.ssl?}&tlsInsecure=true".tap do |uri|
@@ -75,37 +72,103 @@
           logger: logger))
     end
 
-    it 'recreates srv monitor' do
+    let(:wait_for_discovery) do
       client.cluster.next_primary
-      if BSON::Environment.jruby?
-        # Wait for jruby to start SRV monitor thread
-        sleep 1
+    end
+
+    let(:wait_for_discovery_again) do
+      client.cluster.next_primary
+    end
+
+    shared_examples_for 'recreates SRV monitor' do
+      # JRuby produces this error:
+      # RSpec::Expectations::ExpectationNotMetError: expected nil to respond to `alive?`
+      # for this assertion:
+      # expect(thread).not_to be_alive
+      # This is bizarre because if thread was nil, the earlier call to
+      # thread.kill should've similarly failed, but it doesn't.
+      fails_on_jruby
+
+      it 'recreates SRV monitor' do
+        wait_for_discovery
+
+        expect(client.cluster.topology).to be_a(expected_topology_cls)
+        thread = client.cluster.srv_monitor.instance_variable_get('@thread')
+        expect(thread).to be_alive
+
+        thread.kill
+        # context switch to let the thread get killed
+        sleep 0.1
+        expect(thread).not_to be_alive
+
+        client.reconnect
+
+        wait_for_discovery_again
+
+        new_thread = client.cluster.srv_monitor.instance_variable_get('@thread')
+        expect(new_thread).not_to eq(thread)
+        expect(new_thread).to be_alive
       end
-      expect(client.cluster.topology).to be_a(Mongo::Cluster::Topology::Sharded)
-      thread = client.cluster.srv_monitor.instance_variable_get('@thread')
-      expect(thread).to be_alive
+    end
 
-      thread.kill
-      # context switch to let the thread get killed
-      sleep 0.1
-      if BSON::Environment.jruby?
-        # jruby takes a long time here as well
-        15.times do
-          if thread.alive?
-            sleep 1
-          else
-            break
-          end
-        end
+    context 'in sharded topology' do
+      require_topology :sharded
+      require_default_port_deployment
+      require_multi_shard
+
+      let(:expected_topology_cls) { Mongo::Cluster::Topology::Sharded }
+
+      it_behaves_like 'recreates SRV monitor'
+    end
+
+    context 'in unknown topology' do
+
+      # JRuby apparently does not implement non-blocking UDP I/O which is used
+      # by RubyDNS:
+      # NotImplementedError: recvmsg_nonblock is not implemented
+      fails_on_jruby
+
+      let(:uri) do
+        "mongodb+srv://test-fake.test.build.10gen.cc/"
       end
-      expect(thread).not_to be_alive
 
-      client.reconnect
+      let(:client) do
+        ClientRegistry.instance.register_local_client(
+          Mongo::Client.new(uri, server_selection_timeout: 3.89,
+            resolv_options: {
+              nameserver: 'localhost',
+              nameserver_port: [['localhost', 5300], ['127.0.0.1', 5300]],
+            },
+            logger: logger))
+      end
 
-      client.cluster.next_primary
-      new_thread = client.cluster.srv_monitor.instance_variable_get('@thread')
-      expect(new_thread).not_to eq(thread)
-      expect(new_thread).to be_alive
+      let(:expected_topology_cls) { Mongo::Cluster::Topology::Unknown }
+
+      let(:wait_for_discovery) do
+        # Since the entire test is done in unknown topology, we cannot use
+        # next_primary to wait for the client to discover the topology.
+        sleep 5
+      end
+
+      let(:wait_for_discovery_again) do
+        sleep 5
+      end
+
+      around do |example|
+        require 'support/dns'
+
+        rules = [
+          ['_mongodb._tcp.test-fake.test.build.10gen.cc', :srv,
+            [0, 0, 2799, 'localhost.test.build.10gen.cc'],
+          ],
+        ]
+
+        mock_dns(rules) do
+          example.run
+        end
+      end
+
+      it_behaves_like 'recreates SRV monitor'
     end
   end
 end