Skip to content

Remove dead overlay Encoder impl and Connection::split chain #3916

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
tomerweller wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Remove dead overlay Encoder impl and Connection::split chain #3916

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
7a32106
Remove dead overlay Encoder impl and Connection::split chain
Aug 23, 2026
cdc5a72
Trigger fresh CI run for approved dead-code removal
Aug 23, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
47 changes: 12 additions & 35 deletions crates/overlay/src/codec.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand All @@ -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!(
Expand Down Expand Up @@ -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
///
Expand Down Expand Up @@ -268,29 +268,6 @@ impl Decoder for MessageCodec {
}
}

impl Encoder<AuthenticatedMessage> 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.
Expand Down Expand Up @@ -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,
Expand All @@ -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,
Expand Down Expand Up @@ -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();
Expand Down
91 changes: 1 addition & 90 deletions crates/overlay/src/connection.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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<Framed<BoxedIo, MessageCodec>, 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<Framed<BoxedIo, MessageCodec>>,
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<Option<MessageFrame>> {
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.
Expand Down
Loading