From 9fa1365425e08caa17a0f659d070d64dbf036491 Mon Sep 17 00:00:00 2001 From: Nikhil Sharma Date: Mon, 3 Aug 2026 16:00:18 +0530 Subject: [PATCH 1/2] fix(l1): drop peers on undecodable RLPx inbound frames Signed-off-by: Nikhil Sharma --- .../networking/p2p/rlpx/connection/server.rs | 278 +++++++++++++++++- 1 file changed, 275 insertions(+), 3 deletions(-) diff --git a/crates/networking/p2p/rlpx/connection/server.rs b/crates/networking/p2p/rlpx/connection/server.rs index e41443961ed..cc38429435f 100644 --- a/crates/networking/p2p/rlpx/connection/server.rs +++ b/crates/networking/p2p/rlpx/connection/server.rs @@ -130,6 +130,14 @@ pub trait PeerConnectionServerProtocol: Send + Sync { announcement: NewPooledTransactionHashes, hashes: Vec, ) -> Result<(), ActorError>; + /// Inbound RLPx stream failed (codec/decode/IO). The connection must be dropped: + /// after a `Framed` decoder error the read half is finished and the peer can no + /// longer answer requests. + fn inbound_stream_failed( + &self, + reason: DisconnectReason, + detail: String, + ) -> Result<(), ActorError>; } #[cfg(feature = "l2")] @@ -358,6 +366,13 @@ pub struct PeerConnectionServer { impl PeerConnectionServer { #[started] async fn started(&mut self, ctx: &Context) { + // Unit tests may start the actor already in `Established` to exercise teardown + // without a full RLPx handshake. Production always starts as Initiator/Receiver. + #[cfg(test)] + if matches!(self.state, ConnectionState::Established(_)) { + return; + } + // Set a default eth version that we can update after we negotiate peer capabilities // This eth version will only be used to encode & decode the initial `Hello` messages. let eth_version = Arc::new(RwLock::new(EthCapVersion::default())); @@ -718,6 +733,40 @@ impl PeerConnectionServer { } } + #[send_handler] + async fn handle_inbound_stream_failed( + &mut self, + msg: peer_connection_server_protocol::InboundStreamFailed, + ctx: &Context, + ) { + if let ConnectionState::Established(ref mut established_state) = self.state { + debug!( + peer=%established_state.node, + reason=?msg.reason, + detail=%msg.detail, + "Inbound RLPx stream failed, dropping peer", + ); + // Preserve an earlier reason (e.g. peer-sent Disconnect) if one is already set. + // Use the stored reason for both metrics and any wire Disconnect so they agree. + let reason = *established_state + .disconnect_reason + .get_or_insert(msg.reason); + // Protocol breach: outbound may still be writable, so tell the peer why. + // Network/IO failures usually mean the socket is already gone. + if reason == DisconnectReason::ProtocolError { + send_disconnect_message(established_state, Some(reason)).await; + } + ctx.stop(); + } else { + debug!( + reason=?msg.reason, + detail=%msg.detail, + "Inbound RLPx stream failed while connection was not Established", + ); + ctx.stop(); + } + } + fn process_cast_error( state: &ConnectionState, result: Result<(), PeerConnectionError>, @@ -875,13 +924,34 @@ where ); } + // Decoder/`Framed` errors are terminal for the read half: the next poll is EOS, so + // "skipping" is impossible. Notify the actor so it stops promptly instead of leaving + // a zombie peer selectable until missed-pong timeout. + // See https://github.com/lambdaclass/ethrex/issues/7035 + let inbound_ctx = ctx.clone(); spawn_listener( ctx.clone(), - stream.filter_map(|result| match result { + stream.filter_map(move |result| match result { Ok(msg) => Some(peer_connection_server_protocol::IncomingMessage { message: msg }), Err(e) => { - debug!(error=?e, "Error receiving RLPx message"); - // Skipping invalid data + let reason = disconnect_reason_for_inbound_error(&e); + debug!( + error=?e, + ?reason, + "Error receiving RLPx message, notifying connection to drop peer", + ); + if let Err(send_err) = + inbound_ctx.send(peer_connection_server_protocol::InboundStreamFailed { + reason, + detail: e.to_string(), + }) + { + debug!( + error=?send_err, + ?reason, + "Failed to notify connection of inbound RLPx stream failure", + ); + } None } }), @@ -1083,6 +1153,20 @@ fn match_disconnect_reason(error: &PeerConnectionError) -> Option DisconnectReason { + match error { + PeerConnectionError::RLPDecodeError(_) + | PeerConnectionError::InvalidMessageFrame(_) + | PeerConnectionError::CryptographyError(_) + | PeerConnectionError::InvalidMessageLength => DisconnectReason::ProtocolError, + PeerConnectionError::IoError(_) => DisconnectReason::NetworkError, + _ => DisconnectReason::NetworkError, + } +} + async fn exchange_hello_messages( state: &mut Established, stream: &mut S, @@ -1945,3 +2029,191 @@ async fn retry_on_alternates( } } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::{ + peer_table::PeerTableServer, + rlpx::utils::decompress_pubkey, + tx_broadcaster::TxBroadcaster, + }; + use ethrex_rlp::error::RLPDecodeError; + use ethrex_storage::EngineType; + use secp256k1::rand::rngs::OsRng; + use std::{ + io, + net::{IpAddr, Ipv4Addr}, + sync::atomic::AtomicBool, + }; + + #[test] + fn inbound_rlp_decode_error_is_protocol_error() { + let err = PeerConnectionError::RLPDecodeError(RLPDecodeError::MalformedData); + assert_eq!( + disconnect_reason_for_inbound_error(&err), + DisconnectReason::ProtocolError + ); + } + + #[test] + fn inbound_invalid_frame_is_protocol_error() { + let err = PeerConnectionError::InvalidMessageFrame("bad mac".to_string()); + assert_eq!( + disconnect_reason_for_inbound_error(&err), + DisconnectReason::ProtocolError + ); + } + + #[test] + fn inbound_cryptography_error_is_protocol_error() { + let err = PeerConnectionError::CryptographyError("bad keystream".to_string()); + assert_eq!( + disconnect_reason_for_inbound_error(&err), + DisconnectReason::ProtocolError + ); + } + + #[test] + fn inbound_invalid_message_length_is_protocol_error() { + assert_eq!( + disconnect_reason_for_inbound_error(&PeerConnectionError::InvalidMessageLength), + DisconnectReason::ProtocolError + ); + } + + #[test] + fn inbound_io_error_is_network_error() { + let err = PeerConnectionError::IoError(io::Error::new(io::ErrorKind::BrokenPipe, "pipe")); + assert_eq!( + disconnect_reason_for_inbound_error(&err), + DisconnectReason::NetworkError + ); + } + + #[test] + fn inbound_other_errors_default_to_network_error() { + assert_eq!( + disconnect_reason_for_inbound_error(&PeerConnectionError::Timeout), + DisconnectReason::NetworkError + ); + } + + /// Regression for https://github.com/lambdaclass/ethrex/issues/7035: an inbound stream + /// failure must stop the connection actor and remove the peer from selection immediately + /// (not after missed-pong timeout). + #[tokio::test] + async fn inbound_stream_failed_removes_peer_from_selection() { + let store = Store::new("", EngineType::InMemory).expect("in-memory store"); + let blockchain = Arc::new(Blockchain::default_with_store(store.clone())); + let local_id = H256::from_low_u64_be(1); + let peer_table = PeerTableServer::spawn(local_id, 10, store.clone()); + let tx_broadcaster = + TxBroadcaster::spawn(peer_table.clone(), blockchain.clone(), 1_000).expect("tx bc"); + + let signer = SecretKey::new(&mut OsRng); + let node = Node::new( + IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), + 30303, + 30303, + decompress_pubkey(&PublicKey::from_secret_key(secp256k1::SECP256K1, &signer)), + ); + let node_id = node.node_id(); + let caps = vec![Capability::eth(68)]; + + let (outbound_tx, _outbound_rx) = tokio::sync::mpsc::channel(16); + let (broadcast_tx, _broadcast_rx) = broadcast::channel(1); + + let established = Established { + signer, + outbound_tx, + outbound_writer_timed_out: Arc::new(AtomicBool::new(false)), + node: node.clone(), + is_inbound: false, + storage: store, + blockchain, + capabilities: caps.clone(), + negotiated_eth_capability: Some(Capability::eth(68)), + negotiated_snap_capability: None, + last_block_range_update_block: 0, + requested_pooled_txs: HashMap::new(), + pending_tx_requests: Vec::new(), + client_version: "ethrex-test".to_string(), + connection_broadcast_send: broadcast_tx, + peer_table: peer_table.clone(), + #[cfg(feature = "l2")] + l2_state: L2ConnState::Unsupported, + tx_broadcaster, + current_requests: HashMap::new(), + disconnect_reason: None, + is_validated: true, + serve_request_window_start: Instant::now(), + serve_requests_in_window: 0, + txs_sent_to_peer: 0, + received_txs_from_peer: false, + missed_pongs: 0, + }; + + let handle = PeerConnectionServer { + state: ConnectionState::Established(Box::new(established)), + _admission_permit: None, + } + .start(); + + let connection = PeerConnection { + handle: handle.clone(), + }; + peer_table + .new_connected_peer(node, connection, caps, false) + .expect("register peer"); + + assert!( + peer_table + .has_eligible_peer(SUPPORTED_ETH_CAPABILITIES.to_vec()) + .await + .expect("has_eligible_peer"), + "peer must be selectable before inbound failure" + ); + assert!( + peer_table + .get_peer_connection(node_id) + .await + .expect("get_peer_connection") + .is_some(), + "peer connection handle must be present before inbound failure" + ); + + // NetworkError path skips wire Disconnect (no writer task needed). + handle + .inbound_stream_failed( + DisconnectReason::NetworkError, + "simulated framed decode error".to_string(), + ) + .expect("send InboundStreamFailed"); + handle.join().await; + + assert!( + !peer_table + .has_eligible_peer(SUPPORTED_ETH_CAPABILITIES.to_vec()) + .await + .expect("has_eligible_peer"), + "peer must not remain selectable after inbound stream failure" + ); + assert!( + peer_table + .get_peer_connection(node_id) + .await + .expect("get_peer_connection") + .is_none(), + "peer must be removed from the table after inbound stream failure" + ); + assert!( + peer_table + .get_best_peer(SUPPORTED_ETH_CAPABILITIES.to_vec()) + .await + .expect("get_best_peer") + .is_none(), + "get_best_peer must not return the dropped peer" + ); + } +} From cafccb5a8194f1b5fab243b67254ae585a54725a Mon Sep 17 00:00:00 2001 From: Nikhil Sharma Date: Tue, 4 Aug 2026 22:37:56 +0530 Subject: [PATCH 2/2] fix(l1): harden inbound drop path and add handshake regression Signed-off-by: Nikhil Sharma --- .../networking/p2p/rlpx/connection/server.rs | 342 ++++++++++++++---- 1 file changed, 266 insertions(+), 76 deletions(-) diff --git a/crates/networking/p2p/rlpx/connection/server.rs b/crates/networking/p2p/rlpx/connection/server.rs index cc38429435f..da76621305a 100644 --- a/crates/networking/p2p/rlpx/connection/server.rs +++ b/crates/networking/p2p/rlpx/connection/server.rs @@ -366,13 +366,6 @@ pub struct PeerConnectionServer { impl PeerConnectionServer { #[started] async fn started(&mut self, ctx: &Context) { - // Unit tests may start the actor already in `Established` to exercise teardown - // without a full RLPx handshake. Production always starts as Initiator/Receiver. - #[cfg(test)] - if matches!(self.state, ConnectionState::Established(_)) { - return; - } - // Set a default eth version that we can update after we negotiate peer capabilities // This eth version will only be used to encode & decode the initial `Hello` messages. let eth_version = Arc::new(RwLock::new(EthCapVersion::default())); @@ -740,31 +733,16 @@ impl PeerConnectionServer { ctx: &Context, ) { if let ConnectionState::Established(ref mut established_state) = self.state { - debug!( - peer=%established_state.node, - reason=?msg.reason, - detail=%msg.detail, - "Inbound RLPx stream failed, dropping peer", - ); - // Preserve an earlier reason (e.g. peer-sent Disconnect) if one is already set. - // Use the stored reason for both metrics and any wire Disconnect so they agree. - let reason = *established_state - .disconnect_reason - .get_or_insert(msg.reason); - // Protocol breach: outbound may still be writable, so tell the peer why. - // Network/IO failures usually mean the socket is already gone. - if reason == DisconnectReason::ProtocolError { - send_disconnect_message(established_state, Some(reason)).await; - } - ctx.stop(); + apply_inbound_stream_failure(established_state, msg.reason, &msg.detail).await; } else { debug!( reason=?msg.reason, detail=%msg.detail, "Inbound RLPx stream failed while connection was not Established", ); - ctx.stop(); } + // Always stop: inbound is unusable and `stopped()` runs remove_peer when Established. + ctx.stop(); } fn process_cast_error( @@ -1104,6 +1082,31 @@ async fn send_disconnect_message(state: &mut Established, reason: Option Option DisconnectReason { match error { + // Wire/format failures from `RLPxCodec::decode` / message RLP decode. PeerConnectionError::RLPDecodeError(_) | PeerConnectionError::InvalidMessageFrame(_) | PeerConnectionError::CryptographyError(_) | PeerConnectionError::InvalidMessageLength => DisconnectReason::ProtocolError, - PeerConnectionError::IoError(_) => DisconnectReason::NetworkError, - _ => DisconnectReason::NetworkError, + + // Transport errors, local/internal failures, and variants not expected from the + // inbound Framed path. Listed explicitly (fail-closed) rather than `_`. + PeerConnectionError::IoError(_) + | PeerConnectionError::InternalError(_) + | PeerConnectionError::Disconnected + | PeerConnectionError::HandshakeError(_) + | PeerConnectionError::StateError(_) + | PeerConnectionError::NoMatchingCapabilities + | PeerConnectionError::TooManyPeers + | PeerConnectionError::OutboundSendTimeout + | PeerConnectionError::OutboundQueueFull + | PeerConnectionError::DisconnectReceived(_) + | PeerConnectionError::DisconnectSent(_) + | PeerConnectionError::NotFound(_) + | PeerConnectionError::InvalidPeerId + | PeerConnectionError::InvalidRecoveryId + | PeerConnectionError::ExpectedRequestId(_) + | PeerConnectionError::MessageNotHandled(_) + | PeerConnectionError::BadRequest(_) + | PeerConnectionError::RLPEncodeError(_) + | PeerConnectionError::StoreError(_) + | PeerConnectionError::BroadcastError(_) + | PeerConnectionError::RecvError(_) + | PeerConnectionError::SendMessage(_) + | PeerConnectionError::MempoolError(_) + | PeerConnectionError::BlockchainError(_) + | PeerConnectionError::IncompatibleProtocol + | PeerConnectionError::InvalidBlockRange + | PeerConnectionError::L2CapabilityNotNegotiated + | PeerConnectionError::InvalidBlockRangeUpdate + | PeerConnectionError::ActorError(_) + | PeerConnectionError::Timeout + | PeerConnectionError::UnexpectedResponse(_, _) => DisconnectReason::NetworkError, + #[cfg(feature = "l2")] + PeerConnectionError::RollupStoreError(_) => DisconnectReason::NetworkError, } } @@ -2034,10 +2075,10 @@ async fn retry_on_alternates( mod tests { use super::*; use crate::{ - peer_table::PeerTableServer, - rlpx::utils::decompress_pubkey, - tx_broadcaster::TxBroadcaster, + network::P2PContext, peer_table::PeerTableServer, rlpx::utils::decompress_pubkey, + tx_broadcaster::TxBroadcaster, types::NetworkConfig, }; + use ethrex_common::types::Genesis; use ethrex_rlp::error::RLPDecodeError; use ethrex_storage::EngineType; use secp256k1::rand::rngs::OsRng; @@ -2046,6 +2087,8 @@ mod tests { net::{IpAddr, Ipv4Addr}, sync::atomic::AtomicBool, }; + use tokio::net::TcpListener; + use tokio_util::task::TaskTracker; #[test] fn inbound_rlp_decode_error_is_protocol_error() { @@ -2092,25 +2135,28 @@ mod tests { } #[test] - fn inbound_other_errors_default_to_network_error() { + fn inbound_internal_error_is_network_error() { + let err = PeerConnectionError::InternalError("poisoned eth_version lock".to_string()); + assert_eq!( + disconnect_reason_for_inbound_error(&err), + DisconnectReason::NetworkError + ); + } + + #[test] + fn inbound_timeout_is_network_error() { assert_eq!( disconnect_reason_for_inbound_error(&PeerConnectionError::Timeout), DisconnectReason::NetworkError ); } - /// Regression for https://github.com/lambdaclass/ethrex/issues/7035: an inbound stream - /// failure must stop the connection actor and remove the peer from selection immediately - /// (not after missed-pong timeout). - #[tokio::test] - async fn inbound_stream_failed_removes_peer_from_selection() { + fn test_established(outbound_tx: tokio::sync::mpsc::Sender) -> Established { let store = Store::new("", EngineType::InMemory).expect("in-memory store"); let blockchain = Arc::new(Blockchain::default_with_store(store.clone())); - let local_id = H256::from_low_u64_be(1); - let peer_table = PeerTableServer::spawn(local_id, 10, store.clone()); + let peer_table = PeerTableServer::spawn(H256::from_low_u64_be(1), 10, store.clone()); let tx_broadcaster = TxBroadcaster::spawn(peer_table.clone(), blockchain.clone(), 1_000).expect("tx bc"); - let signer = SecretKey::new(&mut OsRng); let node = Node::new( IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), @@ -2118,21 +2164,16 @@ mod tests { 30303, decompress_pubkey(&PublicKey::from_secret_key(secp256k1::SECP256K1, &signer)), ); - let node_id = node.node_id(); - let caps = vec![Capability::eth(68)]; - - let (outbound_tx, _outbound_rx) = tokio::sync::mpsc::channel(16); let (broadcast_tx, _broadcast_rx) = broadcast::channel(1); - - let established = Established { + Established { signer, outbound_tx, outbound_writer_timed_out: Arc::new(AtomicBool::new(false)), - node: node.clone(), + node, is_inbound: false, storage: store, blockchain, - capabilities: caps.clone(), + capabilities: vec![Capability::eth(68)], negotiated_eth_capability: Some(Capability::eth(68)), negotiated_snap_capability: None, last_block_range_update_block: 0, @@ -2140,7 +2181,7 @@ mod tests { pending_tx_requests: Vec::new(), client_version: "ethrex-test".to_string(), connection_broadcast_send: broadcast_tx, - peer_table: peer_table.clone(), + peer_table, #[cfg(feature = "l2")] l2_state: L2ConnState::Unsupported, tx_broadcaster, @@ -2152,63 +2193,212 @@ mod tests { txs_sent_to_peer: 0, received_txs_from_peer: false, missed_pongs: 0, - }; + } + } - let handle = PeerConnectionServer { - state: ConnectionState::Established(Box::new(established)), - _admission_permit: None, + #[tokio::test] + async fn apply_inbound_stream_failure_records_network_error_reason() { + let (outbound_tx, mut outbound_rx) = tokio::sync::mpsc::channel(16); + let mut state = test_established(outbound_tx); + + apply_inbound_stream_failure( + &mut state, + DisconnectReason::NetworkError, + "simulated io failure", + ) + .await; + + assert_eq!( + state.disconnect_reason, + Some(DisconnectReason::NetworkError) + ); + // NetworkError must not enqueue a wire Disconnect. + assert!(outbound_rx.try_recv().is_err()); + } + + #[tokio::test] + async fn apply_inbound_stream_failure_preserves_earlier_reason() { + let (outbound_tx, mut outbound_rx) = tokio::sync::mpsc::channel(16); + let mut state = test_established(outbound_tx); + state.disconnect_reason = Some(DisconnectReason::PingTimeout); + + apply_inbound_stream_failure( + &mut state, + DisconnectReason::ProtocolError, + "decode failed after peer disconnect reason was set", + ) + .await; + + assert_eq!(state.disconnect_reason, Some(DisconnectReason::PingTimeout)); + // Stored reason is not ProtocolError, so no wire Disconnect. + assert!(outbound_rx.try_recv().is_err()); + } + + #[tokio::test] + async fn apply_inbound_stream_failure_sends_disconnect_on_protocol_error() { + let (outbound_tx, mut outbound_rx) = tokio::sync::mpsc::channel(16); + let mut state = test_established(outbound_tx); + + apply_inbound_stream_failure( + &mut state, + DisconnectReason::ProtocolError, + "simulated decode failure", + ) + .await; + + assert_eq!( + state.disconnect_reason, + Some(DisconnectReason::ProtocolError) + ); + match outbound_rx.try_recv() { + Ok(Message::Disconnect(DisconnectMessage { + reason: Some(DisconnectReason::ProtocolError), + })) => {} + other => panic!("expected ProtocolError Disconnect, got {other:?}"), } - .start(); + } - let connection = PeerConnection { - handle: handle.clone(), - }; - peer_table - .new_connected_peer(node, connection, caps, false) - .expect("register peer"); + /// Behavioral regression for https://github.com/lambdaclass/ethrex/issues/7035. + /// + /// Completes a real TCP RLPx handshake (no test-only `started()` seam), then + /// delivers `InboundStreamFailed` — the same message the inbound `Framed` + /// listener sends on codec/decode `Err`. Asserts the peer leaves selection + /// well before missed-pong timeout (~30s). + /// + /// Score is not asserted: this path `remove_peer`s the row rather than + /// applying `record_failure`. + #[tokio::test] + async fn inbound_stream_failed_removes_peer_from_selection_after_handshake() { + let mut store = Store::new("", EngineType::InMemory).expect("in-memory store"); + store + .add_initial_state(Genesis::default()) + .await + .expect("genesis"); + let blockchain = Arc::new(Blockchain::default_with_store(store.clone())); + + let sut_signer = SecretKey::new(&mut OsRng); + let dialer_signer = SecretKey::new(&mut OsRng); + + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("bind loopback"); + let bound = listener.local_addr().expect("local_addr"); + + let sut_node = Node::new( + bound.ip(), + bound.port(), + bound.port(), + decompress_pubkey(&PublicKey::from_secret_key( + secp256k1::SECP256K1, + &sut_signer, + )), + ); + let dialer_node = Node::new( + IpAddr::V4(Ipv4Addr::LOCALHOST), + 0, + 0, + decompress_pubkey(&PublicKey::from_secret_key( + secp256k1::SECP256K1, + &dialer_signer, + )), + ); + let dialer_id = dialer_node.node_id(); + + let sut_table = PeerTableServer::spawn(sut_node.node_id(), 10, store.clone()); + let dialer_table = PeerTableServer::spawn(dialer_node.node_id(), 10, store.clone()); + + let sut_ctx = P2PContext::new( + sut_node.clone(), + NetworkConfig::from_node(&sut_node), + TaskTracker::new(), + sut_signer, + sut_table.clone(), + store.clone(), + blockchain.clone(), + "ethrex-test-sut".to_string(), + None, + 1_000, + 1.0, + ) + .expect("sut P2PContext"); + let dialer_ctx = P2PContext::new( + dialer_node.clone(), + NetworkConfig::from_node(&dialer_node), + TaskTracker::new(), + dialer_signer, + dialer_table, + store, + blockchain, + "ethrex-test-dialer".to_string(), + None, + 1_000, + 1.0, + ) + .expect("dialer P2PContext"); + + let sut_ctx_accept = sut_ctx.clone(); + let accept = tokio::spawn(async move { + let (stream, peer_addr) = listener.accept().await.expect("accept"); + let permit = sut_ctx_accept + .inbound_admission + .clone() + .acquire_owned() + .await + .expect("inbound permit"); + PeerConnection::spawn_as_receiver(sut_ctx_accept, peer_addr, stream, permit) + }); + + // Keep the dialer connection alive for the duration of the test so a + // remote close does not race our simulated inbound failure. + let _dialer_conn = PeerConnection::spawn_as_initiator(dialer_ctx, &sut_node); + + let peer_conn = tokio::time::timeout(Duration::from_secs(10), async { + loop { + if let Ok(Some(conn)) = sut_table.get_peer_connection(dialer_id).await { + break conn; + } + tokio::time::sleep(Duration::from_millis(25)).await; + } + }) + .await + .expect("peer should become selectable after a real RLPx handshake"); + + let _sut_conn = accept.await.expect("accept task"); assert!( - peer_table + sut_table .has_eligible_peer(SUPPORTED_ETH_CAPABILITIES.to_vec()) .await .expect("has_eligible_peer"), "peer must be selectable before inbound failure" ); - assert!( - peer_table - .get_peer_connection(node_id) - .await - .expect("get_peer_connection") - .is_some(), - "peer connection handle must be present before inbound failure" - ); - // NetworkError path skips wire Disconnect (no writer task needed). - handle + peer_conn + .handle .inbound_stream_failed( - DisconnectReason::NetworkError, + DisconnectReason::ProtocolError, "simulated framed decode error".to_string(), ) .expect("send InboundStreamFailed"); - handle.join().await; + peer_conn.handle.join().await; assert!( - !peer_table + !sut_table .has_eligible_peer(SUPPORTED_ETH_CAPABILITIES.to_vec()) .await .expect("has_eligible_peer"), "peer must not remain selectable after inbound stream failure" ); assert!( - peer_table - .get_peer_connection(node_id) + sut_table + .get_peer_connection(dialer_id) .await .expect("get_peer_connection") .is_none(), "peer must be removed from the table after inbound stream failure" ); assert!( - peer_table + sut_table .get_best_peer(SUPPORTED_ETH_CAPABILITIES.to_vec()) .await .expect("get_best_peer")