Implement multipart uploads

2026-08-29 16:37:09 +00:00 · 2020-04-26 20:39:32 +00:00
parent 1999c0ae51
commit 81ecc4999e
9 changed files with 296 additions and 59 deletions
@@ -1,11 +1,13 @@
 use std::collections::VecDeque;
+use std::fmt::Write;
 use std::sync::Arc;

 use futures::stream::*;
-use hyper::Body;
+use hyper::{Body, Request, Response};

 use garage_util::data::*;
 use garage_util::error::Error;
+use garage_table::*;

 use garage_core::block::INLINE_THRESHOLD;
 use garage_core::block_ref_table::*;
@@ -13,14 +15,17 @@ use garage_core::garage::Garage;
 use garage_core::object_table::*;
 use garage_core::version_table::*;

+use crate::http_util::*;
+
 pub async fn handle_put(
 	garage: Arc<Garage>,
-	mime_type: &str,
+	req: Request<Body>,
 	bucket: &str,
 	key: &str,
-	body: Body,
-) -> Result<UUID, Error> {
+) -> Result<Response<BodyType>, Error> {
 	let version_uuid = gen_uuid();
+	let mime_type = get_mime_type(&req)?;
+	let body = req.into_body();

 	let mut chunker = BodyChunker::new(body, garage.config.block_size);
 	let first_block = match chunker.next().await? {
@@ -31,10 +36,10 @@ pub async fn handle_put(
 	let mut object_version = ObjectVersion {
 		uuid: version_uuid,
 		timestamp: now_msec(),
-		mime_type: mime_type.to_string(),
+		mime_type,
 		size: first_block.len() as u64,
 		state: ObjectVersionState::Uploading,
-		data: ObjectVersionData::DeleteMarker,
+		data: ObjectVersionData::Uploading,
 	};

 	if first_block.len() < INLINE_THRESHOLD {
@@ -43,7 +48,7 @@ pub async fn handle_put(

 		let object = Object::new(bucket.into(), key.into(), vec![object_version]);
 		garage.object_table.insert(&object).await?;
-		return Ok(version_uuid);
+		return Ok(put_response(version_uuid));
 	}

 	let version = Version::new(version_uuid, bucket.into(), key.into(), false, vec![]);
@@ -53,9 +58,30 @@ pub async fn handle_put(
 	let object = Object::new(bucket.into(), key.into(), vec![object_version.clone()]);
 	garage.object_table.insert(&object).await?;

+	let total_size = read_and_put_blocks(&garage, version, 1, first_block, first_block_hash, &mut chunker).await?;
+
+	// TODO: if at any step we have an error, we should undo everything we did
+
+	object_version.state = ObjectVersionState::Complete;
+	object_version.size = total_size;
+
+	let object = Object::new(bucket.into(), key.into(), vec![object_version]);
+	garage.object_table.insert(&object).await?;
+
+	Ok(put_response(version_uuid))
+}
+
+async fn read_and_put_blocks(
+	garage: &Arc<Garage>,
+	version: Version,
+	part_number: u64,
+	first_block: Vec<u8>,
+	first_block_hash: Hash,
+	chunker: &mut BodyChunker,
+) -> Result<u64, Error> {
 	let mut next_offset = first_block.len();
 	let mut put_curr_version_block =
-		put_block_meta(garage.clone(), &version, 0, 0, first_block_hash);
+		put_block_meta(garage.clone(), &version, part_number, 0, first_block_hash, first_block.len() as u64);
 	let mut put_curr_block = garage
 		.block_manager
 		.rpc_put_block(first_block_hash, first_block);
@@ -67,7 +93,7 @@ pub async fn handle_put(
 			let block_hash = hash(&block[..]);
 			let block_len = block.len();
 			put_curr_version_block =
-				put_block_meta(garage.clone(), &version, 0, next_offset as u64, block_hash);
+				put_block_meta(garage.clone(), &version, part_number, next_offset as u64, block_hash, block_len as u64);
 			put_curr_block = garage.block_manager.rpc_put_block(block_hash, block);
 			next_offset += block_len;
 		} else {
@@ -75,15 +101,7 @@ pub async fn handle_put(
 		}
 	}

-	// TODO: if at any step we have an error, we should undo everything we did
-
-	object_version.state = ObjectVersionState::Complete;
-	object_version.size = next_offset as u64;
-
-	let object = Object::new(bucket.into(), key.into(), vec![object_version]);
-	garage.object_table.insert(&object).await?;
-
-	Ok(version_uuid)
+	Ok(next_offset as u64)
 }

 async fn put_block_meta(
@@ -92,6 +110,7 @@ async fn put_block_meta(
 	part_number: u64,
 	offset: u64,
 	hash: Hash,
+	size: u64,
 ) -> Result<(), Error> {
 	// TODO: don't clone, restart from empty block list ??
 	let mut version = version.clone();
@@ -100,6 +119,7 @@ async fn put_block_meta(
 			part_number,
 			offset,
 			hash,
+			size,
 		})
 		.unwrap();

@@ -154,6 +174,184 @@ impl BodyChunker {
 	}
 }

+fn put_response(version_uuid: UUID) -> Response<BodyType> {
+	let resp_bytes = format!("{}\n", hex::encode(version_uuid));
+	Response::new(Box::new(BytesBody::from(resp_bytes)))
+}
+
+pub async fn handle_create_multipart_upload(
+	garage: Arc<Garage>,
+	req: &Request<Body>,
+	bucket: &str,
+	key: &str,
+) -> Result<Response<BodyType>, Error> {
+	let version_uuid = gen_uuid();
+	let mime_type = get_mime_type(req)?;
+
+	let object_version = ObjectVersion {
+		uuid: version_uuid,
+		timestamp: now_msec(),
+		mime_type,
+		size: 0,
+		state: ObjectVersionState::Uploading,
+		data: ObjectVersionData::Uploading,
+	};
+	let object = Object::new(bucket.to_string(), key.to_string(), vec![object_version]);
+	garage.object_table.insert(&object).await?;
+
+	let mut xml = String::new();
+	writeln!(&mut xml, r#"<?xml version="1.0" encoding="UTF-8"?>"#).unwrap();
+	writeln!(
+		&mut xml,
+		r#"<InitiateMultipartUploadResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">"#
+	)
+	.unwrap();
+	writeln!(&mut xml, "\t<Bucket>{}</Bucket>", bucket).unwrap();
+	writeln!(&mut xml, "\t<Key>{}</Key>", xml_escape(key)).unwrap();
+	writeln!(
+		&mut xml,
+		"\t<UploadId>{}</UploadId>",
+		hex::encode(version_uuid)
+	)
+	.unwrap();
+	writeln!(&mut xml, "</InitiateMultipartUploadResult>").unwrap();
+
+	Ok(Response::new(Box::new(BytesBody::from(xml.into_bytes()))))
+}
+
+pub async fn handle_put_part(
+	garage: Arc<Garage>,
+	req: Request<Body>,
+	bucket: &str,
+	key: &str,
+	part_number_str: &str,
+	upload_id: &str,
+) -> Result<Response<BodyType>, Error> {
+	// Check parameters
+	let part_number = part_number_str
+		.parse::<u64>()
+		.map_err(|e| Error::BadRequest(format!("Invalid part number: {}", e)))?;
+
+	let version_uuid = uuid_from_str(upload_id).map_err(|_| Error::BadRequest(format!("Invalid upload ID")))?;
+	
+	// Read first chuck, and at the same time try to get object to see if it exists
+	let mut chunker = BodyChunker::new(req.into_body(), garage.config.block_size);
+
+	let bucket = bucket.to_string();
+	let key = key.to_string();
+	let get_object_fut = garage.object_table.get(&bucket, &key);
+	let get_first_block_fut = chunker.next();
+	let (object, first_block) = futures::try_join!(get_object_fut, get_first_block_fut)?;
+
+	// Check object is valid and multipart block can be accepted
+	let first_block = match first_block {
+		None => return Err(Error::BadRequest(format!("Empty body"))),
+		Some(x) => x,
+	};
+	let object = match object {
+		None => return Err(Error::BadRequest(format!("Object not found"))),
+		Some(x) => x,
+	};
+	if !object.versions().iter().any(|v| {
+		v.uuid == version_uuid
+			&& v.state == ObjectVersionState::Uploading
+			&& v.data == ObjectVersionData::Uploading
+	}) {
+		return Err(Error::BadRequest(format!(
+			"Multipart upload does not exist or is otherwise invalid"
+		)));
+	}
+
+	// Copy block to store
+	let version = Version::new(version_uuid, bucket.into(), key.into(), false, vec![]);
+	let first_block_hash = hash(&first_block[..]);
+	read_and_put_blocks(&garage, version, part_number, first_block, first_block_hash, &mut chunker).await?;
+
+	Ok(Response::new(Box::new(BytesBody::from(vec![]))))
+}
+
+pub async fn handle_complete_multipart_upload(
+	garage: Arc<Garage>,
+	_req: Request<Body>,
+	bucket: &str,
+	key: &str,
+	upload_id: &str,
+) -> Result<Response<BodyType>, Error> {
+	let version_uuid = uuid_from_str(upload_id).map_err(|_| Error::BadRequest(format!("Invalid upload ID")))?;
+
+	let bucket = bucket.to_string();
+	let key = key.to_string();
+	let (object, version) = futures::try_join!(
+		garage.object_table.get(&bucket, &key),
+		garage.version_table.get(&version_uuid, &EmptyKey),
+	)?;
+	let object = match object {
+		None => return Err(Error::BadRequest(format!("Object not found"))),
+		Some(x) => x,
+	};
+	let object_version = object.versions().iter().find(|v| {
+		v.uuid == version_uuid
+			&& v.state == ObjectVersionState::Uploading
+			&& v.data == ObjectVersionData::Uploading
+	});
+	let mut object_version = match object_version {
+		None => return Err(Error::BadRequest(format!(
+			"Multipart upload does not exist or has already been completed"
+		))),
+		Some(x) => x.clone(),
+	};
+	let version = match version {
+		None => return Err(Error::BadRequest(format!("Version not found"))),
+		Some(x) => x,
+	};
+	if version.blocks().len() == 0 {
+		return Err(Error::BadRequest(format!("No data was uploaded")));
+	}
+
+	// TODO: check that all the parts that they pretend they gave us are indeed there
+	// TODO: check MD5 sum of all uploaded parts? but that would mean we have to store them somewhere...
+
+	let total_size = version.blocks().iter().map(|x| x.size).fold(0, |x, y| x+y);
+	object_version.size = total_size;
+	object_version.state = ObjectVersionState::Complete;
+	object_version.data = ObjectVersionData::FirstBlock(version.blocks()[0].hash);
+	let final_object = Object::new(bucket.clone(), key.clone(), vec![object_version]);
+	garage.object_table.insert(&final_object).await?;
+
+	let mut xml = String::new();
+	writeln!(&mut xml, r#"<?xml version="1.0" encoding="UTF-8"?>"#).unwrap();
+	writeln!(
+		&mut xml,
+		r#"<CompleteMultipartUploadResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">"#
+	)
+	.unwrap();
+	writeln!(&mut xml, "\t<Location>{}</Location>", garage.config.s3_api.s3_region).unwrap();
+	writeln!(&mut xml, "\t<Bucket>{}</Bucket>", bucket).unwrap();
+	writeln!(&mut xml, "\t<Key>{}</Key>", xml_escape(&key)).unwrap();
+	writeln!(&mut xml, "</CompleteMultipartUploadResult>").unwrap();
+
+	Ok(Response::new(Box::new(BytesBody::from(xml.into_bytes()))))
+}
+
+fn get_mime_type(req: &Request<Body>) -> Result<String, Error> {
+	Ok(req
+		.headers()
+		.get(hyper::header::CONTENT_TYPE)
+		.map(|x| x.to_str())
+		.unwrap_or(Ok("blob"))?
+		.to_string())
+}
+
+fn uuid_from_str(id: &str) -> Result<UUID, ()> {
+	let id_bin = hex::decode(id).map_err(|_| ())?;
+	if id_bin.len() != 32 {
+		return Err(());
+	}
+	let mut uuid = [0u8; 32];
+	uuid.copy_from_slice(&id_bin[..]);
+	Ok(UUID::from(uuid))
+}
+
 pub async fn handle_delete(garage: Arc<Garage>, bucket: &str, key: &str) -> Result<UUID, Error> {
 	let object = match garage
 		.object_table