Implements SingleAttestation handling

beacon/sync/src/main/java/tech/pegasys/teku/beacon/sync/forward/multipeer/chains/ThrottlingSyncSource.java

@@ -65,6 +65,10 @@ public SafeFuture<Void> requestBlocksByRange(
       LOG.debug("Sending request for {} blocks", count);
       return delegate.requestBlocksByRange(startSlot, count, listener);
     } else {
+      LOG.debug(
+          "Rate limiting request for {} blocks. Retry in {} seconds",
+          count,
+          PEER_REQUEST_DELAY.toSeconds());
       return asyncRunner.runAfterDelay(
           () -> requestBlocksByRange(startSlot, count, listener), PEER_REQUEST_DELAY);
     }
@@ -77,6 +81,10 @@ public SafeFuture<Void> requestBlobSidecarsByRange(
       LOG.debug("Sending request for {} blob sidecars", count);
       return delegate.requestBlobSidecarsByRange(startSlot, count, listener);
     } else {
+      LOG.debug(
+          "Rate limiting request for {} blob sidecars. Retry in {} seconds",
+          count,
+          PEER_REQUEST_DELAY.toSeconds());
       return asyncRunner.runAfterDelay(
           () -> requestBlobSidecarsByRange(startSlot, count, listener), PEER_REQUEST_DELAY);
     }

data/beaconrestapi/src/main/java/tech/pegasys/teku/beaconrestapi/handlers/v1/events/EventSubscriptionManager.java

@@ -43,6 +43,7 @@
 import tech.pegasys.teku.spec.datastructures.attestation.ValidatableAttestation;
 import tech.pegasys.teku.spec.datastructures.blobs.versions.deneb.BlobSidecar;
 import tech.pegasys.teku.spec.datastructures.blocks.SignedBeaconBlock;
+import tech.pegasys.teku.spec.datastructures.operations.Attestation;
 import tech.pegasys.teku.spec.datastructures.operations.AttesterSlashing;
 import tech.pegasys.teku.spec.datastructures.operations.ProposerSlashing;
 import tech.pegasys.teku.spec.datastructures.operations.SignedBlsToExecutionChange;
@@ -211,12 +212,13 @@ protected void onSyncCommitteeContribution(
   }
 
   protected void onNewAttestation(final ValidatableAttestation attestation) {
-    if (!attestation.getAttestation().isSingleAttestation()) {
-      final AttestationEvent attestationEvent = new AttestationEvent(attestation.getAttestation());
+    final Attestation actualAttestation = attestation.getUnconvertedAttestation();
+    if (!actualAttestation.isSingleAttestation()) {
+      final AttestationEvent attestationEvent = new AttestationEvent(actualAttestation);
       notifySubscribersOfEvent(EventType.attestation, attestationEvent);
     } else {
       final SingleAttestationEvent attestationEvent =
-          new SingleAttestationEvent(attestation.getAttestation().toSingleAttestationRequired());
+          new SingleAttestationEvent(actualAttestation.toSingleAttestationRequired());
       notifySubscribersOfEvent(EventType.single_attestation, attestationEvent);
     }
   }

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

@@ -13,6 +13,8 @@
 
 package tech.pegasys.teku.spec.datastructures.attestation;
 
+import static com.google.common.base.Preconditions.checkState;
+
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.MoreObjects;
 import com.google.common.base.Objects;
@@ -24,6 +26,7 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Supplier;
 import org.apache.tuweni.bytes.Bytes32;
+import org.jetbrains.annotations.NotNull;
 import tech.pegasys.teku.infrastructure.unsigned.UInt64;
 import tech.pegasys.teku.spec.Spec;
 import tech.pegasys.teku.spec.constants.Domain;
@@ -35,20 +38,32 @@
 
 public class ValidatableAttestation {
   private final Spec spec;
-  private final Attestation attestation;
   private final Optional<SignedAggregateAndProof> maybeAggregate;
   private final Supplier<Bytes32> hashTreeRoot;
   private final AtomicBoolean gossiped = new AtomicBoolean(false);
   private final boolean producedLocally;
   private final OptionalInt receivedSubnetId;
 
+  private final Attestation unconvertedAttestation;
+
+  @NotNull // will help us not forget to initialize if a new constructor is added
+  private volatile Attestation attestation;
+
   private volatile boolean isValidIndexedAttestation = false;
   private volatile boolean acceptedAsGossip = false;
 
   private volatile Optional<IndexedAttestation> indexedAttestation = Optional.empty();
   private volatile Optional<Bytes32> committeeShufflingSeed = Optional.empty();
   private volatile Optional<Int2IntMap> committeesSize = Optional.empty();
 
+  public void convertToAggregatedFormatFromSingleAttestation(
+      final Attestation aggregatedFormatFromSingleAttestation) {
+    checkState(
+        attestation.isSingleAttestation(),
+        "Attestation must be a single attestation to convert to aggregated format");
+    this.attestation = aggregatedFormatFromSingleAttestation;
+  }
+
   public static ValidatableAttestation from(final Spec spec, final Attestation attestation) {
     return new ValidatableAttestation(
         spec, attestation, Optional.empty(), OptionalInt.empty(), false);
@@ -113,6 +128,7 @@ private ValidatableAttestation(
     this.spec = spec;
     this.maybeAggregate = aggregateAndProof;
     this.attestation = attestation;
+    this.unconvertedAttestation = attestation;
     this.receivedSubnetId = receivedSubnetId;
     this.hashTreeRoot = Suppliers.memoize(attestation::hashTreeRoot);
     this.producedLocally = producedLocally;
@@ -128,6 +144,7 @@ private ValidatableAttestation(
     this.spec = spec;
     this.maybeAggregate = aggregateAndProof;
     this.attestation = attestation;
+    this.unconvertedAttestation = attestation;
     this.receivedSubnetId = receivedSubnetId;
     this.hashTreeRoot = Suppliers.memoize(attestation::hashTreeRoot);
     this.producedLocally = producedLocally;
@@ -178,7 +195,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.requiresCommitteeBits()) {
+    if (attestation.isSingleAttestation() || attestation.requiresCommitteeBits()) {
       saveCommitteesSize(state);
     }
   }
@@ -216,10 +233,15 @@ public boolean isAggregate() {
     return maybeAggregate.isPresent();
   }
 
+  @NotNull
   public Attestation getAttestation() {
     return attestation;
   }
 
+  public Attestation getUnconvertedAttestation() {
+    return unconvertedAttestation;
+  }
+
   public SignedAggregateAndProof getSignedAggregateAndProof() {
     return maybeAggregate.orElseThrow(
         () -> new UnsupportedOperationException("ValidatableAttestation is not an aggregate."));
@@ -246,10 +268,9 @@ public boolean equals(final Object o) {
     if (this == o) {
       return true;
     }
-    if (!(o instanceof ValidatableAttestation)) {
+    if (!(o instanceof ValidatableAttestation that)) {
       return false;
     }
-    ValidatableAttestation that = (ValidatableAttestation) o;
     return Objects.equal(getAttestation(), that.getAttestation())
         && Objects.equal(maybeAggregate, that.maybeAggregate);
   }
@@ -273,6 +294,7 @@ public String toString() {
         .add("committeeShufflingSeed", committeeShufflingSeed)
         .add("committeesSize", committeesSize)
         .add("receivedSubnetId", receivedSubnetId)
+        .add("unconvertedAttestation", unconvertedAttestation)
         .toString();
   }
 }

ethereum/spec/src/main/java/tech/pegasys/teku/spec/logic/versions/electra/util/AttestationUtilElectra.java

@@ -13,23 +13,45 @@
 
 package tech.pegasys.teku.spec.logic.versions.electra.util;
 
+import static com.google.common.base.Preconditions.checkArgument;
+import static tech.pegasys.teku.infrastructure.async.SafeFuture.completedFuture;
+
 import it.unimi.dsi.fastutil.ints.IntArrayList;
 import it.unimi.dsi.fastutil.ints.IntList;
 import java.util.List;
+import java.util.Optional;
 import java.util.stream.IntStream;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.apache.tuweni.bytes.Bytes;
+import org.apache.tuweni.bytes.Bytes32;
+import tech.pegasys.teku.bls.BLSPublicKey;
+import tech.pegasys.teku.bls.BLSSignature;
+import tech.pegasys.teku.infrastructure.async.SafeFuture;
 import tech.pegasys.teku.infrastructure.ssz.collections.SszBitlist;
 import tech.pegasys.teku.infrastructure.unsigned.UInt64;
 import tech.pegasys.teku.spec.config.SpecConfig;
+import tech.pegasys.teku.spec.constants.Domain;
+import tech.pegasys.teku.spec.datastructures.attestation.ValidatableAttestation;
 import tech.pegasys.teku.spec.datastructures.blocks.BeaconBlockSummary;
 import tech.pegasys.teku.spec.datastructures.operations.Attestation;
 import tech.pegasys.teku.spec.datastructures.operations.AttestationData;
+import tech.pegasys.teku.spec.datastructures.operations.IndexedAttestation;
+import tech.pegasys.teku.spec.datastructures.operations.IndexedAttestationSchema;
+import tech.pegasys.teku.spec.datastructures.operations.SingleAttestation;
+import tech.pegasys.teku.spec.datastructures.operations.versions.electra.AttestationElectraSchema;
+import tech.pegasys.teku.spec.datastructures.state.Fork;
 import tech.pegasys.teku.spec.datastructures.state.beaconstate.BeaconState;
+import tech.pegasys.teku.spec.datastructures.util.AttestationProcessingResult;
 import tech.pegasys.teku.spec.logic.common.helpers.BeaconStateAccessors;
 import tech.pegasys.teku.spec.logic.common.helpers.MiscHelpers;
+import tech.pegasys.teku.spec.logic.common.util.AsyncBLSSignatureVerifier;
 import tech.pegasys.teku.spec.logic.versions.deneb.util.AttestationUtilDeneb;
 import tech.pegasys.teku.spec.schemas.SchemaDefinitions;
 
 public class AttestationUtilElectra extends AttestationUtilDeneb {
+  private static final Logger LOG = LogManager.getLogger();
+
   public AttestationUtilElectra(
       final SpecConfig specConfig,
       final SchemaDefinitions schemaDefinitions,
@@ -93,4 +115,162 @@ public AttestationData getGenericAttestationData(
       final UInt64 committeeIndex) {
     return super.getGenericAttestationData(slot, state, block, UInt64.ZERO);
   }
+
+  @Override
+  public IndexedAttestation getIndexedAttestation(
+      final BeaconState state, final Attestation attestation) {
+    if (attestation.isSingleAttestation()) {
+      return getIndexedAttestationFromSingleAttestation(attestation.toSingleAttestationRequired());
+    }
+    return super.getIndexedAttestation(state, attestation);
+  }
+
+  private IndexedAttestation getIndexedAttestationFromSingleAttestation(
+      final SingleAttestation attestation) {
+    final IndexedAttestationSchema indexedAttestationSchema =
+        schemaDefinitions.getIndexedAttestationSchema();
+
+    return indexedAttestationSchema.create(
+        indexedAttestationSchema
+            .getAttestingIndicesSchema()
+            .of(attestation.getValidatorIndexRequired()),
+        attestation.getData(),
+        attestation.getSignature());
+  }
+
+  @Override
+  public SafeFuture<AttestationProcessingResult> isValidIndexedAttestationAsync(
+      final Fork fork,
+      final BeaconState state,
+      final ValidatableAttestation attestation,
+      final AsyncBLSSignatureVerifier blsSignatureVerifier) {
+
+    if (!attestation.getAttestation().isSingleAttestation()) {
+      // we can use the default implementation for aggregate attestation
+      return super.isValidIndexedAttestationAsync(fork, state, attestation, blsSignatureVerifier);
+    }
+
+    // single attestation flow
+
+    // 1. verify signature first
+    // 2. verify call getSingleAttestationAggregationBits which also validates the validatorIndex
+    // and the committee against the state
+    // 3. convert attestation inside ValidatableAttestation to AttestationElectra
+    // 4. set the indexed attestation into ValidatableAttestation
+    // 5. set the attestation as valid indexed attestation
+
+    return validateSingleAttestationSignature(
+            fork,
+            state,
+            attestation.getAttestation().toSingleAttestationRequired(),
+            blsSignatureVerifier)
+        .thenApply(
+            result -> {
+              if (result.isSuccessful()) {
+                final SingleAttestation singleAttestation =
+                    attestation.getAttestation().toSingleAttestationRequired();
+                final IndexedAttestation indexedAttestation =
+                    getIndexedAttestationFromSingleAttestation(singleAttestation);
+
+                final SszBitlist singleAttestationAggregationBits =
+                    getSingleAttestationAggregationBits(state, singleAttestation);
+
+                final Attestation convertedAttestation =
+                    convertSingleAttestationToAggregated(
+                        singleAttestation, singleAttestationAggregationBits);
+
+                attestation.convertToAggregatedFormatFromSingleAttestation(convertedAttestation);
+                attestation.saveCommitteeShufflingSeedAndCommitteesSize(state);
+                attestation.setIndexedAttestation(indexedAttestation);
+                attestation.setValidIndexedAttestation();
+              }
+              return result;
+            })
+        .exceptionallyCompose(
+            err -> {
+              if (err.getCause() instanceof IllegalArgumentException) {
+                LOG.debug("on_attestation: Attestation is not valid: ", err);
+                return SafeFuture.completedFuture(
+                    AttestationProcessingResult.invalid(err.getMessage()));
+              } else {
+                return SafeFuture.failedFuture(err);
+              }
+            });
+  }
+
+  private Attestation convertSingleAttestationToAggregated(
+      final SingleAttestation singleAttestation,
+      final SszBitlist singleAttestationAggregationBits) {
+    final AttestationElectraSchema attestationElectraSchema =
+        schemaDefinitions.getAttestationSchema().toVersionElectra().orElseThrow();
+
+    return attestationElectraSchema.create(
+        singleAttestationAggregationBits,
+        singleAttestation.getData(),
+        singleAttestation.getAggregateSignature(),
+        attestationElectraSchema
+            .getCommitteeBitsSchema()
+            .orElseThrow()
+            .ofBits(singleAttestation.getFirstCommitteeIndex().intValue()));
+  }
+
+  private SafeFuture<AttestationProcessingResult> validateSingleAttestationSignature(
+      final Fork fork,
+      final BeaconState state,
+      final SingleAttestation singleAttestation,
+      final AsyncBLSSignatureVerifier signatureVerifier) {
+    final Optional<BLSPublicKey> pubkey =
+        beaconStateAccessors.getValidatorPubKey(
+            state, singleAttestation.getValidatorIndexRequired());
+
+    if (pubkey.isEmpty()) {
+      return completedFuture(
+          AttestationProcessingResult.invalid("Attesting index include non-existent validator"));
+    }
+
+    final BLSSignature signature = singleAttestation.getSignature();
+    final Bytes32 domain =
+        beaconStateAccessors.getDomain(
+            Domain.BEACON_ATTESTER,
+            singleAttestation.getData().getTarget().getEpoch(),
+            fork,
+            state.getGenesisValidatorsRoot());
+    final Bytes signingRoot = miscHelpers.computeSigningRoot(singleAttestation.getData(), domain);
+
+    return signatureVerifier
+        .verify(pubkey.get(), signingRoot, signature)
+        .thenApply(
+            isValidSignature -> {
+              if (isValidSignature) {
+                return AttestationProcessingResult.SUCCESSFUL;
+              } else {
+                return AttestationProcessingResult.invalid("Signature is invalid");
+              }
+            });
+  }
+
+  private SszBitlist getSingleAttestationAggregationBits(
+      final BeaconState state, final SingleAttestation singleAttestation) {
+    final IntList committee =
+        beaconStateAccessors.getBeaconCommittee(
+            state,
+            singleAttestation.getData().getSlot(),
+            singleAttestation.getFirstCommitteeIndex());
+
+    int validatorIndex = singleAttestation.getValidatorIndexRequired().intValue();
+
+    int validatorCommitteeBit = committee.indexOf(validatorIndex);
+
+    checkArgument(
+        validatorCommitteeBit >= 0,
+        "Validator index %s is not part of the committee %s",
+        validatorIndex,
+        singleAttestation.getFirstCommitteeIndex());
+
+    return schemaDefinitions
+        .toVersionElectra()
+        .orElseThrow()
+        .getAttestationSchema()
+        .createAggregationBitsOf(committee.size(), validatorCommitteeBit);
+  }
 }