Merge pull request #157 from saghm/set-version

Implement using `setVersion` to detect stale primaries

Author: saghm

src/topology/mod.rs

@@ -60,6 +60,8 @@ pub struct TopologyDescription {
     // If true, all servers in the topology fall within the compatible
     // mongodb version for this driver.
     compatible: bool,
+    // The largest set version seen from a primary in the topology.
+    max_set_version: Option<i64>,
     compat_error: String,
 }
 
@@ -98,6 +100,7 @@ impl TopologyDescription {
             max_election_id: None,
             compatible: true,
             compat_error: String::new(),
+            max_set_version: None,
         }
     }
 
@@ -560,23 +563,29 @@ impl TopologyDescription {
             return;
         }
 
-        if description.election_id.is_some() {
-            if self.max_election_id.is_some() &&
-                self.max_election_id.as_ref().unwrap() > description.election_id.as_ref().unwrap() {
-                    // Stale primary
-                    if let Some(server) = self.servers.get(&host) {
-                        {
-                            let mut server_description = server.description.write().unwrap();
-                            server_description.server_type = ServerType::Unknown;
-                            server_description.set_name = String::new();
-                            server_description.election_id = None;
+        if description.set_version.is_some() && description.election_id.is_some() {
+            if self.max_set_version.is_some() && self.max_election_id.is_some() &&
+                   (self.max_set_version.unwrap() > description.set_version.unwrap() ||
+                        (self.max_set_version.unwrap() == description.set_version.unwrap() &&
+                        self.max_election_id.as_ref().unwrap() > description.election_id.as_ref().unwrap())) {
+                            // Stale primary
+                            if let Some(server) = self.servers.get(&host) {
+                                {
+                                    let mut server_description = server.description.write().unwrap();
+                                    server_description.server_type = ServerType::Unknown;
+                                    server_description.set_name = String::new();
+                                    server_description.election_id = None;
+                                }
+                            }
+                            self.check_if_has_primary();
+                            return;
+                        } else {
+                            self.max_election_id = description.election_id.clone();
                         }
-                    }
-                    self.check_if_has_primary();
-                    return;
-                } else {
-                    self.max_election_id = description.election_id.clone();
-                }
+        }
+
+        if description.set_version.is_some() && (self.max_set_version.is_none() || description.set_version.unwrap() > self.max_set_version.unwrap()) {
+            self.max_set_version = description.set_version;
         }
 
         // Invalidate any old primaries

src/topology/monitor.rs

@@ -53,6 +53,7 @@ pub struct IsMasterResult {
     pub election_id: Option<oid::ObjectId>,
     pub primary: Option<Host>,
     pub hidden: bool,
+    pub set_version: Option<i64>,
 }
 
 /// Monitors and updates server and topology information.
@@ -112,6 +113,7 @@ impl IsMasterResult {
             election_id: None,
             primary: None,
             hidden: false,
+            set_version: None,
         };
 
         if let Some(&Bson::Boolean(b)) = doc.get("ismaster") {
@@ -183,6 +185,10 @@ impl IsMasterResult {
             result.hidden = *h;
         }
 
+        if let Some(&Bson::I64(v)) = doc.get("setVersion") {
+            result.set_version = Some(v);
+        }
+
         if let Some(&Bson::Document(ref doc)) = doc.get("tags") {
             for (k, v) in doc.into_iter() {
                 if let &Bson::String(ref tag) = v {

src/topology/server.rs

@@ -71,6 +71,8 @@ pub struct ServerDescription {
     pub election_id: Option<oid::ObjectId>,
     /// The server's opinion of who the primary is.
     pub primary: Option<Host>,
+    /// The current replica set version number.
+    pub set_version: Option<i64>,
 }
 
 /// Holds status and connection information about a single server.
@@ -119,6 +121,7 @@ impl ServerDescription {
             set_name: String::new(),
             election_id: None,
             primary: None,
+            set_version: None,
         }
     }
 
@@ -139,6 +142,7 @@ impl ServerDescription {
         self.set_name = ismaster.set_name;
         self.election_id = ismaster.election_id;
         self.primary = ismaster.primary;
+        self.set_version = ismaster.set_version;
         self.round_trip_time = match self.round_trip_time {
             Some(old_rtt) => {
                 // (rtt / div) + (old_rtt * (div-1)/div)