aboutsummaryrefslogtreecommitdiff
path: root/src/table/sync.rs
diff options
context:
space:
mode:
authorAlex Auvolat <alex@adnab.me>2023-12-07 11:16:10 +0100
committerAlex Auvolat <alex@adnab.me>2023-12-07 11:16:10 +0100
commitd90de365b3b30cb631b22fcd62c98bddb5a91549 (patch)
tree9ee7226fa49571179b9fd7ff03e45a838a51a909 /src/table/sync.rs
parent95eb13eb08d517d328e3c8aeb222440a27211ee9 (diff)
downloadgarage-d90de365b3b30cb631b22fcd62c98bddb5a91549.tar.gz
garage-d90de365b3b30cb631b22fcd62c98bddb5a91549.zip
table sync: use write quorums to report global success or failure of sync
Diffstat (limited to 'src/table/sync.rs')
-rw-r--r--src/table/sync.rs51
1 files changed, 30 insertions, 21 deletions
diff --git a/src/table/sync.rs b/src/table/sync.rs
index efeac402..cfcbc4b5 100644
--- a/src/table/sync.rs
+++ b/src/table/sync.rs
@@ -18,6 +18,7 @@ use garage_util::encode::{debug_serialize, nonversioned_encode};
use garage_util::error::{Error, OkOrMessage};
use garage_rpc::layout::*;
+use garage_rpc::rpc_helper::QuorumSetResultTracker;
use garage_rpc::system::System;
use garage_rpc::*;
@@ -106,44 +107,52 @@ impl<F: TableSchema, R: TableReplication> TableSyncer<F, R> {
must_exit: &mut watch::Receiver<bool>,
) -> Result<(), Error> {
let my_id = self.system.id;
- let retain = partition.storage_nodes.contains(&my_id);
+ let retain = partition.storage_sets.iter().any(|x| x.contains(&my_id));
if retain {
debug!(
"({}) Syncing {:?} with {:?}...",
F::TABLE_NAME,
partition,
- partition.storage_nodes
+ partition.storage_sets
);
- let mut sync_futures = partition
- .storage_nodes
+ let mut result_tracker = QuorumSetResultTracker::new(
+ &partition.storage_sets,
+ self.data.replication.write_quorum(),
+ );
+
+ let mut sync_futures = result_tracker
+ .nodes
.iter()
- .filter(|node| **node != my_id)
+ .map(|(node, _)| *node)
.map(|node| {
- self.clone()
- .do_sync_with(&partition, *node, must_exit.clone())
+ let must_exit = must_exit.clone();
+ async move {
+ if node == my_id {
+ (node, Ok(()))
+ } else {
+ (node, self.do_sync_with(&partition, node, must_exit).await)
+ }
+ }
})
.collect::<FuturesUnordered<_>>();
- let mut n_errors = 0;
- while let Some(r) = sync_futures.next().await {
- if let Err(e) = r {
- n_errors += 1;
- warn!("({}) Sync error: {}", F::TABLE_NAME, e);
+ while let Some((node, res)) = sync_futures.next().await {
+ if let Err(e) = &res {
+ warn!("({}) Sync error with {:?}: {}", F::TABLE_NAME, node, e);
}
+ result_tracker.register_result(node, res);
}
- if n_errors > 0 {
- return Err(Error::Message(format!(
- "Sync failed with {} nodes.",
- n_errors
- )));
+
+ if result_tracker.too_many_failures() {
+ return Err(result_tracker.quorum_error());
+ } else {
+ Ok(())
}
} else {
self.offload_partition(&partition.first_hash, &partition.last_hash, must_exit)
- .await?;
+ .await
}
-
- Ok(())
}
// Offload partition: this partition is not something we are storing,
@@ -264,7 +273,7 @@ impl<F: TableSchema, R: TableReplication> TableSyncer<F, R> {
}
async fn do_sync_with(
- self: Arc<Self>,
+ self: &Arc<Self>,
partition: &SyncPartition,
who: Uuid,
must_exit: watch::Receiver<bool>,