First implementation of K2V (#293)

**Specification:**

View spec at [this URL](https://git.deuxfleurs.fr/Deuxfleurs/garage/src/branch/k2v/doc/drafts/k2v-spec.md)

- [x] Specify the structure of K2V triples
- [x] Specify the DVVS format used for causality detection
- [x] Specify the K2V index (just a counter of number of values per partition key)
- [x] Specify single-item endpoints: ReadItem, InsertItem, DeleteItem
- [x] Specify index endpoint: ReadIndex
- [x] Specify multi-item endpoints: InsertBatch, ReadBatch, DeleteBatch
- [x] Move to JSON objects instead of tuples
- [x] Specify endpoints for polling for updates on single values (PollItem)

**Implementation:**

- [x] Table for K2V items, causal contexts
- [x] Indexing mechanism and table for K2V index
- [x] Make API handlers a bit more generic
- [x] K2V API endpoint
- [x] K2V API router
- [x] ReadItem
- [x] InsertItem
- [x] DeleteItem
- [x] PollItem
- [x] ReadIndex
- [x] InsertBatch
- [x] ReadBatch
- [x] DeleteBatch

**Testing:**

- [x] Just a simple Python script that does some requests to check visually that things are going right (does not contain parsing of results or assertions on returned values)
- [x] Actual tests:
  - [x] Adapt testing framework
  - [x] Simple test with InsertItem + ReadItem
  - [x] Test with several Insert/Read/DeleteItem + ReadIndex
  - [x] Test all combinations of return formats for ReadItem
  - [x] Test with ReadBatch, InsertBatch, DeleteBatch
  - [x] Test with PollItem
  - [x] Test error codes
- [ ] Fix most broken stuff
  - [x] test PollItem broken randomly
  - [x] when invalid causality tokens are given, errors should be 4xx not 5xx

**Improvements:**

- [x] Descending range queries
  - [x] Specify
  - [x] Implement
  - [x] Add test
- [x] Batch updates to index counter
- [x] Put K2V behind `k2v` feature flag

Co-authored-by: Alex Auvolat <alex@adnab.me>
Reviewed-on: https://git.deuxfleurs.fr/Deuxfleurs/garage/pulls/293
Co-authored-by: Alex <alex@adnab.me>
Co-committed-by: Alex <alex@adnab.me>

1 files changed, 187 insertions, 1 deletions

diff --git a/src/api/helpers.rs b/src/api/helpers.rs
index c2709bb3..a994b82f 100644
--- a/src/api/helpers.rs
+++ b/src/api/helpers.rs
@@ -1,6 +1,25 @@
-use crate::Error;
 use idna::domain_to_unicode;
 
+use garage_util::data::*;
+
+use garage_model::garage::Garage;
+use garage_model::key_table::Key;
+
+use crate::error::*;
+
+/// What kind of authorization is required to perform a given action
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum Authorization {
+	/// No authorization is required
+	None,
+	/// Having Read permission on bucket
+	Read,
+	/// Having Write permission on bucket
+	Write,
+	/// Having Owner permission on bucket
+	Owner,
+}
+
 /// Host to bucket
 ///
 /// Convert a host, like "bucket.garage-site.tld" to the corresponding bucket "bucket",
@@ -60,11 +79,143 @@ pub fn authority_to_host(authority: &str) -> Result<String, Error> {
 	authority.map(|h| domain_to_unicode(h).0)
 }
 
+#[allow(clippy::ptr_arg)]
+pub async fn resolve_bucket(
+	garage: &Garage,
+	bucket_name: &String,
+	api_key: &Key,
+) -> Result<Uuid, Error> {
+	let api_key_params = api_key
+		.state
+		.as_option()
+		.ok_or_internal_error("Key should not be deleted at this point")?;
+
+	if let Some(Some(bucket_id)) = api_key_params.local_aliases.get(bucket_name) {
+		Ok(*bucket_id)
+	} else {
+		Ok(garage
+			.bucket_helper()
+			.resolve_global_bucket_name(bucket_name)
+			.await?
+			.ok_or(Error::NoSuchBucket)?)
+	}
+}
+
+/// Extract the bucket name and the key name from an HTTP path and possibly a bucket provided in
+/// the host header of the request
+///
+/// S3 internally manages only buckets and keys. This function splits
+/// an HTTP path to get the corresponding bucket name and key.
+pub fn parse_bucket_key<'a>(
+	path: &'a str,
+	host_bucket: Option<&'a str>,
+) -> Result<(&'a str, Option<&'a str>), Error> {
+	let path = path.trim_start_matches('/');
+
+	if let Some(bucket) = host_bucket {
+		if !path.is_empty() {
+			return Ok((bucket, Some(path)));
+		} else {
+			return Ok((bucket, None));
+		}
+	}
+
+	let (bucket, key) = match path.find('/') {
+		Some(i) => {
+			let key = &path[i + 1..];
+			if !key.is_empty() {
+				(&path[..i], Some(key))
+			} else {
+				(&path[..i], None)
+			}
+		}
+		None => (path, None),
+	};
+	if bucket.is_empty() {
+		return Err(Error::BadRequest("No bucket specified".to_string()));
+	}
+	Ok((bucket, key))
+}
+
+const UTF8_BEFORE_LAST_CHAR: char = '\u{10FFFE}';
+
+/// Compute the key after the prefix
+pub fn key_after_prefix(pfx: &str) -> Option<String> {
+	let mut next = pfx.to_string();
+	while !next.is_empty() {
+		let tail = next.pop().unwrap();
+		if tail >= char::MAX {
+			continue;
+		}
+
+		// Circumvent a limitation of RangeFrom that overflow earlier than needed
+		// See: https://doc.rust-lang.org/core/ops/struct.RangeFrom.html
+		let new_tail = if tail == UTF8_BEFORE_LAST_CHAR {
+			char::MAX
+		} else {
+			(tail..).nth(1).unwrap()
+		};
+
+		next.push(new_tail);
+		return Some(next);
+	}
+
+	None
+}
+
 #[cfg(test)]
 mod tests {
 	use super::*;
 
 	#[test]
+	fn parse_bucket_containing_a_key() -> Result<(), Error> {
+		let (bucket, key) = parse_bucket_key("/my_bucket/a/super/file.jpg", None)?;
+		assert_eq!(bucket, "my_bucket");
+		assert_eq!(key.expect("key must be set"), "a/super/file.jpg");
+		Ok(())
+	}
+
+	#[test]
+	fn parse_bucket_containing_no_key() -> Result<(), Error> {
+		let (bucket, key) = parse_bucket_key("/my_bucket/", None)?;
+		assert_eq!(bucket, "my_bucket");
+		assert!(key.is_none());
+		let (bucket, key) = parse_bucket_key("/my_bucket", None)?;
+		assert_eq!(bucket, "my_bucket");
+		assert!(key.is_none());
+		Ok(())
+	}
+
+	#[test]
+	fn parse_bucket_containing_no_bucket() {
+		let parsed = parse_bucket_key("", None);
+		assert!(parsed.is_err());
+		let parsed = parse_bucket_key("/", None);
+		assert!(parsed.is_err());
+		let parsed = parse_bucket_key("////", None);
+		assert!(parsed.is_err());
+	}
+
+	#[test]
+	fn parse_bucket_with_vhost_and_key() -> Result<(), Error> {
+		let (bucket, key) = parse_bucket_key("/a/super/file.jpg", Some("my-bucket"))?;
+		assert_eq!(bucket, "my-bucket");
+		assert_eq!(key.expect("key must be set"), "a/super/file.jpg");
+		Ok(())
+	}
+
+	#[test]
+	fn parse_bucket_with_vhost_no_key() -> Result<(), Error> {
+		let (bucket, key) = parse_bucket_key("", Some("my-bucket"))?;
+		assert_eq!(bucket, "my-bucket");
+		assert!(key.is_none());
+		let (bucket, key) = parse_bucket_key("/", Some("my-bucket"))?;
+		assert_eq!(bucket, "my-bucket");
+		assert!(key.is_none());
+		Ok(())
+	}
+
+	#[test]
 	fn authority_to_host_with_port() -> Result<(), Error> {
 		let domain = authority_to_host("[::1]:3902")?;
 		assert_eq!(domain, "[::1]");
@@ -111,4 +262,39 @@ mod tests {
 		assert_eq!(host_to_bucket("not-garage.tld", "garage.tld"), None);
 		assert_eq!(host_to_bucket("not-garage.tld", ".garage.tld"), None);
 	}
+
+	#[test]
+	fn test_key_after_prefix() {
+		use std::iter::FromIterator;
+
+		assert_eq!(UTF8_BEFORE_LAST_CHAR as u32, (char::MAX as u32) - 1);
+		assert_eq!(key_after_prefix("a/b/").unwrap().as_str(), "a/b0");
+		assert_eq!(key_after_prefix("€").unwrap().as_str(), "₭");
+		assert_eq!(
+			key_after_prefix("􏿽").unwrap().as_str(),
+			String::from(char::from_u32(0x10FFFE).unwrap())
+		);
+
+		// When the last character is the biggest UTF8 char
+		let a = String::from_iter(['a', char::MAX].iter());
+		assert_eq!(key_after_prefix(a.as_str()).unwrap().as_str(), "b");
+
+		// When all characters are the biggest UTF8 char
+		let b = String::from_iter([char::MAX; 3].iter());
+		assert!(key_after_prefix(b.as_str()).is_none());
+
+		// Check utf8 surrogates
+		let c = String::from('\u{D7FF}');
+		assert_eq!(
+			key_after_prefix(c.as_str()).unwrap().as_str(),
+			String::from('\u{E000}')
+		);
+
+		// Check the character before the biggest one
+		let d = String::from('\u{10FFFE}');
+		assert_eq!(
+			key_after_prefix(d.as_str()).unwrap().as_str(),
+			String::from(char::MAX)
+		);
+	}
 }