relocalize logic into block manager

src/block/manager.rs

@@ -336,6 +336,19 @@ impl BlockManager {
 		Err(err)
 	}
 
+	/// Returns the set of nodes that should store a copy of a given block.
+	/// These are the nodes assigned to the block's hash in the current
+	/// layout version only: since blocks are immutable, we don't need to
+	/// do complex logic when several layour versions are active at once,
+	/// just move them directly to the new nodes.
+	pub(crate) fn storage_nodes_of(&self, hash: &Hash) -> Vec<Uuid> {
+		self.system
+			.cluster_layout()
+			.current()
+			.nodes_of(hash)
+			.collect()
+	}
+
 	// ---- Public interface ----
 
 	/// Ask nodes that might have a block for it, return it as a stream
@@ -368,7 +381,7 @@ impl BlockManager {
 		prevent_compression: bool,
 		order_tag: Option<OrderTag>,
 	) -> Result<(), Error> {
-		let who = self.system.cluster_layout().current_storage_nodes_of(&hash);
+		let who = self.storage_nodes_of(&hash);
 
 		let compression_level = self.compression_level.filter(|_| !prevent_compression);
 		let (header, bytes) = DataBlock::from_buffer(data, compression_level)

src/block/resync.rs

@@ -375,10 +375,7 @@ impl BlockResyncManager {
 			info!("Resync block {:?}: offloading and deleting", hash);
 			let existing_path = existing_path.unwrap();
 
-			let mut who = manager
-				.system
-				.cluster_layout()
-				.current_storage_nodes_of(hash);
+			let mut who = manager.storage_nodes_of(hash);
 			if who.len() < manager.write_quorum {
 				return Err(Error::Message("Not trying to offload block because we don't have a quorum of nodes to write to".to_string()));
 			}
@@ -461,10 +458,7 @@ impl BlockResyncManager {
 
 			// First, check whether we are still supposed to store that
 			// block in the latest cluster layout version.
-			let storage_nodes = manager
-				.system
-				.cluster_layout()
-				.current_storage_nodes_of(&hash);
+			let storage_nodes = manager.storage_nodes_of(&hash);
 
 			if !storage_nodes.contains(&manager.system.id) {
 				info!(