fix(l1): handle missing payloads in engine_getPayloadBodiesByRange (#5408)

**Motivation**

When we added RPC metrics for success/error rates in #5335, we found
that we were returning "Internal error" in the
`engine_getPayloadBodiesByRange` method. This was related to changes
done in #2430, which turned missing payloads into a DB error. The spec
says we should return `null`s for any missing payloads, so this deviates
from the spec.

**Description**

This PR fixes the spec deviation by changing the batch-GET methods from
the Store API to return `None` on missing values instead of failing.

Closes #5403

---------

Co-authored-by: Rodrigo Oliveri <[email protected]>

Author: MegaRedHand

crates/networking/p2p/sync.rs

@@ -383,7 +383,12 @@ impl Syncer {
             let final_batch = end_block_number == start + batch_size as u64;
             // Retrieve batch from DB
             if !single_batch {
-                headers = store.read_fullsync_batch(start, batch_size as u64).await?;
+                headers = store
+                    .read_fullsync_batch(start, batch_size as u64)
+                    .await?
+                    .into_iter()
+                    .map(|opt| opt.ok_or(SyncError::MissingFullsyncBatch))
+                    .collect::<Result<Vec<_>, SyncError>>()?;
             }
             let mut blocks = Vec::new();
             // Request block bodies
@@ -1162,6 +1167,8 @@ pub enum SyncError {
     BytecodeFileError,
     #[error("Error in Peer Table: {0}")]
     PeerTableError(#[from] PeerTableError),
+    #[error("Missing fullsync batch")]
+    MissingFullsyncBatch,
 }
 
 impl SyncError {
@@ -1185,7 +1192,8 @@ impl SyncError {
             | SyncError::RocksDBError(_)
             | SyncError::BytecodeFileError
             | SyncError::NoLatestCanonical
-            | SyncError::PeerTableError(_) => false,
+            | SyncError::PeerTableError(_)
+            | SyncError::MissingFullsyncBatch => false,
             SyncError::Chain(_)
             | SyncError::Store(_)
             | SyncError::Send(_)

crates/networking/rpc/engine/payload.rs

@@ -470,9 +470,11 @@ impl RpcHandler for GetPayloadBodiesByRangeV1Request {
             return Err(RpcErr::TooLargeRequest);
         }
         let latest_block_number = context.storage.get_latest_block_number().await?;
+        // NOTE: we truncate the range because the spec says we "MUST NOT return trailing
+        // null values if the request extends past the current latest known block"
         let last = latest_block_number.min(self.start + self.count - 1);
         let bodies = context.storage.get_block_bodies(self.start, last).await?;
-        build_payload_body_response(bodies.into_iter().map(Some).collect())
+        build_payload_body_response(bodies)
     }
 }
 

crates/storage/api.rs

@@ -58,13 +58,13 @@ pub trait StoreEngine: Debug + Send + Sync + RefUnwindSafe {
         &self,
         from: BlockNumber,
         to: BlockNumber,
-    ) -> Result<Vec<BlockBody>, StoreError>;
+    ) -> Result<Vec<Option<BlockBody>>, StoreError>;
 
     /// Obtain block bodies from a list of hashes
     async fn get_block_bodies_by_hash(
         &self,
         hashes: Vec<BlockHash>,
-    ) -> Result<Vec<BlockBody>, StoreError>;
+    ) -> Result<Vec<Option<BlockBody>>, StoreError>;
 
     /// Obtain any block body using the hash
     async fn get_block_body_by_hash(
@@ -372,7 +372,7 @@ pub trait StoreEngine: Debug + Send + Sync + RefUnwindSafe {
         &self,
         start: BlockNumber,
         limit: u64,
-    ) -> Result<Vec<BlockHeader>, StoreError>;
+    ) -> Result<Vec<Option<BlockHeader>>, StoreError>;
 
     /// Clear all headers downloaded during fullsync
     async fn clear_fullsync_headers(&self) -> Result<(), StoreError>;

crates/storage/store.rs

@@ -239,14 +239,14 @@ impl Store {
         &self,
         from: BlockNumber,
         to: BlockNumber,
-    ) -> Result<Vec<BlockBody>, StoreError> {
+    ) -> Result<Vec<Option<BlockBody>>, StoreError> {
         self.engine.get_block_bodies(from, to).await
     }
 
     pub async fn get_block_bodies_by_hash(
         &self,
         hashes: Vec<BlockHash>,
-    ) -> Result<Vec<BlockBody>, StoreError> {
+    ) -> Result<Vec<Option<BlockBody>>, StoreError> {
         self.engine.get_block_bodies_by_hash(hashes).await
     }
 
@@ -1349,7 +1349,7 @@ impl Store {
         &self,
         start: BlockNumber,
         limit: u64,
-    ) -> Result<Vec<BlockHeader>, StoreError> {
+    ) -> Result<Vec<Option<BlockHeader>>, StoreError> {
         self.engine.read_fullsync_batch(start, limit).await
     }
 

crates/storage/store_db/in_memory.rs

@@ -212,31 +212,27 @@ impl StoreEngine for Store {
         &self,
         from: BlockNumber,
         to: BlockNumber,
-    ) -> Result<Vec<BlockBody>, StoreError> {
+    ) -> Result<Vec<Option<BlockBody>>, StoreError> {
         let store = self.inner()?;
         let mut res = Vec::new();
         for block_number in from..=to {
-            if let Some(block) = store
+            let body_opt = store
                 .canonical_hashes
                 .get(&block_number)
-                .and_then(|hash| store.bodies.get(hash))
-            {
-                res.push(block.clone())
-            }
+                .and_then(|hash| store.bodies.get(hash));
+            res.push(body_opt.cloned());
         }
         Ok(res)
     }
 
     async fn get_block_bodies_by_hash(
         &self,
         hashes: Vec<BlockHash>,
-    ) -> Result<Vec<BlockBody>, StoreError> {
+    ) -> Result<Vec<Option<BlockBody>>, StoreError> {
         let store = self.inner()?;
         let mut res = Vec::new();
         for hash in hashes {
-            if let Some(block) = store.bodies.get(&hash).cloned() {
-                res.push(block);
-            }
+            res.push(store.bodies.get(&hash).cloned());
         }
         Ok(res)
     }
@@ -708,19 +704,12 @@ impl StoreEngine for Store {
         &self,
         start: BlockNumber,
         limit: u64,
-    ) -> Result<Vec<BlockHeader>, StoreError> {
+    ) -> Result<Vec<Option<BlockHeader>>, StoreError> {
         let store = self.inner()?;
-        (start..start + limit)
-            .map(|ref n| {
-                store
-                    .fullsync_headers
-                    .get(n)
-                    .cloned()
-                    .ok_or(StoreError::Custom(format!(
-                        "Missing fullsync header for block {n}"
-                    )))
-            })
-            .collect::<Result<Vec<_>, _>>()
+        let batch = (start..start + limit)
+            .map(|ref n| store.fullsync_headers.get(n).cloned())
+            .collect();
+        Ok(batch)
     }
 
     async fn clear_fullsync_headers(&self) -> Result<(), StoreError> {

crates/storage/store_db/rocksdb.rs

@@ -603,7 +603,7 @@ impl Store {
         cf_name: &str,
         keys: Vec<K>,
         deserialize_fn: F,
-    ) -> Result<Vec<V>, StoreError>
+    ) -> Result<Vec<Option<V>>, StoreError>
     where
         K: AsRef<[u8]> + Send + 'static,
         V: Send + 'static,
@@ -623,10 +623,10 @@ impl Store {
                 match db.get_cf(&cf, key)? {
                     Some(bytes) => {
                         let value = deserialize_fn(bytes)?;
-                        results.push(value);
+                        results.push(Some(value));
                     }
                     None => {
-                        return Err(StoreError::Custom("Key not found in bulk read".to_string()));
+                        results.push(None);
                     }
                 }
             }
@@ -1154,7 +1154,7 @@ impl StoreEngine for Store {
         &self,
         from: BlockNumber,
         to: BlockNumber,
-    ) -> Result<Vec<BlockBody>, StoreError> {
+    ) -> Result<Vec<Option<BlockBody>>, StoreError> {
         let numbers: Vec<BlockNumber> = (from..=to).collect();
         let number_keys: Vec<Vec<u8>> = numbers.iter().map(|n| n.to_le_bytes().to_vec()).collect();
 
@@ -1168,24 +1168,41 @@ impl StoreEngine for Store {
 
         let hash_keys: Vec<Vec<u8>> = hashes
             .iter()
-            .map(|hash| BlockHashRLP::from(*hash).bytes().clone())
+            .flat_map(|hash_opt| hash_opt.map(|h| BlockHashRLP::from(h).bytes().clone()))
             .collect();
 
-        let bodies = self
+        let mut bodies = self
             .read_bulk_async(CF_BODIES, hash_keys, |bytes| {
                 BlockBodyRLP::from_bytes(bytes)
                     .to()
                     .map_err(StoreError::from)
             })
             .await?;
 
+        let mut i = 0;
+
+        // Fill in with None for missing bodies
+        let bodies = hashes
+            .into_iter()
+            .map(|opt| {
+                opt.and_then(|_| {
+                    let body_ref = bodies
+                        .get_mut(i)
+                        .expect("bodies length is equal to number of Somes in hashes");
+                    let body = std::mem::take(body_ref);
+                    i += 1;
+                    body
+                })
+            })
+            .collect();
+
         Ok(bodies)
     }
 
     async fn get_block_bodies_by_hash(
         &self,
         hashes: Vec<BlockHash>,
-    ) -> Result<Vec<BlockBody>, StoreError> {
+    ) -> Result<Vec<Option<BlockBody>>, StoreError> {
         let hash_keys: Vec<Vec<u8>> = hashes
             .iter()
             .map(|hash| BlockHashRLP::from(*hash).bytes().clone())
@@ -2052,7 +2069,7 @@ impl StoreEngine for Store {
         &self,
         start: BlockNumber,
         limit: u64,
-    ) -> Result<Vec<BlockHeader>, StoreError> {
+    ) -> Result<Vec<Option<BlockHeader>>, StoreError> {
         self.read_bulk_async(
             CF_FULLSYNC_HEADERS,
             (start..start + limit).map(|n| n.to_le_bytes()).collect(),