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

Squashed commit of the following:

commit 5db1b78
Author: Tan Chee Keong <tanck@sigmaprime.io>
Date:   Fri Feb 7 07:54:00 2025 +0800

    assert

commit 617abb0
Author: Tan Chee Keong <tanck@sigmaprime.io>
Date:   Thu Feb 6 23:31:13 2025 +0800

    remove split_slot

commit f97f50b
Author: Tan Chee Keong <tanck@sigmaprime.io>
Date:   Thu Feb 6 23:30:32 2025 +0800

    Update test

commit f791df1
Author: Tan Chee Keong <tanck@sigmaprime.io>
Date:   Thu Feb 6 22:25:05 2025 +0800

    test

commit b2abae9
Author: Tan Chee Keong <tanck@sigmaprime.io>
Date:   Thu Feb 6 15:11:33 2025 +0800

    Test

commit 42fb4d5
Author: Tan Chee Keong <tanck@sigmaprime.io>
Date:   Thu Feb 6 12:29:14 2025 +0800

    Fix test

commit 424d4c1
Author: Tan Chee Keong <tanck@sigmaprime.io>
Date:   Thu Feb 6 09:37:45 2025 +0800

    test

commit 833d4e2
Merge: e1f81ce 2193f6a
Author: chonghe <44791194+chong-he@users.noreply.github.com>
Date:   Thu Feb 6 09:31:11 2025 +0800

    Merge branch 'unstable' into keep-execution-payload

commit e1f81ce
Author: Tan Chee Keong <tanck@sigmaprime.io>
Date:   Thu Feb 6 09:18:02 2025 +0800

    test

commit beefb92
Author: Tan Chee Keong <tanck@sigmaprime.io>
Date:   Fri Jan 17 09:19:31 2025 +0800

    Add test

commit 2e2efa6
Author: Tan Chee Keong <tanck@sigmaprime.io>
Date:   Wed Jan 15 09:35:48 2025 +0800

    remove info

commit 7b5d5de
Author: Tan Chee Keong <tanck@sigmaprime.io>
Date:   Wed Jan 15 09:27:27 2025 +0800

    Even more simplified logging

commit 9492cab
Author: Tan Chee Keong <tanck@sigmaprime.io>
Date:   Mon Jan 13 11:30:26 2025 +0800

    Simplify logging

commit 0a61483
Author: Tan Chee Keong <tanck@sigmaprime.io>
Date:   Mon Jan 13 10:19:48 2025 +0800

    get to prune_paylods correctly

commit b86434a
Merge: c958809 c9747fb
Author: chonghe <44791194+chong-he@users.noreply.github.com>
Date:   Mon Jan 13 09:51:11 2025 +0800

    Merge branch 'unstable' into keep-execution-payload

commit c958809
Merge: 412e8e9 7c414cc
Author: Tan Chee Keong <tanck@sigmaprime.io>
Date:   Thu Jan 9 11:05:42 2025 +0800

    Merge remote-tracking branch 'lion/anchor_slot_pruning' into keep-execution-payload

commit 7c414cc
Author: Lion - dapplion <35266934+dapplion@users.noreply.github.com>
Date:   Thu Jan 9 09:06:45 2025 +0800

    Update beacon_node/store/src/hot_cold_store.rs

    Co-authored-by: Michael Sproul <micsproul@gmail.com>

commit 412e8e9
Author: Tan Chee Keong <tanck@sigmaprime.io>
Date:   Wed Jan 8 14:52:46 2025 +0800

    Add prune_payload in config.rs

commit a9c2f78
Merge: 9d85619 87b72de
Author: Tan Chee Keong <tanck@sigmaprime.io>
Date:   Wed Jan 8 11:40:16 2025 +0800

    Merge remote-tracking branch 'origin/unstable' into keep-execution-payload

commit 9d85619
Author: Tan Chee Keong <tanck@sigmaprime.io>
Date:   Fri Jan 3 09:41:06 2025 +0800

    add log

commit 016e58e
Author: Tan Chee Keong <tanck@sigmaprime.io>
Date:   Fri Jan 3 09:04:19 2025 +0800

    Add prune_payloads false

commit 791395a
Author: dapplion <35266934+dapplion@users.noreply.github.com>
Date:   Sat Dec 28 21:57:40 2024 +0800

    Use oldest_block_slot to break of pruning payloads

Author: michaelsproul

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,