aboutsummaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/rpc/membership.rs24
-rw-r--r--src/util/persister.rs12
2 files changed, 31 insertions, 5 deletions
diff --git a/src/rpc/membership.rs b/src/rpc/membership.rs
index 330c154f..f8223420 100644
--- a/src/rpc/membership.rs
+++ b/src/rpc/membership.rs
@@ -582,7 +582,7 @@ impl System {
.map(|ip| (*ip, None))
.collect::<Vec<_>>();
- match self.persist_status.load() {
+ match self.persist_status.load_async().await {
Ok(peers) => {
bp2.extend(peers.iter().map(|x| (x.addr, Some(x.id))));
}
@@ -653,9 +653,25 @@ impl System {
async fn update_status(self: &Arc<Self>, updaters: &Updaters, status: Status) {
if status.hash != self.status.borrow().hash {
info!("Persisting new peer list");
- let serializable_status = status.to_serializable_membership(&self);
- self.persist_status.save_async(&serializable_status).await
- .expect("Unable to persist peer list");
+
+ let mut list = status.to_serializable_membership(&self);
+
+ // Combine with old peer list to make sure no peer is lost
+ match self.persist_status.load_async().await {
+ Ok(old_list) => {
+ for pp in old_list {
+ if !list.iter().any(|np| pp.id == np.id) {
+ list.push(pp);
+ }
+ }
+ }
+ _ => (),
+ }
+
+ if list.len() > 0 {
+ self.persist_status.save_async(&list).await
+ .expect("Unable to persist peer list");
+ }
}
let status = Arc::new(status);
diff --git a/src/util/persister.rs b/src/util/persister.rs
index f4e8cd72..93b7cdf4 100644
--- a/src/util/persister.rs
+++ b/src/util/persister.rs
@@ -1,7 +1,7 @@
use std::io::{Read, Write};
use std::path::PathBuf;
-use tokio::io::AsyncWriteExt;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
use serde::{Deserialize, Serialize};
@@ -51,6 +51,16 @@ where
Ok(())
}
+ pub async fn load_async(&self) -> Result<T, Error> {
+ let mut file = tokio::fs::File::open(&self.path).await?;
+
+ let mut bytes = vec![];
+ file.read_to_end(&mut bytes).await?;
+
+ let value = rmp_serde::decode::from_read_ref(&bytes[..])?;
+ Ok(value)
+ }
+
pub async fn save_async(&self, t: &T) -> Result<(), Error> {
let bytes = rmp_to_vec_all_named(t)?;