Error from RPC send_response when request doesn't exist on the active inbound requests (#7663)

Squashed commit of the following:

commit 0ce690c
Author: João Oliveira <[email protected]>
Date:   Fri Jun 27 22:50:11 2025 +0100

    Error from RPC `send_response`

    when request doesn't exist on the active inbound requests.
    And handle the error from the main service to check if it may be a data race or a critical bug

commit 522e00f
Author: Jimmy Chen <[email protected]>
Date:   Sat Jun 28 05:01:46 2025 +1000

    Fix incorrect `waker` update condition (#7656)

    This bug was first found and partially fixed by @VolodymyrBg in #7317 - this PR applies the same fix everywhere else.

    The old logic updated the waker when it already matched the context, and did nothing when it was stale:

    ```rust
    if waker.will_wake(cx.waker()) {
    self.waker = Some(cx.waker().clone());
    }
    ```

    This is the wrong way around. We only want to update the waker if it doesn't match the current context:

    ```rust
    if !waker.will_wake(cx.waker()) {
    self.waker = Some(cx.waker().clone());
    }
    ```

    I don't think we've ever noticed any issues, but it’s a subtle bug that could lead to missed wakeups.

commit 83cad25
Author: Jimmy Chen <[email protected]>
Date:   Sat Jun 28 04:21:17 2025 +1000

    Fix Rust 1.88 clippy errors & execution engine tests (#7657)

    Fix Rust 1.88 clippy errors.

commit 9b1f3ed
Author: Pawan Dhananjay <[email protected]>
Date:   Thu Jun 26 17:26:38 2025 -0700

    Add gossip check (#7652)

    N/A

      Add an additional gossip condition.

commit a0a6b93
Author: Daniel Knopik <[email protected]>
Date:   Wed Jun 25 08:22:24 2025 +0200

    Do not compute sync selection proofs for the sync duty at the current slot (#7551)

commit 8e3c5d1
Author: chonghe <[email protected]>
Date:   Wed Jun 25 13:33:17 2025 +0800

    Rust 1.89 compiler lint fix (#7644)

    Fix lints for Rust 1.89 beta compiler

commit 56b2d4b
Author: Eitan Seri-Levi <[email protected]>
Date:   Tue Jun 24 09:29:10 2025 +0300

    Remove instrumenting log level  (#7636)

    #7155

      Theres some additional places we set instrumenting log levels that wasn't covered in #7620

commit fd643c3
Author: Michael Sproul <[email protected]>
Date:   Mon Jun 23 23:11:46 2025 +1000

    Un-ignore EF test for v1.6.0-alpha.1 (#7632)

    Closes:

    - #7547

      Run the test that was previously ignored when we were between spec versions.

commit cef04ee
Author: chonghe <[email protected]>
Date:   Mon Jun 23 16:37:49 2025 +0800

    Implement `validator_identities` Beacon API endpoint (#7462)

    * #7442

Author: michaelsproul

beacon_node/lighthouse_network/src/rpc/mod.rs

@@ -17,7 +17,7 @@ use std::marker::PhantomData;
 use std::sync::Arc;
 use std::task::{Context, Poll};
 use std::time::Duration;
-use tracing::{debug, error, instrument, trace};
+use tracing::{debug, instrument, trace};
 use types::{EthSpec, ForkContext};
 
 pub(crate) use handler::{HandlerErr, HandlerEvent};
@@ -199,8 +199,7 @@ impl<Id: ReqId, E: EthSpec> RPC<Id, E> {
     }
 
     /// Sends an RPC response.
-    ///
-    /// The peer must be connected for this to succeed.
+    /// Returns an `Err` if the request does exist in the active inbound requests list.
     #[instrument(parent = None,
         level = "trace",
         fields(service = "libp2p_rpc"),
@@ -212,11 +211,10 @@ impl<Id: ReqId, E: EthSpec> RPC<Id, E> {
         peer_id: PeerId,
         request_id: InboundRequestId,
         response: RpcResponse<E>,
-    ) {
+    ) -> Result<(), RpcResponse<E>> {
         let Some((_peer_id, request_type)) = self.active_inbound_requests.remove(&request_id)
         else {
-            error!(%peer_id, ?request_id, %response, "Request not found in active_inbound_requests. Response not sent");
-            return;
+            return Err(response);
         };
 
         // Add the request back to active requests if the response is `Success` and requires stream
@@ -229,6 +227,7 @@ impl<Id: ReqId, E: EthSpec> RPC<Id, E> {
         }
 
         self.send_response_inner(peer_id, request_type.protocol(), request_id, response);
+        Ok(())
     }
 
     fn send_response_inner(
@@ -506,7 +505,8 @@ where
                         RpcResponse::Success(RpcSuccessResponse::Pong(Ping {
                             data: self.seq_number,
                         })),
-                    );
+                    )
+                    .expect("Request to exist");
                 }
 
                 self.events.push(ToSwarm::GenerateEvent(RPCMessage {

beacon_node/lighthouse_network/src/service/mod.rs

@@ -11,8 +11,7 @@ use crate::peer_manager::{MIN_OUTBOUND_ONLY_FACTOR, PEER_EXCESS_FACTOR, PRIORITY
 use crate::rpc::methods::MetadataRequest;
 use crate::rpc::{
     GoodbyeReason, HandlerErr, InboundRequestId, NetworkParams, Protocol, RPCError, RPCMessage,
-    RPCReceived, RequestType, ResponseTermination, RpcErrorResponse, RpcResponse,
-    RpcSuccessResponse, RPC,
+    RPCReceived, RequestType, ResponseTermination, RpcResponse, RpcSuccessResponse, RPC,
 };
 use crate::types::{
     all_topics_at_fork, core_topics_to_subscribe, is_fork_non_core_topic, subnet_from_topic_hash,
@@ -39,7 +38,7 @@ use std::path::PathBuf;
 use std::pin::Pin;
 use std::sync::Arc;
 use std::time::Duration;
-use tracing::{debug, info, instrument, trace, warn};
+use tracing::{debug, error, info, instrument, trace, warn};
 use types::{
     consts::altair::SYNC_COMMITTEE_SUBNET_COUNT, EnrForkId, EthSpec, ForkContext, Slot, SubnetId,
 };
@@ -1146,35 +1145,22 @@ impl<E: EthSpec> Network<E> {
         name = "libp2p",
         skip_all
     )]
-    pub fn send_response(
+    pub fn send_response<T: Into<RpcResponse<E>>>(
         &mut self,
         peer_id: PeerId,
         inbound_request_id: InboundRequestId,
-        response: Response<E>,
+        response: T,
     ) {
-        self.eth2_rpc_mut()
-            .send_response(peer_id, inbound_request_id, response.into())
-    }
-
-    /// Inform the peer that their request produced an error.
-    #[instrument(parent = None,
-        level = "trace",
-        fields(service = "libp2p"),
-        name = "libp2p",
-        skip_all
-    )]
-    pub fn send_error_response(
-        &mut self,
-        peer_id: PeerId,
-        inbound_request_id: InboundRequestId,
-        error: RpcErrorResponse,
-        reason: String,
-    ) {
-        self.eth2_rpc_mut().send_response(
-            peer_id,
-            inbound_request_id,
-            RpcResponse::Error(error, reason.into()),
-        )
+        if let Err(response) =
+            self.eth2_rpc_mut()
+                .send_response(peer_id, inbound_request_id, response.into())
+        {
+            if self.network_globals.peers.read().is_connected(&peer_id) {
+                error!(%peer_id, ?inbound_request_id, %response,
+                    "Request not found in RPC active requests while peer is still connected"
+                );
+            }
+        }
     }
 
     /* Peer management functions */
@@ -1460,19 +1446,6 @@ impl<E: EthSpec> Network<E> {
         name = "libp2p",
         skip_all
     )]
-    fn send_meta_data_response(
-        &mut self,
-        _req: MetadataRequest<E>,
-        inbound_request_id: InboundRequestId,
-        peer_id: PeerId,
-    ) {
-        let metadata = self.network_globals.local_metadata.read().clone();
-        // The encoder is responsible for sending the negotiated version of the metadata
-        let event = RpcResponse::Success(RpcSuccessResponse::MetaData(Arc::new(metadata)));
-        self.eth2_rpc_mut()
-            .send_response(peer_id, inbound_request_id, event);
-    }
-
     // RPC Propagation methods
     /// Queues the response to be sent upwards as long at it was requested outside the Behaviour.
     #[must_use = "return the response"]
@@ -1760,9 +1733,13 @@ impl<E: EthSpec> Network<E> {
                         self.peer_manager_mut().ping_request(&peer_id, ping.data);
                         None
                     }
-                    RequestType::MetaData(req) => {
+                    RequestType::MetaData(_req) => {
                         // send the requested meta-data
-                        self.send_meta_data_response(req, inbound_request_id, peer_id);
+                        let metadata = self.network_globals.local_metadata.read().clone();
+                        // The encoder is responsible for sending the negotiated version of the metadata
+                        let response =
+                            RpcResponse::Success(RpcSuccessResponse::MetaData(Arc::new(metadata)));
+                        self.send_response(peer_id, inbound_request_id, response);
                         None
                     }
                     RequestType::Goodbye(reason) => {

beacon_node/network/src/service.rs

@@ -11,6 +11,7 @@ use futures::channel::mpsc::Sender;
 use futures::future::OptionFuture;
 use futures::prelude::*;
 
+use lighthouse_network::rpc::methods::RpcResponse;
 use lighthouse_network::rpc::InboundRequestId;
 use lighthouse_network::rpc::RequestType;
 use lighthouse_network::service::Network;
@@ -627,10 +628,11 @@ impl<T: BeaconChainTypes> NetworkService<T> {
                 error,
                 inbound_request_id,
                 reason,
-            } => {
-                self.libp2p
-                    .send_error_response(peer_id, inbound_request_id, error, reason);
-            }
+            } => self.libp2p.send_response(
+                peer_id,
+                inbound_request_id,
+                RpcResponse::Error(error, reason.into()),
+            ),
             NetworkMessage::ValidationResult {
                 propagation_source,
                 message_id,