Skip to content

refactor: delete sim error #78

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 2 commits into from
May 9, 2025
Merged

refactor: delete sim error #78

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
bf796ed
fix: metrics previously did not wait for confirmation
prestwich May 7, 2025
b8f712b
refactor: delete sim error
prestwich May 7, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 14 additions & 25 deletions src/tasks/block.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@ use alloy::{
providers::Provider,
};
use chrono::{DateTime, Utc};
use eyre::Report;
use eyre::{Context, bail};
use init4_bin_base::{
deps::tracing::{debug, error, info, warn},
utils::calc::SlotCalculator,
Expand All @@ -28,7 +28,6 @@ use std::{
},
time::{Duration, Instant, SystemTime, UNIX_EPOCH},
};
use thiserror::Error;
use tokio::{
select,
sync::mpsc::{self},
Expand All @@ -46,14 +45,6 @@ use trevm::{
},
};

/// Different error types that the Simulator handles
#[derive(Debug, Error)]
pub enum SimulatorError {
/// Wraps errors encountered when interacting with the RPC
#[error("RPC error: {0}")]
Rpc(#[source] Report),
}

/// `Simulator` is responsible for periodically building blocks and submitting them for
/// signing and inclusion in the blockchain. It wraps a rollup provider and a slot
/// calculator with a builder configuration.
Expand Down Expand Up @@ -106,7 +97,7 @@ impl Simulator {
sim_items: SimCache,
finish_by: Instant,
block: PecorinoBlockEnv,
) -> Result<BuiltBlock, SimulatorError> {
) -> eyre::Result<BuiltBlock> {
let db = self.create_db().await.unwrap();

let block_build: BlockBuild<_, NoOpInspector> = BlockBuild::new(
Expand Down Expand Up @@ -335,7 +326,7 @@ impl Simulator {
/// # Arguments
///
/// - finish_by: The deadline at which block simulation will end.
async fn next_block_env(&self, finish_by: Instant) -> Result<PecorinoBlockEnv, SimulatorError> {
async fn next_block_env(&self, finish_by: Instant) -> eyre::Result<PecorinoBlockEnv> {
let remaining = finish_by.duration_since(Instant::now());
let finish_time = SystemTime::now() + remaining;
let deadline: DateTime<Utc> = finish_time.into();
Expand All @@ -345,8 +336,8 @@ impl Simulator {
let latest_block_number = match self.ru_provider.get_block_number().await {
Ok(num) => num,
Err(err) => {
error!(error = %err, "RPC error during block build");
return Err(SimulatorError::Rpc(Report::new(err)));
error!(%err, "RPC error during block build");
bail!(err)
}
};
debug!(next_block_num = latest_block_number + 1, "preparing block env");
Expand Down Expand Up @@ -379,17 +370,15 @@ impl Simulator {
///
/// The basefee of the previous (latest) block if the request was successful,
/// or a sane default if the RPC failed.
async fn get_basefee(&self) -> Result<Option<u64>, SimulatorError> {
match self.ru_provider.get_block_by_number(Latest).await {
Ok(maybe_block) => match maybe_block {
Some(block) => {
debug!(basefee = ?block.header.base_fee_per_gas, "basefee found");
Ok(block.header.base_fee_per_gas)
}
None => Ok(None),
},
Err(err) => Err(SimulatorError::Rpc(err.into())),
}
async fn get_basefee(&self) -> eyre::Result<Option<u64>> {
let Some(block) =
self.ru_provider.get_block_by_number(Latest).await.wrap_err("basefee error")?
else {
return Ok(None);
};

debug!(basefee = ?block.header.base_fee_per_gas, "basefee found");
Ok(block.header.base_fee_per_gas)
}
}

Expand Down
96 changes: 52 additions & 44 deletions src/tasks/metrics.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,10 +1,13 @@
use crate::config::HostProvider;
use alloy::{primitives::TxHash, providers::Provider as _};
use alloy::{
primitives::TxHash,
providers::{PendingTransactionBuilder, PendingTransactionError, Provider as _, WatchTxError},
};
use init4_bin_base::deps::{
metrics::{counter, histogram},
tracing::{debug, error},
tracing::{Instrument, debug, error, info_span},
};
use std::time::Instant;
use std::time::{Duration, Instant};
use tokio::{sync::mpsc, task::JoinHandle};

/// Collects metrics on transactions sent by the Builder
Expand All @@ -15,64 +18,69 @@ pub struct MetricsTask {
}

impl MetricsTask {
/// Given a transaction hash, record metrics on the result of the transaction mining
pub async fn log_tx(&self, pending_tx_hash: TxHash) {
// start timer when tx hash is received
let start: Instant = Instant::now();
/// Given a transaction hash, record metrics on the result of the
/// transaction mining
pub fn log_tx(&self, tx_hash: TxHash) -> impl Future<Output = ()> + use<> {
let provider = self.host_provider.clone();

// wait for the tx to mine, get its receipt
let receipt_result =
self.host_provider.clone().get_transaction_receipt(pending_tx_hash).await;
async move {
// start timer when tx hash is received
let start: Instant = Instant::now();

match receipt_result {
Ok(maybe_receipt) => {
match maybe_receipt {
Some(receipt) => {
// record how long it took to mine the transaction
// potential improvement: use the block timestamp to calculate the time elapsed
histogram!("metrics.tx_mine_time")
.record(start.elapsed().as_millis() as f64);
let span = info_span!("metrics_submission", %tx_hash);

// log whether the transaction reverted
if receipt.status() {
counter!("metrics.tx_reverted").increment(1);
debug!(tx_hash = %pending_tx_hash, "tx reverted");
} else {
counter!("metrics.tx_succeeded").increment(1);
debug!(tx_hash = %pending_tx_hash, "tx succeeded");
}
}
None => {
counter!("metrics.no_receipt").increment(1);
error!("no receipt found for tx hash");
// wait for the tx to mine, get its receipt
let receipt = PendingTransactionBuilder::new(provider.root().clone(), tx_hash)
.with_required_confirmations(1)
.with_timeout(Some(Duration::from_secs(60)))
.get_receipt()
.instrument(span.clone())
.await;

// enter the span to log the result
let _guard = span.entered();

match receipt {
Ok(receipt) => {
// record how long it took to mine the transaction
// potential improvement: use the block timestamp to calculate the time elapsed
histogram!("metrics.tx_mine_time").record(start.elapsed().as_millis() as f64);

// log whether the transaction reverted
if receipt.status() {
counter!("metrics.tx_reverted").increment(1);
debug!("tx reverted");
} else {
counter!("metrics.tx_succeeded").increment(1);
debug!("tx succeeded");
}
}
}
Err(e) => {
counter!("metrics.rpc_error").increment(1);
error!(error = ?e, "rpc error");
Err(PendingTransactionError::TxWatcher(WatchTxError::Timeout)) => {
// log that the transaction timed out
counter!("metrics.tx_not_mined").increment(1);
debug!("tx not mined");
}
Err(e) => {
counter!("metrics.rpc_error").increment(1);
error!(error = ?e, "rpc error");
}
}
}
}

/// Spawns the task which collects metrics on pending transactions
pub fn spawn(self) -> (mpsc::UnboundedSender<TxHash>, JoinHandle<()>) {
let (sender, mut inbound) = mpsc::unbounded_channel();

let handle = tokio::spawn(async move {
debug!("metrics task spawned");
loop {
if let Some(pending_tx_hash) = inbound.recv().await {
let this = self.clone();
tokio::spawn(async move {
debug!("received tx hash");
let that = this.clone();
that.log_tx(pending_tx_hash).await;
debug!("logged tx metrics");
});
} else {
let Some(tx_hash) = inbound.recv().await else {
debug!("upstream task gone");
break;
}
};
let fut = self.log_tx(tx_hash);
tokio::spawn(fut);
}
});

Expand Down