Update SAMPLES_PER_SLOT to be number of custody groups instead of data columns (#7683)

Update `SAMPLES_PER_SLOT` to be number of custody groups instead of data columns. This should have no impact on the current implementation as config currently maintains a `group:subnet:column` ratio of `1:1:1`.  **In short, this PR doesn't change anything for Fusaka, but ensures compliance with the spec and potential future changes.**

I've added separate methods to compute sampling columns and custody groups for clarity: `spec.sampling_size_columns` and `spec.sampling_size_custod_groups`

See the clarifications in this PR for more details: https://github.com/ethereum/consensus-specs/pull/4251

beacon_node/beacon_chain/src/data_availability_checker/overflow_lru_cache.rs

@@ -481,7 +481,8 @@ impl<T: BeaconChainTypes> DataAvailabilityCheckerInner<T> {
 
         if let Some(available_block) = pending_components.make_available(
             &self.spec,
-            self.custody_context.sampling_size(Some(epoch), &self.spec),
+            self.custody_context
+                .num_of_data_columns_to_sample(Some(epoch), &self.spec),
             |block| self.state_cache.recover_pending_executed_block(block),
         )? {
             // We keep the pending components in the availability cache during block import (#5845).
@@ -526,7 +527,9 @@ impl<T: BeaconChainTypes> DataAvailabilityCheckerInner<T> {
         // Merge in the data columns.
         pending_components.merge_data_columns(kzg_verified_data_columns)?;
 
-        let num_expected_columns = self.custody_context.sampling_size(Some(epoch), &self.spec);
+        let num_expected_columns = self
+            .custody_context
+            .num_of_data_columns_to_sample(Some(epoch), &self.spec);
         debug!(
             component = "data_columns",
             ?block_root,
@@ -622,7 +625,9 @@ impl<T: BeaconChainTypes> DataAvailabilityCheckerInner<T> {
         // Merge in the block.
         pending_components.merge_block(diet_executed_block);
 
-        let num_expected_columns = self.custody_context.sampling_size(Some(epoch), &self.spec);
+        let num_expected_columns = self
+            .custody_context
+            .num_of_data_columns_to_sample(Some(epoch), &self.spec);
         debug!(
             component = "block",
             ?block_root,
@@ -631,11 +636,11 @@ impl<T: BeaconChainTypes> DataAvailabilityCheckerInner<T> {
         );
 
         // Check if we have all components and entire set is consistent.
-        if let Some(available_block) = pending_components.make_available(
+        if let Some(available_block) =
-            &self.spec,
+            pending_components.make_available(&self.spec, num_expected_columns, |block| {
-            self.custody_context.sampling_size(Some(epoch), &self.spec),
+                self.state_cache.recover_pending_executed_block(block)
-            |block| self.state_cache.recover_pending_executed_block(block),
+            })?
-        )? {
+        {
             // We keep the pending components in the availability cache during block import (#5845).
             write_lock.put(block_root, pending_components);
             drop(write_lock);

beacon_node/beacon_chain/src/test_utils.rs

@@ -777,7 +777,7 @@ where
         self.chain
             .data_availability_checker
             .custody_context()
-            .sampling_size(None, &self.chain.spec) as usize
+            .num_of_data_columns_to_sample(None, &self.chain.spec) as usize
     }
 
     pub fn slots_per_epoch(&self) -> u64 {

beacon_node/beacon_chain/src/validator_custody.rs

@@ -215,7 +215,8 @@ impl CustodyContext {
                 );
                 return Some(CustodyCountChanged {
                     new_custody_group_count: updated_cgc,
-                    sampling_count: self.sampling_size(Some(effective_epoch), spec),
+                    sampling_count: self
+                        .num_of_custody_groups_to_sample(Some(effective_epoch), spec),
                 });
             }
         }
@@ -240,9 +241,13 @@ impl CustodyContext {
         }
     }
 
-    /// Returns the count of custody columns this node must sample for a block at `epoch` to import.
+    /// This function is used to determine the custody group count at a given epoch.
-    /// If an `epoch` is not specified, returns the *current* validator custody requirement.
+    ///
-    pub fn sampling_size(&self, epoch_opt: Option<Epoch>, spec: &ChainSpec) -> u64 {
+    /// This differs from the number of custody groups sampled per slot, as the spec requires a
+    /// minimum sampling size which may exceed the custody group count (CGC).
+    ///
+    /// See also: [`Self::num_of_custody_groups_to_sample`].
+    fn custody_group_count_at_epoch(&self, epoch_opt: Option<Epoch>, spec: &ChainSpec) -> u64 {
         let custody_group_count = if self.current_is_supernode {
             spec.number_of_custody_groups
         } else if let Some(epoch) = epoch_opt {
@@ -253,8 +258,26 @@ impl CustodyContext {
         } else {
             self.custody_group_count_at_head(spec)
         };
+        custody_group_count
+    }
 
-        spec.sampling_size(custody_group_count)
+    /// Returns the count of custody groups this node must _sample_ for a block at `epoch` to import.
+    /// If an `epoch` is not specified, returns the *current* validator custody requirement.
+    pub fn num_of_custody_groups_to_sample(
+        &self,
+        epoch_opt: Option<Epoch>,
+        spec: &ChainSpec,
+    ) -> u64 {
+        let custody_group_count = self.custody_group_count_at_epoch(epoch_opt, spec);
+        spec.sampling_size_custody_groups(custody_group_count)
+            .expect("should compute node sampling size from valid chain spec")
+    }
+
+    /// Returns the count of columns this node must _sample_ for a block at `epoch` to import.
+    /// If an `epoch` is not specified, returns the *current* validator custody requirement.
+    pub fn num_of_data_columns_to_sample(&self, epoch_opt: Option<Epoch>, spec: &ChainSpec) -> u64 {
+        let custody_group_count = self.custody_group_count_at_epoch(epoch_opt, spec);
+        spec.sampling_size_columns(custody_group_count)
             .expect("should compute node sampling size from valid chain spec")
     }
 }
@@ -307,7 +330,7 @@ mod tests {
             spec.number_of_custody_groups
         );
         assert_eq!(
-            custody_context.sampling_size(None, &spec),
+            custody_context.num_of_custody_groups_to_sample(None, &spec),
             spec.number_of_custody_groups
         );
     }
@@ -322,7 +345,7 @@ mod tests {
             "head custody count should be minimum spec custody requirement"
         );
         assert_eq!(
-            custody_context.sampling_size(None, &spec),
+            custody_context.num_of_custody_groups_to_sample(None, &spec),
             spec.samples_per_slot
         );
     }
@@ -412,7 +435,7 @@ mod tests {
 
         register_validators_and_assert_cgc(&custody_context, validators_and_expected_cgc, &spec);
         assert_eq!(
-            custody_context.sampling_size(None, &spec),
+            custody_context.num_of_custody_groups_to_sample(None, &spec),
             spec.number_of_custody_groups
         );
     }
@@ -423,7 +446,7 @@ mod tests {
         let spec = E::default_spec();
         let current_slot = Slot::new(10);
         let current_epoch = current_slot.epoch(E::slots_per_epoch());
-        let default_sampling_size = custody_context.sampling_size(None, &spec);
+        let default_sampling_size = custody_context.num_of_custody_groups_to_sample(None, &spec);
         let validator_custody_units = 10;
 
         let _cgc_changed = custody_context.register_validators::<E>(
@@ -437,12 +460,12 @@ mod tests {
 
         // CGC update is not applied for `current_epoch`.
         assert_eq!(
-            custody_context.sampling_size(Some(current_epoch), &spec),
+            custody_context.num_of_custody_groups_to_sample(Some(current_epoch), &spec),
             default_sampling_size
         );
         // CGC update is applied for the next epoch.
         assert_eq!(
-            custody_context.sampling_size(Some(current_epoch + 1), &spec),
+            custody_context.num_of_custody_groups_to_sample(Some(current_epoch + 1), &spec),
             validator_custody_units
         );
     }

beacon_node/lighthouse_network/src/types/globals.rs

@@ -66,7 +66,7 @@ impl<E: EthSpec> NetworkGlobals<E> {
         // The below `expect` calls will panic on start up if the chain spec config values used
         // are invalid
         let sampling_size = spec
-            .sampling_size(custody_group_count)
+            .sampling_size_custody_groups(custody_group_count)
             .expect("should compute node sampling size from valid chain spec");
         let custody_groups = get_custody_groups(node_id, sampling_size, &spec)
             .expect("should compute node custody groups");
@@ -114,7 +114,7 @@ impl<E: EthSpec> NetworkGlobals<E> {
         // are invalid
         let sampling_size = self
             .spec
-            .sampling_size(custody_group_count)
+            .sampling_size_custody_groups(custody_group_count)
             .expect("should compute node sampling size from valid chain spec");
         let custody_groups =
             get_custody_groups(self.local_enr().node_id().raw(), sampling_size, &self.spec)
@@ -298,7 +298,13 @@ mod test {
         spec.fulu_fork_epoch = Some(Epoch::new(0));
 
         let custody_group_count = spec.number_of_custody_groups / 2;
-        let subnet_sampling_size = spec.sampling_size(custody_group_count).unwrap();
+        let sampling_size_custody_groups = spec
+            .sampling_size_custody_groups(custody_group_count)
+            .unwrap();
+        let expected_sampling_subnet_count = sampling_size_custody_groups
+            * spec.data_column_sidecar_subnet_count
+            / spec.number_of_custody_groups;
+
         let metadata = get_metadata(custody_group_count);
         let config = Arc::new(NetworkConfig::default());
 
@@ -310,7 +316,7 @@ mod test {
         );
         assert_eq!(
             globals.sampling_subnets.read().len(),
-            subnet_sampling_size as usize
+            expected_sampling_subnet_count as usize
         );
     }
 
@@ -321,7 +327,7 @@ mod test {
         spec.fulu_fork_epoch = Some(Epoch::new(0));
 
         let custody_group_count = spec.number_of_custody_groups / 2;
-        let subnet_sampling_size = spec.sampling_size(custody_group_count).unwrap();
+        let expected_sampling_columns = spec.sampling_size_columns(custody_group_count).unwrap();
         let metadata = get_metadata(custody_group_count);
         let config = Arc::new(NetworkConfig::default());
 
@@ -333,7 +339,7 @@ mod test {
         );
         assert_eq!(
             globals.sampling_columns.read().len(),
-            subnet_sampling_size as usize
+            expected_sampling_columns as usize
         );
     }
 

consensus/types/src/chain_spec.rs

@@ -720,17 +720,24 @@ impl ChainSpec {
     }
 
     /// Returns the number of column sidecars to sample per slot.
-    pub fn sampling_size(&self, custody_group_count: u64) -> Result<u64, String> {
+    pub fn sampling_size_columns(&self, custody_group_count: u64) -> Result<u64, String> {
+        let sampling_size_groups = self.sampling_size_custody_groups(custody_group_count)?;
+
         let columns_per_custody_group = self
             .number_of_columns
             .safe_div(self.number_of_custody_groups)
             .map_err(|_| "number_of_custody_groups must be greater than 0")?;
 
-        let custody_column_count = columns_per_custody_group
+        let sampling_size_columns = columns_per_custody_group
-            .safe_mul(custody_group_count)
+            .safe_mul(sampling_size_groups)
             .map_err(|_| "Computing sampling size should not overflow")?;
 
-        Ok(std::cmp::max(custody_column_count, self.samples_per_slot))
+        Ok(sampling_size_columns)
+    }
+
+    /// Returns the number of custody groups to sample per slot.
+    pub fn sampling_size_custody_groups(&self, custody_group_count: u64) -> Result<u64, String> {
+        Ok(std::cmp::max(custody_group_count, self.samples_per_slot))
     }
 
     pub fn all_data_column_sidecar_subnets(&self) -> impl Iterator<Item = DataColumnSubnetId> {