layout history merge: rm invalid versions when valid versions are added

2 files changed, 28 insertions, 10 deletions

diff --git a/src/rpc/layout/history.rs b/src/rpc/layout/history.rs
index dce492c9..2346b14a 100644
--- a/src/rpc/layout/history.rs
+++ b/src/rpc/layout/history.rs
@@ -211,6 +211,24 @@ impl LayoutHistory {
 			changed = changed || c;
 		}
 
+		// If there are invalid versions before valid versions, remove them,
+		// and increment update trackers
+		if self.versions.len() > 1 && self.current().check().is_ok() {
+			while self.versions.first().unwrap().check().is_err() {
+				self.versions.remove(0);
+				changed = true;
+			}
+			if changed {
+				let min_v = self.versions.first().unwrap().version;
+				let nodes = self.all_nongateway_nodes().into_owned();
+				for node in nodes {
+					self.update_trackers.ack_map.set_max(node, min_v);
+					self.update_trackers.sync_map.set_max(node, min_v);
+					self.update_trackers.sync_ack_map.set_max(node, min_v);
+				}
+			}
+		}
+
 		// Merge staged layout changes
 		if self.staging != other.staging {
 			self.staging.merge(&other.staging);
diff --git a/src/rpc/layout/version.rs b/src/rpc/layout/version.rs
index 912ee538..947fab56 100644
--- a/src/rpc/layout/version.rs
+++ b/src/rpc/layout/version.rs
@@ -174,6 +174,16 @@ impl LayoutVersion {
 	/// (assignment, roles, parameters, partition size)
 	/// returns true if consistent, false if error
 	pub fn check(&self) -> Result<(), String> {
+		// Check that the assignment data has the correct length
+		let expected_assignment_data_len = (1 << PARTITION_BITS) * self.replication_factor;
+		if self.ring_assignment_data.len() != expected_assignment_data_len {
+			return Err(format!(
+				"ring_assignment_data has incorrect length {} instead of {}",
+				self.ring_assignment_data.len(),
+				expected_assignment_data_len
+			));
+		}
+
 		// Check that node_id_vec contains the correct list of nodes
 		let mut expected_nodes = self
 			.roles
@@ -189,16 +199,6 @@ impl LayoutVersion {
 			return Err(format!("node_id_vec does not contain the correct set of nodes\nnode_id_vec: {:?}\nexpected: {:?}", node_id_vec, expected_nodes));
 		}
 
-		// Check that the assignment data has the correct length
-		let expected_assignment_data_len = (1 << PARTITION_BITS) * self.replication_factor;
-		if self.ring_assignment_data.len() != expected_assignment_data_len {
-			return Err(format!(
-				"ring_assignment_data has incorrect length {} instead of {}",
-				self.ring_assignment_data.len(),
-				expected_assignment_data_len
-			));
-		}
-
 		// Check that the assigned nodes are correct identifiers
 		// of nodes that are assigned a role
 		// and that role is not the role of a gateway nodes