Merge branch 'main' into next

1 files changed, 71 insertions, 8 deletions

diff --git a/src/block/manager.rs b/src/block/manager.rs
index 1b5a5df0..26278974 100644
--- a/src/block/manager.rs
+++ b/src/block/manager.rs
@@ -6,6 +6,7 @@ use std::time::Duration;
 use arc_swap::ArcSwapOption;
 use async_trait::async_trait;
 use bytes::Bytes;
+use rand::prelude::*;
 use serde::{Deserialize, Serialize};
 
 use futures::Stream;
@@ -23,10 +24,12 @@ use garage_rpc::rpc_helper::netapp::stream::{stream_asyncread, ByteStream};
 
 use garage_db as db;
 
-use garage_util::background::BackgroundRunner;
+use garage_util::background::{vars, BackgroundRunner};
 use garage_util::data::*;
 use garage_util::error::*;
 use garage_util::metrics::RecordDuration;
+use garage_util::persister::PersisterShared;
+use garage_util::time::msec_to_rfc3339;
 
 use garage_rpc::rpc_helper::OrderTag;
 use garage_rpc::system::System;
@@ -89,6 +92,7 @@ pub struct BlockManager {
 
 	pub(crate) metrics: BlockManagerMetrics,
 
+	pub scrub_persister: PersisterShared<ScrubWorkerPersisted>,
 	tx_scrub_command: ArcSwapOption<mpsc::Sender<ScrubWorkerCommand>>,
 }
 
@@ -125,8 +129,14 @@ impl BlockManager {
 			.netapp
 			.endpoint("garage_block/manager.rs/Rpc".to_string());
 
-		let metrics =
-			BlockManagerMetrics::new(rc.rc.clone(), resync.queue.clone(), resync.errors.clone());
+		let metrics = BlockManagerMetrics::new(
+			compression_level,
+			rc.rc.clone(),
+			resync.queue.clone(),
+			resync.errors.clone(),
+		);
+
+		let scrub_persister = PersisterShared::new(&system.metadata_dir, "scrub_info");
 
 		let block_manager = Arc::new(Self {
 			replication,
@@ -138,9 +148,11 @@ impl BlockManager {
 			system,
 			endpoint,
 			metrics,
+			scrub_persister,
 			tx_scrub_command: ArcSwapOption::new(None),
 		});
 		block_manager.endpoint.set_handler(block_manager.clone());
+		block_manager.scrub_persister.set_with(|_| ()).unwrap();
 
 		block_manager
 	}
@@ -155,7 +167,31 @@ impl BlockManager {
 		// Spawn scrub worker
 		let (scrub_tx, scrub_rx) = mpsc::channel(1);
 		self.tx_scrub_command.store(Some(Arc::new(scrub_tx)));
-		bg.spawn_worker(ScrubWorker::new(self.clone(), scrub_rx));
+		bg.spawn_worker(ScrubWorker::new(
+			self.clone(),
+			scrub_rx,
+			self.scrub_persister.clone(),
+		));
+	}
+
+	pub fn register_bg_vars(&self, vars: &mut vars::BgVars) {
+		self.resync.register_bg_vars(vars);
+
+		vars.register_rw(
+			&self.scrub_persister,
+			"scrub-tranquility",
+			|p| p.get_with(|x| x.tranquility),
+			|p, tranquility| p.set_with(|x| x.tranquility = tranquility),
+		);
+		vars.register_ro(&self.scrub_persister, "scrub-last-completed", |p| {
+			p.get_with(|x| msec_to_rfc3339(x.time_last_complete_scrub))
+		});
+		vars.register_ro(&self.scrub_persister, "scrub-next-run", |p| {
+			p.get_with(|x| msec_to_rfc3339(x.time_next_run_scrub))
+		});
+		vars.register_ro(&self.scrub_persister, "scrub-corruptions_detected", |p| {
+			p.get_with(|x| x.corruptions_detected)
+		});
 	}
 
 	/// Ask nodes that might have a (possibly compressed) block for it
@@ -649,14 +685,21 @@ impl BlockManagerLocked {
 			}
 		};
 
-		let mut path2 = path.clone();
-		path2.set_extension("tmp");
-		let mut f = fs::File::create(&path2).await?;
+		let mut path_tmp = path.clone();
+		let tmp_extension = format!("tmp{}", hex::encode(thread_rng().gen::<[u8; 4]>()));
+		path_tmp.set_extension(tmp_extension);
+
+		let mut delete_on_drop = DeleteOnDrop(Some(path_tmp.clone()));
+
+		let mut f = fs::File::create(&path_tmp).await?;
 		f.write_all(data).await?;
 		f.sync_all().await?;
 		drop(f);
 
-		fs::rename(path2, path).await?;
+		fs::rename(path_tmp, path).await?;
+
+		delete_on_drop.cancel();
+
 		if let Some(to_delete) = to_delete {
 			fs::remove_file(to_delete).await?;
 		}
@@ -722,3 +765,23 @@ async fn read_stream_to_end(mut stream: ByteStream) -> Result<Bytes, Error> {
 		.concat()
 		.into())
 }
+
+struct DeleteOnDrop(Option<PathBuf>);
+
+impl DeleteOnDrop {
+	fn cancel(&mut self) {
+		drop(self.0.take());
+	}
+}
+
+impl Drop for DeleteOnDrop {
+	fn drop(&mut self) {
+		if let Some(path) = self.0.take() {
+			tokio::spawn(async move {
+				if let Err(e) = fs::remove_file(&path).await {
+					debug!("DeleteOnDrop failed for {}: {}", path.display(), e);
+				}
+			});
+		}
+	}
+}