Custody backfill sync

beacon_node/Cargo.toml

@@ -40,6 +40,7 @@ slasher = { workspace = true }
 store = { workspace = true }
 strum = { workspace = true }
 task_executor = { workspace = true }
+tokio = { workspace = true }
 tracing = { workspace = true }
 types = { workspace = true }
 

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -28,6 +28,7 @@ use crate::data_column_verification::{GossipDataColumnError, GossipVerifiedDataC
 use crate::early_attester_cache::EarlyAttesterCache;
 use crate::errors::{BeaconChainError as Error, BlockProductionError};
 use crate::events::ServerSentEventHandler;
+use crate::events::SyncServiceMessage;
 use crate::execution_payload::{NotifyExecutionLayer, PreparePayloadHandle, get_execution_payload};
 use crate::fetch_blobs::EngineGetBlobsOutput;
 use crate::fork_choice_signal::{ForkChoiceSignalRx, ForkChoiceSignalTx, ForkChoiceWaitResult};
@@ -125,6 +126,7 @@ use store::{
     KeyValueStore, KeyValueStoreOp, StoreItem, StoreOp,
 };
 use task_executor::{ShutdownReason, TaskExecutor};
+use tokio::sync::mpsc;
 use tokio_stream::Stream;
 use tracing::{Span, debug, debug_span, error, info, info_span, instrument, trace, warn};
 use tree_hash::TreeHash;
@@ -490,6 +492,8 @@ pub struct BeaconChain<T: BeaconChainTypes> {
     pub kzg: Arc<Kzg>,
     /// RNG instance used by the chain. Currently used for shuffling column sidecars in block publishing.
     pub rng: Arc<Mutex<Box<dyn RngCore + Send>>>,
+    /// The sending channel for the beacon chain to send messages to sync.
+    pub sync_service_send: mpsc::UnboundedSender<SyncServiceMessage>,
 }
 
 pub enum BeaconBlockResponseWrapper<E: EthSpec> {
@@ -6898,6 +6902,24 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             );
     }
 
+    /// The da boundary for custodying columns. It will just be the DA boundary unless we are near the Fulu fork epoch.
+    pub fn get_column_da_boundary(&self) -> Option<Epoch> {
+        match self.data_availability_boundary() {
+            Some(da_boundary_epoch) => {
+                if let Some(fulu_fork_epoch) = self.spec.fulu_fork_epoch {
+                    if da_boundary_epoch < fulu_fork_epoch {
+                        Some(fulu_fork_epoch)
+                    } else {
+                        Some(da_boundary_epoch)
+                    }
+                } else {
+                    None
+                }
+            }
+            None => None, // If no DA boundary set, dont try to custody backfill
+        }
+    }
+
     /// This method serves to get a sense of the current chain health. It is used in block proposal
     /// to determine whether we should outsource payload production duties.
     ///

beacon_node/beacon_chain/src/builder.rs

@@ -5,6 +5,7 @@ use crate::beacon_chain::{
 };
 use crate::beacon_proposer_cache::BeaconProposerCache;
 use crate::data_availability_checker::DataAvailabilityChecker;
+use crate::events::SyncServiceMessage;
 use crate::fork_choice_signal::ForkChoiceSignalTx;
 use crate::fork_revert::{reset_fork_choice_to_finalization, revert_to_fork_boundary};
 use crate::graffiti_calculator::{GraffitiCalculator, GraffitiOrigin};
@@ -38,6 +39,7 @@ use std::sync::Arc;
 use std::time::Duration;
 use store::{Error as StoreError, HotColdDB, ItemStore, KeyValueStoreOp};
 use task_executor::{ShutdownReason, TaskExecutor};
+use tokio::sync::mpsc;
 use tracing::{debug, error, info};
 use types::{
     BeaconBlock, BeaconState, BlobSidecarList, ChainSpec, DataColumnSidecarList, Epoch, EthSpec,
@@ -102,6 +104,7 @@ pub struct BeaconChainBuilder<T: BeaconChainTypes> {
     validator_monitor_config: Option<ValidatorMonitorConfig>,
     import_all_data_columns: bool,
     rng: Option<Box<dyn RngCore + Send>>,
+    sync_service_send: Option<mpsc::UnboundedSender<SyncServiceMessage>>,
 }
 
 impl<TSlotClock, E, THotStore, TColdStore>
@@ -141,6 +144,7 @@ where
             validator_monitor_config: None,
             import_all_data_columns: false,
             rng: None,
+            sync_service_send: None,
         }
     }
 
@@ -662,6 +666,14 @@ where
         self
     }
 
+    pub fn sync_service_send(
+        mut self,
+        sync_service_send: Option<mpsc::UnboundedSender<SyncServiceMessage>>,
+    ) -> Self {
+        self.sync_service_send = sync_service_send;
+        self
+    }
+
     /// Fetch a reference to the slot clock.
     ///
     /// Can be used for mutation during testing due to `SlotClock`'s internal mutability.
@@ -941,6 +953,10 @@ where
         };
         debug!(?custody_context, "Loading persisted custody context");
 
+        let Some(sync_service_send) = self.sync_service_send else {
+            return Err("Must supply a sync service send channel".to_string());
+        };
+
         let beacon_chain = BeaconChain {
             spec: self.spec.clone(),
             config: self.chain_config,
@@ -1023,6 +1039,7 @@ where
             ),
             kzg: self.kzg.clone(),
             rng: Arc::new(Mutex::new(rng)),
+            sync_service_send,
         };
 
         let head = beacon_chain.head_snapshot();
@@ -1244,6 +1261,8 @@ mod test {
 
         let kzg = get_kzg(&spec);
 
+        let (sync_service_send, _) = mpsc::unbounded_channel::<SyncServiceMessage>();
+
         let chain = Builder::new(MinimalEthSpec, kzg)
             .store(Arc::new(store))
             .task_executor(runtime.task_executor.clone())
@@ -1252,6 +1271,7 @@ mod test {
             .testing_slot_clock(Duration::from_secs(1))
             .expect("should configure testing slot clock")
             .shutdown_sender(shutdown_tx)
+            .sync_service_send(Some(sync_service_send))
             .rng(Box::new(StdRng::seed_from_u64(42)))
             .build()
             .expect("should build");

beacon_node/beacon_chain/src/canonical_head.rs

@@ -31,6 +31,7 @@
 //! the head block root. This is unacceptable for fast-responding functions like the networking
 //! stack.
 
+use crate::events::SyncServiceMessage;
 use crate::persisted_fork_choice::PersistedForkChoice;
 use crate::shuffling_cache::BlockShufflingIds;
 use crate::{
@@ -991,12 +992,49 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                 .process_prune_blobs(data_availability_boundary);
         }
 
+        if self.should_trigger_custody_backfill_sync(new_view.finalized_checkpoint.epoch)
+            && let Err(e) = self
+                .sync_service_send
+                .send(SyncServiceMessage::EarliestCustodyEpochFinalized)
+        {
+            error!(
+                finalized_epoch=?new_view.finalized_checkpoint.epoch,
+                error=?e,
+                "Unable to trigger custody backfill sync at finalized epoch"
+            );
+        }
+
         // Take a write-lock on the canonical head and signal for it to prune.
         self.canonical_head.fork_choice_write_lock().prune()?;
 
         Ok(())
     }
 
+    /// Checks if we should trigger custody backfill sync.
+    /// Returns true if the earliest data column custodied is from an epoch
+    /// that's greater than or equal to the most recently finalized epoch and is within the DA boundary.
+    fn should_trigger_custody_backfill_sync(&self, finalized_epoch: Epoch) -> bool {
+        let Some(earliest_data_column_slot) = self
+            .store
+            .get_data_column_custody_info()
+            .ok()
+            .flatten()
+            .and_then(|custody_info| custody_info.earliest_data_column_slot)
+        else {
+            return false;
+        };
+
+        let Some(column_da_boundary) = self.get_column_da_boundary() else {
+            return false;
+        };
+
+        let earliest_data_column_epoch =
+            earliest_data_column_slot.epoch(T::EthSpec::slots_per_epoch());
+
+        earliest_data_column_epoch >= finalized_epoch
+            && earliest_data_column_epoch > column_da_boundary
+    }
+
     /// Persist fork choice to disk, writing immediately.
     pub fn persist_fork_choice(&self) -> Result<(), Error> {
         let _fork_choice_timer = metrics::start_timer(&metrics::PERSIST_FORK_CHOICE);

beacon_node/beacon_chain/src/events.rs

@@ -1,8 +1,10 @@
+use std::collections::HashSet;
+
 pub use eth2::types::{EventKind, SseBlock, SseFinalizedCheckpoint, SseHead};
 use tokio::sync::broadcast;
 use tokio::sync::broadcast::{Receiver, Sender, error::SendError};
 use tracing::trace;
-use types::EthSpec;
+use types::{ColumnIndex, EthSpec};
 
 const DEFAULT_CHANNEL_CAPACITY: usize = 16;
 
@@ -312,3 +314,9 @@ impl<E: EthSpec> ServerSentEventHandler<E> {
         self.block_gossip_tx.receiver_count() > 0
     }
 }
+
+#[derive(Debug)]
+pub enum SyncServiceMessage {
+    CustodyCountChanged { columns: HashSet<ColumnIndex> },
+    EarliestCustodyEpochFinalized,
+}

beacon_node/beacon_chain/src/historical_data_columns.rs

@@ -0,0 +1,145 @@
+use std::collections::{HashMap, HashSet};
+
+use crate::{
+    BeaconChain, BeaconChainError, BeaconChainTypes,
+    data_column_verification::verify_kzg_for_data_column_list,
+};
+use store::{Error as StoreError, KeyValueStore};
+use types::{ColumnIndex, DataColumnSidecarList, Epoch, EthSpec, Hash256, Slot};
+
+#[derive(Debug)]
+pub enum HistoricalDataColumnError {
+    // The provided data column sidecar pertains to a block that doesn't exist in the database.
+    NoBlockFound {
+        data_column_block_root: Hash256,
+    },
+
+    /// Logic error: should never occur.
+    IndexOutOfBounds,
+
+    /// The provided data column sidecar list doesn't contain columns for the full range of slots for the given epoch.
+    MissingDataColumns {
+        missing_slots_and_data_columns: Vec<(Slot, ColumnIndex)>,
+    },
+
+    /// The provided data column sidecar list contains at least one column with an invalid kzg commitment.
+    InvalidKzg,
+
+    /// Internal store error
+    StoreError(StoreError),
+
+    /// Internal beacon chain error
+    BeaconChainError(Box<BeaconChainError>),
+}
+
+impl From<StoreError> for HistoricalDataColumnError {
+    fn from(e: StoreError) -> Self {
+        Self::StoreError(e)
+    }
+}
+
+impl<T: BeaconChainTypes> BeaconChain<T> {
+    /// Store a batch of historical data columns in the database.
+    ///
+    /// The data columns block roots and proposer signatures are verified with the existing
+    /// block stored in the DB. This function assumes that KZG proofs have already been verified.
+    ///
+    /// This function requires that the data column sidecar list contains columns for a full epoch.
+    ///
+    /// Return the number of `data_columns` successfully imported.
+    pub fn import_historical_data_column_batch(
+        &self,
+        epoch: Epoch,
+        historical_data_column_sidecar_list: DataColumnSidecarList<T::EthSpec>,
+    ) -> Result<usize, HistoricalDataColumnError> {
+        let mut total_imported = 0;
+        let mut ops = vec![];
+
+        let unique_column_indices = historical_data_column_sidecar_list
+            .iter()
+            .map(|item| item.index)
+            .collect::<HashSet<_>>();
+
+        let mut slot_and_column_index_to_data_columns = historical_data_column_sidecar_list
+            .iter()
+            .map(|data_column| ((data_column.slot(), data_column.index), data_column))
+            .collect::<HashMap<_, _>>();
+
+        if historical_data_column_sidecar_list.is_empty() {
+            return Ok(total_imported);
+        }
+
+        let forward_blocks_iter = self
+            .forwards_iter_block_roots_until(
+                epoch.start_slot(T::EthSpec::slots_per_epoch()),
+                epoch.end_slot(T::EthSpec::slots_per_epoch()),
+            )
+            .map_err(|e| HistoricalDataColumnError::BeaconChainError(Box::new(e)))?;
+
+        for block_iter_result in forward_blocks_iter {
+            let (block_root, slot) = block_iter_result
+                .map_err(|e| HistoricalDataColumnError::BeaconChainError(Box::new(e)))?;
+
+            for column_index in unique_column_indices.clone() {
+                if let Some(data_column) =
+                    slot_and_column_index_to_data_columns.remove(&(slot, column_index))
+                {
+                    if self
+                        .store
+                        .get_data_column(&block_root, &data_column.index)?
+                        .is_none()
+                    {
+                        tracing::debug!(
+                            block_root = ?block_root,
+                            column_index = data_column.index,
+                            "Skipping data column import as identical data column exists"
+                        );
+                        continue;
+                    }
+                    if block_root != data_column.block_root() {
+                        return Err(HistoricalDataColumnError::NoBlockFound {
+                            data_column_block_root: data_column.block_root(),
+                        });
+                    }
+                    self.store.data_column_as_kv_store_ops(
+                        &block_root,
+                        data_column.clone(),
+                        &mut ops,
+                    );
+                    total_imported += 1;
+                }
+            }
+        }
+
+        verify_kzg_for_data_column_list(historical_data_column_sidecar_list.iter(), &self.kzg)
+            .map_err(|_| HistoricalDataColumnError::InvalidKzg)?;
+
+        self.store.blobs_db.do_atomically(ops)?;
+
+        if slot_and_column_index_to_data_columns.is_empty() {
+            self.store.put_data_column_custody_info(Some(
+                epoch.start_slot(T::EthSpec::slots_per_epoch()),
+            ))?;
+        } else {
+            tracing::warn!(
+                ?epoch,
+                missing_slots = ?slot_and_column_index_to_data_columns.keys().map(|(slot, _)| slot),
+                "Some data columns are missing from the batch"
+            );
+            return Err(HistoricalDataColumnError::MissingDataColumns {
+                missing_slots_and_data_columns: slot_and_column_index_to_data_columns
+                    .keys()
+                    .cloned()
+                    .collect::<Vec<_>>(),
+            });
+        }
+
+        self.data_availability_checker
+            .custody_context()
+            .backfill_custody_count_at_epoch(epoch);
+
+        tracing::debug!(total_imported, "Imported historical data columns");
+
+        Ok(total_imported)
+    }
+}

beacon_node/beacon_chain/src/lib.rs

@@ -28,6 +28,7 @@ pub mod fork_choice_signal;
 pub mod fork_revert;
 pub mod graffiti_calculator;
 pub mod historical_blocks;
+pub mod historical_data_columns;
 pub mod kzg_utils;
 pub mod light_client_finality_update_verification;
 pub mod light_client_optimistic_update_verification;