RUBY-3909 Close monitoring connection when server marked Unknown from network error

lib/mongo/server.rb

@@ -489,7 +489,8 @@ def handle_handshake_failure!
         unknown!(
           generation: e.generation,
           service_id: e.service_id,
-          stop_push_monitor: true
+          stop_push_monitor: true,
+          network_error: true
         )
       end
       raise
@@ -514,11 +515,13 @@ def handle_auth_failure!
       raise
     rescue Mongo::Error::SocketError, Auth::Unauthorized => e
       # non-timeout network error or auth error, clear the pool and mark the
-      # topology as unknown
+      # topology as unknown. Only a network error makes the monitor connection
+      # suspect; a pure authentication failure must not close it.
       unknown!(
         generation: e.generation,
         service_id: e.service_id,
-        stop_push_monitor: true
+        stop_push_monitor: true,
+        network_error: e.is_a?(Mongo::Error::SocketError)
       )
       raise
     end
@@ -568,8 +571,15 @@ def retry_writes?
     #   on 4.2+ servers).
     # @option options [ TopologyVersion ] :topology_version Topology version
     #   of the error response that is causing the server to be marked unknown.
-    # @option options [ true | false ] :stop_push_monitor Whether to stop
-    #   the PushMonitor associated with the server, if any.
+    # @option options [ true | false ] :stop_push_monitor Set when the server
+    #   is marked Unknown from a connection-establishment error (e.g. an
+    #   authentication failure). Stops the streaming PushMonitor associated with
+    #   the server, if any.
+    # @option options [ true | false ] :network_error Set when the server is
+    #   marked Unknown from a network error. In addition to stopping the
+    #   PushMonitor, cancels the monitor's in-progress check and closes its
+    #   connection so the next check must establish a fresh one, per the Server
+    #   Monitoring spec.
     # @option options [ Object ] :service_id Discard state for the specified
     #   service id only.
     #
@@ -605,7 +615,17 @@ def unknown!(options = {})
         return
       end
 
-      monitor&.stop_push_monitor! if options[:stop_push_monitor]
+      if options[:network_error]
+        # A network error implies every connection to the server is suspect
+        # (Server Monitoring spec, "hello or legacy hello Cancellation"): cancel
+        # the monitor's in-progress check and close its connection so the next
+        # check must establish a fresh one.
+        monitor&.cancel_check!
+      elsif options[:stop_push_monitor]
+        # Non-network errors (e.g. an authentication failure) only tear down the
+        # streaming PushMonitor; the monitor's connection is not suspect.
+        monitor&.stop_push_monitor!
+      end
 
       # SDAM flow will update description on the server without in-place
       # mutations and invoke SDAM transitions as needed.

lib/mongo/server/connection.rb

@@ -423,7 +423,8 @@ def handle_errors
           generation: e.generation,
           # or description.service_id?
           service_id: e.service_id,
-          stop_push_monitor: true
+          stop_push_monitor: true,
+          network_error: true
         )
         raise
       rescue Error::SocketTimeoutError => e

lib/mongo/server/connection_pool.rb

@@ -1058,7 +1058,11 @@ def connect_connection(connection, context = nil)
           @server.unknown!(
             generation: e.generation,
             service_id: e.service_id,
-            stop_push_monitor: true
+            stop_push_monitor: true,
+            # Only a network error makes the monitor connection suspect and
+            # requires cancelling its check; an auth failure does not.
+            network_error: e.is_a?(Mongo::Error::SocketError) ||
+              e.is_a?(Mongo::Error::SocketTimeoutError)
           )
         end
         connection.disconnect!(reason: :error)

lib/mongo/server/monitor.rb

@@ -86,14 +86,23 @@ def initialize(server, event_listeners, monitoring, options = {})
         @sdam_mutex = Mutex.new
         @next_earliest_scan = @next_wanted_scan = Time.now
         @update_mutex = Mutex.new
+        # Guards reads and writes of @connection so the polling connection can
+        # be cancelled from another thread. Per the Server Monitoring spec's
+        # cancelCheck pseudocode, the lock is only held long enough to copy or
+        # assign the reference - never across the blocking check.
+        @connection_lock = Mutex.new
       end
 
       # @return [ Server ] server The server that this monitor is monitoring.
       # @api private
       attr_reader :server
 
-      # @return [ Mongo::Server::Monitor::Connection ] connection The connection to use.
-      attr_reader :connection
+      # @return [ Mongo::Server::Monitor::Connection | nil ] The connection to
+      #   use, read under @connection_lock so callers never observe a stale
+      #   reference after a concurrent cancel_check! clears it.
+      def connection
+        @connection_lock.synchronize { @connection }
+      end
 
       # @return [ Hash ] options The server options.
       attr_reader :options
@@ -159,7 +168,11 @@ def stop!
         end
       end
 
-      def create_push_monitor!(topology_version)
+      # @param [ Mongo::Server::Monitor::Connection ] connection The freshly
+      #   established monitoring connection. Passed explicitly (rather than read
+      #   from @connection) so a concurrent cancel_check! cannot nil it out from
+      #   under us between establishing it and building the PushMonitor.
+      def create_push_monitor!(topology_version, connection)
         @update_mutex.synchronize do
           @push_monitor = nil if @push_monitor && !@push_monitor.running?
 
@@ -170,7 +183,7 @@ def create_push_monitor!(topology_version)
             **Utils.shallow_symbolize_keys(options.merge(
                                              socket_timeout: heartbeat_interval + connection.socket_timeout,
                                              app_metadata: options[:push_monitor_app_metadata],
-                                             check_document: @connection.check_document
+                                             check_document: connection.check_document
                                            ))
           )
         end
@@ -185,6 +198,31 @@ def stop_push_monitor!
         end
       end
 
+      # Cancel the in-progress check and close the monitoring connection.
+      #
+      # Called when the server is marked Unknown from a network error, per the
+      # Server Monitoring spec ("hello or legacy hello Cancellation"). Stops the
+      # streaming PushMonitor (interrupting its awaited hello read) and closes
+      # the polling connection, so the next check must establish a fresh one
+      # rather than re-validating the server over a possibly-dead socket.
+      #
+      # @api private
+      def cancel_check!
+        stop_push_monitor!
+
+        # Copy the connection reference under the lock, then interrupt and close
+        # it outside the lock. Closing the socket interrupts any in-progress
+        # read on the monitor thread; nil-ing the reference forces the next
+        # check to reconnect. The monitor thread is the only writer of a new
+        # connection, so it is safe for this thread to clear it.
+        connection = @connection_lock.synchronize do
+          conn = @connection
+          @connection = nil
+          conn
+        end
+        connection&.disconnect!
+      end
+
       # Perform a check of the server with throttling, and update
       # the server's description and average round trip time.
       #
@@ -310,7 +348,7 @@ def do_scan(publish_heartbeat: true)
       #
       # @return [ true | false ]
       def rtt_measurement_only?
-        return false if @connection.nil?
+        return false if connection.nil?
 
         # Only suppress the check while the server is in a known state and the
         # PushMonitor is the authoritative streaming source. If the server is
@@ -325,23 +363,30 @@ def rtt_measurement_only?
       end
 
       def check
-        if @connection && @connection.pid != Process.pid
-          log_warn("Detected PID change - Mongo client should have been reconnected (old pid #{@connection.pid}, new pid #{Process.pid}")
-          @connection.disconnect!
-          @connection = nil
+        # Snapshot the connection under the lock. A concurrent cancel_check!
+        # may nil @connection from another thread; working on a local copy keeps
+        # this check consistent, and the guarded writeback below never clobbers
+        # a connection the monitor thread did not itself establish.
+        connection = self.connection
+
+        if connection && connection.pid != Process.pid
+          log_warn("Detected PID change - Mongo client should have been reconnected (old pid #{connection.pid}, new pid #{Process.pid}")
+          connection.disconnect!
+          clear_connection(connection)
+          connection = nil
         end
 
-        if @connection
+        if connection
           result = server.round_trip_time_calculator.measure do
-            doc = @connection.check_document
+            doc = connection.check_document
             cmd = Protocol::Query.new(
               Database::ADMIN, Database::COMMAND, doc, limit: -1
             )
-            message = @connection.dispatch_bytes(cmd.serialize.to_s)
+            message = connection.dispatch_bytes(cmd.serialize.to_s)
             message.documents.first
           rescue Mongo::Error
-            @connection.disconnect!
-            @connection = nil
+            connection.disconnect!
+            clear_connection(connection)
             raise
           end
         else
@@ -350,10 +395,13 @@ def check
           result = server.round_trip_time_calculator.measure do
             connection.handshake!
           end
-          @connection = connection
+          store_connection(connection)
           if (tv_doc = result['topologyVersion'])
             if streaming_enabled?
-              create_push_monitor!(TopologyVersion.new(tv_doc))
+              # Run the instance we just created rather than re-reading the
+              # push_monitor getter: a concurrent cancel_check! may have nil'd
+              # @push_monitor between the two calls.
+              push_monitor = create_push_monitor!(TopologyVersion.new(tv_doc), connection)
               push_monitor.run!
             else
               stop_push_monitor!
@@ -367,6 +415,22 @@ def check
         result
       end
 
+      # Store a freshly established monitoring connection.
+      def store_connection(connection)
+        @connection_lock.synchronize do
+          @connection = connection
+        end
+      end
+
+      # Clear the monitoring connection, but only if it is still the one passed
+      # in. A concurrent cancel_check! may have already cleared or replaced it,
+      # in which case we must leave the current connection alone.
+      def clear_connection(connection)
+        @connection_lock.synchronize do
+          @connection = nil if @connection.equal?(connection)
+        end
+      end
+
       # @note If the system clock is set to a time in the past, this method
       #   can sleep for a very long time.
       def throttle_scan_frequency!

spec/mongo/server/monitor_spec.rb

@@ -269,6 +269,83 @@
     end
   end
 
+  describe '#cancel_check!' do
+    # Per the Server Monitoring spec ("hello or legacy hello Cancellation"),
+    # when a server is marked Unknown from an application network error the
+    # monitor MUST cancel its in-progress check and close the current
+    # monitoring connection, so the next check runs over a freshly established
+    # connection.
+
+    context 'when a monitoring connection is established' do
+      let(:existing_connection) do
+        double('monitor connection').tap do |conn|
+          allow(conn).to receive(:disconnect!)
+        end
+      end
+
+      before do
+        monitor.instance_variable_set(:@connection, existing_connection)
+      end
+
+      it 'closes the monitoring connection' do
+        expect(existing_connection).to receive(:disconnect!)
+        monitor.cancel_check!
+      end
+
+      it 'clears the monitoring connection so the next check reconnects' do
+        monitor.cancel_check!
+        expect(monitor.connection).to be_nil
+      end
+    end
+
+    context 'when there is no monitoring connection' do
+      it 'does not raise' do
+        expect(monitor.connection).to be_nil
+        expect { monitor.cancel_check! }.not_to raise_error
+      end
+    end
+
+    context 'when a push monitor is running' do
+      let(:running_push_monitor) do
+        double('push monitor').tap do |push_monitor|
+          allow(push_monitor).to receive(:running?).and_return(true)
+          allow(push_monitor).to receive(:stop!)
+        end
+      end
+
+      before do
+        monitor.instance_variable_set(:@push_monitor, running_push_monitor)
+      end
+
+      it 'stops the push monitor' do
+        expect(running_push_monitor).to receive(:stop!)
+        monitor.cancel_check!
+        expect(monitor.push_monitor).to be_nil
+      end
+    end
+
+    context 'when in the polling protocol with a live server' do
+      let(:monitor_options) do
+        { server_monitoring_mode: :poll }
+      end
+
+      before do
+        monitor.scan!
+        expect(monitor.connection).not_to be_nil
+      end
+
+      it 'establishes a fresh connection on the next check' do
+        old_connection = monitor.connection
+        monitor.cancel_check!
+        expect(monitor.connection).to be_nil
+
+        monitor.scan!
+        expect(monitor.connection).not_to be_nil
+        expect(monitor.connection).not_to equal(old_connection)
+      end
+    end
+  end
+
   # heartbeat interval is now taken out of cluster, monitor has no useful options
   #   describe '#heartbeat_frequency' do
   #

spec/mongo/server_spec.rb

@@ -104,6 +104,50 @@
     end
   end
 
+  describe '#unknown!' do
+    # Per the Server Monitoring spec, marking a server Unknown from a network
+    # error MUST cancel the monitor's in-progress check and close its
+    # connection. A non-network connection error (e.g. an auth failure) only
+    # tears down the streaming PushMonitor.
+    let(:monitor) do
+      double('monitor').tap do |monitor|
+        allow(monitor).to receive(:cancel_check!)
+        allow(monitor).to receive(:stop_push_monitor!)
+        # Called by server teardown (register_server -> close -> monitor.stop!).
+        allow(monitor).to receive(:stop!)
+      end
+    end
+
+    before do
+      allow(server).to receive(:monitor).and_return(monitor)
+      allow(cluster).to receive(:run_sdam_flow)
+    end
+
+    context 'when the network_error option is true' do
+      it 'cancels the monitor check' do
+        expect(monitor).to receive(:cancel_check!)
+        expect(monitor).not_to receive(:stop_push_monitor!)
+        server.unknown!(stop_push_monitor: true, network_error: true)
+      end
+    end
+
+    context 'when only the stop_push_monitor option is true' do
+      it 'stops the push monitor but does not cancel the check' do
+        expect(monitor).to receive(:stop_push_monitor!)
+        expect(monitor).not_to receive(:cancel_check!)
+        server.unknown!(stop_push_monitor: true)
+      end
+    end
+
+    context 'when neither option is set' do
+      it 'does not touch the monitor' do
+        expect(monitor).not_to receive(:cancel_check!)
+        expect(monitor).not_to receive(:stop_push_monitor!)
+        server.unknown!
+      end
+    end
+  end
+
   describe '#disconnect!' do
     context 'with monitoring io' do
       include_context 'with monitoring io'