Fix CGC backfill race condition (#8267)

During custody backfill sync there could be an edge case where we update CGC at the same time where we are importing a batch of columns which may cause us to incorrectly overwrite values when calling `backfill_validator_custody_requirements`. To prevent this race condition, the expected cgc is now passed into this function and is used to check if the expected cgc == the current validator cgc. If the values arent equal, this probably indicates that a very recent CGC occurred so we do not prune/update values in the `epoch_validator_custody_requirements` map.


  


Co-Authored-By: Eitan Seri-Levi <eserilev@ucsc.edu>

beacon_node/network/src/network_beacon_processor/mod.rs

@@ -497,9 +497,11 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
         self: &Arc<Self>,
         batch_id: CustodyBackfillBatchId,
         data_columns: DataColumnSidecarList<T::EthSpec>,
+        expected_cgc: u64,
     ) -> Result<(), Error<T::EthSpec>> {
         let processor = self.clone();
-        let process_fn = move || processor.process_historic_data_columns(batch_id, data_columns);
+        let process_fn =
+            move || processor.process_historic_data_columns(batch_id, data_columns, expected_cgc);
 
         let work = Work::ChainSegmentBackfill(Box::new(process_fn));
 

beacon_node/network/src/network_beacon_processor/sync_methods.rs

@@ -426,6 +426,7 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
         &self,
         batch_id: CustodyBackfillBatchId,
         downloaded_columns: DataColumnSidecarList<T::EthSpec>,
+        expected_cgc: u64,
     ) {
         let _guard = debug_span!(
             SPAN_CUSTODY_BACKFILL_SYNC_IMPORT_COLUMNS,
@@ -435,10 +436,11 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
         .entered();
 
         let sent_columns = downloaded_columns.len();
-        let result = match self
-            .chain
-            .import_historical_data_column_batch(batch_id.epoch, downloaded_columns)
-        {
+        let result = match self.chain.import_historical_data_column_batch(
+            batch_id.epoch,
+            downloaded_columns,
+            expected_cgc,
+        ) {
             Ok(imported_columns) => {
                 metrics::inc_counter_by(
                     &metrics::BEACON_PROCESSOR_CUSTODY_BACKFILL_COLUMN_IMPORT_SUCCESS_TOTAL,

beacon_node/network/src/sync/custody_backfill_sync/mod.rs

@@ -504,6 +504,7 @@ impl<T: BeaconChainTypes> CustodyBackFillSync<T> {
                 run_id: self.run_id,
             },
             data_columns,
+            self.cgc,
         ) {
             crit!(
                 msg = "process_batch",