[PROTOTYPE] epbs-eip7732

beacon/sync/src/main/java/tech/pegasys/teku/beacon/sync/NoopSyncService.java

@@ -145,6 +145,11 @@ public boolean isRunning() {
     return false;
   }
 
+  @Override
+  public void onBlockSeen(final SignedBeaconBlock block) {
+    // No-op
+  }
+
   @Override
   public void onBlockValidated(final SignedBeaconBlock block) {
     // No-op

beacon/sync/src/main/java/tech/pegasys/teku/beacon/sync/gossip/blocks/RecentBlocksFetchService.java

@@ -128,6 +128,9 @@ public void processFetchedResult(final FetchBlockTask task, final SignedBeaconBl
     removeTask(task);
   }
 
+  @Override
+  public void onBlockSeen(final SignedBeaconBlock block) {}
+
   @Override
   public void onBlockValidated(final SignedBeaconBlock block) {}
 

beacon/validator/src/integrationTest/java/tech/pegasys/teku/validator/coordinator/ValidatorApiHandlerIntegrationTest.java

@@ -47,8 +47,10 @@
 import tech.pegasys.teku.spec.datastructures.state.Checkpoint;
 import tech.pegasys.teku.statetransition.attestation.AggregatingAttestationPool;
 import tech.pegasys.teku.statetransition.attestation.AttestationManager;
+import tech.pegasys.teku.statetransition.attestation.PayloadAttestationManager;
 import tech.pegasys.teku.statetransition.blobs.BlockBlobSidecarsTrackersPool;
 import tech.pegasys.teku.statetransition.block.BlockImportChannel;
+import tech.pegasys.teku.statetransition.block.ReceivedBlockEventsChannel;
 import tech.pegasys.teku.statetransition.forkchoice.ForkChoiceTrigger;
 import tech.pegasys.teku.statetransition.forkchoice.ProposersDataManager;
 import tech.pegasys.teku.statetransition.synccommittee.SyncCommitteeContributionPool;
@@ -74,6 +76,8 @@ public class ValidatorApiHandlerIntegrationTest {
   private final BlockFactory blockFactory = mock(BlockFactory.class);
   private final AggregatingAttestationPool attestationPool = mock(AggregatingAttestationPool.class);
   private final AttestationManager attestationManager = mock(AttestationManager.class);
+  private final PayloadAttestationManager payloadAttestationManager =
+      mock(PayloadAttestationManager.class);
   private final AttestationTopicSubscriber attestationTopicSubscriber =
       mock(AttestationTopicSubscriber.class);
   private final ActiveValidatorTracker activeValidatorTracker = mock(ActiveValidatorTracker.class);
@@ -114,6 +118,7 @@ public class ValidatorApiHandlerIntegrationTest {
           blobSidecarGossipChannel,
           attestationPool,
           attestationManager,
+          payloadAttestationManager,
           attestationTopicSubscriber,
           activeValidatorTracker,
           dutyMetrics,
@@ -125,7 +130,8 @@ public class ValidatorApiHandlerIntegrationTest {
           syncCommitteeContributionPool,
           syncCommitteeSubscriptionManager,
           new BlockProductionAndPublishingPerformanceFactory(
-              new SystemTimeProvider(), __ -> UInt64.ZERO, true, 0, 0, 0, 0));
+              new SystemTimeProvider(), __ -> UInt64.ZERO, true, 0, 0, 0, 0),
+          ReceivedBlockEventsChannel.NOOP);
 
   @BeforeEach
   public void setup() {

beacon/validator/src/main/java/tech/pegasys/teku/validator/coordinator/ValidatorApiHandler.java

@@ -53,6 +53,7 @@
 import tech.pegasys.teku.ethereum.json.types.node.PeerCount;
 import tech.pegasys.teku.ethereum.json.types.validator.AttesterDuties;
 import tech.pegasys.teku.ethereum.json.types.validator.BeaconCommitteeSelectionProof;
+import tech.pegasys.teku.ethereum.json.types.validator.PayloadAttesterDuties;
 import tech.pegasys.teku.ethereum.json.types.validator.ProposerDuties;
 import tech.pegasys.teku.ethereum.json.types.validator.ProposerDuty;
 import tech.pegasys.teku.ethereum.json.types.validator.SyncCommitteeDuties;
@@ -77,10 +78,12 @@
 import tech.pegasys.teku.spec.datastructures.blocks.SignedBlockAndState;
 import tech.pegasys.teku.spec.datastructures.blocks.SignedBlockContainer;
 import tech.pegasys.teku.spec.datastructures.builder.SignedValidatorRegistration;
+import tech.pegasys.teku.spec.datastructures.execution.PayloadAttestationData;
 import tech.pegasys.teku.spec.datastructures.genesis.GenesisData;
 import tech.pegasys.teku.spec.datastructures.metadata.BlockContainerAndMetaData;
 import tech.pegasys.teku.spec.datastructures.operations.Attestation;
 import tech.pegasys.teku.spec.datastructures.operations.AttestationData;
+import tech.pegasys.teku.spec.datastructures.operations.PayloadAttestationMessage;
 import tech.pegasys.teku.spec.datastructures.operations.SignedAggregateAndProof;
 import tech.pegasys.teku.spec.datastructures.operations.versions.altair.SignedContributionAndProof;
 import tech.pegasys.teku.spec.datastructures.operations.versions.altair.SyncCommitteeContribution;
@@ -93,8 +96,10 @@
 import tech.pegasys.teku.spec.logic.common.util.SyncCommitteeUtil;
 import tech.pegasys.teku.statetransition.attestation.AggregatingAttestationPool;
 import tech.pegasys.teku.statetransition.attestation.AttestationManager;
+import tech.pegasys.teku.statetransition.attestation.PayloadAttestationManager;
 import tech.pegasys.teku.statetransition.blobs.BlockBlobSidecarsTrackersPool;
 import tech.pegasys.teku.statetransition.block.BlockImportChannel;
+import tech.pegasys.teku.statetransition.block.ReceivedBlockEventsChannel;
 import tech.pegasys.teku.statetransition.forkchoice.ForkChoiceTrigger;
 import tech.pegasys.teku.statetransition.forkchoice.ProposersDataManager;
 import tech.pegasys.teku.statetransition.synccommittee.SyncCommitteeContributionPool;
@@ -135,6 +140,7 @@ public class ValidatorApiHandler implements ValidatorApiChannel {
   private final BlockFactory blockFactory;
   private final AggregatingAttestationPool attestationPool;
   private final AttestationManager attestationManager;
+  private final PayloadAttestationManager payloadAttestationManager;
   private final AttestationTopicSubscriber attestationTopicSubscriber;
   private final ActiveValidatorTracker activeValidatorTracker;
   private final DutyMetrics dutyMetrics;
@@ -145,8 +151,8 @@ public class ValidatorApiHandler implements ValidatorApiChannel {
   private final SyncCommitteeSubscriptionManager syncCommitteeSubscriptionManager;
   private final SyncCommitteeContributionPool syncCommitteeContributionPool;
   private final ProposersDataManager proposersDataManager;
+  private final ReceivedBlockEventsChannel receivedBlockEventsChannel;
   private final BlockPublisher blockPublisher;
-
   private final AttesterDutiesGenerator attesterDutiesGenerator;
 
   public ValidatorApiHandler(
@@ -162,6 +168,7 @@ public ValidatorApiHandler(
       final BlobSidecarGossipChannel blobSidecarGossipChannel,
       final AggregatingAttestationPool attestationPool,
       final AttestationManager attestationManager,
+      final PayloadAttestationManager payloadAttestationManager,
       final AttestationTopicSubscriber attestationTopicSubscriber,
       final ActiveValidatorTracker activeValidatorTracker,
       final DutyMetrics dutyMetrics,
@@ -173,7 +180,8 @@ public ValidatorApiHandler(
       final SyncCommitteeContributionPool syncCommitteeContributionPool,
       final SyncCommitteeSubscriptionManager syncCommitteeSubscriptionManager,
       final BlockProductionAndPublishingPerformanceFactory
-          blockProductionAndPublishingPerformanceFactory) {
+          blockProductionAndPublishingPerformanceFactory,
+      final ReceivedBlockEventsChannel receivedBlockEventsChannel) {
     this.blockProductionAndPublishingPerformanceFactory =
         blockProductionAndPublishingPerformanceFactory;
     this.chainDataProvider = chainDataProvider;
@@ -184,6 +192,7 @@ public ValidatorApiHandler(
     this.blockFactory = blockFactory;
     this.attestationPool = attestationPool;
     this.attestationManager = attestationManager;
+    this.payloadAttestationManager = payloadAttestationManager;
     this.attestationTopicSubscriber = attestationTopicSubscriber;
     this.activeValidatorTracker = activeValidatorTracker;
     this.dutyMetrics = dutyMetrics;
@@ -194,6 +203,7 @@ public ValidatorApiHandler(
     this.syncCommitteeContributionPool = syncCommitteeContributionPool;
     this.syncCommitteeSubscriptionManager = syncCommitteeSubscriptionManager;
     this.proposersDataManager = proposersDataManager;
+    this.receivedBlockEventsChannel = receivedBlockEventsChannel;
     this.blockPublisher =
         new MilestoneBasedBlockPublisher(
             spec,
@@ -268,6 +278,41 @@ public SafeFuture<Optional<AttesterDuties>> getAttestationDuties(
                             combinedChainDataClient.isChainHeadOptimistic())));
   }
 
+  @Override
+  public SafeFuture<Optional<PayloadAttesterDuties>> getPayloadAttestationDuties(
+      final UInt64 epoch, final IntCollection validatorIndices) {
+    if (isSyncActive()) {
+      return NodeSyncingException.failedFuture();
+    }
+    if (epoch.isGreaterThan(
+        combinedChainDataClient
+            .getCurrentEpoch()
+            .plus(spec.getSpecConfig(epoch).getMinSeedLookahead() + DUTY_EPOCH_TOLERANCE))) {
+      return SafeFuture.failedFuture(
+          new IllegalArgumentException(
+              String.format(
+                  "Payload attestation duties were requested %s epochs ahead, only 1 epoch in future is supported.",
+                  epoch.minus(combinedChainDataClient.getCurrentEpoch()).toString())));
+    }
+    // what state can we use? If the current or next epoch, we can use the best state,
+    // which would guarantee no state regeneration
+    final UInt64 slot = spec.getEarliestQueryableSlotForBeaconCommitteeInTargetEpoch(epoch);
+
+    LOG.trace(
+        "Retrieving payload attestation duties from epoch {} using state at slot {}", epoch, slot);
+    return combinedChainDataClient
+        .getStateAtSlotExact(slot)
+        .thenApply(
+            optionalState ->
+                optionalState.map(
+                    state ->
+                        attesterDutiesGenerator.getPayloadAttesterDutiesFromIndicesAndState(
+                            state,
+                            epoch,
+                            validatorIndices,
+                            combinedChainDataClient.isChainHeadOptimistic())));
+  }
+
   @Override
   public SafeFuture<Optional<SyncCommitteeDuties>> getSyncCommitteeDuties(
       final UInt64 epoch, final IntCollection validatorIndices) {
@@ -477,6 +522,61 @@ public SafeFuture<Optional<AttestationData>> createAttestationData(
     }
   }
 
+  @Override
+  public SafeFuture<Optional<PayloadAttestationData>> createPayloadAttestationData(
+      final UInt64 slot) {
+    if (isSyncActive()) {
+      return NodeSyncingException.failedFuture();
+    }
+
+    return combinedChainDataClient
+        .getTimelyExecutionPayload(slot)
+        .thenApply(
+            executionPayloadEnvelope ->
+                executionPayloadEnvelope.map(
+                    payloadStatusAndBeaconBlockRoot ->
+                        PayloadAttestationData.SSZ_SCHEMA.create(
+                            payloadStatusAndBeaconBlockRoot.beaconBlockRoot(),
+                            slot,
+                            payloadStatusAndBeaconBlockRoot.payloadStatus().getCode())));
+  }
+
+  @Override
+  public SafeFuture<List<SubmitDataError>> sendSignedPayloadAttestations(
+      final List<PayloadAttestationMessage> attestations) {
+    return SafeFuture.collectAll(attestations.stream().map(this::processPayloadAttestation))
+        .thenApply(this::convertAttestationProcessingResultsToErrorList);
+  }
+
+  private SafeFuture<InternalValidationResult> processPayloadAttestation(
+      final PayloadAttestationMessage attestation) {
+    return payloadAttestationManager
+        .addPayloadAttestation(attestation, Optional.empty())
+        .thenPeek(
+            result -> {
+              if (!result.isReject()) {
+                // TODO update dutyMetrics and performanceTracker
+              } else {
+                VALIDATOR_LOGGER.producedInvalidPayloadAttestation(
+                    attestation.getData().getSlot(),
+                    result.getDescription().orElse("Unknown reason"));
+              }
+            })
+        .exceptionally(
+            error -> {
+              LOG.error(
+                  "Failed to send signed payload attestation for slot {}, block {}",
+                  attestation.getData().getSlot(),
+                  attestation.getData().getBeaconBlockRoot(),
+                  error);
+              return InternalValidationResult.reject(
+                  "Failed to send signed payload attestation for slot %s, block %s: %s",
+                  attestation.getData().getSlot(),
+                  attestation.getData().getBeaconBlockRoot(),
+                  getMessageOrSimpleName(error));
+            });
+  }
+
   private AttestationData createAttestationData(
       final BeaconBlock block,
       final BeaconState state,
@@ -653,6 +753,7 @@ private SafeFuture<InternalValidationResult> processAggregateAndProof(
   public SafeFuture<SendSignedBlockResult> sendSignedBlock(
       final SignedBlockContainer maybeBlindedBlockContainer,
       final BroadcastValidationLevel broadcastValidationLevel) {
+    receivedBlockEventsChannel.onBlockSeen(maybeBlindedBlockContainer.getSignedBlock());
     final BlockPublishingPerformance blockPublishingPerformance =
         blockProductionAndPublishingPerformanceFactory.createForPublishing(
             maybeBlindedBlockContainer.getSlot());

beacon/validator/src/main/java/tech/pegasys/teku/validator/coordinator/duties/AttesterDutiesGenerator.java

@@ -13,6 +13,7 @@
 
 package tech.pegasys.teku.validator.coordinator.duties;
 
+import it.unimi.dsi.fastutil.ints.Int2ObjectMap;
 import it.unimi.dsi.fastutil.ints.IntCollection;
 import java.util.ArrayList;
 import java.util.List;
@@ -21,6 +22,8 @@
 import org.apache.tuweni.bytes.Bytes32;
 import tech.pegasys.teku.ethereum.json.types.validator.AttesterDuties;
 import tech.pegasys.teku.ethereum.json.types.validator.AttesterDuty;
+import tech.pegasys.teku.ethereum.json.types.validator.PayloadAttesterDuties;
+import tech.pegasys.teku.ethereum.json.types.validator.PayloadAttesterDuty;
 import tech.pegasys.teku.infrastructure.unsigned.UInt64;
 import tech.pegasys.teku.spec.Spec;
 import tech.pegasys.teku.spec.datastructures.state.CommitteeAssignment;
@@ -47,6 +50,20 @@ public AttesterDuties getAttesterDutiesFromIndicesAndState(
     return new AttesterDuties(isChainHeadOptimistic, dependentRoot, duties);
   }
 
+  public PayloadAttesterDuties getPayloadAttesterDutiesFromIndicesAndState(
+      final BeaconState state,
+      final UInt64 epoch,
+      final IntCollection validatorIndices,
+      final boolean isChainHeadOptimistic) {
+    final Bytes32 dependentRoot =
+        epoch.isGreaterThan(spec.getCurrentEpoch(state))
+            ? spec.atEpoch(epoch).getBeaconStateUtil().getCurrentDutyDependentRoot(state)
+            : spec.atEpoch(epoch).getBeaconStateUtil().getPreviousDutyDependentRoot(state);
+    final List<PayloadAttesterDuty> duties =
+        createPayloadAttesterDuties(state, epoch, validatorIndices);
+    return new PayloadAttesterDuties(isChainHeadOptimistic, dependentRoot, duties);
+  }
+
   private List<AttesterDuty> createAttesterDuties(
       final BeaconState state, final UInt64 epoch, final IntCollection validatorIndices) {
     final List<Optional<AttesterDuty>> maybeAttesterDutyList = new ArrayList<>();
@@ -84,4 +101,26 @@ private Optional<AttesterDuty> attesterDutyFromCommitteeAssignment(
                     committeeAssignment.getCommittee().indexOf(validatorIndex),
                     committeeAssignment.getSlot()));
   }
+
+  private List<PayloadAttesterDuty> createPayloadAttesterDuties(
+      final BeaconState state, final UInt64 epoch, final IntCollection validatorIndices) {
+    final List<PayloadAttesterDuty> payloadAttesterDutyList = new ArrayList<>();
+
+    final Int2ObjectMap<UInt64> validatorIndexToCommitteeAssignmentMap =
+        spec.getValidatorIndexToPctAssignmentMap(state, epoch);
+    for (final int validatorIndex : validatorIndices) {
+      final UInt64 slot = validatorIndexToCommitteeAssignmentMap.get(validatorIndex);
+      if (slot != null) {
+        payloadAttesterDutyFromCommitteeAssignment(slot, validatorIndex, state)
+            .ifPresent(payloadAttesterDutyList::add);
+      }
+    }
+    return payloadAttesterDutyList;
+  }
+
+  private Optional<PayloadAttesterDuty> payloadAttesterDutyFromCommitteeAssignment(
+      final UInt64 slot, final int validatorIndex, final BeaconState state) {
+    return spec.getValidatorPubKey(state, UInt64.valueOf(validatorIndex))
+        .map(publicKey -> new PayloadAttesterDuty(publicKey, validatorIndex, slot));
+  }
 }

beacon/validator/src/test/java/tech/pegasys/teku/validator/coordinator/BlockOperationSelectorFactoryTest.java

@@ -68,9 +68,11 @@
 import tech.pegasys.teku.spec.datastructures.execution.FallbackData;
 import tech.pegasys.teku.spec.datastructures.execution.FallbackReason;
 import tech.pegasys.teku.spec.datastructures.execution.GetPayloadResponse;
+import tech.pegasys.teku.spec.datastructures.execution.SignedExecutionPayloadHeader;
 import tech.pegasys.teku.spec.datastructures.operations.Attestation;
 import tech.pegasys.teku.spec.datastructures.operations.AttesterSlashing;
 import tech.pegasys.teku.spec.datastructures.operations.Deposit;
+import tech.pegasys.teku.spec.datastructures.operations.PayloadAttestation;
 import tech.pegasys.teku.spec.datastructures.operations.ProposerSlashing;
 import tech.pegasys.teku.spec.datastructures.operations.SignedBlsToExecutionChange;
 import tech.pegasys.teku.spec.datastructures.operations.SignedVoluntaryExit;
@@ -1296,6 +1298,18 @@ public BeaconBlockBodyBuilder blobKzgCommitments(
       return this;
     }
 
+    @Override
+    public BeaconBlockBodyBuilder signedExecutionPayloadHeader(
+        final SignedExecutionPayloadHeader signedExecutionPayloadHeader) {
+      return this;
+    }
+
+    @Override
+    public BeaconBlockBodyBuilder payloadAttestations(
+        final SszList<PayloadAttestation> payloadAttestations) {
+      return this;
+    }
+
     @Override
     public BeaconBlockBody build() {
       return null;