feat: add request body size configuration

Add VssServiceConfig to make the maximum request body size configurable
through the server configuration file, with a hard limit of 1GB.

Additionally, add test coverage for 1GB maximum supported
value size and verifies that storage backends can
handle the configured maximum value size.

Author: enigbe

rust/api/src/kv_store_tests.rs

@@ -44,6 +44,7 @@ macro_rules! define_kv_store_tests {
 		create_test!(put_should_fail_when_global_version_mismatched);
 		create_test!(put_should_succeed_when_no_global_version_is_given);
 		create_test!(put_and_delete_should_succeed_as_atomic_transaction);
+		create_test!(put_should_succeed_with_maximum_supported_value_size);
 		create_test!(delete_should_succeed_when_item_exists);
 		create_test!(delete_should_succeed_when_item_does_not_exist);
 		create_test!(delete_should_be_idempotent);
@@ -266,6 +267,29 @@ pub trait KvStoreTestSuite {
 		Ok(())
 	}
 
+	async fn put_should_succeed_with_maximum_supported_value_size() -> Result<(), VssError> {
+		const MAXIMUM_SUPPORTED_VALUE_SIZE: usize = 1024 * 1024 * 1024;
+		const PROTOCOL_OVERHEAD_MARGIN: usize = 150;
+		let kv_store = Self::create_store().await;
+		let ctx = TestContext::new(&kv_store);
+
+		// Construct entry that's for a field that's the maximum size of a non-"large_object" object
+		let large_value = vec![0u8; MAXIMUM_SUPPORTED_VALUE_SIZE - PROTOCOL_OVERHEAD_MARGIN];
+		let kv = KeyValue { key: "k1".into(), version: 0, value: Bytes::from(large_value) };
+
+		// Put succeeds
+		ctx.put_objects(None, vec![kv]).await?;
+
+		// Retrieval succeeds
+		let result = ctx.get_object("k1").await?;
+		assert_eq!(result.value.len(), MAXIMUM_SUPPORTED_VALUE_SIZE - PROTOCOL_OVERHEAD_MARGIN);
+		assert!(result.value.iter().all(|&b| b == 0));
+
+		ctx.delete_object(result).await?;
+
+		Ok(())
+	}
+
 	async fn delete_should_succeed_when_item_exists() -> Result<(), VssError> {
 		let kv_store = Self::create_store().await;
 		let ctx = TestContext::new(&kv_store);

rust/server/src/main.rs

@@ -23,7 +23,7 @@ use api::kv_store::KvStore;
 use auth_impls::JWTAuthorizer;
 use impls::postgres_store::{Certificate, PostgresPlaintextBackend, PostgresTlsBackend};
 use util::config::{Config, ServerConfig};
-use vss_service::VssService;
+use vss_service::{VssService, VssServiceConfig};
 
 mod util;
 mod vss_service;
@@ -35,14 +35,17 @@ fn main() {
 		std::process::exit(1);
 	}
 
-	let Config { server_config: ServerConfig { host, port }, jwt_auth_config, postgresql_config } =
-		match util::config::load_config(&args[1]) {
-			Ok(cfg) => cfg,
-			Err(e) => {
-				eprintln!("Failed to load configuration: {}", e);
-				std::process::exit(1);
-			},
-		};
+	let Config {
+		server_config: ServerConfig { host, port, maximum_request_body_size },
+		jwt_auth_config,
+		postgresql_config,
+	} = match util::config::load_config(&args[1]) {
+		Ok(cfg) => cfg,
+		Err(e) => {
+			eprintln!("Failed to load configuration: {}", e);
+			std::process::exit(1);
+		},
+	};
 	let addr: SocketAddr = match format!("{}:{}", host, port).parse() {
 		Ok(addr) => addr,
 		Err(e) => {
@@ -144,7 +147,10 @@ fn main() {
 					match res {
 						Ok((stream, _)) => {
 							let io_stream = TokioIo::new(stream);
-							let vss_service = VssService::new(Arc::clone(&store), Arc::clone(&authorizer));
+							let vss_service_config = if let Some(req_body_size) = maximum_request_body_size {
+								VssServiceConfig::new(req_body_size)
+							} else {VssServiceConfig::default()};
+							let vss_service = VssService::new(Arc::clone(&store), Arc::clone(&authorizer), vss_service_config);
 							runtime.spawn(async move {
 								if let Err(err) = http1::Builder::new().serve_connection(io_stream, vss_service).await {
 									eprintln!("Failed to serve connection: {}", err);

rust/server/src/util/config.rs

@@ -11,6 +11,7 @@ pub(crate) struct Config {
 pub(crate) struct ServerConfig {
 	pub(crate) host: String,
 	pub(crate) port: u16,
+	pub(crate) maximum_request_body_size: Option<usize>,
 }
 
 #[derive(Deserialize)]

rust/server/src/vss_service.rs

@@ -20,15 +20,35 @@ use std::sync::Arc;
 
 const MAXIMUM_REQUEST_BODY_SIZE: usize = 1024 * 1024 * 1024;
 
+#[derive(Clone)]
+pub(crate) struct VssServiceConfig {
+	maximum_request_body_size: usize,
+}
+
+impl VssServiceConfig {
+	pub fn new(maximum_request_body_size: usize) -> Self {
+		Self { maximum_request_body_size: maximum_request_body_size.min(MAXIMUM_REQUEST_BODY_SIZE) }
+	}
+}
+
+impl Default for VssServiceConfig {
+	fn default() -> Self {
+		Self { maximum_request_body_size: MAXIMUM_REQUEST_BODY_SIZE }
+	}
+}
+
 #[derive(Clone)]
 pub struct VssService {
 	store: Arc<dyn KvStore>,
 	authorizer: Arc<dyn Authorizer>,
+	config: VssServiceConfig,
 }
 
 impl VssService {
-	pub(crate) fn new(store: Arc<dyn KvStore>, authorizer: Arc<dyn Authorizer>) -> Self {
-		Self { store, authorizer }
+	pub(crate) fn new(
+		store: Arc<dyn KvStore>, authorizer: Arc<dyn Authorizer>, config: VssServiceConfig,
+	) -> Self {
+		Self { store, authorizer, config }
 	}
 }
 
@@ -43,22 +63,51 @@ impl Service<Request<Incoming>> for VssService {
 		let store = Arc::clone(&self.store);
 		let authorizer = Arc::clone(&self.authorizer);
 		let path = req.uri().path().to_owned();
+		let maximum_request_body_size = self.config.maximum_request_body_size;
 
 		Box::pin(async move {
 			let prefix_stripped_path = path.strip_prefix(BASE_PATH_PREFIX).unwrap_or_default();
 
 			match prefix_stripped_path {
 				"/getObject" => {
-					handle_request(store, authorizer, req, handle_get_object_request).await
+					handle_request(
+						store,
+						authorizer,
+						req,
+						maximum_request_body_size,
+						handle_get_object_request,
+					)
+					.await
 				},
 				"/putObjects" => {
-					handle_request(store, authorizer, req, handle_put_object_request).await
+					handle_request(
+						store,
+						authorizer,
+						req,
+						maximum_request_body_size,
+						handle_put_object_request,
+					)
+					.await
 				},
 				"/deleteObject" => {
-					handle_request(store, authorizer, req, handle_delete_object_request).await
+					handle_request(
+						store,
+						authorizer,
+						req,
+						maximum_request_body_size,
+						handle_delete_object_request,
+					)
+					.await
 				},
 				"/listKeyVersions" => {
-					handle_request(store, authorizer, req, handle_list_object_request).await
+					handle_request(
+						store,
+						authorizer,
+						req,
+						maximum_request_body_size,
+						handle_list_object_request,
+					)
+					.await
 				},
 				_ => {
 					let error_msg = "Invalid request path.".as_bytes();
@@ -99,7 +148,7 @@ async fn handle_request<
 	Fut: Future<Output = Result<R, VssError>> + Send,
 >(
 	store: Arc<dyn KvStore>, authorizer: Arc<dyn Authorizer>, request: Request<Incoming>,
-	handler: F,
+	maximum_request_body_size: usize, handler: F,
 ) -> Result<<VssService as Service<Request<Incoming>>>::Response, hyper::Error> {
 	let (parts, body) = request.into_parts();
 	let headers_map = parts
@@ -113,7 +162,7 @@ async fn handle_request<
 		Err(e) => return Ok(build_error_response(e)),
 	};
 
-	let limited_body = Limited::new(body, MAXIMUM_REQUEST_BODY_SIZE.into());
+	let limited_body = Limited::new(body, maximum_request_body_size);
 	let bytes = match limited_body.collect().await {
 		Ok(body) => body.to_bytes(),
 		Err(_) => {