Add inclusion lists backfilling via InclusionListsByIndices

beacon_node/beacon_chain/src/inclusion_list_verification/gossip_verified_inclusion_list.rs

@@ -1,7 +1,7 @@
 use crate::canonical_head::CanonicalHead;
 use crate::inclusion_list_store::{InclusionListStore, InsertOutcome};
 use crate::inclusion_list_verification::{
-    InclusionListVerificationError, verify_inclusion_list_transactions_bounds,
+    InclusionListSource, InclusionListVerificationError, verify_inclusion_list_transactions_bounds,
 };
 use crate::shuffling_cache::{ShufflingCache, with_cached_shuffling};
 use crate::validator_pubkey_cache::ValidatorPubkeyCache;
@@ -17,6 +17,7 @@ use tracing::debug;
 use types::{ChainSpec, EthSpec, Hash256, SignedInclusionList, Slot};
 
 pub struct GossipVerificationContext<'a, T: BeaconChainTypes> {
+    pub source: InclusionListSource,
     pub canonical_head: &'a CanonicalHead<T>,
     pub inclusion_list_store: &'a RwLock<InclusionListStore<T::EthSpec>>,
     pub shuffling_cache: &'a RwLock<ShufflingCache<T::EthSpec>>,
@@ -46,21 +47,28 @@ impl GossipVerifiedInclusionList {
         let validator_index = inclusion_list.validator_index;
         let dependent_root = inclusion_list.dependent_root;
 
-        // [IGNORE] This is the first or second valid message from this validator.
-        if ctx
-            .inclusion_list_store
-            .read()
-            .seen_twice(slot, dependent_root, validator_index)
-        {
-            return Err(InclusionListVerificationError::AlreadySeenTwice {
-                validator_index,
-                slot,
-                dependent_root,
-            });
-        }
+        match ctx.source {
+            InclusionListSource::Gossip => {
+                // [IGNORE] This is the first or second valid message from this validator.
+                if ctx
+                    .inclusion_list_store
+                    .read()
+                    .seen_twice(slot, dependent_root, validator_index)
+                {
+                    return Err(InclusionListVerificationError::AlreadySeenTwice {
+                        validator_index,
+                        slot,
+                        dependent_root,
+                    });
+                }
 
-        // [IGNORE] `slot` is within the `MAXIMUM_GOSSIP_CLOCK_DISPARITY` allowance.
-        verify_propagation_slot_range(ctx.slot_clock, slot, ctx.spec)?;
+                // [IGNORE] `slot` is within the `MAXIMUM_GOSSIP_CLOCK_DISPARITY` allowance.
+                verify_propagation_slot_range(ctx.slot_clock, slot, ctx.spec)?;
+            }
+            InclusionListSource::Rpc => {
+                verify_retention_slot_range(ctx.slot_clock, slot, ctx.spec)?
+            }
+        }
 
         // [IGNORE] The size of the inclusion list transactions is non-zero.
         let transactions_size: u64 = inclusion_list
@@ -205,9 +213,44 @@ fn verify_propagation_slot_range<S: SlotClock>(
     Ok(())
 }
 
+/// Verify that the `slot` is within the range of slots the inclusion list store retains,
+/// with reference to the current slot of the clock.
+///
+/// Accounts for `MIN_SLOTS_FOR_INCLUSION_LISTS_REQUESTS`.
+fn verify_retention_slot_range<S: SlotClock>(
+    slot_clock: &S,
+    message_slot: Slot,
+    spec: &ChainSpec,
+) -> Result<(), InclusionListVerificationError> {
+    let current_slot = slot_clock
+        .now()
+        .ok_or(InclusionListVerificationError::UnableToReadSlot)?;
+    if message_slot > current_slot {
+        return Err(InclusionListVerificationError::FutureSlot {
+            message_slot,
+            latest_permissible_slot: current_slot,
+        });
+    }
+
+    let earliest_permissible_slot =
+        current_slot.saturating_sub(spec.min_slots_for_inclusion_lists_requests);
+    if message_slot < earliest_permissible_slot {
+        return Err(InclusionListVerificationError::PastSlot {
+            message_slot,
+            earliest_permissible_slot,
+        });
+    }
+
+    Ok(())
+}
+
 impl<T: BeaconChainTypes> BeaconChain<T> {
-    pub fn inclusion_list_gossip_verification_context(&self) -> GossipVerificationContext<'_, T> {
+    pub fn inclusion_list_gossip_verification_context(
+        &self,
+        source: InclusionListSource,
+    ) -> GossipVerificationContext<'_, T> {
         GossipVerificationContext {
+            source,
             canonical_head: &self.canonical_head,
             inclusion_list_store: &self.inclusion_list_store,
             shuffling_cache: &self.shuffling_cache,
@@ -224,18 +267,20 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         &self,
         signed_inclusion_list: SignedInclusionList,
         seen_timestamp: Duration,
+        source: InclusionListSource,
     ) -> Result<GossipVerifiedInclusionList, InclusionListVerificationError> {
         metrics::inc_counter(&metrics::INCLUSION_LIST_PROCESSING_REQUESTS);
         let slot = signed_inclusion_list.message.slot;
         let validator_index = signed_inclusion_list.message.validator_index;
 
-        let ctx = self.inclusion_list_gossip_verification_context();
+        let ctx = self.inclusion_list_gossip_verification_context(source);
         match GossipVerifiedInclusionList::new(signed_inclusion_list, seen_timestamp, &ctx) {
             Ok(verified) => {
                 metrics::inc_counter(&metrics::INCLUSION_LIST_PROCESSING_SUCCESSES);
                 debug!(
                     %slot,
                     %validator_index,
+                    ?source,
                     is_timely = verified.is_timely,
                     "Successfully verified gossip inclusion list"
                 );
@@ -247,6 +292,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                     error = e.to_string(),
                     %slot,
                     %validator_index,
+                    ?source,
                     "Rejected gossip inclusion list"
                 );
                 Err(e)

beacon_node/beacon_chain/src/inclusion_list_verification/mod.rs

@@ -16,6 +16,13 @@ use types::{BeaconStateError, ChainSpec, Hash256, ProgressiveTransactions, Slot}
 
 pub mod gossip_verified_inclusion_list;
 
+/// The path through which an inclusion list reached this node.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, AsRefStr)]
+pub enum InclusionListSource {
+    Gossip,
+    Rpc,
+}
+
 #[cfg(test)]
 mod tests;
 
@@ -69,6 +76,32 @@ pub enum InclusionListVerificationError {
     BeaconStateError(BeaconStateError),
 }
 
+impl InclusionListVerificationError {
+    /// Whether the peer that sent the inclusion list should be penalized.
+    pub fn penalize_peer(&self) -> bool {
+        // This match statement should never have a default case so that we are always forced to
+        // consider whether or not to penalize a peer when we add a new error condition.
+        match self {
+            // The list may be valid but is not useful to this node, or the node cannot tell.
+            InclusionListVerificationError::AlreadySeenTwice { .. }
+            | InclusionListVerificationError::FutureSlot { .. }
+            | InclusionListVerificationError::PastSlot { .. }
+            | InclusionListVerificationError::EmptyTransactions
+            | InclusionListVerificationError::DependentRootUnknown { .. }
+            | InclusionListVerificationError::InvalidDependentRoot { .. }
+            | InclusionListVerificationError::UnableToReadSlot
+            | InclusionListVerificationError::BeaconChainError(_)
+            | InclusionListVerificationError::BeaconStateError(_) => false,
+            // The list is invalid and an honest peer would not have sent it.
+            InclusionListVerificationError::InvalidTransactions(_)
+            | InclusionListVerificationError::DependentRootTooRecent { .. }
+            | InclusionListVerificationError::NotInCommittee { .. }
+            | InclusionListVerificationError::UnknownValidatorIndex(_)
+            | InclusionListVerificationError::InvalidSignature => true,
+        }
+    }
+}
+
 impl std::fmt::Display for InclusionListVerificationError {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         write!(f, "{:?}", self)

beacon_node/beacon_chain/src/inclusion_list_verification/tests.rs

@@ -12,7 +12,7 @@ use crate::{
     WhenSlotSkipped,
     inclusion_list_store::InsertOutcome,
     inclusion_list_verification::{
-        InclusionListVerificationError,
+        InclusionListSource, InclusionListVerificationError,
         gossip_verified_inclusion_list::{GossipVerificationContext, GossipVerifiedInclusionList},
     },
     test_utils::{BeaconChainHarness, EphemeralHarnessType, fork_name_from_env, test_spec},
@@ -66,7 +66,13 @@ impl TestContext {
     fn gossip_ctx(&self) -> GossipVerificationContext<'_, T> {
         self.harness
             .chain
-            .inclusion_list_gossip_verification_context()
+            .inclusion_list_gossip_verification_context(InclusionListSource::Gossip)
+    }
+
+    fn rpc_ctx(&self) -> GossipVerificationContext<'_, T> {
+        self.harness
+            .chain
+            .inclusion_list_gossip_verification_context(InclusionListSource::Rpc)
     }
 
     fn seen_timestamp(&self) -> Duration {
@@ -588,3 +594,68 @@ async fn inclusion_list_in_first_heze_epoch() {
     let result = GossipVerifiedInclusionList::new(signed, ctx.seen_timestamp(), &ctx.gossip_ctx());
     assert!(result.is_ok(), "expected Ok, got: {:?}", result);
 }
+
+#[test]
+fn rpc_list_already_seen_twice_is_verified() {
+    if !fork_name_from_env().is_some_and(|f| f.gloas_enabled()) {
+        return;
+    }
+    let ctx = TestContext::new();
+    let slot = ctx.current_slot();
+
+    for tx in [0xaa, 0xbb] {
+        let signed = ctx.valid_inclusion_list(slot, ctx.genesis_block_root, vec![vec![tx]]);
+        let verified =
+            GossipVerifiedInclusionList::new(signed, ctx.seen_timestamp(), &ctx.gossip_ctx())
+                .expect("should verify inclusion list");
+        ctx.harness.chain.import_inclusion_list(verified);
+    }
+
+    let signed = ctx.valid_inclusion_list(slot, ctx.genesis_block_root, vec![vec![0xcc]]);
+    assert!(GossipVerifiedInclusionList::new(signed, ctx.seen_timestamp(), &ctx.rpc_ctx()).is_ok());
+}
+
+#[test]
+fn rpc_list_within_retention_window() {
+    if !fork_name_from_env().is_some_and(|f| f.gloas_enabled()) {
+        return;
+    }
+    let ctx = TestContext::new();
+    let slot = ctx.current_slot();
+    let signed = ctx.valid_inclusion_list(slot, ctx.genesis_block_root, vec![vec![0xaa]]);
+
+    // A list for a slot the clock hasn't reached yet is rejected, the same as on gossip.
+    let future = ctx.valid_inclusion_list(slot + 1, ctx.genesis_block_root, vec![vec![0xaa]]);
+    assert!(matches!(
+        GossipVerifiedInclusionList::new(future, ctx.seen_timestamp(), &ctx.rpc_ctx()),
+        Err(InclusionListVerificationError::FutureSlot { .. })
+    ));
+
+    let spec = &ctx.harness.spec;
+    let last_retained_slot = slot + spec.min_slots_for_inclusion_lists_requests;
+    let first_unretained_slot = last_retained_slot + 1;
+
+    // Move past the gossip clock disparity into the last retained slot.
+    let last_retained_slot_start = spec.get_slot_duration() * last_retained_slot.as_u64() as u32;
+    ctx.harness
+        .chain
+        .slot_clock
+        .set_current_time(last_retained_slot_start + spec.maximum_gossip_clock_disparity());
+    assert!(matches!(
+        GossipVerifiedInclusionList::new(signed.clone(), ctx.seen_timestamp(), &ctx.gossip_ctx()),
+        Err(InclusionListVerificationError::PastSlot { .. })
+    ));
+    assert!(
+        GossipVerifiedInclusionList::new(signed.clone(), ctx.seen_timestamp(), &ctx.rpc_ctx())
+            .is_ok()
+    );
+
+    ctx.harness
+        .chain
+        .slot_clock
+        .set_slot(first_unretained_slot.as_u64());
+    assert!(matches!(
+        GossipVerifiedInclusionList::new(signed, ctx.seen_timestamp(), &ctx.rpc_ctx()),
+        Err(InclusionListVerificationError::PastSlot { .. })
+    ));
+}

beacon_node/beacon_chain/tests/events.rs

@@ -1,5 +1,6 @@
 use arbitrary::Arbitrary;
 use beacon_chain::data_column_verification::GossipVerifiedDataColumn;
+use beacon_chain::inclusion_list_verification::InclusionListSource;
 use beacon_chain::payload_envelope_verification::EnvelopeSource;
 use beacon_chain::test_utils::{
     BeaconChainHarness, fork_name_from_env, generate_data_column_sidecars_from_block, test_spec,
@@ -554,6 +555,7 @@ async fn inclusion_list_event_on_import() {
             .verify_inclusion_list_for_gossip(
                 signed.clone(),
                 harness.chain.slot_clock.now_duration().unwrap(),
+                InclusionListSource::Gossip,
             )
             .expect("verification should succeed");
         harness.chain.import_inclusion_list(verified);

beacon_node/beacon_processor/src/lib.rs

@@ -435,6 +435,7 @@ pub enum Work<E: EthSpec> {
     },
     RpcCustodyColumn(AsyncFn),
     RpcEnvelope(AsyncFn),
+    RpcInclusionLists(AsyncFn),
     ColumnReconstruction(AsyncFn),
     IgnoredRpcBlock {
         process_fn: BlockingFn,
@@ -506,6 +507,7 @@ pub enum WorkType {
     RpcBlobs,
     RpcCustodyColumn,
     RpcEnvelope,
+    RpcInclusionLists,
     ColumnReconstruction,
     IgnoredRpcBlock,
     ChainSegment,
@@ -574,6 +576,7 @@ impl<E: EthSpec> Work<E> {
             Work::RpcBlobs { .. } => WorkType::RpcBlobs,
             Work::RpcCustodyColumn { .. } => WorkType::RpcCustodyColumn,
             Work::RpcEnvelope(_) => WorkType::RpcEnvelope,
+            Work::RpcInclusionLists(_) => WorkType::RpcInclusionLists,
             Work::ColumnReconstruction(_) => WorkType::ColumnReconstruction,
             Work::IgnoredRpcBlock { .. } => WorkType::IgnoredRpcBlock,
             Work::ChainSegment { .. } => WorkType::ChainSegment,
@@ -1004,6 +1007,8 @@ impl<E: EthSpec> BeaconProcessor<E> {
                         // Check inclusion lists next, they also influence fork choice.
                         } else if let Some(item) = work_queues.gossip_inclusion_list_queue.pop() {
                             Some(item)
+                        } else if let Some(item) = work_queues.rpc_inclusion_lists_queue.pop() {
+                            Some(item)
                         // Check sync committee messages after attestations as their rewards are lesser
                         // and they don't influence fork choice.
                         } else if let Some(item) = work_queues.sync_contribution_queue.pop() {
@@ -1234,6 +1239,9 @@ impl<E: EthSpec> BeaconProcessor<E> {
                             Work::RpcEnvelope(_) => {
                                 work_queues.rpc_envelope_queue.push(work, work_id)
                             }
+                            Work::RpcInclusionLists(_) => {
+                                work_queues.rpc_inclusion_lists_queue.push(work, work_id)
+                            }
                             Work::RpcCustodyColumn { .. } => {
                                 work_queues.rpc_custody_column_queue.push(work, work_id)
                             }
@@ -1390,6 +1398,7 @@ impl<E: EthSpec> BeaconProcessor<E> {
                             work_queues.rpc_blob_queue.len()
                         }
                         WorkType::RpcEnvelope => work_queues.rpc_envelope_queue.len(),
+                        WorkType::RpcInclusionLists => work_queues.rpc_inclusion_lists_queue.len(),
                         WorkType::RpcCustodyColumn => work_queues.rpc_custody_column_queue.len(),
                         WorkType::ColumnReconstruction => {
                             work_queues.column_reconstruction_queue.len()
@@ -1595,6 +1604,7 @@ impl<E: EthSpec> BeaconProcessor<E> {
             | Work::RpcBlobs { process_fn }
             | Work::RpcCustodyColumn(process_fn)
             | Work::RpcEnvelope(process_fn)
+            | Work::RpcInclusionLists(process_fn)
             | Work::ColumnReconstruction(process_fn) => task_spawner.spawn_async(process_fn),
             Work::IgnoredRpcBlock { process_fn } => task_spawner.spawn_blocking(process_fn),
             Work::GossipBlock(work)

beacon_node/beacon_processor/src/scheduler/work_queue.rs

@@ -123,6 +123,7 @@ pub struct BeaconProcessorQueueLengths {
     rpc_blob_queue: usize,
     rpc_custody_column_queue: usize,
     rpc_envelope_queue: usize,
+    rpc_inclusion_lists_queue: usize,
     column_reconstruction_queue: usize,
     chain_segment_queue: usize,
     backfill_chain_segment: usize,
@@ -206,6 +207,7 @@ impl BeaconProcessorQueueLengths {
             rpc_custody_column_queue: 64,
             // Bounded by `PARENT_DEPTH_TOLERANCE`; one envelope per Gloas block.
             rpc_envelope_queue: 1024,
+            rpc_inclusion_lists_queue: 64,
             column_reconstruction_queue: 1,
             chain_segment_queue: 64,
             backfill_chain_segment: 64,
@@ -269,6 +271,7 @@ pub struct WorkQueues<E: EthSpec> {
     pub rpc_blob_queue: FifoQueue<Work<E>>,
     pub rpc_custody_column_queue: FifoQueue<Work<E>>,
     pub rpc_envelope_queue: FifoQueue<Work<E>>,
+    pub rpc_inclusion_lists_queue: FifoQueue<Work<E>>,
     pub column_reconstruction_queue: LifoQueue<Work<E>>,
     pub chain_segment_queue: FifoQueue<Work<E>>,
     pub backfill_chain_segment: FifoQueue<Work<E>>,
@@ -345,6 +348,7 @@ impl<E: EthSpec> WorkQueues<E> {
         let rpc_blob_queue = FifoQueue::new(queue_lengths.rpc_blob_queue);
         let rpc_custody_column_queue = FifoQueue::new(queue_lengths.rpc_custody_column_queue);
         let rpc_envelope_queue = FifoQueue::new(queue_lengths.rpc_envelope_queue);
+        let rpc_inclusion_lists_queue = FifoQueue::new(queue_lengths.rpc_inclusion_lists_queue);
         let column_reconstruction_queue = LifoQueue::new(queue_lengths.column_reconstruction_queue);
         let chain_segment_queue = FifoQueue::new(queue_lengths.chain_segment_queue);
         let backfill_chain_segment = FifoQueue::new(queue_lengths.backfill_chain_segment);
@@ -418,6 +422,7 @@ impl<E: EthSpec> WorkQueues<E> {
             rpc_blob_queue,
             rpc_custody_column_queue,
             rpc_envelope_queue,
+            rpc_inclusion_lists_queue,
             chain_segment_queue,
             column_reconstruction_queue,
             backfill_chain_segment,