Implement DeleteObjects

6 files changed, 145 insertions, 25 deletions

diff --git a/src/api/Cargo.toml b/src/api/Cargo.toml
index 11a42be9..b8674fba 100644
--- a/src/api/Cargo.toml
+++ b/src/api/Cargo.toml
@@ -31,4 +31,4 @@ hyper = "0.13"
 url = "2.1"
 httpdate = "0.3"
 percent-encoding = "2.1.0"
-
+roxmltree = "0.11"
diff --git a/src/api/api_server.rs b/src/api/api_server.rs
index 642697da..913ff0af 100644
--- a/src/api/api_server.rs
+++ b/src/api/api_server.rs
@@ -74,8 +74,7 @@ async fn handler_inner(
 ) -> Result<Response<BodyType>, Error> {
 	let path = req.uri().path().to_string();
 	let path = percent_encoding::percent_decode_str(&path)
-		.decode_utf8()
-		.map_err(|e| Error::BadRequest(format!("Invalid utf8 key ({})", e)))?;
+		.decode_utf8()?;
 
 	let (bucket, key) = parse_bucket_key(&path)?;
 	if bucket.len() == 0 {
@@ -125,10 +124,7 @@ async fn handler_inner(
 					// CopyObject query
 					let copy_source = req.headers().get("x-amz-copy-source").unwrap().to_str()?;
 					let copy_source = percent_encoding::percent_decode_str(&copy_source)
-						.decode_utf8()
-						.map_err(|e| {
-							Error::BadRequest(format!("Invalid utf8 copy_source ({})", e))
-						})?;
+						.decode_utf8()?;
 					let (source_bucket, source_key) = parse_bucket_key(&copy_source)?;
 					if !api_key.allow_read(&source_bucket) {
 						return Err(Error::Forbidden(format!(
@@ -153,8 +149,7 @@ async fn handler_inner(
 					Ok(handle_abort_multipart_upload(garage, &bucket, &key, upload_id).await?)
 				} else {
 					// DeleteObject query
-					let version_uuid = handle_delete(garage, &bucket, &key).await?;
-					Ok(put_response(version_uuid))
+					Ok(handle_delete(garage, &bucket, &key).await?)
 				}
 			}
 			&Method::POST => {
@@ -233,7 +228,22 @@ async fn handler_inner(
 				)
 				.await?)
 			}
-			_ => Err(Error::BadRequest(format!("Invalid method"))),
+            &Method::POST => {
+                if params.contains_key(&"delete".to_string()) {
+                    // DeleteObjects
+                    Ok(handle_delete_objects(garage, bucket, req).await?)
+                } else {
+                    println!(
+                        "Body: {}",
+                        std::str::from_utf8(&hyper::body::to_bytes(req.into_body()).await?)
+                            .unwrap_or("<invalid utf8>")
+                    );
+                    Err(Error::BadRequest(format!("Unsupported call")))
+                }
+            }
+			_ => {
+                Err(Error::BadRequest(format!("Invalid method")))
+            }
 		}
 	}
 }
diff --git a/src/api/s3_delete.rs b/src/api/s3_delete.rs
index 4d6805fb..001eb162 100644
--- a/src/api/s3_delete.rs
+++ b/src/api/s3_delete.rs
@@ -1,4 +1,7 @@
 use std::sync::Arc;
+use std::fmt::Write;
+
+use hyper::{Body, Request, Response};
 
 use garage_util::data::*;
 use garage_util::error::Error;
@@ -6,7 +9,10 @@ use garage_util::error::Error;
 use garage_core::garage::Garage;
 use garage_core::object_table::*;
 
-pub async fn handle_delete(garage: Arc<Garage>, bucket: &str, key: &str) -> Result<UUID, Error> {
+use crate::http_util::*;
+use crate::encoding::*;
+
+async fn handle_delete_internal(garage: &Garage, bucket: &str, key: &str) -> Result<(UUID, UUID), Error> {
 	let object = match garage
 		.object_table
 		.get(&bucket.to_string(), &key.to_string())
@@ -14,7 +20,7 @@ pub async fn handle_delete(garage: Arc<Garage>, bucket: &str, key: &str) -> Resu
 	{
 		None => {
 			// No need to delete
-			return Ok([0u8; 32].into());
+			return Err(Error::NotFound);
 		}
 		Some(o) => o,
 	};
@@ -23,16 +29,19 @@ pub async fn handle_delete(garage: Arc<Garage>, bucket: &str, key: &str) -> Resu
 		v.data != ObjectVersionData::DeleteMarker && v.state != ObjectVersionState::Aborted
 	});
 
-	let mut must_delete = false;
+	let mut must_delete = None;
 	let mut timestamp = now_msec();
 	for v in interesting_versions {
-		must_delete = true;
+        if v.timestamp + 1 > timestamp || must_delete.is_none() {
+		    must_delete = Some(v.uuid);
+        }
 		timestamp = std::cmp::max(timestamp, v.timestamp + 1);
 	}
 
-	if !must_delete {
-		return Ok([0u8; 32].into());
-	}
+    let deleted_version = match must_delete {
+        None => return Err(Error::NotFound),
+        Some(v) => v,
+    };
 
 	let version_uuid = gen_uuid();
 
@@ -50,5 +59,88 @@ pub async fn handle_delete(garage: Arc<Garage>, bucket: &str, key: &str) -> Resu
 	);
 
 	garage.object_table.insert(&object).await?;
-	return Ok(version_uuid);
+	return Ok((deleted_version, version_uuid));
+}
+
+pub async fn handle_delete(garage: Arc<Garage>, bucket: &str, key: &str) -> Result<Response<BodyType>, Error> {
+    let (_deleted_version, delete_marker_version) = handle_delete_internal(&garage, bucket, key).await?;
+
+	Ok(Response::builder()
+		.header("x-amz-version-id", hex::encode(delete_marker_version))
+		.body(empty_body())
+		.unwrap())
+}
+
+pub async fn handle_delete_objects(garage: Arc<Garage>, bucket: &str, req: Request<Body>) -> Result<Response<BodyType>, Error> {
+    let body = hyper::body::to_bytes(req.into_body()).await?;
+    let cmd_xml = roxmltree::Document::parse(&std::str::from_utf8(&body)?)?;
+    let cmd = parse_delete_objects_xml(&cmd_xml)
+        .map_err(|e| Error::BadRequest(format!("Invald delete XML query: {}", e)))?;
+
+    let mut retxml = String::new();
+	writeln!(&mut retxml, r#"<?xml version="1.0" encoding="UTF-8"?>"#).unwrap();
+	writeln!(&mut retxml, "<DeleteObjectsOutput>").unwrap();
+
+    for obj in cmd.objects.iter() {
+        match handle_delete_internal(&garage, bucket, &obj.key).await {
+            Ok((deleted_version, delete_marker_version)) => {
+                writeln!(&mut retxml, "\t<Deleted>").unwrap();
+                writeln!(&mut retxml, "\t\t<Key>{}</Key>", obj.key).unwrap();
+                writeln!(&mut retxml, "\t\t<VersionId>{}</VersionId>", hex::encode(deleted_version)).unwrap();
+                writeln!(&mut retxml, "\t\t<DeleteMarkerVersionId>{}</DeleteMarkerVersionId>", hex::encode(delete_marker_version)).unwrap();
+                writeln!(&mut retxml, "\t</Deleted>").unwrap();
+            }
+            Err(e) => {
+                writeln!(&mut retxml, "\t<Error>").unwrap();
+                writeln!(&mut retxml, "\t\t<Code>{}</Code>", e.http_status_code()).unwrap();
+                writeln!(&mut retxml, "\t\t<Key>{}</Key>", obj.key).unwrap();
+                writeln!(&mut retxml, "\t\t<Message>{}</Message>", xml_escape(&format!("{}", e))).unwrap();
+                writeln!(&mut retxml, "\t</Error>").unwrap();
+            }
+        }
+    }
+
+	writeln!(&mut retxml, "</DeleteObjectsOutput>").unwrap();
+
+	Ok(Response::new(Box::new(BytesBody::from(retxml.into_bytes()))))
+}
+
+struct DeleteRequest {
+    objects: Vec<DeleteObject>,
 }
+
+struct DeleteObject {
+    key: String,
+}
+
+fn parse_delete_objects_xml(xml: &roxmltree::Document) -> Result<DeleteRequest, String> {
+    let mut ret = DeleteRequest{objects: vec![]};
+
+    let root = xml.root();
+    let delete = match root.first_child() {
+        Some(del) => del,
+        None => return Err(format!("Delete tag not found")),
+    };
+    if !delete.has_tag_name("Delete") {
+        return Err(format!("Invalid root tag: {:?}", root));
+    }
+
+    for item in delete.children() {
+        if item.has_tag_name("Object") {
+            if let Some(key) = item.children().find(|e| e.has_tag_name("Key")) {
+                if let Some(key_str) = key.text() {
+                    ret.objects.push(DeleteObject{key: key_str.to_string()});
+                } else {
+                    return Err(format!("No text for key: {:?}", key));
+                }
+            } else {
+                return Err(format!("No delete key for item: {:?}", item));
+            }
+        } else {
+            return Err(format!("Invalid delete item: {:?}", item));
+        }
+    }
+
+    Ok(ret)
+}
+
diff --git a/src/api/s3_list.rs b/src/api/s3_list.rs
index df792da2..d4d8161e 100644
--- a/src/api/s3_list.rs
+++ b/src/api/s3_list.rs
@@ -30,34 +30,38 @@ pub async fn handle_list(
 	let mut result_keys = BTreeMap::<String, ListResultInfo>::new();
 	let mut result_common_prefixes = BTreeSet::<String>::new();
 
-	let mut truncated = true;
 	let mut next_chunk_start = marker.unwrap_or(prefix).to_string();
 
 	debug!("List request: `{}` {} `{}`", delimiter, max_keys, prefix);
 
-	while result_keys.len() + result_common_prefixes.len() < max_keys && truncated {
+	let truncated;
+	'query_loop: loop {
 		let objects = garage
 			.object_table
 			.get_range(
 				&bucket.to_string(),
 				Some(next_chunk_start.clone()),
 				Some(()),
-				max_keys,
+				max_keys + 1,
 			)
 			.await?;
 		debug!(
 			"List: get range {} (max {}), results: {}",
 			next_chunk_start,
-			max_keys,
+			max_keys + 1,
 			objects.len()
 		);
 
 		for object in objects.iter() {
 			if !object.key.starts_with(prefix) {
 				truncated = false;
-				break;
+				break 'query_loop;
 			}
 			if let Some(version) = object.versions().iter().find(|x| x.is_data()) {
+                if result_keys.len() + result_common_prefixes.len() >= max_keys {
+                    truncated = true;
+                    break 'query_loop;
+                }
 				let common_prefix = if delimiter.len() > 0 {
 					let relative_key = &object.key[prefix.len()..];
 					match relative_key.find(delimiter) {
@@ -83,8 +87,9 @@ pub async fn handle_list(
 				};
 			}
 		}
-		if objects.len() < max_keys {
-			truncated = false;
+		if objects.len() < max_keys + 1 {
+            truncated = false;
+            break 'query_loop;
 		}
 		if objects.len() > 0 {
 			next_chunk_start = objects[objects.len() - 1].key.clone();
diff --git a/src/util/Cargo.toml b/src/util/Cargo.toml
index 6f61a586..d8c39c62 100644
--- a/src/util/Cargo.toml
+++ b/src/util/Cargo.toml
@@ -32,4 +32,5 @@ hyper = "0.13"
 rustls = "0.17"
 webpki = "0.21"
 
+roxmltree = "0.11"
 
diff --git a/src/util/error.rs b/src/util/error.rs
index cb99cbbc..d2ed1ccf 100644
--- a/src/util/error.rs
+++ b/src/util/error.rs
@@ -114,3 +114,15 @@ impl<T> From<tokio::sync::mpsc::error::SendError<T>> for Error {
 		Error::Message(format!("MPSC send error"))
 	}
 }
+
+impl From<std::str::Utf8Error> for Error {
+    fn from(e: std::str::Utf8Error) -> Error {
+        Error::BadRequest(format!("Invalid UTF-8: {}", e))
+    }
+}
+
+impl From<roxmltree::Error> for Error {
+    fn from(e: roxmltree::Error) -> Error {
+        Error::BadRequest(format!("Invalid XML: {}", e))
+    }
+}