jleony/nanoreth
1
0
Fork 0
mirror of https://github.com/hl-archive-node/nanoreth.git synced 2025-12-06 10:59:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge c7d1f61817 into 9f952ac2ed

Browse Source
This commit is contained in:
Kamal Nayan
2025-08-25 09:55:14 +00:00
committed by GitHub
parent 9f952ac2ed c7d1f61817
commit e60bc3f09a
6 changed files with 99 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

133
bin/reth/src/block_ingest.rs
View File

@ -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,17 +52,25 @@ struct ScanResult {
new_blocks: Vec<BlockAndReceipts>,
}
fn scan_hour_file(path: &Path, last_line: &mut usize, start_height: u64) -> ScanResult {
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
// );
let file = std::fs::File::open(path).expect("Failed to open hour file path");
let file = std::fs::File::open(path)
.map_err(|e| format!("Failed to open hour file path {}: {}", path.display(), e))?;
let reader = BufReader::new(file);
let mut new_blocks = Vec::<BlockAndReceipts>::new();
let mut last_height = start_height;
let lines: Vec<String> = reader.lines().collect::<Result<_, _>>().unwrap();
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 };
for (line_idx, line) in lines.iter().enumerate().skip(skip) {
@ -109,7 +118,7 @@ fn scan_hour_file(path: &Path, last_line: &mut usize, start_height: u64) -> Scan
}
}
ScanResult { next_expected_height: last_height + 1, new_blocks }
Ok(ScanResult { next_expected_height: last_height + 1, new_blocks })
}
async fn submit_payload<Engine: PayloadTypes + EngineTypes>(
@ -128,7 +137,9 @@ async fn submit_payload<Engine: PayloadTypes + EngineTypes>(
engine_api_client,
envelope.execution_payload,
versioned_hashes,
payload_builder_attributes.parent_beacon_block_root.unwrap(),
payload_builder_attributes
.parent_beacon_block_root
.ok_or("Missing required parent_beacon_block_root")?,
)
.await?
};
@ -144,7 +155,9 @@ fn datetime_from_timestamp(ts_sec: u64) -> OffsetDateTime {
}
fn date_from_datetime(dt: OffsetDateTime) -> String {
dt.format(&format_description::parse("[year][month][day]").unwrap()).unwrap()
// Format string is constant and guaranteed to be valid
dt.format(&format_description::parse("[year][month][day]").expect("Valid format string"))
.expect("DateTime formatting should always succeed with valid format")
}
impl BlockIngest {
@ -166,7 +179,18 @@ impl BlockIngest {
let path = format!("{}/{f}/{s}/{height}.rmp.lz4", self.ingest_dir.to_string_lossy());
let file = std::fs::read(path).ok()?;
let mut decoder = lz4_flex::frame::FrameDecoder::new(&file[..]);
let blocks: Vec<BlockAndReceipts> = rmp_serde::from_read(&mut decoder).unwrap();
let blocks: Vec<BlockAndReceipts> = rmp_serde::from_read(&mut decoder)
.map_err(|e| {
tracing::error!("Failed to deserialize block data for height {}: {}", height, e);
e
})
.ok()?;
if blocks.is_empty() {
tracing::error!("Deserialized empty blocks vector for height {}", height);
return None;
}
info!("Returning s3 synced block for @ Height [{height}]");
Some(blocks[0].clone())
}
@ -186,11 +210,11 @@ impl BlockIngest {
let mut next_height = current_head;
let mut dt = datetime_from_timestamp(current_ts)
.replace_minute(0)
.unwrap()
.expect("Valid minute replacement")
.replace_second(0)
.unwrap()
.expect("Valid second replacement")
.replace_nanosecond(0)
.unwrap();
.expect("Valid nanosecond replacement");
let mut hour = dt.hour();
let mut day_str = date_from_datetime(dt);
@ -200,26 +224,37 @@ impl BlockIngest {
let hour_file = root.join(HOURLY_SUBDIR).join(&day_str).join(format!("{hour}"));
if hour_file.exists() {
let ScanResult { next_expected_height, new_blocks } =
scan_hour_file(&hour_file, &mut last_line, next_height);
if !new_blocks.is_empty() {
let mut u_cache = cache.lock().await;
let mut u_pre_cache = precompiles_cache.lock();
for blk in new_blocks {
let precompiles = PrecompileData {
precompiles: blk.read_precompile_calls.clone(),
highest_precompile_address: blk.highest_precompile_address,
};
let h = match &blk.block {
EvmBlock::Reth115(b) => {
let block_number = b.header().number() as u64;
block_number
let scan_result = scan_hour_file(&hour_file, &mut last_line, next_height);
match scan_result {
Ok(ScanResult { next_expected_height, new_blocks }) => {
if !new_blocks.is_empty() {
let mut u_cache = cache.lock().await;
let mut u_pre_cache = precompiles_cache.lock();
for blk in new_blocks {
let precompiles = PrecompileData {
precompiles: blk.read_precompile_calls.clone(),
highest_precompile_address: blk.highest_precompile_address,
};
let h = match &blk.block {
EvmBlock::Reth115(b) => {
let block_number = b.header().number() as u64;
block_number
}
};
u_cache.insert(h, blk);
u_pre_cache.insert(h, precompiles);
}
};
u_cache.insert(h, blk);
u_pre_cache.insert(h, precompiles);
next_height = next_expected_height;
}
}
Err(e) => {
tracing::error!(
"Failed to scan hour file {}: {}",
hour_file.display(),
e
);
// Continue processing but skip this file
}
next_height = next_expected_height;
}
}
@ -269,8 +304,10 @@ impl BlockIngest {
let mut height = head + 1;
let mut previous_hash = provider.block_hash(head)?.unwrap_or(genesis_hash);
let mut previous_timestamp =
std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_millis();
let mut previous_timestamp = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.expect("System time should be after UNIX epoch")
.as_millis();
let engine_api = node.auth_server_handle().http_client();
let mut evm_map = erc20_contract_to_spot_token(node.chain_spec().chain_id()).await?;
@ -278,8 +315,8 @@ impl BlockIngest {
const MINIMUM_TIMESTAMP: u64 = 1739849780;
let current_block_timestamp: u64 = provider
.block_by_number(head)
.expect("Failed to fetch current block in db")
.expect("Block does not exist")
.map_err(|e| format!("Database error fetching block {}: {}", head, e))?
.ok_or_else(|| format!("Block {} does not exist in database", head))?
.into_header()
.timestamp();
@ -376,11 +413,11 @@ impl BlockIngest {
let current_timestamp = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.expect("System time should be after UNIX epoch")
.as_millis();
if height % 100 == 0 || current_timestamp - previous_timestamp > 100 {
EngineApiClient::<Engine>::fork_choice_updated_v2(
if let Err(e) = EngineApiClient::<Engine>::fork_choice_updated_v2(
&engine_api,
ForkchoiceState {
head_block_hash: block_hash,
@ -390,7 +427,11 @@ impl BlockIngest {
None,
)
.await
.unwrap();
{
tracing::error!("Failed to update fork choice for block {}: {}", height, e);
// Continue processing but log the failure - don't panic the entire
// blockchain
}
previous_timestamp = current_timestamp;
}
previous_hash = block_hash;

5
crates/net/network/src/network.rs
View File

@ -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")
}
}

7
crates/net/network/src/peers.rs
View File

@ -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");
self.queued_actions.push_back(PeerAction::PeerRemoved(peer_id));
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);

4
crates/net/network/src/session/active.rs
View File

@ -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