DVT integration for attestation aggregation (Consensys#8032)

Author: lucassaldanha

ethereum/json-types/src/main/java/tech/pegasys/teku/ethereum/json/types/validator/BeaconCommitteeSelectionProof.java

@@ -14,6 +14,8 @@
 package tech.pegasys.teku.ethereum.json.types.validator;
 
 import java.util.Objects;
+import org.apache.tuweni.bytes.Bytes;
+import tech.pegasys.teku.bls.BLSSignature;
 import tech.pegasys.teku.infrastructure.json.types.CoreTypes;
 import tech.pegasys.teku.infrastructure.json.types.DeserializableTypeDefinition;
 import tech.pegasys.teku.infrastructure.unsigned.UInt64;
@@ -67,6 +69,10 @@ public String getSelectionProof() {
     return selectionProof;
   }
 
+  public BLSSignature getSelectionProofSignature() {
+    return BLSSignature.fromBytesCompressed(Bytes.fromHexString(getSelectionProof()));
+  }
+
   public static BeaconCommitteeSelectionProof.Builder builder() {
     return new BeaconCommitteeSelectionProof.Builder();
   }

teku/src/main/java/tech/pegasys/teku/cli/options/ValidatorClientOptions.java

@@ -13,8 +13,6 @@
 
 package tech.pegasys.teku.cli.options;
 
-import static tech.pegasys.teku.validator.api.ValidatorConfig.DEFAULT_VALIDATOR_CLIENT_SSZ_BLOCKS_ENABLED;
-
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.util.List;
@@ -62,7 +60,8 @@ public class ValidatorClientOptions {
       showDefaultValue = CommandLine.Help.Visibility.ALWAYS,
       arity = "0..1",
       fallbackValue = "true")
-  private boolean validatorClientSszBlocksEnabled = DEFAULT_VALIDATOR_CLIENT_SSZ_BLOCKS_ENABLED;
+  private boolean validatorClientSszBlocksEnabled =
+      ValidatorConfig.DEFAULT_VALIDATOR_CLIENT_SSZ_BLOCKS_ENABLED;
 
   @CommandLine.Option(
       names = {"--Xuse-post-validators-endpoint-enabled"},
@@ -75,6 +74,18 @@ public class ValidatorClientOptions {
   private boolean validatorClientUsePostValidatorsEndpointEnabled =
       ValidatorConfig.DEFAULT_VALIDATOR_CLIENT_USE_POST_VALIDATORS_ENDPOINT_ENABLED;
 
+  @Option(
+      names = {"--Xdvt-integration-enabled"},
+      paramLabel = "<BOOLEAN>",
+      description =
+          "Use DVT endpoints to determine if a distributed validator has aggregation duties.",
+      arity = "0..1",
+      showDefaultValue = CommandLine.Help.Visibility.ALWAYS,
+      hidden = true,
+      fallbackValue = "true")
+  private boolean dvtSelectionsEndpointEnabled =
+      ValidatorConfig.DEFAULT_DVT_SELECTIONS_ENDPOINT_ENABLED;
+
   public void configure(TekuConfiguration.Builder builder) {
     configureBeaconNodeApiEndpoints();
 
@@ -87,7 +98,8 @@ public void configure(TekuConfiguration.Builder builder) {
                     validatorClientUsePostValidatorsEndpointEnabled)
                 .failoversSendSubnetSubscriptionsEnabled(failoversSendSubnetSubscriptionsEnabled)
                 .failoversPublishSignedDutiesEnabled(failoversPublishSignedDutiesEnabled)
-                .sentryNodeConfigurationFile(exclusiveParams.sentryConfigFile));
+                .sentryNodeConfigurationFile(exclusiveParams.sentryConfigFile)
+                .dvtSelectionsEndpointEnabled(dvtSelectionsEndpointEnabled));
   }
 
   private void configureBeaconNodeApiEndpoints() {

teku/src/test/java/tech/pegasys/teku/cli/options/ValidatorOptionsTest.java

@@ -234,4 +234,12 @@ public void shouldSetShutdownWhenValidatorSlashedEnabled() {
             .getValidatorConfig();
     assertThat(config.isShutdownWhenValidatorSlashedEnabled()).isTrue();
   }
+
+  @Test
+  public void shouldNotUseDvtSelectionsEndpointByDefault() {
+    final String[] args = {};
+    final TekuConfiguration config = getTekuConfigurationFromArguments(args);
+    assertThat(config.validatorClient().getValidatorConfig().isDvtSelectionsEndpointEnabled())
+        .isFalse();
+  }
 }

teku/src/test/java/tech/pegasys/teku/cli/subcommand/ValidatorClientCommandTest.java

@@ -210,6 +210,23 @@ public void clientExecutorThreadsShouldThrowOverLimit() {
             "--Xvalidator-client-executor-threads must be greater than 0 and less than 5000.");
   }
 
+  @Test
+  public void shouldSetUseDvtSelectionsEndpoint() {
+    final String[] args = {"vc", "--network", "minimal", "--Xdvt-integration-enabled"};
+    final TekuConfiguration config = getTekuConfigurationFromArguments(args);
+
+    assertThat(config.validatorClient().getValidatorConfig().isDvtSelectionsEndpointEnabled())
+        .isTrue();
+  }
+
+  @Test
+  public void shouldNotUseDvtSelectionsEndpointByDefault() {
+    final String[] args = {"vc", "--network", "minimal"};
+    final TekuConfiguration config = getTekuConfigurationFromArguments(args);
+    assertThat(config.validatorClient().getValidatorConfig().isDvtSelectionsEndpointEnabled())
+        .isFalse();
+  }
+
   private String pathFor(final String filename) {
     return Resources.getResource(ValidatorClientCommandTest.class, filename).toString();
   }

validator/api/src/main/java/tech/pegasys/teku/validator/api/ValidatorConfig.java

@@ -65,6 +65,7 @@ public class ValidatorConfig {
   public static final boolean DEFAULT_VALIDATOR_BLINDED_BLOCKS_ENABLED = false;
   public static final int DEFAULT_VALIDATOR_REGISTRATION_SENDING_BATCH_SIZE = 100;
   public static final UInt64 DEFAULT_BUILDER_REGISTRATION_GAS_LIMIT = UInt64.valueOf(30_000_000);
+  public static final boolean DEFAULT_DVT_SELECTIONS_ENDPOINT_ENABLED = false;
 
   private final List<String> validatorKeys;
   private final List<String> validatorExternalSignerPublicKeySources;
@@ -105,6 +106,7 @@ public class ValidatorConfig {
   private final int executorThreads;
 
   private final boolean isLocalSlashingProtectionSynchronizedModeEnabled;
+  private final boolean dvtSelectionsEndpointEnabled;
 
   private ValidatorConfig(
       final List<String> validatorKeys,
@@ -143,7 +145,8 @@ private ValidatorConfig(
       final int executorMaxQueueSize,
       final int executorThreads,
       final Optional<String> sentryNodeConfigurationFile,
-      boolean isLocalSlashingProtectionSynchronizedModeEnabled) {
+      boolean isLocalSlashingProtectionSynchronizedModeEnabled,
+      boolean dvtSelectionsEndpointEnabled) {
     this.validatorKeys = validatorKeys;
     this.validatorExternalSignerPublicKeySources = validatorExternalSignerPublicKeySources;
     this.validatorExternalSignerUrl = validatorExternalSignerUrl;
@@ -186,6 +189,7 @@ private ValidatorConfig(
     this.sentryNodeConfigurationFile = sentryNodeConfigurationFile;
     this.isLocalSlashingProtectionSynchronizedModeEnabled =
         isLocalSlashingProtectionSynchronizedModeEnabled;
+    this.dvtSelectionsEndpointEnabled = dvtSelectionsEndpointEnabled;
   }
 
   public static Builder builder() {
@@ -348,6 +352,10 @@ public boolean isLocalSlashingProtectionSynchronizedModeEnabled() {
     return isLocalSlashingProtectionSynchronizedModeEnabled;
   }
 
+  public boolean isDvtSelectionsEndpointEnabled() {
+    return dvtSelectionsEndpointEnabled;
+  }
+
   public static final class Builder {
     private List<String> validatorKeys = new ArrayList<>();
     private List<String> validatorExternalSignerPublicKeySources = new ArrayList<>();
@@ -395,11 +403,10 @@ public static final class Builder {
     private Optional<BLSPublicKey> builderRegistrationPublicKeyOverride = Optional.empty();
     private int executorMaxQueueSize = DEFAULT_EXECUTOR_MAX_QUEUE_SIZE;
     private Optional<String> sentryNodeConfigurationFile = Optional.empty();
-
     private int executorThreads = DEFAULT_VALIDATOR_EXECUTOR_THREADS;
-
     private boolean isLocalSlashingProtectionSynchronizedModeEnabled =
         DEFAULT_VALIDATOR_IS_LOCAL_SLASHING_PROTECTION_SYNCHRONIZED_ENABLED;
+    private boolean dvtSelectionsEndpointEnabled = DEFAULT_DVT_SELECTIONS_ENDPOINT_ENABLED;
 
     private Builder() {}
 
@@ -640,6 +647,11 @@ public Builder isLocalSlashingProtectionSynchronizedModeEnabled(
       return this;
     }
 
+    public Builder dvtSelectionsEndpointEnabled(final boolean dvtSelectionsEndpointEnabled) {
+      this.dvtSelectionsEndpointEnabled = dvtSelectionsEndpointEnabled;
+      return this;
+    }
+
     public ValidatorConfig build() {
       validateExternalSignerUrlAndPublicKeys();
       validateExternalSignerKeystoreAndPasswordFileConfig();
@@ -683,7 +695,8 @@ public ValidatorConfig build() {
           executorMaxQueueSize,
           executorThreads,
           sentryNodeConfigurationFile,
-          isLocalSlashingProtectionSynchronizedModeEnabled);
+          isLocalSlashingProtectionSynchronizedModeEnabled,
+          dvtSelectionsEndpointEnabled);
     }
 
     private void validateExternalSignerUrlAndPublicKeys() {

validator/client/src/main/java/tech/pegasys/teku/validator/client/AttestationDutyLoader.java

@@ -45,6 +45,7 @@ public class AttestationDutyLoader
       scheduledDutiesFactory;
   private final BeaconCommitteeSubscriptions beaconCommitteeSubscriptions;
   private final Spec spec;
+  private final boolean useDvtEndpoint;
 
   public AttestationDutyLoader(
       final ValidatorApiChannel validatorApiChannel,
@@ -54,13 +55,15 @@ public AttestationDutyLoader(
       final OwnedValidators validators,
       final ValidatorIndexProvider validatorIndexProvider,
       final BeaconCommitteeSubscriptions beaconCommitteeSubscriptions,
-      final Spec spec) {
+      final Spec spec,
+      final boolean useDvtEndpoint) {
     super(validators, validatorIndexProvider);
     this.validatorApiChannel = validatorApiChannel;
     this.forkProvider = forkProvider;
     this.scheduledDutiesFactory = scheduledDutiesFactory;
     this.beaconCommitteeSubscriptions = beaconCommitteeSubscriptions;
     this.spec = spec;
+    this.useDvtEndpoint = useDvtEndpoint;
   }
 
   @Override
@@ -77,17 +80,27 @@ protected SafeFuture<Optional<AttesterDuties>> requestDuties(
       final UInt64 epoch, final AttesterDuties duties) {
     final SlotBasedScheduledDuties<AttestationProductionDuty, AggregationDuty> scheduledDuties =
         scheduledDutiesFactory.apply(duties.getDependentRoot());
+
+    final Optional<DvtAttestationAggregations> dvtAttestationAggregationsForEpoch =
+        useDvtEndpoint
+            ? Optional.of(
+                new DvtAttestationAggregations(validatorApiChannel, duties.getDuties().size()))
+            : Optional.empty();
+
     return SafeFuture.allOf(
             duties.getDuties().stream()
-                .map(duty -> scheduleDuties(scheduledDuties, duty))
+                .map(
+                    duty ->
+                        scheduleDuties(scheduledDuties, duty, dvtAttestationAggregationsForEpoch))
                 .toArray(SafeFuture[]::new))
         .<SlotBasedScheduledDuties<?, ?>>thenApply(__ -> scheduledDuties)
         .alwaysRun(beaconCommitteeSubscriptions::sendRequests);
   }
 
   private SafeFuture<Void> scheduleDuties(
       final SlotBasedScheduledDuties<AttestationProductionDuty, AggregationDuty> scheduledDuties,
-      final AttesterDuty duty) {
+      final AttesterDuty duty,
+      final Optional<DvtAttestationAggregations> dvtAttestationAggregationLoader) {
     final Optional<Validator> maybeValidator = validators.getValidator(duty.getPublicKey());
     if (maybeValidator.isEmpty()) {
       return SafeFuture.COMPLETE;
@@ -116,7 +129,8 @@ private SafeFuture<Void> scheduleDuties(
         validator,
         duty.getSlot(),
         aggregatorModulo,
-        unsignedAttestationFuture);
+        unsignedAttestationFuture,
+        dvtAttestationAggregationLoader);
   }
 
   private SafeFuture<Optional<AttestationData>> scheduleAttestationProduction(
@@ -147,10 +161,19 @@ private SafeFuture<Void> scheduleAggregation(
       final Validator validator,
       final UInt64 slot,
       final int aggregatorModulo,
-      final SafeFuture<Optional<AttestationData>> unsignedAttestationFuture) {
+      final SafeFuture<Optional<AttestationData>> unsignedAttestationFuture,
+      final Optional<DvtAttestationAggregations> dvtAttestationAggregation) {
     return forkProvider
         .getForkInfo(slot)
         .thenCompose(forkInfo -> validator.getSigner().signAggregationSlot(slot, forkInfo))
+        .thenCompose(
+            slotSignature ->
+                dvtAttestationAggregation
+                    .map(
+                        dvt ->
+                            dvt.getCombinedSelectionProofFuture(
+                                validatorIndex, slot, slotSignature))
+                    .orElse(SafeFuture.completedFuture(slotSignature)))
         .thenAccept(
             slotSignature -> {
               final SpecVersion specVersion = spec.atSlot(slot);