Fuel/Request_Response v0.0.2: More meaningful error messages

CHANGELOG.md

@@ -20,6 +20,9 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 - [2362](https://github.com/FuelLabs/fuel-core/pull/2362): Added a new request_response protocol version `/fuel/req_res/0.0.2`. In comparison with `/fuel/req/0.0.1`, which returns an empty response when a request cannot be fulfilled, this version returns more meaningful error codes. Nodes still support the version `0.0.1` of the protocol to guarantee backward compatibility with fuel-core nodes. Empty responses received from nodes using the old protocol `/fuel/req/0.0.1` are automatically converted into an error `ProtocolV1EmptyResponse` with error code 0, which is also the only error code implemented. More specific error codes will be added in the future.
 - [2386](https://github.com/FuelLabs/fuel-core/pull/2386): Add a flag to define the maximum number of file descriptors that RocksDB can use. By default it's half of the OS limit.
 
+### Changed
+- [2377](https://github.com/FuelLabs/fuel-core/pull/2377): Add more errors that can be returned as responses when using protocol `/fuel/req_res/0.0.2`. The errors supported are `ProtocolV1EmptyResponse` (status code `0`) for converting empty responses sent via protocol `/fuel/req_res/0.0.1`, `RequestedRangeTooLarge`(status code `1`) if the client requests a range of objects such as sealed block headers or transactions too large, `Timeout` (status code `2`) if the remote peer takes too long to fulfill a request, or `SyncProcessorOutOfCapacity` if the remote peer is fulfilling too many requests concurrently.
+
 ## [Version 0.40.0]
 
 ### Added

crates/services/p2p/src/codecs/postcard.rs

@@ -180,16 +180,17 @@ impl NetworkCodec for PostcardCodec {
     fn get_req_res_protocols(
         &self,
     ) -> impl Iterator<Item = <Self as request_response::Codec>::Protocol> {
-        // TODO: Iterating over versions in reverse order should prefer
+        // TODO: https://github.com/FuelLabs/fuel-core/issues/2374
+        // Iterating over versions in reverse order should prefer
         // peers to use V2 over V1 for exchanging messages. However, this is
-        // not guaranteed by the specs for the `request_response` protocol.
+        // not guaranteed by the specs for the `request_response` protocol,
+        // and it should be tested.
         PostcardProtocol::iter().rev()
     }
 }
 
-#[derive(Debug, Default, Clone, EnumIter)]
+#[derive(Debug, Clone, EnumIter)]
 pub enum PostcardProtocol {
-    #[default]
     V1,
     V2,
 }
@@ -206,7 +207,6 @@ impl AsRef<str> for PostcardProtocol {
 #[cfg(test)]
 #[allow(non_snake_case)]
 mod tests {
-
     use fuel_core_types::blockchain::SealedBlockHeader;
     use request_response::Codec as _;
 
@@ -310,10 +310,8 @@ mod tests {
     async fn codec__serialzation_roundtrip_using_v1_on_error_response_returns_predefined_error_code(
     ) {
         // Given
-        // TODO: https://github.com/FuelLabs/fuel-core/issues/1311
-        // Change this to a different ResponseMessageErrorCode once these have been implemented.
         let response = V2ResponseMessage::SealedHeaders(Err(
-            ResponseMessageErrorCode::ProtocolV1EmptyResponse,
+            ResponseMessageErrorCode::RequestedRangeTooLarge,
         ));
         let mut codec = PostcardCodec::new(1024);
         let mut buf = Vec::with_capacity(1024);

crates/services/p2p/src/p2p_service.rs

@@ -677,7 +677,7 @@ impl FuelP2PService {
                             V2ResponseMessage::SealedHeaders(v) => {
                                 // TODO: https://github.com/FuelLabs/fuel-core/issues/1311
                                 // Change type of ResponseSender and remove the .ok() here
-                                c.send((peer, Ok(v.ok()))).is_ok()
+                                c.send((peer, Ok(v))).is_ok()
                             }
                             _ => {
                                 warn!(
@@ -689,7 +689,7 @@ impl FuelP2PService {
                         },
                         ResponseSender::Transactions(c) => match response {
                             V2ResponseMessage::Transactions(v) => {
-                                c.send((peer, Ok(v.ok()))).is_ok()
+                                c.send((peer, Ok(v))).is_ok()
                             }
                             _ => {
                                 warn!(
@@ -701,7 +701,7 @@ impl FuelP2PService {
                         },
                         ResponseSender::TxPoolAllTransactionsIds(c) => match response {
                             V2ResponseMessage::TxPoolAllTransactionsIds(v) => {
-                                c.send((peer, Ok(v.ok()))).is_ok()
+                                c.send((peer, Ok(v))).is_ok()
                             }
                             _ => {
                                 warn!(
@@ -713,7 +713,7 @@ impl FuelP2PService {
                         },
                         ResponseSender::TxPoolFullTransactions(c) => match response {
                             V2ResponseMessage::TxPoolFullTransactions(v) => {
-                                c.send((peer, Ok(v.ok()))).is_ok()
+                                c.send((peer, Ok(v))).is_ok()
                             }
                             _ => {
                                 warn!(
@@ -1717,7 +1717,7 @@ mod tests {
                                         tokio::spawn(async move {
                                             let response_message = rx_orchestrator.await;
 
-                                            if let Ok((_, Ok(Some(transactions)))) = response_message {
+                                            if let Ok((_, Ok(Ok(transactions)))) = response_message {
                                                 let check = transactions.len() == 1 && transactions[0].0.len() == 5;
                                                 let _ = tx_test_end.send(check).await;
                                             } else {
@@ -1733,7 +1733,7 @@ mod tests {
                                         tokio::spawn(async move {
                                             let response_message = rx_orchestrator.await;
 
-                                            if let Ok((_, Ok(Some(transaction_ids)))) = response_message {
+                                            if let Ok((_, Ok(Ok(transaction_ids)))) = response_message {
                                                 let tx_ids: Vec<TxId> = (0..5).map(|_| Transaction::default_test_tx().id(&ChainId::new(1))).collect();
                                                 let check = transaction_ids.len() == 5 && transaction_ids.iter().zip(tx_ids.iter()).all(|(a, b)| a == b);
                                                 let _ = tx_test_end.send(check).await;
@@ -1750,7 +1750,7 @@ mod tests {
                                         tokio::spawn(async move {
                                             let response_message = rx_orchestrator.await;
 
-                                            if let Ok((_, Ok(Some(transactions)))) = response_message {
+                                            if let Ok((_, Ok(Ok(transactions)))) = response_message {
                                                 let txs: Vec<Option<NetworkableTransactionPool>> = tx_ids.iter().enumerate().map(|(i, _)| {
                                                     if i == 0 {
                                                         None

crates/services/p2p/src/request_response/messages.rs

@@ -38,6 +38,12 @@ pub enum ResponseMessageErrorCode {
     /// The peer sent an empty response using protocol `/fuel/req_res/0.0.1`
     #[error("Empty response sent by peer using legacy protocol /fuel/req_res/0.0.1")]
     ProtocolV1EmptyResponse = 0,
+    #[error("The requested range is too large")]
+    RequestedRangeTooLarge = 1,
+    #[error("Timeout while processing request")]
+    Timeout = 2,
+    #[error("Sync processor is out of capacity")]
+    SyncProcessorOutOfCapacity = 3,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -107,10 +113,14 @@ pub type OnResponse<T> = oneshot::Sender<(PeerId, Result<T, ResponseError>)>;
 
 #[derive(Debug)]
 pub enum ResponseSender {
-    SealedHeaders(OnResponse<Option<Vec<SealedBlockHeader>>>),
-    Transactions(OnResponse<Option<Vec<Transactions>>>),
-    TxPoolAllTransactionsIds(OnResponse<Option<Vec<TxId>>>),
-    TxPoolFullTransactions(OnResponse<Option<Vec<Option<NetworkableTransactionPool>>>>),
+    SealedHeaders(OnResponse<Result<Vec<SealedBlockHeader>, ResponseMessageErrorCode>>),
+    Transactions(OnResponse<Result<Vec<Transactions>, ResponseMessageErrorCode>>),
+    TxPoolAllTransactionsIds(OnResponse<Result<Vec<TxId>, ResponseMessageErrorCode>>),
+    TxPoolFullTransactions(
+        OnResponse<
+            Result<Vec<Option<NetworkableTransactionPool>>, ResponseMessageErrorCode>,
+        >,
+    ),
 }
 
 #[derive(Debug, Error)]

crates/services/p2p/src/service.rs

@@ -113,21 +113,23 @@ pub enum TaskRequest {
     },
     GetSealedHeaders {
         block_height_range: Range<u32>,
-        channel: OnResponse<Option<Vec<SealedBlockHeader>>>,
+        channel: OnResponse<Result<Vec<SealedBlockHeader>, ResponseMessageErrorCode>>,
     },
     GetTransactions {
         block_height_range: Range<u32>,
         from_peer: PeerId,
-        channel: OnResponse<Option<Vec<Transactions>>>,
+        channel: OnResponse<Result<Vec<Transactions>, ResponseMessageErrorCode>>,
     },
     TxPoolGetAllTxIds {
         from_peer: PeerId,
-        channel: OnResponse<Option<Vec<TxId>>>,
+        channel: OnResponse<Result<Vec<TxId>, ResponseMessageErrorCode>>,
     },
     TxPoolGetFullTransactions {
         tx_ids: Vec<TxId>,
         from_peer: PeerId,
-        channel: OnResponse<Option<Vec<Option<NetworkableTransactionPool>>>>,
+        channel: OnResponse<
+            Result<Vec<Option<NetworkableTransactionPool>>, ResponseMessageErrorCode>,
+        >,
     },
     // Responds back to the p2p network
     RespondWithGossipsubMessageReport((GossipsubMessageInfo, GossipsubMessageAcceptance)),
@@ -552,11 +554,9 @@ where
             tracing::error!(
                 requested_length = range.len(),
                 max_len,
-                "Requested range is too big"
+                "Requested range is too large"
             );
-            // TODO: https://github.com/FuelLabs/fuel-core/issues/1311
-            // Return helpful error message to requester.
-            let response = Err(ResponseMessageErrorCode::ProtocolV1EmptyResponse);
+            let response = Err(ResponseMessageErrorCode::RequestedRangeTooLarge);
             let _ = self
                 .p2p_service
                 .send_response_msg(request_id, response_sender(response));
@@ -570,12 +570,10 @@ where
                 return;
             }
 
-            // TODO: https://github.com/FuelLabs/fuel-core/issues/1311
-            // Add new error code
             let response = db_lookup(&view, range.clone())
                 .ok()
                 .flatten()
-                .ok_or(ResponseMessageErrorCode::ProtocolV1EmptyResponse);
+                .ok_or(ResponseMessageErrorCode::Timeout);
 
             let _ = response_channel
                 .try_send(task_request(response, request_id))
@@ -585,7 +583,7 @@ where
         // TODO: https://github.com/FuelLabs/fuel-core/issues/1311
         // Handle error cases and return meaningful status codes
         if result.is_err() {
-            let err = Err(ResponseMessageErrorCode::ProtocolV1EmptyResponse);
+            let err = Err(ResponseMessageErrorCode::SyncProcessorOutOfCapacity);
             let _ = self
                 .p2p_service
                 .send_response_msg(request_id, response_sender(err));
@@ -667,9 +665,7 @@ where
         });
 
         if result.is_err() {
-            // TODO: https://github.com/FuelLabs/fuel-core/issues/1311
-            // Return better error code
-            let res = Err(ResponseMessageErrorCode::ProtocolV1EmptyResponse);
+            let res = Err(ResponseMessageErrorCode::SyncProcessorOutOfCapacity);
             let _ = self
                 .p2p_service
                 .send_response_msg(request_id, response_sender(res));
@@ -700,13 +696,11 @@ where
         tx_ids: Vec<TxId>,
         request_id: InboundRequestId,
     ) -> anyhow::Result<()> {
-        // TODO: https://github.com/FuelLabs/fuel-core/issues/1311
-        // Return helpful error message to requester.
         if tx_ids.len() > self.max_txs_per_request {
             self.p2p_service.send_response_msg(
                 request_id,
                 V2ResponseMessage::TxPoolFullTransactions(Err(
-                    ResponseMessageErrorCode::ProtocolV1EmptyResponse,
+                    ResponseMessageErrorCode::RequestedRangeTooLarge,
                 )),
             )?;
             return Ok(());
@@ -1040,8 +1034,17 @@ impl SharedState {
 
         let (peer_id, response) = receiver.await.map_err(|e| anyhow!("{e}"))?;
 
-        let data = response.map_err(|e| anyhow!("Invalid response from peer {e:?}"))?;
-        Ok((peer_id.to_bytes(), data))
+        let data = match response {
+            Err(request_response_protocol_error) => Err(anyhow!(
+                "Invalid response from peer {request_response_protocol_error:?}"
+            )),
+            Ok(Err(response_error_code)) => {
+                warn!("Peer {peer_id:?} failed to respond with sealed headers: {response_error_code:?}");
+                Ok(None)
+            }
+            Ok(Ok(headers)) => Ok(Some(headers)),
+        };
+        data.map(|data| (peer_id.to_bytes(), data))
     }
 
     pub async fn get_transactions_from_peer(
@@ -1067,7 +1070,18 @@ impl SharedState {
             "Bug: response from non-requested peer"
         );
 
-        response.map_err(|e| anyhow!("Invalid response from peer {e:?}"))
+        match response {
+            Err(request_response_protocol_error) => Err(anyhow!(
+                "Invalid response from peer {request_response_protocol_error:?}"
+            )),
+            Ok(Err(response_error_code)) => {
+                warn!(
+                    "Peer {peer_id:?} failed to respond with transactions: {response_error_code:?}"
+                );
+                Ok(None)
+            }
+            Ok(Ok(txs)) => Ok(Some(txs)),
+        }
     }
 
     pub async fn get_all_transactions_ids_from_peer(
@@ -1091,11 +1105,11 @@ impl SharedState {
             "Bug: response from non-requested peer"
         );
 
-        let Some(txs) =
-            response.map_err(|e| anyhow!("Invalid response from peer {e:?}"))?
-        else {
-            return Ok(vec![]);
-        };
+        let response =
+            response.map_err(|e| anyhow!("Invalid response from peer {e:?}"))?;
+
+        let txs = response.inspect_err(|e| { warn!("Peer {peer_id:?} could not response to request to get all transactions ids: {e:?}"); } ).unwrap_or_default();
+
         if txs.len() > self.max_txs_per_request {
             return Err(anyhow!("Too many transactions requested: {}", txs.len()));
         }
@@ -1124,11 +1138,10 @@ impl SharedState {
             "Bug: response from non-requested peer"
         );
 
-        let Some(txs) =
-            response.map_err(|e| anyhow!("Invalid response from peer {e:?}"))?
-        else {
-            return Ok(vec![]);
-        };
+        let response =
+            response.map_err(|e| anyhow!("Invalid response from peer {e:?}"))?;
+        let txs = response.inspect_err(|e| { warn!("Peer {peer_id:?} could not response to request to get full transactions: {e:?}"); } ).unwrap_or_default();
+
         if txs.len() > self.max_txs_per_request {
             return Err(anyhow!("Too many transactions requested: {}", txs.len()));
         }