Merge c7d1f61817 into 9f952ac2ed

bin/reth/src/block_ingest.rs

@@ -1,11 +1,12 @@
-use std::collections::BTreeMap;
-use std::io::{BufRead, BufReader};
-use std::path::{Path, PathBuf};
-use std::sync::Arc;
+use std::{
+    collections::BTreeMap,
+    io::{BufRead, BufReader},
+    path::{Path, PathBuf},
+    sync::Arc,
+};
 
 use alloy_consensus::{BlockBody, BlockHeader, Transaction};
-use alloy_primitives::TxKind;
-use alloy_primitives::{Address, PrimitiveSignature, B256, U256};
+use alloy_primitives::{Address, PrimitiveSignature, TxKind, B256, U256};
 use alloy_rpc_types::engine::{
     ExecutionPayloadEnvelopeV3, ForkchoiceState, PayloadAttributes, PayloadStatusEnum,
 };
@@ -14,9 +15,7 @@ use reth::network::PeersHandleProvider;
 use reth_chainspec::{EthChainSpec, EthereumHardforks};
 use reth_hyperliquid_types::{PrecompileData, PrecompilesCache};
 use reth_node_api::{Block, FullNodeComponents, PayloadTypes};
-use reth_node_builder::EngineTypes;
-use reth_node_builder::NodeTypesWithEngine;
-use reth_node_builder::{rpc::RethRpcAddOns, FullNode};
+use reth_node_builder::{rpc::RethRpcAddOns, EngineTypes, FullNode, NodeTypesWithEngine};
 use reth_payload_builder::{EthBuiltPayload, EthPayloadBuilderAttributes, PayloadId};
 use reth_primitives::{Transaction as TypedTransaction, TransactionSigned};
 use reth_provider::{BlockHashReader, BlockReader, StageCheckpointReader};
@@ -28,8 +27,10 @@ use time::{format_description, Duration, OffsetDateTime};
 use tokio::sync::Mutex;
 use tracing::{debug, info};
 
-use crate::serialized::{BlockAndReceipts, EvmBlock};
-use crate::spot_meta::erc20_contract_to_spot_token;
+use crate::{
+    serialized::{BlockAndReceipts, EvmBlock},
+    spot_meta::erc20_contract_to_spot_token,
+};
 
 /// Poll interval when tailing an *open* hourly file.
 const TAIL_INTERVAL: std::time::Duration = std::time::Duration::from_millis(25);
@@ -51,7 +52,11 @@ struct ScanResult {
     new_blocks: Vec<BlockAndReceipts>,
 }
 
-fn scan_hour_file(path: &Path, last_line: &mut usize, start_height: u64) -> Result<ScanResult, Box<dyn std::error::Error>> {
+fn scan_hour_file(
+    path: &Path,
+    last_line: &mut usize,
+    start_height: u64,
+) -> Result<ScanResult, Box<dyn std::error::Error + Send + Sync>> {
     // info!(
     //     "Scanning hour block file @ {:?} for height [{:?}] | Last Line {:?}",
     //     path, start_height, last_line
@@ -62,7 +67,8 @@ fn scan_hour_file(path: &Path, last_line: &mut usize, start_height: u64) -> Resu
 
     let mut new_blocks = Vec::<BlockAndReceipts>::new();
     let mut last_height = start_height;
-    let lines: Vec<String> = reader.lines()
+    let lines: Vec<String> = reader
+        .lines()
         .collect::<Result<Vec<_>, _>>()
         .map_err(|e| format!("Failed to read lines from file {}: {}", path.display(), e))?;
     let skip = if *last_line == 0 { 0 } else { (last_line.clone()) - 1 };
@@ -131,7 +137,8 @@ async fn submit_payload<Engine: PayloadTypes + EngineTypes>(
             engine_api_client,
             envelope.execution_payload,
             versioned_hashes,
-            payload_builder_attributes.parent_beacon_block_root
+            payload_builder_attributes
+                .parent_beacon_block_root
                 .ok_or("Missing required parent_beacon_block_root")?,
         )
         .await?
@@ -241,7 +248,11 @@ impl BlockIngest {
                             }
                         }
                         Err(e) => {
-                            tracing::error!("Failed to scan hour file {}: {}", hour_file.display(), e);
+                            tracing::error!(
+                                "Failed to scan hour file {}: {}",
+                                hour_file.display(),
+                                e
+                            );
                             // Continue processing but skip this file
                         }
                     }
@@ -415,9 +426,11 @@ impl BlockIngest {
                         },
                         None,
                     )
-                    .await {
+                    .await
+                    {
                         tracing::error!("Failed to update fork choice for block {}: {}", height, e);
-                        // Continue processing but log the failure - don't panic the entire blockchain
+                        // Continue processing but log the failure - don't panic the entire
+                        // blockchain
                     }
                     previous_timestamp = current_timestamp;
                 }

crates/net/network/src/network.rs

@@ -149,7 +149,7 @@ impl<N: NetworkPrimitives> NetworkHandle<N> {
     pub async fn transactions_handle(&self) -> Option<TransactionsHandle<N>> {
         let (tx, rx) = oneshot::channel();
         let _ = self.manager().send(NetworkHandleMessage::GetTransactionsHandle(tx));
-        rx.await.unwrap()
+        rx.await.ok().flatten()
     }
 
     /// Send message to gracefully shutdown node.
@@ -266,7 +266,8 @@ impl<N: NetworkPrimitives> PeersInfo for NetworkHandle<N> {
             builder.udp6(local_node_record.udp_port);
             builder.tcp6(local_node_record.tcp_port);
         }
-        builder.build(&self.inner.secret_key).expect("valid enr")
+        builder.build(&self.inner.secret_key)
+            .expect("ENR builder should always succeed with valid IP and ports")
     }
 }
 

crates/net/network/src/peers.rs

@@ -647,8 +647,11 @@ impl PeersManager {
 
             // remove peer if it has been marked for removal
             if remove_peer {
-                let (peer_id, _) = self.peers.remove_entry(peer_id).expect("peer must exist");
+                if let Some((peer_id, _)) = self.peers.remove_entry(peer_id) {
                     self.queued_actions.push_back(PeerAction::PeerRemoved(peer_id));
+                } else {
+                    tracing::warn!(target: "net::peers", "Attempted to remove non-existent peer: {:?}", peer_id);
+                }
             } else if let Some(backoff_until) = backoff_until {
                 // otherwise, backoff the peer if marked as such
                 self.backoff_peer_until(*peer_id, backoff_until);

crates/net/network/src/session/active.rs

@@ -391,7 +391,7 @@ impl<N: NetworkPrimitives> ActiveSession<N> {
         };
 
         self.terminate_message = Some((self.to_session_manager.inner().clone(), msg));
-        self.poll_terminate_message(cx).expect("message is set")
+        self.poll_terminate_message(cx).unwrap_or(Poll::Ready(()))
     }
 
     /// Report back that this session has been closed due to an error
@@ -402,7 +402,7 @@ impl<N: NetworkPrimitives> ActiveSession<N> {
             error,
         };
         self.terminate_message = Some((self.to_session_manager.inner().clone(), msg));
-        self.poll_terminate_message(cx).expect("message is set")
+        self.poll_terminate_message(cx).unwrap_or(Poll::Ready(()))
     }
 
     /// Starts the disconnect process

crates/storage/provider/src/providers/database/provider.rs

@@ -2278,16 +2278,7 @@ impl<TX: DbTxMut + DbTx + 'static, N: NodeTypesForProvider> StateWriter
             let mut block_receipts = Vec::with_capacity(block_body.tx_count as usize);
             for num in block_body.tx_num_range() {
                 if receipts_iter.peek().is_some_and(|(n, _)| *n == num) {
-                    // Safe to unwrap here since we just peeked and confirmed it exists
-                    // However, for maximum safety in database operations, we still handle it
-                    if let Some((_, receipt)) = receipts_iter.next() {
-                        block_receipts.push(receipt);
-                    } else {
-                        // This should never happen based on peek(), but handle gracefully
-                        return Err(ProviderError::Database(reth_db::DatabaseError::Other(
-                            "Receipt iterator state mismatch during state reconstruction".into()
-                        )));
-                    }
+                    block_receipts.push(receipts_iter.next().unwrap().1);
                 }
             }
             receipts.push(block_receipts);
@@ -3079,13 +3070,9 @@ impl<TX: DbTxMut + DbTx + 'static, N: NodeTypesForProvider + 'static> BlockWrite
             return Ok(());
         }
 
-        // Safe to unwrap after empty check, but use defensive programming for critical DB ops
-        let first_number = blocks.first()
-            .expect("Blocks vector guaranteed non-empty after length check")
-            .number();
+        let first_number = blocks.first().unwrap().number();
 
-        let last = blocks.last()
-            .expect("Blocks vector guaranteed non-empty after length check");
+        let last = blocks.last().unwrap();
         let last_block_number = last.number();
 
         let mut durations_recorder = metrics::DurationsRecorder::default();