Merge branch 'main-v1' into sync-v2-to-v1

src/api/admin/api.rs

@@ -1132,6 +1132,7 @@ pub enum RepairType {
 	Rebalance,
 	Scrub(ScrubCommand),
 	Aliases,
+	ClearResyncQueue,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]

src/api/admin/repair.rs

@@ -93,6 +93,15 @@ impl RequestHandler for LocalLaunchRepairOperationRequest {
 				info!("Repairing bucket aliases (foreground)");
 				garage.locked_helper().await.repair_aliases().await?;
 			}
+			RepairType::ClearResyncQueue => {
+				info!("Clearing resync queue (foreground)");
+				let garage = garage.clone();
+				tokio::task::spawn_blocking(move || {
+					garage.block_manager.resync.clear_resync_queue()
+				})
+				.await
+				.map_err(garage_util::error::Error::from)??;
+			}
 		}
 		Ok(LocalLaunchRepairOperationResponse)
 	}

src/block/resync.rs

@@ -131,6 +131,14 @@ impl BlockResyncManager {
 		)))
 	}
 
+	/// Clear the entire resync queue and list of errored blocks
+	/// Corresponds to `garage repair clear-resync-queue`
+	pub fn clear_resync_queue(&self) -> Result<(), Error> {
+		self.queue.clear()?;
+		self.errors.clear()?;
+		Ok(())
+	}
+
 	pub fn register_bg_vars(&self, vars: &mut vars::BgVars) {
 		let notify = self.notify.clone();
 		vars.register_rw(

src/db/lib.rs

@@ -106,32 +106,44 @@ impl Db {
 			result: Cell::new(None),
 		};
 		let tx_res = self.0.transaction(&f);
-		let ret = f
-			.result
-			.into_inner()
-			.expect("Transaction did not store result");
+		let fn_res = f.result.into_inner();
 
-		match tx_res {
-			Ok(on_commit) => match ret {
-				Ok(value) => {
-					on_commit.into_iter().for_each(|f| f());
-					Ok(value)
-				}
-				_ => unreachable!(),
-			},
-			Err(TxError::Abort(())) => match ret {
-				Err(TxError::Abort(e)) => Err(TxError::Abort(e)),
-				_ => unreachable!(),
-			},
-			Err(TxError::Db(e2)) => match ret {
-				// Ok was stored -> the error occurred when finalizing
-				// transaction
-				Ok(_) => Err(TxError::Db(e2)),
-				// An error was already stored: that's the one we want to
-				// return
-				Err(TxError::Db(e)) => Err(TxError::Db(e)),
-				_ => unreachable!(),
-			},
+		match (tx_res, fn_res) {
+			(Ok(on_commit), Some(Ok(value))) => {
+				// Transaction succeeded
+				// TxFn stored the value to return to the user in fn_res
+				// tx_res contains the on_commit list of callbacks, run them now
+				on_commit.into_iter().for_each(|f| f());
+				Ok(value)
+			}
+			(Err(TxError::Abort(())), Some(Err(TxError::Abort(e)))) => {
+				// Transaction was aborted by user code
+				// The abort error value is stored in fn_res
+				Err(TxError::Abort(e))
+			}
+			(Err(TxError::Db(_tx_e)), Some(Err(TxError::Db(fn_e)))) => {
+				// Transaction encountered a DB error in user code
+				// The error value encountered is the one in fn_res,
+				// tx_res contains only a dummy error message
+				Err(TxError::Db(fn_e))
+			}
+			(Err(TxError::Db(tx_e)), None) => {
+				// Transaction encounterred a DB error when initializing the transaction,
+				// before user code was called
+				Err(TxError::Db(tx_e))
+			}
+			(Err(TxError::Db(tx_e)), Some(Ok(_))) => {
+				// Transaction encounterred a DB error when commiting the transaction,
+				// after user code was called
+				Err(TxError::Db(tx_e))
+			}
+			(tx_res, fn_res) => {
+				panic!(
+					"unexpected error case: tx_res={:?}, fn_res={:?}",
+					tx_res.map(|_| "..."),
+					fn_res.map(|x| x.map(|_| "...").map_err(|_| "..."))
+				);
+			}
 		}
 	}
 

src/db/sqlite_adapter.rs

@@ -151,30 +151,16 @@ impl IDb for SqliteDb {
 	}
 
 	fn snapshot(&self, base_path: &PathBuf) -> Result<()> {
-		fn progress(p: rusqlite::backup::Progress) {
-			use std::sync::atomic::{AtomicU64, Ordering};
-			use std::time::{SystemTime, UNIX_EPOCH};
-
-			static LAST_LOG_TIME: AtomicU64 = AtomicU64::new(0);
-
-			let now = SystemTime::now()
-				.duration_since(UNIX_EPOCH)
-				.expect("Fix your clock :o")
-				.as_millis() as u64;
-			if now >= LAST_LOG_TIME.load(Ordering::Relaxed) + 10 * 1000 {
-				let percent = (p.pagecount - p.remaining) * 100 / p.pagecount;
-				info!("Sqlite snapshot progress: {}%", percent);
-
-				LAST_LOG_TIME.fetch_max(now, Ordering::Relaxed);
-			}
-		}
-
 		std::fs::create_dir_all(base_path)?;
-		let path = Engine::Sqlite.db_path(&base_path);
+		let path = Engine::Sqlite
+			.db_path(&base_path)
+			.into_os_string()
+			.into_string()
+			.map_err(|_| Error("invalid sqlite path string".into()))?;
 
-		self.db
-			.get()?
-			.backup(rusqlite::DatabaseName::Main, path, Some(progress))?;
+		info!("Start sqlite VACUUM INTO `{}`", path);
+		self.db.get()?.execute("VACUUM INTO ?1", params![path])?;
+		info!("Finished sqlite VACUUM INTO `{}`", path);
 
 		Ok(())
 	}

src/garage/cli/remote/node.rs

@@ -92,6 +92,7 @@ impl Cli {
 				ScrubCmd::Resume => ScrubCommand::Resume,
 			}),
 			RepairWhat::Aliases => RepairType::Aliases,
+			RepairWhat::ClearResyncQueue => RepairType::ClearResyncQueue,
 		};
 
 		let res = self

src/garage/cli/structs.rs

@@ -645,6 +645,10 @@ pub enum RepairWhat {
 	/// Repair (resync/rebalance) the set of stored blocks in the cluster
 	#[structopt(name = "blocks", version = garage_version())]
 	Blocks,
+	/// Clear the block resync queue. The list of blocks in errored state
+	/// is cleared as well. You MUST run `garage repair blocks` after invoking this.
+	#[structopt(name = "clear-resync-queue", version = garage_version())]
+	ClearResyncQueue,
 	/// Repropagate object deletions to the version table
 	#[structopt(name = "versions", version = garage_version())]
 	Versions,