feat: implement max/min req/res sizes

we need to prepare for implementation of FCUs deduplication
(this avalanche problem still persists).

that will require assembly of incoming requests before the decision
whether to proxy them or not is made => and for that we will need
buffers + reasonable limits.

Author: 0x416e746f6e

crates/rproxy/src/server/proxy/config/authrpc.rs

@@ -131,6 +131,28 @@ pub(crate) struct ConfigAuthrpc {
     )]
     pub(crate) log_sanitise: bool,
 
+    /// max size of authrpc requests
+    #[arg(
+        default_value = "16",
+        env = "RPROXY_AUTHRPC_MAX_REQUEST_SIZE_MB",
+        help_heading = "authrpc",
+        long("authrpc-max-request-size-mb"),
+        name("authrpc_max_request_size_mb"),
+        value_name = "megabytes"
+    )]
+    pub(crate) max_request_size_mb: usize,
+
+    /// max size of authrpc responses
+    #[arg(
+        default_value = "256",
+        env = "RPROXY_AUTHRPC_MAX_RESPONSE_SIZE_MB",
+        help_heading = "authrpc",
+        long("authrpc-max-response-size-mb"),
+        name("authrpc_max_response_size_mb"),
+        value_name = "megabytes"
+    )]
+    pub(crate) max_response_size_mb: usize,
+
     /// list of authrpc peers urls to mirror the requests to
     #[arg(
         env="RPROXY_AUTHRPC_MIRRORING_PEERS",
@@ -153,6 +175,28 @@ pub(crate) struct ConfigAuthrpc {
     #[clap(value_enum)]
     pub(crate) mirroring_strategy: ConfigProxyHttpMirroringStrategy,
 
+    /// size of preallocated authrpc request buffers
+    #[arg(
+        default_value = "1",
+        env = "RPROXY_AUTHRPC_PREALLOCATED_RESPONSE_BUFFER_SIZE_KB",
+        help_heading = "authrpc",
+        long("authrpc-preallocated-request-buffer-size-kb"),
+        name("authrpc_preallocated_request_buffer_size_kb"),
+        value_name = "kilobytes"
+    )]
+    pub(crate) prealloacated_request_buffer_size_kb: usize,
+
+    /// size of preallocated authrpc response buffers
+    #[arg(
+        default_value = "1",
+        env = "RPROXY_AUTHRPC_PREALLOCATED_RESPONSE_BUFFER_SIZE_KB",
+        help_heading = "authrpc",
+        long("authrpc-preallocated-response-buffer-size-kb"),
+        name("authrpc_preallocated_response_buffer_size_kb"),
+        value_name = "kilobytes"
+    )]
+    pub(crate) prealloacated_response_buffer_size_kb: usize,
+
     /// remove authrpc backend from mirroring peers
     #[arg(
         env = "RPROXY_AUTHRPC_REMOVE_BACKEND_FROM_MIRRORING_PEERS",
@@ -329,6 +373,16 @@ impl ConfigProxyHttp for ConfigAuthrpc {
         self.log_sanitise
     }
 
+    #[inline]
+    fn max_request_size(&self) -> usize {
+        1024 * 1024 * self.max_request_size_mb
+    }
+
+    #[inline]
+    fn max_response_size(&self) -> usize {
+        1024 * 1024 * self.max_response_size_mb
+    }
+
     #[inline]
     fn mirroring_peer_urls(&self) -> Vec<Url> {
         self.mirroring_peer_urls
@@ -341,6 +395,16 @@ impl ConfigProxyHttp for ConfigAuthrpc {
     fn mirroring_strategy(&self) -> &ConfigProxyHttpMirroringStrategy {
         &self.mirroring_strategy
     }
+
+    #[inline]
+    fn prealloacated_request_buffer_size(&self) -> usize {
+        1024 * self.prealloacated_request_buffer_size_kb
+    }
+
+    #[inline]
+    fn prealloacated_response_buffer_size(&self) -> usize {
+        1024 * self.prealloacated_response_buffer_size_kb
+    }
 }
 
 // ConfigAuthrpcError --------------------------------------------------

crates/rproxy/src/server/proxy/config/rpc.rs

@@ -130,6 +130,28 @@ pub(crate) struct ConfigRpc {
     )]
     pub(crate) log_sanitise: bool,
 
+    /// max size of rpc requests
+    #[arg(
+        default_value = "16",
+        env = "RPROXY_RPC_MAX_REQUEST_SIZE_MB",
+        help_heading = "rpc",
+        long("rpc-max-request-size-mb"),
+        name("rpc_max_request_size_mb"),
+        value_name = "megabytes"
+    )]
+    pub(crate) max_request_size_mb: usize,
+
+    /// max size of rpc responses
+    #[arg(
+        default_value = "256",
+        env = "RPROXY_RPC_MAX_RESPONSE_SIZE_MB",
+        help_heading = "rpc",
+        long("rpc-max-response-size-mb"),
+        name("rpc_max_response_size_mb"),
+        value_name = "megabytes"
+    )]
+    pub(crate) max_response_size_mb: usize,
+
     /// whether the requests that returned an error from rpc backend should
     /// be mirrored to peers
     #[arg(
@@ -162,6 +184,28 @@ pub(crate) struct ConfigRpc {
     #[clap(value_enum)]
     pub(crate) mirroring_strategy: ConfigProxyHttpMirroringStrategy,
 
+    /// size of preallocated rpc request buffers
+    #[arg(
+        default_value = "1",
+        env = "RPROXY_RPC_PREALLOCATED_RESPONSE_BUFFER_SIZE_KB",
+        help_heading = "rpc",
+        long("rpc-preallocated-request-buffer-size-kb"),
+        name("rpc_preallocated_request_buffer_size_kb"),
+        value_name = "kilobytes"
+    )]
+    pub(crate) prealloacated_request_buffer_size_kb: usize,
+
+    /// size of preallocated rpc response buffers
+    #[arg(
+        default_value = "256",
+        env = "RPROXY_RPC_PREALLOCATED_RESPONSE_BUFFER_SIZE_KB",
+        help_heading = "rpc",
+        long("rpc-preallocated-response-buffer-size-kb"),
+        name("rpc_preallocated_response_buffer_size_kb"),
+        value_name = "kilobytes"
+    )]
+    pub(crate) prealloacated_response_buffer_size_kb: usize,
+
     /// remove rpc backend from peers
     #[arg(
         env = "RPROXY_RPC_REMOVE_BACKEND_FROM_MIRRORING_PEERS",
@@ -335,6 +379,16 @@ impl ConfigProxyHttp for ConfigRpc {
         self.log_sanitise
     }
 
+    #[inline]
+    fn max_request_size(&self) -> usize {
+        1024 * 1024 * self.max_request_size_mb
+    }
+
+    #[inline]
+    fn max_response_size(&self) -> usize {
+        1024 * 1024 * self.max_response_size_mb
+    }
+
     #[inline]
     fn mirroring_peer_urls(&self) -> Vec<Url> {
         self.mirroring_peer_urls
@@ -347,6 +401,16 @@ impl ConfigProxyHttp for ConfigRpc {
     fn mirroring_strategy(&self) -> &ConfigProxyHttpMirroringStrategy {
         &self.mirroring_strategy
     }
+
+    #[inline]
+    fn prealloacated_request_buffer_size(&self) -> usize {
+        1024 * self.prealloacated_request_buffer_size_kb
+    }
+
+    #[inline]
+    fn prealloacated_response_buffer_size(&self) -> usize {
+        1024 * self.prealloacated_response_buffer_size_kb
+    }
 }
 
 // ConfigRpcError ------------------------------------------------------

crates/rproxy/src/server/proxy/http/config.rs

@@ -15,8 +15,12 @@ pub(crate) trait ConfigProxyHttp: Clone + Send + Unpin + 'static {
     fn log_proxied_requests(&self) -> bool;
     fn log_proxied_responses(&self) -> bool;
     fn log_sanitise(&self) -> bool;
+    fn max_request_size(&self) -> usize;
+    fn max_response_size(&self) -> usize;
     fn mirroring_peer_urls(&self) -> Vec<Url>;
     fn mirroring_strategy(&self) -> &ConfigProxyHttpMirroringStrategy;
+    fn prealloacated_request_buffer_size(&self) -> usize;
+    fn prealloacated_response_buffer_size(&self) -> usize;
 }
 
 // ConfigProxyHttpMirroringStrategy ------------------------------------

crates/rproxy/src/server/proxy/http/proxy.rs

@@ -31,7 +31,7 @@ use awc::{
     ClientResponse,
     Connector,
     body::MessageBody,
-    error::HeaderValue,
+    error::{HeaderValue, PayloadError},
     http::{Method, header::HeaderMap},
 };
 use bytes::Bytes;
@@ -318,7 +318,13 @@ where
         let connection_id = info.connection_id;
 
         let bck_req = this.backend.new_backend_request(&info);
-        let bck_req_body = ProxyHttpRequestBody::new(this.clone(), info, cli_req_body, timestamp);
+        let bck_req_body = ProxyHttpRequestBody::new(
+            this.clone(),
+            info,
+            cli_req_body,
+            this.shared.config().prealloacated_request_buffer_size(),
+            timestamp,
+        );
 
         let bck_res = match bck_req.send_stream(bck_req_body).await {
             Ok(res) => res,
@@ -345,12 +351,14 @@ where
         let status = bck_res.status();
         let mut cli_res = Self::to_client_response(&bck_res);
 
+        let preallocate = this.shared.config().prealloacated_response_buffer_size();
         let bck_body = ProxyHttpResponseBody::new(
             this,
             id,
             status,
             bck_res.headers().clone(),
             bck_res.into_stream(),
+            preallocate,
             timestamp,
         );
 
@@ -1205,6 +1213,7 @@ where
     info: Option<ProxyHttpRequestInfo>,
     start: UtcDateTime,
     body: Vec<u8>,
+    max_size: usize,
 
     #[pin]
     stream: S,
@@ -1216,25 +1225,28 @@ where
     P: ProxyHttpInner<C>,
 {
     fn new(
-        worker: web::Data<ProxyHttp<C, P>>,
+        proxy: web::Data<ProxyHttp<C, P>>,
         info: ProxyHttpRequestInfo,
         body: S,
+        preallocate: usize,
         timestamp: UtcDateTime,
     ) -> Self {
+        let max_size = proxy.shared.config().max_request_size();
         Self {
-            proxy: worker,
+            proxy,
             info: Some(info),
             stream: body,
             start: timestamp,
-            body: Vec::new(), // TODO: preallocate reasonable size
+            body: Vec::with_capacity(preallocate),
+            max_size,
         }
     }
 }
 
 impl<S, E, C, P> Stream for ProxyHttpRequestBody<S, C, P>
 where
     S: Stream<Item = Result<Bytes, E>>,
-    E: Debug,
+    E: From<PayloadError> + Debug,
     C: ConfigProxyHttp,
     P: ProxyHttpInner<C>,
 {
@@ -1247,6 +1259,30 @@ where
             Poll::Pending => Poll::Pending,
 
             Poll::Ready(Some(Ok(chunk))) => {
+                if this.body.len() + chunk.len() > *this.max_size {
+                    let err = format!(
+                        "request is too large: {}+ > {}",
+                        this.body.len() + chunk.len(),
+                        *this.max_size
+                    );
+                    if let Some(info) = mem::take(this.info) {
+                        warn!(
+                            proxy = P::name(),
+                            request_id = %info.id(),
+                            connection_id = %info.connection_id(),
+                            error = err,
+                            "Proxy http request stream error",
+                        );
+                    } else {
+                        warn!(
+                            proxy = P::name(),
+                            error = err,
+                            request_id = "unknown",
+                            "Proxy http request stream error",
+                        );
+                    }
+                    return Poll::Ready(Some(Err(E::from(PayloadError::Overflow))))
+                }
                 this.body.extend_from_slice(&chunk);
                 Poll::Ready(Some(Ok(chunk)))
             }
@@ -1301,6 +1337,7 @@ where
     info: Option<ProxyHttpResponseInfo>,
     start: UtcDateTime,
     body: Vec<u8>,
+    max_size: usize,
 
     #[pin]
     stream: S,
@@ -1317,13 +1354,16 @@ where
         status: StatusCode,
         headers: HeaderMap,
         body: S,
+        preallocate: usize,
         timestamp: UtcDateTime,
     ) -> Self {
+        let max_size = proxy.shared.config().max_response_size();
         Self {
             proxy,
             stream: body,
             start: timestamp,
-            body: Vec::new(), // TODO: preallocate reasonable size
+            body: Vec::with_capacity(preallocate),
+            max_size,
             info: Some(ProxyHttpResponseInfo::new(id, status, headers)),
         }
     }
@@ -1332,7 +1372,7 @@ where
 impl<S, E, C, P> Stream for ProxyHttpResponseBody<S, C, P>
 where
     S: Stream<Item = Result<Bytes, E>>,
-    E: Debug,
+    E: From<PayloadError> + Debug,
     C: ConfigProxyHttp,
     P: ProxyHttpInner<C>,
 {
@@ -1345,6 +1385,29 @@ where
             Poll::Pending => Poll::Pending,
 
             Poll::Ready(Some(Ok(chunk))) => {
+                if this.body.len() + chunk.len() > *this.max_size {
+                    let err = format!(
+                        "response is too large: {}+ > {}",
+                        this.body.len() + chunk.len(),
+                        *this.max_size
+                    );
+                    if let Some(info) = mem::take(this.info) {
+                        warn!(
+                            proxy = P::name(),
+                            request_id = %info.id(),
+                            error = err,
+                            "Proxy http response stream error",
+                        );
+                    } else {
+                        warn!(
+                            proxy = P::name(),
+                            error = err,
+                            request_id = "unknown",
+                            "Proxy http response stream error",
+                        );
+                    }
+                    return Poll::Ready(Some(Err(E::from(PayloadError::Overflow))))
+                }
                 this.body.extend_from_slice(&chunk);
                 Poll::Ready(Some(Ok(chunk)))
             }