diff --git a/crates/overlay/src/codec.rs b/crates/overlay/src/codec.rs index a6d14b95..ec498b0e 100644 --- a/crates/overlay/src/codec.rs +++ b/crates/overlay/src/codec.rs @@ -31,9 +31,9 @@ //! `MAX_UNAUTH_MESSAGE_SIZE = 0x1000`. use crate::{OverlayError, Result}; -use bytes::{Buf, BufMut, BytesMut}; +use bytes::{Buf, BytesMut}; use stellar_xdr::{AuthenticatedMessage, Limits, ReadXdr, WriteXdr}; -use tokio_util::codec::{Decoder, Encoder}; +use tokio_util::codec::Decoder; /// Maximum message size (16 MB) - prevents memory exhaustion. /// Spec: OVERLAY_SPEC §3.3 — MAX_MESSAGE_SIZE = 16,777,216 bytes. @@ -48,11 +48,10 @@ const MIN_MESSAGE_SIZE: usize = 12; /// Rejects an outbound XDR payload whose length exceeds `MAX_MESSAGE_SIZE`. /// -/// This is the single enforcement point for the encode-side size bound, shared -/// by the real send-path encoder (`MessageCodec::encode_message`) and the -/// `Encoder` trait impl. Keeping one implementation prevents the two paths from -/// silently diverging again (see #3774). Mirrors the receive-side rejection in -/// `Decoder::decode` and stellar-core's `TCPPeer.cpp:690-701`. +/// This is the single enforcement point for the encode-side size bound, called +/// from the real send-path encoder (`MessageCodec::encode_message`). Mirrors the +/// receive-side rejection in `Decoder::decode` and stellar-core's +/// `TCPPeer.cpp:690-701`. fn check_encode_size(xdr_len: usize) -> Result<()> { if xdr_len > MAX_MESSAGE_SIZE { return Err(OverlayError::Message(format!( @@ -97,8 +96,9 @@ impl MessageFrame { /// Codec for encoding and decoding Stellar overlay messages. /// -/// Implements tokio's `Encoder` and `Decoder` traits for use with framed -/// TCP streams. Handles the length-prefixed framing protocol automatically. +/// Implements tokio's `Decoder` trait for use with framed TCP streams, and +/// provides [`MessageCodec::encode_message`] for the send path. Handles the +/// length-prefixed framing protocol automatically. /// /// # Usage /// @@ -268,29 +268,6 @@ impl Decoder for MessageCodec { } } -impl Encoder for MessageCodec { - type Error = OverlayError; - - fn encode(&mut self, message: AuthenticatedMessage, dst: &mut BytesMut) -> Result<()> { - // Encode to XDR - let xdr_bytes = message.to_xdr(Limits::none())?; - - // Check size (shared with the real send-path encoder, `encode_message`). - check_encode_size(xdr_bytes.len())?; - - // Write XDR record-marking prefix. Bit 31 is the final-fragment bit, - // not an authentication marker. - let len = xdr_bytes.len() as u32; - dst.reserve(4 + xdr_bytes.len()); - dst.put_u32(len | 0x80000000); - - // Write message body - dst.extend_from_slice(&xdr_bytes); - - Ok(()) - } -} - /// Helper functions for working with Stellar messages. /// /// Provides utilities for message classification and display. @@ -508,7 +485,7 @@ mod tests { message: StellarMessage::Peers(VecM::default()), mac: HmacSha256Mac { mac: [42u8; 32] }, }); - codec.encode(auth_msg, &mut buf).unwrap(); + buf.extend_from_slice(&MessageCodec::encode_message(&auth_msg).unwrap()); let frame = codec.decode(&mut buf).unwrap().unwrap(); assert!( frame.is_last_fragment, @@ -521,7 +498,7 @@ mod tests { message: StellarMessage::Hello(Default::default()), mac: HmacSha256Mac { mac: [0u8; 32] }, }); - codec.encode(hello_msg, &mut buf).unwrap(); + buf.extend_from_slice(&MessageCodec::encode_message(&hello_msg).unwrap()); let frame = codec.decode(&mut buf).unwrap().unwrap(); assert!( frame.is_last_fragment, @@ -601,7 +578,7 @@ mod tests { let mut buf = BytesMut::new(); // Encode - codec.encode(msg, &mut buf).unwrap(); + buf.extend_from_slice(&MessageCodec::encode_message(&msg).unwrap()); // Decode let decoded = codec.decode(&mut buf).unwrap(); diff --git a/crates/overlay/src/connection.rs b/crates/overlay/src/connection.rs index 78921c8b..58b1d27e 100644 --- a/crates/overlay/src/connection.rs +++ b/crates/overlay/src/connection.rs @@ -19,7 +19,7 @@ use crate::{ codec::{MessageCodec, MessageFrame}, OverlayError, Result, }; -use futures::{SinkExt, StreamExt}; +use futures::StreamExt; use std::collections::HashSet; use std::net::{IpAddr, SocketAddr}; use std::pin::Pin; @@ -370,95 +370,6 @@ impl Connection { debug!("Closed connection to {}", self.remote_addr); } } - - /// Splits the connection into separate send and receive halves. - /// - /// This allows concurrent sending and receiving on the same connection. - pub fn split(self) -> (ConnectionSender, ConnectionReceiver) { - let (sink, stream) = self.framed.split(); - ( - ConnectionSender { - sink, - remote_addr: self.remote_addr, - }, - ConnectionReceiver { - stream, - remote_addr: self.remote_addr, - }, - ) - } -} - -/// Send half of a split connection. -/// -/// Created by [`Connection::split`]. Allows sending messages without -/// holding a lock on the full connection. -pub struct ConnectionSender { - sink: futures::stream::SplitSink, AuthenticatedMessage>, - remote_addr: SocketAddr, -} - -impl ConnectionSender { - /// Sends a message to the peer. - /// - /// Includes a timeout to prevent blocking indefinitely on TCP backpressure, - /// matching the timeout behavior of [`Connection::send`]. - pub async fn send(&mut self, message: AuthenticatedMessage) -> Result<()> { - trace!("Sending message to {}", self.remote_addr); - const SEND_TIMEOUT_SECS: u64 = 10; - match timeout( - Duration::from_secs(SEND_TIMEOUT_SECS), - self.sink.send(message), - ) - .await - { - Ok(Ok(())) => Ok(()), - Ok(Err(e)) => Err(e), - Err(_) => Err(OverlayError::ConnectionTimeout(format!( - "send timeout after {}s to {}", - SEND_TIMEOUT_SECS, self.remote_addr - ))), - } - } - - /// Returns the remote peer's socket address. - pub fn remote_addr(&self) -> SocketAddr { - self.remote_addr - } -} - -/// Receive half of a split connection. -/// -/// Created by [`Connection::split`]. Allows receiving messages without -/// holding a lock on the full connection. -pub struct ConnectionReceiver { - stream: futures::stream::SplitStream>, - remote_addr: SocketAddr, -} - -impl ConnectionReceiver { - /// Receives the next message from the peer. - /// - /// Returns `Ok(None)` if the connection was closed. - pub async fn recv(&mut self) -> Result> { - match self.stream.next().await { - Some(Ok(frame)) => { - trace!( - "Received message from {} ({} bytes)", - self.remote_addr, - frame.raw_len - ); - Ok(Some(frame)) - } - Some(Err(e)) => Err(e), - None => Ok(None), - } - } - - /// Returns the remote peer's socket address. - pub fn remote_addr(&self) -> SocketAddr { - self.remote_addr - } } /// TCP listener for accepting incoming peer connections.