Add VC beaconAPI pools CLI params (#9611)

tbenr · web-flow · commit 765e1275ce0d · 2025-06-26T16:36:05.000+02:00
diff --git a/teku/src/main/java/tech/pegasys/teku/cli/options/ValidatorOptions.java b/teku/src/main/java/tech/pegasys/teku/cli/options/ValidatorOptions.java
@@ -138,6 +138,27 @@ public class ValidatorOptions {
       arity = "1")
   private int executorThreads = DEFAULT_VALIDATOR_EXECUTOR_THREADS;
 
+  @Option(
+      names = {"--Xvalidator-client-beacon-api-executor-threads"},
+      paramLabel = "<INTEGER>",
+      showDefaultValue = Visibility.ALWAYS,
+      description = "Set the number of threads for the validator beacon node API executor",
+      hidden = true,
+      converter = OptionalIntConverter.class,
+      arity = "1")
+  private OptionalInt beaconApiExecutorThreads = OptionalInt.empty();
+
+  @Option(
+      names = {"--Xvalidator-client-beacon-api-readiness-executor-threads"},
+      paramLabel = "<INTEGER>",
+      showDefaultValue = Visibility.ALWAYS,
+      description =
+          "Set the number of threads for the validator beacon node API readiness executor",
+      hidden = true,
+      converter = OptionalIntConverter.class,
+      arity = "1")
+  private OptionalInt beaconApiReadinessExecutorThreads = OptionalInt.empty();
+
   @Option(
       names = {"--exit-when-no-validator-keys-enabled"},
       paramLabel = "<BOOLEAN>",
@@ -170,25 +191,26 @@ public class ValidatorOptions {
 
   public void configure(final TekuConfiguration.Builder builder) {
     builder.validator(
-        config -> {
-          config
-              .validatorKeystoreLockingEnabled(validatorKeystoreLockingEnabled)
-              .validatorPerformanceTrackingMode(validatorPerformanceTrackingMode)
-              .validatorExternalSignerSlashingProtectionEnabled(
-                  validatorExternalSignerSlashingProtectionEnabled)
-              .isLocalSlashingProtectionSynchronizedModeEnabled(
-                  isLocalSlashingProtectionSynchronizedEnabled)
-              .graffitiProvider(
-                  new FileBackedGraffitiProvider(
-                      Optional.ofNullable(graffiti), Optional.ofNullable(graffitiFile)))
-              .clientGraffitiAppendFormat(clientGraffitiAppendFormat)
-              .generateEarlyAttestations(generateEarlyAttestations)
-              .doppelgangerDetectionEnabled(doppelgangerDetectionEnabled)
-              .executorThreads(executorThreads)
-              .exitWhenNoValidatorKeysEnabled(exitWhenNoValidatorKeysEnabled)
-              .shutdownWhenValidatorSlashedEnabled(shutdownWhenValidatorSlashed);
-          executorMaxQueueSize.ifPresent(config::executorMaxQueueSize);
-        });
+        config ->
+            config
+                .validatorKeystoreLockingEnabled(validatorKeystoreLockingEnabled)
+                .validatorPerformanceTrackingMode(validatorPerformanceTrackingMode)
+                .validatorExternalSignerSlashingProtectionEnabled(
+                    validatorExternalSignerSlashingProtectionEnabled)
+                .isLocalSlashingProtectionSynchronizedModeEnabled(
+                    isLocalSlashingProtectionSynchronizedEnabled)
+                .graffitiProvider(
+                    new FileBackedGraffitiProvider(
+                        Optional.ofNullable(graffiti), Optional.ofNullable(graffitiFile)))
+                .clientGraffitiAppendFormat(clientGraffitiAppendFormat)
+                .generateEarlyAttestations(generateEarlyAttestations)
+                .doppelgangerDetectionEnabled(doppelgangerDetectionEnabled)
+                .executorThreads(executorThreads)
+                .exitWhenNoValidatorKeysEnabled(exitWhenNoValidatorKeysEnabled)
+                .shutdownWhenValidatorSlashedEnabled(shutdownWhenValidatorSlashed)
+                .executorMaxQueueSize(executorMaxQueueSize)
+                .beaconApiExecutorThreads(beaconApiExecutorThreads)
+                .beaconApiReadinessExecutorThreads(beaconApiReadinessExecutorThreads));
     validatorProposerOptions.configure(builder);
     validatorKeysOptions.configure(builder);
   }
diff --git a/teku/src/test/java/tech/pegasys/teku/cli/options/ValidatorOptionsTest.java b/teku/src/test/java/tech/pegasys/teku/cli/options/ValidatorOptionsTest.java
@@ -251,4 +251,35 @@ public void shouldNotUseDvtSelectionsEndpointByDefault() {
     assertThat(config.validatorClient().getValidatorConfig().isDvtSelectionsEndpointEnabled())
         .isFalse();
   }
+
+  @Test
+  public void validatorClientBeaconApiExecutorsEmptyByDefault() {
+    final TekuConfiguration tekuConfiguration = getTekuConfigurationFromArguments();
+    assertThat(
+            tekuConfiguration.validatorClient().getValidatorConfig().getBeaconApiExecutorThreads())
+        .isEmpty();
+    assertThat(
+            tekuConfiguration
+                .validatorClient()
+                .getValidatorConfig()
+                .getBeaconApiReadinessExecutorThreads())
+        .isEmpty();
+  }
+
+  @Test
+  public void validatorClientBeaconApiExecutorsCanBeSet() {
+    final TekuConfiguration tekuConfiguration =
+        getTekuConfigurationFromArguments(
+            "--Xvalidator-client-beacon-api-executor-threads=2",
+            "--Xvalidator-client-beacon-api-readiness-executor-threads=4");
+    assertThat(
+            tekuConfiguration.validatorClient().getValidatorConfig().getBeaconApiExecutorThreads())
+        .hasValue(2);
+    assertThat(
+            tekuConfiguration
+                .validatorClient()
+                .getValidatorConfig()
+                .getBeaconApiReadinessExecutorThreads())
+        .hasValue(4);
+  }
 }
diff --git a/validator/api/src/main/java/tech/pegasys/teku/validator/api/ValidatorConfig.java b/validator/api/src/main/java/tech/pegasys/teku/validator/api/ValidatorConfig.java
@@ -110,6 +110,9 @@ public class ValidatorConfig {
 
   private final int executorThreads;
 
+  private final OptionalInt beaconApiExecutorThreads;
+  private final OptionalInt beaconApiReadinessExecutorThreads;
+
   private final boolean isLocalSlashingProtectionSynchronizedModeEnabled;
   private final boolean dvtSelectionsEndpointEnabled;
   private final boolean attestationsV2ApisEnabled;
@@ -150,6 +153,8 @@ private ValidatorConfig(
       final Optional<BLSPublicKey> builderRegistrationPublicKeyOverride,
       final int executorMaxQueueSize,
       final int executorThreads,
+      final OptionalInt beaconApiExecutorThreads,
+      final OptionalInt beaconApiReadinessExecutorThreads,
       final Optional<String> sentryNodeConfigurationFile,
       final boolean isLocalSlashingProtectionSynchronizedModeEnabled,
       final boolean dvtSelectionsEndpointEnabled,
@@ -193,6 +198,8 @@ private ValidatorConfig(
     this.builderRegistrationPublicKeyOverride = builderRegistrationPublicKeyOverride;
     this.executorMaxQueueSize = executorMaxQueueSize;
     this.executorThreads = executorThreads;
+    this.beaconApiExecutorThreads = beaconApiExecutorThreads;
+    this.beaconApiReadinessExecutorThreads = beaconApiReadinessExecutorThreads;
     this.sentryNodeConfigurationFile = sentryNodeConfigurationFile;
     this.isLocalSlashingProtectionSynchronizedModeEnabled =
         isLocalSlashingProtectionSynchronizedModeEnabled;
@@ -346,6 +353,14 @@ public int getExecutorThreads() {
     return executorThreads;
   }
 
+  public OptionalInt getBeaconApiExecutorThreads() {
+    return beaconApiExecutorThreads;
+  }
+
+  public OptionalInt getBeaconApiReadinessExecutorThreads() {
+    return beaconApiReadinessExecutorThreads;
+  }
+
   public Optional<String> getSentryNodeConfigurationFile() {
     return sentryNodeConfigurationFile;
   }
@@ -418,6 +433,8 @@ public static final class Builder {
     private Optional<UInt64> builderRegistrationTimestampOverride = Optional.empty();
     private Optional<BLSPublicKey> builderRegistrationPublicKeyOverride = Optional.empty();
     private OptionalInt executorMaxQueueSize = OptionalInt.empty();
+    private OptionalInt beaconApiExecutorThreads = OptionalInt.empty();
+    private OptionalInt beaconApiReadinessExecutorThreads = OptionalInt.empty();
     private Optional<String> sentryNodeConfigurationFile = Optional.empty();
     private int executorThreads = DEFAULT_VALIDATOR_EXECUTOR_THREADS;
     private boolean isLocalSlashingProtectionSynchronizedModeEnabled =
@@ -648,14 +665,25 @@ public Builder builderRegistrationPublicKeyOverride(
       return this;
     }
 
-    public Builder executorMaxQueueSize(final int executorMaxQueueSize) {
-      this.executorMaxQueueSize = OptionalInt.of(executorMaxQueueSize);
+    public Builder executorMaxQueueSize(final OptionalInt executorMaxQueueSize) {
+      this.executorMaxQueueSize = executorMaxQueueSize;
+      return this;
+    }
+
+    public Builder beaconApiExecutorThreads(final OptionalInt beaconApiExecutorThreads) {
+      this.beaconApiExecutorThreads = beaconApiExecutorThreads;
+      return this;
+    }
+
+    public Builder beaconApiReadinessExecutorThreads(
+        final OptionalInt beaconApiReadinessExecutorThreads) {
+      this.beaconApiReadinessExecutorThreads = beaconApiReadinessExecutorThreads;
       return this;
     }
 
     public Builder executorMaxQueueSizeIfDefault(final int executorMaxQueueSize) {
       if (this.executorMaxQueueSize.isEmpty()) {
-        this.executorMaxQueueSize(executorMaxQueueSize);
+        this.executorMaxQueueSize = OptionalInt.of(executorMaxQueueSize);
       }
       return this;
     }
@@ -723,6 +751,8 @@ public ValidatorConfig build() {
           builderRegistrationPublicKeyOverride,
           executorMaxQueueSize.orElse(DEFAULT_EXECUTOR_MAX_QUEUE_SIZE),
           executorThreads,
+          beaconApiExecutorThreads,
+          beaconApiReadinessExecutorThreads,
           sentryNodeConfigurationFile,
           isLocalSlashingProtectionSynchronizedModeEnabled,
           dvtSelectionsEndpointEnabled,
diff --git a/validator/remote/src/main/java/tech/pegasys/teku/validator/remote/RemoteBeaconNodeApi.java b/validator/remote/src/main/java/tech/pegasys/teku/validator/remote/RemoteBeaconNodeApi.java
@@ -84,9 +84,7 @@ public static BeaconNodeApi create(
 
     final int remoteNodeCount = failoverEndpoints.size() + 1;
 
-    final int apiMaxThreads =
-        calculateAPIMaxThreads(
-            remoteNodeCount, validatorConfig.isFailoversPublishSignedDutiesEnabled());
+    final int apiMaxThreads = calculateAPIMaxThreads(remoteNodeCount, validatorConfig);
     final AsyncRunner asyncRunner =
         services.createAsyncRunner(
             "validatorBeaconAPI", apiMaxThreads, MAX_API_EXECUTOR_QUEUE_SIZE);
@@ -97,7 +95,8 @@ public static BeaconNodeApi create(
     } else {
       // Use a separate async runner for the readiness-related api calls, so that they do not
       // interfere with the critical path API calls.
-      final int apiMaxReadinessThreads = calculateReadinessAPIMaxThreads(remoteNodeCount);
+      final int apiMaxReadinessThreads =
+          calculateReadinessAPIMaxThreads(remoteNodeCount, validatorConfig);
       readinessAsyncRunner =
           services.createAsyncRunner(
               "validatorBeaconAPIReadiness", apiMaxReadinessThreads, MAX_API_EXECUTOR_QUEUE_SIZE);
@@ -188,14 +187,23 @@ public static BeaconNodeApi create(
   }
 
   public static int calculateAPIMaxThreads(
-      final int remoteNodeCount, final boolean isFailoversPublishSignedDutiesEnabled) {
+      final int remoteNodeCount, final ValidatorConfig validatorConfig) {
+    if (validatorConfig.getBeaconApiExecutorThreads().isPresent()) {
+      return validatorConfig.getBeaconApiExecutorThreads().getAsInt();
+    }
     // Let's allow at least 4 parallel requests per remote node when publishing signed duties to
     // failovers, to reduce the risk of being affected by a slow remote node.
-    final int remoteNodeCountMultiplier = isFailoversPublishSignedDutiesEnabled ? 4 : 2;
+    final int remoteNodeCountMultiplier =
+        validatorConfig.isFailoversPublishSignedDutiesEnabled() ? 4 : 2;
     return Math.max(5, remoteNodeCount * remoteNodeCountMultiplier);
   }
 
-  public static int calculateReadinessAPIMaxThreads(final int remoteNodeCount) {
+  public static int calculateReadinessAPIMaxThreads(
+      final int remoteNodeCount, final ValidatorConfig validatorConfig) {
+    if (validatorConfig.getBeaconApiReadinessExecutorThreads().isPresent()) {
+      return validatorConfig.getBeaconApiReadinessExecutorThreads().getAsInt();
+    }
+
     // we call two methods per remote node to check readiness, so we need at least 2 threads to call
     // them in parallel
     return remoteNodeCount * 2;
diff --git a/validator/remote/src/main/java/tech/pegasys/teku/validator/remote/sentry/SentryBeaconNodeApi.java b/validator/remote/src/main/java/tech/pegasys/teku/validator/remote/sentry/SentryBeaconNodeApi.java
@@ -81,17 +81,17 @@ public static BeaconNodeApi create(
 
     final int apiMaxThreads =
         calculateAPIMaxThreads(
-            dutiesProviderNodeConfig.getEndpointsAsURIs().size(),
-            validatorConfig.isFailoversPublishSignedDutiesEnabled());
+            dutiesProviderNodeConfig.getEndpointsAsURIs().size(), validatorConfig);
     final AsyncRunner asyncRunner =
         services.createAsyncRunner(
             "validatorBeaconAPI", apiMaxThreads, MAX_API_EXECUTOR_QUEUE_SIZE);
 
-    final int apiMaxReadinessThreads =
-        calculateReadinessAPIMaxThreads(dutiesProviderNodeConfig.getEndpointsAsURIs().size());
+    final int apiReadinessMaxThreads =
+        calculateReadinessAPIMaxThreads(
+            dutiesProviderNodeConfig.getEndpointsAsURIs().size(), validatorConfig);
     final AsyncRunner readinessAsyncRunner =
         services.createAsyncRunner(
-            "validatorBeaconAPIReadiness", apiMaxReadinessThreads, MAX_API_EXECUTOR_QUEUE_SIZE);
+            "validatorBeaconAPIReadiness", apiReadinessMaxThreads, MAX_API_EXECUTOR_QUEUE_SIZE);
 
     final RemoteValidatorApiChannel dutiesProviderPrimaryValidatorApiChannel =
         createPrimaryValidatorApiChannel(
diff --git a/validator/remote/src/test/java/tech/pegasys/teku/validator/remote/RemoteBeaconNodeApiTest.java b/validator/remote/src/test/java/tech/pegasys/teku/validator/remote/RemoteBeaconNodeApiTest.java
@@ -14,11 +14,22 @@
 package tech.pegasys.teku.validator.remote;
 
 import static org.assertj.core.api.AssertionsForClassTypes.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static tech.pegasys.teku.validator.remote.RemoteBeaconNodeApi.MAX_API_EXECUTOR_QUEUE_SIZE;
 
 import java.net.URI;
+import java.net.URISyntaxException;
 import java.util.List;
+import java.util.OptionalInt;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import tech.pegasys.teku.infrastructure.events.EventChannels;
+import tech.pegasys.teku.infrastructure.metrics.StubMetricsSystem;
 import tech.pegasys.teku.service.serviceutils.ServiceConfig;
 import tech.pegasys.teku.spec.Spec;
 import tech.pegasys.teku.spec.TestSpecFactory;
@@ -27,9 +38,17 @@
 class RemoteBeaconNodeApiTest {
 
   private final ServiceConfig serviceConfig = mock(ServiceConfig.class);
+  private final EventChannels eventChannels = mock(EventChannels.class);
   private final ValidatorConfig validatorConfig = mock(ValidatorConfig.class);
+  private final StubMetricsSystem metricsSystem = new StubMetricsSystem();
   private final Spec spec = TestSpecFactory.createMinimalAltair();
 
+  @BeforeEach
+  void setUp() {
+    when(serviceConfig.getEventChannels()).thenReturn(eventChannels);
+    when(serviceConfig.getMetricsSystem()).thenReturn(metricsSystem);
+  }
+
   @Test
   void producesExceptionWhenInvalidUrlPassed() {
     assertThatThrownBy(
@@ -38,4 +57,65 @@ void producesExceptionWhenInvalidUrlPassed() {
                     serviceConfig, validatorConfig, spec, List.of(new URI("notvalid"))))
         .hasMessageContaining("Failed to convert remote api endpoint");
   }
+
+  @Test
+  void shouldConfigureBeaconApiPools_primaryOnly() throws URISyntaxException {
+    RemoteBeaconNodeApi.create(
+        serviceConfig, validatorConfig, spec, List.of(new URI("https://localhost")));
+
+    verify(serviceConfig).createAsyncRunner("validatorBeaconAPI", 5, MAX_API_EXECUTOR_QUEUE_SIZE);
+
+    verify(serviceConfig, times(1)).createAsyncRunner(anyString(), anyInt(), anyInt());
+  }
+
+  @Test
+  void shouldConfigureBeaconApiPools_primaryAndSecondary() throws URISyntaxException {
+    RemoteBeaconNodeApi.create(
+        serviceConfig,
+        validatorConfig,
+        spec,
+        List.of(new URI("https://localhost"), new URI("https://secondary")));
+
+    verify(serviceConfig).createAsyncRunner("validatorBeaconAPI", 5, MAX_API_EXECUTOR_QUEUE_SIZE);
+    verify(serviceConfig)
+        .createAsyncRunner("validatorBeaconAPIReadiness", 4, MAX_API_EXECUTOR_QUEUE_SIZE);
+
+    verify(serviceConfig, times(2)).createAsyncRunner(anyString(), anyInt(), anyInt());
+  }
+
+  @Test
+  void shouldConfigureBeaconApiPools_primaryAndSecondaryWithFailoverPublish()
+      throws URISyntaxException {
+    when(validatorConfig.isFailoversPublishSignedDutiesEnabled()).thenReturn(true);
+
+    RemoteBeaconNodeApi.create(
+        serviceConfig,
+        validatorConfig,
+        spec,
+        List.of(new URI("https://localhost"), new URI("https://secondary")));
+
+    verify(serviceConfig).createAsyncRunner("validatorBeaconAPI", 8, MAX_API_EXECUTOR_QUEUE_SIZE);
+    verify(serviceConfig)
+        .createAsyncRunner("validatorBeaconAPIReadiness", 4, MAX_API_EXECUTOR_QUEUE_SIZE);
+
+    verify(serviceConfig, times(2)).createAsyncRunner(anyString(), anyInt(), anyInt());
+  }
+
+  @Test
+  void shouldConfigureBeaconApiPools_configOverride() throws URISyntaxException {
+    when(validatorConfig.getBeaconApiExecutorThreads()).thenReturn(OptionalInt.of(2));
+    when(validatorConfig.getBeaconApiReadinessExecutorThreads()).thenReturn(OptionalInt.of(4));
+
+    RemoteBeaconNodeApi.create(
+        serviceConfig,
+        validatorConfig,
+        spec,
+        List.of(new URI("https://localhost"), new URI("https://secondary")));
+
+    verify(serviceConfig).createAsyncRunner("validatorBeaconAPI", 2, MAX_API_EXECUTOR_QUEUE_SIZE);
+    verify(serviceConfig)
+        .createAsyncRunner("validatorBeaconAPIReadiness", 4, MAX_API_EXECUTOR_QUEUE_SIZE);
+
+    verify(serviceConfig, times(2)).createAsyncRunner(anyString(), anyInt(), anyInt());
+  }
 }
