perf(blockchain): run the beacon state transition while the engine verdict is in flight

crates/blockchain/src/beacon_engine.rs

@@ -19,6 +19,8 @@
 //! Gloas is the exception: its payload is revealed in an envelope, so its
 //! request ([`gloas_new_payload_request`]) needs the block and the envelope.
 
+use std::time::Instant;
+
 use ethlambda_engine::types::PayloadStatusValue;
 use ethlambda_engine::{EngineClient, EngineError, PayloadStatusV1};
 use ethlambda_state_transition::beacon::containers::{self, SignedBeaconBlock};
@@ -178,6 +180,87 @@ pub fn verdict(status: &PayloadStatusV1) -> PayloadValidity {
     })
 }
 
+/// [`NewPayloadRequest`] with its own copy of the payload, so it can move into
+/// a spawned task.
+///
+/// The chain actor sends the question from a task of its own and runs the
+/// block's state transition meanwhile (see `process_block`), and a task cannot
+/// borrow the block the actor goes on using. Only the payload is copied: the
+/// rest of the request is already owned, and the block itself, which also
+/// carries the attestations and the other operations, is never cloned.
+pub struct OwnedNewPayloadRequest {
+    execution_payload: containers::deneb::ExecutionPayload,
+    versioned_hashes: Vec<Bytes32>,
+    parent_beacon_block_root: Root,
+    execution_requests: Vec<Vec<u8>>,
+}
+
+impl NewPayloadRequest<'_> {
+    /// Copies the borrowed payload so the request outlives the block.
+    pub fn into_owned(self) -> OwnedNewPayloadRequest {
+        OwnedNewPayloadRequest {
+            execution_payload: self.execution_payload.clone(),
+            versioned_hashes: self.versioned_hashes,
+            parent_beacon_block_root: self.parent_beacon_block_root,
+            execution_requests: self.execution_requests,
+        }
+    }
+}
+
+/// [`ask`] for a request already built with [`new_payload_request`], which is
+/// how the actor sends it from a task while it computes the transition.
+///
+/// The answer is read exactly as [`ask`] reads it: `Err` means no answer after
+/// the client's whole retry ladder, and there is no `Ok(None)` because a
+/// request existing means there was something to ask.
+pub async fn ask_request(
+    client: &EngineClient,
+    request: OwnedNewPayloadRequest,
+) -> Result<PayloadValidity, EngineError> {
+    let call = client.new_payload(
+        &request.execution_payload,
+        &request.versioned_hashes,
+        request.parent_beacon_block_root,
+        &request.execution_requests,
+    );
+    let status = timed_new_payload(call).await?;
+    Ok(verdict(&status))
+}
+
+/// Awaits `call`, one `engine_newPayload*` request, and observes the time from
+/// sending it to its answer as `lean_engine_new_payload_seconds`, under the
+/// [`outcome_label`] of the verdict the answer reads as.
+///
+/// Every request this module sends goes through here, a block's payload and an
+/// envelope's alike, so the metric is the execution client's time for every
+/// payload it was asked about. A request refused before it is sent (an
+/// envelope that does not match its block) never reaches it.
+async fn timed_new_payload(
+    call: impl Future<Output = Result<PayloadStatusV1, EngineError>>,
+) -> Result<PayloadStatusV1, EngineError> {
+    let sent = Instant::now();
+    let status = call.await;
+    let validity = status.as_ref().ok().map(verdict);
+    let outcome = outcome_label(validity.as_ref());
+    crate::metrics::observe_engine_new_payload(outcome, sent.elapsed());
+    status
+}
+
+/// The `outcome` an answer is counted under: its verdict, or `no_verdict`
+/// when the retry ladder ran out without one.
+///
+/// `not_required` is listed for completeness: a status from the execution
+/// client always reads as one of the other three verdicts.
+pub fn outcome_label(verdict: Option<&PayloadValidity>) -> &'static str {
+    match verdict {
+        Some(PayloadValidity::Validated) => "valid",
+        Some(PayloadValidity::Optimistic) => "optimistic",
+        Some(PayloadValidity::Invalidated { .. }) => "invalidated",
+        Some(PayloadValidity::NotRequired) => "not_required",
+        None => "no_verdict",
+    }
+}
+
 /// Asks the execution client about `block`'s payload.
 ///
 /// `Ok(None)` means there was nothing to ask, which is not a failure:
@@ -192,14 +275,13 @@ pub async fn ask(
     let Some(request) = new_payload_request(block) else {
         return Ok(None);
     };
-    let status = client
-        .new_payload(
-            request.execution_payload,
-            &request.versioned_hashes,
-            request.parent_beacon_block_root,
-            &request.execution_requests,
-        )
-        .await?;
+    let call = client.new_payload(
+        request.execution_payload,
+        &request.versioned_hashes,
+        request.parent_beacon_block_root,
+        &request.execution_requests,
+    );
+    let status = timed_new_payload(call).await?;
     Ok(Some(verdict(&status)))
 }
 
@@ -230,14 +312,13 @@ pub async fn ask_envelope_status(
     envelope: &containers::gloas::SignedExecutionPayloadEnvelope,
 ) -> Result<PayloadStatusV1, EngineError> {
     let request = gloas_new_payload_request(block, envelope)?;
-    client
-        .new_payload_v5(
-            request.execution_payload,
-            &request.versioned_hashes,
-            request.parent_beacon_block_root,
-            &request.execution_requests,
-        )
-        .await
+    let call = client.new_payload_v5(
+        request.execution_payload,
+        &request.versioned_hashes,
+        request.parent_beacon_block_root,
+        &request.execution_requests,
+    );
+    timed_new_payload(call).await
 }
 
 #[cfg(test)]

crates/blockchain/src/import_timing.rs

@@ -176,8 +176,18 @@ pub struct ImportTimings {
     pub columns_wait_start: Option<Instant>,
     pub columns_wait_end: Option<Instant>,
 
-    /// Beacon: the `engine_newPayload` round trip, including its retry ladder.
-    /// I/O wait, not work.
+    /// Beacon: how long the import sat waiting for the execution client's
+    /// verdict with no consensus work of its own left to do. I/O wait, not
+    /// work.
+    ///
+    /// The request is sent from a task of its own while the state transition
+    /// runs (see [`Self::stf_spec_start`]), so this pair starts where that
+    /// transition ended and ends when the answer arrived. Both are the same
+    /// instant, a recorded zero, when there was no request to send (no
+    /// execution client, or a block with no payload in it: pre-merge or
+    /// gloas) or when the answer was already in once the transition finished.
+    /// How long the execution client itself took is
+    /// `lean_engine_new_payload_seconds`, not this.
     pub engine_start: Option<Instant>,
     pub engine_end: Option<Instant>,
 
@@ -197,6 +207,22 @@ pub struct ImportTimings {
     pub stf_start: Option<Instant>,
     pub stf_end: Option<Instant>,
 
+    /// Beacon: a state transition computed while the execution client's
+    /// verdict was still in flight, whether the import then used it or not.
+    ///
+    /// It runs before the `stf` pair above and overlaps the `engine_newPayload`
+    /// round trip, so it is neither inside `stf_start`..`stf_end` nor a
+    /// section of its own: the `stf` row adds it, since it is state-transition
+    /// work. A discarded one is added too, so the transition an `INVALIDATED`
+    /// verdict makes `on_block` run again shows as the second transition it
+    /// is, and a held block still shows the one it paid for. The `engine` pair
+    /// starts at this pair's end, so the rows still sum to the end-to-end time
+    /// instead of counting the overlap twice. `None` whenever no request was
+    /// sent, which is every import without an execution client and every
+    /// block with no payload to ask about.
+    pub stf_spec_start: Option<Instant>,
+    pub stf_spec_end: Option<Instant>,
+
     /// Beacon: the blocking hand-off of the post-state to the storage crate's
     /// background writer, which waits while the writer's queue is full. It
     /// lies inside the `stf` pair; only the report separates the two.
@@ -376,6 +402,12 @@ impl ImportTimings {
         if let (Some(total), Some(wait)) = (stf.elapsed, writer_wait.elapsed) {
             stf.elapsed = Some(total.saturating_sub(wait));
         }
+        // Also when the import stopped before running its own transition (a
+        // held block), so the speculation it paid for is not lost.
+        if let Some(spec) = span(self.stf_spec_start, self.stf_spec_end) {
+            stf.elapsed = Some(stf.elapsed.unwrap_or_default() + spec);
+            stf.end = stf.end.or(self.stf_spec_end);
+        }
         vec![
             Row::new("decode", self.decode_start, self.decode_end),
             Row::new("queue", self.queue_start, self.queue_end),
@@ -897,6 +929,67 @@ mod tests {
         assert_eq!(row_elapsed(&timings, "writer_wait"), None);
     }
 
+    #[test]
+    fn an_overlapped_transition_is_charged_to_stf_and_not_to_engine() {
+        // A 100 ms round trip with a 70 ms transition running inside it, then
+        // 20 ms of import. Un-overlapped, that is 100 + 20 ms; overlapped, the
+        // engine row keeps the 30 ms that was left and stf gains the 70 ms.
+        let base = Instant::now();
+        let mut timings = ImportTimings::default();
+        (timings.stf_spec_start, timings.stf_spec_end) = pair(base, 0, 70);
+        (timings.engine_start, timings.engine_end) = pair(base, 70, 30);
+        (timings.stf_start, timings.stf_end) = pair(base, 100, 20);
+
+        assert_eq!(
+            row_elapsed(&timings, "engine"),
+            Some(Duration::from_millis(30))
+        );
+        assert_eq!(
+            row_elapsed(&timings, "stf"),
+            Some(Duration::from_millis(90))
+        );
+        let rows: Duration = timings
+            .rows()
+            .into_iter()
+            .filter_map(|row| row.elapsed)
+            .sum();
+        assert_eq!(
+            rows,
+            Duration::from_millis(120),
+            "the rows still tile the span"
+        );
+    }
+
+    #[test]
+    fn a_held_block_still_charges_its_speculation_to_stf() {
+        // No verdict: the 70 ms transition ran, then 300 ms of waiting, then
+        // the block was held without a transition of its own.
+        let base = Instant::now();
+        let mut timings = ImportTimings::default();
+        (timings.stf_spec_start, timings.stf_spec_end) = pair(base, 0, 70);
+        (timings.engine_start, timings.engine_end) = pair(base, 70, 300);
+
+        assert_eq!(
+            row_elapsed(&timings, "stf"),
+            Some(Duration::from_millis(70))
+        );
+        assert_eq!(
+            row_elapsed(&timings, "engine"),
+            Some(Duration::from_millis(300))
+        );
+    }
+
+    #[test]
+    fn an_engine_with_nothing_to_wait_for_is_a_recorded_zero() {
+        // No request sent, or the answer was in before the transition ended:
+        // the row is present and zero, so every beacon block observes it.
+        let base = Instant::now();
+        let mut timings = ImportTimings::default();
+        (timings.engine_start, timings.engine_end) = (Some(base), Some(base));
+
+        assert_eq!(row_elapsed(&timings, "engine"), Some(Duration::ZERO));
+    }
+
     #[test]
     fn a_writer_wait_longer_than_stf_saturates_stf_to_zero() {
         let base = Instant::now();