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 1/6] 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, From 9d6d4c8e9ca007474b5e7cd880447b203d0cf5ea Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Mon, 7 Sep 2026 10:56:24 -0300 Subject: [PATCH 2/6] refactor(p2p): state the watchdog ordering at compile time `matching_block` had one caller and read better in place, so inline it. Pin the watchdog-outlasts-timeout relationship with a `const` assert beside the two constants instead of a unit test. It is a property of the values themselves, so it belongs where they are declared and should fail the build rather than a test run. --- crates/net/p2p/src/lib.rs | 7 ++++ crates/net/p2p/src/req_resp/handlers.rs | 44 ++++--------------------- 2 files changed, 14 insertions(+), 37 deletions(-) diff --git a/crates/net/p2p/src/lib.rs b/crates/net/p2p/src/lib.rs index 46d658c9..4fbaa131 100644 --- a/crates/net/p2p/src/lib.rs +++ b/crates/net/p2p/src/lib.rs @@ -81,6 +81,13 @@ const REQ_RESP_TIMEOUT: Duration = Duration::from_secs(10); /// fails through libp2p's own path first. const ROOT_FETCH_WATCHDOG: Duration = Duration::from_secs(15); +// Inverting these would fire the watchdog on healthy-but-slow requests and +// hide the failure path libp2p reports for itself. +const _: () = assert!( + ROOT_FETCH_WATCHDOG.as_millis() > REQ_RESP_TIMEOUT.as_millis(), + "the fetch watchdog must outlast the libp2p request timeout" +); + pub(crate) struct PendingRequest { pub(crate) attempts: u32, pub(crate) failed_peers: HashSet, diff --git a/crates/net/p2p/src/req_resp/handlers.rs b/crates/net/p2p/src/req_resp/handlers.rs index 2d3b6d81..c0afd030 100644 --- a/crates/net/p2p/src/req_resp/handlers.rs +++ b/crates/net/p2p/src/req_resp/handlers.rs @@ -285,16 +285,6 @@ 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, @@ -305,11 +295,17 @@ async fn handle_blocks_by_root_response( let received = blocks.len(); trace!(%peer, count = received, "Received BlocksByRoot response"); + // Requests carry a single root, so at most one block can answer one and + // anything else the peer sent is unsolicited. + // // 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 { + let answer = blocks + .into_iter() + .find(|block| block.message.hash_tree_root() == requested_root); + let Some(block) = answer else { debug!( %peer, received, @@ -723,24 +719,6 @@ mod tests { )]) } - /// 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] @@ -798,12 +776,4 @@ mod tests { ); 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); - } } From 3ca86574603eeae85ea814293b38e728c00290a5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Mon, 7 Sep 2026 11:03:18 -0300 Subject: [PATCH 3/6] fix(p2p): identify a watched fetch by request id, not attempt number Attempt numbers restart at 1 for every fetch cycle, so a watchdog that outlived its own cycle matched a fresh attempt for the same root and failed it. A give-up finishes ten attempts in about 2.6s while the watchdog runs for 15s, so the stale timer easily outlives the cycle that armed it, and the chain re-asks for a root as soon as another orphan referencing it arrives. Test the request id instead. Every outcome path retires the id from `outbound_requests`, so an id still present is exactly a request libp2p never reported on, and ids are unique across cycles. --- crates/net/p2p/src/lib.rs | 17 +++++++---------- crates/net/p2p/src/req_resp/handlers.rs | 6 ++---- 2 files changed, 9 insertions(+), 14 deletions(-) diff --git a/crates/net/p2p/src/lib.rs b/crates/net/p2p/src/lib.rs index 4fbaa131..8e799733 100644 --- a/crates/net/p2p/src/lib.rs +++ b/crates/net/p2p/src/lib.rs @@ -572,7 +572,6 @@ pub(crate) trait P2PProtocol: Send + Sync { 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>; @@ -614,13 +613,12 @@ impl P2PServer { 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 { + // Every outcome path retires the request id, so an id still present is + // one libp2p never reported on. This has to key off the id rather than + // the attempt number: attempts restart at 1 for each fetch cycle, so a + // watchdog that outlived its own cycle would otherwise fail an + // unrelated attempt for the same root. + if self.outbound_requests.remove(&msg.request_id).is_none() { trace!(root = %msg.root, "Block fetch settled before the watchdog fired"); return; } @@ -628,10 +626,9 @@ impl P2PServer { warn!( root = %msg.root, peer = %msg.peer, - attempt = msg.attempt, + request_id = ?msg.request_id, "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; } diff --git a/crates/net/p2p/src/req_resp/handlers.rs b/crates/net/p2p/src/req_resp/handlers.rs index c0afd030..5a59b6cd 100644 --- a/crates/net/p2p/src/req_resp/handlers.rs +++ b/crates/net/p2p/src/req_resp/handlers.rs @@ -468,14 +468,13 @@ pub async fn fetch_block_from_peer( }; // Track the request if not already tracked (new request) - let attempt = server + 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 @@ -492,7 +491,6 @@ pub async fn fetch_block_from_peer( root, peer, request_id, - attempt, }, ); From 3f67c80760852c88754af791aba2e94fb84e2107 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Mon, 7 Sep 2026 11:12:50 -0300 Subject: [PATCH 4/6] revert(p2p): drop the no-outcome fetch watchdog The watchdog covered a case I could not show to be reachable here. For libp2p to drop a request silently, `send_request_with_protocol` has to take its dial branch and the swarm has to reject that dial with `DialPeerConditionFalse`, which needs either a concurrent dial or an existing connection: - A concurrent dial resolves either way, and both outcomes drain the peer's parked requests: success through `preload_new_handler`, failure through `on_dial_failure` with a real error. That is the invariant upstream PR 6000 relies on, and it holds. - An existing connection would strand the request, but it requires the behaviour's connection map to disagree with the pool. The swarm adapter runs one task selecting between swarm events and commands, so a command never observes a half-updated swarm. The reported symptom agrees: the request-size histogram last moved an hour before the stall, so the stranding request did reach the wire. A dial-path strand never reaches the codec and would not have moved it at all. That leaves the two response paths, which need no race and are fixed directly. Also restores the req/resp config to `Default::default()`, since naming the timeout explicitly only existed to anchor the watchdog against it. --- crates/net/p2p/src/lib.rs | 69 +++---------------------- crates/net/p2p/src/req_resp/handlers.rs | 23 ++------- crates/net/p2p/src/req_resp/mod.rs | 1 - 3 files changed, 9 insertions(+), 84 deletions(-) diff --git a/crates/net/p2p/src/lib.rs b/crates/net/p2p/src/lib.rs index 8e799733..c2b04359 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, handle_fetch_failure, + fetch_block_from_peer, }, swarm_adapter::SwarmHandle, }; @@ -68,26 +68,6 @@ 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); - -// Inverting these would fire the watchdog on healthy-but-slow requests and -// hide the failure path libp2p reports for itself. -const _: () = assert!( - ROOT_FETCH_WATCHDOG.as_millis() > REQ_RESP_TIMEOUT.as_millis(), - "the fetch watchdog must outlast the libp2p request timeout" -); - pub(crate) struct PendingRequest { pub(crate) attempts: u32, pub(crate) failed_peers: HashSet, @@ -318,7 +298,7 @@ pub fn build_swarm(config: SwarmConfig) -> Result { request_response::ProtocolSupport::Full, ), ], - request_response::Config::default().with_request_timeout(REQ_RESP_TIMEOUT), + Default::default(), ); let secret_key = @@ -567,13 +547,6 @@ 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, - ) -> 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>; @@ -585,7 +558,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) @@ -596,42 +569,12 @@ impl P2PServer { trace!(%root, "Retrying block fetch after backoff"); - if !fetch_block_from_peer(self, root, ctx).await { + if !fetch_block_from_peer(self, root).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, - ) { - // Every outcome path retires the request id, so an id still present is - // one libp2p never reported on. This has to key off the id rather than - // the attempt number: attempts restart at 1 for each fetch cycle, so a - // watchdog that outlived its own cycle would otherwise fail an - // unrelated attempt for the same root. - if self.outbound_requests.remove(&msg.request_id).is_none() { - trace!(root = %msg.root, "Block fetch settled before the watchdog fired"); - return; - } - - warn!( - root = %msg.root, - peer = %msg.peer, - request_id = ?msg.request_id, - "BlocksByRoot request produced no libp2p outcome, failing it" - ); - handle_fetch_failure(self, msg.root, msg.peer, ctx).await; - } - #[send_handler] async fn handle_retry_peer_redial( &mut self, @@ -697,14 +640,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, ctx).await; + fetch_block_from_peer(self, root).await; } } diff --git a/crates/net/p2p/src/req_resp/handlers.rs b/crates/net/p2p/src/req_resp/handlers.rs index 5a59b6cd..32213bc6 100644 --- a/crates/net/p2p/src/req_resp/handlers.rs +++ b/crates/net/p2p/src/req_resp/handlers.rs @@ -19,7 +19,7 @@ use super::{ }; use crate::{ BACKOFF_MULTIPLIER, INITIAL_BACKOFF_MS, MAX_FETCH_RETRIES, MAX_SYNC_RANGE, P2PServer, - PendingRequest, PendingRequestKind, ROOT_FETCH_WATCHDOG, RangeSyncState, p2p_protocol, + PendingRequest, PendingRequestKind, RangeSyncState, p2p_protocol, req_resp::RequestedBlockRoots, }; @@ -396,11 +396,7 @@ 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, - ctx: &Context, -) -> bool { +pub async fn fetch_block_from_peer(server: &mut P2PServer, root: H256) -> bool { if server.connected_peers.is_empty() { debug!(%root, "Cannot fetch block: no connected peers"); return false; @@ -481,19 +477,6 @@ pub async fn fetch_block_from_peer( .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, - }, - ); - true } @@ -617,7 +600,7 @@ fn record_fetch_failure( /// 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( +async fn handle_fetch_failure( server: &mut P2PServer, root: H256, peer: PeerId, diff --git a/crates/net/p2p/src/req_resp/mod.rs b/crates/net/p2p/src/req_resp/mod.rs index bbd67d61..11acb79f 100644 --- a/crates/net/p2p/src/req_resp/mod.rs +++ b/crates/net/p2p/src/req_resp/mod.rs @@ -5,7 +5,6 @@ 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, From 3a9c28759f82feb5c2e6abe54448b1050b6782eb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Mon, 7 Sep 2026 11:31:41 -0300 Subject: [PATCH 5/6] chore: simplify comments --- crates/net/p2p/src/req_resp/handlers.rs | 5 ----- 1 file changed, 5 deletions(-) diff --git a/crates/net/p2p/src/req_resp/handlers.rs b/crates/net/p2p/src/req_resp/handlers.rs index 32213bc6..7f3ac87b 100644 --- a/crates/net/p2p/src/req_resp/handlers.rs +++ b/crates/net/p2p/src/req_resp/handlers.rs @@ -297,11 +297,6 @@ async fn handle_blocks_by_root_response( // Requests carry a single root, so at most one block can answer one and // anything else the peer sent is unsolicited. - // - // 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 answer = blocks .into_iter() .find(|block| block.message.hash_tree_root() == requested_root); From 1efd04af19d21e1b4910a664bb625b02a615a92d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Mon, 7 Sep 2026 12:10:58 -0300 Subject: [PATCH 6/6] refactor(p2p): fold the retry accounting back into handle_fetch_failure `record_fetch_failure` and its `FetchFailure` outcome existed to make the accounting reachable from a test without a live actor. Reading it back, the indirection cost more than it bought: the logic is a dozen lines that read fine in place. Its three tests go with it, since they called it directly. The one thing they documented that the code did not is now a comment: a failure for a root nobody is waiting on must stay a no-op, or a late or duplicate event could resurrect a root that already succeeded. --- crates/net/p2p/src/req_resp/handlers.rs | 142 ++++-------------------- 1 file changed, 23 insertions(+), 119 deletions(-) diff --git a/crates/net/p2p/src/req_resp/handlers.rs b/crates/net/p2p/src/req_resp/handlers.rs index 7f3ac87b..56913ad0 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::{HashMap, HashSet}; +use std::collections::HashSet; use ethlambda_network_api::BlockSource; use ethlambda_storage::Store; @@ -549,47 +549,6 @@ fn fail_range_request(server: &mut P2PServer, peer: &PeerId) { } } -/// 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, -) -> FetchFailure { - let Some(pending) = pending_root_requests.get_mut(&root) else { - return FetchFailure::NotPending; - }; - - pending.failed_peers.insert(peer); - let attempts = pending.attempts; - - if attempts >= MAX_FETCH_RETRIES { - pending_root_requests.remove(&root); - return FetchFailure::GaveUp { attempts }; - } - - pending.attempts += 1; - let backoff_ms = INITIAL_BACKOFF_MS * BACKOFF_MULTIPLIER.pow(attempts - 1); - - 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 @@ -601,16 +560,29 @@ async fn handle_fetch_failure( 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 }); - } + // A root nobody is waiting on means a late or duplicate failure, which + // must not resurrect a root that already succeeded. + let Some(pending) = server.pending_root_requests.get_mut(&root) else { + return; + }; + + pending.failed_peers.insert(peer); + + 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; } + + 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; + + send_after(backoff, ctx.clone(), p2p_protocol::RetryBlockFetch { root }); } #[cfg(test)] @@ -684,72 +656,4 @@ 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 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)); - } }