7549 aggregating pool filter phase0 att in electra

ethereum/spec/src/main/java/tech/pegasys/teku/spec/datastructures/attestation/ValidatableAttestation.java

@@ -154,7 +154,7 @@ public void saveCommitteeShufflingSeedAndCommitteesSize(final BeaconState state)
     saveCommitteeShufflingSeed(state);
     // The committees size is only required when the committee_bits field is present in the
     // Attestation
-    if (attestation.getCommitteeBits().isPresent()) {
+    if (attestation.requiresCommitteeBits()) {
       saveCommitteesSize(state);
     }
   }

ethereum/spec/src/main/java/tech/pegasys/teku/spec/datastructures/operations/Attestation.java

@@ -62,4 +62,6 @@ default List<UInt64> getCommitteeIndicesRequired() {
     return getCommitteeIndices()
         .orElseThrow(() -> new IllegalArgumentException("Missing committee indices"));
   }
+
+  boolean requiresCommitteeBits();
 }

ethereum/spec/src/main/java/tech/pegasys/teku/spec/datastructures/operations/AttestationSchema.java

@@ -53,4 +53,6 @@ default Optional<AttestationElectraSchema> toVersionElectra() {
   SszBitlistSchema<?> getAggregationBitsSchema();
 
   Optional<SszBitvectorSchema<?>> getCommitteeBitsSchema();
+
+  boolean requiresCommitteeBits();
 }

ethereum/spec/src/main/java/tech/pegasys/teku/spec/datastructures/operations/versions/electra/AttestationElectra.java

@@ -86,4 +86,9 @@ public Optional<List<UInt64>> getCommitteeIndices() {
     return Optional.of(
         getCommitteeBitsRequired().getAllSetBits().intStream().mapToObj(UInt64::valueOf).toList());
   }
+
+  @Override
+  public boolean requiresCommitteeBits() {
+    return true;
+  }
 }

ethereum/spec/src/main/java/tech/pegasys/teku/spec/datastructures/operations/versions/electra/AttestationElectraSchema.java

@@ -83,4 +83,9 @@ public AttestationElectra create(
   public Optional<AttestationElectraSchema> toVersionElectra() {
     return Optional.of(this);
   }
+
+  @Override
+  public boolean requiresCommitteeBits() {
+    return true;
+  }
 }

ethereum/spec/src/main/java/tech/pegasys/teku/spec/datastructures/operations/versions/phase0/AttestationPhase0.java

@@ -71,4 +71,9 @@ public AttestationData getData() {
   public BLSSignature getAggregateSignature() {
     return getField2().getSignature();
   }
+
+  @Override
+  public boolean requiresCommitteeBits() {
+    return false;
+  }
 }

ethereum/spec/src/main/java/tech/pegasys/teku/spec/datastructures/operations/versions/phase0/AttestationPhase0Schema.java

@@ -63,4 +63,9 @@ public Attestation create(
       final BLSSignature signature) {
     return new AttestationPhase0(this, aggregationBits, data, signature);
   }
+
+  @Override
+  public boolean requiresCommitteeBits() {
+    return false;
+  }
 }

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

@@ -41,6 +41,7 @@
 import tech.pegasys.teku.spec.datastructures.operations.Attestation;
 import tech.pegasys.teku.spec.datastructures.operations.AttestationData;
 import tech.pegasys.teku.spec.datastructures.state.beaconstate.BeaconState;
+import tech.pegasys.teku.spec.schemas.SchemaDefinitions;
 
 /**
  * Maintains a pool of attestations. Attestations can be retrieved either for inclusion in a block
@@ -177,11 +178,14 @@ public synchronized SszList<Attestation> getAttestationsForBlock(
     final UInt64 currentEpoch = spec.getCurrentEpoch(stateAtBlockSlot);
     final int previousEpochLimit = spec.getPreviousEpochAttestationCapacity(stateAtBlockSlot);
 
+    final SchemaDefinitions schemaDefinitions =
+        spec.atSlot(stateAtBlockSlot.getSlot()).getSchemaDefinitions();
+
     final SszListSchema<Attestation, ?> attestationsSchema =
-        spec.atSlot(stateAtBlockSlot.getSlot())
-            .getSchemaDefinitions()
-            .getBeaconBlockBodySchema()
-            .getAttestationsSchema();
+        schemaDefinitions.getBeaconBlockBodySchema().getAttestationsSchema();
+
+    final boolean blockRequiresAttestationsWithCommitteeBits =
+        schemaDefinitions.getAttestationSchema().requiresCommitteeBits();
 
     final AtomicInteger prevEpochCount = new AtomicInteger(0);
     return dataHashBySlot
@@ -196,11 +200,15 @@ public synchronized SszList<Attestation> getAttestationsForBlock(
         .filter(group -> isValid(stateAtBlockSlot, group.getAttestationData()))
         .filter(forkChecker::areAttestationsFromCorrectFork)
         .flatMap(MatchingDataAttestationGroup::stream)
-        .limit(attestationsSchema.getMaxLength())
         .map(ValidatableAttestation::getAttestation)
         .filter(
-            att -> {
-              if (spec.computeEpochAtSlot(att.getData().getSlot()).isLessThan(currentEpoch)) {
+            attestation ->
+                attestation.requiresCommitteeBits() == blockRequiresAttestationsWithCommitteeBits)
+        .limit(attestationsSchema.getMaxLength())
+        .filter(
+            attestation -> {
+              if (spec.computeEpochAtSlot(attestation.getData().getSlot())
+                  .isLessThan(currentEpoch)) {
                 final int currentCount = prevEpochCount.getAndIncrement();
                 return currentCount < previousEpochLimit;
               }