Merge branch 'unstable' into proposer-shuffling-cleanup

Author: michaelsproul

Cargo.lock

@@ -23,6 +23,7 @@ members = [
     "common/compare_fields_derive",
     "common/deposit_contract",
     "common/directory",
+    "common/eip_3076",
     "common/eth2",
     "common/eth2_config",
     "common/eth2_interop_keypairs",
@@ -56,7 +57,6 @@ members = [
     "consensus/int_to_bytes",
     "consensus/merkle_proof",
     "consensus/proto_array",
-    "consensus/safe_arith",
     "consensus/state_processing",
     "consensus/swap_or_not_shuffle",
     "consensus/types",
@@ -136,6 +136,7 @@ directory = { path = "common/directory" }
 dirs = "3"
 discv5 = { version = "0.10", features = ["libp2p"] }
 doppelganger_service = { path = "validator_client/doppelganger_service" }
+eip_3076 = { path = "common/eip_3076" }
 either = "1.9"
 environment = { path = "lighthouse/environment" }
 eth2 = { path = "common/eth2" }
@@ -225,7 +226,7 @@ ring = "0.17"
 rpds = "0.11"
 rusqlite = { version = "0.28", features = ["bundled"] }
 rust_eth_kzg = "0.9"
-safe_arith = { path = "consensus/safe_arith" }
+safe_arith = "0.1"
 sensitive_url = { path = "common/sensitive_url" }
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "beacon_node"
-version = "8.0.0-rc.0"
+version = "8.0.0-rc.1"
 authors = [
     "Paul Hauner <[email protected]>",
     "Age Manning <[email protected]",

beacon_node/Cargo.toml

@@ -3957,7 +3957,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         // See https://github.com/sigp/lighthouse/issues/2028
         let (_, signed_block, block_data) = signed_block.deconstruct();
 
-        match self.get_blobs_or_columns_store_op(block_root, block_data) {
+        match self.get_blobs_or_columns_store_op(block_root, signed_block.slot(), block_data) {
             Ok(Some(blobs_or_columns_store_op)) => {
                 ops.push(blobs_or_columns_store_op);
             }
@@ -7189,6 +7189,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
     pub(crate) fn get_blobs_or_columns_store_op(
         &self,
         block_root: Hash256,
+        block_slot: Slot,
         block_data: AvailableBlockData<T::EthSpec>,
     ) -> Result<Option<StoreOp<'_, T::EthSpec>>, String> {
         match block_data {
@@ -7201,7 +7202,15 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                 );
                 Ok(Some(StoreOp::PutBlobs(block_root, blobs)))
             }
-            AvailableBlockData::DataColumns(data_columns) => {
+            AvailableBlockData::DataColumns(mut data_columns) => {
+                let columns_to_custody = self.custody_columns_for_epoch(Some(
+                    block_slot.epoch(T::EthSpec::slots_per_epoch()),
+                ));
+                // Supernodes need to persist all sampled custody columns
+                if columns_to_custody.len() != self.spec.number_of_custody_groups as usize {
+                    data_columns
+                        .retain(|data_column| columns_to_custody.contains(&data_column.index));
+                }
                 debug!(
                     %block_root,
                     count = data_columns.len(),

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -644,8 +644,17 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
             "Reconstructed columns"
         );
 
+        let columns_to_sample = self
+            .custody_context()
+            .sampling_columns_for_epoch(slot.epoch(T::EthSpec::slots_per_epoch()), &self.spec);
+        let data_columns_to_import: Vec<_> = data_columns_to_publish
+            .iter()
+            .filter(|column| columns_to_sample.contains(&column.index()))
+            .cloned()
+            .collect();
+
         self.availability_cache
-            .put_kzg_verified_data_columns(*block_root, data_columns_to_publish.clone())
+            .put_kzg_verified_data_columns(*block_root, data_columns_to_import)
             .map(|availability| {
                 DataColumnReconstructionResult::Success((
                     availability,
@@ -1082,6 +1091,95 @@ mod test {
         verification_result.expect_err("should have failed to verify blocks");
     }
 
+    #[test]
+    fn should_exclude_reconstructed_columns_not_required_for_sampling() {
+        // SETUP
+        let spec = Arc::new(ForkName::Fulu.make_genesis_spec(E::default_spec()));
+        let mut rng = StdRng::seed_from_u64(0xDEADBEEF0BAD5EEDu64);
+
+        let da_checker = new_da_checker(spec.clone());
+        let custody_context = &da_checker.custody_context;
+        let all_column_indices_ordered =
+            init_custody_context_with_ordered_columns(custody_context, &mut rng, &spec);
+
+        // Set custody requirement to 65 columns (enough to trigger reconstruction)
+        let epoch = Epoch::new(1);
+        custody_context.register_validators(
+            vec![(0, 2_048_000_000_000), (1, 32_000_000_000)], // 64 + 1
+            Slot::new(0),
+            &spec,
+        );
+        let sampling_requirement = custody_context.num_of_data_columns_to_sample(epoch, &spec);
+        assert_eq!(
+            sampling_requirement, 65,
+            "sampling requirement should be 65"
+        );
+
+        let (block, data_columns) = generate_rand_block_and_data_columns::<E>(
+            ForkName::Fulu,
+            NumBlobs::Number(1),
+            &mut rng,
+            &spec,
+        );
+        let block_root = Hash256::random();
+        // Add the block to the DA checker
+        da_checker
+            .availability_cache
+            .put_pre_execution_block(block_root, Arc::new(block), BlockImportSource::Gossip)
+            .expect("should put block");
+
+        // Add 64 columns to the da checker (enough to be able to reconstruct)
+        // Order by all_column_indices_ordered, then take first 64
+        let custody_columns = all_column_indices_ordered
+            .iter()
+            .filter_map(|&col_idx| data_columns.iter().find(|d| d.index == col_idx).cloned())
+            .take(64)
+            .map(|d| {
+                KzgVerifiedCustodyDataColumn::from_asserted_custody(
+                    KzgVerifiedDataColumn::__new_for_testing(d),
+                )
+            })
+            .collect::<Vec<_>>();
+
+        da_checker
+            .availability_cache
+            .put_kzg_verified_data_columns(block_root, custody_columns)
+            .expect("should put custody columns");
+
+        // Try reconstrucing
+        let reconstruction_result = da_checker
+            .reconstruct_data_columns(&block_root)
+            .expect("should reconstruct columns");
+
+        // Reconstruction should succeed
+        let (_availability, reconstructed_columns) = match reconstruction_result {
+            DataColumnReconstructionResult::Success(result) => result,
+            e => {
+                panic!("Expected successful reconstruction {:?}", e);
+            }
+        };
+
+        // Remaining 64 columns should be reconstructed
+        assert_eq!(
+            reconstructed_columns.len(),
+            64,
+            "should reconstruct the remaining 64 columns"
+        );
+
+        // Only the columns required for custody (65) should be imported into the cache
+        let sampling_columns = custody_context.sampling_columns_for_epoch(epoch, &spec);
+        let actual_cached: HashSet<ColumnIndex> = da_checker
+            .cached_data_column_indexes(&block_root)
+            .expect("should have cached data columns")
+            .into_iter()
+            .collect();
+        let expected_sampling_columns = sampling_columns.iter().copied().collect::<HashSet<_>>();
+        assert_eq!(
+            actual_cached, expected_sampling_columns,
+            "should cache only the required custody columns, not all reconstructed columns"
+        );
+    }
+
     fn init_custody_context_with_ordered_columns(
         custody_context: &Arc<CustodyContext<E>>,
         mut rng: &mut StdRng,