merge from unstable

beacon_node/http_api/src/beacon/execution_payload_envelope.rs

@@ -7,7 +7,7 @@ use crate::version::{
     execution_optimistic_finalized_beacon_response,
 };
 use beacon_chain::data_column_verification::{GossipDataColumnError, GossipVerifiedDataColumn};
-use beacon_chain::{BeaconChain, BeaconChainTypes};
+use beacon_chain::{BeaconChain, BeaconChainTypes, NotifyExecutionLayer};
 use bytes::Bytes;
 use eth2::types as api_types;
 use eth2::{CONTENT_TYPE_HEADER, SSZ_CONTENT_TYPE_HEADER};
@@ -18,7 +18,7 @@ use std::future::Future;
 use std::sync::Arc;
 use tokio::sync::mpsc::UnboundedSender;
 use tracing::{debug, error, info, warn};
-use types::{EthSpec, SignedExecutionPayloadEnvelope};
+use types::{BlockImportSource, EthSpec, SignedExecutionPayloadEnvelope};
 use warp::{
     Filter, Rejection, Reply,
     hyper::{Body, Response},
@@ -99,14 +99,12 @@ pub async fn publish_execution_payload_envelope<T: BeaconChainTypes>(
     let slot = envelope.slot();
     let beacon_block_root = envelope.message.beacon_block_root;
 
-    // TODO(gloas): Replace this check once we have gossip validation.
     if !chain.spec.is_gloas_scheduled() {
         return Err(warp_utils::reject::custom_bad_request(
             "Execution payload envelopes are not supported before the Gloas fork".into(),
         ));
     }
 
-    // TODO(gloas): We should probably add validation here i.e. BroadcastValidation::Gossip
     info!(
         %slot,
         %beacon_block_root,
@@ -118,7 +116,7 @@ pub async fn publish_execution_payload_envelope<T: BeaconChainTypes>(
 
     // Spawn the column-build task (CPU-bound KZG cell-and-proof computation) before
     // publishing the envelope so it runs in parallel with envelope gossip, narrowing
-    // the window in which peers see envelope-without-columns. If envelope publication
+    // the window in which peers see envelope-without-columns. If envelope import
     // fails below, dropping this future drops the spawned `JoinHandle` (the running
     // closure on the blocking pool finishes and is then discarded — no work cancellation).
     let column_build_future = match blobs_and_proofs {
@@ -131,17 +129,47 @@ pub async fn publish_execution_payload_envelope<T: BeaconChainTypes>(
         _ => None,
     };
 
-    // Publish the envelope to the network.
-    crate::utils::publish_pubsub_message(
-        network_tx,
-        PubsubMessage::ExecutionPayload(Box::new(envelope)),
-    )
-    .map_err(|_| {
-        warn!(%slot, "Failed to publish execution payload envelope to network");
-        warp_utils::reject::custom_server_error(
-            "Unable to publish execution payload envelope to network".into(),
+    // Gossip-verify the envelope before publishing.
+    let gossip_verified = chain
+        .verify_envelope_for_gossip(Arc::new(envelope))
+        .await
+        .map_err(|e| {
+            warn!(%slot, error = ?e, "Execution payload envelope failed gossip verification");
+            warp_utils::reject::custom_bad_request(format!(
+                "envelope failed gossip verification: {e}"
+            ))
+        })?;
+
+    let network_tx_clone = network_tx.clone();
+    let envelope_for_gossip = gossip_verified.signed_envelope.as_ref().clone();
+    let publish_fn = || {
+        crate::utils::publish_pubsub_message(
+            &network_tx_clone,
+            PubsubMessage::ExecutionPayload(Box::new(envelope_for_gossip)),
         )
-    })?;
+        .map_err(|_| {
+            beacon_chain::payload_envelope_verification::EnvelopeError::BeaconChainError(Arc::new(
+                beacon_chain::BeaconChainError::UnableToPublish,
+            ))
+        })
+    };
+
+    let import_result = chain
+        .process_execution_payload_envelope(
+            beacon_block_root,
+            gossip_verified,
+            NotifyExecutionLayer::Yes,
+            BlockImportSource::HttpApi,
+            publish_fn,
+        )
+        .await;
+
+    if let Err(e) = import_result {
+        warn!(%slot, error = ?e, "Failed to import execution payload envelope");
+        return Err(warp_utils::reject::custom_server_error(format!(
+            "envelope import failed: {e}"
+        )));
+    }
 
     // From here on the envelope is on the wire. `take_blobs` already consumed the cache
     // entry, so a retry would not republish columns; returning Err would mislead the

beacon_node/http_api/tests/tests.rs

@@ -3450,17 +3450,20 @@ impl ApiTester {
                 .unwrap()
                 .unwrap_or(self.chain.head_beacon_block_root());
 
-            // Presently, the beacon chain harness never runs the code that primes the proposer
-            // cache. If this changes in the future then we'll need some smarter logic here, but
-            // this is succinct and effective for the time being.
-            assert!(
-                self.chain
-                    .beacon_proposer_cache
-                    .lock()
-                    .get_epoch::<E>(dependent_root, epoch)
-                    .is_none(),
-                "the proposer cache should miss initially"
-            );
+            // Block import primes the proposer cache for each epoch it runs through (to gate
+            // proposer boost), so epochs `<= current_epoch` are already cached. The only epoch
+            // for which we can observe the endpoint's own caching behaviour is
+            // `current_epoch + 1`, which no block import has touched yet.
+            if epoch == current_epoch + 1 {
+                assert!(
+                    self.chain
+                        .beacon_proposer_cache
+                        .lock()
+                        .get_epoch::<E>(dependent_root, epoch)
+                        .is_none(),
+                    "the proposer cache should miss initially for the next epoch"
+                );
+            }
 
             let result = self
                 .client
@@ -3468,8 +3471,9 @@ impl ApiTester {
                 .await
                 .unwrap();
 
-            // Check that current-epoch requests prime the proposer cache, whilst non-current
-            // requests don't.
+            // A current-epoch request should leave the cache primed (block import already did so,
+            // but this is still a useful end-to-end check). A request for `current_epoch + 1`
+            // should not prime the cache.
             if epoch == current_epoch {
                 assert!(
                     self.chain
@@ -3477,16 +3481,16 @@ impl ApiTester {
                         .lock()
                         .get_epoch::<E>(dependent_root, epoch)
                         .is_some(),
-                    "a current-epoch request should prime the proposer cache"
+                    "the proposer cache should be primed for the current epoch"
                 );
-            } else {
+            } else if epoch == current_epoch + 1 {
                 assert!(
                     self.chain
                         .beacon_proposer_cache
                         .lock()
                         .get_epoch::<E>(dependent_root, epoch)
                         .is_none(),
-                    "a non-current-epoch request should not prime the proposer cache"
+                    "a request for the next epoch should not prime the proposer cache"
                 );
             }
 
@@ -4640,6 +4644,86 @@ impl ApiTester {
         self
     }
 
+    /// Regression test: publishing an envelope via the HTTP API must import it locally so
+    /// that `produce_payload_attestation_data` returns `payload_present = true`. Without
+    /// local import, the `envelope_times_cache` is never populated and PTC voters on the
+    /// same node incorrectly vote MISSING for their own payload.
+    pub async fn test_payload_attestation_present_after_envelope_publish(self) -> Self {
+        if !self.chain.spec.is_gloas_scheduled() {
+            return self;
+        }
+
+        let fork = self.chain.canonical_head.cached_head().head_fork();
+        let genesis_validators_root = self.chain.genesis_validators_root;
+
+        for _ in 0..E::slots_per_epoch() * 3 {
+            let slot = self.chain.slot().unwrap();
+            let epoch = self.chain.epoch().unwrap();
+            let fork_name = self.chain.spec.fork_name_at_slot::<E>(slot);
+
+            if !fork_name.gloas_enabled() {
+                self.chain.slot_clock.set_slot(slot.as_u64() + 1);
+                continue;
+            }
+
+            let (sk, randao_reveal) = self
+                .proposer_setup(slot, epoch, &fork, genesis_validators_root)
+                .await;
+
+            // Produce and publish a block.
+            let (response, _metadata) = self
+                .client
+                .get_validator_blocks_v4::<E>(slot, &randao_reveal, None, None, None)
+                .await
+                .unwrap();
+            let block = response.data;
+            let block_root = block.tree_hash_root();
+
+            let signed_block = block.sign(&sk, &fork, genesis_validators_root, &self.chain.spec);
+            let signed_block_request =
+                PublishBlockRequest::try_from(Arc::new(signed_block)).unwrap();
+            self.client
+                .post_beacon_blocks_v2(&signed_block_request, None)
+                .await
+                .unwrap();
+
+            // Retrieve and publish the envelope.
+            let envelope = self
+                .client
+                .get_validator_execution_payload_envelope::<E>(slot, BUILDER_INDEX_SELF_BUILD)
+                .await
+                .unwrap()
+                .data;
+
+            let signed_envelope =
+                self.sign_envelope(envelope, &sk, epoch, &fork, genesis_validators_root);
+            self.client
+                .post_beacon_execution_payload_envelope(&signed_envelope, fork_name)
+                .await
+                .unwrap();
+
+            // The payload attestation data endpoint must now report the payload as present.
+            let pa_data = self
+                .client
+                .get_validator_payload_attestation_data(slot)
+                .await
+                .unwrap()
+                .into_data();
+
+            assert_eq!(pa_data.beacon_block_root, block_root);
+            assert_eq!(pa_data.slot, slot);
+            assert!(
+                pa_data.payload_present,
+                "payload attestation should report payload_present=true after publishing \
+                 the envelope via the HTTP API (slot {slot})"
+            );
+
+            self.chain.slot_clock.set_slot(slot.as_u64() + 1);
+        }
+
+        self
+    }
+
     pub async fn test_get_validator_payload_attestation_data_pre_gloas(self) -> Self {
         let slot = self.chain.slot().unwrap();
 
@@ -8329,6 +8413,14 @@ async fn get_validator_payload_attestation_data_pre_gloas() {
         .await;
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn payload_attestation_present_after_envelope_publish() {
+    ApiTester::new_with_hard_forks()
+        .await
+        .test_payload_attestation_present_after_envelope_publish()
+        .await;
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn post_beacon_pool_payload_attestations_valid() {
     if !fork_name_from_env().is_some_and(|f| f.gloas_enabled()) {