Implement publish block and blobs endpoint (WIP)

beacon_node/http_api/src/lib.rs

@@ -60,7 +60,7 @@ use types::{
     ProposerPreparationData, ProposerSlashing, RelativeEpoch, SignedAggregateAndProof,
     SignedBeaconBlock, SignedBlindedBeaconBlock, SignedBlsToExecutionChange,
     SignedContributionAndProof, SignedValidatorRegistrationData, SignedVoluntaryExit, Slot,
-    SyncCommitteeMessage, SyncContributionData,
+    SyncCommitteeMessage, SyncContributionData, SignedBlockContents,
 };
 use version::{
     add_consensus_version_header, execution_optimistic_fork_versioned_response,
@@ -1120,11 +1120,11 @@ pub fn serve<T: BeaconChainTypes>(
         .and(network_tx_filter.clone())
         .and(log_filter.clone())
         .and_then(
-            |block: Arc<SignedBeaconBlock<T::EthSpec>>,
+            |block_contents: SignedBlockContents<T::EthSpec>,
              chain: Arc<BeaconChain<T>>,
              network_tx: UnboundedSender<NetworkMessage<T::EthSpec>>,
              log: Logger| async move {
-                publish_blocks::publish_block(None, block, chain, &network_tx, log)
+                publish_blocks::publish_block(None, block_contents, chain, &network_tx, log)
                     .await
                     .map(|()| warp::reply())
             },

beacon_node/http_api/src/publish_blocks.rs

@@ -10,21 +10,20 @@ use slot_clock::SlotClock;
 use std::sync::Arc;
 use tokio::sync::mpsc::UnboundedSender;
 use tree_hash::TreeHash;
-use types::{
+use types::{AbstractExecPayload, BlindedPayload, EthSpec, ExecPayload, ExecutionBlockHash, FullPayload, Hash256, SignedBeaconBlock, SignedBlockContents};
-    AbstractExecPayload, BlindedPayload, EthSpec, ExecPayload, ExecutionBlockHash, FullPayload,
-    Hash256, SignedBeaconBlock,
-};
 use warp::Rejection;
 
 /// Handles a request from the HTTP API for full blocks.
 pub async fn publish_block<T: BeaconChainTypes>(
     block_root: Option<Hash256>,
-    block: Arc<SignedBeaconBlock<T::EthSpec>>,
+    block_contents: SignedBlockContents<T::EthSpec>,
     chain: Arc<BeaconChain<T>>,
     network_tx: &UnboundedSender<NetworkMessage<T::EthSpec>>,
     log: Logger,
 ) -> Result<(), Rejection> {
     let seen_timestamp = timestamp_now();
+    let (block, _maybe_blobs) = block_contents.deconstruct();
+    let block = Arc::new(block);
 
     //FIXME(sean) have to move this to prior to publishing because it's included in the blobs sidecar message.
     //this may skew metrics
@@ -38,20 +37,8 @@ pub async fn publish_block<T: BeaconChainTypes>(
     // Send the block, regardless of whether or not it is valid. The API
     // specification is very clear that this is the desired behaviour.
     let wrapped_block: BlockWrapper<T::EthSpec> =
-        if matches!(block.as_ref(), &SignedBeaconBlock::Eip4844(_)) {
+        if matches!(block.as_ref(), SignedBeaconBlock::Eip4844(_)) {
-            if let Some(sidecar) = chain.blob_cache.pop(&block_root) {
+            todo!("to be implemented")
-                // TODO: Needs to be adjusted
-                // let block_and_blobs = SignedBeaconBlockAndBlobsSidecar {
-                //     beacon_block: block,
-                //     blobs_sidecar: Arc::new(sidecar),
-                // };
-                unimplemented!("Needs to be adjusted")
-            } else {
-                //FIXME(sean): This should probably return a specific no-blob-cached error code, beacon API coordination required
-                return Err(warp_utils::reject::broadcast_without_import(
-                    "no blob cached for block".into(),
-                ));
-            }
         } else {
             crate::publish_pubsub_message(network_tx, PubsubMessage::BeaconBlock(block.clone()))?;
             block.into()
@@ -180,7 +167,7 @@ pub async fn publish_blinded_block<T: BeaconChainTypes>(
     let full_block = reconstruct_block(chain.clone(), block_root, block, log.clone()).await?;
     publish_block::<T>(
         Some(block_root),
-        Arc::new(full_block),
+        SignedBlockContents::Block(full_block),
         chain,
         network_tx,
         log,

common/eth2/src/lib.rs

@@ -612,7 +612,7 @@ impl BeaconNodeHttpClient {
     /// Returns `Ok(None)` on a 404 error.
     pub async fn post_beacon_blocks<T: EthSpec, Payload: AbstractExecPayload<T>>(
         &self,
-        block: &SignedBeaconBlock<T, Payload>,
+        block_contents: &SignedBlockContents<T, Payload>,
     ) -> Result<(), Error> {
         let mut path = self.eth_path(V1)?;
 
@@ -621,7 +621,7 @@ impl BeaconNodeHttpClient {
             .push("beacon")
             .push("blocks");
 
-        self.post_with_timeout(path, block, self.timeouts.proposal)
+        self.post_with_timeout(path, block_contents, self.timeouts.proposal)
             .await?;
 
         Ok(())

consensus/types/src/lib.rs

@@ -106,6 +106,7 @@ pub mod block_contents;
 pub mod signed_blob;
 pub mod signed_block_and_blobs;
 pub mod transaction;
+pub mod signed_block_contents;
 
 use ethereum_types::{H160, H256};
 
@@ -187,8 +188,8 @@ pub use crate::signed_beacon_block::{
 };
 pub use crate::signed_beacon_block_header::SignedBeaconBlockHeader;
 pub use crate::signed_blob::*;
-pub use crate::signed_block_and_blobs::SignedBeaconBlockAndBlobsSidecar;
+pub use crate::signed_block_and_blobs::{SignedBeaconBlockAndBlobsSidecar, SignedBeaconBlockAndBlobsSidecarDecode, SignedBeaconBlockAndBlobSidecars};
-pub use crate::signed_block_and_blobs::SignedBeaconBlockAndBlobsSidecarDecode;
+pub use crate::signed_block_contents::SignedBlockContents;
 pub use crate::signed_bls_to_execution_change::SignedBlsToExecutionChange;
 pub use crate::signed_contribution_and_proof::SignedContributionAndProof;
 pub use crate::signed_voluntary_exit::SignedVoluntaryExit;

consensus/types/src/signed_blob.rs

@@ -3,6 +3,7 @@ use serde_derive::{Deserialize, Serialize};
 use ssz_derive::{Decode, Encode};
 use test_random_derive::TestRandom;
 use tree_hash_derive::TreeHash;
+use derivative::Derivative;
 
 #[derive(
     Debug,
@@ -14,10 +15,12 @@ use tree_hash_derive::TreeHash;
     Decode,
     TestRandom,
     TreeHash,
+    Derivative,
     arbitrary::Arbitrary,
 )]
 #[serde(bound = "T: EthSpec")]
 #[arbitrary(bound = "T: EthSpec")]
+#[derivative(Hash(bound = "T: EthSpec"))]
 pub struct SignedBlobSidecar<T: EthSpec> {
     pub message: BlobSidecar<T>,
     pub signature: Signature,

consensus/types/src/signed_block_and_blobs.rs

@@ -1,9 +1,10 @@
-use crate::{BlobsSidecar, EthSpec, SignedBeaconBlock, SignedBeaconBlockEip4844};
+use crate::{AbstractExecPayload, BlobsSidecar, EthSpec, SignedBeaconBlock, SignedBeaconBlockEip4844, SignedBlobSidecar};
 use derivative::Derivative;
 use serde_derive::{Deserialize, Serialize};
 use ssz::{Decode, DecodeError};
 use ssz_derive::{Decode, Encode};
 use std::sync::Arc;
+use ssz_types::VariableList;
 use tree_hash_derive::TreeHash;
 
 #[derive(Debug, Clone, Serialize, Deserialize, Encode, Decode, TreeHash, PartialEq)]
@@ -13,6 +14,7 @@ pub struct SignedBeaconBlockAndBlobsSidecarDecode<T: EthSpec> {
     pub blobs_sidecar: BlobsSidecar<T>,
 }
 
+// TODO: will be removed once we decouple blobs in Gossip
 #[derive(Debug, Clone, Serialize, Deserialize, Encode, TreeHash, Derivative)]
 #[derivative(PartialEq, Hash(bound = "T: EthSpec"))]
 pub struct SignedBeaconBlockAndBlobsSidecar<T: EthSpec> {
@@ -32,3 +34,11 @@ impl<T: EthSpec> SignedBeaconBlockAndBlobsSidecar<T> {
         })
     }
 }
+
+#[derive(Debug, Clone, Serialize, Deserialize, Encode, TreeHash, Derivative)]
+#[derivative(PartialEq, Hash(bound = "T: EthSpec"))]
+#[serde(bound = "T: EthSpec")]
+pub struct SignedBeaconBlockAndBlobSidecars<T: EthSpec, Payload: AbstractExecPayload<T>> {
+    pub signed_block: SignedBeaconBlock<T, Payload>,
+    pub signed_blob_sidecars: VariableList<SignedBlobSidecar<T>, <T as EthSpec>::MaxBlobsPerBlock>,
+}

consensus/types/src/signed_block_contents.rs

@@ -0,0 +1,40 @@
+use crate::{AbstractExecPayload, EthSpec, FullPayload, SignedBeaconBlock, SignedBlobSidecar};
+use crate::signed_block_and_blobs::SignedBeaconBlockAndBlobSidecars;
+use derivative::Derivative;
+use serde_derive::{Deserialize, Serialize};
+use ssz_types::VariableList;
+
+/// A wrapper over a [`SignedBeaconBlock`] or a [`SignedBeaconBlockAndBlobSidecars`].
+#[derive(Clone, Debug, Derivative, Serialize, Deserialize)]
+#[derivative(PartialEq, Hash(bound = "T: EthSpec"))]
+#[serde(untagged)]
+#[serde(bound = "T: EthSpec")]
+pub enum SignedBlockContents<T: EthSpec, Payload: AbstractExecPayload<T> = FullPayload<T>> {
+    BlockAndBlobSidecars(SignedBeaconBlockAndBlobSidecars<T, Payload>),
+    Block(SignedBeaconBlock<T, Payload>),
+}
+
+impl<T: EthSpec, Payload: AbstractExecPayload<T>> SignedBlockContents<T, Payload> {
+    pub fn signed_block(&self) -> &SignedBeaconBlock<T, Payload> {
+        match self {
+            SignedBlockContents::BlockAndBlobSidecars(block_and_sidecars) => &block_and_sidecars.signed_block,
+            SignedBlockContents::Block(block) => block,
+        }
+    }
+
+    pub fn deconstruct(self) -> (SignedBeaconBlock<T, Payload>, Option<VariableList<SignedBlobSidecar<T>, <T as EthSpec>::MaxBlobsPerBlock>>) {
+        match self {
+            SignedBlockContents::BlockAndBlobSidecars(block_and_sidecars) => (
+                block_and_sidecars.signed_block,
+                Some(block_and_sidecars.signed_blob_sidecars),
+            ),
+            SignedBlockContents::Block(block) => (block, None),
+        }
+    }
+}
+
+impl<T: EthSpec, Payload: AbstractExecPayload<T>> From<SignedBeaconBlock<T, Payload>> for SignedBlockContents<T, Payload> {
+    fn from(block: SignedBeaconBlock<T, Payload>) -> Self {
+        SignedBlockContents::Block(block)
+    }
+}

validator_client/src/block_service.rs

@@ -14,10 +14,7 @@ use std::sync::Arc;
 use std::time::Duration;
 use tokio::sync::mpsc;
 use tokio::time::sleep;
-use types::{
+use types::{AbstractExecPayload, BeaconBlock, BlindedPayload, BlockType, EthSpec, FullPayload, Graffiti, PublicKeyBytes, SignedBlockContents, Slot};
-    AbstractExecPayload, BeaconBlock, BlindedPayload, BlockType, EthSpec, FullPayload, Graffiti,
-    PublicKeyBytes, Slot,
-};
 
 #[derive(Debug)]
 pub enum BlockError {
@@ -410,11 +407,12 @@ impl<T: SlotClock + 'static, E: EthSpec> BlockService<T, E> {
             .await?;
 
         let signing_timer = metrics::start_timer(&metrics::BLOCK_SIGNING_TIMES);
-        let signed_block = self_ref
+        let signed_block_contents: SignedBlockContents<E, Payload> = self_ref
             .validator_store
             .sign_block::<Payload>(*validator_pubkey_ref, block, current_slot)
             .await
-            .map_err(|e| BlockError::Recoverable(format!("Unable to sign block: {:?}", e)))?;
+            .map_err(|e| BlockError::Recoverable(format!("Unable to sign block: {:?}", e)))?
+            .into();
         let signing_time_ms =
             Duration::from_secs_f64(signing_timer.map_or(0.0, |t| t.stop_and_record())).as_millis();
 
@@ -438,7 +436,7 @@ impl<T: SlotClock + 'static, E: EthSpec> BlockService<T, E> {
                                 &[metrics::BEACON_BLOCK_HTTP_POST],
                             );
                             beacon_node
-                                .post_beacon_blocks(&signed_block)
+                                .post_beacon_blocks(&signed_block_contents)
                                 .await
                                 .map_err(|e| {
                                     BlockError::Irrecoverable(format!(
@@ -453,7 +451,8 @@ impl<T: SlotClock + 'static, E: EthSpec> BlockService<T, E> {
                                 &[metrics::BLINDED_BEACON_BLOCK_HTTP_POST],
                             );
                             beacon_node
-                                .post_beacon_blinded_blocks(&signed_block)
+                                // TODO: need to be adjusted for blobs
+                                .post_beacon_blinded_blocks(&signed_block_contents.signed_block())
                                 .await
                                 .map_err(|e| {
                                     BlockError::Irrecoverable(format!(
@@ -472,10 +471,10 @@ impl<T: SlotClock + 'static, E: EthSpec> BlockService<T, E> {
             log,
             "Successfully published block";
             "block_type" => ?Payload::block_type(),
-            "deposits" => signed_block.message().body().deposits().len(),
+            "deposits" => signed_block_contents.signed_block().message().body().deposits().len(),
-            "attestations" => signed_block.message().body().attestations().len(),
+            "attestations" => signed_block_contents.signed_block().message().body().attestations().len(),
             "graffiti" => ?graffiti.map(|g| g.as_utf8_lossy()),
-            "slot" => signed_block.slot().as_u64(),
+            "slot" => signed_block_contents.signed_block().slot().as_u64(),
         );
 
         Ok(())