cli_v2: implement RetryBlockResync and PurgeBlocks

3 files changed, 44 insertions, 165 deletions

diff --git a/src/garage/admin/block.rs b/src/garage/admin/block.rs
deleted file mode 100644
index 1138703a..00000000
--- a/src/garage/admin/block.rs
+++ /dev/null
@@ -1,153 +0,0 @@
-use garage_util::data::*;
-
-use garage_table::*;
-
-use garage_model::helper::error::{Error, OkOrBadRequest};
-use garage_model::s3::object_table::*;
-use garage_model::s3::version_table::*;
-
-use crate::cli::*;
-
-use super::*;
-
-impl AdminRpcHandler {
-	pub(super) async fn handle_block_cmd(&self, cmd: &BlockOperation) -> Result<AdminRpc, Error> {
-		match cmd {
-			BlockOperation::RetryNow { all, blocks } => {
-				self.handle_block_retry_now(*all, blocks).await
-			}
-			BlockOperation::Purge { yes, blocks } => self.handle_block_purge(*yes, blocks).await,
-			_ => unreachable!(),
-		}
-	}
-
-	async fn handle_block_retry_now(
-		&self,
-		all: bool,
-		blocks: &[String],
-	) -> Result<AdminRpc, Error> {
-		if all {
-			if !blocks.is_empty() {
-				return Err(Error::BadRequest(
-					"--all was specified, cannot also specify blocks".into(),
-				));
-			}
-			let blocks = self.garage.block_manager.list_resync_errors()?;
-			for b in blocks.iter() {
-				self.garage.block_manager.resync.clear_backoff(&b.hash)?;
-			}
-			Ok(AdminRpc::Ok(format!(
-				"{} blocks returned in queue for a retry now (check logs to see results)",
-				blocks.len()
-			)))
-		} else {
-			for hash in blocks {
-				let hash = hex::decode(hash).ok_or_bad_request("invalid hash")?;
-				let hash = Hash::try_from(&hash).ok_or_bad_request("invalid hash")?;
-				self.garage.block_manager.resync.clear_backoff(&hash)?;
-			}
-			Ok(AdminRpc::Ok(format!(
-				"{} blocks returned in queue for a retry now (check logs to see results)",
-				blocks.len()
-			)))
-		}
-	}
-
-	async fn handle_block_purge(&self, yes: bool, blocks: &[String]) -> Result<AdminRpc, Error> {
-		if !yes {
-			return Err(Error::BadRequest(
-				"Pass the --yes flag to confirm block purge operation.".into(),
-			));
-		}
-
-		let mut obj_dels = 0;
-		let mut mpu_dels = 0;
-		let mut ver_dels = 0;
-
-		for hash in blocks {
-			let hash = hex::decode(hash).ok_or_bad_request("invalid hash")?;
-			let hash = Hash::try_from(&hash).ok_or_bad_request("invalid hash")?;
-			let block_refs = self
-				.garage
-				.block_ref_table
-				.get_range(&hash, None, None, 10000, Default::default())
-				.await?;
-
-			for br in block_refs {
-				if let Some(version) = self
-					.garage
-					.version_table
-					.get(&br.version, &EmptyKey)
-					.await?
-				{
-					self.handle_block_purge_version_backlink(
-						&version,
-						&mut obj_dels,
-						&mut mpu_dels,
-					)
-					.await?;
-
-					if !version.deleted.get() {
-						let deleted_version = Version::new(version.uuid, version.backlink, true);
-						self.garage.version_table.insert(&deleted_version).await?;
-						ver_dels += 1;
-					}
-				}
-			}
-		}
-
-		Ok(AdminRpc::Ok(format!(
-			"Purged {} blocks, {} versions, {} objects, {} multipart uploads",
-			blocks.len(),
-			ver_dels,
-			obj_dels,
-			mpu_dels,
-		)))
-	}
-
-	async fn handle_block_purge_version_backlink(
-		&self,
-		version: &Version,
-		obj_dels: &mut usize,
-		mpu_dels: &mut usize,
-	) -> Result<(), Error> {
-		let (bucket_id, key, ov_id) = match &version.backlink {
-			VersionBacklink::Object { bucket_id, key } => (*bucket_id, key.clone(), version.uuid),
-			VersionBacklink::MultipartUpload { upload_id } => {
-				if let Some(mut mpu) = self.garage.mpu_table.get(upload_id, &EmptyKey).await? {
-					if !mpu.deleted.get() {
-						mpu.parts.clear();
-						mpu.deleted.set();
-						self.garage.mpu_table.insert(&mpu).await?;
-						*mpu_dels += 1;
-					}
-					(mpu.bucket_id, mpu.key.clone(), *upload_id)
-				} else {
-					return Ok(());
-				}
-			}
-		};
-
-		if let Some(object) = self.garage.object_table.get(&bucket_id, &key).await? {
-			let ov = object.versions().iter().rev().find(|v| v.is_complete());
-			if let Some(ov) = ov {
-				if ov.uuid == ov_id {
-					let del_uuid = gen_uuid();
-					let deleted_object = Object::new(
-						bucket_id,
-						key,
-						vec![ObjectVersion {
-							uuid: del_uuid,
-							timestamp: ov.timestamp + 1,
-							state: ObjectVersionState::Complete(ObjectVersionData::DeleteMarker),
-						}],
-					);
-					self.garage.object_table.insert(&deleted_object).await?;
-					*obj_dels += 1;
-				}
-			}
-		}
-
-		Ok(())
-	}
-}
diff --git a/src/garage/admin/mod.rs b/src/garage/admin/mod.rs
index 1aa9482c..4f734b1a 100644
--- a/src/garage/admin/mod.rs
+++ b/src/garage/admin/mod.rs
@@ -1,5 +1,3 @@
-mod block;
-
 use std::collections::HashMap;
 use std::fmt::Write;
 use std::sync::Arc;
@@ -36,7 +34,6 @@ pub const ADMIN_RPC_PATH: &str = "garage/admin_rpc.rs/Rpc";
 pub enum AdminRpc {
 	LaunchRepair(RepairOpt),
 	Stats(StatsOpt),
-	BlockOperation(BlockOperation),
 	MetaOperation(MetaOperation),
 
 	// Replies
@@ -371,7 +368,6 @@ impl EndpointHandler<AdminRpc> for AdminRpcHandler {
 		match message {
 			AdminRpc::LaunchRepair(opt) => self.handle_launch_repair(opt.clone()).await,
 			AdminRpc::Stats(opt) => self.handle_stats(opt.clone()).await,
-			AdminRpc::BlockOperation(bo) => self.handle_block_cmd(bo).await,
 			AdminRpc::MetaOperation(mo) => self.handle_meta_cmd(mo).await,
 			m => Err(GarageError::unexpected_rpc_message(m).into()),
 		}
diff --git a/src/garage/cli_v2/block.rs b/src/garage/cli_v2/block.rs
index ff3c79e9..7d4595eb 100644
--- a/src/garage/cli_v2/block.rs
+++ b/src/garage/cli_v2/block.rs
@@ -13,14 +13,8 @@ impl Cli {
 		match cmd {
 			BlockOperation::ListErrors => self.cmd_list_block_errors().await,
 			BlockOperation::Info { hash } => self.cmd_get_block_info(hash).await,
-
-			bo => cli_v1::cmd_admin(
-				&self.admin_rpc_endpoint,
-				self.rpc_host,
-				AdminRpc::BlockOperation(bo),
-			)
-			.await
-			.ok_or_message("cli_v1"),
+			BlockOperation::RetryNow { all, blocks } => self.cmd_block_retry_now(all, blocks).await,
+			BlockOperation::Purge { yes, blocks } => self.cmd_block_purge(yes, blocks).await,
 		}
 	}
 
@@ -106,4 +100,46 @@ impl Cli {
 
 		Ok(())
 	}
+
+	pub async fn cmd_block_retry_now(&self, all: bool, blocks: Vec<String>) -> Result<(), Error> {
+		let req = match (all, blocks.len()) {
+			(true, 0) => LocalRetryBlockResyncRequest::All { all: true },
+			(false, n) if n > 0 => LocalRetryBlockResyncRequest::Blocks {
+				block_hashes: blocks,
+			},
+			_ => {
+				return Err(Error::Message(
+					"Please specify block hashes or --all (not both)".into(),
+				))
+			}
+		};
+
+		let res = self.local_api_request(req).await?;
+
+		println!(
+			"{} blocks returned in queue for a retry now (check logs to see results)",
+			res.count
+		);
+
+		Ok(())
+	}
+
+	pub async fn cmd_block_purge(&self, yes: bool, blocks: Vec<String>) -> Result<(), Error> {
+		if !yes {
+			return Err(Error::Message(
+				"Pass the --yes flag to confirm block purge operation.".into(),
+			));
+		}
+
+		let res = self
+			.local_api_request(LocalPurgeBlocksRequest(blocks))
+			.await?;
+
+		println!(
+			"Purged {} blocks: deleted {} versions, {} objects, {} multipart uploads",
+			res.blocks_purged, res.versions_deleted, res.objects_deleted, res.uploads_deleted,
+		);
+
+		Ok(())
+	}
 }