diff --git a/Cargo.lock b/Cargo.lock index 8d0f9fa..994a115 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1279,6 +1279,7 @@ dependencies = [ "snap", "tokio", "tracing", + "tracing-subscriber 0.3.23", "txgen-core", "urlencoding", "uuid", diff --git a/crates/bench-cli/src/send.rs b/crates/bench-cli/src/send.rs index c570800..a3a57aa 100644 --- a/crates/bench-cli/src/send.rs +++ b/crates/bench-cli/src/send.rs @@ -6,14 +6,14 @@ use crate::{ SendArgs, }; use alloy_network::AnyNetwork; -use alloy_provider::{ext::TxPoolApi, DynProvider, Provider, ProviderBuilder}; +use alloy_provider::{ext::TxPoolApi, Provider, ProviderBuilder}; use alloy_rpc_client::RpcClient; use alloy_transport::layers::RetryBackoffLayer; use bench_core::{ collect_block_stats, parse_reporters, start_scrapers, trim_trailing_empty_blocks, ConsoleReporter, FileSource, FinalReport, GeneratedTx, MetricsCollector, ProgressState, - Reporter, RunClock, RunStats, SampleStore, ScraperConfig, Sender, SenderConfig, StdinSource, - TxPhase, TxSource, + Reporter, RpcEndpoint, RunClock, RunStats, SampleStore, ScraperConfig, Sender, SenderConfig, + StdinSource, TxPhase, TxSource, }; use eyre::{bail, Context, Result}; use std::{collections::HashMap, time::Duration}; @@ -44,12 +44,14 @@ pub async fn execute(args: SendArgs) -> Result<()> { let providers = args .rpc_urls .iter() - .map(|url| { - let url = url.parse().context("failed to parse RPC URL")?; + .map(|rpc_url| { + let url = rpc_url.parse().context("failed to parse RPC URL")?; let client = RpcClient::builder() .layer(retry_layer.clone()) .http_with_client(http_client.clone(), url); - Ok(ProviderBuilder::new_with_network::().connect_client(client).erased()) + let provider = + ProviderBuilder::new_with_network::().connect_client(client).erased(); + Ok(RpcEndpoint::new(rpc_url.clone(), provider)) }) .collect::>>()?; @@ -68,12 +70,12 @@ pub async fn execute(args: SendArgs) -> Result<()> { async fn execute_source( args: &SendArgs, metadata: &HashMap, - providers: Vec>, + providers: Vec, source: &mut S, scraper_configs: &[ScraperConfig], ) -> Result<()> { let config = SenderConfig { rate_limit: args.tps, max_concurrent: args.max_concurrent }; - let query_provider = &providers[0]; + let query_provider = providers[0].provider(); let first_workload = run_setup_phase(args, source, &providers, &config).await?; @@ -100,7 +102,7 @@ async fn execute_source( Vec::new() }; - let mut sender = Sender::new(providers.clone(), config.clone(), metrics.clone()); + let mut sender = Sender::new_with_endpoints(providers.clone(), config.clone(), metrics.clone()); let mut reporters = parse_reporters(&args.reports, "send", metadata)?; if reporters.is_empty() { @@ -235,12 +237,13 @@ async fn execute_source( async fn run_setup_phase( args: &SendArgs, source: &mut S, - providers: &[DynProvider], + providers: &[RpcEndpoint], config: &SenderConfig, ) -> Result> { let setup_clock = RunClock::new(); let setup_metrics = MetricsCollector::new_with_latencies(setup_clock, false); - let mut setup_sender = Sender::new(providers.to_vec(), config.clone(), setup_metrics.clone()); + let mut setup_sender = + Sender::new_with_endpoints(providers.to_vec(), config.clone(), setup_metrics.clone()); let mut setup_seen = 0u64; while let Some(tx) = source.next_tx().await? { @@ -287,6 +290,11 @@ async fn finish_setup_phase( bail!("setup phase failed: {failed} setup transaction(s) failed or reverted"); } + setup_sender + .wait_for_setup_convergence(setup_seen) + .await + .wrap_err("setup RPC convergence barrier failed")?; + Ok(()) } diff --git a/crates/bench-core/Cargo.toml b/crates/bench-core/Cargo.toml index 9bc0cfc..8d8d957 100644 --- a/crates/bench-core/Cargo.toml +++ b/crates/bench-core/Cargo.toml @@ -43,3 +43,4 @@ flate2 = "1" prost = "0.12" snap = "1" tokio = { version = "1", features = ["rt", "macros"] } +tracing-subscriber = "0.3" diff --git a/crates/bench-core/src/lib.rs b/crates/bench-core/src/lib.rs index 77a7127..826f53f 100644 --- a/crates/bench-core/src/lib.rs +++ b/crates/bench-core/src/lib.rs @@ -40,6 +40,6 @@ pub use reth_api::{ }; pub use sample::{Sample, SampleArchive, SampleStore}; pub use scraper::{start_scrapers, SampleCallback, ScraperConfig, ScraperHandle}; -pub use sender::{Sender, SenderConfig}; +pub use sender::{RpcEndpoint, Sender, SenderConfig}; pub use source::{FileSource, SourceTx, StdinSource, TxSource}; pub use txgen_core::{GeneratedTx, TxPhase}; diff --git a/crates/bench-core/src/sender.rs b/crates/bench-core/src/sender.rs index efdc231..4d439c6 100644 --- a/crates/bench-core/src/sender.rs +++ b/crates/bench-core/src/sender.rs @@ -6,12 +6,14 @@ use crate::metrics::MetricsCollector; use alloy_network::{primitives::ReceiptResponse, AnyNetwork}; -use alloy_primitives::Bytes; +use alloy_primitives::{keccak256, BlockHash, Bytes, TxHash}; use alloy_provider::{DynProvider, Provider}; -use eyre::Result; +use eyre::{bail, Result, WrapErr}; +use futures::future::try_join_all; use rand::seq::IndexedRandom; use std::{ collections::{HashSet, VecDeque}, + fmt, sync::Arc, time::{Duration, Instant}, }; @@ -48,8 +50,62 @@ type KeySets = (SchedulingKeys, SchedulingKeys); const RECEIPT_POLL_INTERVAL: Duration = Duration::from_millis(100); const RECEIPT_TIMEOUT: Duration = Duration::from_secs(300); +const FAILURE_HEAD_QUERY_TIMEOUT: Duration = Duration::from_secs(2); const PENDING_BACKLOG_FACTOR: usize = 4; +/// An RPC endpoint used for transaction submission and receipt queries. +/// +/// Keeping the URL next to the erased provider makes failures attributable to +/// the endpoint that produced them. +#[derive(Clone)] +pub struct RpcEndpoint { + url: Arc, + provider: DynProvider, +} + +impl RpcEndpoint { + /// Create an endpoint from its configured URL and provider. + pub fn new(url: impl Into, provider: DynProvider) -> Self { + Self { url: Arc::from(url.into()), provider } + } + + /// The configured RPC URL. + pub fn url(&self) -> &str { + &self.url + } + + /// The provider connected to this endpoint. + pub fn provider(&self) -> &DynProvider { + &self.provider + } +} + +impl fmt::Debug for RpcEndpoint { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("RpcEndpoint").field("url", &self.url).finish_non_exhaustive() + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct SetupReceiptCheckpoint { + tx_hash: TxHash, + block_number: u64, + block_hash: BlockHash, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct ReceiptObservation { + status: bool, + block_number: Option, + block_hash: Option, +} + +#[derive(Debug, PartialEq, Eq)] +struct ProviderHeadDiagnostic { + block_number: Option, + error: Option, +} + /// A transaction to be sent. struct PendingTx { phase: TxPhase, @@ -67,7 +123,7 @@ impl PendingTx { /// Transaction sender. pub struct Sender { - providers: Vec>, + providers: Vec, metrics: Arc, semaphore: Arc, /// Transactions waiting for all of their scheduling keys to become free. @@ -76,6 +132,8 @@ pub struct Sender { active_keys: HashSet, completion_tx: mpsc::UnboundedSender, completion_rx: mpsc::UnboundedReceiver, + setup_receipt_tx: mpsc::UnboundedSender, + setup_receipt_rx: mpsc::UnboundedReceiver, /// Worker tasks for awaiting completion and reaping completed task state. worker_tasks: JoinSet<()>, /// Rate limiter tokens. @@ -86,11 +144,25 @@ pub struct Sender { } impl Sender { - /// Create a new sender. + /// Create a new sender from providers without configured URL labels. pub fn new( providers: Vec>, config: SenderConfig, metrics: Arc, + ) -> Self { + let endpoints = providers + .into_iter() + .enumerate() + .map(|(index, provider)| RpcEndpoint::new(format!("provider[{index}]"), provider)) + .collect(); + Self::new_with_endpoints(endpoints, config, metrics) + } + + /// Create a new sender from providers paired with their configured URLs. + pub fn new_with_endpoints( + providers: Vec, + config: SenderConfig, + metrics: Arc, ) -> Self { let semaphore = Arc::new(Semaphore::new(config.max_concurrent)); @@ -102,6 +174,7 @@ impl Sender { let max_buffered = max_buffered_transactions(&config, rate_limiter.as_deref()); let (completion_tx, completion_rx) = mpsc::unbounded_channel(); + let (setup_receipt_tx, setup_receipt_rx) = mpsc::unbounded_channel(); Self { providers, @@ -111,6 +184,8 @@ impl Sender { active_keys: HashSet::new(), completion_tx, completion_rx, + setup_receipt_tx, + setup_receipt_rx, worker_tasks: JoinSet::new(), rate_limiter, max_buffered, @@ -155,6 +230,48 @@ impl Sender { } } + /// Wait until every configured endpoint has imported the successful setup + /// receipts observed during this sender's most recent flush. + /// + /// A matching transaction hash alone is not sufficient: every endpoint + /// must report success and the same inclusion block number and hash. This + /// prevents workload submission through a lagging or divergent endpoint. + /// `expected_setup_txs` is also checked so a missing worker checkpoint + /// cannot silently weaken the barrier. + pub async fn wait_for_setup_convergence(&mut self, expected_setup_txs: u64) -> Result<()> { + let mut checkpoints = Vec::new(); + while let Ok(checkpoint) = self.setup_receipt_rx.try_recv() { + checkpoints.push(checkpoint); + } + if checkpoints.len() as u64 != expected_setup_txs { + bail!( + "setup receipt checkpoint count mismatch: expected {expected_setup_txs}, observed {}", + checkpoints.len() + ); + } + + tracing::info!( + setup_txs = checkpoints.len(), + rpc_endpoints = self.providers.len(), + "Waiting for setup state on all RPC endpoints" + ); + + wait_for_all_provider_receipts( + &self.providers, + &checkpoints, + RECEIPT_TIMEOUT, + RECEIPT_POLL_INTERVAL, + ) + .await?; + + tracing::info!( + setup_txs = checkpoints.len(), + rpc_endpoints = self.providers.len(), + "Setup state converged on all RPC endpoints" + ); + Ok(()) + } + fn drain_completions(&mut self) { while let Ok(keys) = self.completion_rx.try_recv() { self.release_keys(&keys); @@ -244,9 +361,10 @@ impl Sender { let providers = self.providers.clone(); let metrics = self.metrics.clone(); let completion_tx = self.completion_tx.clone(); + let setup_receipt_tx = self.setup_receipt_tx.clone(); self.worker_tasks.spawn(async move { - submit_tx(pending, providers, metrics, permit, completion_tx).await; + submit_tx(pending, providers, metrics, permit, completion_tx, setup_receipt_tx).await; }); } } @@ -277,10 +395,11 @@ fn normalize_key_sets( async fn submit_tx( pending: PendingTx, - providers: Vec>, + providers: Vec, metrics: Arc, permit: OwnedSemaphorePermit, completion_tx: mpsc::UnboundedSender, + setup_receipt_tx: mpsc::UnboundedSender, ) { let release_all_keys = || { let mut keys = pending.submission_keys.clone(); @@ -292,23 +411,50 @@ async fn submit_tx( // Pick a random provider for this request. // SAFETY: `providers` is guaranteed to be non-empty by construction. - let provider = providers.choose(&mut rand::rng()).unwrap(); + let endpoint = providers.choose(&mut rand::rng()).unwrap(); + let local_tx_hash = keccak256(&pending.raw); let start = Instant::now(); - let tx_hash = match provider.send_raw_transaction(&pending.raw).await { + let tx_hash = match endpoint.provider().send_raw_transaction(&pending.raw).await { Ok(pending_tx) => { metrics.record_success(start.elapsed()); *pending_tx.tx_hash() } Err(e) => { - tracing::warn!(error = %e, "Failed to send transaction"); + let error = e.to_string(); metrics.record_failure(); - drop(permit); release_keys(&completion_tx, release_all_keys()); + + // A rejected transaction has no receipt or canonical inclusion + // block. Query the same endpoint immediately after rejection to + // capture the approximate head against which it was validated. + // Retain the permit so diagnostic requests remain bounded by the + // configured maximum RPC concurrency if failures arrive in bulk. + let head = query_provider_head(endpoint).await; + drop(permit); + tracing::warn!( + error = %error, + rpc_url = endpoint.url(), + tx_hash = %local_tx_hash, + id = pending.id.as_deref(), + phase = ?pending.phase, + validation_head = head.block_number, + validation_head_error = head.error.as_deref(), + "Failed to send transaction" + ); return; } }; + if tx_hash != local_tx_hash { + tracing::warn!( + rpc_url = endpoint.url(), + %local_tx_hash, + provider_tx_hash = %tx_hash, + "RPC endpoint returned a transaction hash that differs from the locally computed hash" + ); + } + drop(permit); release_keys(&completion_tx, pending.submission_keys); @@ -317,14 +463,53 @@ async fn submit_tx( return; } - match wait_for_receipt(provider, tx_hash).await { - Ok(true) => {} - Ok(false) => { - tracing::error!(id = pending.id.as_deref(), phase = ?pending.phase, %tx_hash, "Transaction reverted"); + match wait_for_receipt(endpoint, tx_hash).await { + Ok(receipt) if !receipt.status => { + tracing::error!( + id = pending.id.as_deref(), + phase = ?pending.phase, + rpc_url = endpoint.url(), + %tx_hash, + block_number = receipt.block_number, + block_hash = ?receipt.block_hash, + "Transaction reverted" + ); metrics.record_failure(); } + Ok(receipt) => { + if pending.phase == TxPhase::Setup { + match (receipt.block_number, receipt.block_hash) { + (Some(block_number), Some(block_hash)) => { + let _ = setup_receipt_tx.send(SetupReceiptCheckpoint { + tx_hash, + block_number, + block_hash, + }); + } + _ => { + tracing::error!( + id = pending.id.as_deref(), + phase = ?pending.phase, + rpc_url = endpoint.url(), + %tx_hash, + block_number = receipt.block_number, + block_hash = ?receipt.block_hash, + "Successful setup receipt is missing its inclusion block identity" + ); + metrics.record_failure(); + } + } + } + } Err(e) => { - tracing::error!(error = %e, %tx_hash, "Failed waiting for transaction receipt"); + tracing::error!( + error = %e, + rpc_url = endpoint.url(), + %tx_hash, + id = pending.id.as_deref(), + phase = ?pending.phase, + "Failed waiting for transaction receipt" + ); metrics.record_failure(); } } @@ -338,25 +523,136 @@ fn release_keys(completion_tx: &mpsc::UnboundedSender, keys: Sch } } -async fn wait_for_receipt( - provider: &DynProvider, - tx_hash: alloy_primitives::TxHash, -) -> Result { +async fn query_provider_head(endpoint: &RpcEndpoint) -> ProviderHeadDiagnostic { + match tokio::time::timeout(FAILURE_HEAD_QUERY_TIMEOUT, endpoint.provider().get_block_number()) + .await + { + Ok(Ok(block_number)) => { + ProviderHeadDiagnostic { block_number: Some(block_number), error: None } + } + Ok(Err(error)) => { + ProviderHeadDiagnostic { block_number: None, error: Some(error.to_string()) } + } + Err(_) => ProviderHeadDiagnostic { + block_number: None, + error: Some(format!( + "head query timed out after {}s", + FAILURE_HEAD_QUERY_TIMEOUT.as_secs() + )), + }, + } +} + +async fn wait_for_receipt(endpoint: &RpcEndpoint, tx_hash: TxHash) -> Result { let deadline = tokio::time::Instant::now() + RECEIPT_TIMEOUT; loop { - if let Some(receipt) = provider.get_transaction_receipt(tx_hash).await? { - return Ok(receipt.status()); + if let Some(receipt) = + endpoint.provider().get_transaction_receipt(tx_hash).await.wrap_err_with(|| { + format!( + "failed to query transaction receipt from rpc_url={} tx_hash={tx_hash}", + endpoint.url() + ) + })? + { + return Ok(ReceiptObservation { + status: receipt.status(), + block_number: receipt.block_number(), + block_hash: receipt.block_hash(), + }); } if tokio::time::Instant::now() >= deadline { - eyre::bail!("timed out waiting for transaction receipt"); + bail!( + "timed out waiting for transaction receipt from rpc_url={} tx_hash={tx_hash}", + endpoint.url() + ); } tokio::time::sleep(RECEIPT_POLL_INTERVAL).await; } } +async fn wait_for_all_provider_receipts( + endpoints: &[RpcEndpoint], + checkpoints: &[SetupReceiptCheckpoint], + timeout: Duration, + poll_interval: Duration, +) -> Result<()> { + let deadline = tokio::time::Instant::now() + timeout; + + try_join_all(endpoints.iter().map(|endpoint| async move { + for checkpoint in checkpoints { + wait_for_matching_setup_receipt(endpoint, checkpoint, deadline, poll_interval).await?; + } + Ok::<(), eyre::Report>(()) + })) + .await?; + + Ok(()) +} + +async fn wait_for_matching_setup_receipt( + endpoint: &RpcEndpoint, + expected: &SetupReceiptCheckpoint, + deadline: tokio::time::Instant, + poll_interval: Duration, +) -> Result<()> { + loop { + let receipt = + endpoint.provider().get_transaction_receipt(expected.tx_hash).await.wrap_err_with( + || { + format!( + "failed to query setup receipt from rpc_url={} tx_hash={}", + endpoint.url(), + expected.tx_hash + ) + }, + )?; + + if let Some(receipt) = receipt { + if !receipt.status() { + bail!( + "setup receipt reverted on rpc_url={} tx_hash={} block_number={:?} block_hash={:?}", + endpoint.url(), + expected.tx_hash, + receipt.block_number(), + receipt.block_hash() + ); + } + + let observed_number = receipt.block_number(); + let observed_hash = receipt.block_hash(); + if observed_number != Some(expected.block_number) || + observed_hash != Some(expected.block_hash) + { + bail!( + "setup receipt block mismatch on rpc_url={} tx_hash={}: expected block_number={} block_hash={}, observed block_number={observed_number:?} block_hash={observed_hash:?}", + endpoint.url(), + expected.tx_hash, + expected.block_number, + expected.block_hash + ); + } + + return Ok(()); + } + + let now = tokio::time::Instant::now(); + if now >= deadline { + bail!( + "timed out waiting for setup state on rpc_url={} tx_hash={} expected_block_number={} expected_block_hash={}", + endpoint.url(), + expected.tx_hash, + expected.block_number, + expected.block_hash + ); + } + + tokio::time::sleep(poll_interval.min(deadline.saturating_duration_since(now))).await; + } +} + const RATE_LIMITER_MAX_BURST: Duration = Duration::from_millis(10); /// Token-bucket rate limiter with a bounded burst budget. @@ -417,6 +713,76 @@ impl RateLimiterState { #[cfg(test)] mod tests { use super::*; + use alloy_provider::ProviderBuilder; + use alloy_transport::mock::Asserter; + use serde_json::{json, Value}; + use std::{ + io::{self, Write}, + sync::Mutex, + }; + use tracing_subscriber::fmt::MakeWriter; + + #[derive(Clone, Default)] + struct CapturedLogs(Arc>>); + + struct CapturedLogWriter(Arc>>); + + impl Write for CapturedLogWriter { + fn write(&mut self, buf: &[u8]) -> io::Result { + self.0.lock().unwrap().extend_from_slice(buf); + Ok(buf.len()) + } + + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } + } + + impl<'a> MakeWriter<'a> for CapturedLogs { + type Writer = CapturedLogWriter; + + fn make_writer(&'a self) -> Self::Writer { + CapturedLogWriter(self.0.clone()) + } + } + + impl CapturedLogs { + fn contents(&self) -> String { + String::from_utf8(self.0.lock().unwrap().clone()).unwrap() + } + } + + fn mock_endpoint(url: &str) -> (RpcEndpoint, Asserter) { + let asserter = Asserter::new(); + let provider = ProviderBuilder::new_with_network::() + .connect_mocked_client(asserter.clone()) + .erased(); + (RpcEndpoint::new(url, provider), asserter) + } + + fn mock_receipt( + tx_hash: TxHash, + block_number: u64, + block_hash: BlockHash, + status: bool, + ) -> Value { + json!({ + "type": "0x0", + "status": if status { "0x1" } else { "0x0" }, + "cumulativeGasUsed": "0x5208", + "logsBloom": format!("0x{}", "00".repeat(256)), + "logs": [], + "transactionHash": tx_hash, + "transactionIndex": "0x0", + "blockHash": block_hash, + "blockNumber": format!("0x{block_number:x}"), + "gasUsed": "0x5208", + "effectiveGasPrice": "0x1", + "from": "0x0000000000000000000000000000000000000001", + "to": "0x0000000000000000000000000000000000000002", + "contractAddress": null + }) + } #[test] fn test_sender_config_default() { @@ -459,4 +825,123 @@ mod tests { let limiter = RateLimiter::new(config.rate_limit); assert_eq!(max_buffered_transactions(&config, Some(&limiter)), 2_000); } + + #[tokio::test] + async fn setup_barrier_waits_for_a_lagging_provider_receipt() { + let tx_hash = TxHash::repeat_byte(0x11); + let block_hash = BlockHash::repeat_byte(0x22); + let checkpoint = SetupReceiptCheckpoint { tx_hash, block_number: 42, block_hash }; + let receipt = mock_receipt(tx_hash, 42, block_hash, true); + + let (synced, synced_rpc) = mock_endpoint("http://synced.example"); + synced_rpc.push_success(&receipt); + + let (lagging, lagging_rpc) = mock_endpoint("http://lagging.example"); + lagging_rpc.push_success(&Value::Null); + lagging_rpc.push_success(&receipt); + + wait_for_all_provider_receipts( + &[synced, lagging], + &[checkpoint], + Duration::from_secs(1), + Duration::from_millis(1), + ) + .await + .unwrap(); + + assert!(synced_rpc.read_q().is_empty()); + assert!(lagging_rpc.read_q().is_empty()); + } + + #[tokio::test] + async fn setup_barrier_requires_matching_block_identity_from_every_provider() { + let tx_hash = TxHash::repeat_byte(0x33); + let expected_hash = BlockHash::repeat_byte(0x44); + let checkpoint = + SetupReceiptCheckpoint { tx_hash, block_number: 100, block_hash: expected_hash }; + + let (synced, synced_rpc) = mock_endpoint("http://synced.example"); + synced_rpc.push_success(&mock_receipt(tx_hash, 100, expected_hash, true)); + + let (divergent, divergent_rpc) = mock_endpoint("http://divergent.example"); + divergent_rpc.push_success(&mock_receipt(tx_hash, 100, BlockHash::repeat_byte(0x55), true)); + + let error = wait_for_all_provider_receipts( + &[synced, divergent], + &[checkpoint], + Duration::from_secs(1), + Duration::from_millis(1), + ) + .await + .unwrap_err() + .to_string(); + + assert!(error.contains("http://divergent.example"), "{error}"); + assert!(error.contains("setup receipt block mismatch"), "{error}"); + } + + #[tokio::test] + async fn setup_barrier_rejects_a_missing_receipt_checkpoint() { + let (endpoint, _) = mock_endpoint("http://setup.example"); + let metrics = MetricsCollector::new(crate::clock::RunClock::new()); + let mut sender = + Sender::new_with_endpoints(vec![endpoint], SenderConfig::default(), metrics); + + let error = sender.wait_for_setup_convergence(1).await.unwrap_err().to_string(); + + assert!(error.contains("expected 1, observed 0"), "{error}"); + } + + #[test] + fn send_failure_logs_provider_hash_and_head() { + let (endpoint, rpc) = mock_endpoint("http://rejecting.example"); + rpc.push_failure_msg("transaction rejected"); + rpc.push_success(&123u64); + + let raw = Bytes::from_static(&[0x01, 0x02, 0x03]); + let expected_hash = keccak256(&raw); + let pending = PendingTx { + phase: TxPhase::Workload, + id: Some("rejected-workload".to_string()), + raw, + submission_keys: Vec::new(), + inclusion_keys: Vec::new(), + }; + let logs = CapturedLogs::default(); + let subscriber = tracing_subscriber::fmt() + .without_time() + .with_ansi(false) + .with_writer(logs.clone()) + .finish(); + let dispatcher = tracing::Dispatch::new(subscriber); + + let metrics = tracing::dispatcher::with_default(&dispatcher, || { + tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap().block_on( + async { + let metrics = MetricsCollector::new(crate::clock::RunClock::new()); + let permit = Arc::new(Semaphore::new(1)).acquire_owned().await.unwrap(); + let (completion_tx, _) = mpsc::unbounded_channel(); + let (setup_receipt_tx, _) = mpsc::unbounded_channel(); + + submit_tx( + pending, + vec![endpoint], + metrics.clone(), + permit, + completion_tx, + setup_receipt_tx, + ) + .await; + metrics + }, + ) + }); + + assert!(rpc.read_q().is_empty(), "head-number response was not consumed"); + assert_eq!(metrics.counts(), (1, 0, 1)); + let output = logs.contents(); + assert!(output.contains("rpc_url=\"http://rejecting.example\""), "{output}"); + assert!(output.contains(&format!("tx_hash={expected_hash}")), "{output}"); + assert!(output.contains("validation_head=123"), "{output}"); + } }