CSHARP-1823: Update MaxStaleness implementation to reflect recent spec changes.

Author: rstam

src/MongoDB.Driver.Core/Core/Clusters/ServerSelectors/ReadPreferenceServerSelector.cs

@@ -46,7 +46,7 @@ public static ReadPreferenceServerSelector Primary
         #endregion
 
         // fields
-        private readonly TimeSpan? _maxStaleness; // with Zero and InfiniteTimespan converted to null
+        private readonly TimeSpan? _maxStaleness; // with InfiniteTimespan converted to null
         private readonly ReadPreference _readPreference;
 
         // constructors
@@ -57,7 +57,7 @@ public static ReadPreferenceServerSelector Primary
         public ReadPreferenceServerSelector(ReadPreference readPreference)
         {
             _readPreference = Ensure.IsNotNull(readPreference, nameof(readPreference));
-            if (readPreference.MaxStaleness == TimeSpan.Zero || readPreference.MaxStaleness == Timeout.InfiniteTimeSpan)
+            if (readPreference.MaxStaleness == Timeout.InfiniteTimeSpan)
             {
                 _maxStaleness = null;
             }
@@ -132,53 +132,42 @@ private IEnumerable<ServerDescription> SelectByTagSets(IEnumerable<ServerDescrip
 
         private IEnumerable<ServerDescription> SelectForReplicaSet(ClusterDescription cluster, IEnumerable<ServerDescription> servers)
         {
-            if (_maxStaleness.HasValue)
-            {
-                var minHeartBeatIntervalTicks = servers.Select(s => s.HeartbeatInterval.Ticks).Min();
-                if (_maxStaleness.Value.Ticks < 2 * minHeartBeatIntervalTicks)
-                {
-                    throw new MongoClientException("MaxStaleness must be at least twice the heartbeat interval.");
-                }
+            EnsureMaxStalenessIsValid(cluster);
 
-                servers = new CachedEnumerable<ServerDescription>(SelectFreshServers(cluster, servers)); // prevent multiple enumeration
-            }
-            else
-            {
-                servers = new CachedEnumerable<ServerDescription>(servers); // prevent multiple enumeration
-            }
+            servers = new CachedEnumerable<ServerDescription>(servers); // prevent multiple enumeration
 
             switch (_readPreference.ReadPreferenceMode)
             {
                 case ReadPreferenceMode.Primary:
-                    return servers.Where(n => n.Type == ServerType.ReplicaSetPrimary);
+                    return SelectPrimary(servers);
 
                 case ReadPreferenceMode.PrimaryPreferred:
-                    var primary = servers.FirstOrDefault(n => n.Type == ServerType.ReplicaSetPrimary);
-                    if (primary != null)
+                    var primary = SelectPrimary(servers);
+                    if (primary.Count != 0)
                     {
-                        return new[] { primary };
+                        return primary;
                     }
                     else
                     {
-                        return SelectByTagSets(servers.Where(n => n.Type == ServerType.ReplicaSetSecondary));
+                        return SelectByTagSets(SelectFreshSecondaries(cluster, servers));
                     }
 
                 case ReadPreferenceMode.Secondary:
-                    return SelectByTagSets(servers.Where(n => n.Type == ServerType.ReplicaSetSecondary));
+                    return SelectByTagSets(SelectFreshSecondaries(cluster, servers));
 
                 case ReadPreferenceMode.SecondaryPreferred:
-                    var matchingSecondaries = SelectByTagSets(servers.Where(n => n.Type == ServerType.ReplicaSetSecondary)).ToList();
-                    if (matchingSecondaries.Count != 0)
+                    var selectedSecondaries = SelectByTagSets(SelectFreshSecondaries(cluster, servers)).ToList();
+                    if (selectedSecondaries.Count != 0)
                     {
-                        return matchingSecondaries;
+                        return selectedSecondaries;
                     }
                     else
                     {
-                        return servers.Where(n => n.Type == ServerType.ReplicaSetPrimary);
+                        return SelectPrimary(servers);
                     }
 
                 case ReadPreferenceMode.Nearest:
-                    return SelectByTagSets(servers.Where(n => n.Type == ServerType.ReplicaSetPrimary || n.Type == ServerType.ReplicaSetSecondary));
+                    return SelectByTagSets(SelectPrimary(servers).Concat(SelectFreshSecondaries(cluster, servers)));
 
                 default:
                     throw new ArgumentException("Invalid ReadPreferenceMode.");
@@ -195,44 +184,93 @@ private IEnumerable<ServerDescription> SelectForStandaloneCluster(IEnumerable<Se
             return servers.Where(n => n.Type == ServerType.Standalone); // standalone servers match any ReadPreference (to facilitate testing)
         }
 
-        private IReadOnlyList<ServerDescription> SelectFreshServers(ClusterDescription cluster, IEnumerable<ServerDescription> servers)
+        private List<ServerDescription> SelectPrimary(IEnumerable<ServerDescription> servers)
         {
-            var primary = cluster.Servers.SingleOrDefault(s => s.Type == ServerType.ReplicaSetPrimary);
-            if (primary == null)
+            var primary = servers.Where(s => s.Type == ServerType.ReplicaSetPrimary).ToList();
+            if (primary.Count > 1)
             {
-                return SelectFreshServersWithNoPrimary(cluster, servers);
+                throw new MongoClientException($"More than one primary found: [{string.Join(", ", servers.Select(s => s.ToString()))}].");
+            }
+            return primary; // returned as a list because otherwise some callers would have to create a new list
+        }
+
+        private IEnumerable<ServerDescription> SelectSecondaries(IEnumerable<ServerDescription> servers)
+        {
+            return servers.Where(s => s.Type == ServerType.ReplicaSetSecondary);
+        }
+
+        private IEnumerable<ServerDescription> SelectFreshSecondaries(ClusterDescription cluster, IEnumerable<ServerDescription> servers)
+        {
+            var secondaries = SelectSecondaries(servers);
+
+            if (_maxStaleness.HasValue)
+            {
+                var primary = SelectPrimary(cluster.Servers).SingleOrDefault();
+                if (primary == null)
+                {
+                    return SelectFreshSecondariesWithNoPrimary(secondaries);
+                }
+                else
+                {
+
+                    return SelectFreshSecondariesWithPrimary(primary, secondaries);
+                }
             }
             else
             {
-                return SelectFreshServersWithPrimary(cluster, primary, servers);
+                return secondaries;
             }
         }
 
-        private IReadOnlyList<ServerDescription> SelectFreshServersWithNoPrimary(ClusterDescription cluster, IEnumerable<ServerDescription> servers)
+        private IEnumerable<ServerDescription> SelectFreshSecondariesWithNoPrimary(IEnumerable<ServerDescription> secondaries)
         {
-            var smax = servers
-                .Where(s => s.Type == ServerType.ReplicaSetSecondary)
+            var smax = secondaries
                 .OrderByDescending(s => s.LastWriteTimestamp)
                 .FirstOrDefault();
-            return servers
+            if (smax == null)
+            {
+                return Enumerable.Empty<ServerDescription>();
+            }
+
+            return secondaries
                 .Where(s =>
                 {
                     var estimatedStaleness = smax.LastWriteTimestamp.Value - s.LastWriteTimestamp.Value + s.HeartbeatInterval;
                     return estimatedStaleness <= _maxStaleness;
-                })
-                .ToList();
+                });
         }
 
-        private IReadOnlyList<ServerDescription> SelectFreshServersWithPrimary(ClusterDescription cluster, ServerDescription primary, IEnumerable<ServerDescription> servers)
+        private IEnumerable<ServerDescription> SelectFreshSecondariesWithPrimary(ServerDescription primary, IEnumerable<ServerDescription> secondaries)
         {
             var p = primary;
-            return servers
+            return secondaries
                 .Where(s =>
-                {   
+                {
                     var estimatedStaleness = (s.LastUpdateTimestamp - s.LastWriteTimestamp.Value) - (p.LastUpdateTimestamp - p.LastWriteTimestamp.Value) + s.HeartbeatInterval;
                     return estimatedStaleness <= _maxStaleness;
-                })
-                .ToList();
+                });
+        }
+
+        private void EnsureMaxStalenessIsValid(ClusterDescription cluster)
+        {
+            if (_maxStaleness.HasValue)
+            {
+                var primary = SelectPrimary(cluster.Servers).SingleOrDefault();
+                var primaryIdleWritePeriod = primary?.IdleWritePeriod;
+
+                foreach (var server in cluster.Servers)
+                {
+                    if (server.Type == ServerType.ReplicaSetPrimary || server.Type == ServerType.ReplicaSetSecondary)
+                    {
+                        var heartbeatInterval = server.HeartbeatInterval;
+                        var idleWriteTime = primaryIdleWritePeriod ?? server.IdleWritePeriod;
+                        if (_maxStaleness.Value < heartbeatInterval + idleWriteTime)
+                        {
+                            throw new Exception("Max staleness must greater than or equal to heartbeat interval plus idle write period.");
+                        }
+                    }
+                }
+            }
         }
     }
 }

src/MongoDB.Driver.Core/Core/Configuration/ConnectionString.cs

@@ -490,7 +490,7 @@ private void Parse()
 
         private void ParseOption(string name, string value)
         {
-            switch (name.ToLower())
+            switch (name.ToLowerInvariant())
             {
                 case "appname":
                     string invalidApplicationNameMessage;
@@ -554,8 +554,12 @@ private void ParseOption(string name, string value)
                     _maxPoolSize = ParseInt32(name, value);
                     break;
                 case "maxstaleness":
-                case "maxstalenessms":
+                case "maxstalenessseconds":
                     _maxStaleness = ParseTimeSpan(name, value);
+                    if (_maxStaleness.Value == TimeSpan.FromSeconds(-1))
+                    {
+                        _maxStaleness = null;
+                    }
                     break;
                 case "minpoolsize":
                     _minPoolSize = ParseInt32(name, value);
@@ -763,6 +767,10 @@ private static TimeSpan ParseTimeSpan(string name, string value)
             {
                 multiplier = 1;
             }
+            else if (lowerName.EndsWith("seconds", StringComparison.Ordinal))
+            {
+                multiplier = 1000;
+            }
             else if (lowerValue.EndsWith("ms", StringComparison.Ordinal))
             {
                 lowerValue = lowerValue.Substring(0, lowerValue.Length - 2);

src/MongoDB.Driver.Core/Core/Connections/IsMasterResult.cs

@@ -63,6 +63,25 @@ public ElectionId ElectionId
             }
         }
 
+        /// <summary>
+        /// Gets the idle write period.
+        /// </summary>
+        public TimeSpan IdleWritePeriod
+        {
+            get
+            {
+                BsonValue value;
+                if (_wrapped.TryGetValue("idleWritePeriodMS", out value))
+                {
+                    return TimeSpan.FromMilliseconds(value.ToDouble());
+                }
+                else
+                {
+                    return TimeSpan.FromSeconds(10);
+                }
+            }
+        }
+
         /// <summary>
         /// Gets a value indicating whether this instance is an arbiter.
         /// </summary>

src/MongoDB.Driver.Core/Core/Misc/Ensure.cs

@@ -1,4 +1,4 @@
-/* Copyright 2013-2015 MongoDB Inc.
+/* Copyright 2013-2016 MongoDB Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
@@ -192,6 +192,22 @@ public static TimeSpan IsInfiniteOrGreaterThanOrEqualToZero(TimeSpan value, stri
             return value;
         }
 
+        /// <summary>
+        /// Ensures that the value of a parameter is infinite or greater than zero.
+        /// </summary>
+        /// <param name="value">The value of the parameter.</param>
+        /// <param name="paramName">The name of the parameter.</param>
+        /// <returns>The value of the parameter.</returns>
+        public static TimeSpan IsInfiniteOrGreaterThanZero(TimeSpan value, string paramName)
+        {
+            if (value == Timeout.InfiniteTimeSpan || value > TimeSpan.Zero)
+            {
+                return value;
+            }
+            var message = string.Format("Value is not infinite or greater than zero: {0}.", TimeSpanParser.ToString(value));
+            throw new ArgumentOutOfRangeException(paramName, message);
+        }
+
         /// <summary>
         /// Ensures that the value of a parameter is not null.
         /// </summary>

src/MongoDB.Driver.Core/Core/Operations/QueryHelper.cs

@@ -1,4 +1,4 @@
-/* Copyright 2013-2014 MongoDB Inc.
+/* Copyright 2013-2016 MongoDB Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
@@ -81,7 +81,7 @@ public static BsonDocument CreateReadPreferenceDocument(ServerType serverType, R
             {
                 { "mode", modeString },
                 { "tags", tagSets, tagSets != null },
-                { "maxStalenessMS", () => (long)readPreference.MaxStaleness.Value.TotalMilliseconds, readPreference.MaxStaleness.HasValue }
+                { "maxStalenessSeconds", () => readPreference.MaxStaleness.Value.TotalSeconds, readPreference.MaxStaleness.HasValue }
             };
         }
     }