add error case for layout not ready, and fail earlier in many places

src/table/data.rs

@@ -254,7 +254,7 @@ impl<F: TableSchema, R: TableReplication> TableData<F, R> {
 				// any node of the partition is unavailable.
 				let pk_hash = Hash::try_from(&tree_key[..32]).unwrap();
 				// TODO: this probably breaks when the layout changes
-				let nodes = self.replication.storage_nodes(&pk_hash);
+				let nodes = self.replication.storage_nodes(&pk_hash)?;
 				if nodes.first() == Some(&self.system.id) {
 					GcTodoEntry::new(tree_key, new_bytes_hash).save(&self.gc_todo)?;
 				}

src/table/gc.rs

@@ -153,7 +153,7 @@ impl<F: TableSchema, R: TableReplication> TableGc<F, R> {
 		let mut partitions = HashMap::new();
 		for entry in entries {
 			let pkh = Hash::try_from(&entry.key[..32]).unwrap();
-			let mut nodes = self.data.replication.storage_nodes(&pkh);
+			let mut nodes = self.data.replication.storage_nodes(&pkh)?;
 			nodes.retain(|x| *x != self.system.id);
 			nodes.sort();
 

src/table/merkle.rs

@@ -102,7 +102,7 @@ impl<F: TableSchema, R: TableReplication> MerkleUpdater<F, R> {
 			partition: self
 				.data
 				.replication
-				.partition_of(&Hash::try_from(&k[0..32]).unwrap()),
+				.partition_of(&Hash::try_from(&k[0..32]).unwrap())?,
 			prefix: vec![],
 		};
 		self.data

src/table/replication/fullcopy.rs

@@ -4,6 +4,7 @@ use std::time::Duration;
 use garage_rpc::layout::*;
 use garage_rpc::{replication_mode::ConsistencyMode, system::System};
 use garage_util::data::*;
+use garage_util::error::Error;
 
 use crate::replication::*;
 
@@ -34,63 +35,64 @@ impl TableReplication for TableFullReplication {
 	// Also, it's generally a much bigger problem for fullcopy tables to be out of sync.
 	const ANTI_ENTROPY_INTERVAL: Duration = Duration::from_secs(10);
 
-	fn storage_nodes(&self, _hash: &Hash) -> Vec<Uuid> {
-		self.system.cluster_layout().all_nodes().to_vec()
+	fn storage_nodes(&self, _hash: &Hash) -> Result<Vec<Uuid>, Error> {
+		Ok(self.system.cluster_layout().all_nodes()?.to_vec())
 	}
 
-	fn read_nodes(&self, _hash: &Hash) -> Vec<Uuid> {
-		self.system
+	fn read_nodes(&self, _hash: &Hash) -> Result<Vec<Uuid>, Error> {
+		Ok(self
+			.system
 			.cluster_layout()
-			.read_version()
+			.read_version()?
 			.all_nodes()
-			.to_vec()
+			.to_vec())
 	}
-	fn read_quorum(&self) -> usize {
+	fn read_quorum(&self) -> Result<usize, Error> {
 		match self.consistency_mode {
-			ConsistencyMode::Dangerous | ConsistencyMode::Degraded => 1,
+			ConsistencyMode::Dangerous | ConsistencyMode::Degraded => Ok(1),
 			ConsistencyMode::Consistent => {
 				let layout = self.system.cluster_layout();
-				let nodes = layout.read_version().all_nodes();
-				nodes.len().div_euclid(2) + 1
+				let nodes = layout.read_version()?.all_nodes();
+				Ok(nodes.len().div_euclid(2) + 1)
 			}
 		}
 	}
 
-	fn write_sets(&self, _hash: &Hash) -> Self::WriteSets {
+	fn write_sets(&self, _hash: &Hash) -> Result<Self::WriteSets, Error> {
 		self.system.layout_manager.write_lock_with(write_sets)
 	}
-	fn write_quorum(&self) -> usize {
+	fn write_quorum(&self) -> Result<usize, Error> {
 		match self.consistency_mode {
-			ConsistencyMode::Dangerous => 1,
+			ConsistencyMode::Dangerous => Ok(1),
 			ConsistencyMode::Degraded | ConsistencyMode::Consistent => {
 				let layout = self.system.cluster_layout();
 				let min_len = layout
-					.versions()
+					.versions()?
 					.iter()
 					.map(|x| x.all_nodes().len())
 					.min()
 					.unwrap();
 				let max_quorum = layout
-					.versions()
+					.versions()?
 					.iter()
 					.map(|x| x.all_nodes().len().div_euclid(2) + 1)
 					.max()
 					.unwrap();
 				if min_len < max_quorum {
 					warn!("Write quorum will not be respected for TableFullReplication operations due to multiple active layout versions with vastly different number of nodes");
-					min_len
+					Ok(std::cmp::max(1, min_len))
 				} else {
-					max_quorum
+					Ok(max_quorum)
 				}
 			}
 		}
 	}
 
-	fn partition_of(&self, _hash: &Hash) -> Partition {
-		0u16
+	fn partition_of(&self, _hash: &Hash) -> Result<Partition, Error> {
+		Ok(0u16)
 	}
 
-	fn sync_partitions(&self) -> SyncPartitions {
+	fn sync_partitions(&self) -> Result<SyncPartitions, Error> {
 		let layout = self.system.cluster_layout();
 		let layout_version = layout.ack_map_min();
 
@@ -98,19 +100,18 @@ impl TableReplication for TableFullReplication {
 			partition: 0u16,
 			first_hash: [0u8; 32].into(),
 			last_hash: [0xff; 32].into(),
-			storage_sets: write_sets(&layout),
+			storage_sets: write_sets(layout.versions()?),
 		}];
 
-		SyncPartitions {
+		Ok(SyncPartitions {
 			layout_version,
 			partitions,
-		}
+		})
 	}
 }
 
-fn write_sets(layout: &LayoutHelper) -> Vec<Vec<Uuid>> {
-	layout
-		.versions()
+fn write_sets(layout_versions: &[LayoutVersion]) -> Vec<Vec<Uuid>> {
+	layout_versions
 		.iter()
 		.map(|x| x.all_nodes().to_vec())
 		.collect()

src/table/replication/parameters.rs

@@ -2,6 +2,7 @@ use std::time::Duration;
 
 use garage_rpc::layout::*;
 use garage_util::data::*;
+use garage_util::error::Error;
 
 /// Trait to describe how a table shall be replicated
 pub trait TableReplication: Send + Sync + 'static {
@@ -13,23 +14,23 @@ pub trait TableReplication: Send + Sync + 'static {
 	// To understand various replication methods
 
 	/// The entire list of all nodes that store a partition
-	fn storage_nodes(&self, hash: &Hash) -> Vec<Uuid>;
+	fn storage_nodes(&self, hash: &Hash) -> Result<Vec<Uuid>, Error>;
 
 	/// Which nodes to send read requests to
-	fn read_nodes(&self, hash: &Hash) -> Vec<Uuid>;
+	fn read_nodes(&self, hash: &Hash) -> Result<Vec<Uuid>, Error>;
 	/// Responses needed to consider a read successful
-	fn read_quorum(&self) -> usize;
+	fn read_quorum(&self) -> Result<usize, Error>;
 
 	/// Which nodes to send writes to
-	fn write_sets(&self, hash: &Hash) -> Self::WriteSets;
+	fn write_sets(&self, hash: &Hash) -> Result<Self::WriteSets, Error>;
 	/// Responses needed to consider a write successful in each set
-	fn write_quorum(&self) -> usize;
+	fn write_quorum(&self) -> Result<usize, Error>;
 
 	// Accessing partitions, for Merkle tree & sync
 	/// Get partition for data with given hash
-	fn partition_of(&self, hash: &Hash) -> Partition;
+	fn partition_of(&self, hash: &Hash) -> Result<Partition, Error>;
 	/// List of partitions and nodes to sync with in current layout
-	fn sync_partitions(&self) -> SyncPartitions;
+	fn sync_partitions(&self) -> Result<SyncPartitions, Error>;
 }
 
 #[derive(Debug)]

src/table/replication/sharded.rs

@@ -4,6 +4,7 @@ use std::time::Duration;
 use garage_rpc::layout::*;
 use garage_rpc::replication_mode::ConsistencyMode;
 use garage_util::data::*;
+use garage_util::error::Error;
 
 use crate::replication::sharded::manager::LayoutManager;
 use crate::replication::*;
@@ -27,59 +28,64 @@ impl TableReplication for TableShardedReplication {
 
 	type WriteSets = WriteLock<Vec<Vec<Uuid>>>;
 
-	fn storage_nodes(&self, hash: &Hash) -> Vec<Uuid> {
+	fn storage_nodes(&self, hash: &Hash) -> Result<Vec<Uuid>, Error> {
 		let mut ret = vec![];
-		for version in self.layout_manager.layout().versions().iter() {
+		for version in self.layout_manager.layout().versions()?.iter() {
 			ret.extend(version.nodes_of(hash));
 		}
 		ret.sort();
 		ret.dedup();
-		ret
+		Ok(ret)
 	}
 
-	fn read_nodes(&self, hash: &Hash) -> Vec<Uuid> {
-		self.layout_manager
+	fn read_nodes(&self, hash: &Hash) -> Result<Vec<Uuid>, Error> {
+		Ok(self
+			.layout_manager
 			.layout()
-			.read_version()
+			.read_version()?
 			.nodes_of(hash)
-			.collect()
+			.collect())
 	}
 
-	fn read_quorum(&self) -> usize {
-		self.layout_manager
+	fn read_quorum(&self) -> Result<usize, Error> {
+		Ok(self
+			.layout_manager
 			.layout()
-			.read_version()
-			.read_quorum(self.consistency_mode)
+			.read_version()?
+			.read_quorum(self.consistency_mode))
 	}
 
-	fn write_sets(&self, hash: &Hash) -> Self::WriteSets {
-		self.layout_manager.write_lock_with(|l| write_sets(l, hash))
-	}
-
-	fn write_quorum(&self) -> usize {
+	fn write_sets(&self, hash: &Hash) -> Result<Self::WriteSets, Error> {
 		self.layout_manager
+			.write_lock_with(|lvs| write_sets(lvs, hash))
+	}
+
+	fn write_quorum(&self) -> Result<usize, Error> {
+		Ok(self
+			.layout_manager
 			.layout()
-			.current()
-			.write_quorum(self.consistency_mode)
+			.current()?
+			.write_quorum(self.consistency_mode))
 	}
 
-	fn partition_of(&self, hash: &Hash) -> Partition {
-		self.layout_manager.layout().current().partition_of(hash)
+	fn partition_of(&self, hash: &Hash) -> Result<Partition, Error> {
+		Ok(self.layout_manager.layout().current()?.partition_of(hash))
 	}
 
-	fn sync_partitions(&self) -> SyncPartitions {
+	fn sync_partitions(&self) -> Result<SyncPartitions, Error> {
 		let layout = self.layout_manager.layout();
+		let layout_versions = layout.versions()?;
 		let layout_version = layout.ack_map_min();
 
 		let mut partitions = layout
-			.current()
+			.current()?
 			.partitions()
 			.map(|(partition, first_hash)| {
 				SyncPartition {
 					partition,
 					first_hash,
 					last_hash: [0u8; 32].into(), // filled in just after
-					storage_sets: write_sets(&layout, &first_hash),
+					storage_sets: write_sets(layout_versions, &first_hash),
 				}
 			})
 			.collect::<Vec<_>>();
@@ -92,16 +98,15 @@ impl TableReplication for TableShardedReplication {
 			};
 		}
 
-		SyncPartitions {
+		Ok(SyncPartitions {
 			layout_version,
 			partitions,
-		}
+		})
 	}
 }
 
-fn write_sets(layout: &LayoutHelper, hash: &Hash) -> Vec<Vec<Uuid>> {
-	layout
-		.versions()
+fn write_sets(layout_versions: &[LayoutVersion], hash: &Hash) -> Vec<Vec<Uuid>> {
+	layout_versions
 		.iter()
 		.map(|x| x.nodes_of(hash).collect())
 		.collect()

src/table/sync.rs

@@ -115,7 +115,7 @@ impl<F: TableSchema, R: TableReplication> TableSyncer<F, R> {
 			);
 			let mut result_tracker = QuorumSetResultTracker::new(
 				&partition.storage_sets,
-				self.data.replication.write_quorum(),
+				self.data.replication.write_quorum()?,
 			);
 
 			let mut sync_futures = result_tracker
@@ -179,7 +179,7 @@ impl<F: TableSchema, R: TableReplication> TableSyncer<F, R> {
 			}
 
 			if !items.is_empty() {
-				let nodes = self.data.replication.storage_nodes(begin);
+				let nodes = self.data.replication.storage_nodes(begin)?;
 				if nodes.contains(&self.system.id) {
 					warn!(
 						"({}) Interrupting offload as partitions seem to have changed",
@@ -187,7 +187,7 @@ impl<F: TableSchema, R: TableReplication> TableSyncer<F, R> {
 					);
 					break;
 				}
-				if nodes.len() < self.data.replication.write_quorum() {
+				if nodes.len() < self.data.replication.write_quorum()? {
 					return Err(Error::Message(
 						"Not offloading as we don't have a quorum of nodes to write to."
 							.to_string(),
@@ -502,15 +502,21 @@ impl<F: TableSchema, R: TableReplication> SyncWorker<F, R> {
 	}
 
 	fn add_full_sync(&mut self) {
-		let mut partitions = self.syncer.data.replication.sync_partitions();
-		debug!(
-			"{}: Adding full sync for ack layout version {}",
-			F::TABLE_NAME,
-			partitions.layout_version
-		);
+		match self.syncer.data.replication.sync_partitions() {
+			Ok(mut partitions) => {
+				debug!(
+					"{}: Adding full sync for ack layout version {}",
+					F::TABLE_NAME,
+					partitions.layout_version
+				);
 
-		partitions.partitions.shuffle(&mut thread_rng());
-		self.todo = Some(partitions);
+				partitions.partitions.shuffle(&mut thread_rng());
+				self.todo = Some(partitions);
+			}
+			Err(e) => {
+				debug!("{}: Not adding full sync: {}", F::TABLE_NAME, e);
+			}
+		}
 		self.next_full_sync = Instant::now() + R::ANTI_ENTROPY_INTERVAL;
 	}
 }

src/table/table.rs

@@ -119,7 +119,7 @@ impl<F: TableSchema, R: TableReplication> Table<F, R> {
 
 	async fn insert_internal(&self, e: &F::E) -> Result<(), Error> {
 		let hash = e.partition_key().hash();
-		let who = self.data.replication.write_sets(&hash);
+		let who = self.data.replication.write_sets(&hash)?;
 
 		let e_enc = Arc::new(ByteBuf::from(e.encode()?));
 		let rpc = TableRpc::<F>::Update(vec![e_enc]);
@@ -131,7 +131,7 @@ impl<F: TableSchema, R: TableReplication> Table<F, R> {
 				who.as_ref(),
 				rpc,
 				RequestStrategy::with_priority(PRIO_NORMAL)
-					.with_quorum(self.data.replication.write_quorum()),
+					.with_quorum(self.data.replication.write_quorum()?),
 			)
 			.await?;
 
@@ -180,7 +180,7 @@ impl<F: TableSchema, R: TableReplication> Table<F, R> {
 		// a quorum of nodes has answered OK, then the insert has succeeded and
 		// consistency properties (read-after-write) are preserved.
 
-		let quorum = self.data.replication.write_quorum();
+		let quorum = self.data.replication.write_quorum()?;
 
 		// Serialize all entries and compute the write sets for each of them.
 		// In the case of sharded table replication, this also takes an "ack lock"
@@ -193,7 +193,7 @@ impl<F: TableSchema, R: TableReplication> Table<F, R> {
 		for entry in entries.into_iter() {
 			let entry = entry.borrow();
 			let hash = entry.partition_key().hash();
-			let mut write_sets = self.data.replication.write_sets(&hash);
+			let mut write_sets = self.data.replication.write_sets(&hash)?;
 			for set in write_sets.as_mut().iter_mut() {
 				// Sort nodes in each write sets to merge write sets with same
 				// nodes but in possibly different orders
@@ -309,7 +309,7 @@ impl<F: TableSchema, R: TableReplication> Table<F, R> {
 		sort_key: &F::S,
 	) -> Result<Option<F::E>, Error> {
 		let hash = partition_key.hash();
-		let who = self.data.replication.read_nodes(&hash);
+		let who = self.data.replication.read_nodes(&hash)?;
 
 		let rpc = TableRpc::<F>::ReadEntry(partition_key.clone(), sort_key.clone());
 		let resps = self
@@ -320,7 +320,7 @@ impl<F: TableSchema, R: TableReplication> Table<F, R> {
 				&who,
 				rpc,
 				RequestStrategy::with_priority(PRIO_NORMAL)
-					.with_quorum(self.data.replication.read_quorum()),
+					.with_quorum(self.data.replication.read_quorum()?),
 			)
 			.await?;
 
@@ -397,7 +397,7 @@ impl<F: TableSchema, R: TableReplication> Table<F, R> {
 		enumeration_order: EnumerationOrder,
 	) -> Result<Vec<F::E>, Error> {
 		let hash = partition_key.hash();
-		let who = self.data.replication.read_nodes(&hash);
+		let who = self.data.replication.read_nodes(&hash)?;
 
 		let rpc = TableRpc::<F>::ReadRange {
 			partition: partition_key.clone(),
@@ -415,7 +415,7 @@ impl<F: TableSchema, R: TableReplication> Table<F, R> {
 				&who,
 				rpc,
 				RequestStrategy::with_priority(PRIO_NORMAL)
-					.with_quorum(self.data.replication.read_quorum()),
+					.with_quorum(self.data.replication.read_quorum()?),
 			)
 			.await?;