More security: don't delete stuff too easily

This commit is contained in:
Alex Auvolat 2021-03-11 19:06:27 +01:00
parent f7c2cd1cd7
commit 3f7a496355
3 changed files with 12 additions and 1 deletions

View File

@ -305,6 +305,9 @@ impl BlockManager {
let ring = self.system.ring.borrow().clone(); let ring = self.system.ring.borrow().clone();
let mut who = self.replication.replication_nodes(&hash, &ring); let mut who = self.replication.replication_nodes(&hash, &ring);
if who.len() < self.replication.write_quorum(&self.system) {
return Err(Error::Message(format!("Not trying to offload block because we don't have a quorum of nodes to write to")));
}
who.retain(|id| *id != self.system.id); who.retain(|id| *id != self.system.id);
let msg = Arc::new(Message::NeedBlockQuery(*hash)); let msg = Arc::new(Message::NeedBlockQuery(*hash));

View File

@ -41,7 +41,12 @@ impl TableReplication for TableFullReplication {
self.replication_nodes(hash, system.ring.borrow().as_ref()) self.replication_nodes(hash, system.ring.borrow().as_ref())
} }
fn write_quorum(&self, system: &System) -> usize { fn write_quorum(&self, system: &System) -> usize {
system.ring.borrow().config.members.len() - self.max_faults let nmembers = system.ring.borrow().config.members.len();
if nmembers > self.max_faults {
nmembers - self.max_faults
} else {
1
}
} }
fn max_write_errors(&self) -> usize { fn max_write_errors(&self) -> usize {
self.max_faults self.max_faults

View File

@ -277,6 +277,9 @@ where
warn!("Interrupting offload as partitions seem to have changed"); warn!("Interrupting offload as partitions seem to have changed");
break; break;
} }
if nodes.len() < self.aux.replication.write_quorum(&self.aux.system) {
return Err(Error::Message(format!("Not offloading as we don't have a quorum of nodes to write to.")));
}
counter += 1; counter += 1;
debug!( debug!(