merge ptc branch chagnes

beacon_node/http_api/src/beacon/pool.rs

@@ -1,5 +1,8 @@
 use crate::task_spawner::{Priority, TaskSpawner};
-use crate::utils::{NetworkTxFilter, OptionalConsensusVersionHeaderFilter, ResponseFilter};
+use crate::utils::{
+    ChainFilter, EthV1Filter, NetworkTxFilter, OptionalConsensusVersionHeaderFilter,
+    ResponseFilter, TaskSpawnerFilter,
+};
 use crate::version::{
     ResponseIncludesVersion, V1, V2, add_consensus_version_header, beacon_response,
     unsupported_version_rejection,
@@ -7,11 +10,13 @@ use crate::version::{
 use crate::{sync_committees, utils};
 use beacon_chain::observed_operations::ObservationOutcome;
 use beacon_chain::{BeaconChain, BeaconChainTypes};
+use bytes::Bytes;
 use eth2::types::{AttestationPoolQuery, EndpointVersion, Failure, GenericResponse};
 use lighthouse_network::PubsubMessage;
 use network::NetworkMessage;
 use operation_pool::ReceivedPreCapella;
 use slot_clock::SlotClock;
+use ssz::{Decode, Encode};
 use std::collections::HashSet;
 use std::sync::Arc;
 use tokio::sync::mpsc::UnboundedSender;
@@ -522,9 +527,10 @@ pub fn post_beacon_pool_attestations_v2<T: BeaconChainTypes>(
         .boxed()
 }
 
-/// POST beacon/pool/payload_attestations
+/// POST beacon/pool/payload_attestations (JSON)
 pub fn post_beacon_pool_payload_attestations<T: BeaconChainTypes>(
     network_tx_filter: &NetworkTxFilter<T>,
+    optional_consensus_version_header_filter: OptionalConsensusVersionHeaderFilter,
     beacon_pool_path: &BeaconPoolPathFilter<T>,
 ) -> ResponseFilter {
     beacon_pool_path
@@ -532,23 +538,80 @@ pub fn post_beacon_pool_payload_attestations<T: BeaconChainTypes>(
         .and(warp::path("payload_attestations"))
         .and(warp::path::end())
         .and(warp_utils::json::json())
+        .and(optional_consensus_version_header_filter)
         .and(network_tx_filter.clone())
         .then(
             |task_spawner: TaskSpawner<T::EthSpec>,
              _chain: Arc<BeaconChain<T>>,
              messages: Vec<PayloadAttestationMessage>,
+             _fork_name: Option<ForkName>,
              network_tx: UnboundedSender<NetworkMessage<T::EthSpec>>| {
                 task_spawner.blocking_json_task(Priority::P0, move || {
-                    // TODO(gloas): add proper verification once payload_attestation_verification is implemented
-                    for message in messages {
-                        utils::publish_pubsub_message(
-                            &network_tx,
-                            PubsubMessage::PayloadAttestation(Box::new(message)),
-                        )?;
-                    }
-                    Ok(())
+                    publish_payload_attestation_messages(&network_tx, messages)
                 })
             },
         )
         .boxed()
 }
+
+/// POST beacon/pool/payload_attestations (SSZ)
+pub fn post_beacon_pool_payload_attestations_ssz<T: BeaconChainTypes>(
+    eth_v1: EthV1Filter,
+    task_spawner_filter: TaskSpawnerFilter<T>,
+    chain_filter: ChainFilter<T>,
+    network_tx_filter: NetworkTxFilter<T>,
+) -> ResponseFilter {
+    eth_v1
+        .and(warp::path("beacon"))
+        .and(warp::path("pool"))
+        .and(warp::path("payload_attestations"))
+        .and(warp::path::end())
+        .and(warp::body::bytes())
+        .and(task_spawner_filter)
+        .and(chain_filter)
+        .and(network_tx_filter)
+        .then(
+            |body_bytes: Bytes,
+             task_spawner: TaskSpawner<T::EthSpec>,
+             _chain: Arc<BeaconChain<T>>,
+             network_tx: UnboundedSender<NetworkMessage<T::EthSpec>>| {
+                task_spawner.blocking_json_task(Priority::P0, move || {
+                    let item_len = <PayloadAttestationMessage as Encode>::ssz_fixed_len();
+                    if body_bytes.len() % item_len != 0 {
+                        return Err(warp_utils::reject::custom_bad_request(format!(
+                            "SSZ body length {} is not a multiple of PayloadAttestationMessage size {}",
+                            body_bytes.len(),
+                            item_len,
+                        )));
+                    }
+                    let messages: Vec<PayloadAttestationMessage> = body_bytes
+                        .chunks(item_len)
+                        .map(|chunk| {
+                            PayloadAttestationMessage::from_ssz_bytes(chunk).map_err(|e| {
+                                warp_utils::reject::custom_bad_request(format!(
+                                    "invalid SSZ: {e:?}"
+                                ))
+                            })
+                        })
+                        .collect::<Result<_, _>>()?;
+
+                    publish_payload_attestation_messages(&network_tx, messages)
+                })
+            },
+        )
+        .boxed()
+}
+
+fn publish_payload_attestation_messages<E: types::EthSpec>(
+    network_tx: &UnboundedSender<NetworkMessage<E>>,
+    messages: Vec<PayloadAttestationMessage>,
+) -> Result<(), warp::Rejection> {
+    // TODO(gloas): add proper gossip verification and store in ptc op pool.
+    for message in messages {
+        utils::publish_pubsub_message(
+            network_tx,
+            PubsubMessage::PayloadAttestation(Box::new(message)),
+        )?;
+    }
+    Ok(())
+}

beacon_node/http_api/src/lib.rs

@@ -1454,7 +1454,7 @@ pub fn serve<T: BeaconChainTypes>(
 
     let post_beacon_pool_attestations_v2 = post_beacon_pool_attestations_v2(
         &network_tx_filter,
-        optional_consensus_version_header_filter,
+        optional_consensus_version_header_filter.clone(),
         &beacon_pool_path_v2,
     );
 
@@ -1488,8 +1488,19 @@ pub fn serve<T: BeaconChainTypes>(
         post_beacon_pool_sync_committees(&network_tx_filter, &beacon_pool_path);
 
     // POST beacon/pool/payload_attestations
-    let post_beacon_pool_payload_attestations =
-        post_beacon_pool_payload_attestations(&network_tx_filter, &beacon_pool_path);
+    let post_beacon_pool_payload_attestations = post_beacon_pool_payload_attestations(
+        &network_tx_filter,
+        optional_consensus_version_header_filter,
+        &beacon_pool_path,
+    );
+
+    // POST beacon/pool/payload_attestations (SSZ)
+    let post_beacon_pool_payload_attestations_ssz = post_beacon_pool_payload_attestations_ssz(
+        eth_v1.clone(),
+        task_spawner_filter.clone(),
+        chain_filter.clone(),
+        network_tx_filter.clone(),
+    );
 
     // GET beacon/pool/bls_to_execution_changes
     let get_beacon_pool_bls_to_execution_changes =
@@ -3404,7 +3415,8 @@ pub fn serve<T: BeaconChainTypes>(
                             .uor(post_beacon_blocks_v2_ssz)
                             .uor(post_beacon_blinded_blocks_ssz)
                             .uor(post_beacon_blinded_blocks_v2_ssz)
-                            .uor(post_beacon_execution_payload_envelope_ssz),
+                            .uor(post_beacon_execution_payload_envelope_ssz)
+                            .uor(post_beacon_pool_payload_attestations_ssz),
                     )
                     .uor(post_beacon_blocks)
                     .uor(post_beacon_blinded_blocks)

beacon_node/http_api/tests/tests.rs

@@ -2793,6 +2793,62 @@ impl ApiTester {
         self
     }
 
+    pub async fn test_post_beacon_pool_payload_attestations_valid(mut self) -> Self {
+        let slot = self.chain.slot().unwrap();
+        let head_root = self.chain.head_beacon_block_root();
+
+        let message = PayloadAttestationMessage {
+            validator_index: 0,
+            data: PayloadAttestationData {
+                beacon_block_root: head_root,
+                slot,
+                payload_present: true,
+                blob_data_available: true,
+            },
+            signature: Signature::empty(),
+        };
+
+        self.client
+            .post_beacon_pool_payload_attestations(&[message])
+            .await
+            .unwrap();
+
+        assert!(
+            self.network_rx.network_recv.recv().await.is_some(),
+            "valid payload attestation should be sent to network"
+        );
+
+        self
+    }
+
+    pub async fn test_post_beacon_pool_payload_attestations_valid_ssz(mut self) -> Self {
+        let slot = self.chain.slot().unwrap();
+        let head_root = self.chain.head_beacon_block_root();
+
+        let message = PayloadAttestationMessage {
+            validator_index: 0,
+            data: PayloadAttestationData {
+                beacon_block_root: head_root,
+                slot,
+                payload_present: true,
+                blob_data_available: true,
+            },
+            signature: Signature::empty(),
+        };
+
+        self.client
+            .post_beacon_pool_payload_attestations_ssz(&[message])
+            .await
+            .unwrap();
+
+        assert!(
+            self.network_rx.network_recv.recv().await.is_some(),
+            "valid payload attestation (SSZ) should be sent to network"
+        );
+
+        self
+    }
+
     pub async fn test_get_config_fork_schedule(self) -> Self {
         let result = self.client.get_config_fork_schedule().await.unwrap().data;
 
@@ -8246,6 +8302,19 @@ async fn get_validator_payload_attestation_data_pre_gloas() {
         .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()) {
+        return;
+    }
+    ApiTester::new()
+        .await
+        .test_post_beacon_pool_payload_attestations_valid()
+        .await
+        .test_post_beacon_pool_payload_attestations_valid_ssz()
+        .await;
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn get_validator_aggregate_attestation_v1() {
     ApiTester::new()