feat: remove chunking from rpc

comms/core/src/protocol/messaging/inbound.rs

@@ -22,7 +22,7 @@
 
 use std::io;
 
-use futures::{future::Either, SinkExt, StreamExt};
+use futures::{future, future::Either, SinkExt, StreamExt};
 use log::*;
 use tari_shutdown::ShutdownSignal;
 use tokio::{
@@ -33,7 +33,7 @@ use tokio::{
 #[cfg(feature = "metrics")]
 use super::metrics;
 use super::{MessagingEvent, MessagingProtocol};
-use crate::{message::InboundMessage, peer_manager::NodeId, protocol::rpc::__macro_reexports::future};
+use crate::{message::InboundMessage, peer_manager::NodeId};
 
 const LOG_TARGET: &str = "comms::protocol::messaging::inbound";
 

comms/core/src/protocol/rpc/client/mod.rs

@@ -75,7 +75,6 @@ use crate::{
             RpcError,
             RpcServerError,
             RpcStatus,
-            RPC_CHUNKING_MAX_CHUNKS,
         },
         ProtocolId,
     },
@@ -932,53 +931,17 @@ where TSubstream: AsyncRead + AsyncWrite + Unpin
 
     pub async fn read_response(&mut self) -> Result<proto::rpc::RpcResponse, RpcError> {
         let timer = Instant::now();
-        let mut resp = self.next().await?;
+        let resp = self.next().await?;
         self.time_to_first_msg = Some(timer.elapsed());
         self.check_response(&resp)?;
-        let mut chunk_count = 1;
-        let mut last_chunk_flags =
-            RpcMessageFlags::from_bits(u8::try_from(resp.flags).map_err(|_| {
-                RpcStatus::protocol_error(&format!("invalid message flag: must be less than {}", u8::MAX))
-            })?)
-            .ok_or(RpcStatus::protocol_error(&format!(
-                "invalid message flag, does not match any flags ({})",
-                resp.flags
-            )))?;
-        let mut last_chunk_size = resp.payload.len();
-        self.bytes_read += last_chunk_size;
-        loop {
-            trace!(
-                target: LOG_TARGET,
-                "Chunk {} received (flags={:?}, {} bytes, {} total)",
-                chunk_count,
-                last_chunk_flags,
-                last_chunk_size,
-                resp.payload.len()
-            );
-            if !last_chunk_flags.is_more() {
-                return Ok(resp);
-            }
-
-            if chunk_count >= RPC_CHUNKING_MAX_CHUNKS {
-                return Err(RpcError::RemotePeerExceededMaxChunkCount {
-                    expected: RPC_CHUNKING_MAX_CHUNKS,
-                });
-            }
-
-            let msg = self.next().await?;
-            last_chunk_flags = RpcMessageFlags::from_bits(u8::try_from(msg.flags).map_err(|_| {
-                RpcStatus::protocol_error(&format!("invalid message flag: must be less than {}", u8::MAX))
-            })?)
-            .ok_or(RpcStatus::protocol_error(&format!(
-                "invalid message flag, does not match any flags ({})",
-                resp.flags
-            )))?;
-            last_chunk_size = msg.payload.len();
-            self.bytes_read += last_chunk_size;
-            self.check_response(&resp)?;
-            resp.payload.extend(msg.payload);
-            chunk_count += 1;
-        }
+        self.bytes_read = resp.payload.len();
+        trace!(
+            target: LOG_TARGET,
+            "Received {} bytes in {:.2?}",
+            resp.payload.len(),
+            self.time_to_first_msg.unwrap_or_default()
+        );
+        Ok(resp)
     }
 
     pub async fn read_ack(&mut self) -> Result<proto::rpc::RpcResponse, RpcError> {

comms/core/src/protocol/rpc/message.rs

@@ -24,19 +24,24 @@ use std::{convert::TryFrom, fmt, time::Duration};
 
 use bitflags::bitflags;
 use bytes::Bytes;
+use log::warn;
 
 use super::RpcError;
 use crate::{
     proto,
     proto::rpc::rpc_session_reply::SessionResult,
-    protocol::rpc::{
-        body::{Body, IntoBody},
-        context::RequestContext,
-        error::HandshakeRejectReason,
-        RpcStatusCode,
+    protocol::{
+        rpc,
+        rpc::{
+            body::{Body, IntoBody},
+            context::RequestContext,
+            error::HandshakeRejectReason,
+            RpcStatusCode,
+        },
     },
 };
 
+const LOG_TARGET: &str = "comms::rpc::message";
 #[derive(Debug)]
 pub struct Request<T> {
     pub(super) context: Option<RequestContext>,
@@ -203,8 +208,6 @@ bitflags! {
         const FIN = 0x01;
         /// Typically sent with empty contents and used to confirm a substream is alive.
         const ACK = 0x02;
-        /// Another chunk to be received
-        const MORE = 0x04;
     }
 }
 impl RpcMessageFlags {
@@ -215,10 +218,6 @@ impl RpcMessageFlags {
     pub fn is_ack(self) -> bool {
         self.contains(Self::ACK)
     }
-
-    pub fn is_more(self) -> bool {
-        self.contains(Self::MORE)
-    }
 }
 
 impl Default for RpcMessageFlags {
@@ -276,6 +275,21 @@ impl RpcResponse {
             payload: self.payload.to_vec(),
         }
     }
+
+    pub fn exceeded_message_size(self) -> RpcResponse {
+        let msg = format!(
+            "The response size exceeded the maximum allowed payload size. Max = {} bytes, Got = {} bytes",
+            rpc::max_response_payload_size() as f32,
+            self.payload.len() as f32,
+        );
+        warn!(target: LOG_TARGET, "{}", msg);
+        RpcResponse {
+            request_id: self.request_id,
+            status: RpcStatusCode::MalformedResponse,
+            flags: RpcMessageFlags::FIN,
+            payload: msg.into_bytes().into(),
+        }
+    }
 }
 
 impl Default for RpcResponse {

comms/core/src/protocol/rpc/mod.rs

@@ -30,31 +30,22 @@ mod test;
 
 /// Maximum frame size of each RPC message. This is enforced in tokio's length delimited codec.
 /// This can be thought of as the hard limit on message size.
-pub const RPC_MAX_FRAME_SIZE: usize = 3 * 1024 * 1024; // 3 MiB
-/// Maximum number of chunks into which a message can be broken up.
-const RPC_CHUNKING_MAX_CHUNKS: usize = 16; // 16 x 256 Kib = 4 MiB max combined message size
-const RPC_CHUNKING_THRESHOLD: usize = 256 * 1024;
-const RPC_CHUNKING_SIZE_LIMIT: usize = 384 * 1024;
+pub const RPC_MAX_FRAME_SIZE: usize = 4 * 1024 * 1024; // 4 MiB
 
 /// The maximum request payload size
 const fn max_request_size() -> usize {
     RPC_MAX_FRAME_SIZE
 }
 
-/// The maximum size for a single RPC response message
-const fn max_response_size() -> usize {
-    RPC_CHUNKING_MAX_CHUNKS * RPC_CHUNKING_THRESHOLD
-}
-
-/// The maximum size for a single RPC response excluding overhead
+/// The maximum size for a single RPC response body excluding response header overhead
 const fn max_response_payload_size() -> usize {
     // RpcResponse overhead is:
     // - 4 varint protobuf fields, each field ID is 1 byte
     // - 3 u32 fields, VarInt(u32::MAX) is 5 bytes
     // - 1 length varint for the payload, allow for 5 bytes to be safe (max_payload_size being technically too small is
     //   fine, being too large isn't)
     const MAX_HEADER_SIZE: usize = 4 + 4 * 5;
-    max_response_size() - MAX_HEADER_SIZE
+    RPC_MAX_FRAME_SIZE - MAX_HEADER_SIZE
 }
 
 mod body;