diff --git a/CLAUDE.md b/CLAUDE.md index d08ee404f..e7f659038 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -83,6 +83,20 @@ Fork choice head update (Store buffer fields are `new_payloads`/`known_payloads`; the accessors are named `extract_latest_new_attestations`/`extract_latest_known_attestations`.) +### Asynchronous `engine_newPayload` (beacon) +`crates/blockchain/src/beacon_new_payload.rs`. On the running actor a pre-gloas +block (optimistic candidate) imports as optimistic and a gloas envelope applies +as `NOT_VALIDATED` before the EL answers; `engine_newPayload` goes out first from +a tokio task that reports back with a `NewPayloadAnswered` message, applied as +`optimistic-sync.md` applies a verdict. Sent parent-before-child, at most +`MAX_NEW_PAYLOADS_IN_FLIGHT` in flight; `forkchoiceUpdated` names the newest +payload on the head's chain whose request is not queued or in flight (so catch-up +still feeds the EL a head); at most `MAX_PAYLOADS_AWAITING_VERDICT` wait (past +that the import is held). No answer: +kept optimistic, re-asked once a slot (a documented deviation). Without a +mailbox (offline replay) and for non-candidates the import still waits inline. +The actor records its own `ActorRef` in a `#[started]` hook for this. + ### State Transition Phases 1. **process_slots()**: Advance through empty slots, update historical roots 2. **process_block()**: Validate header → process attestations → update justifications/finality @@ -819,10 +833,13 @@ transitions are in `ethlambda-types`, per the section above. Nothing above the duties. `attestation_data.index` at a gloas slot comes from `Store::head_payload_status`. With `--execution-endpoint`, an envelope that passes the consensus checks is - judged by `engine_newPayloadV5` (INVALID with a null `latestValidHash` is a - hash mismatch that refuses only that envelope; INVALID with one condemns the - payload via the execution-chain walk; an engine error keeps the envelope and - retries once a slot, probing with the oldest), and a gloas head sends + applied at once (payload `NOT_VALIDATED`) and judged by `engine_newPayloadV5` + without waiting, like a pre-gloas block (see `beacon_new_payload` below): + INVALID with a null `latestValidHash` is a hash mismatch (the only answer + Engine API V2+ gives one) that takes the envelope back out and refuses only + that copy; INVALID with one condemns the payload via the execution-chain + walk; no answer keeps it applied and re-asks once a slot. Blocks built on a + FULL payload wait for its answer, not just its envelope. A gloas head sends `engine_forkchoiceUpdatedV4` with the custody columns (head hash by the head node's FULL or EMPTY status, finalized and safe from the checkpoint blocks' `bid.parent_block_hash`, safe being a fallback for fast confirmation). Building diff --git a/crates/blockchain/src/beacon_envelope.rs b/crates/blockchain/src/beacon_envelope.rs index a893a651c..d607ce53d 100644 --- a/crates/blockchain/src/beacon_envelope.rs +++ b/crates/blockchain/src/beacon_envelope.rs @@ -470,6 +470,29 @@ impl BlockChainServer { // for every pre-gloas block it imports (`NotRequired`). let status = match self.engine.clone() { None => PayloadStatusEnum::Valid, + // The running actor applies the envelope now, its payload + // `NOT_VALIDATED`, and applies the answer when it arrives (see + // `beacon_new_payload`). Blocks built on this payload wait for + // that answer, not for this, so a refused envelope can be taken + // back out with nothing on top of it. + Some(_) if self.answers_payloads_later() => { + if let Err(err) = beacon_engine::gloas_new_payload_request(&block, &held.envelope) { + warn!( + %slot, + block_root = %ShortRoot(&root.0), + %err, + "Dropping an envelope that does not match its block" + ); + self.refetch_envelope_if_children_wait(root); + return false; + } + if !self.new_payloads.has_room() { + self.hold_envelope_awaiting_engine(root, slot, held); + return false; + } + self.ask_about_envelope_payload(&block, &held.envelope, held.envelope_root); + PayloadStatusEnum::Syncing + } Some(client) => { let asked = beacon_engine::ask_envelope_status(&client, &block, &held.envelope).await; @@ -483,11 +506,12 @@ impl BlockChainServer { // this envelope is bad, the payload the bid names may not // be. Only this envelope is refused, so the builder's real // one can still arrive. - // Some execution clients have no `INVALID_BLOCK_HASH` and - // answer a hash mismatch with `INVALID` and a null - // `latestValidHash` (ethrex does), whereas a payload that - // executed and failed names its last valid ancestor. So a - // null one is read the same way. + // Since `engine_newPayloadV2` a hash mismatch is answered + // `INVALID` with a null `latestValidHash` + // (`INVALID_BLOCK_HASH` is "supplanted by `INVALID`", + // execution-apis `shanghai.md`), whereas a payload that + // executed and failed normally names its last valid + // ancestor. So a null one is read the same way. Ok(raw) if raw.status == PayloadStatusValue::InvalidBlockHash || (raw.status == PayloadStatusValue::Invalid @@ -573,10 +597,53 @@ impl BlockChainServer { slot, held.envelope.message.payload.block_hash, ); - self.envelopes.verified.push_back(root); + // The children waiting on this payload go once it is answered, which + // is now unless the question is still out (then + // `beacon_new_payload` releases them). + if !self.new_payloads.awaits_envelope_answer(&root) { + self.envelopes.verified.push_back(root); + } true } + /// Release the blocks waiting on `root`'s payload, now that the execution + /// client answered for it (`VALID`, or `SYNCING`/`ACCEPTED`). + pub(crate) fn release_children_of_answered_payload(&mut self, root: H256) { + self.envelopes.verified.push_back(root); + } + + /// Take an applied envelope back out: the execution client refused its + /// contents (`INVALID` with no `latestValidHash`), which says this copy is + /// bad, not that the payload its bid commits to is. + /// + /// The block keeps its EMPTY branch and goes back to having no payload, so + /// a correct envelope can still be applied; this one is refused if it + /// comes again. Nothing was built on the FULL node, since blocks building + /// on it wait for the answer (`parent_payload_unverified`), and those + /// waiting keep waiting: the envelope is asked for again for them. + /// Lighthouse and Prysm reach the same state by never applying such an + /// envelope, since they wait for the answer first. + pub(crate) fn revert_envelope(&mut self, root: H256, envelope_root: H256) { + warn!( + block_root = %ShortRoot(&root.0), + envelope_root = %ShortRoot(&envelope_root.0), + "Execution client refused an envelope's contents; taking it back out" + ); + self.store.remove_verified_payload(&root); + self.store.remove_beacon_block_payload_status(root); + self.store.remove_beacon_envelope_seen(&root); + self.envelopes.verified.retain(|verified| *verified != root); + self.envelopes.reject(envelope_root); + self.envelopes.head_dirty = true; + self.refetch_envelope_if_children_wait(root); + } + + /// Queue every envelope held because too many payloads were waiting for + /// the execution client, now that there is room again. + pub(crate) fn release_envelopes_awaiting_engine_room(&mut self) { + self.release_envelopes_awaiting_engine(); + } + /// Keep a consensus-valid envelope until the execution client answers. fn hold_envelope_awaiting_engine(&mut self, root: H256, slot: u64, mut held: HeldEnvelope) { held.slot = slot; @@ -942,14 +1009,23 @@ impl BlockChainServer { /// Whether `block` is a gloas block whose parent is FULL while that /// parent's payload is not yet verified: `on_block` would reject it with /// `is_payload_verified(store, block.parent_root)`. + /// + /// Also while the parent's envelope is applied but the execution client + /// has not answered for its payload yet. `on_block` would accept the block + /// then, but an envelope refused for its contents is taken back out (see + /// `revert_envelope`), and nothing may be built on it by that time. + /// Lighthouse and Prysm hold the same blocks, by applying an envelope only + /// after its answer. pub(crate) fn parent_payload_unverified(&self, block: &SignedBeaconBlock) -> bool { if !block.fork_name().has_payload_envelopes() { return false; } + let parent_root = block.parent_root(); matches!( fork_choice::get_parent_payload_status(&self.store, block), Ok(PayloadStatus::Full) - ) && !fork_choice::is_payload_verified(&self.store, block.parent_root()) + ) && (!fork_choice::is_payload_verified(&self.store, parent_root) + || self.new_payloads.awaits_envelope_answer(&parent_root)) } /// Keep `block` until its parent's envelope is verified. @@ -3029,8 +3105,9 @@ mod tests { ignore = "needs the fork choice fixtures; run `make consensus-spec-tests`" )] async fn an_invalid_verdict_with_a_null_latest_valid_hash_rejects_only_that_envelope() { - // How ethrex reports a payload whose contents do not hash to its block - // hash: INVALID, `latestValidHash` null. + // How the Engine API reports a payload whose contents do not hash to + // its block hash since `engine_newPayloadV2`: INVALID, + // `latestValidHash` null. let (client, mock) = mock_engine(status_json("INVALID", None), status_json("SYNCING", None)).await; let mut server = server_with_held_child(client).await; @@ -3054,6 +3131,196 @@ mod tests { assert_eq!(count(&mock, "engine_newPayloadV5"), 1); } + /// The running actor's path: the envelope is applied at once and the + /// answer comes back as a message, which the test hands over itself. + fn answer_later( + server: &mut BlockChainServer, + ) -> tokio::sync::mpsc::UnboundedReceiver { + let (tx, rx) = tokio::sync::mpsc::unbounded_channel(); + server.verdict_sink = Some(crate::beacon_new_payload::VerdictSink::Channel(tx)); + rx + } + + #[tokio::test] + #[cfg_attr( + not(feature = "beacon-spec-tests"), + ignore = "needs the fork choice fixtures; run `make consensus-spec-tests`" + )] + async fn an_envelope_applies_before_its_answer_and_its_full_child_waits_for_it() { + let (client, mock) = + mock_engine(status_json("VALID", None), status_json("SYNCING", None)).await; + let mut server = server_with_held_child(client).await; + let mut answers = answer_later(&mut server); + + server + .receive_envelope(valid_envelope_for_full_parent(), 0) + .await; + server.settle_envelopes().await; + + // Applied, not yet judged: the FULL node exists, the child still waits. + assert!(server.store.has_verified_payload(&root(FULL_PARENT))); + assert!( + server + .store + .beacon_block_payload_status(root(FULL_PARENT)) + .is_not_validated() + ); + assert!(server.envelopes.is_block_held(&root(FULL_CHILD))); + + let answered = answers.recv().await.expect("an answer"); + server.on_new_payload_answered(answered).await; + + assert_eq!( + server.store.beacon_block_payload_status(root(FULL_PARENT)), + PayloadStatusEnum::Valid + ); + assert!(server.store.has_state(&root(FULL_CHILD)).unwrap()); + assert_eq!(count(&mock, "engine_newPayloadV5"), 1); + } + + /// The other order: the child arrives after the envelope is applied but + /// before its answer. `on_block` alone would take it then; it waits. + #[tokio::test] + #[cfg_attr( + not(feature = "beacon-spec-tests"), + ignore = "needs the fork choice fixtures; run `make consensus-spec-tests`" + )] + async fn a_child_arriving_while_its_parent_payload_awaits_an_answer_is_held() { + let (client, _mock) = + mock_engine(status_json("VALID", None), status_json("SYNCING", None)).await; + let case = Case::new("on_execution_payload_envelope", FULL_CHILD_CASE); + let mut server = case.server(12); + server.engine = Some(client); + let mut answers = answer_later(&mut server); + server + .on_block(case.block(FULL_PARENT), ImportTimings::default()) + .await; + fork_choice::on_tick(&mut server.store, 24, &fixture_config()); + server + .receive_envelope(valid_envelope_for_full_parent(), 0) + .await; + server.settle_envelopes().await; + assert!(server.store.has_verified_payload(&root(FULL_PARENT))); + + server + .on_block(case.block(FULL_CHILD), ImportTimings::default()) + .await; + assert!(server.envelopes.is_block_held(&root(FULL_CHILD))); + assert!(!server.store.has_state(&root(FULL_CHILD)).unwrap()); + + let answered = answers.recv().await.expect("an answer"); + server.on_new_payload_answered(answered).await; + assert!(server.store.has_state(&root(FULL_CHILD)).unwrap()); + } + + #[tokio::test] + #[cfg_attr( + not(feature = "beacon-spec-tests"), + ignore = "needs the fork choice fixtures; run `make consensus-spec-tests`" + )] + async fn an_envelope_refused_for_its_contents_is_taken_back_out() { + let (client, mock) = + mock_engine(status_json("INVALID", None), status_json("SYNCING", None)).await; + let mut server = server_with_held_child(client).await; + let mut answers = answer_later(&mut server); + let envelope = valid_envelope_for_full_parent(); + + server.receive_envelope(envelope.clone(), 0).await; + server.settle_envelopes().await; + assert!(server.store.has_verified_payload(&root(FULL_PARENT))); + + let answered = answers.recv().await.expect("an answer"); + server.on_new_payload_answered(answered).await; + + // Back to no payload, not to an invalid one: a correct envelope can + // still be applied, and the child keeps waiting for it. + assert!(!server.store.has_verified_payload(&root(FULL_PARENT))); + assert!( + !server + .store + .beacon_block_payload_status(root(FULL_PARENT)) + .is_invalidated() + ); + assert_eq!( + server.store.beacon_envelope_seen_ms(&root(FULL_PARENT)), + None + ); + assert!(server.envelopes.is_block_held(&root(FULL_CHILD))); + assert_eq!(head(&server).1, PayloadStatus::Empty); + + // The same copy is refused if it comes again, without asking. + server.receive_envelope(envelope, 0).await; + server.settle_envelopes().await; + assert!(!server.store.has_verified_payload(&root(FULL_PARENT))); + assert_eq!(count(&mock, "engine_newPayloadV5"), 1); + } + + #[tokio::test] + #[cfg_attr( + not(feature = "beacon-spec-tests"), + ignore = "needs the fork choice fixtures; run `make consensus-spec-tests`" + )] + async fn an_answer_that_condemns_the_payload_drops_the_child_waiting_on_it() { + // A non-null `latestValidHash`: the payload executed and failed. + let (client, _mock) = mock_engine( + status_json("INVALID", Some(H256::repeat_byte(0x11))), + status_json("SYNCING", None), + ) + .await; + let mut server = server_with_held_child(client).await; + let mut answers = answer_later(&mut server); + + server + .receive_envelope(valid_envelope_for_full_parent(), 0) + .await; + server.settle_envelopes().await; + let answered = answers.recv().await.expect("an answer"); + server.on_new_payload_answered(answered).await; + + assert!(!server.store.has_verified_payload(&root(FULL_PARENT))); + assert_eq!( + server.store.beacon_block_payload_status(root(FULL_PARENT)), + PayloadStatusEnum::Invalid + ); + assert!(!server.envelopes.is_block_held(&root(FULL_CHILD))); + assert!(!server.store.has_state(&root(FULL_CHILD)).unwrap()); + assert_eq!(head(&server), (root(FULL_PARENT), PayloadStatus::Empty)); + } + + #[tokio::test] + #[cfg_attr( + not(feature = "beacon-spec-tests"), + ignore = "needs the fork choice fixtures; run `make consensus-spec-tests`" + )] + async fn no_answer_keeps_the_envelope_and_its_child_waiting() { + let (client, mock) = mock_engine("ERROR".to_string(), status_json("SYNCING", None)).await; + let mut server = server_with_held_child(client).await; + let mut answers = answer_later(&mut server); + + server + .receive_envelope(valid_envelope_for_full_parent(), 0) + .await; + server.settle_envelopes().await; + let answered = answers.recv().await.expect("an answer"); + server.on_new_payload_answered(answered).await; + + assert!(server.store.has_verified_payload(&root(FULL_PARENT))); + assert!(server.envelopes.is_block_held(&root(FULL_CHILD))); + assert!( + server + .new_payloads + .awaits_envelope_answer(&root(FULL_PARENT)) + ); + + // Asked again on the next slot; this time the client answers. + mock.lock().unwrap().new_payload = status_json("SYNCING", None); + server.redrive_new_payloads().await; + let answered = answers.recv().await.expect("a second answer"); + server.on_new_payload_answered(answered).await; + assert!(server.store.has_state(&root(FULL_CHILD)).unwrap()); + assert!(count(&mock, "engine_newPayloadV5") >= 2); + } + #[tokio::test] #[cfg_attr( not(feature = "beacon-spec-tests"), diff --git a/crates/blockchain/src/beacon_new_payload.rs b/crates/blockchain/src/beacon_new_payload.rs new file mode 100644 index 000000000..2b962c163 --- /dev/null +++ b/crates/blockchain/src/beacon_new_payload.rs @@ -0,0 +1,1076 @@ +//! Asking the execution client about payloads without waiting for the answer. +//! +//! A pre-gloas block, and from gloas on an envelope, enters fork choice as soon +//! as its consensus checks pass, with its payload marked as not yet validated +//! (`optimistic-sync.md`'s `NOT_VALIDATED`). The `engine_newPayload` question +//! goes out first, from a tokio task of its own, and the answer comes back to +//! the actor as a [`NewPayloadAnswered`] message, where it is applied the way +//! the specification applies a `newPayload` verdict: +//! +//! | Answer | Pre-gloas block | Gloas payload | +//! |---|---|---| +//! | `VALID` | validated, with its optimistic ancestors | `VALID`, held children released | +//! | `SYNCING` / `ACCEPTED` | stays optimistic | stays `NOT_VALIDATED`, held children released | +//! | `INVALID` with a hash | removed with its subtree | condemned (see `condemn_payload_chain`) | +//! | `INVALID` with no hash | as above | taken back out: only that envelope is refused | +//! | no answer | kept, asked again next slot | kept, asked again next slot | +//! +//! ```text +//! import ─► queued ─(parent answered, < MAX in flight)─► in flight ─► answered +//! ▲ │ no answer +//! └──────────────── next slot ── unanswered ◄────┘ +//! ``` +//! +//! # Order +//! +//! The Engine API orders only `forkchoiceUpdated` (`common.md`, "Message +//! ordering"), and the engine client's connection pool can deliver two +//! in-flight requests in either order. A payload whose parent the execution +//! client has not seen is answered `SYNCING`, and the client may start a sync +//! from its peers for it. So a payload is sent only once no request for the +//! payload it builds on is outstanding, and at most +//! [`MAX_NEW_PAYLOADS_IN_FLIGHT`] at once: siblings and forks go in parallel, a +//! linear chain goes one at a time. For the same reason `forkchoiceUpdated` +//! never names a payload whose request is still queued or in flight: it names +//! the newest one on the head's chain the client was asked about (see +//! [`NewPayloads::forkchoice_head`]), and is sent again when the head's own +//! answer arrives. +//! +//! # Children of a gloas payload +//! +//! A block that builds on a FULL parent waits until the parent payload's +//! answer is in, not merely until its envelope is applied, as Lighthouse and +//! Prysm do (neither applies an envelope before the answer). That is what lets +//! a refused envelope be taken back out without anything built on it. +//! +//! # Bounds +//! +//! At most [`MAX_PAYLOADS_AWAITING_VERDICT`] payloads wait for an answer. Past +//! that the import itself waits: a pre-gloas block in +//! [`NewPayloads::blocks_awaiting_room`], an envelope in `awaiting_engine`, +//! each released as answers free room. That only happens during catch-up, +//! where the execution client is the slower side. +//! +//! # Deviation +//! +//! `optimistic-sync.md` says a block the execution client gave no answer for +//! MUST NOT be applied to fork choice. Here it already is, so it stays, +//! optimistic, and is asked about again once a slot. See +//! `docs/spec_deviations.md`. +//! +//! Without a way to hear back (offline replay builds the server without an +//! actor) the import waits for the answer inline, as it always did. + +use std::collections::{HashMap, HashSet, VecDeque}; +use std::sync::Arc; + +use ethlambda_engine::types::PayloadStatusValue; +use ethlambda_engine::{EngineError, PayloadStatusV1}; +use ethlambda_state_transition::beacon::containers::{SignedBeaconBlock, gloas}; +use ethlambda_state_transition::beacon::fork_choice::{self, PayloadStatusEnum, PayloadValidity}; +use ethlambda_types::{ShortRoot, primitives::H256}; +use spawned_concurrency::message::Message; +use spawned_concurrency::tasks::{ActorRef, Context, Handler}; +use tracing::{info, warn}; + +use crate::import_timing::ImportTimings; +use crate::{BlockChainServer, beacon_engine, metrics}; + +/// How many `engine_newPayload` requests may be in flight at once. +/// +/// The Engine API sets no limit. Execution clients validate one payload at a +/// time on the canonical chain, so more in flight only queues work on their +/// side; a few let side branches and the next request's transfer overlap the +/// current validation. +pub(crate) const MAX_NEW_PAYLOADS_IN_FLIGHT: usize = 4; + +/// How many imported payloads may wait for an answer, in flight or not. +/// +/// One range-sync batch: enough that catch-up keeps the execution client +/// busy, small enough that the consensus side cannot run unboundedly far +/// ahead of what has been validated. +pub(crate) const MAX_PAYLOADS_AWAITING_VERDICT: usize = 128; + +/// What a request is about. +pub(crate) enum PayloadQuestion { + /// A pre-gloas block, whose payload is in its body. + Block(SignedBeaconBlock), + /// A gloas envelope, with the block whose bid it fulfils. + Envelope { + block: SignedBeaconBlock, + envelope: Box, + envelope_root: H256, + }, +} + +impl PayloadQuestion { + fn kind(&self) -> PayloadKind { + match self { + Self::Block(_) => PayloadKind::Block, + Self::Envelope { .. } => PayloadKind::Envelope, + } + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum PayloadKind { + Block, + Envelope, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum AskState { + /// Waiting for its turn: room in flight, and its parent payload answered. + Queued, + /// Sent; the task holding it has not reported back. + InFlight, + /// The last request got no answer after the whole retry ladder; asked + /// again on the next slot. + Unanswered, +} + +struct Ask { + slot: u64, + block_hash: H256, + parent_hash: H256, + state: AskState, + question: Arc, +} + +/// The execution client's answer, as the task that asked reports it. +pub(crate) enum PayloadAnswer { + Block(Result, EngineError>), + /// The raw status: an envelope's hash failure and an executed failure are + /// handled differently, and only the status tells them apart. + Envelope { + envelope_root: H256, + result: Result, + }, +} + +/// A request task's report to the actor. +pub(crate) struct NewPayloadAnswered { + pub(crate) root: H256, + pub(crate) answer: PayloadAnswer, +} + +impl Message for NewPayloadAnswered { + type Result = (); +} + +/// Where a request task sends its answer. +#[derive(Clone)] +pub(crate) enum VerdictSink { + Actor(ActorRef), + /// Tests drive the actor's methods directly and feed the answers back + /// themselves. + #[cfg(test)] + Channel(tokio::sync::mpsc::UnboundedSender), +} + +impl VerdictSink { + fn send(&self, answered: NewPayloadAnswered) { + match self { + Self::Actor(actor) => { + let _ = actor.send(answered); + } + #[cfg(test)] + Self::Channel(tx) => { + let _ = tx.send(answered); + } + } + } +} + +/// Every payload imported and not yet answered for, and what waits on them. +#[derive(Default)] +pub(crate) struct NewPayloads { + /// By the root of the block carrying (pre-gloas) or bidding for (gloas) + /// the payload. A root has at most one payload to ask about. + asks: HashMap, + /// Queued roots, in the order they were imported. + queue: VecDeque, + /// Roots whose task has not reported back, including ones forgotten since + /// (their answer is then ignored, but they still occupy a request slot). + in_flight: HashSet, + /// Pre-gloas blocks held because the waiting set was full, oldest first, + /// with their slots. Written to the store as pending blocks. + blocks_awaiting_room: VecDeque<(H256, u64)>, + /// Set when a `forkchoiceUpdated` was skipped because the payload it would + /// name had not reached the execution client yet. + forkchoice_deferred: bool, +} + +impl NewPayloads { + /// Whether another payload may be imported before an answer. + pub(crate) fn has_room(&self) -> bool { + self.asks.len() < MAX_PAYLOADS_AWAITING_VERDICT + } + + /// Whether `root`'s gloas payload is applied and its answer not in yet. + pub(crate) fn awaits_envelope_answer(&self, root: &H256) -> bool { + self.asks + .get(root) + .is_some_and(|ask| ask.question.kind() == PayloadKind::Envelope) + } + + /// The payload `forkchoiceUpdated` may name as the head, given the head's + /// own payload `block_hash`: that one, unless its request is queued or in + /// flight, and then the nearest payload it builds on whose request is not. + /// + /// Walks each held-back payload's parent hash, which is the chain the + /// execution client validates along. During catch-up the newest block's + /// request always waits behind its parent's, so skipping the call instead + /// would leave the client without a head for the whole catch-up. + pub(crate) fn forkchoice_head(&self, block_hash: H256) -> H256 { + // One that got no answer does not hold the head back: an execution + // client that cannot be reached answers neither call. + let mut head = block_hash; + for _ in 0..=self.asks.len() { + let Some(ask) = self + .asks + .values() + .find(|ask| ask.block_hash == head && ask.state != AskState::Unanswered) + else { + break; + }; + head = ask.parent_hash; + } + head + } + + pub(crate) fn defer_forkchoice_updated(&mut self) { + self.forkchoice_deferred = true; + } + + fn take_deferred_forkchoice_updated(&mut self) -> bool { + std::mem::take(&mut self.forkchoice_deferred) + } + + fn insert( + &mut self, + root: H256, + slot: u64, + block_hash: H256, + parent_hash: H256, + question: PayloadQuestion, + ) { + if self.asks.contains_key(&root) { + return; + } + self.queue.push_back(root); + self.asks.insert( + root, + Ask { + slot, + block_hash, + parent_hash, + state: AskState::Queued, + question: Arc::new(question), + }, + ); + } + + /// The oldest queued payload that may go now, marked in flight. + fn next_to_send(&mut self) -> Option<(H256, Arc)> { + if self.in_flight.len() >= MAX_NEW_PAYLOADS_IN_FLIGHT { + return None; + } + let position = self.queue.iter().position(|root| { + let Some(ask) = self.asks.get(root) else { + return false; + }; + !self + .asks + .values() + .any(|other| other.block_hash == ask.parent_hash) + })?; + let root = self.queue.remove(position)?; + let ask = self.asks.get_mut(&root)?; + ask.state = AskState::InFlight; + self.in_flight.insert(root); + Some((root, Arc::clone(&ask.question))) + } + + /// Frees `root`'s request slot. `true` if its ask is still tracked, so the + /// answer is to be applied. + fn finish(&mut self, root: &H256) -> bool { + self.in_flight.remove(root) + && self + .asks + .get(root) + .is_some_and(|ask| ask.state == AskState::InFlight) + } + + /// `root` was answered: it no longer waits. Its slot, if it was waiting. + fn complete(&mut self, root: &H256) -> Option { + self.asks.remove(root).map(|ask| ask.slot) + } + + fn mark_unanswered(&mut self, root: &H256) { + if let Some(ask) = self.asks.get_mut(root) { + ask.state = AskState::Unanswered; + } + } + + /// Put every unanswered ask back in the queue, oldest slot first. + fn requeue_unanswered(&mut self) { + let mut unanswered: Vec<(u64, H256)> = self + .asks + .iter() + .filter(|(_, ask)| ask.state == AskState::Unanswered) + .map(|(root, ask)| (ask.slot, *root)) + .collect(); + unanswered.sort(); + for (_, root) in unanswered { + if let Some(ask) = self.asks.get_mut(&root) { + ask.state = AskState::Queued; + } + self.queue.push_back(root); + } + } + + /// Stop waiting for `root`. An answer already on its way is ignored. + pub(crate) fn forget(&mut self, root: &H256) { + self.asks.remove(root); + self.queue.retain(|queued| queued != root); + } + + /// Stop waiting for every payload `keep` says no longer matters, given its + /// root, its block's slot and what it is. + fn retain(&mut self, mut keep: impl FnMut(&H256, u64, PayloadKind) -> bool) { + let dropped: Vec = self + .asks + .iter() + .filter(|(root, ask)| !keep(root, ask.slot, ask.question.kind())) + .map(|(root, _)| *root) + .collect(); + for root in dropped { + self.forget(&root); + } + } + + pub(crate) fn len(&self) -> usize { + self.asks.len() + } + + pub(crate) fn in_flight_len(&self) -> usize { + self.in_flight.len() + } + + pub(crate) fn is_block_awaiting_room(&self, root: &H256) -> bool { + self.blocks_awaiting_room + .iter() + .any(|(held, _)| held == root) + } +} + +/// `INVALID` with no `latestValidHash`: since `engine_newPayloadV2` this is +/// how a payload whose contents do not hash to its claimed block hash is +/// answered (`INVALID_BLOCK_HASH` is "supplanted by `INVALID`", +/// execution-apis `shanghai.md`), and also an executed failure whose last valid +/// ancestor the client cannot name. Either way nothing says the payload the +/// bid commits to is bad, only this copy of it. +fn refuses_only_this_envelope(status: &PayloadStatusV1) -> bool { + status.status == PayloadStatusValue::InvalidBlockHash + || (status.status == PayloadStatusValue::Invalid && status.latest_valid_hash.is_none()) +} + +impl BlockChainServer { + /// Whether payloads are imported before their answer: there is an + /// execution client to ask, and a way for the answer to come back. + pub(crate) fn answers_payloads_later(&self) -> bool { + self.engine.is_some() && self.verdict_sink.is_some() + } + + /// Ask about `block`'s payload without waiting. The caller imports the + /// block as optimistic right after. + pub(crate) fn ask_about_block_payload(&mut self, block: &SignedBeaconBlock) { + let Some(request) = beacon_engine::new_payload_request(block) else { + return; + }; + let block_hash = request.execution_payload.block_hash; + let parent_hash = request.execution_payload.parent_hash; + let root = block.message_hash_tree_root(); + let question = PayloadQuestion::Block(block.clone()); + self.new_payloads + .insert(root, block.slot(), block_hash, parent_hash, question); + self.send_new_payloads(); + } + + /// Ask about the payload `envelope` reveals for `block` without waiting. + /// The caller has checked that the two belong together + /// ([`beacon_engine::gloas_new_payload_request`]) and applies the envelope + /// right after. + pub(crate) fn ask_about_envelope_payload( + &mut self, + block: &SignedBeaconBlock, + envelope: &gloas::SignedExecutionPayloadEnvelope, + envelope_root: H256, + ) { + let root = envelope.message.beacon_block_root; + let payload = &envelope.message.payload; + let (block_hash, parent_hash) = (payload.block_hash, payload.parent_hash); + let question = PayloadQuestion::Envelope { + block: block.clone(), + envelope: Box::new(envelope.clone()), + envelope_root, + }; + self.new_payloads + .insert(root, block.slot(), block_hash, parent_hash, question); + self.send_new_payloads(); + } + + /// Send every queued request that may go now, each from a task of its own + /// that reports back with a [`NewPayloadAnswered`]. + fn send_new_payloads(&mut self) { + let (Some(client), Some(sink)) = (self.engine.clone(), self.verdict_sink.clone()) else { + return; + }; + while let Some((root, question)) = self.new_payloads.next_to_send() { + let client = client.clone(); + let sink = sink.clone(); + tokio::spawn(async move { + let answer = match &*question { + PayloadQuestion::Block(block) => { + PayloadAnswer::Block(beacon_engine::ask(&client, block).await) + } + PayloadQuestion::Envelope { + block, + envelope, + envelope_root, + } => PayloadAnswer::Envelope { + envelope_root: *envelope_root, + result: beacon_engine::ask_envelope_status(&client, block, envelope).await, + }, + }; + sink.send(NewPayloadAnswered { root, answer }); + }); + } + self.publish_new_payload_gauges(); + } + + fn publish_new_payload_gauges(&self) { + metrics::set_new_payloads(self.new_payloads.in_flight_len(), self.new_payloads.len()); + } + + /// Apply an answer, then let through whatever it made room for. + pub(crate) async fn on_new_payload_answered(&mut self, answered: NewPayloadAnswered) { + let NewPayloadAnswered { root, answer } = answered; + let tracked = self.new_payloads.finish(&root); + let mut head_moved = false; + match answer { + PayloadAnswer::Block(result) if tracked => { + head_moved = self.apply_block_answer(root, result); + } + PayloadAnswer::Envelope { + envelope_root, + result, + } if tracked => { + if self.apply_envelope_answer(root, envelope_root, result) { + self.forget_payloads_left_behind(); + } + } + PayloadAnswer::Block(_) => metrics::inc_new_payload_answer("block", "stale"), + PayloadAnswer::Envelope { .. } => metrics::inc_new_payload_answer("envelope", "stale"), + } + if head_moved { + self.forget_payloads_left_behind(); + } + + self.send_new_payloads(); + self.release_blocks_awaiting_engine_room().await; + if self.new_payloads.has_room() { + self.release_envelopes_awaiting_engine_room(); + } + // Releases the children of a gloas payload that was just answered, and + // recomputes the head when an envelope was taken back out. + self.settle_envelopes().await; + if head_moved || self.new_payloads.take_deferred_forkchoice_updated() { + self.recompute_beacon_head().await; + } + self.publish_new_payload_gauges(); + } + + /// Apply the answer for a pre-gloas block already in fork choice. `true` + /// when blocks left fork choice. + fn apply_block_answer( + &mut self, + root: H256, + result: Result, EngineError>, + ) -> bool { + let validity = match result { + Ok(validity) => validity, + Err(err) => { + warn!( + block_root = %ShortRoot(&root.0), + %err, + "No verdict from the execution client; the block stays optimistic \ + and is asked about again next slot" + ); + metrics::inc_engine_no_verdict(); + metrics::inc_new_payload_answer("block", "no_answer"); + self.new_payloads.mark_unanswered(&root); + return false; + } + }; + let slot = self.new_payloads.complete(&root); + // The block failed its own import after the question went out, or + // left fork choice since. + if !slot.is_some_and(|slot| self.store.has_live_chain_entry(slot, &root)) { + metrics::inc_new_payload_answer("block", "stale"); + return false; + } + match validity { + Some(PayloadValidity::Validated) | Some(PayloadValidity::NotRequired) | None => { + let head = self.store.beacon_head().map(|(_, head)| head); + let head_was_optimistic = + head.is_some_and(|head| self.store.is_beacon_optimistic(head)); + fork_choice::mark_validated(&mut self.store, root); + metrics::inc_new_payload_answer("block", "valid"); + if let Some(head) = head + && head_was_optimistic + && !self.store.is_beacon_optimistic(head) + { + let wall_clock_slot = self.wall_clock_slot(); + self.events + .publish_beacon_head_validated(&self.store, head, wall_clock_slot); + } + false + } + Some(PayloadValidity::Optimistic) => { + metrics::inc_new_payload_answer("block", "optimistic"); + false + } + Some(PayloadValidity::Invalidated { latest_valid_hash }) => { + metrics::inc_new_payload_answer("block", "invalid"); + self.invalidate_block_payload(root, latest_valid_hash) > 0 + } + } + } + + /// Apply the answer for a gloas payload whose envelope is already applied. + /// `true` when payloads or blocks left fork choice. + fn apply_envelope_answer( + &mut self, + root: H256, + envelope_root: H256, + result: Result, + ) -> bool { + let status = match result { + Ok(status) => status, + Err(err) => { + warn!( + block_root = %ShortRoot(&root.0), + %err, + "No verdict from the execution client; the payload stays not validated \ + and is asked about again next slot" + ); + metrics::inc_engine_no_verdict(); + metrics::inc_new_payload_answer("envelope", "no_answer"); + self.new_payloads.mark_unanswered(&root); + return false; + } + }; + self.new_payloads.complete(&root); + // Taken back out, or condemned with an ancestor, since it was asked. + if !self.store.has_verified_payload(&root) { + metrics::inc_new_payload_answer("envelope", "stale"); + return false; + } + let Some((slot, _parent)) = self.store.block_entry(&root) else { + metrics::inc_new_payload_answer("envelope", "stale"); + return false; + }; + + if refuses_only_this_envelope(&status) { + metrics::inc_new_payload_answer("envelope", "reverted"); + self.revert_envelope(root, envelope_root); + return false; + } + match beacon_engine::verdict(&status) { + PayloadValidity::Validated | PayloadValidity::NotRequired => { + metrics::inc_new_payload_answer("envelope", "valid"); + self.store + .insert_beacon_block_payload_status(root, slot, PayloadStatusEnum::Valid); + crate::beacon_payloads::mark_payloads_validated(&mut self.store, root); + self.release_children_of_answered_payload(root); + false + } + PayloadValidity::Optimistic => { + metrics::inc_new_payload_answer("envelope", "optimistic"); + self.release_children_of_answered_payload(root); + false + } + PayloadValidity::Invalidated { latest_valid_hash } => { + metrics::inc_new_payload_answer("envelope", "invalid"); + warn!( + %slot, + block_root = %ShortRoot(&root.0), + "Execution client found an envelope's payload invalid" + ); + self.envelopes.head_dirty = true; + self.condemn_payload_chain(root, latest_valid_hash) + } + } + } + + /// The block at or above `root` on its chain whose payload is + /// `block_hash`, for a `forkchoiceUpdated` that names an ancestor's payload + /// instead of the head's (see [`NewPayloads::forkchoice_head`]). Pre-gloas + /// only: a block's payload is its own there. + pub(crate) fn ancestor_with_payload(&self, root: H256, block_hash: H256) -> Option { + let mut current = root; + for _ in 0..=self.new_payloads.len() { + if self.store.beacon_el_block_hash(current) == Some(block_hash) { + return Some(current); + } + current = self.store.block_entry(¤t)?.1; + } + None + } + + /// Asks whose block or payload is gone after blocks left fork choice: a + /// question about them would only be answered to be ignored. + fn forget_payloads_left_behind(&mut self) { + let store = &self.store; + self.new_payloads.retain(|root, slot, kind| match kind { + PayloadKind::Block => store.has_live_chain_entry(slot, root), + PayloadKind::Envelope => store.has_verified_payload(root), + }); + } + + /// Hold a pre-gloas block whose payload there is no room to wait for. + /// + /// Written the way a block held for columns is (see + /// `hold_block_for_columns`): no `LiveChain` entry, so it stays invisible + /// to fork choice, and readable back by root once room frees. + pub(crate) fn hold_block_for_engine_room( + &mut self, + block: SignedBeaconBlock, + timings: ImportTimings, + ) { + let slot = block.slot(); + let block_root = block.message_hash_tree_root(); + if self.new_payloads.is_block_awaiting_room(&block_root) { + return; + } + info!( + %slot, + block_root = %ShortRoot(&block_root.0), + awaiting = self.new_payloads.len(), + "Holding block: too many payloads are waiting for the execution client" + ); + self.store + .insert_pending_block(block_root, block) + .expect("DB insert should succeed"); + self.new_payloads + .blocks_awaiting_room + .push_back((block_root, slot)); + self.held_timings.insert(block_root, timings); + } + + /// Import the blocks held for room, oldest first, while there is room. + async fn release_blocks_awaiting_engine_room(&mut self) { + while self.new_payloads.has_room() { + let Some((block_root, slot)) = self.new_payloads.blocks_awaiting_room.pop_front() + else { + return; + }; + let timings = self + .held_timings + .remove(&block_root) + .unwrap_or_else(ImportTimings::starting_now); + let Ok(Some(block)) = self.store.get_signed_block(&block_root) else { + warn!( + %slot, + block_root = %ShortRoot(&block_root.0), + "A block held for engine room vanished from the store" + ); + continue; + }; + // The import path every block takes, which cascades its pending + // children and may hold it again if the room is gone by then. + self.on_block(block, timings).await; + } + } + + /// Once a slot: ask again about every payload the execution client did not + /// answer, and let through what the room allows. + pub(crate) async fn redrive_new_payloads(&mut self) { + if self.new_payloads.len() == 0 && self.new_payloads.blocks_awaiting_room.is_empty() { + return; + } + self.new_payloads.requeue_unanswered(); + self.send_new_payloads(); + self.release_blocks_awaiting_engine_room().await; + } + + /// Stop waiting for payloads at or below the finalized slot, whose verdict + /// no longer changes what fork choice can pick, and drop the blocks held + /// for room there. + pub(crate) fn evict_new_payloads_at_or_below_finality(&mut self) { + if self.new_payloads.len() == 0 && self.new_payloads.blocks_awaiting_room.is_empty() { + return; + } + let finalized_slot = self + .store + .latest_finalized() + .expect("finalized checkpoint exists") + .slot; + let below: Vec = self + .new_payloads + .asks + .iter() + .filter(|(_, ask)| ask.slot <= finalized_slot) + .map(|(root, _)| *root) + .collect(); + for root in below { + self.new_payloads.forget(&root); + } + let mut dropped = Vec::new(); + self.new_payloads + .blocks_awaiting_room + .retain(|(root, slot)| { + let keep = *slot > finalized_slot; + if !keep { + dropped.push(*root); + } + keep + }); + for root in dropped { + self.held_timings.remove(&root); + self.discard_pending_subtree(root); + } + self.publish_new_payload_gauges(); + } +} + +impl Handler for BlockChainServer { + async fn handle(&mut self, answered: NewPayloadAnswered, _ctx: &Context) { + self.on_new_payload_answered(answered).await; + } +} + +#[cfg(test)] +mod tests { + use ethlambda_engine::{EngineClient, JwtSecret}; + use tokio::sync::mpsc; + + use super::*; + use crate::events::ChainEvent; + use crate::tests::{ + A, B, G, GENESIS_TIME, beacon_server, beacon_store, event_store, fulu_block, + insert_event_block, set_head, + }; + + fn hash(byte: u8) -> H256 { + H256::repeat_byte(byte) + } + + /// A distinct root per `n`, for filling the waiting set. + fn numbered(n: u64) -> H256 { + let mut bytes = [0u8; 32]; + bytes[24..].copy_from_slice(&n.to_be_bytes()); + H256(bytes) + } + + fn question(slot: u64) -> PayloadQuestion { + PayloadQuestion::Block(fulu_block(H256::ZERO, slot, 0)) + } + + /// `root` asks about payload `block_hash`, built on `parent_hash`. + fn ask(payloads: &mut NewPayloads, root: u8, block_hash: u8, parent_hash: u8) { + payloads.insert( + hash(root), + 1, + hash(block_hash), + hash(parent_hash), + question(1), + ); + } + + fn sent(payloads: &mut NewPayloads) -> Option { + payloads.next_to_send().map(|(root, _)| root) + } + + #[test] + fn a_payload_waits_while_the_one_it_builds_on_is_outstanding() { + let mut payloads = NewPayloads::default(); + ask(&mut payloads, 1, 0xa1, 0xa0); + ask(&mut payloads, 2, 0xa2, 0xa1); + + assert_eq!(sent(&mut payloads), Some(hash(1))); + assert_eq!(sent(&mut payloads), None, "the child's parent is in flight"); + + assert!(payloads.finish(&hash(1))); + payloads.complete(&hash(1)); + assert_eq!(sent(&mut payloads), Some(hash(2))); + } + + #[test] + fn siblings_go_in_parallel_up_to_the_cap() { + let mut payloads = NewPayloads::default(); + for root in 1..=MAX_NEW_PAYLOADS_IN_FLIGHT as u8 + 2 { + ask(&mut payloads, root, 0xb0 + root, 0xa0); + } + for root in 1..=MAX_NEW_PAYLOADS_IN_FLIGHT as u8 { + assert_eq!(sent(&mut payloads), Some(hash(root))); + } + assert_eq!(sent(&mut payloads), None, "the cap is reached"); + + assert!(payloads.finish(&hash(1))); + payloads.complete(&hash(1)); + assert_eq!( + sent(&mut payloads), + Some(hash(MAX_NEW_PAYLOADS_IN_FLIGHT as u8 + 1)) + ); + } + + #[test] + fn an_unanswered_payload_holds_its_children_until_asked_again() { + let mut payloads = NewPayloads::default(); + ask(&mut payloads, 1, 0xa1, 0xa0); + ask(&mut payloads, 2, 0xa2, 0xa1); + assert_eq!(sent(&mut payloads), Some(hash(1))); + assert!(payloads.finish(&hash(1))); + payloads.mark_unanswered(&hash(1)); + + assert_eq!(sent(&mut payloads), None, "the parent is still unanswered"); + // An unreachable client answers no forkchoiceUpdated either, so the + // unanswered payload does not hold the head back; the queued one does. + assert_eq!(payloads.forkchoice_head(hash(0xa1)), hash(0xa1)); + assert_eq!(payloads.forkchoice_head(hash(0xa2)), hash(0xa1)); + + payloads.requeue_unanswered(); + assert_eq!(sent(&mut payloads), Some(hash(1))); + } + + #[test] + fn forkchoice_updated_names_the_newest_payload_the_client_was_asked_about() { + let mut payloads = NewPayloads::default(); + ask(&mut payloads, 1, 0xa1, 0xa0); + ask(&mut payloads, 2, 0xa2, 0xa1); + ask(&mut payloads, 3, 0xa3, 0xa2); + + // Nothing answered yet: the chain falls back to the payload the first + // one builds on, which the client already has. + assert_eq!(payloads.forkchoice_head(hash(0xa3)), hash(0xa0)); + + assert_eq!(sent(&mut payloads), Some(hash(1))); + assert!(payloads.finish(&hash(1))); + payloads.complete(&hash(1)); + assert_eq!(payloads.forkchoice_head(hash(0xa3)), hash(0xa1)); + + // An unanswered request does not hold the head back. + assert_eq!(sent(&mut payloads), Some(hash(2))); + assert!(payloads.finish(&hash(2))); + payloads.mark_unanswered(&hash(2)); + assert_eq!(payloads.forkchoice_head(hash(0xa3)), hash(0xa2)); + + // A payload nobody is waiting on is named as it is. + assert_eq!(payloads.forkchoice_head(hash(0xff)), hash(0xff)); + } + + #[test] + fn a_forgotten_payload_frees_its_slot_and_its_answer_is_ignored() { + let mut payloads = NewPayloads::default(); + ask(&mut payloads, 1, 0xa1, 0xa0); + ask(&mut payloads, 2, 0xa2, 0xa1); + assert_eq!(sent(&mut payloads), Some(hash(1))); + + payloads.forget(&hash(1)); + assert!(!payloads.finish(&hash(1)), "nothing applies the answer"); + assert_eq!(payloads.in_flight_len(), 0); + assert_eq!( + sent(&mut payloads), + Some(hash(2)), + "the child is free to go" + ); + } + + #[test] + fn the_waiting_set_is_bounded() { + let mut payloads = NewPayloads::default(); + for root in 0..MAX_PAYLOADS_AWAITING_VERDICT { + let root = numbered(root as u64 + 1); + payloads.insert(root, 1, root, H256::ZERO, question(1)); + } + assert!(!payloads.has_room()); + payloads.forget(&numbered(1)); + assert!(payloads.has_room()); + } + + /// `A` (slot 1) and `B` (slot 2, the head) imported optimistically, with + /// `B`'s question in flight. + fn server_waiting_on_b() -> BlockChainServer { + let mut store = event_store(GENESIS_TIME); + insert_event_block(&mut store, A, 1, G, &[(0, G)]); + insert_event_block(&mut store, B, 2, A, &[(0, G), (1, A)]); + set_head(&mut store, B); + store.insert_beacon_optimistic_root(A, 1); + store.insert_beacon_optimistic_root(B, 2); + let mut server = beacon_server(store); + server + .new_payloads + .insert(B, 2, hash(0xb2), hash(0xa1), question(2)); + assert_eq!(sent(&mut server.new_payloads), Some(B)); + server + } + + fn block_answer(validity: Result, EngineError>) -> NewPayloadAnswered { + NewPayloadAnswered { + root: B, + answer: PayloadAnswer::Block(validity), + } + } + + #[tokio::test] + async fn a_valid_answer_validates_the_block_and_its_ancestors() { + let mut server = server_waiting_on_b(); + + server + .on_new_payload_answered(block_answer(Ok(Some(PayloadValidity::Validated)))) + .await; + + assert!(!server.store.is_beacon_optimistic(B)); + assert!( + !server.store.is_beacon_optimistic(A), + "ancestors go with it" + ); + assert_eq!(server.new_payloads.len(), 0); + } + + /// The second `head` for a head whose payload was validated after it was + /// announced. Driven with an explicit wall-clock slot: the handler reads + /// the real clock, against which this fixture's head is long stale. + #[tokio::test] + async fn a_validated_head_is_announced_again_without_the_optimistic_flag() { + let mut server = server_waiting_on_b(); + let mut events = server.events.subscribe(); + fork_choice::mark_validated(&mut server.store, B); + + server + .events + .publish_beacon_head_validated(&server.store, A, 2); + assert!(events.try_recv().is_err(), "not the announced head"); + + server + .events + .publish_beacon_head_validated(&server.store, B, 2); + let heads: Vec<_> = std::iter::from_fn(|| events.try_recv().ok()) + .filter_map(|event| match event { + ChainEvent::BeaconHead(head) => Some(head), + _ => None, + }) + .collect(); + assert_eq!(heads.len(), 1, "{heads:?}"); + assert_eq!(heads[0].block, B); + assert!(!heads[0].execution_optimistic); + } + + #[tokio::test] + async fn a_syncing_answer_leaves_the_block_optimistic() { + let mut server = server_waiting_on_b(); + + server + .on_new_payload_answered(block_answer(Ok(Some(PayloadValidity::Optimistic)))) + .await; + + assert!(server.store.is_beacon_optimistic(B)); + assert_eq!(server.new_payloads.len(), 0, "answered, no longer waiting"); + } + + #[tokio::test] + async fn an_invalid_answer_removes_the_block_from_fork_choice() { + let mut server = server_waiting_on_b(); + + server + .on_new_payload_answered(block_answer(Ok(Some(PayloadValidity::Invalidated { + latest_valid_hash: None, + })))) + .await; + + assert!(!server.store.has_live_chain_entry(2, &B)); + assert!( + server.store.has_live_chain_entry(1, &A), + "only B is condemned" + ); + } + + #[tokio::test] + async fn no_answer_keeps_the_block_and_asks_again_next_slot() { + let mut server = server_waiting_on_b(); + + server + .on_new_payload_answered(block_answer(Err(EngineError::Transport("down".into())))) + .await; + + assert!(server.store.is_beacon_optimistic(B)); + assert!(server.store.has_live_chain_entry(2, &B)); + assert_eq!(server.new_payloads.len(), 1, "still waiting"); + assert_eq!( + sent(&mut server.new_payloads), + None, + "not before the next slot" + ); + + server.new_payloads.requeue_unanswered(); + assert_eq!(sent(&mut server.new_payloads), Some(B)); + } + + #[tokio::test] + async fn an_answer_for_a_forgotten_payload_changes_nothing() { + let mut server = server_waiting_on_b(); + server.new_payloads.forget(&B); + + server + .on_new_payload_answered(block_answer(Ok(Some(PayloadValidity::Invalidated { + latest_valid_hash: None, + })))) + .await; + + assert!(server.store.has_live_chain_entry(2, &B)); + assert_eq!(server.new_payloads.in_flight_len(), 0); + } + + /// With the waiting set full, a block that would be imported optimistically + /// is held instead, and goes back to the import path once room frees. + #[tokio::test] + async fn a_full_waiting_set_holds_the_import_until_room_frees() { + let mut server = beacon_server(beacon_store(GENESIS_TIME, 0)); + // Never reached: the set is full, so nothing is sent. + server.engine = Some( + EngineClient::new("http://127.0.0.1:1".to_string(), JwtSecret::new([0u8; 32])) + .expect("client builds"), + ); + let (tx, _rx) = mpsc::unbounded_channel(); + server.verdict_sink = Some(VerdictSink::Channel(tx)); + // Any parent qualifies as an optimistic candidate. + server.safe_slots_to_import_optimistically = 0; + for root in 0..MAX_PAYLOADS_AWAITING_VERDICT { + let root = numbered(root as u64 + 1); + server + .new_payloads + .insert(root, 1, root, H256::ZERO, question(1)); + } + + let block = fulu_block(H256::ZERO, 1, 0); + let block_root = block.message_hash_tree_root(); + let (_timings, outcome) = server.process_block(block, ImportTimings::default()).await; + + assert!(matches!(outcome, Ok(crate::ImportOutcome::Held))); + assert!(server.new_payloads.is_block_awaiting_room(&block_root)); + assert!( + server + .store + .get_signed_block(&block_root) + .unwrap() + .is_some() + ); + + server.new_payloads.forget(&numbered(1)); + server.release_blocks_awaiting_engine_room().await; + assert!(!server.new_payloads.is_block_awaiting_room(&block_root)); + } +} diff --git a/crates/blockchain/src/events.rs b/crates/blockchain/src/events.rs index 1e82f89e8..f3f1dbdc3 100644 --- a/crates/blockchain/src/events.rs +++ b/crates/blockchain/src/events.rs @@ -593,6 +593,27 @@ impl ChainEvents { } } + /// `head` again for the head already announced, now that the execution + /// client validated its payload. + /// + /// A block is imported before its payload is judged, so the `head` event + /// for a new head goes out with `execution_optimistic: true` and a + /// validator client that reacts to it is refused attestation data until the + /// verdict arrives. This second event, same head and `false`, is its cue + /// to ask again. Only when the view's head is still `head`: a head that + /// has moved on was announced on its own. + pub(crate) fn publish_beacon_head_validated( + &self, + store: &Store, + head: H256, + wall_clock_slot: u64, + ) { + if self.view.head != head || !self.bus.has_subscribers() { + return; + } + publish_beacon_head_change(store, &self.bus, &self.view, head, wall_clock_slot); + } + /// Move the view to `store`'s current values, whether or not anything is /// emitted, and hand back the previous one when there is a subscriber to /// report the difference to. diff --git a/crates/blockchain/src/lib.rs b/crates/blockchain/src/lib.rs index 251048b7a..9c38df2c8 100644 --- a/crates/blockchain/src/lib.rs +++ b/crates/blockchain/src/lib.rs @@ -58,6 +58,7 @@ mod beacon_aggregates; mod beacon_columns; pub mod beacon_engine; mod beacon_envelope; +mod beacon_new_payload; mod beacon_payloads; pub use beacon_payloads::checkpoint_hash; pub mod block_builder; @@ -423,6 +424,8 @@ impl BlockChain { custody_columns, engine, safe_slots_to_import_optimistically, + new_payloads: Default::default(), + verdict_sink: None, last_tick_instant: None, epoch_precompute_in_flight: None, sync_status, @@ -548,6 +551,15 @@ pub struct BlockChainServer { /// specification requires the value to be operator-configurable. safe_slots_to_import_optimistically: u64, + /// Payloads imported before the execution client answered for them, and + /// what waits on that answer. See [`crate::beacon_new_payload`]. + new_payloads: beacon_new_payload::NewPayloads, + + /// Where a `newPayload` request task reports its answer: this actor's own + /// reference, recorded when it starts. `None` on a server with no mailbox + /// (offline replay), which then waits for every answer inline. + verdict_sink: Option, + /// Last tick instant for measuring interval duration. last_tick_instant: Option, @@ -745,14 +757,19 @@ pub enum ImportOutcome { /// The block left the cascade without a post-state under its root, so /// nothing pending on that root may be unblocked. /// - /// Two reasons reach this, and they differ in what happens next: + /// Three reasons reach this, and they differ in what happens next: /// /// - A fulu block whose custody columns have not all arrived. The block is /// persisted and readable back by root, and /// [`BlockChainServer::release_block_if_columns_complete`] eventually /// re-imports it and reaches `collect_pending_children` for real. - /// - No verdict from the execution client, or a `NOT_VALIDATED` verdict on - /// a block that is not an optimistic candidate. Nothing is recorded and + /// - Too many payloads already waiting for the execution client's answer + /// (see [`crate::beacon_new_payload`]). Persisted the same way, and + /// re-imported as answers free room. + /// - On the inline path (a block that is not an optimistic candidate, or a + /// server with no mailbox): no verdict from the execution client, or a + /// `NOT_VALIDATED` verdict on a block that is not an optimistic + /// candidate. Nothing is recorded and /// the block is simply dropped: unlike a column hold, no timer re-drives /// it, which is what the terminal retry policy on the engine ladder /// means. See `EngineClient::call`'s own documentation. The next arrival @@ -1075,7 +1092,9 @@ impl BlockChainServer { self.evict_held_blocks_at_or_below_finality(); self.evict_sidecars_awaiting_parent_at_or_below_finality(); self.evict_envelope_queues_at_or_below_finality(); + self.evict_new_payloads_at_or_below_finality(); self.redrive_held_blocks().await; + self.redrive_new_payloads().await; self.redrive_envelopes_awaiting_columns(); self.redrive_envelopes_awaiting_engine(); self.redrive_missing_envelopes(); @@ -1790,63 +1809,47 @@ impl BlockChainServer { fork_choice::DataAvailability::NotRequired }; - // The engine round trip sits here, between the + // The engine question sits here, between the // data-availability gate above and `fork_choice::on_block` // below. That order is the specification's: // `is_data_available` runs before `state_transition`, so a // block about to be held for its custody columns is never one // this node asks an execution client about. + // + // On the running actor the question goes out without waiting + // and the block imports as optimistic; the answer is applied + // when it arrives (see `beacon_new_payload`). Only an + // optimistic candidate may be imported that way, so a block + // that is not one (around the merge transition, never on a + // network this node follows today) still waits for its answer + // inline, as does a server with no mailbox to hear back on. timings.engine_start = Some(Instant::now()); - let validity = match &self.engine { - None => fork_choice::PayloadValidity::NotRequired, - Some(client) => match beacon_engine::ask(client, &beacon_block).await { - Ok(None) => fork_choice::PayloadValidity::NotRequired, - Ok(Some(validity)) => validity, - // No answer after the whole ladder. - // `optimistic-sync.md`: a consensus engine MUST NOT - // import the block and MUST NOT apply it to the fork - // choice store. Returning `Held` leaves the block out - // of the store without marking it imported, so the - // cascade does not treat it as having produced a - // post-state. - Err(err) => { - warn!( - %slot, - block_root = %ShortRoot(&block_root.0), - %err, - "No verdict from the execution client; not importing" - ); - metrics::inc_engine_no_verdict(); - timings.engine_end = Some(Instant::now()); - return (timings, Ok(ImportOutcome::Held)); - } - }, - }; - timings.engine_end = Some(Instant::now()); - - // An optimistic import is only permitted for a block that - // qualifies. A block that does not, and got a NOT_VALIDATED - // answer, is not imported at all. - if matches!(validity, fork_choice::PayloadValidity::Optimistic) { - let current_slot = self.wall_clock_slot(); - let parent_root = beacon_block.parent_root(); - if !fork_choice::is_optimistic_candidate_block( + let parent_root = beacon_block.parent_root(); + let answered_later = self.answers_payloads_later() + && beacon_engine::new_payload_request(&beacon_block).is_some() + && fork_choice::is_optimistic_candidate_block( &self.store, - current_slot, + self.wall_clock_slot(), slot, parent_root, self.safe_slots_to_import_optimistically, - ) { - warn!( - %slot, - block_root = %ShortRoot(&block_root.0), - "Not importing: the execution client has not validated this \ - block and it is not an optimistic candidate" - ); - metrics::inc_engine_not_optimistic_candidate(); - return (timings, Ok(ImportOutcome::Held)); - } + ); + if answered_later && !self.new_payloads.has_room() { + timings.engine_end = Some(Instant::now()); + self.hold_block_for_engine_room(beacon_block, timings); + return (timings, Ok(ImportOutcome::Held)); } + let validity = if answered_later { + self.ask_about_block_payload(&beacon_block); + Some(fork_choice::PayloadValidity::Optimistic) + } else { + self.ask_payload_inline(&beacon_block, slot, block_root) + .await + }; + timings.engine_end = Some(Instant::now()); + let Some(validity) = validity else { + return (timings, Ok(ImportOutcome::Held)); + }; // Drop any hand-off span left by an insert outside this call, // so the one taken below is this block's alone. @@ -1867,6 +1870,8 @@ impl BlockChainServer { None => (None, None), }; if let Err(err) = imported { + // Its question may already be out; the answer is ignored. + self.new_payloads.forget(&block_root); return (timings, Err(err.into())); } @@ -1961,6 +1966,7 @@ impl BlockChainServer { self.evict_held_blocks_at_or_below_finality(); self.evict_sidecars_awaiting_parent_at_or_below_finality(); self.evict_envelope_queues_at_or_below_finality(); + self.evict_new_payloads_at_or_below_finality(); self.refresh_chain_metrics(); @@ -1979,6 +1985,65 @@ impl BlockChainServer { (timings, Ok(outcome)) } + /// Ask the execution client about `block`'s payload and wait for the + /// answer: the path for a block that may not be imported optimistically, + /// and for a server with no mailbox to hear an answer back on. + /// + /// `None` when the block must not be imported: no answer after the whole + /// retry ladder (`optimistic-sync.md`: a consensus engine MUST NOT import + /// the block and MUST NOT apply it to the fork choice store), or a + /// `NOT_VALIDATED` answer for a block that is not an optimistic candidate. + /// The caller then reports [`ImportOutcome::Held`], which leaves the block + /// out of the store without marking it imported, so the cascade does not + /// treat it as having produced a post-state. + async fn ask_payload_inline( + &mut self, + block: &SignedBeaconBlock, + slot: u64, + block_root: H256, + ) -> Option { + let validity = match &self.engine { + None => fork_choice::PayloadValidity::NotRequired, + Some(client) => match beacon_engine::ask(client, block).await { + Ok(None) => fork_choice::PayloadValidity::NotRequired, + Ok(Some(validity)) => validity, + Err(err) => { + warn!( + %slot, + block_root = %ShortRoot(&block_root.0), + %err, + "No verdict from the execution client; not importing" + ); + metrics::inc_engine_no_verdict(); + return None; + } + }, + }; + + // An optimistic import is only permitted for a block that qualifies. + // A block that does not, and got a NOT_VALIDATED answer, is not + // imported at all. + if matches!(validity, fork_choice::PayloadValidity::Optimistic) + && !fork_choice::is_optimistic_candidate_block( + &self.store, + self.wall_clock_slot(), + slot, + block.parent_root(), + self.safe_slots_to_import_optimistically, + ) + { + warn!( + %slot, + block_root = %ShortRoot(&block_root.0), + "Not importing: the execution client has not validated this \ + block and it is not an optimistic candidate" + ); + metrics::inc_engine_not_optimistic_candidate(); + return None; + } + Some(validity) + } + /// Build an unspawned server for offline replay. /// /// No mailbox, no tick loop, no p2p: the caller drives every import @@ -2016,6 +2081,8 @@ impl BlockChainServer { custody_columns: Vec::new(), engine, safe_slots_to_import_optimistically, + new_payloads: Default::default(), + verdict_sink: None, last_tick_instant: None, epoch_precompute_in_flight: None, sync_status: SyncStatusTracker::new(false), @@ -2558,9 +2625,30 @@ impl BlockChainServer { // for a pre-gloas head, which keeps its own cached payload hash; // `None` there means a follower still on its checkpoint anchor. let plan = beacon_payloads::forkchoice_plan(&self.store, head_root)?; - let state = plan.state; + let mut state = plan.state; let gloas = plan.gloas; + // A head whose own `newPayload` is still queued or in flight is one the + // execution client may not have yet: named as the head, it is answered + // `SYNCING` and may send the client syncing from its peers. The call + // names the nearest payload on its chain the client has been asked + // about instead, and the answer that releases the head sends it again + // (`beacon_new_payload`). The verdict it brings back is then about + // that ancestor: a pre-gloas one is found by its payload hash, a gloas + // one is resolved from the hash by `apply_gloas_forkchoice_verdict`. + let mut verdict_root = head_root; + let named = self.new_payloads.forkchoice_head(state.head_block_hash); + if named != state.head_block_hash { + self.new_payloads.defer_forkchoice_updated(); + if named.is_zero() { + return None; + } + if !gloas { + verdict_root = self.ancestor_with_payload(head_root, named)?; + } + state.head_block_hash = named; + } + // `head_root` stays valid across the await: this is a single-threaded // actor, so no other message is handled until this one returns. An // `Invalidated` verdict leaves `KEY_HEAD` and `Table::BlockRoots` @@ -2580,7 +2668,7 @@ impl BlockChainServer { }; match response { Ok(status) => { - self.apply_forkchoice_verdict(head_root, state.head_block_hash, gloas, &status) + self.apply_forkchoice_verdict(verdict_root, state.head_block_hash, gloas, &status) } Err(err) => warn!(%err, "forkchoiceUpdated failed"), } @@ -2614,50 +2702,63 @@ impl BlockChainServer { fork_choice::mark_validated(&mut self.store, head_root); } fork_choice::PayloadValidity::Invalidated { latest_valid_hash } => { - let index = self.store.block_index(); - let parent_root = index - .get(&head_root) - .map(|(_slot, parent)| *parent) - .unwrap_or(H256::ZERO); - let condemned = fork_choice::resolve_invalid_block( - &self.store, - &index, - head_root, - parent_root, - latest_valid_hash, - ); - // Unlike the `newPayload` path, `head_root` *is* in the index - // here: this verdict is about a block already imported, which is - // why an invalidation reached through `forkchoiceUpdated` can - // remove the head itself rather than only its descendants. - let removed = fork_choice::invalidate_subtree(&mut self.store, condemned); - warn!( - condemned = %ShortRoot(&condemned.0), - removed, - "Execution client invalidated the head's branch" - ); - - // The rows are gone from fork choice, but `KEY_HEAD` and - // `Table::BlockRoots` still name them, and the p2p req/resp - // handlers read both: `Status` would advertise the - // invalidated root, and `BlocksByRange`/`BlocksByRoot` would - // serve the invalidated block to peers, until the next tick or - // cascade recomputed the head. Do it here instead, so no peer - // is handed a block this node has just refused. - // - // Head only, deliberately not `recompute_beacon_head`: this - // runs inside `forkchoiceUpdated`'s own response handling, and - // announcing the new head from here would re-enter the call - // being answered. - if removed > 0 { - self.update_head_from_fork_choice(); - } + self.invalidate_block_payload(head_root, latest_valid_hash); } fork_choice::PayloadValidity::Optimistic | fork_choice::PayloadValidity::NotRequired => {} } } + /// Apply an `INVALID` verdict about the payload of `root`, a pre-gloas + /// block already imported: cut the branch the verdict condemns out of fork + /// choice and move the head off it. Returns how many blocks left. + /// + /// Two callers, both about a block in the index: a `forkchoiceUpdated` + /// answer about the head, and an asynchronous `newPayload` answer (see + /// `beacon_new_payload`). Either can remove `root` itself, not only its + /// descendants. + pub(crate) fn invalidate_block_payload( + &mut self, + root: H256, + latest_valid_hash: Option, + ) -> usize { + let index = self.store.block_index(); + let parent_root = index + .get(&root) + .map(|(_slot, parent)| *parent) + .unwrap_or(H256::ZERO); + let condemned = fork_choice::resolve_invalid_block( + &self.store, + &index, + root, + parent_root, + latest_valid_hash, + ); + let removed = fork_choice::invalidate_subtree(&mut self.store, condemned); + warn!( + condemned = %ShortRoot(&condemned.0), + removed, + "Execution client invalidated a block's branch" + ); + + // The rows are gone from fork choice, but `KEY_HEAD` and + // `Table::BlockRoots` still name them, and the p2p req/resp handlers + // read both: `Status` would advertise the invalidated root, and + // `BlocksByRange`/`BlocksByRoot` would serve the invalidated block to + // peers, until the next tick or cascade recomputed the head. Do it here + // instead, so no peer is handed a block this node has just refused. + // + // Head only, deliberately not `recompute_beacon_head`: the + // `forkchoiceUpdated` caller runs inside that call's own response + // handling, and announcing the new head from here would re-enter the + // call being answered. The `newPayload` caller recomputes in full + // afterwards. + if removed > 0 { + self.update_head_from_fork_choice(); + } + removed + } + /// Try to process a single block. If its parent state is missing, store it /// as pending. On success, collect any unblocked children into `queue` for /// the caller to process next (iteratively, avoiding deep recursion). @@ -3792,6 +3893,15 @@ impl BlockChainServer { ); } + /// Actor lifecycle hook: record this actor's own reference, which is where + /// a `newPayload` request task reports back (see + /// [`crate::beacon_new_payload`]). Until it runs, and on a server that is + /// never started, payloads are asked about inline. + #[started] + async fn on_started(&mut self, ctx: &Context) { + self.verdict_sink = Some(beacon_new_payload::VerdictSink::Actor(ctx.actor_ref())); + } + /// Actor lifecycle hook: wait for any in-flight aggregation worker to exit /// before the actor is fully stopped. We cancel the session's token and /// wait up to PRIOR_WORKER_JOIN_TIMEOUT for the worker's current @@ -4488,6 +4598,8 @@ mod tests { custody_columns: Vec::new(), engine: None, safe_slots_to_import_optimistically: constants::SAFE_SLOTS_TO_IMPORT_OPTIMISTICALLY, + new_payloads: Default::default(), + verdict_sink: None, last_tick_instant: None, epoch_precompute_in_flight: None, sync_status: SyncStatusTracker::new(false), @@ -5833,10 +5945,10 @@ mod tests { use ethlambda_storage::ForkCheckpoints; use tokio::sync::broadcast; - const G: H256 = H256::repeat_byte(0x10); - const A: H256 = H256::repeat_byte(0x11); - const B: H256 = H256::repeat_byte(0x12); - const C: H256 = H256::repeat_byte(0x13); + pub(crate) const G: H256 = H256::repeat_byte(0x10); + pub(crate) const A: H256 = H256::repeat_byte(0x11); + pub(crate) const B: H256 = H256::repeat_byte(0x12); + pub(crate) const C: H256 = H256::repeat_byte(0x13); /// The post-state root a test block names, derived so each block's is /// distinct from every root in play. @@ -5847,7 +5959,7 @@ mod tests { /// A beacon store anchored at `G`, a real block at slot zero, so the /// head can move off it: the canonical-index diff walks from the old head /// to the new one and needs a block at both ends. - fn event_store(genesis_time: u64) -> Store { + pub(crate) fn event_store(genesis_time: u64) -> Store { let anchor = Checkpoint { root: G, slot: 0 }; let mut store = Store::init_beacon( Arc::new(InMemoryBackend::default()), @@ -5864,7 +5976,7 @@ mod tests { /// Insert `root` at `slot` under `parent`, with a post-state whose /// `block_roots` names, for every slot below `slot`, the latest of /// `ancestors` at or before it, as a state transition would have left it. - fn insert_event_block( + pub(crate) fn insert_event_block( store: &mut Store, root: H256, slot: u64, @@ -5895,7 +6007,7 @@ mod tests { store.insert_state(root, state).expect("insert"); } - fn set_head(store: &mut Store, root: H256) { + pub(crate) fn set_head(store: &mut Store, root: H256) { store .update_checkpoints(ForkCheckpoints::head_only(root)) .expect("update head"); diff --git a/crates/blockchain/src/metrics.rs b/crates/blockchain/src/metrics.rs index 34c67398f..82ba399c3 100644 --- a/crates/blockchain/src/metrics.rs +++ b/crates/blockchain/src/metrics.rs @@ -1489,12 +1489,62 @@ pub fn set_sidecars_awaiting_parent(count: u64) { LEAN_SIDECARS_AWAITING_PARENT.set(count as i64); } +/// `engine_newPayload` requests in flight right now, and payloads (blocks +/// before gloas, envelopes from it) imported and still waiting for an +/// answer, in flight or not. +/// +/// The second is bounded by `MAX_PAYLOADS_AWAITING_VERDICT`; sitting at that +/// bound means imports are being held for room, which is expected during +/// catch-up and nowhere else. +pub fn set_new_payloads(in_flight: usize, awaiting_verdict: usize) { + static LEAN_ENGINE_NEW_PAYLOAD_IN_FLIGHT: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + register_int_gauge!( + "lean_engine_new_payload_in_flight", + "engine_newPayload requests sent and not yet answered" + ) + .unwrap() + }); + static LEAN_ENGINE_PAYLOADS_AWAITING_VERDICT: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + register_int_gauge!( + "lean_engine_payloads_awaiting_verdict", + "Imported payloads the execution client has not answered for yet" + ) + .unwrap() + }); + LEAN_ENGINE_NEW_PAYLOAD_IN_FLIGHT.set(in_flight as i64); + LEAN_ENGINE_PAYLOADS_AWAITING_VERDICT.set(awaiting_verdict as i64); +} + +/// One answer to an asynchronous `engine_newPayload`, by what was asked about +/// (`block` before gloas, `envelope` from it) and how it was answered +/// (`valid`, `optimistic`, `invalid`, `reverted`, `no_answer`, `stale`). +pub fn inc_new_payload_answer(kind: &str, outcome: &str) { + static LEAN_ENGINE_NEW_PAYLOAD_ANSWERS_TOTAL: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + register_int_counter_vec!( + "lean_engine_new_payload_answers_total", + "Answers to engine_newPayload requests sent without waiting, by kind and outcome", + &["kind", "outcome"] + ) + .unwrap() + }); + LEAN_ENGINE_NEW_PAYLOAD_ANSWERS_TOTAL + .with_label_values(&[kind, outcome]) + .inc(); +} + /// Blocks not imported because no execution client verdict was obtained. /// /// A non-zero rate here means the follower is dropping blocks it cannot ask /// about, and its head will park behind the first one. Nothing re-drives a /// given-up call, so this is the metric that says the follower has stopped /// following rather than merely slowed down. +/// +/// With answers taken asynchronously (the running actor's path) a payload +/// already imported is kept and asked about again each slot instead, and this +/// counts those failed asks. pub fn inc_engine_no_verdict() { static LEAN_ENGINE_NO_VERDICT_TOTAL: std::sync::LazyLock = std::sync::LazyLock::new(|| { diff --git a/crates/storage/src/store.rs b/crates/storage/src/store.rs index 14ac9116f..b3c63bc23 100644 --- a/crates/storage/src/store.rs +++ b/crates/storage/src/store.rs @@ -2425,6 +2425,18 @@ impl Store { batch.commit().expect("commit"); } + /// Whether `root`, a block at `slot`, has a live-chain row: whether it is + /// in fork choice. One point read, where [`block_index`](Self::block_index) + /// scans the whole table. A block removed by + /// [`delete_live_chain_entries`](Self::delete_live_chain_entries) answers + /// `false` while its block and state rows remain. + pub fn has_live_chain_entry(&self, slot: u64, root: &H256) -> bool { + let view = self.backend.begin_read().expect("read view"); + view.get(Table::LiveChain, &encode_slot_root_key(slot, root)) + .expect("get") + .is_some() + } + /// Deletes the named live-chain index rows. /// /// Unlike [`prune_live_chain`](Self::prune_live_chain), which drops a whole @@ -4457,6 +4469,18 @@ impl Store { .insert(root, (slot, status)); } + /// Forgets `root`'s payload verdict, for an envelope taken back out of the + /// store before its payload was judged (see + /// [`Self::remove_verified_payload`]): the payload is unknown again, not + /// invalid, so a later envelope for the same block can still be applied. + pub fn remove_beacon_block_payload_status(&mut self, root: H256) { + self.beacon + .lock() + .unwrap() + .block_payload_statuses + .remove(&root); + } + /// Drops payload verdicts strictly below `finalized_slot`, which no /// attestation rule can ask about any more. pub fn prune_beacon_block_payload_statuses(&mut self, finalized_slot: u64) { @@ -4575,6 +4599,13 @@ impl Store { .or_insert((slot, seen_ms)); } + /// Forgets when `root`'s envelope arrived, for an envelope taken back out + /// of the store: a refused envelope says nothing about whether the + /// payload was revealed on time. + pub fn remove_beacon_envelope_seen(&mut self, root: &H256) { + self.beacon.lock().unwrap().envelope_seen_ms.remove(root); + } + /// When `root`'s envelope reached this node, in milliseconds past genesis, /// if it has. pub fn beacon_envelope_seen_ms(&self, root: &H256) -> Option { diff --git a/docs/beacon_engine.md b/docs/beacon_engine.md index 1aa13b184..6758d97a6 100644 --- a/docs/beacon_engine.md +++ b/docs/beacon_engine.md @@ -72,6 +72,37 @@ This diagram is the pre-gloas path. A gloas block carries no payload, only a builder's bid, so nothing is asked about it at import; the question moves to the envelope that reveals the payload (see [Gloas](#gloas) below). +## Asking without waiting + +On the running actor the `engine_newPayloadV4` arrow above does not wait. The +question goes out first, from a tokio task of its own, and the block is imported +right after as optimistic (it joins `optimistic_roots`, so every Beacon API +response about it says `execution_optimistic` and attestation data for it is a +`503`). The task reports back with a `NewPayloadAnswered` message, and the +answer is applied then, the way the table below says, to a block already in fork +choice. A gloas envelope goes the same way: it is applied at once with its +payload `NOT_VALIDATED`. `crates/blockchain/src/beacon_new_payload.rs` holds +the queue, the task and the answer handler. + +``` +block ─► DA gate ─► newPayload queued ─► on_block (optimistic) ─► head may move + │ + └─► task ─► engine_newPayloadV4 ─► NewPayloadAnswered ─► verdict applied +``` + +| Rule | Why | +|---|---| +| A payload is sent only once no request for the payload it builds on is outstanding, and at most `MAX_NEW_PAYLOADS_IN_FLIGHT` at once | The Engine API orders only `forkchoiceUpdated`, and the client's connection pool can deliver two requests in either order; a payload whose parent the execution client has not seen is answered `SYNCING` and may start a sync from its peers | +| `forkchoiceUpdated` names the newest payload on the head's chain whose request is not queued or in flight, and is sent again when the head's own answer arrives | Same reason: the execution client may not have the head's payload yet. Naming an ancestor instead of skipping the call keeps the client fed a head during catch-up, when the newest block's request always waits behind its parent's | +| At most `MAX_PAYLOADS_AWAITING_VERDICT` imported payloads wait for an answer; past that a block is held (written as a pending block) and an envelope waits in `awaiting_engine`, until answers free room | Bounds how far the consensus side can run ahead of what has been validated; only reached during catch-up | +| A block built on a gloas FULL parent waits for the parent payload's answer, not only for its envelope | An envelope refused for its contents is taken back out, and nothing may be built on it by then. Lighthouse and Prysm hold the same blocks by applying an envelope only after its answer | +| No answer after the retry ladder: the block or payload stays, optimistic, and is asked again once a slot | It is already in fork choice; see [spec_deviations.md](./spec_deviations.md#payloads-are-applied-before-the-execution-client-answers) | +| A `VALID` answer for the current head sends `head` again with `execution_optimistic: false` | A validator client that reacts to `head` was refused attestation data for the optimistic one | + +Two paths still wait for the answer inline: a block that is not an optimistic +candidate (around the merge transition; see below), and a server with no mailbox +to hear back on (offline replay). + The order is the specification's own: `is_data_available` runs before `state_transition`, so a block about to be held for its custody columns is never one this node asks an execution client about. @@ -97,9 +128,13 @@ specification. | Wire status | Alias | What happens | |---|---|---| -| `VALID` | | Imported. The block and every optimistic ancestor leave `optimistic_roots` | -| `SYNCING`, `ACCEPTED` | `NOT_VALIDATED` | Imported *if* `is_optimistic_candidate_block` allows it, and joins `optimistic_roots` | -| `INVALID`, `INVALID_BLOCK_HASH` | `INVALIDATED` | Not imported. `latestValidHash` decides how much of the branch dies with it | +| `VALID` | | The block and every optimistic ancestor leave `optimistic_roots` | +| `SYNCING`, `ACCEPTED` | `NOT_VALIDATED` | The block stays in `optimistic_roots` (it was imported optimistic, which only an `is_optimistic_candidate_block` may be) | +| `INVALID`, `INVALID_BLOCK_HASH` | `INVALIDATED` | The block leaves fork choice. `latestValidHash` decides how much of the branch dies with it | + +On the inline path the same answers decide whether the block is imported at +all: `NOT_VALIDATED` imports it only if it is an optimistic candidate, and +`INVALIDATED` does not import it. A block in `optimistic_roots` carries normal fork-choice weight and can be the head. That is the point of optimistic sync, and the `sync/optimistic` fixture @@ -217,21 +252,26 @@ honest block hash and a different body cannot earn an `INVALID` that lands on th honest payload, and the engine never spends time on an envelope this node would refuse anyway. +The envelope is applied as soon as those checks pass, its payload recorded +`NOT_VALIDATED`, and the answer is applied when it arrives (see +[Asking without waiting](#asking-without-waiting)): + | Answer | What happens to the envelope | |---|---| -| `VALID` | Applied; the payload is recorded `VALID` | -| `SYNCING`, `ACCEPTED` | Applied and recorded `SYNCING` (not yet validated); a later `forkchoiceUpdated` `VALID` promotes it | -| `INVALID` with a `latestValidHash` | The payload is invalid: not applied, so the block's FULL node never exists; the block and its EMPTY branch stay | -| `INVALID` with a null `latestValidHash`, or `INVALID_BLOCK_HASH` | The envelope's contents do not hash to the block hash it claims: only that envelope is refused, since the builder's real one may still arrive | -| no answer after the retry ladder | Kept (it is consensus-valid) and asked again once a slot | - -The null-`latestValidHash` row exists because ethrex has no `INVALID_BLOCK_HASH` -status: it answers a mismatch with a null hash and an executed-and-failed payload -with its last valid ancestor. `latestValidHash` and `VALID` follow the execution +| `VALID` | The payload is recorded `VALID`; blocks waiting on it are released | +| `SYNCING`, `ACCEPTED` | Stays `NOT_VALIDATED`; blocks waiting on it are released; a later `forkchoiceUpdated` `VALID` promotes it | +| `INVALID` with a `latestValidHash` | The payload is invalid: its FULL node is removed and the blocks waiting on it are dropped; the block and its EMPTY branch stay | +| `INVALID` with a null `latestValidHash`, or `INVALID_BLOCK_HASH` | The envelope's contents do not hash to the block hash it claims: the envelope is taken back out (no payload, not an invalid one), only that envelope is refused, and the blocks waiting on it keep waiting, since the builder's real one may still arrive | +| no answer after the retry ladder | Stays applied and `NOT_VALIDATED`, blocks waiting on it keep waiting, and it is asked again once a slot | + +The null-`latestValidHash` row exists because since `engine_newPayloadV2` a +block hash mismatch is answered `INVALID` with a null hash (`INVALID_BLOCK_HASH` +is "supplanted by `INVALID`", execution-apis `shanghai.md`), while an +executed-and-failed payload normally names its last valid ancestor. `latestValidHash` and `VALID` follow the execution chain (each block's `bid.parent_block_hash`), not every verified ancestor, since a block built past an ancestor's EMPTY node skips that ancestor's payload. The details, and why each departs from the specification, are in -[spec_deviations.md](./spec_deviations.md#the-execution-client-judges-a-gloas-payload-before-the-envelope-is-applied). +[spec_deviations.md](./spec_deviations.md#payloads-are-applied-before-the-execution-client-answers). With the engine down, the per-slot retry asks about the oldest held envelope only and releases the rest once that ask is answered, so an unreachable client @@ -314,7 +354,7 @@ cosmetic mismatch into an outage. ## Limitations -### The retry ladder gives up, and nothing re-drives it +### The retry ladder gives up Each call is attempted `ENGINE_MAX_ATTEMPTS` times, with a per-attempt timeout of `ENGINE_TIMEOUT` and a backoff starting at `ENGINE_INITIAL_BACKOFF` and doubling. @@ -322,17 +362,16 @@ An RPC *error* is not retried: that is the execution client answering, with a refusal but an answer, so asking again would get the same refusal and burn the ladder for nothing. -After the last attempt the block is **dropped**. `optimistic-sync.md` requires -exactly that much ("a consensus engine MUST NOT import the block and MUST NOT -apply it to the fork choice store"), but nothing here re-drives the call -afterwards, so a persistently unreachable execution client parks the follower's -head behind the first block it could not ask about, permanently, until a -restart. - -This was accepted knowingly for this change. The intended fix is a tick-driven -re-drive of the set of blocks that got no verdict. `lean_engine_no_verdict_total` is the -metric that says it is happening: a non-zero rate means the follower has stopped -following rather than merely slowed down. +On the running actor a block or payload that got no answer is already in fork +choice; it stays, optimistic, and is asked again once a slot (see +[Asking without waiting](#asking-without-waiting)). + +On the inline path (a block that is not an optimistic candidate, or offline +replay) the block is **dropped** after the last attempt. `optimistic-sync.md` +requires exactly that much ("a consensus engine MUST NOT import the block and +MUST NOT apply it to the fork choice store"), and nothing re-drives the call +afterwards. `lean_engine_no_verdict_total` counts both: a non-zero rate means +the execution client is not answering. ### `optimistic_roots` does not survive a restart diff --git a/docs/metrics.md b/docs/metrics.md index 9ac92fd4f..e82e6edf2 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -559,6 +559,26 @@ block raced it, see `lean_beacon_epoch_precompute_seconds` against the slot duration), the node was syncing, or the entry was evicted from the state cache. A `timer` start with no matching `head` start means the last slot was skipped. +### Beacon Execution Payload Verdicts + +A beacon block (before gloas) or envelope (from it) is imported before the +execution client answers `engine_newPayload`, and the answer is applied when it +arrives. See "Asking without waiting" in [`beacon_engine.md`](beacon_engine.md). +This is ethlambda-specific, not part of the leanMetrics spec. + +| Name | Type | Usage | Sample collection event | Labels | Buckets | +|------|------|-------|-------------------------|--------|---------| +| `lean_engine_new_payload_in_flight` | Gauge | `engine_newPayload` requests sent and not yet answered | On every send, answer and per-slot redrive | | | +| `lean_engine_payloads_awaiting_verdict` | Gauge | Imported payloads with no answer yet, in flight or queued | On every send, answer and per-slot redrive | | | +| `lean_engine_new_payload_answers_total` | Counter | Answers by what was asked about and how it was answered | When an answer is applied | kind=block,envelope; outcome=valid,optimistic,invalid,reverted,no_answer,stale | | +| `lean_engine_no_verdict_total` | Counter | Requests that got no answer after the whole retry ladder (the payload stays, and is asked again next slot) | When the ladder is exhausted | | | + +**Read the waiting gauge against `MAX_PAYLOADS_AWAITING_VERDICT`.** At the cap, +imports are held for room: expected during catch-up, a sign of a slow or stuck +execution client otherwise. `outcome=stale` is an answer for a payload that left +fork choice (or was finalized) while its request was out, and `reverted` an +envelope taken back out because its contents do not hash to its block hash. + ### Beacon Pubkey Cache Every BLS signature check `ethlambda beacon` runs (block import, fork choice, diff --git a/docs/spec_deviations.md b/docs/spec_deviations.md index 49ee15c03..90387b5fb 100644 --- a/docs/spec_deviations.md +++ b/docs/spec_deviations.md @@ -571,50 +571,92 @@ newer than the release the fixtures are pinned to. before it checks the head. Remove this entry when the pin reaches `v1.7.0-beta.4`. -## The execution client judges a gloas payload before the envelope is applied - -`verify_execution_payload_envelope` asks the execution engine last, as part of -one boolean function. The follower runs the pure consensus checks first -(`check_execution_payload_envelope`: signature, bid, state, data availability), -then asks the engine with `engine_newPayloadV5` only about an envelope that -passed them, and records it with `accept_execution_payload_envelope`. The -engine's part is answered "valid" inside the checks because its answer has -three outcomes and the specification's `ExecutionEngine` has two. - -- **`VALID`:** applied; the payload is recorded `VALID`. -- **`SYNCING` / `ACCEPTED`:** applied, since the consensus checks passed and the - payload is `NOT_VALIDATED` (`optimistic-sync.md`); recorded `SYNCING`, which - the attestation gossip rules already read as not yet validated. A later - `forkchoiceUpdated` `VALID` for the head, or for a descendant payload, moves - it and every optimistic payload on that chain to `VALID`. Unlike a pre-gloas - block there is no `is_optimistic_candidate_block` gate: a payload is applied - to a block the consensus layer already imported, so there is no - merge-transition poisoning to guard against. -- **`INVALID`:** not applied, so the FULL node of the block never exists; the - payload is recorded `INVALID`, children held for it are dropped and later - blocks that name it as their FULL parent are dropped instead of held. The - block itself and its EMPTY branch stay. `latestValidHash` can reach further - back, in which case the first payload after it is condemned the same way +## Payloads are applied before the execution client answers + +The specification asks the execution engine inside the state transition +(`process_execution_payload` before gloas, `verify_execution_payload_envelope` +from it), so a block or envelope enters the store only with an answer in hand, +and `optimistic-sync.md` relaxes that only for an answer of `NOT_VALIDATED`. The +running follower does not wait for the answer at all +(`crates/blockchain/src/beacon_new_payload.rs`): + +- A pre-gloas block that is an optimistic candidate is imported as optimistic + (`optimistic_roots`) as soon as its consensus checks pass, with its + `engine_newPayloadV4` sent first from a task of its own. A gloas envelope is + applied as soon as the consensus checks below pass, its payload recorded + `NOT_VALIDATED`. Everything that reads optimism reads both as optimistic, so + the Beacon API refuses attestation and payload attestation data for them with + a `503` until the answer arrives. +- The answer is applied when it arrives, to what is already in fork choice: + `VALID` validates (and sends `head` again for a head that was optimistic), + `SYNCING`/`ACCEPTED` changes nothing, `INVALID` removes the block's branch or + condemns the gloas payload, exactly as a `forkchoiceUpdated` verdict about an + imported block already did. +- **No answer after the retry ladder** is where this departs from the letter of + `optimistic-sync.md`, which says such a block MUST NOT be applied to fork + choice. It already is, so it stays, optimistic, and is asked again once a + slot. Taking it back out would need an un-import that re-holds its + descendants, for a state the next answer resolves anyway. +- A payload is sent only once the one it builds on has been answered, at most + `MAX_NEW_PAYLOADS_IN_FLIGHT` at once, and `forkchoiceUpdated` names the newest + payload on the head's chain whose request is not queued or in flight: the + Engine API orders only `forkchoiceUpdated`, and a payload the execution client + receives before its parent is answered `SYNCING` and may start a sync from + peers. +- At most `MAX_PAYLOADS_AWAITING_VERDICT` payloads wait for an answer; past that + the import waits (a pre-gloas block is held, an envelope waits in + `awaiting_engine`). Only catch-up reaches it. +- A block that is not an optimistic candidate (around the merge transition) + still waits for its answer inline, as does a server with no mailbox + (offline replay): those are the specification's order. + +For a gloas payload the follower also orders the checks differently: +`verify_execution_payload_envelope` asks the engine last, as part of one boolean +function, while the follower runs the pure consensus checks first +(`check_execution_payload_envelope`: signature, bid, state, data availability) +and asks with `engine_newPayloadV5` only about an envelope that passed them, +recording it with `accept_execution_payload_envelope`. The engine's part is +answered "valid" inside the checks because its answer has three outcomes and +the specification's `ExecutionEngine` has two. + +- **`VALID`:** the payload is recorded `VALID`. +- **`SYNCING` / `ACCEPTED`:** recorded `SYNCING`, which the attestation gossip + rules already read as not yet validated. A later `forkchoiceUpdated` `VALID` + for the head, or for a descendant payload, moves it and every optimistic + payload on that chain to `VALID`. Unlike a pre-gloas block there is no + `is_optimistic_candidate_block` gate: a payload is applied to a block the + consensus layer already imported, so there is no merge-transition poisoning + to guard against. +- **`INVALID`:** the FULL node of the block is removed and the payload is + recorded `INVALID`; blocks waiting on it are dropped and later blocks that + name it as their FULL parent are dropped instead of held. The block itself and + its EMPTY branch stay. `latestValidHash` can reach further back, in which case + the first payload after it is condemned the same way (`beacon_payloads::resolve_invalid_payload`, which skips blocks that carry no payload, unlike the pre-gloas `resolve_invalid_block`). A `forkchoiceUpdated` - `INVALID` for a payload already applied removes it again (its FULL node, its - stored envelope, and every child built on it). + `INVALID` for a payload already applied removes it the same way. - **A hash mismatch (`INVALID_BLOCK_HASH`, or `INVALID` with a null `latestValidHash`):** the envelope's contents do not hash to the block hash it - claims, so only that envelope (by its own root) is refused; the root's - payload is not condemned, since the builder's real envelope may still arrive. - ethrex has no `INVALID_BLOCK_HASH` status: it answers a block hash mismatch - with `INVALID` and a null `latestValidHash`, and an executed-and-failed - payload with `INVALID` and its last valid ancestor. So a null one is read as - a mismatch and a non-null one as the payload being invalid. A payload that - failed with no valid ancestor to name is read as a mismatch too, which only - costs a refused envelope. -- **No answer after the retry ladder:** the payload is not applied, as for a - block, but unlike a block the envelope is consensus-valid and already held, - so it is kept and the per-slot redrive asks again, about the oldest held - envelope only and the rest once that ask is answered, so a down client costs - one retry ladder per slot however many wait. It is not fetched again: - that would repeat the ladder for every copy that arrives. + claims, so the envelope is taken back out (`revert_envelope`: no FULL node, + no recorded verdict, no arrival time) and only that envelope (by its own root) + is refused; the root's payload is not condemned, since the builder's real + envelope may still arrive. Since `engine_newPayloadV2` a block hash mismatch + is answered `INVALID` with a null `latestValidHash` (`INVALID_BLOCK_HASH` is + "supplanted by `INVALID`", execution-apis `shanghai.md`), and an + executed-and-failed payload `INVALID` with its last valid ancestor. So a null + one is read as a mismatch and a non-null one as the payload being invalid. A + payload that failed with no valid ancestor to name is read as a mismatch too, + which only costs a refused envelope. Lighthouse and Prysm reject only the + envelope too; they never apply it before the answer, so they have nothing to + take back out. +- **Blocks built on a FULL payload wait for its answer**, not only for its + envelope, which is what lets a refused envelope be taken back out with nothing + on top of it. `on_block` would accept them as soon as the envelope is applied; + Lighthouse (`get_parent_import_status`) and Prysm (`ParentPayloadReady`) hold + the same blocks, since their FULL node exists only after the answer. +- **No answer after the retry ladder:** the envelope stays applied and + `NOT_VALIDATED`, the blocks waiting on it keep waiting, and it is asked again + once a slot, as for a pre-gloas block. `latestValidHash` and `VALID` follow the execution-layer chain, not every verified ancestor: a gloas block's EL parent is the ancestor whose payload hash