feat(net): add BAL requirement to block access list requests

crates/net/network/src/fetch/client.rs

@@ -6,7 +6,7 @@ use futures::{future, future::Either};
 use reth_eth_wire::{BlockAccessLists, EthNetworkPrimitives, NetworkPrimitives};
 use reth_network_api::test_utils::PeersHandle;
 use reth_network_p2p::{
-    block_access_lists::client::BlockAccessListsClient,
+    block_access_lists::client::{BalRequirement, BlockAccessListsClient},
     bodies::client::{BodiesClient, BodiesFut},
     download::DownloadClient,
     error::{PeerRequestResult, RequestError},
@@ -135,11 +135,29 @@ impl<N: NetworkPrimitives> BlockAccessListsClient for FetchClient<N> {
         &self,
         hashes: Vec<B256>,
         priority: Priority,
+    ) -> Self::Output {
+        self.get_block_access_lists_with_priority_and_requirement(
+            hashes,
+            priority,
+            BalRequirement::Mandatory,
+        )
+    }
+
+    fn get_block_access_lists_with_priority_and_requirement(
+        &self,
+        hashes: Vec<B256>,
+        priority: Priority,
+        requirement: BalRequirement,
     ) -> Self::Output {
         let (response, rx) = oneshot::channel();
         if self
             .request_tx
-            .send(DownloadRequest::GetBlockAccessLists { request: hashes, response, priority })
+            .send(DownloadRequest::GetBlockAccessLists {
+                request: hashes,
+                response,
+                priority,
+                requirement,
+            })
             .is_ok()
         {
             Box::pin(FlattenedResponse::from(rx))

crates/net/network/src/fetch/mod.rs

@@ -13,6 +13,7 @@ use reth_eth_wire::{
 };
 use reth_network_api::test_utils::PeersHandle;
 use reth_network_p2p::{
+    block_access_lists::client::BalRequirement,
     error::{EthResponseValidator, PeerRequestResult, RequestError, RequestResult},
     headers::client::HeadersRequest,
     priority::Priority,
@@ -195,28 +196,39 @@ impl<N: NetworkPrimitives> StateFetcher<N> {
         Some(*best_peer.0)
     }
 
+    /// Returns whether any connected peer can serve BAL requests.
+    fn has_bal_capable_peer(&self) -> bool {
+        self.peers.values().any(|peer| {
+            !matches!(peer.state, PeerState::Closing) &&
+                peer.capabilities.supports_eth_at_least(&EthVersion::Eth71)
+        })
+    }
+
     /// Returns the next action to return
     fn poll_action(&mut self) -> PollAction {
-        // we only check and not pop here since we don't know yet whether a peer is available.
-        if self.queued_requests.is_empty() {
-            return PollAction::NoRequests
-        }
+        loop {
+            // we only check and not pop here since we don't know yet whether a peer is available.
+            if self.queued_requests.is_empty() {
+                return PollAction::NoRequests
+            }
 
-        if self.peers.is_empty() {
-            return PollAction::NoPeersAvailable
-        }
+            let request = self.queued_requests.pop_front().expect("not empty");
+            let Some(peer_id) = self.next_best_peer(request.best_peer_requirements()) else {
+                if request.should_complete_without_capable_peer(self.has_bal_capable_peer()) {
+                    request.send_err_response(RequestError::UnsupportedCapability);
+                    continue
+                }
 
-        let request = self.queued_requests.pop_front().expect("not empty");
-        let Some(peer_id) = self.next_best_peer(request.best_peer_requirements()) else {
-            // no peer matches this request's requirements; requeue at the back so other
-            // queued requests get a chance on the next poll instead of head-of-line blocking.
-            self.queued_requests.push_back(request);
-            return PollAction::NoPeersAvailable
-        };
+                // no peer matches this request's requirements; requeue at the back so other
+                // queued requests get a chance on the next poll instead of head-of-line blocking.
+                self.queued_requests.push_back(request);
+                return PollAction::NoPeersAvailable
+            };
 
-        let request = self.prepare_block_request(peer_id, request);
+            let request = self.prepare_block_request(peer_id, request);
 
-        PollAction::Ready(FetchAction::BlockRequest { peer_id, request })
+            return PollAction::Ready(FetchAction::BlockRequest { peer_id, request })
+        }
     }
 
     /// Advance the state the syncer
@@ -602,6 +614,7 @@ pub(crate) enum DownloadRequest<N: NetworkPrimitives> {
         request: Vec<B256>,
         response: oneshot::Sender<PeerRequestResult<BlockAccessLists>>,
         priority: Priority,
+        requirement: BalRequirement,
     },
     /// Download receipts for the given block hashes and send response through channel
     GetReceipts {
@@ -639,6 +652,22 @@ impl<N: NetworkPrimitives> DownloadRequest<N> {
         self.get_priority().is_normal()
     }
 
+    /// Returns whether the request may complete locally when no capable peer exists.
+    const fn should_complete_without_capable_peer(&self, has_bal_capable_peer: bool) -> bool {
+        matches!(self, Self::GetBlockAccessLists { requirement: BalRequirement::Optional, .. }) &&
+            !has_bal_capable_peer
+    }
+
+    /// Sends an error response to the waiting caller.
+    fn send_err_response(self, err: RequestError) {
+        let _ = match self {
+            Self::GetBlockHeaders { response, .. } => response.send(Err(err)).ok(),
+            Self::GetBlockBodies { response, .. } => response.send(Err(err)).ok(),
+            Self::GetBlockAccessLists { response, .. } => response.send(Err(err)).ok(),
+            Self::GetReceipts { response, .. } => response.send(Err(err)).ok(),
+        };
+    }
+
     /// Returns the best peer requirements for this request.
     fn best_peer_requirements(&self) -> BestPeerRequirements {
         match self {
@@ -1541,6 +1570,7 @@ mod tests {
             request: vec![],
             response: tx,
             priority: Priority::Normal,
+            requirement: BalRequirement::Mandatory,
         });
 
         let waker = noop_waker();
@@ -1583,4 +1613,76 @@ mod tests {
             assert_eq!(peer_id, peer_71);
         }
     }
+
+    #[tokio::test]
+    async fn test_optional_bal_request_completes_without_capable_peer() {
+        use futures::task::noop_waker;
+        use std::task::{Context, Poll};
+
+        let manager = PeersManager::new(PeersConfig::default());
+        let mut fetcher =
+            StateFetcher::<EthNetworkPrimitives>::new(manager.handle(), Default::default());
+
+        let peer_old = B512::random();
+        let caps_old = Arc::new(Capabilities::new(vec![]));
+        fetcher.new_active_peer(
+            peer_old,
+            B256::random(),
+            100,
+            caps_old,
+            Arc::new(AtomicU64::new(10)),
+            None,
+        );
+
+        let (tx, rx) = oneshot::channel();
+        fetcher.queued_requests.push_back(DownloadRequest::GetBlockAccessLists {
+            request: vec![],
+            response: tx,
+            priority: Priority::Normal,
+            requirement: BalRequirement::Optional,
+        });
+
+        let waker = noop_waker();
+        let mut cx = Context::from_waker(&waker);
+
+        assert!(matches!(fetcher.poll(&mut cx), Poll::Pending));
+        assert!(fetcher.queued_requests.is_empty());
+        assert_eq!(rx.await.unwrap().unwrap_err(), RequestError::UnsupportedCapability);
+    }
+
+    #[tokio::test]
+    async fn test_optional_bal_request_waits_for_busy_capable_peer() {
+        use futures::task::noop_waker;
+        use std::task::{Context, Poll};
+
+        let manager = PeersManager::new(PeersConfig::default());
+        let mut fetcher =
+            StateFetcher::<EthNetworkPrimitives>::new(manager.handle(), Default::default());
+
+        let peer_71 = B512::random();
+        let caps_71 = Arc::new(Capabilities::from(vec![Capability::new("eth".into(), 71)]));
+        fetcher.new_active_peer(
+            peer_71,
+            B256::random(),
+            100,
+            caps_71,
+            Arc::new(AtomicU64::new(10)),
+            None,
+        );
+        fetcher.peers.get_mut(&peer_71).expect("peer exists").state = PeerState::GetBlockHeaders;
+
+        let (tx, _rx) = oneshot::channel();
+        fetcher.queued_requests.push_back(DownloadRequest::GetBlockAccessLists {
+            request: vec![],
+            response: tx,
+            priority: Priority::Normal,
+            requirement: BalRequirement::Optional,
+        });
+
+        let waker = noop_waker();
+        let mut cx = Context::from_waker(&waker);
+
+        assert!(matches!(fetcher.poll(&mut cx), Poll::Pending));
+        assert_eq!(fetcher.queued_requests.len(), 1);
+    }
 }

crates/net/p2p/src/block_access_lists/client.rs

@@ -4,6 +4,17 @@ use auto_impl::auto_impl;
 use futures::Future;
 use reth_eth_wire_types::BlockAccessLists;
 
+/// Controls whether a BAL request must wait for a capable peer or may complete early when none are
+/// available.
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+pub enum BalRequirement {
+    /// Keep waiting until an eth/71-capable peer is available.
+    #[default]
+    Mandatory,
+    /// Return early if no connected peer can serve BALs.
+    Optional,
+}
+
 /// A client capable of downloading block access lists.
 #[auto_impl(&, Arc, Box)]
 pub trait BlockAccessListsClient: DownloadClient {
@@ -12,13 +23,44 @@ pub trait BlockAccessListsClient: DownloadClient {
 
     ///  Fetches the block access lists for given hashes.
     fn get_block_access_lists(&self, hashes: Vec<B256>) -> Self::Output {
-        self.get_block_access_lists_with_priority(hashes, Priority::Normal)
+        self.get_block_access_lists_with_priority_and_requirement(
+            hashes,
+            Priority::Normal,
+            BalRequirement::Mandatory,
+        )
+    }
+
+    /// Fetches the block access lists for given hashes with the requested BAL availability policy.
+    fn get_block_access_lists_with_requirement(
+        &self,
+        hashes: Vec<B256>,
+        requirement: BalRequirement,
+    ) -> Self::Output {
+        self.get_block_access_lists_with_priority_and_requirement(
+            hashes,
+            Priority::Normal,
+            requirement,
+        )
     }
 
     ///  Fetches the block access lists for given hashes with priority
     fn get_block_access_lists_with_priority(
         &self,
         hashes: Vec<B256>,
         priority: Priority,
+    ) -> Self::Output {
+        self.get_block_access_lists_with_priority_and_requirement(
+            hashes,
+            priority,
+            BalRequirement::Mandatory,
+        )
+    }
+
+    /// Fetches the block access lists for given hashes with priority and BAL availability policy.
+    fn get_block_access_lists_with_priority_and_requirement(
+        &self,
+        hashes: Vec<B256>,
+        priority: Priority,
+        requirement: BalRequirement,
     ) -> Self::Output;
 }

crates/net/p2p/src/error.rs

@@ -57,8 +57,8 @@ impl<H: BlockHeader> EthResponseValidator for RequestResult<Vec<H>> {
     /// [`RequestError::ConnectionDropped`] should be ignored here because this is already handled
     /// when the dropped connection is handled.
     ///
-    /// [`RequestError::UnsupportedCapability`] is not used yet because we only support active
-    /// session for eth protocol.
+    /// [`RequestError::UnsupportedCapability`] is also used for locally rejected optional requests,
+    /// which should not affect peer reputation.
     fn reputation_change_err(&self) -> Option<ReputationChangeKind> {
         if let Err(err) = self {
             match err {

crates/net/p2p/src/full_block.rs

@@ -1110,10 +1110,11 @@ mod tests {
     impl BlockAccessListsClient for FullBlockWithAccessListsClient {
         type Output = futures::future::Ready<PeerRequestResult<BlockAccessLists>>;
 
-        fn get_block_access_lists_with_priority(
+        fn get_block_access_lists_with_priority_and_requirement(
             &self,
             hashes: Vec<B256>,
             _priority: Priority,
+            _requirement: crate::block_access_lists::client::BalRequirement,
         ) -> Self::Output {
             self.access_list_requests.fetch_add(1, Ordering::SeqCst);
 

crates/net/p2p/src/lib.rs

@@ -54,7 +54,7 @@ pub mod snap;
 #[cfg(any(test, feature = "test-utils"))]
 pub mod test_utils;
 
-pub use block_access_lists::client::BlockAccessListsClient;
+pub use block_access_lists::client::{BalRequirement, BlockAccessListsClient};
 pub use bodies::client::BodiesClient;
 pub use headers::client::HeadersClient;
 pub use receipts::client::ReceiptsClient;