From 4c139bcbca2ef538e9e99aa84e29124cc6ae0128 Mon Sep 17 00:00:00 2001 From: Alex Auvolat Date: Sat, 13 Sep 2025 20:24:19 +0200 Subject: [PATCH] add error case for layout not ready, and fail earlier in many places --- src/api/admin/api_server.rs | 50 ++++++++-------- src/api/admin/cluster.rs | 92 ++++++++++++++++------------- src/api/k2v/index.rs | 4 +- src/block/manager.rs | 13 ++-- src/block/resync.rs | 4 +- src/model/helper/bucket.rs | 2 +- src/model/index_counter.rs | 9 ++- src/model/k2v/rpc.rs | 12 ++-- src/rpc/layout/helper.rs | 49 +++++++++------ src/rpc/layout/manager.rs | 17 +++--- src/rpc/layout/version.rs | 17 +++--- src/rpc/rpc_helper.rs | 28 +++++---- src/rpc/system.rs | 70 ++++++++++++++++------ src/rpc/system_metrics.rs | 18 ++++-- src/table/data.rs | 2 +- src/table/gc.rs | 2 +- src/table/merkle.rs | 2 +- src/table/replication/fullcopy.rs | 53 +++++++++-------- src/table/replication/parameters.rs | 15 ++--- src/table/replication/sharded.rs | 61 ++++++++++--------- src/table/sync.rs | 28 +++++---- src/table/table.rs | 16 ++--- src/util/error.rs | 3 + 23 files changed, 323 insertions(+), 244 deletions(-) diff --git a/src/api/admin/api_server.rs b/src/api/admin/api_server.rs index 5d071bbf..19a88024 100644 --- a/src/api/admin/api_server.rs +++ b/src/api/admin/api_server.rs @@ -306,38 +306,36 @@ fn verify_authorization( } pub(crate) fn find_matching_nodes(garage: &Garage, spec: &str) -> Result, Error> { - let mut res = vec![]; - if spec == "*" { - res = garage.system.cluster_layout().all_nodes().to_vec(); + if spec == "self" { + Ok(vec![garage.system.id]) + } else { + // Collect all nodes currently up and/or in cluster layout + let mut res = vec![]; + if let Ok(all_nodes) = garage.system.cluster_layout().all_nodes() { + res = all_nodes.to_vec(); + } for node in garage.system.get_known_nodes() { if node.is_up && !res.contains(&node.id) { res.push(node.id); } } - } else if spec == "self" { - res.push(garage.system.id); - } else { - let layout = garage.system.cluster_layout(); - let known_nodes = garage.system.get_known_nodes(); - let all_nodes = layout - .all_nodes() - .iter() - .copied() - .chain(known_nodes.iter().filter(|x| x.is_up).map(|x| x.id)); - for node in all_nodes { - if !res.contains(&node) && hex::encode(node).starts_with(spec) { - res.push(node); + + if spec == "*" { + // match all nodes + Ok(res) + } else { + // filter nodes that match spec + res.retain(|node| hex::encode(node).starts_with(spec)); + if res.is_empty() { + Err(Error::bad_request(format!("No nodes matching {}", spec))) + } else if res.len() > 1 { + Err(Error::bad_request(format!( + "Multiple nodes matching {}: {:?}", + spec, res + ))) + } else { + Ok(res) } } - if res.is_empty() { - return Err(Error::bad_request(format!("No nodes matching {}", spec))); - } - if res.len() > 1 { - return Err(Error::bad_request(format!( - "Multiple nodes matching {}: {:?}", - spec, res - ))); - } } - Ok(res) } diff --git a/src/api/admin/cluster.rs b/src/api/admin/cluster.rs index f42c84cf..6a97c471 100644 --- a/src/api/admin/cluster.rs +++ b/src/api/admin/cluster.rs @@ -56,48 +56,52 @@ impl RequestHandler for GetClusterStatusRequest { }) .collect::>(); - for (id, _, role) in layout.current().roles.items().iter() { - if let layout::NodeRoleV(Some(r)) = role { - let role = NodeAssignedRole { - zone: r.zone.to_string(), - capacity: r.capacity, - tags: r.tags.clone(), - }; - match nodes.get_mut(id) { - None => { - nodes.insert( - *id, - NodeResp { - id: hex::encode(id), - role: Some(role), - ..Default::default() - }, - ); - } - Some(n) => { - n.role = Some(role); + if let Ok(current_layout) = layout.current() { + for (id, _, role) in current_layout.roles.items().iter() { + if let layout::NodeRoleV(Some(r)) = role { + let role = NodeAssignedRole { + zone: r.zone.to_string(), + capacity: r.capacity, + tags: r.tags.clone(), + }; + match nodes.get_mut(id) { + None => { + nodes.insert( + *id, + NodeResp { + id: hex::encode(id), + role: Some(role), + ..Default::default() + }, + ); + } + Some(n) => { + n.role = Some(role); + } } } } } - for ver in layout.versions().iter().rev().skip(1) { - for (id, _, role) in ver.roles.items().iter() { - if let layout::NodeRoleV(Some(r)) = role { - if r.capacity.is_some() { - if let Some(n) = nodes.get_mut(id) { - if n.role.is_none() { - n.draining = true; + if let Ok(layout_versions) = layout.versions() { + for ver in layout_versions.iter().rev().skip(1) { + for (id, _, role) in ver.roles.items().iter() { + if let layout::NodeRoleV(Some(r)) = role { + if r.capacity.is_some() { + if let Some(n) = nodes.get_mut(id) { + if n.role.is_none() { + n.draining = true; + } + } else { + nodes.insert( + *id, + NodeResp { + id: hex::encode(id), + draining: true, + ..Default::default() + }, + ); } - } else { - nodes.insert( - *id, - NodeResp { - id: hex::encode(id), - draining: true, - ..Default::default() - }, - ); } } } @@ -108,7 +112,7 @@ impl RequestHandler for GetClusterStatusRequest { nodes.sort_by(|x, y| x.id.cmp(&y.id)); Ok(GetClusterStatusResponse { - layout_version: layout.current().version, + layout_version: layout.inner().current().version, nodes, }) } @@ -159,9 +163,11 @@ impl RequestHandler for GetClusterStatisticsRequest { // Gather storage node and free space statistics for current nodes let layout = &garage.system.cluster_layout(); let mut node_partition_count = HashMap::::new(); - for short_id in layout.current().ring_assignment_data.iter() { - let id = layout.current().node_id_vec[*short_id as usize]; - *node_partition_count.entry(id).or_default() += 1; + if let Ok(current_layout) = layout.current() { + for short_id in current_layout.ring_assignment_data.iter() { + let id = current_layout.node_id_vec[*short_id as usize]; + *node_partition_count.entry(id).or_default() += 1; + } } let node_info = garage .system @@ -174,7 +180,11 @@ impl RequestHandler for GetClusterStatisticsRequest { for (id, parts) in node_partition_count.iter() { let info = node_info.get(id); let status = info.map(|x| &x.status); - let role = layout.current().roles.get(id).and_then(|x| x.0.as_ref()); + let role = layout + .current() + .ok() + .and_then(|l| l.roles.get(id)) + .and_then(|x| x.0.as_ref()); let hostname = status.and_then(|x| x.hostname.as_deref()).unwrap_or("?"); let zone = role.map(|x| x.zone.as_str()).unwrap_or("?"); let capacity = role diff --git a/src/api/k2v/index.rs b/src/api/k2v/index.rs index fbfaad98..f4beba24 100644 --- a/src/api/k2v/index.rs +++ b/src/api/k2v/index.rs @@ -28,7 +28,7 @@ pub async fn handle_read_index( let node_id_vec = garage .system .cluster_layout() - .all_nongateway_nodes() + .all_nongateway_nodes()? .to_vec(); let (partition_keys, more, next_start) = read_range( @@ -66,7 +66,7 @@ pub async fn handle_read_index( bytes: *vals.get(&s_bytes).unwrap_or(&0), } }) - .collect::>(), + .collect(), more, next_start, }; diff --git a/src/block/manager.rs b/src/block/manager.rs index bd0b7611..1f5b72af 100644 --- a/src/block/manager.rs +++ b/src/block/manager.rs @@ -285,7 +285,7 @@ impl BlockManager { let who = self .system .rpc_helper() - .block_read_nodes_of(hash, self.system.rpc_helper()); + .block_read_nodes_of(hash, self.system.rpc_helper())?; for node in who.iter() { let node_id = NodeID::from(*node); @@ -341,12 +341,9 @@ impl BlockManager { /// 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 { - self.system - .cluster_layout() - .current() - .nodes_of(hash) - .collect() + pub(crate) fn storage_nodes_of(&self, hash: &Hash) -> Result, Error> { + let cluster_layout = self.system.cluster_layout(); + Ok(cluster_layout.current()?.nodes_of(hash).collect()) } // ---- Public interface ---- @@ -381,7 +378,7 @@ impl BlockManager { prevent_compression: bool, order_tag: Option, ) -> Result<(), Error> { - let who = self.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) diff --git a/src/block/resync.rs b/src/block/resync.rs index 307f7c48..ad2b816a 100644 --- a/src/block/resync.rs +++ b/src/block/resync.rs @@ -375,7 +375,7 @@ impl BlockResyncManager { info!("Resync block {:?}: offloading and deleting", hash); let existing_path = existing_path.unwrap(); - let mut who = manager.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())); } @@ -458,7 +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.storage_nodes_of(&hash); + let storage_nodes = manager.storage_nodes_of(&hash)?; if !storage_nodes.contains(&manager.system.id) { info!( diff --git a/src/model/helper/bucket.rs b/src/model/helper/bucket.rs index ebbe95ca..c82dd683 100644 --- a/src/model/helper/bucket.rs +++ b/src/model/helper/bucket.rs @@ -227,7 +227,7 @@ impl<'a> BucketHelper<'a> { .0 .system .cluster_layout() - .all_nongateway_nodes() + .all_nongateway_nodes()? .to_vec(); let k2vindexes = self .0 diff --git a/src/model/index_counter.rs b/src/model/index_counter.rs index aa13ee7b..50abdec3 100644 --- a/src/model/index_counter.rs +++ b/src/model/index_counter.rs @@ -84,17 +84,16 @@ impl Entry for CounterEntry { impl CounterEntry { pub fn filtered_values(&self, layout: &LayoutHelper) -> HashMap { - let nodes = layout.all_nongateway_nodes(); - self.filtered_values_with_nodes(&nodes) + self.filtered_values_internal(layout.all_nongateway_nodes().ok()) } - pub fn filtered_values_with_nodes(&self, nodes: &[Uuid]) -> HashMap { + fn filtered_values_internal(&self, nodes_opt: Option<&[Uuid]>) -> HashMap { let mut ret = HashMap::new(); for (name, vals) in self.values.iter() { let new_vals = vals .node_values .iter() - .filter(|(n, _)| nodes.contains(n)) + .filter(|(n, _)| nodes_opt.map(|nodes| nodes.contains(n)).unwrap_or(true)) .map(|(_, (_, v))| *v) .collect::>(); if !new_vals.is_empty() { @@ -153,7 +152,7 @@ impl TableSchema for CounterTable { } let is_tombstone = entry - .filtered_values_with_nodes(&filter.1[..]) + .filtered_values_internal(Some(&filter.1[..])) .iter() .all(|(_, v)| *v == 0); filter.0.apply(is_tombstone) diff --git a/src/model/k2v/rpc.rs b/src/model/k2v/rpc.rs index ddc356b5..8fcf8309 100644 --- a/src/model/k2v/rpc.rs +++ b/src/model/k2v/rpc.rs @@ -126,7 +126,7 @@ impl K2VRpcHandler { .item_table .data .replication - .storage_nodes(&partition.hash()); + .storage_nodes(&partition.hash())?; who.sort(); self.system @@ -165,7 +165,7 @@ impl K2VRpcHandler { .item_table .data .replication - .storage_nodes(&partition.hash()); + .storage_nodes(&partition.hash())?; who.sort(); call_list.entry(who).or_default().push(InsertedItem { @@ -222,7 +222,7 @@ impl K2VRpcHandler { .item_table .data .replication - .storage_nodes(&poll_key.partition.hash()); + .storage_nodes(&poll_key.partition.hash())?; let rpc = self.system.rpc_helper().try_call_many( &self.endpoint, @@ -233,7 +233,7 @@ impl K2VRpcHandler { timeout_msec, }, RequestStrategy::with_priority(PRIO_NORMAL) - .with_quorum(self.item_table.data.replication.read_quorum()) + .with_quorum(self.item_table.data.replication.read_quorum()?) .send_all_at_once(true) .without_timeout(), ); @@ -283,8 +283,8 @@ impl K2VRpcHandler { .item_table .data .replication - .storage_nodes(&range.partition.hash()); - let quorum = self.item_table.data.replication.read_quorum(); + .storage_nodes(&range.partition.hash())?; + let quorum = self.item_table.data.replication.read_quorum()?; let msg = K2VRpc::PollRange { range, seen_str, diff --git a/src/rpc/layout/helper.rs b/src/rpc/layout/helper.rs index 088ffb2f..ec4137bc 100644 --- a/src/rpc/layout/helper.rs +++ b/src/rpc/layout/helper.rs @@ -4,6 +4,7 @@ use std::sync::atomic::{AtomicUsize, Ordering}; use serde::{Deserialize, Serialize}; use garage_util::data::*; +use garage_util::error::Error; use super::*; use crate::replication_mode::*; @@ -145,24 +146,31 @@ impl LayoutHelper { } /// Returns the current layout version - pub fn current(&self) -> &LayoutVersion { - self.inner().current() + pub fn current(&self) -> Result<&LayoutVersion, Error> { + if !self.is_check_ok { + return Err(Error::LayoutNotReady); + } + Ok(self.inner().current()) } /// Returns all layout versions currently active in the cluster - pub fn versions(&self) -> &[LayoutVersion] { - &self.inner().versions + pub fn versions(&self) -> Result<&[LayoutVersion], Error> { + if !self.is_check_ok { + return Err(Error::LayoutNotReady); + } + Ok(&self.inner().versions) } /// Returns the latest layout version for which it is safe to read data from, /// i.e. the version whose version number is sync_map_min - pub fn read_version(&self) -> &LayoutVersion { + pub fn read_version(&self) -> Result<&LayoutVersion, Error> { let sync_min = self.sync_map_min; - self.versions() + let versions = self.versions()?; + Ok(versions .iter() .find(|x| x.version == sync_min) - .or(self.versions().last()) - .unwrap() + .or(versions.last()) + .unwrap()) } pub fn is_check_ok(&self) -> bool { @@ -171,14 +179,20 @@ impl LayoutHelper { /// Return all nodes that have a role (gateway or storage) /// in one of the currently active layout versions - pub fn all_nodes(&self) -> &[Uuid] { - &self.all_nodes + pub fn all_nodes(&self) -> Result<&[Uuid], Error> { + if !self.is_check_ok { + return Err(Error::LayoutNotReady); + } + Ok(&self.all_nodes) } /// Return all nodes that are configured to store data /// in one of the currently active layout versions - pub fn all_nongateway_nodes(&self) -> &[Uuid] { - &self.all_nongateway_nodes + pub fn all_nongateway_nodes(&self) -> Result<&[Uuid], Error> { + if !self.is_check_ok { + return Err(Error::LayoutNotReady); + } + Ok(&self.all_nongateway_nodes) } pub fn ack_map_min(&self) -> u64 { @@ -193,7 +207,7 @@ impl LayoutHelper { pub fn sync_digest(&self) -> SyncLayoutDigest { SyncLayoutDigest { - current: self.current().version, + current: self.inner().current().version, ack_map_min: self.ack_map_min(), min_stored: self.inner().min_stored(), } @@ -201,8 +215,8 @@ impl LayoutHelper { pub(crate) fn digest(&self) -> RpcLayoutDigest { RpcLayoutDigest { - current_version: self.current().version, - active_versions: self.versions().len(), + current_version: self.inner().current().version, + active_versions: self.inner().versions.len(), trackers_hash: self.trackers_hash, staging_hash: self.staging_hash, } @@ -246,7 +260,8 @@ impl LayoutHelper { pub(crate) fn update_ack_to_max_free(&mut self, local_node_id: Uuid) -> bool { let max_free = self - .versions() + .inner() + .versions .iter() .map(|x| x.version) .skip_while(|v| { @@ -256,7 +271,7 @@ impl LayoutHelper { .unwrap_or(true) }) .next() - .unwrap_or(self.current().version); + .unwrap_or(self.inner().current().version); let changed = self.update(|layout| { layout .update_trackers diff --git a/src/rpc/layout/manager.rs b/src/rpc/layout/manager.rs index c5bba0b6..aa6dda6f 100644 --- a/src/rpc/layout/manager.rs +++ b/src/rpc/layout/manager.rs @@ -105,7 +105,7 @@ impl LayoutManager { } pub fn add_table(&self, table_name: &'static str) { - let first_version = self.layout().versions().first().unwrap().version; + let first_version = self.layout().inner().versions.first().unwrap().version; self.table_sync_version .lock() @@ -139,19 +139,20 @@ impl LayoutManager { // ---- ACK LOCKING ---- - pub fn write_lock_with(self: &Arc, f: F) -> WriteLock + pub fn write_lock_with(self: &Arc, f: F) -> Result, Error> where - F: FnOnce(&LayoutHelper) -> T, + F: FnOnce(&[LayoutVersion]) -> T, { let layout = self.layout(); - let version = layout.current().version; - let value = f(&layout); + let current_version = layout.current()?.version; + let versions = layout.versions()?; + let value = f(versions); layout .ack_lock - .get(&version) + .get(¤t_version) .unwrap() .fetch_add(1, Ordering::Relaxed); - WriteLock::new(version, self, value) + Ok(WriteLock::new(current_version, self, value)) } // ---- INTERNALS --- @@ -369,7 +370,7 @@ impl Drop for WriteLock { let layout = self.layout_manager.layout(); // acquire read lock if let Some(counter) = layout.ack_lock.get(&self.layout_version) { let prev_lock = counter.fetch_sub(1, Ordering::Relaxed); - if prev_lock == 1 && layout.current().version > self.layout_version { + if prev_lock == 1 && layout.current().unwrap().version > self.layout_version { drop(layout); // release read lock, write lock will be acquired self.layout_manager.ack_new_version(); } diff --git a/src/rpc/layout/version.rs b/src/rpc/layout/version.rs index 4dd2963c..840c1144 100644 --- a/src/rpc/layout/version.rs +++ b/src/rpc/layout/version.rs @@ -118,15 +118,14 @@ impl LayoutVersion { pub fn nodes_of(&self, position: &Hash) -> impl Iterator + '_ { let data = &self.ring_assignment_data; - let partition_nodes = if data.len() == self.replication_factor * (1 << PARTITION_BITS) { - let partition_idx = self.partition_of(position) as usize; - let partition_start = partition_idx * self.replication_factor; - let partition_end = (partition_idx + 1) * self.replication_factor; - &data[partition_start..partition_end] - } else { - warn!("Ring not yet ready, read/writes will be lost!"); - &[] - }; + if data.len() != self.replication_factor * (1 << PARTITION_BITS) { + panic!(".nodes_of() called on invalid LayoutVersion (this is a bug)"); + } + + let partition_idx = self.partition_of(position) as usize; + let partition_start = partition_idx * self.replication_factor; + let partition_end = (partition_idx + 1) * self.replication_factor; + let partition_nodes = &data[partition_start..partition_end]; partition_nodes .iter() diff --git a/src/rpc/rpc_helper.rs b/src/rpc/rpc_helper.rs index 87fff5d6..3d56957e 100644 --- a/src/rpc/rpc_helper.rs +++ b/src/rpc/rpc_helper.rs @@ -345,7 +345,7 @@ impl RpcHelper { // Reorder requests to priorize closeness / low latency let request_order = - self.request_order(&self.0.layout.read().unwrap().current(), to.iter().copied()); + self.request_order(self.0.layout.read().unwrap().current()?, to.iter().copied()); let send_all_at_once = strategy.rs_send_all_at_once.unwrap_or(false); // Build future for each request @@ -567,25 +567,29 @@ impl RpcHelper { /// The preference order, for each layout version, is given by `request_order`, /// based on factors such as nodes being in the same datacenter, /// having low ping, etc. - pub fn block_read_nodes_of(&self, position: &Hash, rpc_helper: &RpcHelper) -> Vec { + pub fn block_read_nodes_of( + &self, + position: &Hash, + rpc_helper: &RpcHelper, + ) -> Result, Error> { let layout = self.0.layout.read().unwrap(); + let current_layout = layout.current()?; // Compute, for each layout version, the set of nodes that might store // the block, and put them in their preferred order as of `request_order`. - let mut vernodes = layout.versions().iter().map(|ver| { + let mut vernodes = vec![]; + for ver in layout.versions()?.iter() { let nodes = ver.nodes_of(position); - rpc_helper.request_order(layout.current(), nodes) - }); + vernodes.push(rpc_helper.request_order(current_layout, nodes)) + } - let mut ret = if layout.versions().len() == 1 { + let mut ret = if vernodes.len() == 1 { // If we have only one active layout version, then these are the // only nodes we ask in step 1 - vernodes.next().unwrap() + vernodes.into_iter().next().unwrap() } else { - let vernodes = vernodes.collect::>(); - let mut nodes = Vec::::with_capacity(12); - for i in 0..layout.current().replication_factor { + for i in 0..current_layout.replication_factor { for vn in vernodes.iter() { if let Some(n) = vn.get(i) { if !nodes.contains(&n) { @@ -608,14 +612,14 @@ impl RpcHelper { let old_ver_iter = layout.inner().old_versions.iter().rev(); for ver in old_ver_iter { let nodes = ver.nodes_of(position); - for node in rpc_helper.request_order(layout.current(), nodes) { + for node in rpc_helper.request_order(current_layout, nodes) { if !ret.contains(&node) { ret.push(node); } } } - ret + Ok(ret) } fn request_order( diff --git a/src/rpc/system.rs b/src/rpc/system.rs index 97901130..ee6a3547 100644 --- a/src/rpc/system.rs +++ b/src/rpc/system.rs @@ -453,11 +453,28 @@ impl System { // Acquire a rwlock read-lock to the current cluster layout let layout = self.cluster_layout(); + let layout_versions = match layout.versions() { + Ok(v) => v, + Err(_) => { + // Layout not yet configured, special case + return ClusterHealth { + status: ClusterHealthStatus::Unavailable, + known_nodes: nodes.len(), + connected_nodes, + storage_nodes: 0, + storage_nodes_ok: 0, + partitions: 0, + partitions_quorum: 0, + partitions_all_ok: 0, + }; + } + }; + let current_layout = layout_versions.last().unwrap(); // Obtain information about nodes that have a role as storage nodes // in one of the active layout versions let mut storage_nodes = HashSet::::with_capacity(16); - for ver in layout.versions().iter() { + for ver in layout_versions.iter() { storage_nodes.extend( ver.roles .items() @@ -471,11 +488,11 @@ impl System { // Determine the number of partitions that have: // - a quorum of up nodes for all write sets (i.e. are available) // - for which all nodes in all write sets are up (i.e. are fully healthy) - let partitions = layout.current().partitions().collect::>(); + let partitions = current_layout.partitions().collect::>(); let mut partitions_quorum = 0; let mut partitions_all_ok = 0; for (_, hash) in partitions.iter() { - let mut write_sets = layout.versions().iter().map(|x| x.nodes_of(hash)); + let mut write_sets = layout_versions.iter().map(|x| x.nodes_of(hash)); let has_quorum = write_sets .clone() .all(|set| set.filter(|x| node_up(x)).count() >= quorum); @@ -630,21 +647,37 @@ impl System { async fn discovery_loop(self: &Arc, mut stop_signal: watch::Receiver) { while !*stop_signal.borrow() { - let n_connected = self + let peers_up = self .peering .get_peer_list() .iter() .filter(|p| p.is_up()) - .count(); + .map(|p| Uuid::from(p.id)) + .collect::>(); - let not_configured = !self.cluster_layout().is_check_ok(); - let no_peers = n_connected < self.replication_factor.into(); - let expected_n_nodes = self.cluster_layout().all_nodes().len(); - let bad_peers = n_connected != 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); + let do_bootstrap = match self.cluster_layout().all_nodes() { + Err(_) => { + debug!("doing bootstrap/discovery step (layout not configured)"); + true + } + Ok(all_nodes) => { + // Do bootstrap if we have fewer peers than the replication + // factor, + // or if some peers in the layout are not connected + let do_bootstrap = peers_up.len() < self.replication_factor.into() + || all_nodes.iter().any(|x| !peers_up.contains(x)); + if do_bootstrap { + debug!( + "doing bootstrap/discovery step (peers_up: {}, all_nodes: {})", + peers_up.len(), + all_nodes.len() + ); + } + do_bootstrap + } + }; + if do_bootstrap { let mut ping_list = resolve_peers(&self.bootstrap_peers).await; // Add peer list from list stored on disk @@ -687,12 +720,13 @@ impl System { } } - if !not_configured && !no_peers { - // If the layout is configured, and we already have some connections - // to other nodes in the cluster, we can skip trying to connect to - // nodes that are not in the cluster layout. - let layout = self.cluster_layout(); - ping_list.retain(|(id, _)| layout.all_nodes().contains(&(*id).into())); + if let Ok(all_nodes) = self.cluster_layout().all_nodes() { + if peers_up.len() >= self.replication_factor.into() { + // If the layout is configured, and we already have some connections + // to other nodes in the cluster, we can skip trying to connect to + // nodes that are not in the cluster layout. + ping_list.retain(|(id, _)| all_nodes.contains(&(*id).into())); + } } for (node_id, node_addr) in ping_list { diff --git a/src/rpc/system_metrics.rs b/src/rpc/system_metrics.rs index 31c2e9ff..937c6db2 100644 --- a/src/rpc/system_metrics.rs +++ b/src/rpc/system_metrics.rs @@ -216,10 +216,13 @@ impl SystemMetrics { .u64_value_observer("cluster_layout_node_connected", move |observer| { let layout = system.cluster_layout(); let nodes = system.get_known_nodes(); - for id in layout.all_nodes().iter() { + for id in layout.all_nodes().unwrap_or_default().iter() { let mut kv = vec![KeyValue::new("id", format!("{:?}", id))]; - if let Some(role) = - layout.current().roles.get(id).and_then(|r| r.0.as_ref()) + if let Some(role) = layout + .current() + .ok() + .and_then(|l| l.roles.get(id)) + .and_then(|r| r.0.as_ref()) { kv.push(KeyValue::new("role_zone", role.zone.clone())); match role.capacity { @@ -260,10 +263,13 @@ impl SystemMetrics { .u64_value_observer("cluster_layout_node_disconnected_time", move |observer| { let layout = system.cluster_layout(); let nodes = system.get_known_nodes(); - for id in layout.all_nodes().iter() { + for id in layout.all_nodes().unwrap_or_default().iter() { let mut kv = vec![KeyValue::new("id", format!("{:?}", id))]; - if let Some(role) = - layout.current().roles.get(id).and_then(|r| r.0.as_ref()) + if let Some(role) = layout + .current() + .ok() + .and_then(|l| l.roles.get(id)) + .and_then(|r| r.0.as_ref()) { kv.push(KeyValue::new("role_zone", role.zone.clone())); match role.capacity { diff --git a/src/table/data.rs b/src/table/data.rs index c589c777..ffd55e8d 100644 --- a/src/table/data.rs +++ b/src/table/data.rs @@ -254,7 +254,7 @@ impl TableData { // 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)?; } diff --git a/src/table/gc.rs b/src/table/gc.rs index 28ea119d..e010efe1 100644 --- a/src/table/gc.rs +++ b/src/table/gc.rs @@ -153,7 +153,7 @@ impl TableGc { 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(); diff --git a/src/table/merkle.rs b/src/table/merkle.rs index 596d5805..5fe35bab 100644 --- a/src/table/merkle.rs +++ b/src/table/merkle.rs @@ -102,7 +102,7 @@ impl MerkleUpdater { 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 diff --git a/src/table/replication/fullcopy.rs b/src/table/replication/fullcopy.rs index 236b8c72..b6b95ae6 100644 --- a/src/table/replication/fullcopy.rs +++ b/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 { - self.system.cluster_layout().all_nodes().to_vec() + fn storage_nodes(&self, _hash: &Hash) -> Result, Error> { + Ok(self.system.cluster_layout().all_nodes()?.to_vec()) } - fn read_nodes(&self, _hash: &Hash) -> Vec { - self.system + fn read_nodes(&self, _hash: &Hash) -> Result, 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 { 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.system.layout_manager.write_lock_with(write_sets) } - fn write_quorum(&self) -> usize { + fn write_quorum(&self) -> Result { 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 { + Ok(0u16) } - fn sync_partitions(&self) -> SyncPartitions { + fn sync_partitions(&self) -> Result { 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> { - layout - .versions() +fn write_sets(layout_versions: &[LayoutVersion]) -> Vec> { + layout_versions .iter() .map(|x| x.all_nodes().to_vec()) .collect() diff --git a/src/table/replication/parameters.rs b/src/table/replication/parameters.rs index 327f2cbf..0bdfebd0 100644 --- a/src/table/replication/parameters.rs +++ b/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; + fn storage_nodes(&self, hash: &Hash) -> Result, Error>; /// Which nodes to send read requests to - fn read_nodes(&self, hash: &Hash) -> Vec; + fn read_nodes(&self, hash: &Hash) -> Result, Error>; /// Responses needed to consider a read successful - fn read_quorum(&self) -> usize; + fn read_quorum(&self) -> Result; /// Which nodes to send writes to - fn write_sets(&self, hash: &Hash) -> Self::WriteSets; + fn write_sets(&self, hash: &Hash) -> Result; /// Responses needed to consider a write successful in each set - fn write_quorum(&self) -> usize; + fn write_quorum(&self) -> Result; // 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; /// List of partitions and nodes to sync with in current layout - fn sync_partitions(&self) -> SyncPartitions; + fn sync_partitions(&self) -> Result; } #[derive(Debug)] diff --git a/src/table/replication/sharded.rs b/src/table/replication/sharded.rs index cd2e8044..b8983a4a 100644 --- a/src/table/replication/sharded.rs +++ b/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>>; - fn storage_nodes(&self, hash: &Hash) -> Vec { + fn storage_nodes(&self, hash: &Hash) -> Result, 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 { - self.layout_manager + fn read_nodes(&self, hash: &Hash) -> Result, 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 { + 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.layout_manager + .write_lock_with(|lvs| write_sets(lvs, hash)) + } + + fn write_quorum(&self) -> Result { + 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 { + Ok(self.layout_manager.layout().current()?.partition_of(hash)) } - fn sync_partitions(&self) -> SyncPartitions { + fn sync_partitions(&self) -> Result { 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::>(); @@ -92,16 +98,15 @@ impl TableReplication for TableShardedReplication { }; } - SyncPartitions { + Ok(SyncPartitions { layout_version, partitions, - } + }) } } -fn write_sets(layout: &LayoutHelper, hash: &Hash) -> Vec> { - layout - .versions() +fn write_sets(layout_versions: &[LayoutVersion], hash: &Hash) -> Vec> { + layout_versions .iter() .map(|x| x.nodes_of(hash).collect()) .collect() diff --git a/src/table/sync.rs b/src/table/sync.rs index e72286ed..a4c558bb 100644 --- a/src/table/sync.rs +++ b/src/table/sync.rs @@ -115,7 +115,7 @@ impl TableSyncer { ); 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 TableSyncer { } 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 TableSyncer { ); 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 SyncWorker { } 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; } } diff --git a/src/table/table.rs b/src/table/table.rs index 565d27a5..8ddd8378 100644 --- a/src/table/table.rs +++ b/src/table/table.rs @@ -119,7 +119,7 @@ impl Table { 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::::Update(vec![e_enc]); @@ -131,7 +131,7 @@ impl Table { 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 Table { // 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 Table { 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 Table { sort_key: &F::S, ) -> Result, 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::::ReadEntry(partition_key.clone(), sort_key.clone()); let resps = self @@ -320,7 +320,7 @@ impl Table { &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 Table { enumeration_order: EnumerationOrder, ) -> Result, 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::::ReadRange { partition: partition_key.clone(), @@ -415,7 +415,7 @@ impl Table { &who, rpc, RequestStrategy::with_priority(PRIO_NORMAL) - .with_quorum(self.data.replication.read_quorum()), + .with_quorum(self.data.replication.read_quorum()?), ) .await?; diff --git a/src/util/error.rs b/src/util/error.rs index 75fd3f9c..d33bb19d 100644 --- a/src/util/error.rs +++ b/src/util/error.rs @@ -54,6 +54,9 @@ pub enum Error { #[error(display = "Timeout")] Timeout, + #[error(display = "Layout not ready")] + LayoutNotReady, + #[error( display = "Could not reach quorum of {} (sets={:?}). {} of {} request succeeded, others returned errors: {:?}", _0,