SWIFT-1456 Implement selectServer for MongoClient (#737)

Author: isabelatkinson

Package.swift

@@ -15,7 +15,6 @@ let package = Package(
         .package(url: "https://github.com/mongodb/swift-bson", .upToNextMajor(from: "3.0.0"))
     ],
     targets: [
-        .target(name: "MongoSwift", dependencies: ["CLibMongoC", "NIO", "NIOConcurrencyHelpers", "SwiftBSON",]),
         .target(name: "MongoSwiftSync", dependencies: ["MongoSwift", "NIO"]),
         .target(name: "AtlasConnectivity", dependencies: ["MongoSwiftSync"]),
         .target(name: "TestsCommon", dependencies: ["MongoSwift", "Nimble"]),
@@ -34,3 +33,10 @@ let package = Package(
         )
     ]
 )
+
+#if compiler(>=5.3)
+package.dependencies += [.package(url: "https://github.com/apple/swift-atomics", .upToNextMajor(from: "1.0.0"))]
+package.targets += [.target(name: "MongoSwift", dependencies: ["Atomics", "CLibMongoC", "NIO", "NIOConcurrencyHelpers", "SwiftBSON"])]
+#else
+package.targets += [.target(name: "MongoSwift", dependencies: ["CLibMongoC", "NIO", "NIOConcurrencyHelpers", "SwiftBSON"])]
+#endif

Sources/MongoSwift/MongoClient.swift

@@ -248,6 +248,9 @@ public class MongoClient {
     /// The pool of connections backing this client.
     internal let connectionPool: ConnectionPool
 
+    /// The connection string used to create this client.
+    internal let connectionString: MongoConnectionString
+
     /// Executor responsible for executing operations on behalf of this client and its child objects.
     internal let operationExecutor: OperationExecutor
 
@@ -320,6 +323,7 @@ public class MongoClient {
             try connString.applyOptions(options)
         }
         try connString.validate()
+        self.connectionString = connString
 
         self.operationExecutor = OperationExecutor(
             eventLoopGroup: eventLoopGroup,

Sources/MongoSwift/SDAM.swift

@@ -5,6 +5,10 @@ internal enum SDAMConstants {
     internal static let defaultHeartbeatFrequencyMS = 10000
     internal static let idleWritePeriodMS = 10000
     internal static let smallestMaxStalenessSeconds = 90
+    internal static let defaultLocalThresholdMS = 15
+    internal static let defaultServerSelectionTimeoutMS = 30000
+    internal static let minWireVersion = 6 // 3.6
+    internal static let maxWireVersion = 15 // 5.2
 }
 
 /// A struct representing a network address, consisting of a host and port.
@@ -243,20 +247,21 @@ public struct ServerDescription {
         tags: [String: String]?,
         lastWriteDate: Date?,
         maxWireVersion: Int?,
-        lastUpdateTime: Date?
+        lastUpdateTime: Date?,
+        averageRoundTripTimeMS: Double?
     ) {
         self.address = address
         self.type = type
         self.tags = tags ?? [:]
         self.lastWriteDate = lastWriteDate
         self.lastUpdateTime = lastUpdateTime ?? Date()
-        self.maxWireVersion = maxWireVersion ?? 0
+        self.maxWireVersion = maxWireVersion ?? SDAMConstants.maxWireVersion
+        self.averageRoundTripTimeMS = averageRoundTripTimeMS
 
         // these fields are not used by the server selection tests
         self.serverId = 0
         self.roundTripTime = 0
-        self.averageRoundTripTimeMS = nil
-        self.minWireVersion = 0
+        self.minWireVersion = SDAMConstants.minWireVersion
         self.me = self.address
         self.setName = nil
         self.setVersion = nil
@@ -339,7 +344,7 @@ public struct TopologyDescription: Equatable {
 
         /// Internal representation of topology type. If enums could be marked non-exhaustive in Swift, this would be
         /// the public representation too.
-        fileprivate enum _TopologyType: String, Equatable {
+        internal enum _TopologyType: String, Equatable {
             case single = "Single"
             case replicaSetNoPrimary = "ReplicaSetNoPrimary"
             case replicaSetWithPrimary = "ReplicaSetWithPrimary"
@@ -348,7 +353,7 @@ public struct TopologyDescription: Equatable {
             case loadBalanced = "LoadBalanced"
         }
 
-        fileprivate let _topologyType: _TopologyType
+        internal let _topologyType: _TopologyType
 
         private init(_ _type: _TopologyType) {
             self._topologyType = _type
@@ -429,191 +434,3 @@ public struct TopologyDescription: Equatable {
         self.servers = servers
     }
 }
-
-extension TopologyDescription {
-    internal func findSuitableServers(
-        readPreference: ReadPreference? = nil,
-        heartbeatFrequencyMS: Int
-    ) throws -> [ServerDescription] {
-        try readPreference?.validateMaxStalenessSeconds(
-            heartbeatFrequencyMS: heartbeatFrequencyMS,
-            topologyType: self.type
-        )
-        switch self.type._topologyType {
-        case .unknown:
-            return []
-        case .single, .loadBalanced:
-            return self.servers
-        case .replicaSetNoPrimary, .replicaSetWithPrimary:
-            switch readPreference?.mode {
-            case .secondary:
-                return self.filterReplicaSetServers(
-                    readPreference: readPreference,
-                    heartbeatFrequencyMS: heartbeatFrequencyMS,
-                    includePrimary: false
-                )
-            case .nearest:
-                return self.filterReplicaSetServers(
-                    readPreference: readPreference,
-                    heartbeatFrequencyMS: heartbeatFrequencyMS,
-                    includePrimary: true
-                )
-            case .secondaryPreferred:
-                // If mode is 'secondaryPreferred', attempt the selection algorithm with mode 'secondary' and the
-                // user's maxStalenessSeconds and tag_sets. If no server matches, select the primary.
-                let secondaryMatches = self.filterReplicaSetServers(
-                    readPreference: readPreference,
-                    heartbeatFrequencyMS: heartbeatFrequencyMS,
-                    includePrimary: false
-                )
-                return secondaryMatches.isEmpty ? self.servers.filter { $0.type == .rsPrimary } : secondaryMatches
-            case .primaryPreferred:
-                // If mode is 'primaryPreferred' or a readPreference is not provided, select the primary if it is known,
-                // otherwise attempt the selection algorithm with mode 'secondary' and the user's
-                // maxStalenessSeconds and tag_sets.
-                let primaries = self.servers.filter { $0.type == .rsPrimary }
-                if !primaries.isEmpty {
-                    return primaries
-                }
-                return self.filterReplicaSetServers(
-                    readPreference: readPreference,
-                    heartbeatFrequencyMS: heartbeatFrequencyMS,
-                    includePrimary: false
-                )
-            default: // or .primary
-                // the default mode is 'primary'.
-                return self.servers.filter { $0.type == .rsPrimary }
-            }
-        case .sharded:
-            return self.servers.filter { $0.type == .mongos }
-        }
-    }
-
-    /// Filters the replica set servers in this topology first by max staleness and then by tag sets.
-    private func filterReplicaSetServers(
-        readPreference: ReadPreference?,
-        heartbeatFrequencyMS: Int,
-        includePrimary: Bool
-    ) -> [ServerDescription] {
-        // The initial set of servers from which to filter. Only include the secondaries unless includePrimary is true.
-        var servers = self.servers.filter { ($0.type == .rsPrimary && includePrimary) || $0.type == .rsSecondary }
-
-        // Filter by max staleness. If maxStalenessSeconds is not configured as a positive number, all servers are
-        // eligible.
-        if let maxStalenessSeconds = readPreference?.maxStalenessSeconds, maxStalenessSeconds > 0 {
-            let primary = self.servers.first { $0.type == .rsPrimary }
-            let maxLastWriteDate = self.getMaxLastWriteDate()
-            servers.removeAll {
-                guard let staleness = $0.calculateStalenessSeconds(
-                    primary: primary,
-                    maxLastWriteDate: maxLastWriteDate,
-                    heartbeatFrequencyMS: heartbeatFrequencyMS
-                ) else {
-                    return false
-                }
-                return staleness > maxStalenessSeconds
-            }
-        }
-
-        // Filter by tag sets.
-        guard let tagSets = readPreference?.tagSets else {
-            return servers
-        }
-        for tagSet in tagSets {
-            let matches = servers.filter { server in tagSet.allSatisfy { server.tags[$0.key] == $0.value.stringValue } }
-            if !matches.isEmpty {
-                return matches
-            }
-        }
-
-        // If no matches were found during tag set filtering, return an empty list.
-        return []
-    }
-
-    /// Returns a `Date` representing the latest `lastWriteDate` configured on a secondary in the topology, or `nil`
-    /// if none is found.
-    private func getMaxLastWriteDate() -> Date? {
-        let secondaryLastWriteDates = self.servers.compactMap {
-            $0.type == .rsSecondary ? $0.lastWriteDate : nil
-        }
-        return secondaryLastWriteDates.max()
-    }
-}
-
-extension ServerDescription {
-    /// Calculates the staleness of this server. If the server is not a secondary, the staleness is 0. Otherwise,
-    /// compare against the primary if one is present, or the maximum last write date seen in the topology if present.
-    /// If staleness cannot be calculated due to an absence of values, `nil` is returned.
-    fileprivate func calculateStalenessSeconds(
-        primary: ServerDescription?,
-        maxLastWriteDate: Date?,
-        heartbeatFrequencyMS: Int
-    ) -> Int? {
-        guard self.type == .rsSecondary else {
-            return 0
-        }
-        guard let lastWriteDate = self.lastWriteDate else {
-            return nil
-        }
-        if let primary = primary {
-            guard let primaryLastWriteDate = primary.lastWriteDate else {
-                return nil
-            }
-            let selfInterval = self.lastUpdateTime.timeIntervalSince(lastWriteDate)
-            let primaryInterval = primary.lastUpdateTime.timeIntervalSince(primaryLastWriteDate)
-            // timeIntervalSince returns a TimeInterval in seconds, so heartbeatFrequencyMS needs to be converted from
-            // milliseconds to seconds.
-            let stalenessSeconds = selfInterval - primaryInterval + Double(heartbeatFrequencyMS) / 1000.0
-            return Int(stalenessSeconds.rounded(.up))
-        } else {
-            guard let maxLastWriteDate = maxLastWriteDate else {
-                return nil
-            }
-            let interval = maxLastWriteDate.timeIntervalSince(lastWriteDate)
-            let stalenessSeconds = interval + Double(heartbeatFrequencyMS) / 1000.0
-            return Int(stalenessSeconds.rounded(.up))
-        }
-    }
-}
-
-extension ReadPreference {
-    fileprivate func validateMaxStalenessSeconds(
-        heartbeatFrequencyMS: Int,
-        topologyType: TopologyDescription.TopologyType
-    ) throws {
-        if let maxStalenessSeconds = self.maxStalenessSeconds {
-            if self.mode == .primary && maxStalenessSeconds > 0 {
-                throw MongoError.InvalidArgumentError(
-                    message: "A positive maxStalenessSeconds cannot be specified when the read preference mode is"
-                        + " primary"
-                )
-            }
-            if topologyType == .replicaSetWithPrimary || topologyType == .replicaSetNoPrimary {
-                if maxStalenessSeconds * 1000 < heartbeatFrequencyMS + SDAMConstants.idleWritePeriodMS {
-                    throw MongoError.InvalidArgumentError(
-                        message: "maxStalenessSeconds must be at least the sum of the heartbeatFrequencyMS configured"
-                            + " on the client (\(heartbeatFrequencyMS)) and the idleWritePeriodMS"
-                            + " (\(SDAMConstants.idleWritePeriodMS))"
-                    )
-                }
-                if maxStalenessSeconds < SDAMConstants.smallestMaxStalenessSeconds {
-                    throw MongoError.InvalidArgumentError(
-                        message: "The maxStalenessSeconds configured for a replica set must be at least"
-                            + " \(SDAMConstants.smallestMaxStalenessSeconds)"
-                    )
-                }
-            }
-        }
-    }
-}
-
-extension ServerDescription {
-    internal mutating func updateAverageRoundTripTime(roundTripTime: Double) {
-        if let oldAverageRTT = self.averageRoundTripTimeMS {
-            let alpha = 0.2
-            self.averageRoundTripTimeMS = alpha * roundTripTime + (1 - alpha) * oldAverageRTT
-        } else {
-            self.averageRoundTripTimeMS = roundTripTime
-        }
-    }
-}