Introduce AttestationMatchingGroupV2

ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/attestation/MatchingDataAttestationGroup.java

@@ -178,8 +178,8 @@ public int size() {
   }
 
   /**
-   * Updates {@code seenAggregationBits} and removes any attestation from this group whose
-   * aggregation bits have all been seen.
+   * Updates includedValidators bits and removes any attestation from this group whose aggregation
+   * bits have all been seen.
    *
    * <p>This is well suited for removing attestations that have been included in a block.
    *

ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/attestation/utils/AttestationBits.java

@@ -15,6 +15,7 @@
 
 import it.unimi.dsi.fastutil.ints.Int2IntMap;
 import java.util.Optional;
+import java.util.stream.IntStream;
 import tech.pegasys.teku.infrastructure.ssz.collections.SszBitlist;
 import tech.pegasys.teku.infrastructure.ssz.collections.SszBitvector;
 import tech.pegasys.teku.spec.datastructures.attestation.ValidatableAttestation;
@@ -80,6 +81,12 @@ static AttestationBits of(
 
   boolean isExclusivelyFromCommittee(int committeeIndex);
 
+  boolean isFromCommittee(int committeeIndex);
+
+  int getFirstCommitteeIndex();
+
+  IntStream streamCommitteeIndices();
+
   /** Creates an independent copy of this instance */
   AttestationBits copy();
 }

ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/attestation/utils/AttestationBitsElectra.java

@@ -22,6 +22,7 @@
 import java.util.BitSet;
 import java.util.List;
 import java.util.Objects;
+import java.util.stream.IntStream;
 import tech.pegasys.teku.infrastructure.ssz.collections.SszBitlist;
 import tech.pegasys.teku.infrastructure.ssz.collections.SszBitvector;
 import tech.pegasys.teku.infrastructure.ssz.schema.collections.SszBitlistSchema;
@@ -329,6 +330,21 @@ public boolean isExclusivelyFromCommittee(final int committeeIndex) {
         && committeeAggregationBitsMap.containsKey(committeeIndex);
   }
 
+  @Override
+  public boolean isFromCommittee(final int committeeIndex) {
+    return committeeAggregationBitsMap.containsKey(committeeIndex);
+  }
+
+  @Override
+  public int getFirstCommitteeIndex() {
+    return committeeBits.nextSetBit(0);
+  }
+
+  @Override
+  public IntStream streamCommitteeIndices() {
+    return committeeBits.stream();
+  }
+
   @Override
   public String toString() {
     long totalSetBits = 0;

ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/attestation/utils/AttestationBitsPhase0.java

@@ -16,6 +16,7 @@
 import com.google.common.base.MoreObjects;
 import it.unimi.dsi.fastutil.ints.Int2IntMap;
 import java.util.Objects;
+import java.util.stream.IntStream;
 import tech.pegasys.teku.infrastructure.ssz.collections.SszBitlist;
 import tech.pegasys.teku.infrastructure.ssz.collections.SszBitvector;
 import tech.pegasys.teku.spec.datastructures.operations.Attestation;
@@ -99,6 +100,21 @@ public boolean isExclusivelyFromCommittee(final int committeeIndex) {
     throw new IllegalStateException("Committee bits not available in phase0");
   }
 
+  @Override
+  public boolean isFromCommittee(final int committeeIndex) {
+    throw new IllegalStateException("Committee bits not available in phase0");
+  }
+
+  @Override
+  public int getFirstCommitteeIndex() {
+    throw new IllegalStateException("Committee bits not available in phase0");
+  }
+
+  @Override
+  public IntStream streamCommitteeIndices() {
+    throw new IllegalStateException("Committee bits not available in phase0");
+  }
+
   @Override
   public String toString() {
     return MoreObjects.toStringHelper(this).add("aggregationBits", aggregationBits).toString();

ethereum/statetransition/src/main/java/tech/pegasys/teku/statetransition/attestation/utils/TimeLimitingIterator.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright Consensys Software Inc., 2025
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package tech.pegasys.teku.statetransition.attestation.utils;
+
+import java.util.Iterator;
+import java.util.function.LongConsumer;
+import java.util.function.LongSupplier;
+
+public record TimeLimitingIterator<T>(
+    LongSupplier nanosSupplier, long timeLimitNanos, Iterator<T> delegate, LongConsumer onTimeLimit)
+    implements Iterator<T> {
+
+  @Override
+  public boolean hasNext() {
+    if (nanosSupplier.getAsLong() <= timeLimitNanos) {
+      return delegate.hasNext();
+    }
+
+    onTimeLimit.accept(timeLimitNanos);
+
+    return false;
+  }
+
+  @Override
+  public T next() {
+    return delegate.next();
+  }
+}

ethereum/statetransition/src/test/java/tech/pegasys/teku/statetransition/attestation/AggregatingAttestationPoolTest.java

@@ -206,7 +206,7 @@ public void createAggregateFor_shouldAggregateAttestationsWithMatchingData() {
 
     final Optional<Attestation> result =
         aggregatingPool.createAggregateFor(attestationData.hashTreeRoot(), committeeIndex);
-    assertThat(result).contains(aggregateAttestations(attestation1, attestation2));
+    assertThat(result).contains(aggregateAttestations(committeeSizes, attestation1, attestation2));
   }
 
   @TestTemplate
@@ -218,7 +218,7 @@ public void createAggregateFor_shouldReturnBestAggregateForMatchingDataWhenSomeO
 
     final Optional<Attestation> result =
         aggregatingPool.createAggregateFor(attestationData.hashTreeRoot(), committeeIndex);
-    assertThat(result).contains(aggregateAttestations(attestation1, attestation2));
+    assertThat(result).contains(aggregateAttestations(committeeSizes, attestation1, attestation2));
   }
 
   @TestTemplate
@@ -291,7 +291,7 @@ public void getAttestationsForBlock_shouldAggregateAttestationsWhenPossible() {
     final BeaconState stateAtBlockSlot = dataStructureUtil.randomBeaconState(SLOT.increment());
 
     assertThat(aggregatingPool.getAttestationsForBlock(stateAtBlockSlot, forkChecker))
-        .containsExactly(aggregateAttestations(attestation1, attestation2));
+        .containsExactly(aggregateAttestations(committeeSizes, attestation1, attestation2));
   }
 
   @TestTemplate
@@ -305,7 +305,8 @@ public void getAttestationsForBlock_shouldIncludeAttestationsWithDifferentData()
     final BeaconState stateAtBlockSlot = dataStructureUtil.randomBeaconState(ONE);
 
     assertThat(aggregatingPool.getAttestationsForBlock(stateAtBlockSlot, forkChecker))
-        .containsExactlyInAnyOrder(aggregateAttestations(attestation1, attestation2), attestation3);
+        .containsExactlyInAnyOrder(
+            aggregateAttestations(committeeSizes, attestation1, attestation2), attestation3);
   }
 
   @TestTemplate

ethereum/statetransition/src/test/java/tech/pegasys/teku/statetransition/attestation/MatchingDataAttestationGroupTest.java

@@ -120,7 +120,8 @@ public void remove_shouldRemoveAttestationsThatAreAggregatedIntoRemovedAttestati
     int numRemoved =
         group.onAttestationIncludedInBlock(
             UInt64.ZERO,
-            aggregateAttestations(toAttestation(attestation1), toAttestation(attestation2)));
+            aggregateAttestations(
+                committeeSizes, toAttestation(attestation1), toAttestation(attestation2)));
 
     assertThat(group.stream(Optional.of(UInt64.ZERO)))
         .containsExactly(toPooledAttestationWithData(attestation3));
@@ -155,7 +156,8 @@ public void add_shouldAggregateAttestationsFromSameCommittee(final SpecContext s
         .containsExactly(toPooledAttestationWithData(attestation1));
 
     final Attestation expected =
-        aggregateAttestations(toAttestation(attestation2), toAttestation(attestation3));
+        aggregateAttestations(
+            committeeSizes, toAttestation(attestation2), toAttestation(attestation3));
 
     assertThat(group.stream(Optional.of(UInt64.ONE)))
         .containsExactly(
@@ -184,7 +186,8 @@ public void iterator_shouldAggregateAttestationsWhereValidatorsDoNotOverlap() {
     final PooledAttestation attestation2 = addPooledAttestation(2);
 
     final Attestation expected =
-        aggregateAttestations(toAttestation(attestation1), toAttestation(attestation2));
+        aggregateAttestations(
+            committeeSizes, toAttestation(attestation1), toAttestation(attestation2));
 
     assertThat(group.stream(Optional.of(UInt64.ZERO)))
         .containsExactlyInAnyOrder(
@@ -205,7 +208,9 @@ public void iterator_shouldAggregateAttestationsWithMoreValidatorsFirst() {
                 ValidatableAttestation.from(
                     spec,
                     aggregateAttestations(
-                        toAttestation(bigAttestation), toAttestation(littleAttestation)),
+                        committeeSizes,
+                        toAttestation(bigAttestation),
+                        toAttestation(littleAttestation)),
                     committeeSizes)),
             mediumAttestation);
   }
@@ -250,7 +255,8 @@ void iterator_shouldOmitAttestationsThatOverlapWithFirstAttestationAndAreRedunda
             PooledAttestation.fromValidatableAttestation(
                 ValidatableAttestation.from(
                     spec,
-                    aggregateAttestations(toAttestation(useful1), toAttestation(useful2)),
+                    aggregateAttestations(
+                        committeeSizes, toAttestation(useful1), toAttestation(useful2)),
                     committeeSizes)));
 
     assertThat(group.stream(Optional.of(UInt64.ZERO))).containsExactly(expected);
@@ -363,6 +369,7 @@ public void size() {
         group.onAttestationIncludedInBlock(
             UInt64.ZERO,
             aggregateAttestations(
+                committeeSizes,
                 attestation1.toAttestation(attestationSchema),
                 attestation2.toAttestation(attestationSchema)));