From b274cf15ecae599f78f85daca65fe41b3276e42a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Fri, 4 Sep 2026 19:45:33 -0300 Subject: [PATCH] fix(p2p): never leave a block root pending without an outcome A root sits in `pending_root_requests` from the moment a BlocksByRoot request is sent until something retires it, and the `FetchBlock` handler deduplicates every later fetch against that table. A root that is never retired is therefore unreachable for the life of the process: the chain actor can ask for it any number of times and no request reaches the wire, with no error to show for it. Three paths ended an attempt without retiring the root: - An error response for a root request re-inserted the request id into `outbound_requests` and returned. `handle_fetch_failure` never reads that map, and an error response completes the exchange, so no `OutboundFailure` followed to pick the root up. - A non-empty response whose blocks all failed the requested-root check fell out of the loop having removed nothing. - libp2p drops a request whose dial it rejects with `DialError::DialPeerConditionFalse` and emits no event at all, so nothing downstream could react. Route the first two through `handle_fetch_failure`, and arm a watchdog at send time for anything libp2p never reports on. The watchdog is pinned above the req/resp timeout, now set explicitly rather than inherited from libp2p's default, so a request that did reach a connection still fails through libp2p's own path first. Split the retry accounting out of `handle_fetch_failure` so it can be tested without a live actor, and make the response path pick the one block that can answer a single-root request instead of looping. --- crates/net/p2p/src/lib.rs | 65 +++++- crates/net/p2p/src/req_resp/handlers.rs | 268 ++++++++++++++++++------ crates/net/p2p/src/req_resp/mod.rs | 1 + 3 files changed, 269 insertions(+), 65 deletions(-) diff --git a/crates/net/p2p/src/lib.rs b/crates/net/p2p/src/lib.rs index c2b04359..46d658c9 100644 --- a/crates/net/p2p/src/lib.rs +++ b/crates/net/p2p/src/lib.rs @@ -48,7 +48,7 @@ use crate::{ req_resp::{ BLOCKS_BY_RANGE_PROTOCOL_V1, BLOCKS_BY_ROOT_PROTOCOL_V1, Codec, MAX_COMPRESSED_PAYLOAD_SIZE, MAX_REQUEST_BLOCKS, Request, STATUS_PROTOCOL_V1, build_status, - fetch_block_from_peer, + fetch_block_from_peer, handle_fetch_failure, }, swarm_adapter::SwarmHandle, }; @@ -68,6 +68,19 @@ const BACKOFF_MULTIPLIER: u64 = 2; const PEER_REDIAL_INTERVAL_SECS: u64 = 12; const MAX_SYNC_RANGE: u64 = MAX_REQUEST_BLOCKS * 64; // 65,536 slots (~3 days) +/// Timeout libp2p applies to an in-flight req/resp exchange. +const REQ_RESP_TIMEOUT: Duration = Duration::from_secs(10); + +/// Backstop for a `BlocksByRoot` request that libp2p never reports an outcome +/// for. libp2p drops a request whose dial it rejects with +/// `DialError::DialPeerConditionFalse` without emitting `OutboundFailure`, and +/// a root left in `pending_root_requests` is deduplicated out of every later +/// fetch, so the block can never be recovered for the life of the process. +/// +/// Kept above `REQ_RESP_TIMEOUT` so a request that did reach a connection +/// fails through libp2p's own path first. +const ROOT_FETCH_WATCHDOG: Duration = Duration::from_secs(15); + pub(crate) struct PendingRequest { pub(crate) attempts: u32, pub(crate) failed_peers: HashSet, @@ -298,7 +311,7 @@ pub fn build_swarm(config: SwarmConfig) -> Result { request_response::ProtocolSupport::Full, ), ], - Default::default(), + request_response::Config::default().with_request_timeout(REQ_RESP_TIMEOUT), ); let secret_key = @@ -547,6 +560,14 @@ pub(crate) trait P2PProtocol: Send + Sync { #[allow(dead_code)] // invoked via send_after, not called directly fn retry_block_fetch(&self, root: H256) -> Result<(), ActorError>; #[allow(dead_code)] // invoked via send_after, not called directly + fn block_fetch_timeout( + &self, + root: H256, + peer: PeerId, + request_id: OutboundRequestId, + attempt: u32, + ) -> Result<(), ActorError>; + #[allow(dead_code)] // invoked via send_after, not called directly fn retry_peer_redial(&self, peer_id: PeerId) -> Result<(), ActorError>; #[allow(dead_code)] // invoked via send_after, not called directly fn discover_peers(&self) -> Result<(), ActorError>; @@ -558,7 +579,7 @@ impl P2PServer { async fn handle_retry_block_fetch( &mut self, msg: p2p_protocol::RetryBlockFetch, - _ctx: &Context, + ctx: &Context, ) { let root = msg.root; // Check if still pending (might have succeeded during backoff) @@ -569,12 +590,44 @@ impl P2PServer { trace!(%root, "Retrying block fetch after backoff"); - if !fetch_block_from_peer(self, root).await { + if !fetch_block_from_peer(self, root, ctx).await { tracing::error!(%root, "Failed to retry block fetch, giving up"); self.pending_root_requests.remove(&root); } } + /// Fail an attempt that libp2p never reported an outcome for. + /// + /// Without this the root stays in `pending_root_requests` forever and the + /// deduplication in the `FetchBlock` handler swallows every later attempt + /// to fetch that block. + #[send_handler] + async fn handle_block_fetch_timeout( + &mut self, + msg: p2p_protocol::BlockFetchTimeout, + ctx: &Context, + ) { + // Any outcome for this attempt either cleared the entry or moved it on + // to a later attempt, which arms a watchdog of its own. + let outstanding = self + .pending_root_requests + .get(&msg.root) + .is_some_and(|pending| pending.attempts == msg.attempt); + if !outstanding { + trace!(root = %msg.root, "Block fetch settled before the watchdog fired"); + return; + } + + warn!( + root = %msg.root, + peer = %msg.peer, + attempt = msg.attempt, + "BlocksByRoot request produced no libp2p outcome, failing it" + ); + self.outbound_requests.remove(&msg.request_id); + handle_fetch_failure(self, msg.root, msg.peer, ctx).await; + } + #[send_handler] async fn handle_retry_peer_redial( &mut self, @@ -640,14 +693,14 @@ impl Handler for P2PServer { } impl Handler for P2PServer { - async fn handle(&mut self, msg: FetchBlock, _ctx: &Context) { + async fn handle(&mut self, msg: FetchBlock, ctx: &Context) { let root = msg.root; // Deduplicate - if already pending, ignore if self.pending_root_requests.contains_key(&root) { trace!(%root, "Block fetch already in progress, ignoring duplicate"); return; } - fetch_block_from_peer(self, root).await; + fetch_block_from_peer(self, root, ctx).await; } } diff --git a/crates/net/p2p/src/req_resp/handlers.rs b/crates/net/p2p/src/req_resp/handlers.rs index 897ade8f..2d3b6d81 100644 --- a/crates/net/p2p/src/req_resp/handlers.rs +++ b/crates/net/p2p/src/req_resp/handlers.rs @@ -1,4 +1,4 @@ -use std::collections::HashSet; +use std::collections::{HashMap, HashSet}; use ethlambda_network_api::BlockSource; use ethlambda_storage::Store; @@ -19,7 +19,7 @@ use super::{ }; use crate::{ BACKOFF_MULTIPLIER, INITIAL_BACKOFF_MS, MAX_FETCH_RETRIES, MAX_SYNC_RANGE, P2PServer, - PendingRequest, PendingRequestKind, RangeSyncState, p2p_protocol, + PendingRequest, PendingRequestKind, ROOT_FETCH_WATCHDOG, RangeSyncState, p2p_protocol, req_resp::RequestedBlockRoots, }; @@ -80,10 +80,8 @@ pub async fn handle_req_resp_message( .await; } Some(PendingRequestKind::Root(root)) => { - handle_blocks_by_root_response( - server, blocks, peer, request_id, root, ctx, - ) - .await; + handle_blocks_by_root_response(server, blocks, peer, root, ctx) + .await; } None => { debug!(%peer, ?request_id, "Received blocks response for unknown request_id"); @@ -99,8 +97,12 @@ pub async fn handle_req_resp_message( Some(PendingRequestKind::Range { .. }) => { fail_range_request(server, &peer); } - Some(request @ PendingRequestKind::Root(_)) => { - server.outbound_requests.insert(request_id, request); + Some(PendingRequestKind::Root(root)) => { + // An error response completes the exchange, so + // no `OutboundFailure` follows to retire the + // root. Fail it here or it stays pending + // forever and deduplicates every later fetch. + handle_fetch_failure(server, root, peer, ctx).await; } None => {} } @@ -283,48 +285,48 @@ fn canonical_blocks_by_range(store: &Store, start_slot: u64, count: u64) -> Vec< .unwrap_or_default() } +/// Pick the block that answers a `BlocksByRoot` request out of the response. +/// +/// Requests carry a single root, so at most one block can answer one. Anything +/// else the peer sent is unsolicited and dropped. +fn matching_block(blocks: Vec, requested_root: H256) -> Option { + blocks + .into_iter() + .find(|block| block.message.hash_tree_root() == requested_root) +} + async fn handle_blocks_by_root_response( server: &mut P2PServer, blocks: Vec, peer: PeerId, - request_id: request_response::OutboundRequestId, requested_root: H256, ctx: &Context, ) { - trace!(%peer, count = blocks.len(), "Received BlocksByRoot response"); - - if blocks.is_empty() { - // Re-insert so failure handling can find it - server - .outbound_requests - .insert(request_id, PendingRequestKind::Root(requested_root)); - debug!(%peer, "Received empty BlocksByRoot response"); + let received = blocks.len(); + trace!(%peer, count = received, "Received BlocksByRoot response"); + + // A response that answers nothing is a failed attempt, whether it was empty + // or carried only blocks we never asked for. Treating the latter as a + // no-op would leave the root pending forever, and the deduplication in the + // `FetchBlock` handler would swallow every later attempt to fetch it. + let Some(block) = matching_block(blocks, requested_root) else { + debug!( + %peer, + received, + expected_root = %ethlambda_types::ShortRoot(&requested_root.0), + "BlocksByRoot response carried no matching block" + ); handle_fetch_failure(server, requested_root, peer, ctx).await; return; - } - - for block in blocks { - let root = block.message.hash_tree_root(); - - // Validate that this block matches what we requested - if root != requested_root { - debug!( - %peer, - received_root = %ethlambda_types::ShortRoot(&root.0), - expected_root = %ethlambda_types::ShortRoot(&requested_root.0), - "Received block with mismatched root, ignoring" - ); - continue; - } + }; - // Clean up tracking for this root - server.pending_root_requests.remove(&root); + // Clean up tracking for this root + server.pending_root_requests.remove(&requested_root); - if let Some(ref blockchain) = server.blockchain { - let _ = blockchain - .new_block(block, BlockSource::Sync) - .inspect_err(|err| error!(%err, "Failed to forward fetched block to blockchain")); - } + if let Some(ref blockchain) = server.blockchain { + let _ = blockchain + .new_block(block, BlockSource::Sync) + .inspect_err(|err| error!(%err, "Failed to forward fetched block to blockchain")); } } @@ -398,7 +400,11 @@ pub fn build_status(store: &Store) -> Status { /// Fetch a missing block from a random connected peer. /// Handles tracking in both pending_requests and request_id_map. -pub async fn fetch_block_from_peer(server: &mut P2PServer, root: H256) -> bool { +pub async fn fetch_block_from_peer( + server: &mut P2PServer, + root: H256, + ctx: &Context, +) -> bool { if server.connected_peers.is_empty() { debug!(%root, "Cannot fetch block: no connected peers"); return false; @@ -466,19 +472,34 @@ pub async fn fetch_block_from_peer(server: &mut P2PServer, root: H256) -> bool { }; // Track the request if not already tracked (new request) - server + let attempt = server .pending_root_requests .entry(root) .or_insert(PendingRequest { attempts: 1, failed_peers: HashSet::new(), - }); + }) + .attempts; // Map request_id to root for failure handling server .outbound_requests .insert(request_id, PendingRequestKind::Root(root)); + // Every other exit from this attempt runs off a libp2p event, and libp2p + // does not always emit one, so arm a backstop that fails the attempt if + // nothing else does. + send_after( + ROOT_FETCH_WATCHDOG, + ctx.clone(), + p2p_protocol::BlockFetchTimeout { + root, + peer, + request_id, + attempt, + }, + ); + true } @@ -556,33 +577,68 @@ fn fail_range_request(server: &mut P2PServer, peer: &PeerId) { } } -async fn handle_fetch_failure( - server: &mut P2PServer, +/// What follows a failed attempt to fetch a block by root. +#[derive(Debug, PartialEq, Eq)] +enum FetchFailure { + /// Wait `backoff`, then try another peer. + Retry { attempts: u32, backoff: Duration }, + /// Out of attempts. The root is no longer pending. + GaveUp { attempts: u32 }, + /// Nothing was pending for this root, so the failure is late or duplicate. + NotPending, +} + +/// Charge a failed attempt to the pending table and decide what follows. +/// +/// Split out of [`handle_fetch_failure`] so the retry accounting is testable +/// without a live actor. +fn record_fetch_failure( + pending_root_requests: &mut HashMap, root: H256, peer: PeerId, - ctx: &Context, -) { - let Some(pending) = server.pending_root_requests.get_mut(&root) else { - return; +) -> FetchFailure { + let Some(pending) = pending_root_requests.get_mut(&root) else { + return FetchFailure::NotPending; }; pending.failed_peers.insert(peer); + let attempts = pending.attempts; - if pending.attempts >= MAX_FETCH_RETRIES { - error!(%root, %peer, attempts=%pending.attempts, - "Block fetch failed after max retries, giving up"); - server.pending_root_requests.remove(&root); - return; + if attempts >= MAX_FETCH_RETRIES { + pending_root_requests.remove(&root); + return FetchFailure::GaveUp { attempts }; } - let backoff_ms = INITIAL_BACKOFF_MS * BACKOFF_MULTIPLIER.pow(pending.attempts - 1); - let backoff = Duration::from_millis(backoff_ms); - - debug!(%root, %peer, attempts=%pending.attempts, ?backoff, "Block fetch failed, scheduling retry"); - pending.attempts += 1; + let backoff_ms = INITIAL_BACKOFF_MS * BACKOFF_MULTIPLIER.pow(attempts - 1); - send_after(backoff, ctx.clone(), p2p_protocol::RetryBlockFetch { root }); + FetchFailure::Retry { + attempts, + backoff: Duration::from_millis(backoff_ms), + } +} + +/// Retire one failed attempt at fetching `root`. +/// +/// Every path that ends an attempt must come through here: a root left in +/// `pending_root_requests` is deduplicated out of every later fetch, so a +/// silent exit loses that block for the life of the process. +pub(crate) async fn handle_fetch_failure( + server: &mut P2PServer, + root: H256, + peer: PeerId, + ctx: &Context, +) { + match record_fetch_failure(&mut server.pending_root_requests, root, peer) { + FetchFailure::NotPending => {} + FetchFailure::GaveUp { attempts } => { + error!(%root, %peer, attempts, "Block fetch failed after max retries, giving up"); + } + FetchFailure::Retry { attempts, backoff } => { + debug!(%root, %peer, attempts, ?backoff, "Block fetch failed, scheduling retry"); + send_after(backoff, ctx.clone(), p2p_protocol::RetryBlockFetch { root }); + } + } } #[cfg(test)] @@ -656,4 +712,98 @@ mod tests { assert_eq!(roots, vec![root_1, root_2, root_4]); assert!(!roots.contains(&side_root_3)); } + + fn pending_root(root: H256, attempts: u32) -> HashMap { + HashMap::from([( + root, + PendingRequest { + attempts, + failed_peers: HashSet::new(), + }, + )]) + } + + /// A response that answers nothing must not look like a success. Returning + /// `Some` for an unrequested block would leave the requested root pending + /// forever, and the `FetchBlock` deduplication would then swallow every + /// later attempt to fetch it. + #[test] + fn matching_block_rejects_a_response_that_answers_a_different_root() { + let wanted = signed_block(1, H256::ZERO); + let wanted_root = wanted.message.hash_tree_root(); + let other = signed_block(2, H256::ZERO); + + assert!(matching_block(vec![other.clone()], wanted_root).is_none()); + assert!(matching_block(Vec::new(), wanted_root).is_none()); + + let found = matching_block(vec![other, wanted], wanted_root) + .expect("the requested block answers the request"); + assert_eq!(found.message.hash_tree_root(), wanted_root); + } + + /// A failure for a root nobody is waiting on must stay a no-op, so a late + /// or duplicate event cannot resurrect a root that already succeeded. + #[test] + fn record_fetch_failure_ignores_an_untracked_root() { + let mut pending = HashMap::new(); + let outcome = record_fetch_failure(&mut pending, H256::ZERO, PeerId::random()); + + assert_eq!(outcome, FetchFailure::NotPending); + assert!(pending.is_empty()); + } + + #[test] + fn record_fetch_failure_backs_off_and_excludes_the_failing_peer() { + let root = H256::ZERO; + let peer = PeerId::random(); + let mut pending = pending_root(root, 1); + + let first = record_fetch_failure(&mut pending, root, peer); + assert_eq!( + first, + FetchFailure::Retry { + attempts: 1, + backoff: Duration::from_millis(INITIAL_BACKOFF_MS), + } + ); + + let second = record_fetch_failure(&mut pending, root, PeerId::random()); + assert_eq!( + second, + FetchFailure::Retry { + attempts: 2, + backoff: Duration::from_millis(INITIAL_BACKOFF_MS * BACKOFF_MULTIPLIER), + } + ); + + let entry = pending.get(&root).expect("root is still pending"); + assert_eq!(entry.attempts, 3); + assert!(entry.failed_peers.contains(&peer)); + } + + /// Giving up has to clear the entry: leaving it behind is the same + /// permanent lock as never failing the attempt at all. + #[test] + fn record_fetch_failure_clears_the_root_when_it_gives_up() { + let root = H256::ZERO; + let mut pending = pending_root(root, MAX_FETCH_RETRIES); + + let outcome = record_fetch_failure(&mut pending, root, PeerId::random()); + + assert_eq!( + outcome, + FetchFailure::GaveUp { + attempts: MAX_FETCH_RETRIES + } + ); + assert!(!pending.contains_key(&root)); + } + + /// The watchdog is a backstop for requests libp2p never reports on, so it + /// must outlast libp2p's own timeout. Inverting these makes it fire on + /// healthy-but-slow requests and hides the real failure path. + #[test] + fn the_fetch_watchdog_outlasts_the_libp2p_request_timeout() { + assert!(ROOT_FETCH_WATCHDOG > crate::REQ_RESP_TIMEOUT); + } } diff --git a/crates/net/p2p/src/req_resp/mod.rs b/crates/net/p2p/src/req_resp/mod.rs index 11acb79f..bbd67d61 100644 --- a/crates/net/p2p/src/req_resp/mod.rs +++ b/crates/net/p2p/src/req_resp/mod.rs @@ -5,6 +5,7 @@ mod messages; pub use codec::Codec; pub use encoding::{MAX_COMPRESSED_PAYLOAD_SIZE, MAX_PAYLOAD_SIZE}; +pub(crate) use handlers::handle_fetch_failure; pub use handlers::{build_status, fetch_block_from_peer, handle_req_resp_message}; pub use messages::{ BLOCKS_BY_RANGE_PROTOCOL_V1, BLOCKS_BY_ROOT_PROTOCOL_V1, BlocksByRangeRequest,