diff --git a/beacon_node/beacon_chain/src/block_verification.rs b/beacon_node/beacon_chain/src/block_verification.rs index f4d8b80c42..84df30e75f 100644 --- a/beacon_node/beacon_chain/src/block_verification.rs +++ b/beacon_node/beacon_chain/src/block_verification.rs @@ -942,12 +942,18 @@ impl GossipVerifiedBlock { // Check that we've received the parent envelope. If not, issue a single envelope // lookup for the parent and queue this block in the reprocess queue. + // + // The anchor block (proto-array root) is implicitly considered to have its payload + // received: there is no envelope to fetch for the anchor (per spec, the anchor is + // never added to `store.payloads`), and the anchor is trusted by definition. let parent_is_gloas = chain .spec .fork_name_at_slot::(parent_block.slot) .gloas_enabled(); + let parent_is_anchor = parent_block.parent_root.is_none(); if parent_is_gloas + && !parent_is_anchor && !fork_choice_read_lock.is_payload_received(&block.message().parent_root()) { return Err(BlockError::ParentEnvelopeUnknown { diff --git a/beacon_node/beacon_chain/src/payload_envelope_verification/import.rs b/beacon_node/beacon_chain/src/payload_envelope_verification/import.rs index b40e8337fb..bb1f1b53c7 100644 --- a/beacon_node/beacon_chain/src/payload_envelope_verification/import.rs +++ b/beacon_node/beacon_chain/src/payload_envelope_verification/import.rs @@ -14,7 +14,8 @@ use super::{ }; use crate::{ AvailabilityProcessingStatus, BeaconChain, BeaconChainError, BeaconChainTypes, - NotifyExecutionLayer, block_verification_types::AvailableBlockData, metrics, + NotifyExecutionLayer, block_verification::PayloadVerificationOutcome, + block_verification_types::AvailableBlockData, metrics, payload_envelope_verification::ExecutionPendingEnvelope, validator_monitor::get_slot_delay_ms, }; @@ -99,9 +100,18 @@ impl BeaconChain { self.import_available_execution_payload_envelope(Box::new(envelope)) .await } - ExecutedEnvelope::AvailabilityPending() => Err(EnvelopeError::InternalError( - "Pending payload envelope not yet implemented".to_owned(), - )), + ExecutedEnvelope::AvailabilityPending { + signed_envelope, + import_data, + payload_verification_outcome, + } => { + self.import_pending_execution_payload_envelope( + signed_envelope, + import_data, + payload_verification_outcome, + ) + .await + } } }; @@ -186,6 +196,39 @@ impl BeaconChain { )) } + /// Import an envelope whose data column availability has not yet been satisfied. + /// + /// Marks the block's payload as received in fork choice and persists the envelope to the + /// store, but does not write data column ops. Columns are expected to arrive separately + /// (gossip, engineGetBlobs, or reconstruction). + #[instrument(skip_all)] + pub async fn import_pending_execution_payload_envelope( + self: &Arc, + signed_envelope: Arc>, + import_data: EnvelopeImportData, + payload_verification_outcome: PayloadVerificationOutcome, + ) -> Result { + let EnvelopeImportData { + block_root, + _phantom, + } = import_data; + let block_root = { + let chain = self.clone(); + self.spawn_blocking_handle( + move || { + chain.import_execution_payload_envelope_pending_columns( + signed_envelope, + block_root, + payload_verification_outcome.payload_verification_status, + ) + }, + "payload_verification_handle", + ) + .await?? + }; + Ok(AvailabilityProcessingStatus::Imported(block_root)) + } + #[instrument(skip_all)] pub async fn import_available_execution_payload_envelope( self: &Arc, @@ -220,6 +263,50 @@ impl BeaconChain { Ok(AvailabilityProcessingStatus::Imported(block_root)) } + /// Same as `import_execution_payload_envelope` but for envelopes whose data columns + /// have not yet been received. Marks the payload as received in fork choice and + /// persists the envelope; columns are persisted separately as they arrive. + #[instrument(skip_all)] + fn import_execution_payload_envelope_pending_columns( + &self, + signed_envelope: Arc>, + block_root: Hash256, + payload_verification_status: PayloadVerificationStatus, + ) -> Result { + let fork_choice_reader = self.canonical_head.fork_choice_upgradable_read_lock(); + if !fork_choice_reader.contains_block(&block_root) { + return Err(EnvelopeError::BlockRootUnknown { block_root }); + } + + let mut fork_choice = parking_lot::RwLockUpgradableReadGuard::upgrade(fork_choice_reader); + fork_choice + .on_valid_payload_envelope_received(block_root) + .map_err(|e| EnvelopeError::InternalError(format!("{e:?}")))?; + + let db_write_timer = metrics::start_timer(&metrics::ENVELOPE_PROCESSING_DB_WRITE); + let ops = vec![StoreOp::PutPayloadEnvelope( + block_root, + signed_envelope.clone(), + )]; + let db_span = info_span!("persist_envelope_pending_columns").entered(); + if let Err(e) = self.store.do_atomically_with_block_and_blobs_cache(ops) { + error!(error = ?e, "Database write failed for pending-columns envelope"); + return Err(e.into()); + } + drop(db_span); + drop(fork_choice); + + let envelope_time_imported = self.slot_clock.now_duration().unwrap_or(Duration::MAX); + metrics::stop_timer(db_write_timer); + self.import_envelope_update_metrics_and_events( + signed_envelope, + block_root, + payload_verification_status, + envelope_time_imported, + ); + Ok(block_root) + } + /// Accepts a fully-verified and available envelope and imports it into the chain without performing any /// additional verification. /// diff --git a/beacon_node/beacon_chain/src/payload_envelope_verification/mod.rs b/beacon_node/beacon_chain/src/payload_envelope_verification/mod.rs index b153a3cd6a..5558cbc50c 100644 --- a/beacon_node/beacon_chain/src/payload_envelope_verification/mod.rs +++ b/beacon_node/beacon_chain/src/payload_envelope_verification/mod.rs @@ -119,12 +119,16 @@ pub struct EnvelopeProcessingSnapshot { /// 1. `Available`: This envelope has been executed and also contains all data to consider it /// fully available. /// 2. `AvailabilityPending`: This envelope hasn't received all required blobs to consider it -/// fully available. -#[allow(dead_code)] +/// fully available. The envelope is still imported (fork-choice marks the block's payload +/// as received and the envelope is persisted); column persistence is handled separately +/// via gossip / engineGetBlobs as columns arrive. pub enum ExecutedEnvelope { Available(AvailableExecutedEnvelope), - // TODO(gloas): check data column availability via DA checker - AvailabilityPending(), + AvailabilityPending { + signed_envelope: Arc>, + import_data: EnvelopeImportData, + payload_verification_outcome: PayloadVerificationOutcome, + }, } impl ExecutedEnvelope { @@ -142,15 +146,14 @@ impl ExecutedEnvelope { payload_verification_outcome, )) } - // TODO(gloas): check data column availability via DA checker MaybeAvailableEnvelope::AvailabilityPending { - block_hash, - envelope, - } => Self::Available(AvailableExecutedEnvelope::new( - AvailableEnvelope::new(block_hash, envelope, vec![], None, spec), + block_hash: _, + envelope: signed_envelope, + } => Self::AvailabilityPending { + signed_envelope, import_data, payload_verification_outcome, - )), + }, } } } diff --git a/beacon_node/network/src/network_beacon_processor/gossip_methods.rs b/beacon_node/network/src/network_beacon_processor/gossip_methods.rs index 0f132d5dd6..5a0dffed4d 100644 --- a/beacon_node/network/src/network_beacon_processor/gossip_methods.rs +++ b/beacon_node/network/src/network_beacon_processor/gossip_methods.rs @@ -3978,8 +3978,13 @@ impl NetworkBeaconProcessor { // register_process_result_metrics(&result, metrics::BlockSource::Gossip, "envelope"); match &result { - Ok(AvailabilityProcessingStatus::Imported(_)) - | Ok(AvailabilityProcessingStatus::MissingComponents(_, _)) => { + Ok(AvailabilityProcessingStatus::Imported(block_root)) => { + // Notify sync so any pending child lookup awaiting this parent envelope unblocks. + self.send_sync_message(SyncMessage::GossipEnvelopeImported { + block_root: *block_root, + }); + } + Ok(AvailabilityProcessingStatus::MissingComponents(_, _)) => { // Nothing to do } Err(e) => match e { diff --git a/beacon_node/network/src/network_beacon_processor/tests.rs b/beacon_node/network/src/network_beacon_processor/tests.rs index 76c6ba812d..2a7542e73b 100644 --- a/beacon_node/network/src/network_beacon_processor/tests.rs +++ b/beacon_node/network/src/network_beacon_processor/tests.rs @@ -9,6 +9,7 @@ use crate::{ sync::{SyncMessage, manager::BlockProcessType}, }; use beacon_chain::block_verification_types::LookupBlock; +use beacon_chain::chain_config::ChainConfig; use beacon_chain::custody_context::NodeCustodyType; use beacon_chain::data_column_verification::validate_data_column_sidecar_for_gossip_fulu; use beacon_chain::kzg_utils::blobs_to_data_column_sidecars; @@ -134,7 +135,10 @@ impl TestRig { .fresh_ephemeral_store() .mock_execution_layer() .node_custody_type(NodeCustodyType::Fullnode) - .chain_config(<_>::default()) + .chain_config(ChainConfig { + disable_get_blobs: true, + ..ChainConfig::default() + }) .build(); harness.advance_slot(); @@ -169,7 +173,10 @@ impl TestRig { .fresh_ephemeral_store() .mock_execution_layer() .node_custody_type(node_custody_type) - .chain_config(<_>::default()) + .chain_config(ChainConfig { + disable_get_blobs: true, + ..ChainConfig::default() + }) .build(); harness.advance_slot(); @@ -1001,14 +1008,30 @@ async fn data_column_reconstruction_at_deadline() { rig.enqueue_gossip_data_columns(i); } - // Expect all gossip events + reconstruction - let mut expected_events: Vec = (0..min_columns_for_reconstruction) - .map(|_| WorkType::GossipDataColumnSidecar) - .collect(); - expected_events.push(WorkType::ColumnReconstruction); - - rig.assert_event_journal_contains_ordered(&expected_events) - .await; + // Drain the journal until we've seen all gossip events plus at least one + // reconstruction. Under real crypto the reprocess queue can dispatch the + // reconstruction work item more than once (the second is a no-op via + // `reconstruction_started`), so we don't pin the count — we just require >= 1. + let gsc: &str = WorkType::GossipDataColumnSidecar.into(); + let cr: &str = WorkType::ColumnReconstruction.into(); + let (mut gossip_seen, mut recon_seen) = (0usize, 0usize); + let drain = async { + while let Some(event) = rig.work_journal_rx.recv().await { + if event == gsc { + gossip_seen += 1; + } else if event == cr { + recon_seen += 1; + } + if gossip_seen == min_columns_for_reconstruction && recon_seen >= 1 { + break; + } + } + }; + if tokio::time::timeout(STANDARD_TIMEOUT, drain).await.is_err() { + panic!("timeout: gossip_seen={gossip_seen}, recon_seen={recon_seen}"); + } + assert_eq!(gossip_seen, min_columns_for_reconstruction); + assert!(recon_seen >= 1); } // Test the column reconstruction is delayed for columns that arrive for a previous slot. diff --git a/beacon_node/network/src/sync/manager.rs b/beacon_node/network/src/sync/manager.rs index f60c3949c9..869e7e32b0 100644 --- a/beacon_node/network/src/sync/manager.rs +++ b/beacon_node/network/src/sync/manager.rs @@ -154,6 +154,10 @@ pub enum SyncMessage { /// A block's parent is known but its execution payload envelope has not been received yet. UnknownParentEnvelope(PeerId, Arc>, Hash256), + /// An execution payload envelope has been imported via the local gossip path. + /// Sync uses this to unblock any child lookups that were awaiting this parent envelope. + GossipEnvelopeImported { block_root: Hash256 }, + /// A partial data column with an unknown parent has been received. UnknownParentPartialDataColumn { peer_id: PeerId, @@ -961,6 +965,14 @@ impl SyncManager { }), ); } + SyncMessage::GossipEnvelopeImported { block_root } => { + debug!( + %block_root, + "Gossip-imported envelope; unblocking awaiting child lookups" + ); + self.block_lookups + .continue_envelope_child_lookups(block_root, &mut self.network); + } SyncMessage::UnknownParentPartialDataColumn { peer_id, block_root, @@ -1096,6 +1108,21 @@ impl SyncManager { slot: Slot, block_component: BlockComponent, ) { + // Defensive: if the parent's payload envelope was already received between when + // gossip-verification raised `ParentEnvelopeUnknown` and now, no lookup is needed. + if self + .chain + .canonical_head + .fork_choice_read_lock() + .is_payload_received(&parent_root) + { + debug!( + %block_root, + %parent_root, + "Parent envelope already received, skipping envelope lookup" + ); + return; + } match self.should_search_for_block(Some(slot), &peer_id) { Ok(_) => { if self.block_lookups.search_child_and_parent_envelope( diff --git a/beacon_node/network/src/sync/tests/lookups.rs b/beacon_node/network/src/sync/tests/lookups.rs index a26996ec5e..35c45eb928 100644 --- a/beacon_node/network/src/sync/tests/lookups.rs +++ b/beacon_node/network/src/sync/tests/lookups.rs @@ -37,7 +37,7 @@ use tokio::sync::mpsc; use tracing::info; use types::{ BlobSidecar, BlockImportSource, ColumnIndex, DataColumnSidecar, EthSpec, ForkContext, ForkName, - Hash256, MinimalEthSpec as E, SignedBeaconBlock, Slot, + Hash256, MinimalEthSpec as E, SignedBeaconBlock, SignedExecutionPayloadEnvelope, Slot, test_utils::{SeedableRng, XorShiftRng}, }; @@ -85,6 +85,9 @@ pub struct SimulateConfig { ee_offline_for_n_range_responses: Option, /// Disconnect all peers after this many successful BlocksByRange responses. successful_range_responses_before_disconnect: Option, + /// Number of `PayloadEnvelopesByRoot` responses that return an envelope for a + /// different block_root than requested. + return_wrong_envelopes_n_times: usize, } impl SimulateConfig { @@ -116,6 +119,11 @@ impl SimulateConfig { self } + fn return_wrong_envelope_once(mut self) -> Self { + self.return_wrong_envelopes_n_times = 1; + self + } + fn return_wrong_sidecar_for_block_once(mut self) -> Self { self.return_wrong_sidecar_for_block_n_times = 1; self @@ -209,6 +217,9 @@ pub(crate) struct TestRigConfig { fulu_test_type: FuluTestType, /// Override the node custody type derived from `fulu_test_type` node_custody_type_override: Option, + /// Override the number of validators in the harness genesis state. Defaults to 1. + /// Some forks (e.g. Gloas) cannot initialise a state with a single validator. + validator_count_override: Option, } impl TestRig { @@ -222,9 +233,9 @@ impl TestRig { ); // Initialise a new beacon chain - let harness = BeaconChainHarness::>::builder(E) + let mut builder = BeaconChainHarness::>::builder(E) .spec(spec.clone()) - .deterministic_keypairs(1) + .deterministic_keypairs(test_rig_config.validator_count_override.unwrap_or(1)) .fresh_ephemeral_store() .mock_execution_layer() .testing_slot_clock(clock.clone()) @@ -232,8 +243,17 @@ impl TestRig { test_rig_config .node_custody_type_override .unwrap_or_else(|| test_rig_config.fulu_test_type.we_node_custody_type()), - ) - .build(); + ); + // Post-Electra forks need validators with effective balance close to + // `max_effective_balance_electra` for balance-weighted committee + // selection (sync committee, PTC) to converge during genesis. + if spec.electra_fork_epoch == Some(types::Epoch::new(0)) { + let max_eb = spec.max_effective_balance_electra; + builder = builder.with_genesis_state_builder(move |b| { + b.set_initial_balance_fn(Box::new(move |_| max_eb)) + }); + } + let harness = builder.build(); let chain = harness.chain.clone(); let fork_context = Arc::new(ForkContext::new::( @@ -305,6 +325,7 @@ impl TestRig { fork_name, network_blocks_by_root: <_>::default(), network_blocks_by_slot: <_>::default(), + network_envelopes_by_root: <_>::default(), penalties: <_>::default(), seen_lookups: <_>::default(), requests: <_>::default(), @@ -319,6 +340,7 @@ impl TestRig { Self::new(TestRigConfig { fulu_test_type: FuluTestType::WeFullnodeThemSupernode, node_custody_type_override: None, + validator_count_override: None, }) } @@ -327,6 +349,7 @@ impl TestRig { Self::new(TestRigConfig { fulu_test_type: FuluTestType::WeFullnodeThemSupernode, node_custody_type_override: Some(node_custody_type), + validator_count_override: None, }) } @@ -429,9 +452,9 @@ impl TestRig { process_fn.await } } - Work::RpcBlobs { process_fn } | Work::RpcCustodyColumn(process_fn) => { - process_fn.await - } + Work::RpcBlobs { process_fn } + | Work::RpcCustodyColumn(process_fn) + | Work::RpcPayloadEnvelope { process_fn } => process_fn.await, Work::ChainSegment { process_fn, process_id: (chain_id, batch_epoch), @@ -671,6 +694,45 @@ impl TestRig { self.send_rpc_columns_response(req_id, peer_id, &columns); } + (RequestType::PayloadEnvelopesByRoot(req), AppRequestId::Sync(req_id)) => { + if self.complete_strategy.return_no_data_n_times > 0 { + self.complete_strategy.return_no_data_n_times -= 1; + return self.send_rpc_envelopes_response(req_id, peer_id, &[]); + } + + if self.complete_strategy.return_wrong_envelopes_n_times > 0 { + self.complete_strategy.return_wrong_envelopes_n_times -= 1; + // Return any envelope that doesn't match the request, so the + // request items layer raises `UnrequestedBlockRoot`. + let requested = req + .beacon_block_roots + .iter() + .copied() + .collect::>(); + let wrong = self + .network_envelopes_by_root + .iter() + .find(|(root, _)| !requested.contains(*root)) + .map(|(_, envelope)| envelope.clone()) + .expect("test fixture must produce at least one extra envelope"); + return self.send_rpc_envelopes_response(req_id, peer_id, &[wrong]); + } + + let envelopes = req + .beacon_block_roots + .iter() + .map(|block_root| { + self.network_envelopes_by_root + .get(block_root) + .unwrap_or_else(|| { + panic!("Test consumer requested unknown envelope: {block_root:?}") + }) + .clone() + }) + .collect::>(); + self.send_rpc_envelopes_response(req_id, peer_id, &envelopes); + } + (RequestType::BlocksByRange(req), AppRequestId::Sync(req_id)) => { if self.complete_strategy.skip_by_range_routes { return; @@ -894,6 +956,36 @@ impl TestRig { }); } + fn send_rpc_envelopes_response( + &mut self, + sync_request_id: SyncRequestId, + peer_id: PeerId, + envelopes: &[Arc>], + ) { + let block_roots = envelopes + .iter() + .map(|e| e.beacon_block_root()) + .collect::>(); + self.log(&format!( + "Completing request {sync_request_id:?} to {peer_id} with envelopes for {block_roots:?}" + )); + + for envelope in envelopes { + self.push_sync_message(SyncMessage::RpcPayloadEnvelope { + sync_request_id, + peer_id, + envelope: Some(envelope.clone()), + seen_timestamp: D, + }); + } + self.push_sync_message(SyncMessage::RpcPayloadEnvelope { + sync_request_id, + peer_id, + envelope: None, + seen_timestamp: D, + }); + } + fn send_rpc_columns_response( &mut self, sync_request_id: SyncRequestId, @@ -936,16 +1028,25 @@ impl TestRig { pub(super) async fn build_chain(&mut self, block_count: usize) -> Hash256 { let mut blocks = vec![]; - // Initialise a new beacon chain - let external_harness = BeaconChainHarness::>::builder(E) + // Initialise a new beacon chain. Match the local harness's validator count and + // balance hooks so post-Electra forks (where genesis-time committee selection is + // balance-weighted) can initialise. + let validator_count = self.harness.validator_keypairs.len(); + let mut builder = BeaconChainHarness::>::builder(E) .spec(self.harness.spec.clone()) - .deterministic_keypairs(1) + .deterministic_keypairs(validator_count) .fresh_ephemeral_store() .mock_execution_layer() .testing_slot_clock(self.harness.chain.slot_clock.clone()) // Make the external harness a supernode so all columns are available - .node_custody_type(NodeCustodyType::Supernode) - .build(); + .node_custody_type(NodeCustodyType::Supernode); + if self.harness.spec.electra_fork_epoch == Some(types::Epoch::new(0)) { + let max_eb = self.harness.spec.max_effective_balance_electra; + builder = builder.with_genesis_state_builder(move |b| { + b.set_initial_balance_fn(Box::new(move |_| max_eb)) + }); + } + let external_harness = builder.build(); // Ensure all blocks have data. Otherwise, the triggers for unknown blob parent and unknown // data column parent fail. external_harness @@ -974,6 +1075,16 @@ impl TestRig { self.network_blocks_by_root .insert(block_root, block.clone()); self.network_blocks_by_slot.insert(block_slot, block); + // Post-Gloas, also capture the execution payload envelope so peers can serve it. + if self.is_after_gloas() + && let Ok(Some(envelope)) = external_harness + .chain + .store + .get_payload_envelope(&block_root) + { + self.network_envelopes_by_root + .insert(block_root, Arc::new(envelope)); + } self.log(&format!( "Produced block {} index {i} in external harness", block_slot, @@ -1002,6 +1113,21 @@ impl TestRig { self.re_insert_block(Arc::new(block), blobs, columns); } + /// Replace the cached envelope's signature for `block_root` with one signed by an + /// unrelated key, so it fails verification against the proposer's pubkey. + fn corrupt_envelope_signature_for(&mut self, block_root: Hash256) { + let envelope = self + .network_envelopes_by_root + .get(&block_root) + .expect("no envelope cached for block_root") + .as_ref() + .clone(); + let mut envelope = envelope; + envelope.signature = self.valid_signature(); + self.network_envelopes_by_root + .insert(block_root, Arc::new(envelope)); + } + fn valid_signature(&mut self) -> bls::Signature { let keypair = bls::Keypair::random(); let msg = Hash256::random(); @@ -1178,6 +1304,32 @@ impl TestRig { self.harness.chain.recompute_head_at_current_slot().await; } + /// Persist a Gloas execution payload envelope into the local chain and mark the + /// block as "payload received" in fork choice. Mimics the side-effects of the + /// gossip-import path, including the `GossipEnvelopeImported` sync notification. + /// The caller is responsible for ensuring the corresponding beacon block is + /// already imported. + async fn import_envelope_for_block_root(&mut self, block_root: Hash256) { + let envelope = self + .network_envelopes_by_root + .get(&block_root) + .unwrap_or_else(|| panic!("no envelope cached for {block_root:?}")) + .as_ref() + .clone(); + self.harness + .chain + .store + .put_payload_envelope(&block_root, &envelope) + .expect("should store envelope"); + self.harness + .chain + .canonical_head + .fork_choice_write_lock() + .on_valid_payload_envelope_received(block_root) + .expect("should update fork choice with envelope"); + self.push_sync_message(SyncMessage::GossipEnvelopeImported { block_root }); + } + /// Import a block directly into the chain without going through lookup sync async fn import_block_by_root(&mut self, block_root: Hash256) { let range_sync_block = self @@ -1444,6 +1596,7 @@ impl TestRig { Self::new(TestRigConfig { fulu_test_type, node_custody_type_override: None, + validator_count_override: None, }) }) } @@ -1460,6 +1613,22 @@ impl TestRig { self.fork_name.fulu_enabled() } + pub fn is_after_gloas(&self) -> bool { + self.fork_name.gloas_enabled() + } + + fn new_after_gloas() -> Option { + // Gloas requires more than 1 validator to initialise the genesis state + // (committee/sampling computations fail with `InvalidIndicesCount`). + genesis_fork().gloas_enabled().then(|| { + Self::new(TestRigConfig { + fulu_test_type: FuluTestType::WeFullnodeThemSupernode, + node_custody_type_override: None, + validator_count_override: Some(1024), + }) + }) + } + fn trigger_unknown_parent_block(&mut self, peer_id: PeerId, block: Arc>) { let block_root = block.canonical_root(); self.send_sync_message(SyncMessage::UnknownParentBlock(peer_id, block, block_root)) @@ -1483,6 +1652,18 @@ impl TestRig { )); } + /// Trigger an envelope-unknown lookup for the last block in the chain. Caller is + /// expected to have already imported the parent block (via `import_blocks_up_to_slot`) + /// without registering its envelope. + fn trigger_with_last_unknown_parent_envelope(&mut self) { + let peer_id = self.new_connected_supernode_peer(); + let last_block = self.get_last_block().block_cloned(); + let block_root = last_block.canonical_root(); + self.send_sync_message(SyncMessage::UnknownParentEnvelope( + peer_id, last_block, block_root, + )); + } + fn rand_block(&mut self) -> SignedBeaconBlock { self.rand_block_and_blobs(NumBlobs::None).0 } @@ -2639,3 +2820,172 @@ async fn crypto_on_fail_with_bad_column_kzg_proof() { r.assert_penalties_of_type("lookup_custody_column_processing_failure"); } } + +// --------------------------------------------------------------------------- +// Gloas: parent envelope unknown lookup +// --------------------------------------------------------------------------- +// +// These tests exercise the lookup-sync state machine introduced in PR #9039: +// when a gossip block's parent execution payload envelope is missing, +// `SyncManager` is expected to create two single-block lookups — an envelope-only +// lookup for the parent block_root and a "child" lookup that holds the gossip +// block and waits on `AwaitingParent::Envelope(parent_root)`. The envelope-only +// lookup issues a `PayloadEnvelopesByRoot` RPC; on completion it unblocks the +// child via `continue_envelope_child_lookups`. +// +// The tests below cover lookup creation, RPC routing, and drop-cascade +// behaviour. The end-to-end happy path is gated on +// `process_execution_payload_envelope` supporting `AvailabilityPending` (today +// it returns `InternalError("Pending payload envelope not yet implemented")`), +// which is tracked separately. See `process_rpc_envelope` in `sync_methods.rs`. + +/// Builds a 2-block gloas chain in the external harness and locally imports block 1 +/// (parent) WITHOUT registering its envelope, leaving `is_payload_received(parent_root)` +/// false — the precondition for `BlockError::ParentEnvelopeUnknown`. +async fn setup_unknown_parent_envelope_scenario() -> Option { + let mut r = TestRig::new_after_gloas()?; + r.build_chain(2).await; + r.import_blocks_up_to_slot(1).await; + Some(r) +} + +fn payload_envelope_request_count(rig: &TestRig) -> usize { + rig.requests + .iter() + .filter(|(request, _)| matches!(request, RequestType::PayloadEnvelopesByRoot(_))) + .count() +} + +/// Triggering `UnknownParentEnvelope` creates exactly two lookups: an envelope-only +/// lookup for the parent and a child lookup for the gossip block awaiting that envelope. +#[tokio::test] +async fn unknown_parent_envelope_creates_two_lookups() { + let Some(mut r) = setup_unknown_parent_envelope_scenario().await else { + return; + }; + r.trigger_with_last_unknown_parent_envelope(); + r.assert_single_lookups_count(2); +} + +/// Repeated `UnknownParentEnvelope` triggers for the same parent must not spawn extra +/// lookups (peers are merged into the existing envelope lookup). +#[tokio::test] +async fn happy_path_unknown_parent_envelope_multiple_triggers() { + let Some(mut r) = setup_unknown_parent_envelope_scenario().await else { + return; + }; + r.trigger_with_last_unknown_parent_envelope(); + r.trigger_with_last_unknown_parent_envelope(); + r.assert_single_lookups_count(2); +} + +/// The envelope-only lookup must dispatch a `PayloadEnvelopesByRoot` RPC for the +/// parent block_root. +#[tokio::test] +async fn envelope_lookup_issues_by_root_rpc() { + let Some(mut r) = setup_unknown_parent_envelope_scenario().await else { + return; + }; + r.trigger_with_last_unknown_parent_envelope(); + r.simulate(SimulateConfig::new()).await; + assert_eq!( + payload_envelope_request_count(&r), + 1, + "expected exactly one PayloadEnvelopesByRoot request" + ); +} + +/// One transient RPC error on the envelope request → lookup retries with the same peer +/// and completes successfully. Mirrors the `bad_peer_rpc_failure` shape used for blocks. +#[tokio::test] +async fn bad_peer_envelope_rpc_failure() { + let Some(mut r) = setup_unknown_parent_envelope_scenario().await else { + return; + }; + r.trigger_with_last_unknown_parent_envelope(); + r.simulate(SimulateConfig::new().return_rpc_error(RPCError::IoError("test".into()))) + .await; + r.assert_successful_lookup_sync(); + r.assert_head_slot(2); +} + +/// Peer responds once with an envelope for a different block_root than requested. +/// The request-items layer raises `UnrequestedBlockRoot`, the peer is penalised, and +/// the lookup retries successfully on the next request. +#[tokio::test] +async fn bad_peer_wrong_envelope_response() { + let Some(mut r) = setup_unknown_parent_envelope_scenario().await else { + return; + }; + r.trigger_with_last_unknown_parent_envelope(); + r.simulate(SimulateConfig::new().return_wrong_envelope_once()) + .await; + r.assert_penalties_of_type("UnrequestedBlockRoot"); + r.assert_successful_lookup_sync(); + r.assert_head_slot(2); +} + +/// Trigger `UnknownParentEnvelope` when the parent's payload envelope is already +/// in fork choice. Sync should treat the trigger as a no-op and create no lookups. +#[tokio::test] +async fn envelope_already_received_skips_lookup() { + let Some(mut r) = setup_unknown_parent_envelope_scenario().await else { + return; + }; + let parent_root = r.get_last_block().block_cloned().parent_root(); + r.import_envelope_for_block_root(parent_root).await; + r.trigger_with_last_unknown_parent_envelope(); + r.assert_single_lookups_count(0); +} + +/// End-to-end: an envelope-only RPC lookup completes, the cached child block is +/// processed, and the head advances to the gossip block. +#[tokio::test] +async fn happy_path_unknown_parent_envelope() { + let Some(mut r) = setup_unknown_parent_envelope_scenario().await else { + return; + }; + r.trigger_with_last_unknown_parent_envelope(); + r.simulate(SimulateConfig::happy_path()).await; + r.assert_successful_lookup_sync(); + r.assert_head_slot(2); + r.assert_no_penalties(); +} + +/// While an envelope-only RPC lookup is pending, the same envelope is imported +/// via the gossip path. The child lookup should still unblock and import. +#[tokio::test] +async fn happy_path_unknown_parent_envelope_via_gossip() { + let Some(mut r) = setup_unknown_parent_envelope_scenario().await else { + return; + }; + let parent_root = r.get_last_block().block_cloned().parent_root(); + r.trigger_with_last_unknown_parent_envelope(); + // Import the envelope via the local gossip path before any RPC response arrives. + r.import_envelope_for_block_root(parent_root).await; + r.simulate(SimulateConfig::happy_path()).await; + r.assert_successful_lookup_sync(); + r.assert_head_slot(2); +} + +/// Peer returns the requested envelope but with a corrupted signature. Gossip +/// verification rejects it; the lookup retries (single peer → exhaust → drop) +/// and reports `lookup_envelope_processing_failure` against the peer. +#[tokio::test] +async fn crypto_on_fail_with_bad_envelope_signature() { + let Some(mut r) = setup_unknown_parent_envelope_scenario().await else { + return; + }; + let parent_root = r.get_last_block().block_cloned().parent_root(); + r.corrupt_envelope_signature_for(parent_root); + r.trigger_with_last_unknown_parent_envelope(); + r.simulate(SimulateConfig::happy_path()).await; + if cfg!(feature = "fake_crypto") { + // Under fake_crypto, signature checks are no-ops, so a "corrupted" + // signature still passes. Skip — analogous to the existing + // `crypto_on_fail_with_invalid_block_signature` test. + return; + } + r.assert_failed_lookup_sync(); + r.assert_penalties_of_type("lookup_envelope_processing_failure"); +} diff --git a/beacon_node/network/src/sync/tests/mod.rs b/beacon_node/network/src/sync/tests/mod.rs index 6e948e4726..29dd7b898e 100644 --- a/beacon_node/network/src/sync/tests/mod.rs +++ b/beacon_node/network/src/sync/tests/mod.rs @@ -22,7 +22,7 @@ use tokio::sync::mpsc; use tracing_subscriber::fmt::MakeWriter; use tracing_subscriber::layer::SubscriberExt; use tracing_subscriber::util::SubscriberInitExt; -use types::{ForkName, Hash256, MinimalEthSpec as E, Slot}; +use types::{ForkName, Hash256, MinimalEthSpec as E, SignedExecutionPayloadEnvelope, Slot}; mod lookups; mod range; @@ -79,6 +79,8 @@ struct TestRig { /// Blocks that will be used in the test but may not be known to `harness` yet. network_blocks_by_root: HashMap>, network_blocks_by_slot: HashMap>, + /// Execution payload envelopes (Gloas) keyed by beacon block root, available to peers. + network_envelopes_by_root: HashMap>>, penalties: Vec, /// All seen lookups through the test run seen_lookups: HashMap,