diff --git a/.mod/drivers-evergreen-tools b/.mod/drivers-evergreen-tools index d8aa120a15..18aed4f176 160000 --- a/.mod/drivers-evergreen-tools +++ b/.mod/drivers-evergreen-tools @@ -1 +1 @@ -Subproject commit d8aa120a1590a68bfb337b98da1d240914a843c2 +Subproject commit 18aed4f176dab1ed505c9a2a53466ddf3f4796ec diff --git a/lib/mongo/server.rb b/lib/mongo/server.rb index 1373428e8c..b852032a95 100644 --- a/lib/mongo/server.rb +++ b/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. diff --git a/lib/mongo/server/connection.rb b/lib/mongo/server/connection.rb index e1acc5b43e..bfb822a903 100644 --- a/lib/mongo/server/connection.rb +++ b/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 diff --git a/lib/mongo/server/connection_pool.rb b/lib/mongo/server/connection_pool.rb index a86582596c..b467c87edb 100644 --- a/lib/mongo/server/connection_pool.rb +++ b/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) diff --git a/lib/mongo/server/monitor.rb b/lib/mongo/server/monitor.rb index 75d373ab22..5444b418b8 100644 --- a/lib/mongo/server/monitor.rb +++ b/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 = @connection_lock.synchronize { @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! diff --git a/spec/mongo/server/monitor_spec.rb b/spec/mongo/server/monitor_spec.rb index 562085bcae..ac016c15c7 100644 --- a/spec/mongo/server/monitor_spec.rb +++ b/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 # diff --git a/spec/mongo/server_spec.rb b/spec/mongo/server_spec.rb index d963601e75..d84756c29e 100644 --- a/spec/mongo/server_spec.rb +++ b/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'