Ensure /eth/v2/beacon/pool/attestations honors committee_index

beacon_node/http_api/src/lib.rs

@@ -85,13 +85,14 @@ use tokio_stream::{
     wrappers::{errors::BroadcastStreamRecvError, BroadcastStream},
     StreamExt,
 };
+use types::AttestationData;
 use types::{
-    fork_versioned_response::EmptyMetadata, Attestation, AttestationData, AttestationShufflingId,
-    AttesterSlashing, BeaconStateError, ChainSpec, Checkpoint, CommitteeCache, ConfigAndPreset,
-    Epoch, EthSpec, ForkName, ForkVersionedResponse, Hash256, ProposerPreparationData,
-    ProposerSlashing, RelativeEpoch, SignedAggregateAndProof, SignedBlindedBeaconBlock,
-    SignedBlsToExecutionChange, SignedContributionAndProof, SignedValidatorRegistrationData,
-    SignedVoluntaryExit, Slot, SyncCommitteeMessage, SyncContributionData,
+    fork_versioned_response::EmptyMetadata, Attestation, AttestationShufflingId, AttesterSlashing,
+    BeaconStateError, ChainSpec, Checkpoint, CommitteeCache, ConfigAndPreset, Epoch, EthSpec,
+    ForkName, ForkVersionedResponse, Hash256, ProposerPreparationData, ProposerSlashing,
+    RelativeEpoch, SignedAggregateAndProof, SignedBlindedBeaconBlock, SignedBlsToExecutionChange,
+    SignedContributionAndProof, SignedValidatorRegistrationData, SignedVoluntaryExit, Slot,
+    SyncCommitteeMessage, SyncContributionData,
 };
 use validator::pubkey_to_validator_index;
 use version::{
@@ -1999,11 +2000,14 @@ pub fn serve<T: BeaconChainTypes>(
              chain: Arc<BeaconChain<T>>,
              query: api_types::AttestationPoolQuery| {
                 task_spawner.blocking_response_task(Priority::P1, move || {
-                    let query_filter = |data: &AttestationData| {
+                    let query_filter = |data: &AttestationData, committee_index: Option<u64>| {
                         query.slot.is_none_or(|slot| slot == data.slot)
-                            && query
-                                .committee_index
-                                .is_none_or(|index| index == data.index)
+                            && query.committee_index.is_none_or(|index| {
+                                if let Some(committee_index) = committee_index {
+                                    return index == committee_index;
+                                }
+                                false
+                            })
                     };
 
                     let mut attestations = chain.op_pool.get_filtered_attestations(query_filter);
@@ -2012,7 +2016,7 @@ pub fn serve<T: BeaconChainTypes>(
                             .naive_aggregation_pool
                             .read()
                             .iter()
-                            .filter(|&att| query_filter(att.data()))
+                            .filter(|&att| query_filter(att.data(), att.committee_index()))
                             .cloned(),
                     );
                     // Use the current slot to find the fork version, and convert all messages to the

beacon_node/http_api/tests/tests.rs

@@ -2111,8 +2111,47 @@ impl ApiTester {
             .await
             .unwrap()
             .data;
+
         assert_eq!(result, expected);
 
+        let result_committee_index_filtered = self
+            .client
+            .get_beacon_pool_attestations_v1(None, Some(1))
+            .await
+            .unwrap()
+            .data;
+
+        let expected_committee_index_filtered = expected
+            .clone()
+            .into_iter()
+            .filter(|att| att.committee_index() == Some(1))
+            .collect::<Vec<_>>();
+
+        assert_eq!(
+            result_committee_index_filtered,
+            expected_committee_index_filtered
+        );
+        assert_ne!(result_committee_index_filtered, expected);
+
+        let result_committee_index_filtered = self
+            .client
+            .get_beacon_pool_attestations_v2(None, Some(2))
+            .await
+            .unwrap()
+            .data;
+
+        let expected_committee_index_filtered = expected
+            .clone()
+            .into_iter()
+            .filter(|att| att.committee_index() == Some(2))
+            .collect::<Vec<_>>();
+
+        assert_eq!(
+            result_committee_index_filtered,
+            expected_committee_index_filtered
+        );
+        assert_ne!(result_committee_index_filtered, expected);
+
         self
     }
 

beacon_node/operation_pool/src/attestation_storage.rs

@@ -119,6 +119,18 @@ impl<E: EthSpec> CompactAttestationRef<'_, E> {
         }
     }
 
+    pub fn committee_index(&self) -> Option<u64> {
+        match self.indexed {
+            CompactIndexedAttestation::Base(_) => Some(self.data.index),
+            CompactIndexedAttestation::Electra(indexed_att) => indexed_att
+                .committee_bits
+                .iter()
+                .enumerate()
+                .find(|&(_, bit)| bit)
+                .map(|(index, _)| index as u64),
+        }
+    }
+
     pub fn clone_as_attestation(&self) -> Attestation<E> {
         match self.indexed {
             CompactIndexedAttestation::Base(indexed_att) => Attestation::Base(AttestationBase {
@@ -268,7 +280,11 @@ impl<E: EthSpec> CompactIndexedAttestationElectra<E> {
     }
 
     pub fn committee_index(&self) -> Option<u64> {
-        self.get_committee_indices().first().copied()
+        self.committee_bits
+            .iter()
+            .enumerate()
+            .find(|&(_, bit)| bit)
+            .map(|(index, _)| index as u64)
     }
 
     pub fn get_committee_indices(&self) -> Vec<u64> {

beacon_node/operation_pool/src/lib.rs

@@ -673,12 +673,12 @@ impl<E: EthSpec> OperationPool<E> {
     /// This method may return objects that are invalid for block inclusion.
     pub fn get_filtered_attestations<F>(&self, filter: F) -> Vec<Attestation<E>>
     where
-        F: Fn(&AttestationData) -> bool,
+        F: Fn(&AttestationData, Option<u64>) -> bool,
     {
         self.attestations
             .read()
             .iter()
-            .filter(|att| filter(&att.attestation_data()))
+            .filter(|att| filter(&att.attestation_data(), att.committee_index()))
             .map(|att| att.clone_as_attestation())
             .collect()
     }

consensus/types/src/attestation.rs

@@ -292,7 +292,11 @@ impl<E: EthSpec> AttestationRef<'_, E> {
 
 impl<E: EthSpec> AttestationElectra<E> {
     pub fn committee_index(&self) -> Option<u64> {
-        self.get_committee_indices().first().cloned()
+        self.committee_bits
+            .iter()
+            .enumerate()
+            .find(|&(_, bit)| bit)
+            .map(|(index, _)| index as u64)
     }
 
     pub fn get_aggregation_bits(&self) -> Vec<u64> {