Add experimental complete-blob-backfill flag (#7751)

A different (and complementary) approach for:

- https://github.com/sigp/lighthouse/issues/5391


  This PR adds a flag to set the DA boundary to the Deneb fork. The effect of this change is that Lighthouse will try to backfill _all_ blobs.

Most peers do not have this data, but I'm thinking that combined with `trusted-peers` this could be quite effective.


Co-Authored-By: Michael Sproul <michael@sigmaprime.io>

beacon_node/beacon_chain/src/data_availability_checker.rs

@@ -78,6 +78,7 @@ pub const STATE_LRU_CAPACITY: usize = STATE_LRU_CAPACITY_NON_ZERO.get();
 /// proposer. Having a capacity > 1 is an optimization to prevent sync lookup from having re-fetch
 /// data during moments of unstable network conditions.
 pub struct DataAvailabilityChecker<T: BeaconChainTypes> {
+    complete_blob_backfill: bool,
     availability_cache: Arc<DataAvailabilityCheckerInner<T>>,
     slot_clock: T::SlotClock,
     kzg: Arc<Kzg>,
@@ -116,6 +117,7 @@ impl<E: EthSpec> Debug for Availability<E> {
 
 impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
     pub fn new(
+        complete_blob_backfill: bool,
         slot_clock: T::SlotClock,
         kzg: Arc<Kzg>,
         store: BeaconStore<T>,
@@ -129,6 +131,7 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
             spec.clone(),
         )?;
         Ok(Self {
+            complete_blob_backfill,
             availability_cache: Arc::new(inner),
             slot_clock,
             kzg,
@@ -518,9 +521,15 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
     /// The epoch at which we require a data availability check in block processing.
     /// `None` if the `Deneb` fork is disabled.
     pub fn data_availability_boundary(&self) -> Option<Epoch> {
-        let current_epoch = self.slot_clock.now()?.epoch(T::EthSpec::slots_per_epoch());
-        self.spec
-            .min_epoch_data_availability_boundary(current_epoch)
+        let fork_epoch = self.spec.deneb_fork_epoch?;
+
+        if self.complete_blob_backfill {
+            Some(fork_epoch)
+        } else {
+            let current_epoch = self.slot_clock.now()?.epoch(T::EthSpec::slots_per_epoch());
+            self.spec
+                .min_epoch_data_availability_boundary(current_epoch)
+        }
     }
 
     /// Returns true if the given epoch lies within the da boundary and false otherwise.
@@ -1076,7 +1085,15 @@ mod test {
         let kzg = get_kzg(&spec);
         let store = Arc::new(HotColdDB::open_ephemeral(<_>::default(), spec.clone()).unwrap());
         let custody_context = Arc::new(CustodyContext::new(false));
-        DataAvailabilityChecker::new(slot_clock, kzg, store, custody_context, spec)
-            .expect("should initialise data availability checker")
+        let complete_blob_backfill = false;
+        DataAvailabilityChecker::new(
+            complete_blob_backfill,
+            slot_clock,
+            kzg,
+            store,
+            custody_context,
+            spec,
+        )
+        .expect("should initialise data availability checker")
     }
 }