RUBY-3741 Support deprioritizing for all topologies (#2984)

Author: comandeo-mongo

lib/mongo/server_selector/base.rb

@@ -165,9 +165,8 @@ def ==(other)
       # @param [ true | false ] write_aggregation Whether we need a server that
       #   supports writing aggregations (e.g. with $merge/$out) on secondaries.
       # @param [ Array<Server> ] deprioritized A list of servers that should
-      #   be selected from only if no other servers are available. This is
-      #   used to avoid selecting the same server twice in a row when
-      #   retrying a command.
+      #   be selected from only if no other servers are available.
+      #
       # @param [ Float | nil ] :timeout Timeout in seconds for the operation,
       #   if any.
       #
@@ -287,6 +286,9 @@ def select_server(
           end
 
           server = try_select_server(cluster, write_aggregation: write_aggregation, deprioritized: deprioritized)
+          if server.nil? && deprioritized.any?
+            server = try_select_server(cluster, write_aggregation: write_aggregation, deprioritized: [])
+          end
 
           if server
             unless cluster.topology.compatible?
@@ -358,20 +360,20 @@ def try_select_server(cluster, write_aggregation: false, deprioritized: [])
 
           if is_write_supported
             # 2. If all servers support secondary writes, we respect read preference.
-            suitable_servers(cluster)
+            suitable_servers(cluster, deprioritized)
           else
             # 3. Otherwise we fallback to primary for replica set.
             [cluster.servers.detect(&:primary?)]
           end
         else
-          suitable_servers(cluster)
+          suitable_servers(cluster, deprioritized)
         end
 
         # This list of servers may be ordered in a specific way
         # by the selector (e.g. for secondary preferred, the first
         # server may be a secondary and the second server may be primary)
         # and we should take the first server here respecting the order
-        server = suitable_server(servers, deprioritized)
+        server = suitable_server(servers)
 
         if server
           if Lint.enabled?
@@ -396,12 +398,14 @@ def try_select_server(cluster, write_aggregation: false, deprioritized: [])
       # latency filtering.
       #
       # @param [ Cluster ] cluster The cluster.
+      # @param [ Array<Server> ] deprioritized A list of servers that should
+      #   be selected from only if no other servers are available.
       #
       # @return [ Array<Server> ] The candidate servers.
       #
       # @api private
-      def candidates(cluster)
-        servers = cluster.servers
+      def candidates(cluster, deprioritized = [])
+        servers = cluster.servers.reject { |s| deprioritized.include?(s) }
         servers.each do |server|
           validate_max_staleness_support!(server)
         end
@@ -420,44 +424,57 @@ def candidates(cluster)
       # Returns servers satisfying the server selector from the cluster.
       #
       # @param [ Cluster ] cluster The cluster.
+      # @param [ Array<Server> ] deprioritized A list of servers that should
+      #   be selected from only if no other servers are available.
+      #
+      # @return [ Array<Server> ] The suitable servers.
+      #
+      # @api private
+      def suitable_servers(cluster, deprioritized = [])
+        result = suitable_servers_impl(cluster, deprioritized)
+        if result.empty? && deprioritized.any?
+          result = suitable_servers_impl(cluster, [])
+        end
+        result
+      end
+
+      private
+
+      # Internal implementation of suitable_servers that applies deprioritization
+      # filtering to the candidate servers.
+      #
+      # @param [ Cluster ] cluster The cluster.
+      # @param [ Array<Server> ] deprioritized A list of servers that should
+      #   be excluded from the candidate pool.
       #
       # @return [ Array<Server> ] The suitable servers.
       #
       # @api private
-      def suitable_servers(cluster)
+      def suitable_servers_impl(cluster, deprioritized)
         if cluster.single?
-          candidates(cluster)
+          candidates(cluster, deprioritized)
         elsif cluster.sharded?
           local_threshold = local_threshold_with_cluster(cluster)
-          servers = candidates(cluster)
+          servers = candidates(cluster, deprioritized)
           near_servers(servers, local_threshold)
         elsif cluster.replica_set?
           validate_max_staleness_value!(cluster)
-          candidates(cluster)
+          candidates(cluster, deprioritized)
         else
           # Unknown cluster - no servers
           []
         end
       end
 
-      private
-
       # Returns a server from the list of servers that is suitable for
       # executing the operation.
       #
       # @param [ Array<Server> ] servers The candidate servers.
-      # @param [ Array<Server> ] deprioritized A list of servers that should
-      #  be selected from only if no other servers are available.
       #
       # @return [ Server | nil ] The suitable server or nil if no suitable
       #  server is available.
-      def suitable_server(servers, deprioritized)
-        preferred = servers - deprioritized
-        if preferred.empty?
-          servers.first
-        else
-          preferred.first
-        end
+      def suitable_server(servers)
+        servers.first
       end
 
       # Convert this server preference definition into a format appropriate

spec/runners/server_selection.rb

@@ -53,6 +53,10 @@ class Spec
         # @since 2.0.0
         attr_reader :suitable_servers
 
+        # @return [ Array<Hash> ] deprioritized_servers The set of servers that are deprioritized
+        #   and must only be selected if no other suitable server exists.
+        attr_reader :deprioritized_servers
+
         # @return [ Mongo::Cluster::Topology ] type The topology type.
         #
         # @since 2.0.0
@@ -72,6 +76,7 @@ def initialize(test_path)
           @max_staleness = @read_preference['maxStalenessSeconds']
           @candidate_servers = @test['topology_description']['servers']
           @suitable_servers = @test['suitable_servers'] || []
+          @deprioritized_servers = @test['deprioritized_servers'] || []
           @in_latency_window = @test['in_latency_window'] || []
           @type = Mongo::Cluster::Topology.const_get(@test['topology_description']['type'])
         end
@@ -123,7 +128,7 @@ def candidate_servers
   end
 end
 
-def define_server_selection_spec_tests(test_paths)
+def define_server_selection_spec_tests(test_paths, skipped_tests = {})
   # Linter insists that a server selection semaphore is present when
   # performing server selection.
   require_no_linting
@@ -132,6 +137,13 @@ def define_server_selection_spec_tests(test_paths)
 
     spec = Mongo::ServerSelection::Read::Spec.new(file)
 
+    if skipped_tests.keys.include?(File.basename(file))
+      it spec.description do
+        skip("Skipped due to #{skipped_tests[File.basename(file)]}")
+      end
+      next
+    end
+
     context(spec.description) do
       # Cluster needs a topology and topology needs a cluster...
       # This temporary cluster is used for topology construction.
@@ -234,6 +246,13 @@ def define_server_selection_spec_tests(test_paths)
         end
       end
 
+      let(:deprioritized_servers) do
+        spec.deprioritized_servers.collect do |server|
+          Mongo::Server.new(Mongo::Address.new(server['address']), cluster, monitoring, listeners,
+            options.merge(monitoring_io: false))
+        end
+      end
+
       let(:in_latency_window) do
         spec.in_latency_window.collect do |server|
           Mongo::Server.new(Mongo::Address.new(server['address']), cluster, monitoring, listeners,
@@ -286,21 +305,28 @@ def define_server_selection_spec_tests(test_paths)
           if spec.in_latency_window.length == 1
 
             it 'selects the expected server' do
-              [server_selector.select_server(cluster)].should == in_latency_window
+              [
+                server_selector.select_server(cluster, deprioritized: deprioritized_servers)
+              ].should == in_latency_window
             end
 
           else
 
             it 'selects a server in the suitable list' do
-              in_latency_window.should include(server_selector.select_server(cluster))
+              expect(in_latency_window)
+                .to include(
+                      server_selector.select_server(
+                        cluster,
+                        deprioritized: deprioritized_servers)
+                    )
             end
 
             let(:expected_addresses) do
               in_latency_window.map(&:address).map(&:seed).sort
             end
 
             let(:actual_addresses) do
-              server_selector.suitable_servers(cluster).map(&:address).map(&:seed).sort
+              server_selector.suitable_servers(cluster, deprioritized_servers).map(&:address).map(&:seed).sort
             end
 
             it 'identifies expected suitable servers' do
@@ -317,7 +343,7 @@ def define_server_selection_spec_tests(test_paths)
             end
 
             let(:actual_addresses) do
-              servers = server_selector.send(:suitable_servers, cluster)
+              servers = server_selector.send(:suitable_servers, cluster, deprioritized_servers)
 
               # The tests expect that only secondaries are "suitable" for
               # server selection with secondary preferred read preference.
@@ -354,7 +380,7 @@ def define_server_selection_spec_tests(test_paths)
 
           it 'Raises a NoServerAvailable Exception' do
             expect do
-              server_selector.select_server(cluster)
+              server_selector.select_server(cluster, deprioritized: deprioritized_servers)
             end.to raise_exception(Mongo::Error::NoServerAvailable)
           end
 

spec/spec_tests/data/server_selection/ReplicaSetWithPrimary/read/DeprioritizedAllPrimaryPreferred.yml

@@ -0,0 +1,34 @@
+topology_description:
+  type: ReplicaSetWithPrimary
+  servers:
+  - &1
+    address: b:27017
+    avg_rtt_ms: 5
+    type: RSSecondary
+    tags:
+      data_center: nyc
+  - &2
+    address: c:27017
+    avg_rtt_ms: 5
+    type: RSSecondary
+    tags:
+      data_center: nyc
+  - &3
+    address: a:27017
+    avg_rtt_ms: 5
+    type: RSPrimary
+    tags:
+      data_center: nyc
+operation: read
+read_preference:
+  mode: PrimaryPreferred
+  tag_sets:
+  - {}
+deprioritized_servers:
+- *1
+- *2
+- *3
+suitable_servers:
+- *3
+in_latency_window:
+- *3

spec/spec_tests/data/server_selection/ReplicaSetWithPrimary/read/DeprioritizedAllSecondaryPreferred.yml

@@ -0,0 +1,36 @@
+topology_description:
+  type: ReplicaSetWithPrimary
+  servers:
+  - &1
+    address: b:27017
+    avg_rtt_ms: 5
+    type: RSSecondary
+    tags:
+      data_center: nyc
+  - &2
+    address: c:27017
+    avg_rtt_ms: 5
+    type: RSSecondary
+    tags:
+      data_center: nyc
+  - &3
+    address: a:27017
+    avg_rtt_ms: 5
+    type: RSPrimary
+    tags:
+      data_center: nyc
+operation: read
+read_preference:
+  mode: SecondaryPreferred
+  tag_sets:
+  - {}
+deprioritized_servers:
+- *1
+- *2
+- *3
+suitable_servers:
+- *1
+- *2
+in_latency_window:
+- *1
+- *2

spec/spec_tests/data/server_selection/ReplicaSetWithPrimary/read/DeprioritizedNearest.yml

@@ -0,0 +1,33 @@
+topology_description:
+  type: ReplicaSetWithPrimary
+  servers:
+  - &1
+    address: b:27017
+    avg_rtt_ms: 5
+    type: RSSecondary
+    tags:
+      data_center: nyc
+  - &2
+    address: c:27017
+    avg_rtt_ms: 100
+    type: RSSecondary
+    tags:
+      data_center: nyc
+  - &3
+    address: a:27017
+    avg_rtt_ms: 26
+    type: RSPrimary
+    tags:
+      data_center: nyc
+operation: read
+read_preference:
+  mode: Nearest
+  tag_sets:
+  - data_center: nyc
+deprioritized_servers:
+- *1
+suitable_servers:
+- *3
+- *2
+in_latency_window:
+- *3

spec/spec_tests/data/server_selection/ReplicaSetWithPrimary/read/DeprioritizedNearestStateChange.yml

@@ -0,0 +1,41 @@
+topology_description:
+  type: ReplicaSetWithPrimary
+  servers:
+    - address: b:27017
+      avg_rtt_ms: 5
+      type: RSSecondary
+      tags:
+        data_center: nyc
+    - &2
+      address: c:27017
+      avg_rtt_ms: 100
+      type: RSSecondary
+      tags:
+        data_center: nyc
+    - &3
+      address: a:27017
+      avg_rtt_ms: 25
+      type: RSPrimary
+      tags:
+        data_center: nyc
+
+operation: read
+
+read_preference:
+  mode: Nearest
+  tag_sets:
+    - data_center: nyc
+
+deprioritized_servers:
+  - address: "b:27017"
+    avg_rtt_ms: 50
+    type: RSPrimary
+    tags:
+      data_center: nyc
+
+suitable_servers:
+  - *2
+  - *3
+
+in_latency_window:
+  - *3