Improve how node roles are assigned in Garage

- change the terminology: the network configuration becomes the role
  table, the configuration of a nodes becomes a node's role
- the modification of the role table takes place in two steps: first,
  changes are staged in a CRDT data structure. Then, once the user is
  happy with the changes, they can commit them all at once (or revert
  them).
- update documentation
- fix tests
- implement smarter partition assignation algorithm

This patch breaks the format of the network configuration: when
migrating, the cluster will be in a state where no roles are assigned.
All roles must be re-assigned and commited at once. This migration
should not pose an issue.

src/rpc/Cargo.toml

@@ -1,11 +1,12 @@
 [package]
 name = "garage_rpc"
-version = "0.4.0"
+version = "0.5.0"
 authors = ["Alex Auvolat <alex@adnab.me>"]
 edition = "2018"
 license = "AGPL-3.0"
 description = "Cluster membership management and RPC protocol for the Garage object store"
 repository = "https://git.deuxfleurs.fr/Deuxfleurs/garage"
+readme = "../../README.md"
 
 [lib]
 path = "lib.rs"
@@ -13,7 +14,7 @@ path = "lib.rs"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-garage_util = { version = "0.4.0", path = "../util" }
+garage_util = { version = "0.5.0", path = "../util" }
 
 arc-swap = "1.0"
 bytes = "1.0"
@@ -26,6 +27,7 @@ sodiumoxide = { version = "0.2.5-0", package = "kuska-sodiumoxide" }
 async-trait = "0.1.7"
 rmp-serde = "0.15"
 serde = { version = "1.0", default-features = false, features = ["derive", "rc"] }
+serde_bytes = "0.11"
 serde_json = "1.0"
 
 futures = "0.3"

src/rpc/layout.rs

@@ -0,0 +1,579 @@
+use std::cmp::Ordering;
+use std::collections::{HashMap, HashSet};
+
+use serde::{Deserialize, Serialize};
+
+use garage_util::crdt::{AutoCrdt, Crdt, LwwMap};
+use garage_util::data::*;
+
+use crate::ring::*;
+
+/// The layout of the cluster, i.e. the list of roles
+/// which are assigned to each cluster node
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct ClusterLayout {
+	pub version: u64,
+
+	pub replication_factor: usize,
+	pub roles: LwwMap<Uuid, NodeRoleV>,
+
+	/// node_id_vec: a vector of node IDs with a role assigned
+	/// in the system (this includes gateway nodes).
+	/// The order here is different than the vec stored by `roles`, because:
+	/// 1. non-gateway nodes are first so that they have lower numbers
+	/// 2. nodes that don't have a role are excluded (but they need to
+	///    stay in the CRDT as tombstones)
+	pub node_id_vec: Vec<Uuid>,
+	/// the assignation of data partitions to node, the values
+	/// are indices in node_id_vec
+	#[serde(with = "serde_bytes")]
+	pub ring_assignation_data: Vec<CompactNodeType>,
+
+	/// Role changes which are staged for the next version of the layout
+	pub staging: LwwMap<Uuid, NodeRoleV>,
+	pub staging_hash: Hash,
+}
+
+#[derive(PartialEq, Eq, PartialOrd, Ord, Clone, Debug, Serialize, Deserialize)]
+pub struct NodeRoleV(pub Option<NodeRole>);
+
+impl AutoCrdt for NodeRoleV {
+	const WARN_IF_DIFFERENT: bool = true;
+}
+
+/// The user-assigned roles of cluster nodes
+#[derive(PartialEq, Eq, PartialOrd, Ord, Clone, Debug, Serialize, Deserialize)]
+pub struct NodeRole {
+	/// Datacenter at which this entry belong. This information might be used to perform a better
+	/// geodistribution
+	pub zone: String,
+	/// The (relative) capacity of the node
+	/// If this is set to None, the node does not participate in storing data for the system
+	/// and is only active as an API gateway to other nodes
+	pub capacity: Option<u32>,
+	/// A set of tags to recognize the node
+	pub tags: Vec<String>,
+}
+
+impl NodeRole {
+	pub fn capacity_string(&self) -> String {
+		match self.capacity {
+			Some(c) => format!("{}", c),
+			None => "gateway".to_string(),
+		}
+	}
+}
+
+impl ClusterLayout {
+	pub fn new(replication_factor: usize) -> Self {
+		let empty_lwwmap = LwwMap::new();
+		let empty_lwwmap_hash = blake2sum(&rmp_to_vec_all_named(&empty_lwwmap).unwrap()[..]);
+
+		ClusterLayout {
+			version: 0,
+			replication_factor,
+			roles: LwwMap::new(),
+			node_id_vec: Vec::new(),
+			ring_assignation_data: Vec::new(),
+			staging: empty_lwwmap,
+			staging_hash: empty_lwwmap_hash,
+		}
+	}
+
+	pub fn merge(&mut self, other: &ClusterLayout) -> bool {
+		match other.version.cmp(&self.version) {
+			Ordering::Greater => {
+				*self = other.clone();
+				true
+			}
+			Ordering::Equal => {
+				self.staging.merge(&other.staging);
+
+				let new_staging_hash = blake2sum(&rmp_to_vec_all_named(&self.staging).unwrap()[..]);
+				let changed = new_staging_hash != self.staging_hash;
+
+				self.staging_hash = new_staging_hash;
+
+				changed
+			}
+			Ordering::Less => false,
+		}
+	}
+
+	/// Returns a list of IDs of nodes that currently have
+	/// a role in the cluster
+	pub fn node_ids(&self) -> &[Uuid] {
+		&self.node_id_vec[..]
+	}
+
+	pub fn num_nodes(&self) -> usize {
+		self.node_id_vec.len()
+	}
+
+	/// Returns the role of a node in the layout
+	pub fn node_role(&self, node: &Uuid) -> Option<&NodeRole> {
+		match self.roles.get(node) {
+			Some(NodeRoleV(Some(v))) => Some(v),
+			_ => None,
+		}
+	}
+
+	/// Check a cluster layout for internal consistency
+	/// returns true if consistent, false if error
+	pub fn check(&self) -> bool {
+		// Check that the hash of the staging data is correct
+		let staging_hash = blake2sum(&rmp_to_vec_all_named(&self.staging).unwrap()[..]);
+		if staging_hash != self.staging_hash {
+			return false;
+		}
+
+		// Check that node_id_vec contains the correct list of nodes
+		let mut expected_nodes = self
+			.roles
+			.items()
+			.iter()
+			.filter(|(_, _, v)| v.0.is_some())
+			.map(|(id, _, _)| *id)
+			.collect::<Vec<_>>();
+		expected_nodes.sort();
+		let mut node_id_vec = self.node_id_vec.clone();
+		node_id_vec.sort();
+		if expected_nodes != node_id_vec {
+			return false;
+		}
+
+		// Check that the assignation data has the correct length
+		if self.ring_assignation_data.len() != (1 << PARTITION_BITS) * self.replication_factor {
+			return false;
+		}
+
+		// 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
+		for x in self.ring_assignation_data.iter() {
+			if *x as usize >= self.node_id_vec.len() {
+				return false;
+			}
+			let node = self.node_id_vec[*x as usize];
+			match self.roles.get(&node) {
+				Some(NodeRoleV(Some(x))) if x.capacity.is_some() => (),
+				_ => return false,
+			}
+		}
+
+		true
+	}
+
+	/// Calculate an assignation of partitions to nodes
+	pub fn calculate_partition_assignation(&mut self) -> bool {
+		let (configured_nodes, zones) = self.configured_nodes_and_zones();
+		let n_zones = zones.len();
+
+		println!("Calculating updated partition assignation, this may take some time...");
+		println!();
+
+		let old_partitions = self.parse_assignation_data();
+
+		let mut partitions = old_partitions.clone();
+		for part in partitions.iter_mut() {
+			part.nodes
+				.retain(|(_, info)| info.map(|x| x.capacity.is_some()).unwrap_or(false));
+		}
+
+		// When nodes are removed, or when bootstraping an assignation from
+		// scratch for a new cluster, the old partitions will have holes (or be empty).
+		// Here we add more nodes to make a complete (sub-optimal) assignation,
+		// using an initial partition assignation that is calculated using the multi-dc maglev trick
+		match self.initial_partition_assignation() {
+			Some(initial_partitions) => {
+				for (part, ipart) in partitions.iter_mut().zip(initial_partitions.iter()) {
+					for (id, info) in ipart.nodes.iter() {
+						if part.nodes.len() < self.replication_factor {
+							part.add(part.nodes.len() + 1, n_zones, id, info.unwrap());
+						}
+					}
+					assert!(part.nodes.len() == self.replication_factor);
+				}
+			}
+			None => {
+				return false;
+			}
+		}
+
+		// Calculate how many partitions each node should ideally store,
+		// and how many partitions they are storing with the current assignation
+		// This defines our target for which we will optimize in the following loop.
+		let total_capacity = configured_nodes
+			.iter()
+			.map(|(_, info)| info.capacity.unwrap_or(0))
+			.sum::<u32>() as usize;
+		let total_partitions = self.replication_factor * (1 << PARTITION_BITS);
+		let target_partitions_per_node = configured_nodes
+			.iter()
+			.map(|(id, info)| {
+				(
+					*id,
+					info.capacity.unwrap_or(0) as usize * total_partitions / total_capacity,
+				)
+			})
+			.collect::<HashMap<&Uuid, usize>>();
+
+		let mut partitions_per_node = self.partitions_per_node(&partitions[..]);
+
+		println!("Target number of partitions per node:");
+		for (node, npart) in target_partitions_per_node.iter() {
+			println!("{:?}\t{}", node, npart);
+		}
+		println!();
+
+		// Shuffle partitions between nodes so that nodes will reach (or better approach)
+		// their target number of stored partitions
+		loop {
+			let mut option = None;
+			for (i, part) in partitions.iter_mut().enumerate() {
+				for (irm, (idrm, _)) in part.nodes.iter().enumerate() {
+					let suprm = partitions_per_node.get(*idrm).cloned().unwrap_or(0) as i32
+						- target_partitions_per_node.get(*idrm).cloned().unwrap_or(0) as i32;
+
+					for (idadd, infoadd) in configured_nodes.iter() {
+						// skip replacing a node by itself
+						// and skip replacing by gateway nodes
+						if idadd == idrm || infoadd.capacity.is_none() {
+							continue;
+						}
+
+						let supadd = partitions_per_node.get(*idadd).cloned().unwrap_or(0) as i32
+							- target_partitions_per_node.get(*idadd).cloned().unwrap_or(0) as i32;
+
+						// We want to try replacing node idrm by node idadd
+						// if that brings us close to our goal.
+						let square = |i: i32| i * i;
+						let oldcost = square(suprm) + square(supadd);
+						let newcost = square(suprm - 1) + square(supadd + 1);
+						if newcost >= oldcost {
+							// not closer to our goal
+							continue;
+						}
+						let gain = oldcost - newcost;
+
+						let mut newpart = part.clone();
+
+						newpart.nodes.remove(irm);
+						if !newpart.add(newpart.nodes.len() + 1, n_zones, idadd, infoadd) {
+							continue;
+						}
+						assert!(newpart.nodes.len() == self.replication_factor);
+
+						if !old_partitions[i]
+							.is_valid_transition_to(&newpart, self.replication_factor)
+						{
+							continue;
+						}
+
+						if option
+							.as_ref()
+							.map(|(old_gain, _, _, _, _)| gain > *old_gain)
+							.unwrap_or(true)
+						{
+							option = Some((gain, i, idadd, idrm, newpart));
+						}
+					}
+				}
+			}
+			if let Some((_gain, i, idadd, idrm, newpart)) = option {
+				*partitions_per_node.entry(idadd).or_insert(0) += 1;
+				*partitions_per_node.get_mut(idrm).unwrap() -= 1;
+				partitions[i] = newpart;
+			} else {
+				break;
+			}
+		}
+
+		// Check we completed the assignation correctly
+		// (this is a set of checks for the algorithm's consistency)
+		assert!(partitions.len() == (1 << PARTITION_BITS));
+		assert!(partitions
+			.iter()
+			.all(|p| p.nodes.len() == self.replication_factor));
+
+		let new_partitions_per_node = self.partitions_per_node(&partitions[..]);
+		assert!(new_partitions_per_node == partitions_per_node);
+
+		// Show statistics
+		println!("New number of partitions per node:");
+		for (node, npart) in partitions_per_node.iter() {
+			println!("{:?}\t{}", node, npart);
+		}
+		println!();
+
+		let mut diffcount = HashMap::new();
+		for (oldpart, newpart) in old_partitions.iter().zip(partitions.iter()) {
+			let nminus = oldpart.txtplus(newpart);
+			let nplus = newpart.txtplus(oldpart);
+			if nminus != "[...]" || nplus != "[...]" {
+				let tup = (nminus, nplus);
+				*diffcount.entry(tup).or_insert(0) += 1;
+			}
+		}
+		if diffcount.is_empty() {
+			println!("No data will be moved between nodes.");
+		} else {
+			let mut diffcount = diffcount.into_iter().collect::<Vec<_>>();
+			diffcount.sort();
+			println!("Number of partitions that move:");
+			for ((nminus, nplus), npart) in diffcount {
+				println!("\t{}\t{} -> {}", npart, nminus, nplus);
+			}
+		}
+		println!();
+
+		// Calculate and save new assignation data
+		let (nodes, assignation_data) =
+			self.compute_assignation_data(&configured_nodes[..], &partitions[..]);
+
+		self.node_id_vec = nodes;
+		self.ring_assignation_data = assignation_data;
+
+		true
+	}
+
+	fn initial_partition_assignation(&self) -> Option<Vec<PartitionAss<'_>>> {
+		let (configured_nodes, zones) = self.configured_nodes_and_zones();
+		let n_zones = zones.len();
+
+		// Create a vector of partition indices (0 to 2**PARTITION_BITS-1)
+		let partitions_idx = (0usize..(1usize << PARTITION_BITS)).collect::<Vec<_>>();
+
+		// Prepare ring
+		let mut partitions: Vec<PartitionAss> = partitions_idx
+			.iter()
+			.map(|_i| PartitionAss::new())
+			.collect::<Vec<_>>();
+
+		// Create MagLev priority queues for each node
+		let mut queues = configured_nodes
+			.iter()
+			.filter(|(_id, info)| info.capacity.is_some())
+			.map(|(node_id, node_info)| {
+				let mut parts = partitions_idx
+					.iter()
+					.map(|i| {
+						let part_data =
+							[&u16::to_be_bytes(*i as u16)[..], node_id.as_slice()].concat();
+						(*i, fasthash(&part_data[..]))
+					})
+					.collect::<Vec<_>>();
+				parts.sort_by_key(|(_i, h)| *h);
+				let parts_i = parts.iter().map(|(i, _h)| *i).collect::<Vec<_>>();
+				(node_id, node_info, parts_i, 0)
+			})
+			.collect::<Vec<_>>();
+
+		let max_capacity = configured_nodes
+			.iter()
+			.filter_map(|(_, node_info)| node_info.capacity)
+			.fold(0, std::cmp::max);
+
+		// Fill up ring
+		for rep in 0..self.replication_factor {
+			queues.sort_by_key(|(ni, _np, _q, _p)| {
+				let queue_data = [&u16::to_be_bytes(rep as u16)[..], ni.as_slice()].concat();
+				fasthash(&queue_data[..])
+			});
+
+			for (_, _, _, pos) in queues.iter_mut() {
+				*pos = 0;
+			}
+
+			let mut remaining = partitions_idx.len();
+			while remaining > 0 {
+				let remaining0 = remaining;
+				for i_round in 0..max_capacity {
+					for (node_id, node_info, q, pos) in queues.iter_mut() {
+						if i_round >= node_info.capacity.unwrap() {
+							continue;
+						}
+						for (pos2, &qv) in q.iter().enumerate().skip(*pos) {
+							if partitions[qv].add(rep + 1, n_zones, node_id, node_info) {
+								remaining -= 1;
+								*pos = pos2 + 1;
+								break;
+							}
+						}
+					}
+				}
+				if remaining == remaining0 {
+					// No progress made, exit
+					return None;
+				}
+			}
+		}
+
+		Some(partitions)
+	}
+
+	fn configured_nodes_and_zones(&self) -> (Vec<(&Uuid, &NodeRole)>, HashSet<&str>) {
+		let configured_nodes = self
+			.roles
+			.items()
+			.iter()
+			.filter(|(_id, _, info)| info.0.is_some())
+			.map(|(id, _, info)| (id, info.0.as_ref().unwrap()))
+			.collect::<Vec<(&Uuid, &NodeRole)>>();
+
+		let zones = configured_nodes
+			.iter()
+			.filter(|(_id, info)| info.capacity.is_some())
+			.map(|(_id, info)| info.zone.as_str())
+			.collect::<HashSet<&str>>();
+
+		(configured_nodes, zones)
+	}
+
+	fn compute_assignation_data<'a>(
+		&self,
+		configured_nodes: &[(&'a Uuid, &'a NodeRole)],
+		partitions: &[PartitionAss<'a>],
+	) -> (Vec<Uuid>, Vec<CompactNodeType>) {
+		assert!(partitions.len() == (1 << PARTITION_BITS));
+
+		// Make a canonical order for nodes
+		let mut nodes = configured_nodes
+			.iter()
+			.filter(|(_id, info)| info.capacity.is_some())
+			.map(|(id, _)| **id)
+			.collect::<Vec<_>>();
+		let nodes_rev = nodes
+			.iter()
+			.enumerate()
+			.map(|(i, id)| (*id, i as CompactNodeType))
+			.collect::<HashMap<Uuid, CompactNodeType>>();
+
+		let mut assignation_data = vec![];
+		for partition in partitions.iter() {
+			assert!(partition.nodes.len() == self.replication_factor);
+			for (id, _) in partition.nodes.iter() {
+				assignation_data.push(*nodes_rev.get(id).unwrap());
+			}
+		}
+
+		nodes.extend(
+			configured_nodes
+				.iter()
+				.filter(|(_id, info)| info.capacity.is_none())
+				.map(|(id, _)| **id),
+		);
+
+		(nodes, assignation_data)
+	}
+
+	fn parse_assignation_data(&self) -> Vec<PartitionAss<'_>> {
+		if self.ring_assignation_data.len() == self.replication_factor * (1 << PARTITION_BITS) {
+			// If the previous assignation data is correct, use that
+			let mut partitions = vec![];
+			for i in 0..(1 << PARTITION_BITS) {
+				let mut part = PartitionAss::new();
+				for node_i in self.ring_assignation_data
+					[i * self.replication_factor..(i + 1) * self.replication_factor]
+					.iter()
+				{
+					let node_id = &self.node_id_vec[*node_i as usize];
+
+					if let Some(NodeRoleV(Some(info))) = self.roles.get(node_id) {
+						part.nodes.push((node_id, Some(info)));
+					} else {
+						part.nodes.push((node_id, None));
+					}
+				}
+				partitions.push(part);
+			}
+			partitions
+		} else {
+			// Otherwise start fresh
+			(0..(1 << PARTITION_BITS))
+				.map(|_| PartitionAss::new())
+				.collect()
+		}
+	}
+
+	fn partitions_per_node<'a>(&self, partitions: &[PartitionAss<'a>]) -> HashMap<&'a Uuid, usize> {
+		let mut partitions_per_node = HashMap::<&Uuid, usize>::new();
+		for p in partitions.iter() {
+			for (id, _) in p.nodes.iter() {
+				*partitions_per_node.entry(*id).or_insert(0) += 1;
+			}
+		}
+		partitions_per_node
+	}
+}
+
+// ---- Internal structs for partition assignation in layout ----
+
+#[derive(Clone)]
+struct PartitionAss<'a> {
+	nodes: Vec<(&'a Uuid, Option<&'a NodeRole>)>,
+}
+
+impl<'a> PartitionAss<'a> {
+	fn new() -> Self {
+		Self { nodes: Vec::new() }
+	}
+
+	fn nplus(&self, other: &PartitionAss<'a>) -> usize {
+		self.nodes
+			.iter()
+			.filter(|x| !other.nodes.contains(x))
+			.count()
+	}
+
+	fn txtplus(&self, other: &PartitionAss<'a>) -> String {
+		let mut nodes = self
+			.nodes
+			.iter()
+			.filter(|x| !other.nodes.contains(x))
+			.map(|x| format!("{:?}", x.0))
+			.collect::<Vec<_>>();
+		nodes.sort();
+		if self.nodes.iter().any(|x| other.nodes.contains(x)) {
+			nodes.push("...".into());
+		}
+		format!("[{}]", nodes.join(" "))
+	}
+
+	fn is_valid_transition_to(&self, other: &PartitionAss<'a>, replication_factor: usize) -> bool {
+		let min_keep_nodes_per_part = (replication_factor + 1) / 2;
+		let n_removed = self.nplus(other);
+
+		if self.nodes.len() <= min_keep_nodes_per_part {
+			n_removed == 0
+		} else {
+			n_removed <= self.nodes.len() - min_keep_nodes_per_part
+		}
+	}
+
+	fn add(
+		&mut self,
+		target_len: usize,
+		n_zones: usize,
+		node: &'a Uuid,
+		role: &'a NodeRole,
+	) -> bool {
+		if self.nodes.len() != target_len - 1 {
+			return false;
+		}
+
+		let p_zns = self
+			.nodes
+			.iter()
+			.map(|(_id, info)| info.unwrap().zone.as_str())
+			.collect::<HashSet<&str>>();
+		if (p_zns.len() < n_zones && !p_zns.contains(&role.zone.as_str()))
+			|| (p_zns.len() == n_zones && !self.nodes.iter().any(|(id, _)| *id == node))
+		{
+			self.nodes.push((node, Some(role)));
+			true
+		} else {
+			false
+		}
+	}
+}

src/rpc/lib.rs

@@ -5,6 +5,7 @@ extern crate log;
 
 mod consul;
 
+pub mod layout;
 pub mod ring;
 pub mod system;
 

src/rpc/ring.rs

@@ -1,12 +1,11 @@
 //! Module containing types related to computing nodes which should receive a copy of data blocks
 //! and metadata
-use std::collections::{HashMap, HashSet};
 use std::convert::TryInto;
 
-use serde::{Deserialize, Serialize};
-
 use garage_util::data::*;
 
+use crate::layout::ClusterLayout;
+
 /// A partition id, which is stored on 16 bits
 /// i.e. we have up to 2**16 partitions.
 /// (in practice we have exactly 2**PARTITION_BITS partitions)
@@ -22,47 +21,6 @@ pub const PARTITION_BITS: usize = 8;
 
 const PARTITION_MASK_U16: u16 = ((1 << PARTITION_BITS) - 1) << (16 - PARTITION_BITS);
 
-/// The user-defined configuration of the cluster's nodes
-#[derive(Clone, Debug, Serialize, Deserialize)]
-pub struct NetworkConfig {
-	/// Map of each node's id to it's configuration
-	pub members: HashMap<Uuid, NetworkConfigEntry>,
-	/// Version of this config
-	pub version: u64,
-}
-
-impl NetworkConfig {
-	pub(crate) fn new() -> Self {
-		Self {
-			members: HashMap::new(),
-			version: 0,
-		}
-	}
-}
-
-/// The overall configuration of one (possibly remote) node
-#[derive(Clone, Debug, Serialize, Deserialize)]
-pub struct NetworkConfigEntry {
-	/// Datacenter at which this entry belong. This infromation might be used to perform a better
-	/// geodistribution
-	pub zone: String,
-	/// The (relative) capacity of the node
-	/// If this is set to None, the node does not participate in storing data for the system
-	/// and is only active as an API gateway to other nodes
-	pub capacity: Option<u32>,
-	/// A tag to recognize the entry, not used for other things than display
-	pub tag: String,
-}
-
-impl NetworkConfigEntry {
-	pub fn capacity_string(&self) -> String {
-		match self.capacity {
-			Some(c) => format!("{}", c),
-			None => "gateway".to_string(),
-		}
-	}
-}
-
 /// A ring distributing fairly objects to nodes
 #[derive(Clone)]
 pub struct Ring {
@@ -70,7 +28,7 @@ pub struct Ring {
 	pub replication_factor: usize,
 
 	/// The network configuration used to generate this ring
-	pub config: NetworkConfig,
+	pub layout: ClusterLayout,
 
 	// Internal order of nodes used to make a more compact representation of the ring
 	nodes: Vec<Uuid>,
@@ -81,7 +39,7 @@ pub struct Ring {
 
 // Type to store compactly the id of a node in the system
 // Change this to u16 the day we want to have more than 256 nodes in a cluster
-type CompactNodeType = u8;
+pub type CompactNodeType = u8;
 
 // The maximum number of times an object might get replicated
 // This must be at least 3 because Garage supports 3-way replication
@@ -102,132 +60,26 @@ struct RingEntry {
 }
 
 impl Ring {
-	// TODO this function MUST be refactored, it's 100 lines long, with a 50 lines loop, going up to 6
-	// levels of imbrication. It is basically impossible to test, maintain, or understand for an
-	// outsider.
-	pub(crate) fn new(config: NetworkConfig, replication_factor: usize) -> Self {
-		// Create a vector of partition indices (0 to 2**PARTITION_BITS-1)
-		let partitions_idx = (0usize..(1usize << PARTITION_BITS)).collect::<Vec<_>>();
-
-		let zones = config
-			.members
-			.iter()
-			.filter(|(_id, info)| info.capacity.is_some())
-			.map(|(_id, info)| info.zone.as_str())
-			.collect::<HashSet<&str>>();
-		let n_zones = zones.len();
-
-		// Prepare ring
-		let mut partitions: Vec<Vec<(&Uuid, &NetworkConfigEntry)>> = partitions_idx
-			.iter()
-			.map(|_i| Vec::new())
-			.collect::<Vec<_>>();
-
-		// Create MagLev priority queues for each node
-		let mut queues = config
-			.members
-			.iter()
-			.filter(|(_id, info)| info.capacity.is_some())
-			.map(|(node_id, node_info)| {
-				let mut parts = partitions_idx
-					.iter()
-					.map(|i| {
-						let part_data =
-							[&u16::to_be_bytes(*i as u16)[..], node_id.as_slice()].concat();
-						(*i, fasthash(&part_data[..]))
-					})
-					.collect::<Vec<_>>();
-				parts.sort_by_key(|(_i, h)| *h);
-				let parts_i = parts.iter().map(|(i, _h)| *i).collect::<Vec<_>>();
-				(node_id, node_info, parts_i, 0)
-			})
-			.collect::<Vec<_>>();
-
-		let max_capacity = config
-			.members
-			.iter()
-			.filter_map(|(_, node_info)| node_info.capacity)
-			.fold(0, std::cmp::max);
-
-		assert!(replication_factor <= MAX_REPLICATION);
-
-		// Fill up ring
-		for rep in 0..replication_factor {
-			queues.sort_by_key(|(ni, _np, _q, _p)| {
-				let queue_data = [&u16::to_be_bytes(rep as u16)[..], ni.as_slice()].concat();
-				fasthash(&queue_data[..])
-			});
-
-			for (_, _, _, pos) in queues.iter_mut() {
-				*pos = 0;
-			}
-
-			let mut remaining = partitions_idx.len();
-			while remaining > 0 {
-				let remaining0 = remaining;
-				for i_round in 0..max_capacity {
-					for (node_id, node_info, q, pos) in queues.iter_mut() {
-						if i_round >= node_info.capacity.unwrap() {
-							continue;
-						}
-						for (pos2, &qv) in q.iter().enumerate().skip(*pos) {
-							if partitions[qv].len() != rep {
-								continue;
-							}
-							let p_zns = partitions[qv]
-								.iter()
-								.map(|(_id, info)| info.zone.as_str())
-								.collect::<HashSet<&str>>();
-							if (p_zns.len() < n_zones && !p_zns.contains(&node_info.zone.as_str()))
-								|| (p_zns.len() == n_zones
-									&& !partitions[qv].iter().any(|(id, _i)| id == node_id))
-							{
-								partitions[qv].push((node_id, node_info));
-								remaining -= 1;
-								*pos = pos2 + 1;
-								break;
-							}
-						}
-					}
-				}
-				if remaining == remaining0 {
-					// No progress made, exit
-					warn!("Could not build ring, not enough nodes configured.");
-					return Self {
-						replication_factor,
-						config,
-						nodes: vec![],
-						ring: vec![],
-					};
-				}
-			}
+	pub(crate) fn new(layout: ClusterLayout, replication_factor: usize) -> Self {
+		if replication_factor != layout.replication_factor {
+			warn!("Could not build ring: replication factor does not match between local configuration and network role assignation.");
+			return Self::empty(layout, replication_factor);
 		}
 
-		// Make a canonical order for nodes
-		let nodes = config
-			.members
-			.iter()
-			.filter(|(_id, info)| info.capacity.is_some())
-			.map(|(id, _)| *id)
-			.collect::<Vec<_>>();
-		let nodes_rev = nodes
-			.iter()
-			.enumerate()
-			.map(|(i, id)| (*id, i as CompactNodeType))
-			.collect::<HashMap<Uuid, CompactNodeType>>();
+		if layout.ring_assignation_data.len() != replication_factor * (1 << PARTITION_BITS) {
+			warn!("Could not build ring: network role assignation data has invalid length");
+			return Self::empty(layout, replication_factor);
+		}
 
-		let ring = partitions
-			.iter()
-			.enumerate()
-			.map(|(i, nodes)| {
+		let nodes = layout.node_id_vec.clone();
+		let ring = (0..(1 << PARTITION_BITS))
+			.map(|i| {
 				let top = (i as u16) << (16 - PARTITION_BITS);
-				let nodes = nodes
-					.iter()
-					.map(|(id, _info)| *nodes_rev.get(id).unwrap())
-					.collect::<Vec<CompactNodeType>>();
-				assert!(nodes.len() == replication_factor);
 				let mut nodes_buf = [0u8; MAX_REPLICATION];
-				nodes_buf[..replication_factor].copy_from_slice(&nodes[..]);
+				nodes_buf[..replication_factor].copy_from_slice(
+					&layout.ring_assignation_data
+						[replication_factor * i..replication_factor * (i + 1)],
+				);
 				RingEntry {
 					hash_prefix: top,
 					nodes_buf,
@@ -237,12 +89,21 @@ impl Ring {
 
 		Self {
 			replication_factor,
-			config,
+			layout,
 			nodes,
 			ring,
 		}
 	}
 
+	fn empty(layout: ClusterLayout, replication_factor: usize) -> Self {
+		Self {
+			replication_factor,
+			layout,
+			nodes: vec![],
+			ring: vec![],
+		}
+	}
+
 	/// Get the partition in which data would fall on
 	pub fn partition_of(&self, position: &Hash) -> Partition {
 		let top = u16::from_be_bytes(position.as_slice()[0..2].try_into().unwrap());

src/rpc/rpc_helper.rs

@@ -225,7 +225,7 @@ impl RpcHelper {
 			// Retrieve some status variables that we will use to sort requests
 			let peer_list = self.0.fullmesh.get_peer_list();
 			let ring: Arc<Ring> = self.0.ring.borrow().clone();
-			let our_zone = match ring.config.members.get(&self.0.our_node_id) {
+			let our_zone = match ring.layout.node_role(&self.0.our_node_id) {
 				Some(pc) => &pc.zone,
 				None => "",
 			};
@@ -238,7 +238,7 @@ impl RpcHelper {
 			// and within a same zone we priorize nodes with the lowest latency.
 			let mut requests = requests
 				.map(|(to, fut)| {
-					let peer_zone = match ring.config.members.get(&to) {
+					let peer_zone = match ring.layout.node_role(&to) {
 						Some(pc) => &pc.zone,
 						None => "",
 					};

src/rpc/system.rs

@@ -23,12 +23,13 @@ use netapp::{NetApp, NetworkKey, NodeID, NodeKey};
 
 use garage_util::background::BackgroundRunner;
 use garage_util::config::Config;
-use garage_util::data::Uuid;
+use garage_util::data::*;
 use garage_util::error::*;
 use garage_util::persister::Persister;
 use garage_util::time::*;
 
 use crate::consul::*;
+use crate::layout::*;
 use crate::ring::*;
 use crate::rpc_helper::*;
 
@@ -48,13 +49,13 @@ pub enum SystemRpc {
 	Ok,
 	/// Request to connect to a specific node (in <pubkey>@<host>:<port> format)
 	Connect(String),
-	/// Ask other node its config. Answered with AdvertiseConfig
-	PullConfig,
+	/// Ask other node its cluster layout. Answered with AdvertiseClusterLayout
+	PullClusterLayout,
 	/// Advertise Garage status. Answered with another AdvertiseStatus.
 	/// Exchanged with every node on a regular basis.
 	AdvertiseStatus(NodeStatus),
-	/// Advertisement of nodes config. Sent spontanously or in response to PullConfig
-	AdvertiseConfig(NetworkConfig),
+	/// Advertisement of cluster layout. Sent spontanously or in response to PullClusterLayout
+	AdvertiseClusterLayout(ClusterLayout),
 	/// Get known nodes states
 	GetKnownNodes,
 	/// Return known nodes
@@ -70,7 +71,7 @@ pub struct System {
 	/// The id of this node
 	pub id: Uuid,
 
-	persist_config: Persister<NetworkConfig>,
+	persist_cluster_layout: Persister<ClusterLayout>,
 	persist_peer_list: Persister<Vec<(Uuid, SocketAddr)>>,
 
 	local_status: ArcSwap<NodeStatus>,
@@ -103,8 +104,10 @@ pub struct NodeStatus {
 	pub hostname: String,
 	/// Replication factor configured on the node
 	pub replication_factor: usize,
-	/// Configuration version
-	pub config_version: u64,
+	/// Cluster layout version
+	pub cluster_layout_version: u64,
+	/// Hash of cluster layout staging data
+	pub cluster_layout_staging_hash: Hash,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -187,17 +190,17 @@ impl System {
 			gen_node_key(&config.metadata_dir).expect("Unable to read or generate node ID");
 		info!("Node public key: {}", hex::encode(&node_key.public_key()));
 
-		let persist_config = Persister::new(&config.metadata_dir, "network_config");
+		let persist_cluster_layout = Persister::new(&config.metadata_dir, "cluster_layout");
 		let persist_peer_list = Persister::new(&config.metadata_dir, "peer_list");
 
-		let net_config = match persist_config.load() {
+		let cluster_layout = match persist_cluster_layout.load() {
 			Ok(x) => x,
 			Err(e) => {
 				info!(
-					"No valid previous network configuration stored ({}), starting fresh.",
+					"No valid previous cluster layout stored ({}), starting fresh.",
 					e
 				);
-				NetworkConfig::new()
+				ClusterLayout::new(replication_factor)
 			}
 		};
 
@@ -206,10 +209,11 @@ impl System {
 				.into_string()
 				.unwrap_or_else(|_| "<invalid utf-8>".to_string()),
 			replication_factor,
-			config_version: net_config.version,
+			cluster_layout_version: cluster_layout.version,
+			cluster_layout_staging_hash: cluster_layout.staging_hash,
 		};
 
-		let ring = Ring::new(net_config, replication_factor);
+		let ring = Ring::new(cluster_layout, replication_factor);
 		let (update_ring, ring) = watch::channel(Arc::new(ring));
 
 		if let Some(addr) = config.rpc_public_addr {
@@ -229,7 +233,7 @@ impl System {
 
 		let sys = Arc::new(System {
 			id: netapp.id.into(),
-			persist_config,
+			persist_cluster_layout,
 			persist_peer_list,
 			local_status: ArcSwap::new(Arc::new(local_status)),
 			node_status: RwLock::new(HashMap::new()),
@@ -292,12 +296,12 @@ impl System {
 	}
 
 	/// Save network configuration to disc
-	async fn save_network_config(self: Arc<Self>) -> Result<(), Error> {
+	async fn save_cluster_layout(self: Arc<Self>) -> Result<(), Error> {
 		let ring: Arc<Ring> = self.ring.borrow().clone();
-		self.persist_config
-			.save_async(&ring.config)
+		self.persist_cluster_layout
+			.save_async(&ring.layout)
 			.await
-			.expect("Cannot save current cluster configuration");
+			.expect("Cannot save current cluster layout");
 		Ok(())
 	}
 
@@ -305,7 +309,8 @@ impl System {
 		let mut new_si: NodeStatus = self.local_status.load().as_ref().clone();
 
 		let ring = self.ring.borrow();
-		new_si.config_version = ring.config.version;
+		new_si.cluster_layout_version = ring.layout.version;
+		new_si.cluster_layout_staging_hash = ring.layout.staging_hash;
 		self.local_status.swap(Arc::new(new_si));
 	}
 
@@ -337,9 +342,9 @@ impl System {
 		)));
 	}
 
-	fn handle_pull_config(&self) -> SystemRpc {
+	fn handle_pull_cluster_layout(&self) -> SystemRpc {
 		let ring = self.ring.borrow().clone();
-		SystemRpc::AdvertiseConfig(ring.config.clone())
+		SystemRpc::AdvertiseClusterLayout(ring.layout.clone())
 	}
 
 	fn handle_get_known_nodes(&self) -> SystemRpc {
@@ -360,7 +365,8 @@ impl System {
 					.unwrap_or(NodeStatus {
 						hostname: "?".to_string(),
 						replication_factor: 0,
-						config_version: 0,
+						cluster_layout_version: 0,
+						cluster_layout_staging_hash: Hash::from([0u8; 32]),
 					}),
 			})
 			.collect::<Vec<_>>();
@@ -381,10 +387,12 @@ impl System {
 			std::process::exit(1);
 		}
 
-		if info.config_version > local_info.config_version {
+		if info.cluster_layout_version > local_info.cluster_layout_version
+			|| info.cluster_layout_staging_hash != local_info.cluster_layout_staging_hash
+		{
 			let self2 = self.clone();
 			self.background.spawn_cancellable(async move {
-				self2.pull_config(from).await;
+				self2.pull_cluster_layout(from).await;
 				Ok(())
 			});
 		}
@@ -397,32 +405,39 @@ impl System {
 		Ok(SystemRpc::Ok)
 	}
 
-	async fn handle_advertise_config(
+	async fn handle_advertise_cluster_layout(
 		self: Arc<Self>,
-		adv: &NetworkConfig,
+		adv: &ClusterLayout,
 	) -> Result<SystemRpc, Error> {
 		let update_ring = self.update_ring.lock().await;
-		let ring: Arc<Ring> = self.ring.borrow().clone();
+		let mut layout: ClusterLayout = self.ring.borrow().layout.clone();
 
-		if adv.version > ring.config.version {
-			let ring = Ring::new(adv.clone(), self.replication_factor);
+		let prev_layout_check = layout.check();
+		if layout.merge(adv) {
+			if prev_layout_check && !layout.check() {
+				error!("New cluster layout is invalid, discarding.");
+				return Err(Error::Message(
+					"New cluster layout is invalid, discarding.".into(),
+				));
+			}
+
+			let ring = Ring::new(layout.clone(), self.replication_factor);
 			update_ring.send(Arc::new(ring))?;
 			drop(update_ring);
 
 			let self2 = self.clone();
-			let adv = adv.clone();
 			self.background.spawn_cancellable(async move {
 				self2
 					.rpc
 					.broadcast(
 						&self2.system_endpoint,
-						SystemRpc::AdvertiseConfig(adv),
+						SystemRpc::AdvertiseClusterLayout(layout),
 						RequestStrategy::with_priority(PRIO_HIGH),
 					)
 					.await;
 				Ok(())
 			});
-			self.background.spawn(self.clone().save_network_config());
+			self.background.spawn(self.clone().save_cluster_layout());
 		}
 
 		Ok(SystemRpc::Ok)
@@ -456,14 +471,15 @@ impl System {
 		};
 
 		while !*stop_signal.borrow() {
-			let not_configured = self.ring.borrow().config.members.is_empty();
+			let not_configured = !self.ring.borrow().layout.check();
 			let no_peers = self.fullmesh.get_peer_list().len() < self.replication_factor;
+			let expected_n_nodes = self.ring.borrow().layout.num_nodes();
 			let bad_peers = self
 				.fullmesh
 				.get_peer_list()
 				.iter()
 				.filter(|p| p.is_up())
-				.count() != self.ring.borrow().config.members.len();
+				.count() != expected_n_nodes;
 
 			if not_configured || no_peers || bad_peers {
 				info!("Doing a bootstrap/discovery step (not_configured: {}, no_peers: {}, bad_peers: {})", not_configured, no_peers, bad_peers);
@@ -533,18 +549,18 @@ impl System {
 		self.persist_peer_list.save_async(&peer_list).await
 	}
 
-	async fn pull_config(self: Arc<Self>, peer: Uuid) {
+	async fn pull_cluster_layout(self: Arc<Self>, peer: Uuid) {
 		let resp = self
 			.rpc
 			.call(
 				&self.system_endpoint,
 				peer,
-				SystemRpc::PullConfig,
+				SystemRpc::PullClusterLayout,
 				RequestStrategy::with_priority(PRIO_HIGH).with_timeout(PING_TIMEOUT),
 			)
 			.await;
-		if let Ok(SystemRpc::AdvertiseConfig(config)) = resp {
-			let _: Result<_, _> = self.handle_advertise_config(&config).await;
+		if let Ok(SystemRpc::AdvertiseClusterLayout(layout)) = resp {
+			let _: Result<_, _> = self.handle_advertise_cluster_layout(&layout).await;
 		}
 	}
 }
@@ -554,9 +570,11 @@ impl EndpointHandler<SystemRpc> for System {
 	async fn handle(self: &Arc<Self>, msg: &SystemRpc, from: NodeID) -> Result<SystemRpc, Error> {
 		match msg {
 			SystemRpc::Connect(node) => self.handle_connect(node).await,
-			SystemRpc::PullConfig => Ok(self.handle_pull_config()),
+			SystemRpc::PullClusterLayout => Ok(self.handle_pull_cluster_layout()),
 			SystemRpc::AdvertiseStatus(adv) => self.handle_advertise_status(from.into(), adv).await,
-			SystemRpc::AdvertiseConfig(adv) => self.clone().handle_advertise_config(adv).await,
+			SystemRpc::AdvertiseClusterLayout(adv) => {
+				self.clone().handle_advertise_cluster_layout(adv).await
+			}
 			SystemRpc::GetKnownNodes => Ok(self.handle_get_known_nodes()),
 			_ => Err(Error::BadRpc("Unexpected RPC message".to_string())),
 		}