Fix data column rpc request (#8247)

Fixes an issue mentioned in this comment regarding data column rpc requests:
https://github.com/sigp/lighthouse/issues/6572#issuecomment-3400076236


  


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

Co-Authored-By: Michael Sproul <micsproul@gmail.com>

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -6946,9 +6946,49 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
     pub fn update_data_column_custody_info(&self, slot: Option<Slot>) {
         self.store
             .put_data_column_custody_info(slot)
-            .unwrap_or_else(
-                |e| tracing::error!(error = ?e, "Failed to update data column custody info"),
-            );
+            .unwrap_or_else(|e| error!(error = ?e, "Failed to update data column custody info"));
+    }
+
+    /// Get the earliest epoch in which the node has met its custody requirements.
+    /// A `None` response indicates that we've met our custody requirements up to the
+    /// column data availability window
+    pub fn earliest_custodied_data_column_epoch(&self) -> Option<Epoch> {
+        self.store
+            .get_data_column_custody_info()
+            .inspect_err(
+                |e| error!(error=?e, "Failed to get data column custody info from the store"),
+            )
+            .ok()
+            .flatten()
+            .and_then(|info| info.earliest_data_column_slot)
+            .map(|slot| {
+                let mut epoch = slot.epoch(T::EthSpec::slots_per_epoch());
+                // If the earliest custodied slot isn't the first slot in the epoch
+                // The node has only met its custody requirements for the next epoch.
+                if slot > epoch.start_slot(T::EthSpec::slots_per_epoch()) {
+                    epoch += 1;
+                }
+                epoch
+            })
+    }
+
+    /// The data availability boundary for custodying columns. It will just be the
+    /// regular data availability boundary unless we are near the Fulu fork epoch.
+    pub fn column_data_availability_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 // Fulu hasn't been enabled
+                }
+            }
+            None => None, // Deneb hasn't been enabled
+        }
     }
 
     /// This method serves to get a sense of the current chain health. It is used in block proposal

beacon_node/beacon_chain/tests/store_tests.rs

@@ -4369,6 +4369,65 @@ async fn fulu_prune_data_columns_fork_boundary() {
     check_data_column_existence(&harness, pruned_slot, harness.head_slot(), true);
 }
 
+#[tokio::test]
+async fn test_column_da_boundary() {
+    let mut spec = ForkName::Electra.make_genesis_spec(E::default_spec());
+    let fulu_fork_epoch = Epoch::new(4);
+    spec.fulu_fork_epoch = Some(fulu_fork_epoch);
+    let db_path = tempdir().unwrap();
+    let store = get_store_generic(&db_path, StoreConfig::default(), spec);
+
+    if !store.get_chain_spec().is_peer_das_scheduled() {
+        // No-op if PeerDAS not scheduled.
+        panic!("PeerDAS not scheduled");
+    }
+
+    let harness = get_harness(store.clone(), LOW_VALIDATOR_COUNT);
+
+    // The column da boundary should be the fulu fork epoch
+    assert_eq!(
+        harness.chain.column_data_availability_boundary(),
+        Some(fulu_fork_epoch)
+    );
+}
+
+#[tokio::test]
+async fn test_earliest_custodied_data_column_epoch() {
+    let spec = ForkName::Fulu.make_genesis_spec(E::default_spec());
+    let db_path = tempdir().unwrap();
+    let store = get_store_generic(&db_path, StoreConfig::default(), spec);
+    let custody_info_epoch = Epoch::new(4);
+
+    if !store.get_chain_spec().is_peer_das_scheduled() {
+        // No-op if PeerDAS not scheduled.
+        panic!("PeerDAS not scheduled");
+    }
+
+    let harness = get_harness(store.clone(), LOW_VALIDATOR_COUNT);
+
+    // earliest custody info is set to the last slot in `custody_info_epoch`
+    harness
+        .chain
+        .update_data_column_custody_info(Some(custody_info_epoch.end_slot(E::slots_per_epoch())));
+
+    // earliest custodied data column epoch should be `custody_info_epoch` + 1
+    assert_eq!(
+        harness.chain.earliest_custodied_data_column_epoch(),
+        Some(custody_info_epoch + 1)
+    );
+
+    // earliest custody info is set to the first slot in `custody_info_epoch`
+    harness
+        .chain
+        .update_data_column_custody_info(Some(custody_info_epoch.start_slot(E::slots_per_epoch())));
+
+    // earliest custodied data column epoch should be `custody_info_epoch`
+    assert_eq!(
+        harness.chain.earliest_custodied_data_column_epoch(),
+        Some(custody_info_epoch)
+    );
+}
+
 /// Check that blob pruning prunes data columns older than the data availability boundary with
 /// margin applied.
 #[tokio::test]

beacon_node/network/src/network_beacon_processor/rpc_methods.rs

@@ -1204,33 +1204,42 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
 
         let request_start_slot = Slot::from(req.start_slot);
 
-        let data_availability_boundary_slot = match self.chain.data_availability_boundary() {
-            Some(boundary) => boundary.start_slot(T::EthSpec::slots_per_epoch()),
-            None => {
-                debug!("Deneb fork is disabled");
-                return Err((RpcErrorResponse::InvalidRequest, "Deneb fork is disabled"));
-            }
-        };
+        let column_data_availability_boundary_slot =
+            match self.chain.column_data_availability_boundary() {
+                Some(boundary) => boundary.start_slot(T::EthSpec::slots_per_epoch()),
+                None => {
+                    debug!("Fulu fork is disabled");
+                    return Err((RpcErrorResponse::InvalidRequest, "Fulu fork is disabled"));
+                }
+            };
 
-        let oldest_data_column_slot = self
-            .chain
-            .store
-            .get_data_column_info()
-            .oldest_data_column_slot
-            .unwrap_or(data_availability_boundary_slot);
+        let earliest_custodied_data_column_slot =
+            match self.chain.earliest_custodied_data_column_epoch() {
+                Some(earliest_custodied_epoch) => {
+                    let earliest_custodied_slot =
+                        earliest_custodied_epoch.start_slot(T::EthSpec::slots_per_epoch());
+                    // Ensure the earliest columns we serve are within the data availability window
+                    if earliest_custodied_slot < column_data_availability_boundary_slot {
+                        column_data_availability_boundary_slot
+                    } else {
+                        earliest_custodied_slot
+                    }
+                }
+                None => column_data_availability_boundary_slot,
+            };
 
-        if request_start_slot < oldest_data_column_slot {
+        if request_start_slot < earliest_custodied_data_column_slot {
             debug!(
                 %request_start_slot,
-                %oldest_data_column_slot,
-                %data_availability_boundary_slot,
-                "Range request start slot is older than data availability boundary."
+                %earliest_custodied_data_column_slot,
+                %column_data_availability_boundary_slot,
+                "Range request start slot is older than the earliest custodied data column slot."
             );
 
-            return if data_availability_boundary_slot < oldest_data_column_slot {
+            return if earliest_custodied_data_column_slot > column_data_availability_boundary_slot {
                 Err((
                     RpcErrorResponse::ResourceUnavailable,
-                    "blobs pruned within boundary",
+                    "columns pruned within boundary",
                 ))
             } else {
                 Err((