garage/src/garage/admin/block.rs

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::ListErrors => Ok(AdminRpc::BlockErrorList(
				self.garage.block_manager.list_resync_errors()?,
			)),
			BlockOperation::Info { hash } => self.handle_block_info(hash).await,
			BlockOperation::RetryNow { all, blocks } => {
				self.handle_block_retry_now(*all, blocks).await
			}
			BlockOperation::Purge { yes, blocks } => self.handle_block_purge(*yes, blocks).await,
		}
	}

	async fn handle_block_info(&self, hash: &String) -> Result<AdminRpc, Error> {
		let hash = hex::decode(hash).ok_or_bad_request("invalid hash")?;
		let hash = Hash::try_from(&hash).ok_or_bad_request("invalid hash")?;
		let refcount = self.garage.block_manager.get_block_rc(&hash)?;
		let block_refs = self
			.garage
			.block_ref_table
			.get_range(&hash, None, None, 10000, Default::default())
			.await?;
		let mut versions = vec![];
		let mut uploads = vec![];
		for br in block_refs {
			if let Some(v) = self
				.garage
				.version_table
				.get(&br.version, &EmptyKey)
				.await?
			{
				if let VersionBacklink::MultipartUpload { upload_id } = &v.backlink {
					if let Some(u) = self.garage.mpu_table.get(upload_id, &EmptyKey).await? {
						uploads.push(u);
					}
				}
				versions.push(Ok(v));
			} else {
				versions.push(Err(br.version));
			}
		}
		Ok(AdminRpc::BlockInfo {
			hash,
			refcount,
			versions,
			uploads,
		})
	}

	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(())
	}
}