Follow headV2 event spec: new emit on payload status change, no new emit for headV1

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

@@ -84,6 +84,13 @@ public class EventSubscriptionManager
   // collection of subscribers
   private final Collection<EventSubscriber> eventSubscribers;
 
+  // The v1 head event carries no payload status, so a Gloas empty -> full head update would produce
+  // a byte identical duplicate. Only head_v2 is expected to emit a second event for the same beacon
+  // block and slot (https://github.com/ethereum/beacon-APIs/pull/628), so the duplicate v1 event is
+  // suppressed by comparing against the last emitted one. Only the immediately preceding event is
+  // retained: reverting to an earlier head is a re-org, which must be re-emitted.
+  private volatile HeadEvent.HeadData lastHeadEventData;
+
   public EventSubscriptionManager(
       final Spec spec,
       final NodeDataProvider nodeDataProvider,
@@ -176,7 +183,10 @@ public void chainHeadUpdated(
             executionOptimistic,
             previousDutyDependentRoot,
             currentDutyDependentRoot);
-    notifySubscribersOfEvent(EventType.head, headEvent);
+    if (!headEvent.getData().equals(lastHeadEventData)) {
+      lastHeadEventData = headEvent.getData();
+      notifySubscribersOfEvent(EventType.head, headEvent);
+    }
 
     final HeadV2Event headV2Event =
         HeadV2Event.create(

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

@@ -18,6 +18,7 @@
 import static tech.pegasys.teku.infrastructure.json.types.CoreTypes.BYTES32_TYPE;
 import static tech.pegasys.teku.infrastructure.json.types.CoreTypes.UINT64_TYPE;
 
+import java.util.Objects;
 import org.apache.tuweni.bytes.Bytes32;
 import tech.pegasys.teku.infrastructure.json.types.SerializableTypeDefinition;
 import tech.pegasys.teku.infrastructure.unsigned.UInt64;
@@ -111,5 +112,35 @@ public Bytes32 getPreviousDutyDependentRoot() {
     public Bytes32 getCurrentDutyDependentRoot() {
       return currentDutyDependentRoot;
     }
+
+    @Override
+    public boolean equals(final Object o) {
+      if (this == o) {
+        return true;
+      }
+      if (o == null || getClass() != o.getClass()) {
+        return false;
+      }
+      final HeadData headData = (HeadData) o;
+      return epochTransition == headData.epochTransition
+          && executionOptimistic == headData.executionOptimistic
+          && Objects.equals(slot, headData.slot)
+          && Objects.equals(block, headData.block)
+          && Objects.equals(state, headData.state)
+          && Objects.equals(previousDutyDependentRoot, headData.previousDutyDependentRoot)
+          && Objects.equals(currentDutyDependentRoot, headData.currentDutyDependentRoot);
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(
+          slot,
+          block,
+          state,
+          epochTransition,
+          executionOptimistic,
+          previousDutyDependentRoot,
+          currentDutyDependentRoot);
+    }
   }
 }

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

@@ -242,6 +242,90 @@ void shouldPropagateHeadV2Event() throws IOException {
     assertThat(outputStream.getString()).contains("\"next_epoch_dependent_root\"");
   }
 
+  /**
+   * beacon-APIs requires (`should`) a second head_v2 event for the same beacon block and slot when
+   * the payload status changes from empty to full. See
+   * https://github.com/ethereum/beacon-APIs/pull/628.
+   */
+  @Test
+  void shouldPropagateSecondHeadV2EventWhenPayloadStatusChangesFromEmptyToFull() {
+    when(req.getQueryString()).thenReturn("&topics=head_v2");
+    manager.registerClient(client1);
+
+    triggerHeadV2Event(ForkChoicePayloadStatus.PAYLOAD_STATUS_EMPTY);
+    triggerHeadV2Event(ForkChoicePayloadStatus.PAYLOAD_STATUS_FULL);
+
+    final String eventString = outputStream.getString();
+    assertThat(countOccurrences(eventString, "event: head_v2\n")).isEqualTo(2);
+    assertThat(eventString.indexOf("\"payload_status\":\"empty\""))
+        .isLessThan(eventString.indexOf("\"payload_status\":\"full\""));
+  }
+
+  /**
+   * Emission on payload status transitions other than empty -> full is optional and
+   * implementation-defined (https://github.com/ethereum/beacon-APIs/pull/629). Teku emits them,
+   * because fork choice treats a payload status change as a new head.
+   */
+  @Test
+  void shouldPropagateHeadV2EventWhenPayloadStatusChangesFromFullToEmpty() {
+    when(req.getQueryString()).thenReturn("&topics=head_v2");
+    manager.registerClient(client1);
+
+    triggerHeadV2Event(ForkChoicePayloadStatus.PAYLOAD_STATUS_FULL);
+    triggerHeadV2Event(ForkChoicePayloadStatus.PAYLOAD_STATUS_EMPTY);
+
+    final String eventString = outputStream.getString();
+    assertThat(countOccurrences(eventString, "event: head_v2\n")).isEqualTo(2);
+    assertThat(eventString.indexOf("\"payload_status\":\"full\""))
+        .isLessThan(eventString.indexOf("\"payload_status\":\"empty\""));
+  }
+
+  /**
+   * The second emission is a head_v2 concern only: the v1 head event carries no payload status, so
+   * re-emitting it for the same head would be a byte identical duplicate.
+   */
+  @Test
+  void shouldNotPropagateDuplicateHeadV1EventWhenOnlyPayloadStatusChanges() {
+    when(req.getQueryString()).thenReturn("&topics=head");
+    manager.registerClient(client1);
+
+    triggerHeadV2Event(ForkChoicePayloadStatus.PAYLOAD_STATUS_EMPTY);
+    triggerHeadV2Event(ForkChoicePayloadStatus.PAYLOAD_STATUS_FULL);
+
+    assertThat(countOccurrences(outputStream.getString(), "event: head\n")).isEqualTo(1);
+  }
+
+  /**
+   * Duplicate suppression must only consider the immediately preceding head event: re-orging back
+   * to a head that was already reported is a genuine head change and has to be re-emitted.
+   */
+  @Test
+  void shouldPropagateHeadEventWhenReorgingBackToAPreviouslyReportedHead() {
+    when(req.getQueryString()).thenReturn("&topics=head");
+    manager.registerClient(client1);
+
+    final Bytes32 originalBlockRoot = headEvent.getData().getBlock();
+    final Bytes32 forkBlockRoot = data.randomBytes32();
+
+    triggerHeadEvent(originalBlockRoot, ForkChoicePayloadStatus.PAYLOAD_STATUS_EMPTY);
+    // The payload arrives: head_v2 emits again, the v1 head event is a duplicate.
+    triggerHeadEvent(originalBlockRoot, ForkChoicePayloadStatus.PAYLOAD_STATUS_FULL);
+    triggerHeadEvent(forkBlockRoot, ForkChoicePayloadStatus.PAYLOAD_STATUS_FULL);
+    triggerHeadEvent(originalBlockRoot, ForkChoicePayloadStatus.PAYLOAD_STATUS_FULL);
+
+    assertThat(countOccurrences(outputStream.getString(), "event: head\n")).isEqualTo(3);
+  }
+
+  private int countOccurrences(final String haystack, final String needle) {
+    int count = 0;
+    int index = haystack.indexOf(needle);
+    while (index >= 0) {
+      count++;
+      index = haystack.indexOf(needle, index + needle.length());
+    }
+    return count;
+  }
+
   @Test
   void shouldPropagateContributions() {
     when(req.getQueryString()).thenReturn("&topics=contribution_and_proof");
@@ -722,6 +806,25 @@ private void triggerHeadEvent() {
   }
 
   private void triggerHeadV2Event() {
+    triggerHeadV2Event(ForkChoicePayloadStatus.PAYLOAD_STATUS_FULL);
+  }
+
+  private void triggerHeadEvent(
+      final Bytes32 bestBlockRoot, final ForkChoicePayloadStatus payloadStatus) {
+    manager.chainHeadUpdated(
+        headEvent.getData().getSlot(),
+        headEvent.getData().getState(),
+        bestBlockRoot,
+        false,
+        true,
+        headEvent.getData().getPreviousDutyDependentRoot(),
+        headEvent.getData().getCurrentDutyDependentRoot(),
+        Optional.of(payloadStatus),
+        Optional.empty());
+    asyncRunner.executeQueuedActions();
+  }
+
+  private void triggerHeadV2Event(final ForkChoicePayloadStatus payloadStatus) {
     manager.chainHeadUpdated(
         headV2Event.getData().data().slot(),
         headV2Event.getData().data().state(),
@@ -730,7 +833,7 @@ private void triggerHeadV2Event() {
         true,
         headV2Event.getData().data().currentEpochDependentRoot(),
         headV2Event.getData().data().nextEpochDependentRoot(),
-        Optional.of(ForkChoicePayloadStatus.PAYLOAD_STATUS_FULL),
+        Optional.of(payloadStatus),
         Optional.empty());
     asyncRunner.executeQueuedActions();
   }

ethereum/statetransition/src/test/java/tech/pegasys/teku/statetransition/forkchoice/ForkChoiceTest.java

@@ -123,6 +123,7 @@
 import tech.pegasys.teku.statetransition.validation.BlockBroadcastValidator;
 import tech.pegasys.teku.statetransition.validation.BlockBroadcastValidator.BroadcastValidationResult;
 import tech.pegasys.teku.storage.api.LateBlockReorgPreparationHandler;
+import tech.pegasys.teku.storage.api.TrackingChainHeadChannel.HeadEvent;
 import tech.pegasys.teku.storage.api.TrackingChainHeadChannel.ReorgEvent;
 import tech.pegasys.teku.storage.client.ChainHead;
 import tech.pegasys.teku.storage.client.ChainUpdater;
@@ -1790,6 +1791,42 @@ void onAttestation_gloasFullVoteShouldRequireReprocessingWhenPayloadArrives() {
         .isEqualTo(ZERO);
   }
 
+  /**
+   * beacon-APIs requires (`should`) a second head event for the same beacon block and slot when the
+   * head's payload status changes from empty to full. See
+   * https://github.com/ethereum/beacon-APIs/pull/628.
+   */
+  @Test
+  void onExecutionPayloadEnvelope_shouldUpdateHeadWhenPayloadStatusChangesFromEmptyToFull() {
+    setupWithSpec(
+        TestSpecFactory.createMinimalGloas(
+            builder -> builder.blsSignatureVerifier(BLSSignatureVerifier.NOOP)));
+    assertThat(forkChoice.applyGenesisExecutionPayloadForGloas()).isCompleted();
+
+    final SignedBlockAndState block = chainBuilder.generateBlockAtSlot(ONE);
+    importBlock(block);
+
+    // The payload hasn't been revealed yet, so the head is the EMPTY node of the new block.
+    final List<HeadEvent> headEvents = storageSystem.chainHeadChannel().getHeadEvents();
+    assertThat(headEvents).isNotEmpty();
+    final HeadEvent emptyHeadEvent = headEvents.getLast();
+    assertThat(emptyHeadEvent.getSlot()).isEqualTo(block.getSlot());
+    assertThat(emptyHeadEvent.getBestBlockRoot()).isEqualTo(block.getRoot());
+    assertThat(emptyHeadEvent.getPayloadStatus())
+        .contains(ForkChoicePayloadStatus.PAYLOAD_STATUS_EMPTY);
+    headEvents.clear();
+
+    importPayload(block);
+
+    // A second head event for the same block and slot, now reporting the FULL payload status.
+    assertThat(headEvents).hasSize(1);
+    final HeadEvent fullHeadEvent = headEvents.getFirst();
+    assertThat(fullHeadEvent.getSlot()).isEqualTo(block.getSlot());
+    assertThat(fullHeadEvent.getBestBlockRoot()).isEqualTo(block.getRoot());
+    assertThat(fullHeadEvent.getPayloadStatus())
+        .contains(ForkChoicePayloadStatus.PAYLOAD_STATUS_FULL);
+  }
+
   @Test
   void applyIndexedAttestations_gloasFullVoteShouldNotApplyWhenExecutionPayloadMissing() {
     setupWithSpec(

storage/src/testFixtures/java/tech/pegasys/teku/storage/api/TrackingChainHeadChannel.java

@@ -45,7 +45,8 @@ public void chainHeadUpdated(
             epochTransition,
             executionOptimistic,
             previousDutyDependentRoot,
-            currentDutyDependentRoot));
+            currentDutyDependentRoot,
+            payloadStatus));
     optionalReorgContext.ifPresent(
         context ->
             reorgEvents.add(
@@ -74,6 +75,7 @@ public static class HeadEvent {
     private final boolean executionOptimistic;
     private final Bytes32 previousDutyDependentRoot;
     private final Bytes32 currentDutyDependentRoot;
+    private final Optional<ForkChoicePayloadStatus> payloadStatus;
 
     public HeadEvent(
         final UInt64 slot,
@@ -83,19 +85,48 @@ public HeadEvent(
         final boolean executionOptimistic,
         final Bytes32 previousDutyDependentRoot,
         final Bytes32 currentDutyDependentRoot) {
+      this(
+          slot,
+          stateRoot,
+          bestBlockRoot,
+          epochTransition,
+          executionOptimistic,
+          previousDutyDependentRoot,
+          currentDutyDependentRoot,
+          Optional.empty());
+    }
+
+    public HeadEvent(
+        final UInt64 slot,
+        final Bytes32 stateRoot,
+        final Bytes32 bestBlockRoot,
+        final boolean epochTransition,
+        final boolean executionOptimistic,
+        final Bytes32 previousDutyDependentRoot,
+        final Bytes32 currentDutyDependentRoot,
+        final Optional<ForkChoicePayloadStatus> payloadStatus) {
       this.slot = slot;
       this.stateRoot = stateRoot;
       this.bestBlockRoot = bestBlockRoot;
       this.epochTransition = epochTransition;
       this.executionOptimistic = executionOptimistic;
       this.previousDutyDependentRoot = previousDutyDependentRoot;
       this.currentDutyDependentRoot = currentDutyDependentRoot;
+      this.payloadStatus = payloadStatus;
     }
 
     public UInt64 getSlot() {
       return slot;
     }
 
+    public Bytes32 getBestBlockRoot() {
+      return bestBlockRoot;
+    }
+
+    public Optional<ForkChoicePayloadStatus> getPayloadStatus() {
+      return payloadStatus;
+    }
+
     public Bytes32 getPreviousDutyDependentRoot() {
       return previousDutyDependentRoot;
     }
@@ -119,7 +150,8 @@ public boolean equals(final Object o) {
           && Objects.equals(stateRoot, headEvent.stateRoot)
           && Objects.equals(bestBlockRoot, headEvent.bestBlockRoot)
           && Objects.equals(previousDutyDependentRoot, headEvent.previousDutyDependentRoot)
-          && Objects.equals(currentDutyDependentRoot, headEvent.currentDutyDependentRoot);
+          && Objects.equals(currentDutyDependentRoot, headEvent.currentDutyDependentRoot)
+          && Objects.equals(payloadStatus, headEvent.payloadStatus);
     }
 
     @Override
@@ -131,7 +163,8 @@ public int hashCode() {
           epochTransition,
           executionOptimistic,
           previousDutyDependentRoot,
-          currentDutyDependentRoot);
+          currentDutyDependentRoot,
+          payloadStatus);
     }
 
     @Override
@@ -144,6 +177,7 @@ public String toString() {
           .add("executionOptimistic", executionOptimistic)
           .add("previousDutyDependentRoot", previousDutyDependentRoot)
           .add("currentDutyDependentRoot", currentDutyDependentRoot)
+          .add("payloadStatus", payloadStatus)
           .toString();
     }
   }