Merge pull request 'Garage v0.9' (#473) from next into main

Reviewed-on: https://git.deuxfleurs.fr/Deuxfleurs/garage/pulls/473

5 files changed, 254 insertions, 110 deletions

diff --git a/src/garage/cli/cmd.rs b/src/garage/cli/cmd.rs
index cb7a898c..48359614 100644
--- a/src/garage/cli/cmd.rs
+++ b/src/garage/cli/cmd.rs
@@ -85,7 +85,7 @@ pub async fn cmd_status(rpc_cli: &Endpoint<SystemRpc, ()>, rpc_host: NodeID) ->
 				));
 			}
 			_ => {
-				let new_role = match layout.staging.get(&adv.id) {
+				let new_role = match layout.staging_roles.get(&adv.id) {
 					Some(NodeRoleV(Some(_))) => "(pending)",
 					_ => "NO ROLE ASSIGNED",
 				};
@@ -190,8 +190,9 @@ pub async fn cmd_admin(
 			bucket,
 			relevant_keys,
 			counters,
+			mpu_counters,
 		} => {
-			print_bucket_info(&bucket, &relevant_keys, &counters);
+			print_bucket_info(&bucket, &relevant_keys, &counters, &mpu_counters);
 		}
 		AdminRpc::KeyList(kl) => {
 			print_key_list(kl);
@@ -215,8 +216,9 @@ pub async fn cmd_admin(
 			hash,
 			refcount,
 			versions,
+			uploads,
 		} => {
-			print_block_info(hash, refcount, versions);
+			print_block_info(hash, refcount, versions, uploads);
 		}
 		r => {
 			error!("Unexpected response: {:?}", r);
diff --git a/src/garage/cli/convert_db.rs b/src/garage/cli/convert_db.rs
index f590a4b6..3c6ce69c 100644
--- a/src/garage/cli/convert_db.rs
+++ b/src/garage/cli/convert_db.rs
@@ -41,6 +41,8 @@ fn open_db(path: PathBuf, engine: String) -> Result<Db> {
 		#[cfg(feature = "sqlite")]
 		"sqlite" | "sqlite3" | "rusqlite" => {
 			let db = sqlite_adapter::rusqlite::Connection::open(&path)?;
+			db.pragma_update(None, "journal_mode", &"WAL")?;
+			db.pragma_update(None, "synchronous", &"NORMAL")?;
 			Ok(sqlite_adapter::SqliteDb::init(db))
 		}
 		#[cfg(feature = "lmdb")]
@@ -51,11 +53,13 @@ fn open_db(path: PathBuf, engine: String) -> Result<Db> {
 
 			let map_size = lmdb_adapter::recommended_map_size();
 
-			let db = lmdb_adapter::heed::EnvOpenOptions::new()
-				.max_dbs(100)
-				.map_size(map_size)
-				.open(&path)
-				.unwrap();
+			let mut env_builder = lmdb_adapter::heed::EnvOpenOptions::new();
+			env_builder.max_dbs(100);
+			env_builder.map_size(map_size);
+			unsafe {
+				env_builder.flag(lmdb_adapter::heed::flags::Flags::MdbNoMetaSync);
+			}
+			let db = env_builder.open(&path)?;
 			Ok(lmdb_adapter::LmdbDb::init(db))
 		}
 		e => Err(Error(
diff --git a/src/garage/cli/layout.rs b/src/garage/cli/layout.rs
index dc5315a1..ce2b11e0 100644
--- a/src/garage/cli/layout.rs
+++ b/src/garage/cli/layout.rs
@@ -1,3 +1,5 @@
+use bytesize::ByteSize;
+
 use format_table::format_table;
 use garage_util::crdt::Crdt;
 use garage_util::error::*;
@@ -14,8 +16,8 @@ pub async fn cli_layout_command_dispatch(
 	rpc_host: NodeID,
 ) -> Result<(), Error> {
 	match cmd {
-		LayoutOperation::Assign(configure_opt) => {
-			cmd_assign_role(system_rpc_endpoint, rpc_host, configure_opt).await
+		LayoutOperation::Assign(assign_opt) => {
+			cmd_assign_role(system_rpc_endpoint, rpc_host, assign_opt).await
 		}
 		LayoutOperation::Remove(remove_opt) => {
 			cmd_remove_role(system_rpc_endpoint, rpc_host, remove_opt).await
@@ -27,6 +29,9 @@ pub async fn cli_layout_command_dispatch(
 		LayoutOperation::Revert(revert_opt) => {
 			cmd_revert_layout(system_rpc_endpoint, rpc_host, revert_opt).await
 		}
+		LayoutOperation::Config(config_opt) => {
+			cmd_config_layout(system_rpc_endpoint, rpc_host, config_opt).await
+		}
 	}
 }
 
@@ -60,14 +65,14 @@ pub async fn cmd_assign_role(
 		.collect::<Result<Vec<_>, _>>()?;
 
 	let mut roles = layout.roles.clone();
-	roles.merge(&layout.staging);
+	roles.merge(&layout.staging_roles);
 
 	for replaced in args.replace.iter() {
 		let replaced_node = find_matching_node(layout.node_ids().iter().cloned(), replaced)?;
 		match roles.get(&replaced_node) {
 			Some(NodeRoleV(Some(_))) => {
 				layout
-					.staging
+					.staging_roles
 					.merge(&roles.update_mutator(replaced_node, NodeRoleV(None)));
 			}
 			_ => {
@@ -83,7 +88,7 @@ pub async fn cmd_assign_role(
 		return Err(Error::Message(
 				"-c and -g are mutually exclusive, please configure node either with c>0 to act as a storage node or with -g to act as a gateway node".into()));
 	}
-	if args.capacity == Some(0) {
+	if args.capacity == Some(ByteSize::b(0)) {
 		return Err(Error::Message("Invalid capacity value: 0".into()));
 	}
 
@@ -91,7 +96,7 @@ pub async fn cmd_assign_role(
 		let new_entry = match roles.get(&added_node) {
 			Some(NodeRoleV(Some(old))) => {
 				let capacity = match args.capacity {
-					Some(c) => Some(c),
+					Some(c) => Some(c.as_u64()),
 					None if args.gateway => None,
 					None => old.capacity,
 				};
@@ -108,7 +113,7 @@ pub async fn cmd_assign_role(
 			}
 			_ => {
 				let capacity = match args.capacity {
-					Some(c) => Some(c),
+					Some(c) => Some(c.as_u64()),
 					None if args.gateway => None,
 					None => return Err(Error::Message(
 							"Please specify a capacity with the -c flag, or set node explicitly as gateway with -g".into())),
@@ -125,7 +130,7 @@ pub async fn cmd_assign_role(
 		};
 
 		layout
-			.staging
+			.staging_roles
 			.merge(&roles.update_mutator(added_node, NodeRoleV(Some(new_entry))));
 	}
 
@@ -145,13 +150,13 @@ pub async fn cmd_remove_role(
 	let mut layout = fetch_layout(rpc_cli, rpc_host).await?;
 
 	let mut roles = layout.roles.clone();
-	roles.merge(&layout.staging);
+	roles.merge(&layout.staging_roles);
 
 	let deleted_node =
 		find_matching_node(roles.items().iter().map(|(id, _, _)| *id), &args.node_id)?;
 
 	layout
-		.staging
+		.staging_roles
 		.merge(&roles.update_mutator(deleted_node, NodeRoleV(None)));
 
 	send_layout(rpc_cli, rpc_host, layout).await?;
@@ -166,40 +171,45 @@ pub async fn cmd_show_layout(
 	rpc_cli: &Endpoint<SystemRpc, ()>,
 	rpc_host: NodeID,
 ) -> Result<(), Error> {
-	let mut layout = fetch_layout(rpc_cli, rpc_host).await?;
+	let layout = fetch_layout(rpc_cli, rpc_host).await?;
 
 	println!("==== CURRENT CLUSTER LAYOUT ====");
-	if !print_cluster_layout(&layout) {
-		println!("No nodes currently have a role in the cluster.");
-		println!("See `garage status` to view available nodes.");
-	}
+	print_cluster_layout(&layout, "No nodes currently have a role in the cluster.\nSee `garage status` to view available nodes.");
 	println!();
 	println!("Current cluster layout version: {}", layout.version);
 
-	if print_staging_role_changes(&layout) {
-		layout.roles.merge(&layout.staging);
-
-		println!();
-		println!("==== NEW CLUSTER LAYOUT AFTER APPLYING CHANGES ====");
-		if !print_cluster_layout(&layout) {
-			println!("No nodes have a role in the new layout.");
-		}
-		println!();
+	let has_role_changes = print_staging_role_changes(&layout);
+	if has_role_changes {
+		let v = layout.version;
+		let res_apply = layout.apply_staged_changes(Some(v + 1));
 
 		// this will print the stats of what partitions
 		// will move around when we apply
-		if layout.calculate_partition_assignation() {
-			println!("To enact the staged role changes, type:");
-			println!();
-			println!("    garage layout apply --version {}", layout.version + 1);
-			println!();
-			println!(
-				"You can also revert all proposed changes with: garage layout revert --version {}",
-				layout.version + 1
-			);
-		} else {
-			println!("Not enough nodes have an assigned role to maintain enough copies of data.");
-			println!("This new layout cannot yet be applied.");
+		match res_apply {
+			Ok((layout, msg)) => {
+				println!();
+				println!("==== NEW CLUSTER LAYOUT AFTER APPLYING CHANGES ====");
+				print_cluster_layout(&layout, "No nodes have a role in the new layout.");
+				println!();
+
+				for line in msg.iter() {
+					println!("{}", line);
+				}
+				println!("To enact the staged role changes, type:");
+				println!();
+				println!("    garage layout apply --version {}", v + 1);
+				println!();
+				println!(
+                    "You can also revert all proposed changes with: garage layout revert --version {}",
+                    v + 1)
+			}
+			Err(e) => {
+				println!("Error while trying to compute the assignment: {}", e);
+				println!("This new layout cannot yet be applied.");
+				println!(
+                    "You can also revert all proposed changes with: garage layout revert --version {}",
+                    v + 1)
+			}
 		}
 	}
 
@@ -213,11 +223,14 @@ pub async fn cmd_apply_layout(
 ) -> Result<(), Error> {
 	let layout = fetch_layout(rpc_cli, rpc_host).await?;
 
-	let layout = layout.apply_staged_changes(apply_opt.version)?;
+	let (layout, msg) = layout.apply_staged_changes(apply_opt.version)?;
+	for line in msg.iter() {
+		println!("{}", line);
+	}
 
 	send_layout(rpc_cli, rpc_host, layout).await?;
 
-	println!("New cluster layout with updated role assignation has been applied in cluster.");
+	println!("New cluster layout with updated role assignment has been applied in cluster.");
 	println!("Data will now be moved around between nodes accordingly.");
 
 	Ok(())
@@ -238,6 +251,52 @@ pub async fn cmd_revert_layout(
 	Ok(())
 }
 
+pub async fn cmd_config_layout(
+	rpc_cli: &Endpoint<SystemRpc, ()>,
+	rpc_host: NodeID,
+	config_opt: ConfigLayoutOpt,
+) -> Result<(), Error> {
+	let mut layout = fetch_layout(rpc_cli, rpc_host).await?;
+
+	let mut did_something = false;
+	match config_opt.redundancy {
+		None => (),
+		Some(r_str) => {
+			let r = r_str
+				.parse::<ZoneRedundancy>()
+				.ok_or_message("invalid zone redundancy value")?;
+			if let ZoneRedundancy::AtLeast(r_int) = r {
+				if r_int > layout.replication_factor {
+					return Err(Error::Message(format!(
+						"The zone redundancy must be smaller or equal to the \
+                    replication factor ({}).",
+						layout.replication_factor
+					)));
+				} else if r_int < 1 {
+					return Err(Error::Message(
+						"The zone redundancy must be at least 1.".into(),
+					));
+				}
+			}
+
+			layout
+				.staging_parameters
+				.update(LayoutParameters { zone_redundancy: r });
+			println!("The zone redundancy parameter has been set to '{}'.", r);
+			did_something = true;
+		}
+	}
+
+	if !did_something {
+		return Err(Error::Message(
+			"Please specify an action for `garage layout config`".into(),
+		));
+	}
+
+	send_layout(rpc_cli, rpc_host, layout).await?;
+	Ok(())
+}
+
 // --- utility ---
 
 pub async fn fetch_layout(
@@ -268,59 +327,84 @@ pub async fn send_layout(
 	Ok(())
 }
 
-pub fn print_cluster_layout(layout: &ClusterLayout) -> bool {
-	let mut table = vec!["ID\tTags\tZone\tCapacity".to_string()];
+pub fn print_cluster_layout(layout: &ClusterLayout, empty_msg: &str) {
+	let mut table = vec!["ID\tTags\tZone\tCapacity\tUsable capacity".to_string()];
 	for (id, _, role) in layout.roles.items().iter() {
 		let role = match &role.0 {
 			Some(r) => r,
 			_ => continue,
 		};
 		let tags = role.tags.join(",");
-		table.push(format!(
-			"{:?}\t{}\t{}\t{}",
-			id,
-			tags,
-			role.zone,
-			role.capacity_string()
-		));
+		let usage = layout.get_node_usage(id).unwrap_or(0);
+		let capacity = layout.get_node_capacity(id).unwrap_or(0);
+		if capacity > 0 {
+			table.push(format!(
+				"{:?}\t{}\t{}\t{}\t{} ({:.1}%)",
+				id,
+				tags,
+				role.zone,
+				role.capacity_string(),
+				ByteSize::b(usage as u64 * layout.partition_size).to_string_as(false),
+				(100.0 * usage as f32 * layout.partition_size as f32) / (capacity as f32)
+			));
+		} else {
+			table.push(format!(
+				"{:?}\t{}\t{}\t{}",
+				id,
+				tags,
+				role.zone,
+				role.capacity_string()
+			));
+		};
 	}
-	if table.len() == 1 {
-		false
-	} else {
+	if table.len() > 1 {
 		format_table(table);
-		true
+		println!();
+		println!("Zone redundancy: {}", layout.parameters.zone_redundancy);
+	} else {
+		println!("{}", empty_msg);
 	}
 }
 
 pub fn print_staging_role_changes(layout: &ClusterLayout) -> bool {
-	let has_changes = layout
-		.staging
+	let has_role_changes = layout
+		.staging_roles
 		.items()
 		.iter()
 		.any(|(k, _, v)| layout.roles.get(k) != Some(v));
+	let has_layout_changes = *layout.staging_parameters.get() != layout.parameters;
 
-	if has_changes {
+	if has_role_changes || has_layout_changes {
 		println!();
 		println!("==== STAGED ROLE CHANGES ====");
-		let mut table = vec!["ID\tTags\tZone\tCapacity".to_string()];
-		for (id, _, role) in layout.staging.items().iter() {
-			if layout.roles.get(id) == Some(role) {
-				continue;
-			}
-			if let Some(role) = &role.0 {
-				let tags = role.tags.join(",");
-				table.push(format!(
-					"{:?}\t{}\t{}\t{}",
-					id,
-					tags,
-					role.zone,
-					role.capacity_string()
-				));
-			} else {
-				table.push(format!("{:?}\tREMOVED", id));
+		if has_role_changes {
+			let mut table = vec!["ID\tTags\tZone\tCapacity".to_string()];
+			for (id, _, role) in layout.staging_roles.items().iter() {
+				if layout.roles.get(id) == Some(role) {
+					continue;
+				}
+				if let Some(role) = &role.0 {
+					let tags = role.tags.join(",");
+					table.push(format!(
+						"{:?}\t{}\t{}\t{}",
+						id,
+						tags,
+						role.zone,
+						role.capacity_string()
+					));
+				} else {
+					table.push(format!("{:?}\tREMOVED", id));
+				}
 			}
+			format_table(table);
+			println!();
+		}
+		if has_layout_changes {
+			println!(
+				"Zone redundancy: {}",
+				layout.staging_parameters.get().zone_redundancy
+			);
 		}
-		format_table(table);
 		true
 	} else {
 		false
diff --git a/src/garage/cli/structs.rs b/src/garage/cli/structs.rs
index ee7c2988..aba57551 100644
--- a/src/garage/cli/structs.rs
+++ b/src/garage/cli/structs.rs
@@ -19,7 +19,7 @@ pub enum Command {
 	#[structopt(name = "node", version = garage_version())]
 	Node(NodeOperation),
 
-	/// Operations on the assignation of node roles in the cluster layout
+	/// Operations on the assignment of node roles in the cluster layout
 	#[structopt(name = "layout", version = garage_version())]
 	Layout(LayoutOperation),
 
@@ -97,6 +97,10 @@ pub enum LayoutOperation {
 	#[structopt(name = "remove", version = garage_version())]
 	Remove(RemoveRoleOpt),
 
+	/// Configure parameters value for the layout computation
+	#[structopt(name = "config", version = garage_version())]
+	Config(ConfigLayoutOpt),
+
 	/// Show roles currently assigned to nodes and changes staged for commit
 	#[structopt(name = "show", version = garage_version())]
 	Show,
@@ -120,9 +124,9 @@ pub struct AssignRoleOpt {
 	#[structopt(short = "z", long = "zone")]
 	pub(crate) zone: Option<String>,
 
-	/// Capacity (in relative terms, use 1 to represent your smallest server)
+	/// Storage capacity, in bytes (supported suffixes: B, KB, MB, GB, TB, PB)
 	#[structopt(short = "c", long = "capacity")]
-	pub(crate) capacity: Option<u32>,
+	pub(crate) capacity: Option<bytesize::ByteSize>,
 
 	/// Gateway-only node
 	#[structopt(short = "g", long = "gateway")]
@@ -144,6 +148,13 @@ pub struct RemoveRoleOpt {
 }
 
 #[derive(StructOpt, Debug)]
+pub struct ConfigLayoutOpt {
+	/// Zone redundancy parameter ('none'/'max' or integer)
+	#[structopt(short = "r", long = "redundancy")]
+	pub(crate) redundancy: Option<String>,
+}
+
+#[derive(StructOpt, Debug)]
 pub struct ApplyLayoutOpt {
 	/// Version number of new configuration: this command will fail if
 	/// it is not exactly 1 + the previous configuration's version
@@ -323,11 +334,11 @@ pub enum KeyOperation {
 
 	/// Get key info
 	#[structopt(name = "info", version = garage_version())]
-	Info(KeyOpt),
+	Info(KeyInfoOpt),
 
 	/// Create new key
-	#[structopt(name = "new", version = garage_version())]
-	New(KeyNewOpt),
+	#[structopt(name = "create", version = garage_version())]
+	Create(KeyNewOpt),
 
 	/// Rename key
 	#[structopt(name = "rename", version = garage_version())]
@@ -351,15 +362,18 @@ pub enum KeyOperation {
 }
 
 #[derive(Serialize, Deserialize, StructOpt, Debug)]
-pub struct KeyOpt {
+pub struct KeyInfoOpt {
 	/// ID or name of the key
 	pub key_pattern: String,
+	/// Whether to display the secret key
+	#[structopt(long = "show-secret")]
+	pub show_secret: bool,
 }
 
 #[derive(Serialize, Deserialize, StructOpt, Debug)]
 pub struct KeyNewOpt {
 	/// Name of the key
-	#[structopt(long = "name", default_value = "Unnamed key")]
+	#[structopt(default_value = "Unnamed key")]
 	pub name: String,
 }
 
@@ -403,6 +417,10 @@ pub struct KeyImportOpt {
 	/// Key name
 	#[structopt(short = "n", default_value = "Imported key")]
 	pub name: String,
+
+	/// Confirm key import
+	#[structopt(long = "yes")]
+	pub yes: bool,
 }
 
 #[derive(Serialize, Deserialize, StructOpt, Debug, Clone)]
@@ -438,24 +456,30 @@ pub struct RepairOpt {
 
 #[derive(Serialize, Deserialize, StructOpt, Debug, Eq, PartialEq, Clone)]
 pub enum RepairWhat {
-	/// Only do a full sync of metadata tables
+	/// Do a full sync of metadata tables
 	#[structopt(name = "tables", version = garage_version())]
 	Tables,
-	/// Only repair (resync/rebalance) the set of stored blocks
+	/// Repair (resync/rebalance) the set of stored blocks in the cluster
 	#[structopt(name = "blocks", version = garage_version())]
 	Blocks,
-	/// Only redo the propagation of object deletions to the version table (slow)
+	/// Repropagate object deletions to the version table
 	#[structopt(name = "versions", version = garage_version())]
 	Versions,
-	/// Only redo the propagation of version deletions to the block ref table (extremely slow)
+	/// Repropagate object deletions to the multipart upload table
+	#[structopt(name = "mpu", version = garage_version())]
+	MultipartUploads,
+	/// Repropagate version deletions to the block ref table
 	#[structopt(name = "block_refs", version = garage_version())]
 	BlockRefs,
-	/// Verify integrity of all blocks on disc (extremely slow, i/o intensive)
+	/// Verify integrity of all blocks on disc
 	#[structopt(name = "scrub", version = garage_version())]
 	Scrub {
 		#[structopt(subcommand)]
 		cmd: ScrubCmd,
 	},
+	/// Rebalance data blocks among HDDs on individual nodes
+	#[structopt(name = "rebalance", version = garage_version())]
+	Rebalance,
 }
 
 #[derive(Serialize, Deserialize, StructOpt, Debug, Eq, PartialEq, Clone)]
diff --git a/src/garage/cli/util.rs b/src/garage/cli/util.rs
index 1140cf22..2232d395 100644
--- a/src/garage/cli/util.rs
+++ b/src/garage/cli/util.rs
@@ -12,8 +12,9 @@ use garage_block::manager::BlockResyncErrorInfo;
 
 use garage_model::bucket_table::*;
 use garage_model::key_table::*;
-use garage_model::s3::object_table::{BYTES, OBJECTS, UNFINISHED_UPLOADS};
-use garage_model::s3::version_table::Version;
+use garage_model::s3::mpu_table::{self, MultipartUpload};
+use garage_model::s3::object_table;
+use garage_model::s3::version_table::*;
 
 use crate::cli::structs::WorkerListOpt;
 
@@ -135,6 +136,7 @@ pub fn print_bucket_info(
 	bucket: &Bucket,
 	relevant_keys: &HashMap<String, Key>,
 	counters: &HashMap<String, i64>,
+	mpu_counters: &HashMap<String, i64>,
 ) {
 	let key_name = |k| {
 		relevant_keys
@@ -148,7 +150,7 @@ pub fn print_bucket_info(
 		Deletable::Deleted => println!("Bucket is deleted."),
 		Deletable::Present(p) => {
 			let size =
-				bytesize::ByteSize::b(counters.get(BYTES).cloned().unwrap_or_default() as u64);
+				bytesize::ByteSize::b(*counters.get(object_table::BYTES).unwrap_or(&0) as u64);
 			println!(
 				"\nSize: {} ({})",
 				size.to_string_as(true),
@@ -156,14 +158,22 @@ pub fn print_bucket_info(
 			);
 			println!(
 				"Objects: {}",
-				counters.get(OBJECTS).cloned().unwrap_or_default()
+				*counters.get(object_table::OBJECTS).unwrap_or(&0)
+			);
+			println!(
+				"Unfinished uploads (multipart and non-multipart): {}",
+				*counters.get(object_table::UNFINISHED_UPLOADS).unwrap_or(&0)
 			);
 			println!(
 				"Unfinished multipart uploads: {}",
-				counters
-					.get(UNFINISHED_UPLOADS)
-					.cloned()
-					.unwrap_or_default()
+				*mpu_counters.get(mpu_table::UPLOADS).unwrap_or(&0)
+			);
+			let mpu_size =
+				bytesize::ByteSize::b(*mpu_counters.get(mpu_table::BYTES).unwrap_or(&0) as u64);
+			println!(
+				"Size of unfinished multipart uploads: {} ({})",
+				mpu_size.to_string_as(true),
+				mpu_size.to_string_as(false),
 			);
 
 			println!("\nWebsite access: {}", p.website_config.get().is_some());
@@ -390,29 +400,49 @@ pub fn print_block_error_list(el: Vec<BlockResyncErrorInfo>) {
 	format_table(table);
 }
 
-pub fn print_block_info(hash: Hash, refcount: u64, versions: Vec<Result<Version, Uuid>>) {
+pub fn print_block_info(
+	hash: Hash,
+	refcount: u64,
+	versions: Vec<Result<Version, Uuid>>,
+	uploads: Vec<MultipartUpload>,
+) {
 	println!("Block hash: {}", hex::encode(hash.as_slice()));
 	println!("Refcount: {}", refcount);
 	println!();
 
-	let mut table = vec!["Version\tBucket\tKey\tDeleted".into()];
+	let mut table = vec!["Version\tBucket\tKey\tMPU\tDeleted".into()];
 	let mut nondeleted_count = 0;
 	for v in versions.iter() {
 		match v {
 			Ok(ver) => {
-				table.push(format!(
-					"{:?}\t{:?}\t{}\t{:?}",
-					ver.uuid,
-					ver.bucket_id,
-					ver.key,
-					ver.deleted.get()
-				));
+				match &ver.backlink {
+					VersionBacklink::Object { bucket_id, key } => {
+						table.push(format!(
+							"{:?}\t{:?}\t{}\t\t{:?}",
+							ver.uuid,
+							bucket_id,
+							key,
+							ver.deleted.get()
+						));
+					}
+					VersionBacklink::MultipartUpload { upload_id } => {
+						let upload = uploads.iter().find(|x| x.upload_id == *upload_id);
+						table.push(format!(
+							"{:?}\t{:?}\t{}\t{:?}\t{:?}",
+							ver.uuid,
+							upload.map(|u| u.bucket_id).unwrap_or_default(),
+							upload.map(|u| u.key.as_str()).unwrap_or_default(),
+							upload_id,
+							ver.deleted.get()
+						));
+					}
+				}
 				if !ver.deleted.get() {
 					nondeleted_count += 1;
 				}
 			}
 			Err(vh) => {
-				table.push(format!("{:?}\t\t\tyes", vh));
+				table.push(format!("{:?}\t\t\t\tyes", vh));
 			}
 		}
 	}