Repository: mintlayer/mintlayer-core
Pull request #782 (2023-03-23 21:01:17)

Use one config parameter instead of two constants (message size and announcement size)
Closes https://github.com/mintlayer/mintlayer-core/issues/518.


Patch:
diff --git a/dns_server/src/main.rs b/dns_server/src/main.rs
index 594ddfc60..87f17b6da 100644
--- a/dns_server/src/main.rs
+++ b/dns_server/src/main.rs
@@ -62,6 +62,7 @@ async fn run(config: Arc<DnsServerConfig>) -> Result<void::Void, error::DnsServe
         msg_max_locator_count: Default::default(),
         max_request_blocks_count: Default::default(),
         user_agent,
+        max_message_size: Default::default(),
     });
 
     let transport = p2p::make_p2p_transport();
diff --git a/node/src/config_files/p2p.rs b/node/src/config_files/p2p.rs
index 2c41d84b3..11ae7f107 100644
--- a/node/src/config_files/p2p.rs
+++ b/node/src/config_files/p2p.rs
@@ -102,6 +102,7 @@ impl From<P2pConfigFile> for P2pConfig {
             msg_max_locator_count: Default::default(),
             max_request_blocks_count: Default::default(),
             user_agent: mintlayer_core_user_agent(),
+            max_message_size: Default::default(),
         }
     }
 }
diff --git a/p2p/src/config.rs b/p2p/src/config.rs
index 75e6ca1ae..27395f1c1 100644
--- a/p2p/src/config.rs
+++ b/p2p/src/config.rs
@@ -31,6 +31,7 @@ make_config_setting!(PingTimeout, Duration, Duration::from_secs(150));
 make_config_setting!(HeaderLimit, usize, 2000);
 make_config_setting!(MaxLocatorSize, usize, 101);
 make_config_setting!(RequestedBlocksLimit, usize, 500);
+make_config_setting!(MaxMessageSize, usize, 10 * 1024 * 1024);
 
 /// A node type.
 #[derive(Debug, Copy, Clone)]
@@ -102,4 +103,6 @@ pub struct P2pConfig {
     pub max_request_blocks_count: RequestedBlocksLimit,
     /// User agent value of this node (sent to peers over the network).
     pub user_agent: UserAgent,
+    /// A maximum size of a p2p message in bytes.
+    pub max_message_size: MaxMessageSize,
 }
diff --git a/p2p/src/constants.rs b/p2p/src/constants.rs
deleted file mode 100644
index 561b7c1ec..000000000
--- a/p2p/src/constants.rs
+++ /dev/null
@@ -1,21 +0,0 @@
-// Copyright (c) 2022 RBB S.r.l
-// opensource@mintlayer.org
-// SPDX-License-Identifier: MIT
-// Licensed under the MIT License;
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-// https://github.com/mintlayer/mintlayer-core/blob/master/LICENSE
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-//! P2P constants
-//!
-//! See protocol specification for more details
-
-/// Maximum message size
-pub const MAX_MESSAGE_SIZE: usize = 10 * 1024 * 1024;
diff --git a/p2p/src/error.rs b/p2p/src/error.rs
index 9dcddce45..b290984f7 100644
--- a/p2p/src/error.rs
+++ b/p2p/src/error.rs
@@ -71,13 +71,6 @@ pub enum PeerError {
     Pending(String),
 }
 
-/// PubSub errors for announcements
-#[derive(Error, Debug, PartialEq, Eq)]
-pub enum PublishError {
-    #[error("Message is too large. Tried to send {0:?} bytes when limit is {1:?}")]
-    MessageTooLarge(usize, usize),
-}
-
 /// Errors related to establishing a connection with a remote peer
 #[derive(Error, Debug, PartialEq, Eq)]
 pub enum DialError {
@@ -106,8 +99,6 @@ pub enum ConversionError {
 pub enum P2pError {
     #[error("Protocol violation: `{0}`")]
     ProtocolError(ProtocolError),
-    #[error("Failed to publish message: `{0}`")]
-    PublishError(PublishError),
     #[error("Failed to dial peer: `{0}`")]
     DialError(DialError),
     #[error("Connection to other task lost")]
@@ -178,7 +169,6 @@ impl BanScore for P2pError {
     fn ban_score(&self) -> u32 {
         match self {
             P2pError::ProtocolError(err) => err.ban_score(),
-            P2pError::PublishError(err) => err.ban_score(),
             P2pError::DialError(_) => 0,
             P2pError::ChannelClosed => 0,
             P2pError::PeerError(_) => 0,
@@ -216,14 +206,6 @@ impl BanScore for ProtocolError {
     }
 }
 
-impl BanScore for PublishError {
-    fn ban_score(&self) -> u32 {
-        match self {
-            PublishError::MessageTooLarge(_, _) => 100,
-        }
-    }
-}
-
 impl BanScore for ConversionError {
     fn ban_score(&self) -> u32 {
         match self {
diff --git a/p2p/src/lib.rs b/p2p/src/lib.rs
index 7769ac435..1b3ff045e 100644
--- a/p2p/src/lib.rs
+++ b/p2p/src/lib.rs
@@ -14,7 +14,6 @@
 // limitations under the License.
 
 pub mod config;
-pub mod constants;
 pub mod error;
 pub mod event;
 pub mod interface;
diff --git a/p2p/src/net/default_backend/backend.rs b/p2p/src/net/default_backend/backend.rs
index 68322b137..cf6a7d792 100644
--- a/p2p/src/net/default_backend/backend.rs
+++ b/p2p/src/net/default_backend/backend.rs
@@ -25,16 +25,15 @@ use tokio::{sync::mpsc, time::timeout};
 use common::chain::ChainConfig;
 use crypto::random::{make_pseudo_rng, Rng, SliceRandom};
 use logging::log;
-use serialization::{DecodeAll, Encode};
+use serialization::DecodeAll;
 use utils::set_flag::SetFlag;
 
 use crate::{
     config::P2pConfig,
-    error::{DialError, P2pError, PeerError, PublishError},
+    error::{DialError, P2pError, PeerError},
     message::{PeerManagerMessage, SyncMessage},
     net::{
         default_backend::{
-            constants::ANNOUNCEMENT_MAX_SIZE,
             peer,
             transport::{TransportListener, TransportSocket},
             types::{Command, Event, Message, PeerEvent},
@@ -242,19 +241,6 @@ where
         peer_id: PeerId,
         announcement: Announcement,
     ) -> crate::Result<()> {
-        let size = announcement.encode().len();
-        if size > ANNOUNCEMENT_MAX_SIZE {
-            self.conn_tx
-                .send(ConnectivityEvent::Misbehaved {
-                    peer_id,
-                    error: P2pError::PublishError(PublishError::MessageTooLarge(
-                        size,
-                        ANNOUNCEMENT_MAX_SIZE,
-                    )),
-                })
-                .map_err(P2pError::from)?;
-        }
-
         Self::send_sync_event(
             &self.sync_tx,
             SyncingEvent::Announcement {
diff --git a/p2p/src/net/default_backend/constants.rs b/p2p/src/net/default_backend/constants.rs
deleted file mode 100644
index e56e90f6f..000000000
--- a/p2p/src/net/default_backend/constants.rs
+++ /dev/null
@@ -1,18 +0,0 @@
-// Copyright (c) 2021-2022 RBB S.r.l
-// opensource@mintlayer.org
-// SPDX-License-Identifier: MIT
-// Licensed under the MIT License;
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-// https://github.com/mintlayer/mintlayer-core/blob/master/LICENSE
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-// TODO: Move constants to the config.
-
-pub const ANNOUNCEMENT_MAX_SIZE: usize = 2 * 1024 * 1024;
diff --git a/p2p/src/net/default_backend/mod.rs b/p2p/src/net/default_backend/mod.rs
index d1cfe3f25..ed8685241 100644
--- a/p2p/src/net/default_backend/mod.rs
+++ b/p2p/src/net/default_backend/mod.rs
@@ -14,7 +14,6 @@
 // limitations under the License.
 
 pub mod backend;
-pub mod constants;
 pub mod peer;
 pub mod transport;
 pub mod types;
@@ -28,18 +27,15 @@ use logging::log;
 use serialization::Encode;
 
 use crate::{
-    config,
-    error::{P2pError, PublishError},
+    error::P2pError,
     message::{Announcement, PeerManagerMessage, SyncMessage},
     net::{
-        default_backend::{
-            constants::ANNOUNCEMENT_MAX_SIZE,
-            transport::{TransportListener, TransportSocket},
-        },
+        default_backend::transport::{TransportListener, TransportSocket},
         types::{ConnectivityEvent, SyncingEvent},
         ConnectivityService, MessagingService, NetworkingService, SyncingEventReceiver,
     },
     types::peer_id::PeerId,
+    P2pConfig,
 };
 
 use super::types::services::Service;
@@ -79,12 +75,26 @@ impl<S: NetworkingService, T: TransportSocket> ConnectivityHandle<S, T> {
 #[derive(Debug)]
 pub struct MessagingHandle<T: TransportSocket> {
     command_sender: mpsc::UnboundedSender<types::Command<T::Address>>,
+    p2p_config: Arc<P2pConfig>,
+}
+
+impl<T: TransportSocket> MessagingHandle<T> {
+    pub fn new(
+        command_sender: mpsc::UnboundedSender<types::Command<T::Address>>,
+        p2p_config: Arc<P2pConfig>,
+    ) -> Self {
+        Self {
+            command_sender,
+            p2p_config,
+        }
+    }
 }
 
 impl<T: TransportSocket> Clone for MessagingHandle<T> {
     fn clone(&self) -> Self {
         Self {
             command_sender: self.command_sender.clone(),
+            p2p_config: self.p2p_config.clone(),
         }
     }
 }
@@ -107,7 +117,7 @@ impl<T: TransportSocket> NetworkingService for DefaultNetworkingService<T> {
         transport: Self::Transport,
         bind_addresses: Vec<Self::Address>,
         chain_config: Arc<common::chain::ChainConfig>,
-        p2p_config: Arc<config::P2pConfig>,
+        p2p_config: Arc<P2pConfig>,
     ) -> crate::Result<(
         Self::ConnectivityHandle,
         Self::MessagingHandle,
@@ -119,12 +129,13 @@ impl<T: TransportSocket> NetworkingService for DefaultNetworkingService<T> {
         let socket = transport.bind(bind_addresses).await?;
         let local_addresses = socket.local_addresses().expect("to have bind address available");
 
+        let p2p_config_ = Arc::clone(&p2p_config);
         tokio::spawn(async move {
             let mut backend = backend::Backend::<T>::new(
                 transport,
                 socket,
                 chain_config,
-                p2p_config,
+                p2p_config_,
                 cmd_rx,
                 conn_tx,
                 sync_tx,
@@ -138,9 +149,7 @@ impl<T: TransportSocket> NetworkingService for DefaultNetworkingService<T> {
 
         Ok((
             ConnectivityHandle::new(local_addresses, cmd_tx.clone(), conn_rx),
-            Self::MessagingHandle {
-                command_sender: cmd_tx,
-            },
+            MessagingHandle::new(cmd_tx, p2p_config),
             Self::SyncingEventReceiver { sync_rx },
         ))
     }
@@ -202,19 +211,13 @@ impl<T: TransportSocket> MessagingService for MessagingHandle<T> {
     }
 
     fn make_announcement(&mut self, announcement: Announcement) -> crate::Result<()> {
-        let message = announcement.encode();
-        if message.len() > ANNOUNCEMENT_MAX_SIZE {
-            return Err(P2pError::PublishError(PublishError::MessageTooLarge(
-                message.len(),
-                ANNOUNCEMENT_MAX_SIZE,
-            )));
-        }
-
         let service = match &announcement {
             Announcement::Block(_) => Service::Blocks,
             Announcement::Transaction(_) => Service::Transactions,
         };
 
+        let message = announcement.encode();
+        assert!(message.len() < *self.p2p_config.max_message_size);
         self.command_sender
             .send(types::Command::AnnounceData { service, message })
             .map_err(P2pError::from)
diff --git a/p2p/src/net/default_backend/peer.rs b/p2p/src/net/default_backend/peer.rs
index bc233cf1c..f4475b758 100644
--- a/p2p/src/net/default_backend/peer.rs
+++ b/p2p/src/net/default_backend/peer.rs
@@ -93,7 +93,7 @@ where
         tx: mpsc::UnboundedSender<(PeerId, PeerEvent)>,
         rx: mpsc::UnboundedReceiver<Event>,
     ) -> Self {
-        let socket = BufferedTranscoder::new(socket);
+        let socket = BufferedTranscoder::new(socket, *p2p_config.max_message_size);
 
         Self {
             peer_id,
@@ -293,7 +293,7 @@ mod tests {
             peer
         });
 
-        let mut socket2 = BufferedTranscoder::new(socket2);
+        let mut socket2 = BufferedTranscoder::new(socket2, *p2p_config.max_message_size);
         assert!(socket2.recv().now_or_never().is_none());
         assert!(socket2
             .send(types::Message::Handshake(types::HandshakeMessage::Hello {
@@ -364,7 +364,7 @@ mod tests {
             peer
         });
 
-        let mut socket2 = BufferedTranscoder::new(socket2);
+        let mut socket2 = BufferedTranscoder::new(socket2, *p2p_config.max_message_size);
         socket2.recv().await.unwrap();
         assert!(socket2
             .send(types::Message::Handshake(
@@ -436,7 +436,7 @@ mod tests {
 
         let handle = tokio::spawn(async move { peer.handshake().await });
 
-        let mut socket2 = BufferedTranscoder::new(socket2);
+        let mut socket2 = BufferedTranscoder::new(socket2, *p2p_config.max_message_size);
         assert!(socket2.recv().now_or_never().is_none());
         assert!(socket2
             .send(types::Message::Handshake(types::HandshakeMessage::Hello {
@@ -484,7 +484,7 @@ mod tests {
             peer_id2,
             PeerRole::Inbound,
             chain_config,
-            p2p_config,
+            Arc::clone(&p2p_config),
             socket1,
             None,
             tx1,
@@ -493,7 +493,7 @@ mod tests {
 
         let handle = tokio::spawn(async move { peer.handshake().await });
 
-        let mut socket2 = BufferedTranscoder::new(socket2);
+        let mut socket2 = BufferedTranscoder::new(socket2, *p2p_config.max_message_size);
         assert!(socket2.recv().now_or_never().is_none());
         socket2
             .send(types::Message::HeaderListRequest(
diff --git a/p2p/src/net/default_backend/transport/buffered_transcoder.rs b/p2p/src/net/default_backend/transport/buffered_transcoder.rs
new file mode 100644
index 000000000..3a41cc406
--- /dev/null
+++ b/p2p/src/net/default_backend/transport/buffered_transcoder.rs
@@ -0,0 +1,71 @@
+// Copyright (c) 2022 RBB S.r.l
+// opensource@mintlayer.org
+// SPDX-License-Identifier: MIT
+// Licensed under the MIT License;
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// https://github.com/mintlayer/mintlayer-core/blob/master/LICENSE
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::io;
+
+use bytes::BytesMut;
+use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
+use tokio_util::codec::{Decoder, Encoder};
+
+use crate::{
+    net::default_backend::{transport::message_codec::EncoderDecoder, types::Message},
+    Result,
+};
+
+pub struct BufferedTranscoder<S> {
+    stream: S,
+    buffer: BytesMut,
+    encoder_decoder: EncoderDecoder,
+}
+
+impl<S: AsyncWrite + AsyncRead + Unpin> BufferedTranscoder<S> {
+    pub fn new(stream: S, max_message_size: usize) -> BufferedTranscoder<S> {
+        let encoder_decoder = EncoderDecoder::new(max_message_size);
+        BufferedTranscoder {
+            stream,
+            buffer: BytesMut::new(),
+            encoder_decoder,
+        }
+    }
+
+    pub async fn send(&mut self, msg: Message) -> Result<()> {
+        let mut buf = BytesMut::new();
+        self.encoder_decoder.encode(msg, &mut buf)?;
+        self.stream.write_all(&buf).await?;
+        self.stream.flush().await?;
+        Ok(())
+    }
+
+    /// Read a framed message from socket
+    ///
+    /// First try to decode whatever may be in the stream's buffer and if it's empty
+    /// or the frame hasn't been completely received, wait on the socket until the buffer
+    /// has all data. If the buffer has a full frame that can be decoded, return that without
+    /// calling the socket first.
+    pub async fn recv(&mut self) -> Result<Message> {
+        loop {
+            match self.encoder_decoder.decode(&mut self.buffer) {
+                Ok(None) => {
+                    if self.stream.read_buf(&mut self.buffer).await? == 0 {
+                        return Err(io::Error::from(io::ErrorKind::UnexpectedEof).into());
+                    }
+                    continue;
+                }
+                Ok(Some(msg)) => return Ok(msg),
+                Err(e) => return Err(e),
+            }
+        }
+    }
+}
diff --git a/p2p/src/net/default_backend/transport/impls/channel.rs b/p2p/src/net/default_backend/transport/impls/channel.rs
index 8dd0bccc4..e10c43920 100644
--- a/p2p/src/net/default_backend/transport/impls/channel.rs
+++ b/p2p/src/net/default_backend/transport/impls/channel.rs
@@ -194,14 +194,22 @@ impl PeerStream for ChannelStream {}
 mod tests {
     use std::net::SocketAddrV4;
 
+    use crypto::random::Rng;
+    use test_utils::random::Seed;
+
     use super::*;
     use crate::{
         message::BlockListRequest,
         net::default_backend::{transport::BufferedTranscoder, types::Message},
     };
 
+    #[rstest::rstest]
+    #[trace]
+    #[case(Seed::from_entropy())]
     #[tokio::test]
-    async fn send_recv() {
+    async fn send_recv(#[case] seed: Seed) {
+        let mut rng = test_utils::random::make_seedable_rng(seed);
+
         let transport = MpscChannelTransport::new();
         let address: SocketAddr = SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, 0).into();
         let mut server = transport.bind(vec![address]).await.unwrap();
@@ -212,10 +220,10 @@ mod tests {
         let peer_stream = peer_res.unwrap();
 
         let message = Message::BlockListRequest(BlockListRequest::new(vec![]));
-        let mut peer_stream = BufferedTranscoder::new(peer_stream);
+        let mut peer_stream = BufferedTranscoder::new(peer_stream, rng.gen_range(128..1024));
         peer_stream.send(message.clone()).await.unwrap();
 
-        let mut server_stream = BufferedTranscoder::new(server_stream);
+        let mut server_stream = BufferedTranscoder::new(server_stream, rng.gen_range(128..1024));
         assert_eq!(server_stream.recv().await.unwrap(), message);
     }
 }
diff --git a/p2p/src/net/default_backend/transport/impls/tcp.rs b/p2p/src/net/default_backend/transport/impls/tcp.rs
index ec4a4cf97..72613bd03 100644
--- a/p2p/src/net/default_backend/transport/impls/tcp.rs
+++ b/p2p/src/net/default_backend/transport/impls/tcp.rs
@@ -171,6 +171,8 @@ mod tests {
         chain::block::Block,
         primitives::{Id, H256},
     };
+    use crypto::random::Rng;
+    use test_utils::random::Seed;
 
     use crate::{
         message::BlockListRequest,
@@ -180,8 +182,13 @@ mod tests {
     use super::*;
     use crate::net::default_backend::{transport::BufferedTranscoder, types::Message};
 
+    #[rstest::rstest]
+    #[trace]
+    #[case(Seed::from_entropy())]
     #[tokio::test]
-    async fn send_recv() {
+    async fn send_recv(#[case] seed: Seed) {
+        let mut rng = test_utils::random::make_seedable_rng(seed);
+
         let transport = TcpTransportSocket::new();
         let mut server = transport.bind(vec![TestTransportTcp::make_address()]).await.unwrap();
         let peer_fut = transport.connect(server.local_addresses().unwrap()[0]);
@@ -191,15 +198,20 @@ mod tests {
         let peer_stream = peer_res.unwrap();
 
         let message = Message::BlockListRequest(BlockListRequest::new(vec![]));
-        let mut peer_stream = BufferedTranscoder::new(peer_stream);
+        let mut peer_stream = BufferedTranscoder::new(peer_stream, rng.gen_range(128..1024));
         peer_stream.send(message.clone()).await.unwrap();
 
-        let mut server_stream = BufferedTranscoder::new(server_stream);
+        let mut server_stream = BufferedTranscoder::new(server_stream, rng.gen_range(128..1024));
         assert_eq!(server_stream.recv().await.unwrap(), message);
     }
 
+    #[rstest::rstest]
+    #[trace]
+    #[case(Seed::from_entropy())]
     #[tokio::test]
-    async fn send_2_reqs() {
+    async fn send_2_reqs(#[case] seed: Seed) {
+        let mut rng = test_utils::random::make_seedable_rng(seed);
+
         let transport = TcpTransportSocket::new();
         let mut server = transport.bind(vec![TestTransportTcp::make_address()]).await.unwrap();
         let peer_fut = transport.connect(server.local_addresses().unwrap()[0]);
@@ -209,16 +221,14 @@ mod tests {
         let peer_stream = peer_res.unwrap();
 
         let message_1 = Message::BlockListRequest(BlockListRequest::new(vec![]));
-        let mut rng =
-            test_utils::random::make_seedable_rng(test_utils::random::Seed::from_entropy());
         let id: Id<Block> = H256::random_using(&mut rng).into();
         let message_2 = Message::BlockListRequest(BlockListRequest::new(vec![id]));
 
-        let mut peer_stream = BufferedTranscoder::new(peer_stream);
+        let mut peer_stream = BufferedTranscoder::new(peer_stream, rng.gen_range(512..2048));
         peer_stream.send(message_1.clone()).await.unwrap();
         peer_stream.send(message_2.clone()).await.unwrap();
 
-        let mut server_stream = BufferedTranscoder::new(server_stream);
+        let mut server_stream = BufferedTranscoder::new(server_stream, rng.gen_range(512..2048));
         assert_eq!(server_stream.recv().await.unwrap(), message_1);
         assert_eq!(server_stream.recv().await.unwrap(), message_2);
     }
diff --git a/p2p/src/net/default_backend/transport/message_codec.rs b/p2p/src/net/default_backend/transport/message_codec.rs
index e658f9b24..dc770f577 100644
--- a/p2p/src/net/default_backend/transport/message_codec.rs
+++ b/p2p/src/net/default_backend/transport/message_codec.rs
@@ -17,12 +17,21 @@ use std::io;
 
 use bytes::{Buf, BytesMut};
 use serialization::{DecodeAll, Encode};
-use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
 use tokio_util::codec::{Decoder, Encoder};
 
-use crate::{constants::MAX_MESSAGE_SIZE, net::default_backend::types::Message, P2pError, Result};
+use crate::{net::default_backend::types::Message, P2pError, Result};
 
-struct EncoderDecoder {}
+const HEADER_LEN: usize = 4;
+
+pub struct EncoderDecoder {
+    max_message_size: usize,
+}
+
+impl EncoderDecoder {
+    pub fn new(max_message_size: usize) -> Self {
+        Self { max_message_size }
+    }
+}
 
 impl Decoder for EncoderDecoder {
     type Item = Message;
@@ -33,12 +42,12 @@ impl Decoder for EncoderDecoder {
             return Ok(None);
         }
 
-        let (header, remaining_bytes) = src.split_at_mut(4);
+        let (header, remaining_bytes) = src.split_at_mut(HEADER_LEN);
 
         // Unwrap is safe here because the header size is 4 bytes
         let length = u32::from_le_bytes(header.try_into().expect("valid size")) as usize;
 
-        if length > MAX_MESSAGE_SIZE {
+        if length > self.max_message_size {
             return Err(io::Error::new(
                 io::ErrorKind::InvalidData,
                 format!("Frame of length {length} is too large"),
@@ -70,7 +79,7 @@ impl Encoder<Message> for EncoderDecoder {
     fn encode(&mut self, msg: Message, dst: &mut BytesMut) -> Result<()> {
         let encoded = msg.encode();
 
-        if encoded.len() > MAX_MESSAGE_SIZE {
+        if encoded.len() > self.max_message_size {
             return Err(io::Error::new(
                 io::ErrorKind::InvalidData,
                 format!("Frame of length {} is too large", encoded.len()),
@@ -88,45 +97,105 @@ impl Encoder<Message> for EncoderDecoder {
     }
 }
 
-pub struct BufferedTranscoder<S> {
-    stream: S,
-    buffer: BytesMut,
-}
+#[cfg(test)]
+mod tests {
+    use std::{
+        io::ErrorKind,
+        net::{IpAddr, Ipv4Addr, SocketAddr},
+    };
+
+    use crypto::random::Rng;
+    use test_utils::random::Seed;
+
+    use super::*;
+    use crate::{
+        error::DialError,
+        message::{AddrListRequest, AnnounceAddrRequest, HeaderListResponse, PingRequest},
+    };
+
+    #[rstest::rstest]
+    #[trace]
+    #[case(Seed::from_entropy())]
+    fn size_limit_encode(#[case] seed: Seed) {
+        let mut rng = test_utils::random::make_seedable_rng(seed);
+
+        let message = Message::AnnounceAddrRequest(AnnounceAddrRequest {
+            address: SocketAddr::new(
+                IpAddr::V4(Ipv4Addr::new(rng.gen(), rng.gen(), rng.gen(), rng.gen())),
+                rng.gen(),
+            )
+            .into(),
+        });
 
-impl<S: AsyncWrite + AsyncRead + Unpin> BufferedTranscoder<S> {
-    pub fn new(stream: S) -> BufferedTranscoder<S> {
-        BufferedTranscoder {
-            stream,
-            buffer: BytesMut::new(),
-        }
+        let mut buf = BytesMut::new();
+        // Encode to determine the serialized message length.
+        EncoderDecoder::new(rng.gen_range(64..128))
+            .encode(message.clone(), &mut buf)
+            .unwrap();
+        assert!(buf.len() > HEADER_LEN);
+        let message_length = buf.len() - HEADER_LEN;
+
+        let mut encoder = EncoderDecoder::new(rng.gen_range(0..message_length));
+        assert_eq!(
+            Err(P2pError::DialError(DialError::IoError(
+                ErrorKind::InvalidData
+            ))),
+            encoder.encode(message, &mut buf)
+        );
     }
 
-    pub async fn send(&mut self, msg: Message) -> Result<()> {
-        let mut buf = BytesMut::new();
-        EncoderDecoder {}.encode(msg, &mut buf)?;
-        self.stream.write_all(&buf).await?;
-        self.stream.flush().await?;
-        Ok(())
+    #[rstest::rstest]
+    #[trace]
+    #[case(Seed::from_entropy())]
+    fn size_limit_decode(#[case] seed: Seed) {
+        let mut rng = test_utils::random::make_seedable_rng(seed);
+
+        let message = Message::AnnounceAddrRequest(AnnounceAddrRequest {
+            address: SocketAddr::new(
+                IpAddr::V4(Ipv4Addr::new(rng.gen(), rng.gen(), rng.gen(), rng.gen())),
+                rng.gen(),
+            )
+            .into(),
+        });
+        let mut encoded = BytesMut::new();
+        EncoderDecoder::new(rng.gen_range(126..512))
+            .encode(message, &mut encoded)
+            .unwrap();
+
+        let mut decoder = EncoderDecoder::new(rng.gen_range(0..(encoded.len() - HEADER_LEN)));
+        assert_eq!(
+            Err(P2pError::DialError(DialError::IoError(
+                ErrorKind::InvalidData
+            ))),
+            decoder.decode(&mut encoded)
+        );
     }
 
-    /// Read a framed message from socket
-    ///
-    /// First try to decode whatever may be in the stream's buffer and if it's empty
-    /// or the frame hasn't been completely received, wait on the socket until the buffer
-    /// has all data. If the buffer has a full frame that can be decoded, return that without
-    /// calling the socket first.
-    pub async fn recv(&mut self) -> Result<Message> {
-        loop {
-            match (EncoderDecoder {}.decode(&mut self.buffer)) {
-                Ok(None) => {
-                    if self.stream.read_buf(&mut self.buffer).await? == 0 {
-                        return Err(io::Error::from(io::ErrorKind::UnexpectedEof).into());
-                    }
-                    continue;
-                }
-                Ok(Some(msg)) => return Ok(msg),
-                Err(e) => return Err(e),
-            }
+    #[rstest::rstest]
+    #[trace]
+    #[case(Seed::from_entropy())]
+    fn roundtrip(#[case] seed: Seed) {
+        let mut rng = test_utils::random::make_seedable_rng(seed);
+
+        let messages = [
+            Message::PingRequest(PingRequest { nonce: 1 }),
+            Message::HeaderListResponse(HeaderListResponse::new(Vec::new())),
+            Message::AnnounceAddrRequest(AnnounceAddrRequest {
+                address: SocketAddr::new(
+                    IpAddr::V4(Ipv4Addr::new(rng.gen(), rng.gen(), rng.gen(), rng.gen())),
+                    rng.gen(),
+                )
+                .into(),
+            }),
+            Message::AddrListRequest(AddrListRequest {}),
+        ];
+
+        let mut encoder = EncoderDecoder::new(rng.gen_range(128..2048));
+        for message in messages {
+            let mut buf = BytesMut::new();
+            encoder.encode(message.clone(), &mut buf).unwrap();
+            let decoded = encoder.decode(&mut buf).unwrap().unwrap();
+            assert_eq!(message, decoded);
         }
     }
 }
diff --git a/p2p/src/net/default_backend/transport/mod.rs b/p2p/src/net/default_backend/transport/mod.rs
index 39f3840ed..d4a72cc91 100644
--- a/p2p/src/net/default_backend/transport/mod.rs
+++ b/p2p/src/net/default_backend/transport/mod.rs
@@ -13,6 +13,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+mod buffered_transcoder;
 mod impls;
 mod message_codec;
 mod traits;
@@ -20,8 +21,8 @@ mod traits;
 use impls::{channel, socks5, stream_adapter, tcp};
 
 pub use self::{
+    buffered_transcoder::BufferedTranscoder,
     channel::{ChannelListener, ChannelStream, MpscChannelTransport},
-    message_codec::BufferedTranscoder,
     socks5::Socks5TransportSocket,
     stream_adapter::{
         identity::IdentityStreamAdapter, noise::NoiseEncryptionAdapter,
diff --git a/p2p/src/sync/peer.rs b/p2p/src/sync/peer.rs
index 13d516248..032d2df8e 100644
--- a/p2p/src/sync/peer.rs
+++ b/p2p/src/sync/peer.rs
@@ -474,7 +474,6 @@ where
             | P2pError::ConversionError(_)
             | P2pError::PeerError(_)
             | P2pError::NoiseHandshakeError(_)
-            | P2pError::PublishError(_)
             | P2pError::InvalidConfigurationValue(_)
             | P2pError::ChainstateError(_)) => Err(e),
             // Fatal errors, simply propagate them to stop the sync manager.


Test patch:
diff --git a/p2p/backend-test-suite/src/block_announcement.rs b/p2p/backend-test-suite/src/block_announcement.rs
index c6c73ca4a..aec10c871 100644
--- a/p2p/backend-test-suite/src/block_announcement.rs
+++ b/p2p/backend-test-suite/src/block_announcement.rs
@@ -16,38 +16,21 @@
 use std::{fmt::Debug, sync::Arc};
 
 use common::{
-    chain::{
-        block::{
-            consensus_data::{ConsensusData, PoSData},
-            timestamp::BlockTimestamp,
-            Block, BlockReward,
-        },
-        signature::{
-            inputsig::{standard_signature::StandardInputSignature, InputWitness},
-            sighashtype,
-        },
-    },
-    primitives::{user_agent::mintlayer_core_user_agent, Compact, Id, H256},
+    chain::block::{consensus_data::ConsensusData, timestamp::BlockTimestamp, Block, BlockReward},
+    primitives::{user_agent::mintlayer_core_user_agent, Id, H256},
 };
-use crypto::vrf::{transcript::TranscriptAssembler, VRFKeyKind, VRFPrivateKey};
-use serialization::Encode;
 
 use p2p::{
     config::{NodeType, P2pConfig},
-    error::{P2pError, PublishError},
     message::Announcement,
     net::{
-        default_backend::constants::ANNOUNCEMENT_MAX_SIZE, types::SyncingEvent,
-        ConnectivityService, MessagingService, NetworkingService, SyncingEventReceiver,
+        types::SyncingEvent, ConnectivityService, MessagingService, NetworkingService,
+        SyncingEventReceiver,
     },
     testing_utils::{connect_and_accept_services, test_p2p_config, TestTransportMaker},
 };
 
-tests![
-    block_announcement,
-    block_announcement_no_subscription,
-    block_announcement_too_big_message,
-];
+tests![block_announcement, block_announcement_no_subscription,];
 
 async fn block_announcement<T, N, A>()
 where
@@ -167,6 +150,7 @@ where
         msg_max_locator_count: Default::default(),
         max_request_blocks_count: Default::default(),
         user_agent: mintlayer_core_user_agent(),
+        max_message_size: Default::default(),
     });
     let (mut conn1, mut messaging_handle1, _sync1) = N::start(
         T::make_transport(),
@@ -199,70 +183,3 @@ where
         .make_announcement(Announcement::Block(Box::new(block.header().clone())))
         .unwrap();
 }
-
-async fn block_announcement_too_big_message<T, N, A>()
-where
-    T: TestTransportMaker<Transport = N::Transport, Address = N::Address>,
-    N: NetworkingService + Debug,
-    N::MessagingHandle: MessagingService,
-    N::SyncingEventReceiver: SyncingEventReceiver,
-    N::ConnectivityHandle: ConnectivityService<N>,
-{
-    // TODO: Use seedable random.
-    let mut rng = crypto::random::make_true_rng();
-
-    let config = Arc::new(common::chain::config::create_mainnet());
-    let p2p_config = Arc::new(test_p2p_config());
-    let (mut conn1, mut messaging_handle1, _sync1) = N::start(
-        T::make_transport(),
-        vec![T::make_address()],
-        Arc::clone(&config),
-        Arc::clone(&p2p_config),
-    )
-    .await
-    .unwrap();
-
-    let (mut conn2, _messaging_handle2, _sync2) = N::start(
-        T::make_transport(),
-        vec![T::make_address()],
-        Arc::clone(&config),
-        Arc::clone(&p2p_config),
-    )
-    .await
-    .unwrap();
-
-    connect_and_accept_services::<N>(&mut conn1, &mut conn2).await;
-
-    let signature = (0..ANNOUNCEMENT_MAX_SIZE).map(|_| 0).collect::<Vec<u8>>();
-    let signatures = vec![InputWitness::Standard(StandardInputSignature::new(
-        sighashtype::SigHashType::try_from(sighashtype::SigHashType::ALL).unwrap(),
-        signature,
-    ))];
-    let (sk, _pk) = VRFPrivateKey::new_from_rng(&mut rng, VRFKeyKind::Schnorrkel);
-    let vrf_data = sk.produce_vrf_data(TranscriptAssembler::new(b"abc").finalize().into());
-    let pos = PoSData::new(
-        Vec::new(),
-        signatures,
-        H256::random_using(&mut rng).into(),
-        vrf_data,
-        Compact(0),
-    );
-    let block = Block::new(
-        Vec::new(),
-        Id::new(H256([0x04; 32])),
-        BlockTimestamp::from_int_seconds(1337u64),
-        ConsensusData::PoS(Box::new(pos)),
-        BlockReward::new(Vec::new()),
-    )
-    .unwrap();
-    let message = Announcement::Block(Box::new(block.header().clone()));
-    let encoded_size = message.encode().len();
-
-    assert_eq!(
-        messaging_handle1.make_announcement(message),
-        Err(P2pError::PublishError(PublishError::MessageTooLarge(
-            encoded_size,
-            ANNOUNCEMENT_MAX_SIZE
-        )))
-    );
-}
diff --git a/p2p/src/net/default_backend/transport/impls/stream_adapter/wrapped_transport/tests.rs b/p2p/src/net/default_backend/transport/impls/stream_adapter/wrapped_transport/tests.rs
index 7270bbf7c..f4343cfe8 100644
--- a/p2p/src/net/default_backend/transport/impls/stream_adapter/wrapped_transport/tests.rs
+++ b/p2p/src/net/default_backend/transport/impls/stream_adapter/wrapped_transport/tests.rs
@@ -26,13 +26,12 @@ use tokio::{
     time::timeout,
 };
 
-use crate::testing_utils::{
-    P2pTokioTestTimeGetter, TestTransportChannel, TestTransportMaker, TestTransportTcp,
-};
 use common::{
     chain::Block,
     primitives::{Id, H256},
 };
+use crypto::random::Rng;
+use test_utils::random::Seed;
 
 use crate::{
     message::BlockListRequest,
@@ -45,6 +44,9 @@ use crate::{
         },
         types::Message,
     },
+    testing_utils::{
+        P2pTokioTestTimeGetter, TestTransportChannel, TestTransportMaker, TestTransportTcp,
+    },
 };
 
 use super::wrapped_socket::WrappedTransportSocket;
@@ -204,8 +206,13 @@ async fn test_bind_port_closed() {
     assert!(!*transport.base_transport.port_open.lock().unwrap());
 }
 
+#[rstest::rstest]
+#[trace]
+#[case(Seed::from_entropy())]
 #[tokio::test]
-async fn send_2_reqs() {
+async fn send_2_reqs(#[case] seed: Seed) {
+    let mut rng = test_utils::random::make_seedable_rng(seed);
+
     let transport = WrappedTransportSocket::<NoiseEncryptionAdapter, TcpTransportSocket>::new(
         NoiseEncryptionAdapter::gen_new(),
         TcpTransportSocket::new(),
@@ -218,15 +225,14 @@ async fn send_2_reqs() {
     let peer_stream = peer_res.unwrap();
 
     let message_1 = Message::BlockListRequest(BlockListRequest::new(vec![]));
-    let mut rng = test_utils::random::make_seedable_rng(test_utils::random::Seed::from_entropy());
     let id: Id<Block> = H256::random_using(&mut rng).into();
     let message_2 = Message::BlockListRequest(BlockListRequest::new(vec![id]));
-    let mut peer_stream = BufferedTranscoder::new(peer_stream);
+    let mut peer_stream = BufferedTranscoder::new(peer_stream, rng.gen_range(512..1024));
     peer_stream.send(message_1.clone()).await.unwrap();
 
     peer_stream.send(message_2.clone()).await.unwrap();
 
-    let mut server_stream = BufferedTranscoder::new(server_stream);
+    let mut server_stream = BufferedTranscoder::new(server_stream, rng.gen_range(512..1024));
     assert_eq!(server_stream.recv().await.unwrap(), message_1);
     assert_eq!(server_stream.recv().await.unwrap(), message_2);
 }
diff --git a/p2p/src/peer_manager/peerdb/tests.rs b/p2p/src/peer_manager/peerdb/tests.rs
index 989aa4db0..963472e91 100644
--- a/p2p/src/peer_manager/peerdb/tests.rs
+++ b/p2p/src/peer_manager/peerdb/tests.rs
@@ -53,6 +53,7 @@ fn unban_peer() {
             msg_max_locator_count: Default::default(),
             max_request_blocks_count: Default::default(),
             user_agent: mintlayer_core_user_agent(),
+            max_message_size: Default::default(),
         }),
         time_getter.get_time_getter(),
         db_store,
diff --git a/p2p/src/peer_manager/tests/connections.rs b/p2p/src/peer_manager/tests/connections.rs
index 69bce022e..539b62240 100644
--- a/p2p/src/peer_manager/tests/connections.rs
+++ b/p2p/src/peer_manager/tests/connections.rs
@@ -676,6 +676,7 @@ where
         msg_max_locator_count: Default::default(),
         max_request_blocks_count: Default::default(),
         user_agent: mintlayer_core_user_agent(),
+        max_message_size: Default::default(),
     });
     let tx1 = run_peer_manager::<T>(
         A::make_transport(),
@@ -711,6 +712,7 @@ where
         msg_max_locator_count: Default::default(),
         max_request_blocks_count: Default::default(),
         user_agent: mintlayer_core_user_agent(),
+        max_message_size: Default::default(),
     });
     let tx1 = run_peer_manager::<T>(
         A::make_transport(),
@@ -791,6 +793,7 @@ where
         msg_max_locator_count: Default::default(),
         max_request_blocks_count: Default::default(),
         user_agent: mintlayer_core_user_agent(),
+        max_message_size: Default::default(),
     });
     let tx1 = run_peer_manager::<T>(
         A::make_transport(),
@@ -827,6 +830,7 @@ where
         msg_max_locator_count: Default::default(),
         max_request_blocks_count: Default::default(),
         user_agent: mintlayer_core_user_agent(),
+        max_message_size: Default::default(),
     });
     let tx2 = run_peer_manager::<T>(
         A::make_transport(),
@@ -856,6 +860,7 @@ where
         msg_max_locator_count: Default::default(),
         max_request_blocks_count: Default::default(),
         user_agent: mintlayer_core_user_agent(),
+        max_message_size: Default::default(),
     });
     let tx3 = run_peer_manager::<T>(
         A::make_transport(),
diff --git a/p2p/src/peer_manager/tests/ping.rs b/p2p/src/peer_manager/tests/ping.rs
index d0d8db0c9..d6c6a61a3 100644
--- a/p2p/src/peer_manager/tests/ping.rs
+++ b/p2p/src/peer_manager/tests/ping.rs
@@ -58,6 +58,7 @@ async fn ping_timeout() {
         msg_max_locator_count: Default::default(),
         max_request_blocks_count: Default::default(),
         user_agent: mintlayer_core_user_agent(),
+        max_message_size: Default::default(),
     });
     let ping_check_period = *p2p_config.ping_check_period;
     let ping_timeout = *p2p_config.ping_timeout;
diff --git a/p2p/src/testing_utils.rs b/p2p/src/testing_utils.rs
index f6884eb11..efb6d0d95 100644
--- a/p2p/src/testing_utils.rs
+++ b/p2p/src/testing_utils.rs
@@ -326,5 +326,6 @@ pub fn test_p2p_config() -> P2pConfig {
         msg_max_locator_count: Default::default(),
         max_request_blocks_count: Default::default(),
         user_agent: mintlayer_core_user_agent(),
+        max_message_size: Default::default(),
     }
 }