diff --git a/crates/engine/primitives/src/message.rs b/crates/engine/primitives/src/message.rs index fca83328777..430beee3f1b 100644 --- a/crates/engine/primitives/src/message.rs +++ b/crates/engine/primitives/src/message.rs @@ -196,8 +196,6 @@ pub enum BeaconEngineMessage { payload: Payload::ExecutionData, /// The sender for returning payload status result. tx: oneshot::Sender>, - /// When this message was enqueued, used to measure backpressure wait time. - enqueued_at: Instant, }, /// Message with new payload used by `reth_newPayload` endpoint. /// @@ -290,11 +288,7 @@ where payload: Payload::ExecutionData, ) -> Result { let (tx, rx) = oneshot::channel(); - let _ = self.to_engine.send(BeaconEngineMessage::NewPayload { - payload, - tx, - enqueued_at: Instant::now(), - }); + let _ = self.to_engine.send(BeaconEngineMessage::NewPayload { payload, tx }); rx.await.map_err(|_| BeaconOnNewPayloadError::EngineUnavailable)? } diff --git a/crates/engine/tree/src/tree/metrics.rs b/crates/engine/tree/src/tree/metrics.rs index e9e0f593287..f8a96e3c6bb 100644 --- a/crates/engine/tree/src/tree/metrics.rs +++ b/crates/engine/tree/src/tree/metrics.rs @@ -436,9 +436,9 @@ impl NewPayloadStatusMetrics { latest_forkchoice_updated_at: &mut Option, result: &Result, InsertBlockFatalError>, gas_used: u64, - latency: Duration, ) { let finish = Instant::now(); + let elapsed = finish - start; if let Some(prev_finish) = self.latest_finish_at { self.time_between_new_payloads.record(start - prev_finish); @@ -455,13 +455,13 @@ impl NewPayloadStatusMetrics { if !outcome.already_seen { self.new_payload_total_gas.record(gas_used as f64); self.new_payload_total_gas_last.set(gas_used as f64); - let gas_per_second = gas_used as f64 / latency.as_secs_f64(); + let gas_per_second = gas_used as f64 / elapsed.as_secs_f64(); self.new_payload_gas_per_second.record(gas_per_second); self.new_payload_gas_per_second_last.set(gas_per_second); - self.new_payload_latency.record(latency); - self.new_payload_last.set(latency); - self.gas_bucket.record(gas_used, latency); + self.new_payload_latency.record(elapsed); + self.new_payload_last.set(elapsed); + self.gas_bucket.record(gas_used, elapsed); } } PayloadStatusEnum::Syncing => self.new_payload_syncing.increment(1), diff --git a/crates/engine/tree/src/tree/mod.rs b/crates/engine/tree/src/tree/mod.rs index e7188f75016..e5230db98b6 100644 --- a/crates/engine/tree/src/tree/mod.rs +++ b/crates/engine/tree/src/tree/mod.rs @@ -1608,18 +1608,16 @@ where warn!(target: "engine::tree", ?state, elapsed=?start.elapsed(), "Failed to deliver forkchoiceUpdated response, receiver dropped (request cancelled): {err:?}"); } } - BeaconEngineMessage::NewPayload { payload, tx, enqueued_at } => { + BeaconEngineMessage::NewPayload { payload, tx } => { let start = Instant::now(); let gas_used = payload.gas_used(); let num_hash = payload.num_hash(); let mut output = self.on_new_payload(payload); - let latency = enqueued_at.elapsed(); self.metrics.engine.new_payload.update_response_metrics( start, &mut self.metrics.engine.forkchoice_updated.latest_finish_at, &output, gas_used, - latency, ); let maybe_event = @@ -1695,20 +1693,12 @@ where let gas_used = payload.gas_used(); let num_hash = payload.num_hash(); let mut output = self.on_new_payload(payload); - - // Latency measures time from enqueue to completion, excluding - // only the explicit persistence wait. This means backpressure - // (time spent queued due to the engine being busy) is included, - // reflecting real-world engine responsiveness. - let latency = - enqueued_at.elapsed().saturating_sub(explicit_persistence_wait); - + let latency = start.elapsed(); self.metrics.engine.new_payload.update_response_metrics( start, &mut self.metrics.engine.forkchoice_updated.latest_finish_at, &output, gas_used, - latency, ); let maybe_event = diff --git a/crates/engine/tree/src/tree/tests.rs b/crates/engine/tree/src/tree/tests.rs index 23dcdd217cc..bbb6c4c59b1 100644 --- a/crates/engine/tree/src/tree/tests.rs +++ b/crates/engine/tree/src/tree/tests.rs @@ -682,7 +682,6 @@ async fn test_holesky_payload() { sidecar: ExecutionPayloadSidecar::none(), }, tx, - enqueued_at: std::time::Instant::now(), } .into(), )) diff --git a/crates/engine/util/src/reorg.rs b/crates/engine/util/src/reorg.rs index 8d33559a5a0..adfec89ba0e 100644 --- a/crates/engine/util/src/reorg.rs +++ b/crates/engine/util/src/reorg.rs @@ -25,7 +25,6 @@ use std::{ future::Future, pin::Pin, task::{ready, Context, Poll}, - time::Instant, }; use tokio::sync::oneshot; use tracing::*; @@ -144,7 +143,7 @@ where let next = ready!(this.stream.poll_next_unpin(cx)); let item = match (next, &this.last_forkchoice_state) { ( - Some(BeaconEngineMessage::NewPayload { payload, tx, enqueued_at }), + Some(BeaconEngineMessage::NewPayload { payload, tx }), Some(last_forkchoice_state), ) if this.forkchoice_states_forwarded > this.frequency && // Only enter reorg state if new payload attaches to current head. @@ -174,7 +173,6 @@ where return Poll::Ready(Some(BeaconEngineMessage::NewPayload { payload, tx, - enqueued_at, })) } }; @@ -193,12 +191,11 @@ where let queue = VecDeque::from([ // Current payload - BeaconEngineMessage::NewPayload { payload, tx, enqueued_at }, + BeaconEngineMessage::NewPayload { payload, tx }, // Reorg payload BeaconEngineMessage::NewPayload { payload: T::block_to_payload(reorg_block), tx: reorg_payload_tx, - enqueued_at: Instant::now(), }, // Reorg forkchoice state BeaconEngineMessage::ForkchoiceUpdated { diff --git a/crates/engine/util/src/skip_new_payload.rs b/crates/engine/util/src/skip_new_payload.rs index 22c637d5d58..fa818f54538 100644 --- a/crates/engine/util/src/skip_new_payload.rs +++ b/crates/engine/util/src/skip_new_payload.rs @@ -41,7 +41,7 @@ where loop { let next = ready!(this.stream.poll_next_unpin(cx)); let item = match next { - Some(BeaconEngineMessage::NewPayload { payload, tx, enqueued_at }) => { + Some(BeaconEngineMessage::NewPayload { payload, tx }) => { if this.skipped < this.threshold { *this.skipped += 1; tracing::warn!( @@ -56,7 +56,7 @@ where continue } *this.skipped = 0; - Some(BeaconEngineMessage::NewPayload { payload, tx, enqueued_at }) + Some(BeaconEngineMessage::NewPayload { payload, tx }) } next => next, }; diff --git a/examples/bsc-p2p/src/block_import/service.rs b/examples/bsc-p2p/src/block_import/service.rs index 63d368e979c..4e4029d7e95 100644 --- a/examples/bsc-p2p/src/block_import/service.rs +++ b/examples/bsc-p2p/src/block_import/service.rs @@ -413,7 +413,7 @@ mod tests { tokio::spawn(async move { while let Some(message) = from_engine.recv().await { match message { - BeaconEngineMessage::NewPayload { payload: _, tx, .. } => { + BeaconEngineMessage::NewPayload { payload: _, tx } => { tx.send(Ok(PayloadStatus::new(responses.new_payload.clone(), None))) .unwrap(); }