diff options
| -rw-r--r-- | Cargo.lock | 1 | ||||
| -rw-r--r-- | src/api/s3_put.rs | 2 | ||||
| -rw-r--r-- | src/model/block.rs | 2 | ||||
| -rw-r--r-- | src/model/bucket_table.rs | 1 | ||||
| -rw-r--r-- | src/table/lib.rs | 2 | ||||
| -rw-r--r-- | src/table/schema.rs | 4 | ||||
| -rw-r--r-- | src/web/Cargo.toml | 1 | ||||
| -rw-r--r-- | src/web/error.rs | 3 | ||||
| -rw-r--r-- | src/web/web_server.rs | 134 |
9 files changed, 21 insertions, 129 deletions
@@ -655,6 +655,7 @@ dependencies = [ "err-derive", "futures", "futures-util", + "garage_api", "garage_model 0.1.1", "garage_table 0.1.1", "garage_util 0.1.0", diff --git a/src/api/s3_put.rs b/src/api/s3_put.rs index 72613323..a1681d77 100644 --- a/src/api/s3_put.rs +++ b/src/api/s3_put.rs @@ -322,7 +322,7 @@ pub async fn handle_put_part( 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 = first_block.ok_or(Error::BadRequest(format!("Empty body")))?; + let first_block = first_block.ok_or(Error::BadRequest(format!("Empty body")))?; let object = object.ok_or(Error::BadRequest(format!("Object not found")))?; if !object diff --git a/src/model/block.rs b/src/model/block.rs index 6a5d9c5b..8a513a3c 100644 --- a/src/model/block.rs +++ b/src/model/block.rs @@ -20,7 +20,7 @@ use garage_rpc::rpc_client::*; use garage_rpc::rpc_server::*; use garage_table::table_sharded::TableShardedReplication; -use garage_table::{TableReplication, DeletedFilter}; +use garage_table::{DeletedFilter, TableReplication}; use crate::block_ref_table::*; diff --git a/src/model/bucket_table.rs b/src/model/bucket_table.rs index 35c0cc27..11f853f9 100644 --- a/src/model/bucket_table.rs +++ b/src/model/bucket_table.rs @@ -104,7 +104,6 @@ impl Entry<EmptyKey, String> for Bucket { pub struct BucketTable; - #[async_trait] impl TableSchema for BucketTable { type P = EmptyKey; diff --git a/src/table/lib.rs b/src/table/lib.rs index 7684fe9d..a10f78c2 100644 --- a/src/table/lib.rs +++ b/src/table/lib.rs @@ -12,5 +12,5 @@ pub mod table_sharded; pub mod table_sync; pub use schema::*; -pub use util::*; pub use table::*; +pub use util::*; diff --git a/src/table/schema.rs b/src/table/schema.rs index 49cede0a..d2ec9450 100644 --- a/src/table/schema.rs +++ b/src/table/schema.rs @@ -20,7 +20,6 @@ impl PartitionKey for Hash { } } - pub trait SortKey { fn sort_key(&self) -> &[u8]; } @@ -37,7 +36,6 @@ impl SortKey for Hash { } } - pub trait Entry<P: PartitionKey, S: SortKey>: PartialEq + Clone + Serialize + for<'de> Deserialize<'de> + Send + Sync { @@ -47,7 +45,6 @@ pub trait Entry<P: PartitionKey, S: SortKey>: fn merge(&mut self, other: &Self); } - #[async_trait] pub trait TableSchema: Send + Sync { type P: PartitionKey + Clone + PartialEq + Serialize + for<'de> Deserialize<'de> + Send + Sync; @@ -66,4 +63,3 @@ pub trait TableSchema: Send + Sync { true } } - diff --git a/src/web/Cargo.toml b/src/web/Cargo.toml index 819b51c1..0d08fdbf 100644 --- a/src/web/Cargo.toml +++ b/src/web/Cargo.toml @@ -16,6 +16,7 @@ path = "lib.rs" garage_util = { version = "0.1", path = "../util" } garage_table = { version = "0.1.1", path = "../table" } garage_model = { version = "0.1.1", path = "../model" } +garage_api = { version = "0.1.1", path = "../api" } rand = "0.7" hex = "0.3" diff --git a/src/web/error.rs b/src/web/error.rs index 094b22d0..59810f0f 100644 --- a/src/web/error.rs +++ b/src/web/error.rs @@ -5,6 +5,9 @@ use garage_util::error::Error as GarageError; #[derive(Debug, Error)] pub enum Error { + #[error(display = "API error: {}", _0)] + ApiError(#[error(source)] garage_api::error::Error), + // Category: internal error #[error(display = "Internal error: {}", _0)] InternalError(#[error(source)] GarageError), diff --git a/src/web/web_server.rs b/src/web/web_server.rs index 8a222738..4f79a9ec 100644 --- a/src/web/web_server.rs +++ b/src/web/web_server.rs @@ -1,26 +1,20 @@ -use std::borrow::Cow; -use std::convert::Infallible; -use std::net::SocketAddr; -use std::sync::Arc; -use std::time::{Duration, UNIX_EPOCH}; +use std::{borrow::Cow, convert::Infallible, net::SocketAddr, sync::Arc}; use futures::future::Future; -use futures::stream::*; use hyper::{ header::HOST, - body::Bytes, server::conn::AddrStream, service::{make_service_fn, service_fn}, - Body, Request, Response, Server, StatusCode}; + Body, Request, Response, Server, +}; use idna::domain_to_unicode; +use crate::error::*; +use garage_api::s3_get::handle_get; use garage_model::garage::Garage; -use garage_model::object_table::*; -use garage_table::EmptyKey; use garage_util::error::Error as GarageError; -use crate::error::*; pub async fn run_web_server( garage: Arc<Garage>, @@ -89,109 +83,9 @@ async fn serve_file(garage: Arc<Garage>, req: Request<Body>) -> Result<Response< info!("Selected bucket: \"{}\", selected key: \"{}\"", bucket, key); - // Get bucket descriptor - let object = garage - .object_table - .get(&bucket.to_string(), &key.to_string()) - .await? - .ok_or(Error::NotFound)?; - - // Get last complete version descriptor - let last_v = object - .versions() - .iter() - .rev() - .filter(|v| v.is_complete()) - .next() - .ok_or(Error::NotFound)?; - - // Unwrap version - let last_v_data = match &last_v.state { - ObjectVersionState::Complete(x) => x, - _ => unreachable!(), - }; - - // Get metadata from version - let last_v_meta = match last_v_data { - ObjectVersionData::DeleteMarker => return Err(Error::NotFound), - ObjectVersionData::Inline(meta, _) => meta, - ObjectVersionData::FirstBlock(meta, _) => meta, - }; - - // @FIXME Support range - - - // Set headers - let resp_builder = object_headers(&last_v, last_v_meta).status(StatusCode::OK); - + let r = handle_get(garage, &req, bucket, &key).await?; - // Stream body - match &last_v_data { - ObjectVersionData::DeleteMarker => unreachable!(), - ObjectVersionData::Inline(_, bytes) => { - let body: Body = Body::from(bytes.to_vec()); - Ok(resp_builder.body(body)?) - } - ObjectVersionData::FirstBlock(_, first_block_hash) => { - let read_first_block = garage.block_manager.rpc_get_block(&first_block_hash); - let get_next_blocks = garage.version_table.get(&last_v.uuid, &EmptyKey); - - let (first_block, version) = futures::try_join!(read_first_block, get_next_blocks)?; - let version = version.ok_or(Error::NotFound)?; - - let mut blocks = version - .blocks() - .iter() - .map(|vb| (vb.hash, None)) - .collect::<Vec<_>>(); - blocks[0].1 = Some(first_block); - - let body_stream = futures::stream::iter(blocks) - .map(move |(hash, data_opt)| { - let garage = garage.clone(); - async move { - if let Some(data) = data_opt { - Ok(Bytes::from(data)) - } else { - garage - .block_manager - .rpc_get_block(&hash) - .await - .map(Bytes::from) - } - } - }) - .buffered(2); - //let body: Body = Box::new(StreamBody::new(Box::pin(body_stream))); - let body = hyper::body::Body::wrap_stream(body_stream); - Ok(resp_builder.body(body)?) - } - } -} - -// Copied from api/s3_get.rs -fn object_headers( - version: &ObjectVersion, - version_meta: &ObjectVersionMeta, -) -> http::response::Builder { - let date = UNIX_EPOCH + Duration::from_millis(version.timestamp); - let date_str = httpdate::fmt_http_date(date); - - let mut resp = Response::builder() - .header( - "Content-Type", - version_meta.headers.content_type.to_string(), - ) - .header("Content-Length", format!("{}", version_meta.size)) - .header("ETag", version_meta.etag.to_string()) - .header("Last-Modified", date_str) - .header("Accept-Ranges", format!("bytes")); - - for (k, v) in version_meta.headers.other.iter() { - resp = resp.header(k, v.to_string()); - } - - resp + Ok(r) } /// Extract host from the authority section given by the HTTP host header @@ -253,11 +147,11 @@ fn host_to_bucket<'a>(host: &'a str, root: &str) -> &'a str { /// which is also AWS S3 behavior. fn path_to_key<'a>(path: &'a str, index: &str) -> Result<Cow<'a, str>, Error> { let path_utf8 = percent_encoding::percent_decode_str(&path).decode_utf8()?; - + if path_utf8.chars().next() != Some('/') { return Err(Error::BadRequest(format!( "Path must start with a / (slash)" - ))) + ))); } match path_utf8.chars().last() { @@ -270,12 +164,10 @@ fn path_to_key<'a>(path: &'a str, index: &str) -> Result<Cow<'a, str>, Error> { key.push_str(index); Ok(key.into()) } - Some(_) => { - match path_utf8 { - Cow::Borrowed(pu8) => Ok((&pu8[1..]).into()), - Cow::Owned(pu8) => Ok((&pu8[1..]).to_string().into()), - } - } + Some(_) => match path_utf8 { + Cow::Borrowed(pu8) => Ok((&pu8[1..]).into()), + Cow::Owned(pu8) => Ok((&pu8[1..]).to_string().into()), + }, } } |