Merge pull request #836 from estolfo/RUBY-1174-update-max-staleness

RUBY-1174 RUBY-1178 Update Max Staleness implementation

Author: estolfo

lib/mongo/cluster.rb

@@ -37,6 +37,11 @@ class Cluster
     # @since 2.1.1
     READ_RETRY_INTERVAL = 5
 
+    # How often an idle primary writes a no-op to the oplog.
+    #
+    # @since 2.4.0
+    IDLE_WRITE_PERIOD_SECONDS = 10
+
     # @return [ Hash ] The options hash.
     attr_reader :options
 

lib/mongo/error/invalid_server_preference.rb

@@ -33,7 +33,9 @@ class InvalidServerPreference < Error
       # Error message for when the max staleness is not at least twice the heartbeat frequency.
       #
       # @since 2.4.0
-      INVALID_MAX_STALENESS = "max_staleness must be at least twice the client's heartbeat frequency.".freeze
+      INVALID_MAX_STALENESS = "`max_staleness` value is too small. It must be at least " +
+        "`ServerSelector::SMALLEST_MAX_STALENESS_SECONDS` and (the cluster's heartbeat_frequency " +
+          "setting + `Cluster::IDLE_WRITE_PERIOD_SECONDS`).".freeze
 
       # Error message when max staleness cannot be used because one or more servers has version < 3.4.
       #

lib/mongo/server.rb

@@ -45,6 +45,7 @@ class Server
 
     # Get the description from the monitor and scan on monitor.
     def_delegators :monitor, :description, :scan!, :heartbeat_frequency, :last_scan
+    alias :heartbeat_frequency_seconds :heartbeat_frequency
 
     # Delegate convenience methods to the monitor description.
     def_delegators :description,

lib/mongo/server_selector.rb

@@ -38,6 +38,11 @@ module ServerSelector
     # @since 2.0.0
     SERVER_SELECTION_TIMEOUT = 30.freeze
 
+    # The smallest allowed max staleness value, in seconds.
+    #
+    # @since 2.4.0
+    SMALLEST_MAX_STALENESS_SECONDS = 90
+
     # Primary read preference.
     #
     # @since 2.1.0

lib/mongo/server_selector/nearest.rb

@@ -67,7 +67,7 @@ def tags_allowed?
       def to_mongos
         preference = { :mode => 'nearest' }
         preference.merge!({ :tags => tag_sets }) unless tag_sets.empty?
-        preference.merge!({ maxStalenessMS: max_staleness * 1000 }) if max_staleness
+        preference.merge!({ maxStalenessSeconds: max_staleness }) if max_staleness
         preference
       end
 

lib/mongo/server_selector/primary_preferred.rb

@@ -68,7 +68,7 @@ def tags_allowed?
       def to_mongos
         preference = { :mode => 'primaryPreferred' }
         preference.merge!({ :tags => tag_sets }) unless tag_sets.empty?
-        preference.merge!({ maxStalenessMS: max_staleness * 1000 }) if max_staleness
+        preference.merge!({ maxStalenessSeconds: max_staleness }) if max_staleness
         preference
       end
 

lib/mongo/server_selector/secondary.rb

@@ -68,7 +68,7 @@ def tags_allowed?
       def to_mongos
         preference = { :mode => 'secondary' }
         preference.merge!({ :tags => tag_sets }) unless tag_sets.empty?
-        preference.merge!({ maxStalenessMS: max_staleness * 1000 }) if max_staleness
+        preference.merge!({ maxStalenessSeconds: max_staleness }) if max_staleness
         preference
       end
 

lib/mongo/server_selector/secondary_preferred.rb

@@ -71,7 +71,7 @@ def to_mongos
         return nil if tag_sets.empty? && max_staleness.nil?
         preference = { mode: 'secondaryPreferred' }
         preference.merge!({ tags: tag_sets }) unless tag_sets.empty?
-        preference.merge!({ maxStalenessMS: max_staleness * 1000 }) if max_staleness
+        preference.merge!({ maxStalenessSeconds: max_staleness }) if max_staleness
         preference
       end
 

lib/mongo/server_selector/selectable.rb

@@ -26,7 +26,7 @@ module Selectable
       # @return [ Array ] tag_sets The tag sets used to select servers.
       attr_reader :tag_sets
 
-      # @return [ Float ] max_staleness The maximum replication lag, in seconds, that a
+      # @return [ Integer ] max_staleness The maximum replication lag, in seconds, that a
       #   secondary can suffer and still be eligible for a read.
       #
       # @since 2.4.0
@@ -68,7 +68,7 @@ def ==(other)
       def initialize(options = {})
         @options = (options || {}).freeze
         @tag_sets = (options[:tag_sets] || []).freeze
-        @max_staleness = options[:max_staleness] if options[:max_staleness] && options[:max_staleness] > 0
+        @max_staleness = options[:max_staleness] unless options[:max_staleness] == -1
         validate!
       end
 
@@ -154,7 +154,7 @@ def candidates(cluster)
         elsif cluster.sharded?
           near_servers(cluster.servers).each { |server| validate_max_staleness_support!(server) }
         else
-          validate_max_staleness_value!(cluster)
+          validate_max_staleness_value!(cluster) unless cluster.unknown?
           select(cluster.servers)
         end
       end
@@ -230,14 +230,14 @@ def filter_stale_servers(candidates, primary = nil)
             validate_max_staleness_support!(server)
             staleness = (server.last_scan - server.last_write_date) -
                         (primary.last_scan - primary.last_write_date)  +
-                        (server.heartbeat_frequency * 1000)
+                        (server.heartbeat_frequency_seconds * 1000)
             staleness <= max_staleness_ms
           end
         else
           max_write_date = candidates.collect(&:last_write_date).max
           candidates.select do |server|
             validate_max_staleness_support!(server)
-            staleness = max_write_date - server.last_write_date + (server.heartbeat_frequency * 1000)
+            staleness = max_write_date - server.last_write_date + (server.heartbeat_frequency_seconds * 1000)
             staleness <= max_staleness_ms
           end
         end
@@ -258,10 +258,12 @@ def validate_max_staleness_support!(server)
       end
 
       def validate_max_staleness_value!(cluster)
-        return unless @max_staleness
-        heartbeat_frequency = cluster.options[:heartbeat_frequency] || Server::Monitor::HEARTBEAT_FREQUENCY
-        if @max_staleness < heartbeat_frequency * 2
-          raise Error::InvalidServerPreference.new(Error::InvalidServerPreference::INVALID_MAX_STALENESS)
+        if @max_staleness
+          heartbeat_frequency_seconds = cluster.options[:heartbeat_frequency] || Server::Monitor::HEARTBEAT_FREQUENCY
+          unless @max_staleness >= [ SMALLEST_MAX_STALENESS_SECONDS,
+                                     (heartbeat_frequency_seconds  + Cluster::IDLE_WRITE_PERIOD_SECONDS) ].max
+            raise Error::InvalidServerPreference.new(Error::InvalidServerPreference::INVALID_MAX_STALENESS)
+          end
         end
       end
     end

lib/mongo/uri.rb

@@ -374,7 +374,7 @@ def self.uri_option(uri_key, name, extra = {})
     # Read Options
     uri_option 'readpreference', :mode, :group => :read, :type => :read_mode
     uri_option 'readpreferencetags', :tag_sets, :group => :read, :type => :read_tags
-    uri_option 'maxstalenessms', :max_staleness, :group => :read, :type => :ms_convert
+    uri_option 'maxstalenessseconds', :max_staleness, :group => :read
 
     # Pool options
     uri_option 'minpoolsize', :min_pool_size