Merge pull request 'Dependency upgrades: http, hyper, aws-sdk, smaller deps' (#703) from dep-upgrade-202402 into main

Reviewed-on: https://git.deuxfleurs.fr/Deuxfleurs/garage/pulls/703
author: Alex <alex@adnab.me> 2024-02-07 14:59:40 +0000
committer: Alex <alex@adnab.me> 2024-02-07 14:59:40 +0000
commit: 5d941e0100489fff552b59f0679a2a010403a21c (patch)
tree: 5c5cef9af72d48dd7347922341e43f0013380c60 /src/web/web_server.rs
parent: feeb076b7f5db7fe6fdbe3d2903fae054cde6219 (diff)
parent: e011941964b1c1e0b90f85014d166d64a83ae8e2 (diff)
download: garage-5d941e0100489fff552b59f0679a2a010403a21c.tar.gz
garage-5d941e0100489fff552b59f0679a2a010403a21c.zip
1 files changed, 44 insertions, 66 deletions
diff --git a/src/web/web_server.rs b/src/web/web_server.rs
index 73780efb..766e3829 100644
--- a/src/web/web_server.rs
+++ b/src/web/web_server.rs
@@ -5,15 +5,12 @@ use std::{convert::Infallible, sync::Arc};
 use futures::future::Future;
 
 use hyper::{
+	body::Incoming as IncomingBody,
 	header::{HeaderValue, HOST},
-	server::conn::AddrStream,
-	service::{make_service_fn, service_fn},
-	Body, Method, Request, Response, Server, StatusCode,
+	Method, Request, Response, StatusCode,
 };
 
-use hyperlocal::UnixServerExt;
-
-use tokio::net::UnixStream;
+use tokio::net::{TcpListener, UnixListener};
 
 use opentelemetry::{
 	global,
@@ -24,7 +21,8 @@ use opentelemetry::{
 
 use crate::error::*;
 
-use garage_api::helpers::{authority_to_host, host_to_bucket};
+use garage_api::generic_server::{server_loop, UnixListenerOn};
+use garage_api::helpers::*;
 use garage_api::s3::cors::{add_cors_headers, find_matching_cors_rule, handle_options_for_bucket};
 use garage_api::s3::error::{
 	CommonErrorDerivative, Error as ApiError, OkOrBadRequest, OkOrInternalError,
@@ -74,78 +72,52 @@ pub struct WebServer {
 
 impl WebServer {
 	/// Run a web server
-	pub async fn run(
-		garage: Arc<Garage>,
-		addr: UnixOrTCPSocketAddress,
-		root_domain: String,
-		shutdown_signal: impl Future<Output = ()>,
-	) -> Result<(), GarageError> {
+	pub fn new(garage: Arc<Garage>, root_domain: String) -> Arc<Self> {
 		let metrics = Arc::new(WebMetrics::new());
-		let web_server = Arc::new(WebServer {
+		Arc::new(WebServer {
 			garage,
 			metrics,
 			root_domain,
-		});
-
-		let tcp_service = make_service_fn(|conn: &AddrStream| {
-			let web_server = web_server.clone();
-
-			let client_addr = conn.remote_addr();
-			async move {
-				Ok::<_, Error>(service_fn(move |req: Request<Body>| {
-					let web_server = web_server.clone();
-
-					web_server.handle_request(req, client_addr.to_string())
-				}))
-			}
-		});
-
-		let unix_service = make_service_fn(|_: &UnixStream| {
-			let web_server = web_server.clone();
-
-			let path = addr.to_string();
-			async move {
-				Ok::<_, Error>(service_fn(move |req: Request<Body>| {
-					let web_server = web_server.clone();
-
-					web_server.handle_request(req, path.clone())
-				}))
-			}
-		});
+		})
+	}
 
-		info!("Web server listening on {}", addr);
+	pub async fn run(
+		self: Arc<Self>,
+		bind_addr: UnixOrTCPSocketAddress,
+		shutdown_signal: impl Future<Output = ()>,
+	) -> Result<(), GarageError> {
+		info!("Web server listening on {}", bind_addr);
 
-		match addr {
+		match bind_addr {
 			UnixOrTCPSocketAddress::TCPSocket(addr) => {
-				Server::bind(&addr)
-					.serve(tcp_service)
-					.with_graceful_shutdown(shutdown_signal)
-					.await?
+				let listener = TcpListener::bind(addr).await?;
+
+				let handler =
+					move |stream, socketaddr| self.clone().handle_request(stream, socketaddr);
+				server_loop(listener, handler, shutdown_signal).await
 			}
 			UnixOrTCPSocketAddress::UnixSocket(ref path) => {
 				if path.exists() {
 					fs::remove_file(path)?
 				}
 
-				let bound = Server::bind_unix(path)?;
+				let listener = UnixListener::bind(path)?;
+				let listener = UnixListenerOn(listener, path.display().to_string());
 
 				fs::set_permissions(path, Permissions::from_mode(0o222))?;
 
-				bound
-					.serve(unix_service)
-					.with_graceful_shutdown(shutdown_signal)
-					.await?;
+				let handler =
+					move |stream, socketaddr| self.clone().handle_request(stream, socketaddr);
+				server_loop(listener, handler, shutdown_signal).await
 			}
-		};
-
-		Ok(())
+		}
 	}
 
 	async fn handle_request(
 		self: Arc<Self>,
-		req: Request<Body>,
+		req: Request<IncomingBody>,
 		addr: String,
-	) -> Result<Response<Body>, Infallible> {
+	) -> Result<Response<BoxBody<Error>>, http::Error> {
 		if let Ok(forwarded_for_ip_addr) =
 			forwarded_headers::handle_forwarded_for_headers(req.headers())
 		{
@@ -187,7 +159,8 @@ impl WebServer {
 		match res {
 			Ok(res) => {
 				debug!("{} {} {}", req.method(), res.status(), req.uri());
-				Ok(res)
+				Ok(res
+					.map(|body| BoxBody::new(http_body_util::BodyExt::map_err(body, Error::from))))
 			}
 			Err(error) => {
 				info!(
@@ -220,7 +193,10 @@ impl WebServer {
 		Ok(exists)
 	}
 
-	async fn serve_file(self: &Arc<Self>, req: &Request<Body>) -> Result<Response<Body>, Error> {
+	async fn serve_file(
+		self: &Arc<Self>,
+		req: &Request<IncomingBody>,
+	) -> Result<Response<BoxBody<ApiError>>, Error> {
 		// Get http authority string (eg. [::1]:3902 or garage.tld:80)
 		let authority = req
 			.headers()
@@ -267,9 +243,11 @@ impl WebServer {
 		);
 
 		let ret_doc = match *req.method() {
-			Method::OPTIONS => handle_options_for_bucket(req, &bucket),
-			Method::HEAD => handle_head(self.garage.clone(), req, bucket_id, &key, None).await,
-			Method::GET => handle_get(self.garage.clone(), req, bucket_id, &key, None).await,
+			Method::OPTIONS => handle_options_for_bucket(req, &bucket)
+				.map_err(ApiError::from)
+				.map(|res| res.map(|_empty_body: EmptyBody| empty_body())),
+			Method::HEAD => handle_head(self.garage.clone(), &req, bucket_id, &key, None).await,
+			Method::GET => handle_get(self.garage.clone(), &req, bucket_id, &key, None).await,
 			_ => Err(ApiError::bad_request("HTTP method not supported")),
 		};
 
@@ -281,7 +259,7 @@ impl WebServer {
 				Ok(Response::builder()
 					.status(StatusCode::FOUND)
 					.header("Location", url)
-					.body(Body::empty())
+					.body(empty_body())
 					.unwrap())
 			}
 			_ => ret_doc,
@@ -310,7 +288,7 @@ impl WebServer {
 				// Create a fake HTTP request with path = the error document
 				let req2 = Request::builder()
 					.uri(format!("http://{}/{}", host, &error_document))
-					.body(Body::empty())
+					.body(empty_body::<Infallible>())
 					.unwrap();
 
 				match handle_get(self.garage.clone(), &req2, bucket_id, &error_document, None).await
@@ -358,7 +336,7 @@ impl WebServer {
 	}
 }
 
-fn error_to_res(e: Error) -> Response<Body> {
+fn error_to_res(e: Error) -> Response<BoxBody<Error>> {
 	// If we are here, it is either that:
 	// - there was an error before trying to get the requested URL
 	//   from the bucket (e.g. bucket not found)
@@ -366,7 +344,7 @@ fn error_to_res(e: Error) -> Response<Body> {
 	//   was a HEAD request or we couldn't get the error document)
 	// We do NOT enter this code path when returning the bucket's
 	// error document (this is handled in serve_file)
-	let body = Body::from(format!("{}\n", e));
+	let body = string_body(format!("{}\n", e));
 	let mut http_error = Response::new(body);
 	*http_error.status_mut() = e.http_status_code();
 	e.add_headers(http_error.headers_mut());
author	Alex <alex@adnab.me>	2024-02-07 14:59:40 +0000
committer	Alex <alex@adnab.me>	2024-02-07 14:59:40 +0000
commit	5d941e0100489fff552b59f0679a2a010403a21c (patch)
tree	5c5cef9af72d48dd7347922341e43f0013380c60 /src/web/web_server.rs
parent	feeb076b7f5db7fe6fdbe3d2903fae054cde6219 (diff)
parent	e011941964b1c1e0b90f85014d166d64a83ae8e2 (diff)
download	garage-5d941e0100489fff552b59f0679a2a010403a21c.tar.gz garage-5d941e0100489fff552b59f0679a2a010403a21c.zip