Added SingleAttestationEvent (#8873)

 - if the attestation arriving is a singleAttestation, will fire a single_attestation event, otherwise will trigger an attestation event.

- at the moment we only see attestations, but added unit test and can re-verify once we have single attestations being produced...

 - can essentially listen to both by :
 `curl -H "Accept:text/event-stream" -N http://localhost:5051/eth/v1/events\?topics\=attestation,single_attestation`

Signed-off-by: Paul Harris <[email protected]>

CHANGELOG.md

@@ -9,8 +9,9 @@
 ### Breaking Changes
 
 ### Additions and Improvements
-- improve block publishing performance, especially relevant with locally produced blocks
-- delay blobs publishing until the block is published to at least 1 peer, especially relevant with locally produced blocks with low upload bandwidth connections. Can be disabled via `--Xp2p-gossip-blobs-after-block-enabled=false`
+- Improved block publishing performance, especially relevant with locally produced blocks
+- Delayed blob publishing until the block is published to at least 1 peer, especially relevant with locally produced blocks with low upload bandwidth connections. Can be disabled via `--Xp2p-gossip-blobs-after-block-enabled=false`
+- Added single_attestation event type for electra attestations
 
 ### Bug Fixes
 - Added a startup script for unix systems to ensure that when jemalloc is installed the script sets the LD_PRELOAD environment variable to the use the jemalloc library

data/beaconrestapi/src/integration-test/resources/tech/pegasys/teku/beaconrestapi/beacon/paths/_eth_v1_events.json

@@ -9,7 +9,7 @@
       "in" : "query",
       "schema" : {
         "type" : "string",
-        "description" : "Event types to subscribe to. Available values include: [`head`, `finalized_checkpoint`, `chain_reorg`, `block`, `attestation`, `voluntary_exit`, `contribution_and_proof`, `blob_sidecar`]\n\n",
+        "description" : "Event types to subscribe to. Supported event types: [`attestation`, `attester_slashing`, `blob_sidecar`, `block_gossip`, `block`, `bls_to_execution_change`, `chain_reorg`, `contribution_and_proof`, `finalized_checkpoint`, `head`, `payload_attributes`, `proposer_slashing`, `single_attestation`, `sync_state`, `voluntary_exit`",
         "example" : "head"
       }
     } ],

data/beaconrestapi/src/main/java/tech/pegasys/teku/beaconrestapi/BeaconRestApiTypes.java

@@ -61,9 +61,12 @@
 import static tech.pegasys.teku.infrastructure.json.types.CoreTypes.STRING_TYPE;
 import static tech.pegasys.teku.infrastructure.json.types.CoreTypes.UINT64_TYPE;
 
+import java.util.EnumSet;
 import java.util.Locale;
 import java.util.function.Function;
+import java.util.stream.Collectors;
 import org.apache.tuweni.bytes.Bytes32;
+import tech.pegasys.teku.api.response.v1.EventType;
 import tech.pegasys.teku.bls.BLSSignature;
 import tech.pegasys.teku.ethereum.json.types.beacon.StatusParameter;
 import tech.pegasys.teku.infrastructure.http.RestApiConstants;
@@ -208,8 +211,11 @@ public class BeaconRestApiTypes {
           TOPICS,
           CoreTypes.string(
               "Event types to subscribe to."
-                  + " Available values include: [`head`, `finalized_checkpoint`, `chain_reorg`, `block`, "
-                  + "`attestation`, `voluntary_exit`, `contribution_and_proof`, `blob_sidecar`]\n\n",
+                  + " Supported event types: ["
+                  + EnumSet.allOf(EventType.class).stream()
+                      .map(val -> "`" + val.toString() + "`")
+                      .sorted()
+                      .collect(Collectors.joining(", ")),
               "head"));
 
   public static final SerializableTypeDefinition<Bytes32> ROOT_TYPE =

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

@@ -211,8 +211,14 @@ protected void onSyncCommitteeContribution(
   }
 
   protected void onNewAttestation(final ValidatableAttestation attestation) {
-    final AttestationEvent attestationEvent = new AttestationEvent(attestation.getAttestation());
-    notifySubscribersOfEvent(EventType.attestation, attestationEvent);
+    if (!attestation.getAttestation().isSingleAttestation()) {
+      final AttestationEvent attestationEvent = new AttestationEvent(attestation.getAttestation());
+      notifySubscribersOfEvent(EventType.attestation, attestationEvent);
+    } else {
+      final SingleAttestationEvent attestationEvent =
+          new SingleAttestationEvent(attestation.getAttestation().toSingleAttestationRequired());
+      notifySubscribersOfEvent(EventType.single_attestation, attestationEvent);
+    }
   }
 
   protected void onNewBlock(final SignedBeaconBlock block, final boolean executionOptimistic) {

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

@@ -0,0 +1,23 @@
+/*
+ * Copyright Consensys Software Inc., 2024
+ *
+ * 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.beaconrestapi.handlers.v1.events;
+
+import tech.pegasys.teku.spec.datastructures.operations.SingleAttestation;
+
+public class SingleAttestationEvent extends Event<SingleAttestation> {
+  SingleAttestationEvent(final SingleAttestation attestation) {
+    super(
+        attestation.toSingleAttestationRequired().getSchema().getJsonTypeDefinition(), attestation);
+  }
+}

data/beaconrestapi/src/test/java/tech/pegasys/teku/beaconrestapi/handlers/v1/events/EventSubscriptionManagerTest.java

@@ -55,6 +55,7 @@
 import tech.pegasys.teku.spec.datastructures.operations.ProposerSlashing;
 import tech.pegasys.teku.spec.datastructures.operations.SignedBlsToExecutionChange;
 import tech.pegasys.teku.spec.datastructures.operations.SignedVoluntaryExit;
+import tech.pegasys.teku.spec.datastructures.operations.SingleAttestation;
 import tech.pegasys.teku.spec.datastructures.operations.versions.altair.SignedContributionAndProof;
 import tech.pegasys.teku.spec.datastructures.state.Checkpoint;
 import tech.pegasys.teku.spec.executionlayer.ForkChoiceState;
@@ -65,7 +66,7 @@
 import tech.pegasys.teku.storage.api.ReorgContext;
 
 public class EventSubscriptionManagerTest {
-  private final Spec spec = TestSpecFactory.createMinimalDeneb();
+  private final Spec spec = TestSpecFactory.createMinimalElectra();
   private final SpecConfig specConfig = spec.getGenesisSpecConfig();
   private final DataStructureUtil data = new DataStructureUtil(spec);
   protected final NodeDataProvider nodeDataProvider = mock(NodeDataProvider.class);
@@ -107,6 +108,7 @@ public class EventSubscriptionManagerTest {
       SignedBeaconBlock.create(data.randomSignedBeaconBlock(0));
   private final BlobSidecar sampleBlobSidecar = data.randomBlobSidecar();
   private final Attestation sampleAttestation = data.randomAttestation(0);
+  private final SingleAttestation singleAttestation = data.randomSingleAttestation();
 
   private final AttesterSlashing sampleAttesterSlashing =
       spec.getGenesisSchemaDefinitions()
@@ -123,7 +125,7 @@ public class EventSubscriptionManagerTest {
       data.randomPayloadBuildingAttributes(true);
   final PayloadAttributesData samplePayloadAttributesData =
       new PayloadAttributesData(
-          SpecMilestone.DENEB,
+          SpecMilestone.ELECTRA,
           new Data(
               samplePayloadAttributes.getProposalSlot(),
               samplePayloadAttributes.getParentBeaconBlockRoot(),
@@ -299,6 +301,15 @@ void shouldPropagateAttestation() throws IOException {
     checkEvent("attestation", new AttestationEvent(sampleAttestation));
   }
 
+  @Test
+  void shouldPropagateSingleAttestation() throws IOException {
+    when(req.getQueryString()).thenReturn("&topics=single_attestation");
+    manager.registerClient(client1);
+
+    triggerSingleAttestationEvent();
+    checkEvent("single_attestation", new SingleAttestationEvent(singleAttestation));
+  }
+
   @Test
   void shouldPropagateAttesterSlashing() throws IOException {
     when(req.getQueryString()).thenReturn("&topics=attester_slashing");
@@ -438,6 +449,11 @@ private void triggerAttestationEvent() {
     asyncRunner.executeQueuedActions();
   }
 
+  private void triggerSingleAttestationEvent() {
+    manager.onNewAttestation(ValidatableAttestation.from(spec, singleAttestation));
+    asyncRunner.executeQueuedActions();
+  }
+
   private void triggerAttesterSlashingEvent() {
     manager.onNewAttesterSlashing(sampleAttesterSlashing, InternalValidationResult.ACCEPT, false);
     asyncRunner.executeQueuedActions();

data/serializer/src/main/java/tech/pegasys/teku/api/response/v1/EventType.java

@@ -30,7 +30,8 @@ public enum EventType {
   attester_slashing,
   proposer_slashing,
   payload_attributes,
-  block_gossip;
+  block_gossip,
+  single_attestation;
 
   public static List<EventType> getTopics(final List<String> topics) {
     return topics.stream().map(EventType::valueOf).toList();