Keep execution payload during historical backfill when prune-payloads set to false (#6766)

- #6510


  - Keep execution payload during historical backfill when `--prune-payloads false` is set
- Add a field in the historical backfill debug log to indicate if execution payload is kept
- Add a test to check historical blocks has execution payload when `--prune-payloads false is set
- Very minor typo correction that I notice when working on this

beacon_node/beacon_chain/src/historical_blocks.rs

@@ -130,10 +130,20 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                 });
             }
 
-            let blinded_block = block.clone_as_blinded();
-            // Store block in the hot database without payload.
-            self.store
-                .blinded_block_as_kv_store_ops(&block_root, &blinded_block, &mut hot_batch);
+            if !self.store.get_config().prune_payloads {
+                // If prune-payloads is set to false, store the block which includes the execution payload
+                self.store
+                    .block_as_kv_store_ops(&block_root, (*block).clone(), &mut hot_batch)?;
+            } else {
+                let blinded_block = block.clone_as_blinded();
+                // Store block in the hot database without payload.
+                self.store.blinded_block_as_kv_store_ops(
+                    &block_root,
+                    &blinded_block,
+                    &mut hot_batch,
+                );
+            }
+
             // Store the blobs too
             if let Some(blobs) = maybe_blobs {
                 new_oldest_blob_slot = Some(block.slot());

beacon_node/beacon_chain/tests/store_tests.rs

@@ -47,7 +47,11 @@ type E = MinimalEthSpec;
 type TestHarness = BeaconChainHarness<DiskHarnessType<E>>;
 
 fn get_store(db_path: &TempDir) -> Arc<HotColdDB<E, BeaconNodeBackend<E>, BeaconNodeBackend<E>>> {
-    get_store_generic(db_path, StoreConfig::default(), test_spec::<E>())
+    let store_config = StoreConfig {
+        prune_payloads: false,
+        ..StoreConfig::default()
+    };
+    get_store_generic(db_path, store_config, test_spec::<E>())
 }
 
 fn get_store_generic(
@@ -2571,6 +2575,15 @@ async fn weak_subjectivity_sync_test(slots: Vec<Slot>, checkpoint_slot: Slot) {
         if block_root != prev_block_root {
             assert_eq!(block.slot(), slot);
         }
+
+        // Prune_payloads is set to false in the default config, so the payload should exist
+        if block.message().execution_payload().is_ok() {
+            assert!(beacon_chain
+                .store
+                .execution_payload_exists(&block_root)
+                .unwrap(),);
+        }
+
         prev_block_root = block_root;
     }
 
@@ -3558,7 +3571,6 @@ fn check_split_slot(
 /// Check that all the states in a chain dump have the correct tree hash.
 fn check_chain_dump(harness: &TestHarness, expected_len: u64) {
     let mut chain_dump = harness.chain.chain_dump().unwrap();
-    let split_slot = harness.chain.store.get_split_slot();
 
     assert_eq!(chain_dump.len() as u64, expected_len);
 
@@ -3585,13 +3597,12 @@ fn check_chain_dump(harness: &TestHarness, expected_len: u64) {
 
         // Check presence of execution payload on disk.
         if harness.chain.spec.bellatrix_fork_epoch.is_some() {
-            assert_eq!(
+            assert!(
                 harness
                     .chain
                     .store
                     .execution_payload_exists(&checkpoint.beacon_block_root)
                     .unwrap(),
-                checkpoint.beacon_block.slot() >= split_slot,
                 "incorrect payload storage for block at slot {}: {:?}",
                 checkpoint.beacon_block.slot(),
                 checkpoint.beacon_block_root,

beacon_node/network/src/network_beacon_processor/sync_methods.rs

@@ -483,6 +483,7 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
                         debug!(self.log, "Backfill batch processed";
                             "batch_epoch" => epoch,
                             "first_block_slot" => start_slot,
+                            "keep_execution_payload" => !self.chain.store.get_config().prune_payloads,
                             "last_block_slot" => end_slot,
                             "processed_blocks" => sent_blocks,
                             "processed_blobs" => n_blobs,

beacon_node/store/src/hot_cold_store.rs

@@ -516,7 +516,7 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> HotColdDB<E, Hot, Cold>
             .ok_or(Error::AddPayloadLogicError)
     }
 
-    /// Prepare a signed beacon block for storage in the datbase *without* its payload.
+    /// Prepare a signed beacon block for storage in the database *without* its payload.
     pub fn blinded_block_as_kv_store_ops(
         &self,
         key: &Hash256,