feat: fetch unknown blocks from peers (#61)

* feat: add BlocksByRootCodec

* refactor: unify both Codecs

* refactor: restructure code in gossipsub and req_resp modules

* refactor: remove the messages module

* refactor: handle response result explicitly

* refactor: make Response fields public

* refactor: move req-resp handlers to new module

* fix: make response include only a single block

* refactor: move handle_outgoing_gossip to gossipsub module

* refactor: add new FetchBlock P2PMessage

* refactor: rename to handle_p2p_message

* feat: request missing blocks

* refactor: move block fetching helper to req_resp::handlers

* refactor: store event loop state in struct

* feat: log req-resp failures

* fix: treat encoded-length as uncompressed size

* fix: split req-resp behaviour

libp2p::request_response expects protocols to be equivalent between them,
and negotiates with the peer which one to use. Since we specified the status
first in the list of protocols, it seems to default to that one when we
send a request.

* refactor: use imports

* Revert "fix: split req-resp behaviour"

This reverts commit b4ffabd.

* fix: implement protocol selection in libp2p fork

* feat: add block fetch request retrying

* refactor: move bookkeeping to fetch_block_from_peer

* refactor: remove BlockchainServer.in_flight_requests

* chore: tweak retries, initial backoff, and multiplier

* fix: avoid calling store::on_block if parent is missing

Author: MegaRedHand

Cargo.lock

@@ -3,6 +3,7 @@ use std::time::{Duration, SystemTime};
 
 use ethlambda_state_transition::is_proposer;
 use ethlambda_storage::Store;
+use ethlambda_types::primitives::H256;
 use ethlambda_types::{
     attestation::{Attestation, AttestationData, SignedAttestation},
     block::{BlockSignatures, BlockWithAttestation, SignedBlockWithAttestation},
@@ -14,21 +15,23 @@ use spawned_concurrency::tasks::{
     CallResponse, CastResponse, GenServer, GenServerHandle, send_after,
 };
 use tokio::sync::mpsc;
-use tracing::{error, info, warn};
+use tracing::{error, info, trace, warn};
 
 use crate::store::StoreError;
 
 pub mod key_manager;
 pub mod metrics;
 pub mod store;
 
-/// Messages sent from the blockchain to the P2P layer for publishing.
+/// Messages sent from the blockchain to the P2P layer.
 #[derive(Clone, Debug)]
-pub enum OutboundGossip {
+pub enum P2PMessage {
     /// Publish an attestation to the gossip network.
     PublishAttestation(SignedAttestation),
     /// Publish a block to the gossip network.
     PublishBlock(SignedBlockWithAttestation),
+    /// Fetch a block by its root hash.
+    FetchBlock(H256),
 }
 
 pub struct BlockChain {
@@ -41,7 +44,7 @@ pub const SECONDS_PER_SLOT: u64 = 4;
 impl BlockChain {
     pub fn spawn(
         store: Store,
-        p2p_tx: mpsc::UnboundedSender<OutboundGossip>,
+        p2p_tx: mpsc::UnboundedSender<P2PMessage>,
         validator_keys: HashMap<u64, ValidatorSecretKey>,
     ) -> BlockChain {
         let genesis_time = store.config().genesis_time;
@@ -50,6 +53,7 @@ impl BlockChain {
             store,
             p2p_tx,
             key_manager,
+            pending_blocks: HashMap::new(),
         }
         .start();
         let time_until_genesis = (SystemTime::UNIX_EPOCH + Duration::from_secs(genesis_time))
@@ -84,8 +88,11 @@ impl BlockChain {
 
 struct BlockChainServer {
     store: Store,
-    p2p_tx: mpsc::UnboundedSender<OutboundGossip>,
+    p2p_tx: mpsc::UnboundedSender<P2PMessage>,
     key_manager: key_manager::KeyManager,
+
+    // Pending blocks waiting for their parent
+    pending_blocks: HashMap<H256, Vec<SignedBlockWithAttestation>>,
 }
 
 impl BlockChainServer {
@@ -173,7 +180,7 @@ impl BlockChainServer {
             // Publish to gossip network
             let Ok(_) = self
                 .p2p_tx
-                .send(OutboundGossip::PublishAttestation(signed_attestation))
+                .send(P2PMessage::PublishAttestation(signed_attestation))
                 .inspect_err(
                     |err| error!(%slot, %validator_id, %err, "Failed to publish attestation"),
                 )
@@ -244,7 +251,7 @@ impl BlockChainServer {
         // Publish to gossip network
         let Ok(()) = self
             .p2p_tx
-            .send(OutboundGossip::PublishBlock(signed_block))
+            .send(P2PMessage::PublishBlock(signed_block))
             .inspect_err(|err| error!(%slot, %validator_id, %err, "Failed to publish block"))
         else {
             return;
@@ -268,8 +275,60 @@ impl BlockChainServer {
 
     fn on_block(&mut self, signed_block: SignedBlockWithAttestation) {
         let slot = signed_block.message.block.slot;
-        if let Err(err) = self.process_block(signed_block) {
-            warn!(%slot, %err, "Failed to process block");
+        let block_root = signed_block.message.block.tree_hash_root();
+        let parent_root = signed_block.message.block.parent_root;
+
+        // Check if parent block exists before attempting to process
+        if !self.store.contains_block(&parent_root) {
+            info!(%slot, %parent_root, %block_root, "Block parent missing, storing as pending");
+
+            // Store block for later processing
+            self.pending_blocks
+                .entry(parent_root)
+                .or_default()
+                .push(signed_block);
+
+            // Request missing parent from network
+            self.request_missing_block(parent_root);
+            return;
+        }
+
+        // Parent exists, proceed with processing
+        match self.process_block(signed_block) {
+            Ok(_) => {
+                info!(%slot, "Block processed successfully");
+
+                // Check if any pending blocks can now be processed
+                self.process_pending_children(block_root);
+            }
+            Err(err) => {
+                warn!(%slot, %err, "Failed to process block");
+            }
+        }
+    }
+
+    fn request_missing_block(&mut self, block_root: H256) {
+        // Send request to P2P layer (deduplication handled by P2P module)
+        if let Err(err) = self.p2p_tx.send(P2PMessage::FetchBlock(block_root)) {
+            error!(%block_root, %err, "Failed to send FetchBlock message to P2P");
+        } else {
+            info!(%block_root, "Requested missing block from network");
+        }
+    }
+
+    fn process_pending_children(&mut self, parent_root: H256) {
+        // Remove and process all blocks that were waiting for this parent
+        if let Some(children) = self.pending_blocks.remove(&parent_root) {
+            info!(%parent_root, num_children=%children.len(),
+                  "Processing pending blocks after parent arrival");
+
+            for child_block in children {
+                let slot = child_block.message.block.slot;
+                trace!(%parent_root, %slot, "Processing pending child block");
+
+                // Process recursively - might unblock more descendants
+                self.on_block(child_block);
+            }
         }
     }
 

crates/blockchain/src/lib.rs

@@ -299,8 +299,9 @@ pub fn on_block(
         return Ok(());
     }
 
-    // Verify parent chain is available
-    // TODO: sync parent chain if parent is missing
+    // Verify parent state is available
+    // Note: Parent block existence is checked by the caller before calling this function.
+    // This check ensures the state has been computed for the parent block.
     let parent_state =
         store
             .get_state(&block.parent_root)

crates/blockchain/src/store.rs

@@ -17,13 +17,18 @@ ethlambda-types.workspace = true
 
 async-trait = "0.1"
 
-libp2p = { version = "0.56", features = ["full"] }
+# Fork with request-response feature for outbound protocol selection
+libp2p = { git = "https://github.com/lambdaclass/rust-libp2p.git", rev = "cd6cc3b1e5db2c5e23e133c2201c23b063fc4895", features = [
+    "full",
+] }
 
 snap = "1.1"
 
 tokio.workspace = true
 tracing.workspace = true
 
+rand = "0.8"
+
 # Required for NodeEnr parsing
 ethrex-p2p = { git = "https://github.com/lambdaclass/ethrex", rev = "1af63a4de7c93eb7413b9b003df1be82e1484c69" }
 ethrex-rlp = { git = "https://github.com/lambdaclass/ethrex", rev = "1af63a4de7c93eb7413b9b003df1be82e1484c69" }

crates/net/p2p/Cargo.toml

@@ -0,0 +1,32 @@
+/// Decompress data using raw snappy format (for gossipsub messages).
+pub fn decompress_message(data: &[u8]) -> snap::Result<Vec<u8>> {
+    let uncompressed_size = snap::raw::decompress_len(data)?;
+    let mut uncompressed_data = vec![0u8; uncompressed_size];
+    snap::raw::Decoder::new().decompress(data, &mut uncompressed_data)?;
+    Ok(uncompressed_data)
+}
+
+/// Compress data using raw snappy format (for gossipsub messages).
+pub fn compress_message(data: &[u8]) -> Vec<u8> {
+    let max_compressed_len = snap::raw::max_compress_len(data.len());
+    let mut compressed = vec![0u8; max_compressed_len];
+    let compressed_len = snap::raw::Encoder::new()
+        .compress(data, &mut compressed)
+        .expect("snappy compression should not fail");
+    compressed.truncate(compressed_len);
+    compressed
+}
+
+#[cfg(test)]
+mod tests {
+    use ethlambda_types::block::SignedBlockWithAttestation;
+    use ssz::Decode;
+
+    #[test]
+    #[ignore = "Test data uses old BlockSignatures field order (proposer_signature, attestation_signatures). Needs regeneration with correct order (attestation_signatures, proposer_signature)."]
+    fn test_decode_block() {
+        // Sample uncompressed block sent by Zeam (commit b153373806aa49f65aadc47c41b68ead4fab7d6e)
+        let block_bytes = include_bytes!("../../test_data/signed_block_with_attestation.ssz");
+        let _block = SignedBlockWithAttestation::from_ssz_bytes(block_bytes).unwrap();
+    }
+}