diff --git a/CHANGELOG.md b/CHANGELOG.md index d17cb5d93..da0ccd840 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,10 @@ # Changelog +## Next release +### Features Added +- Bulk loading keys from Azure Key Vault is significantly faster. Secrets and keys are now fetched concurrently, up to a configurable limit, while the vault is still being listed, instead of one 25-item page at a time. New options: `--azure-bulk-load-max-concurrency` (default 20) and `--azure-bulk-load-timeout` (default 900 seconds). Retries for throttled or transient failures are left to the Azure SDK's own default retry policy. + +--- ## 26.7.0 ### Features Added - Support for Hashicorp Vault Kubernetes authentication [PR 1195](https://github.com/Consensys/web3signer/pull/1195) diff --git a/commandline/build.gradle b/commandline/build.gradle index 3529800ca..ec5ff55cc 100644 --- a/commandline/build.gradle +++ b/commandline/build.gradle @@ -16,6 +16,7 @@ dependencies { implementation project(":common") implementation project(":core") + implementation project(":keystorage") implementation project(":slashing-protection") implementation project(":signing") implementation 'info.picocli:picocli' diff --git a/commandline/src/main/java/tech/pegasys/web3signer/commandline/PicoCliAzureKeyVaultParameters.java b/commandline/src/main/java/tech/pegasys/web3signer/commandline/PicoCliAzureKeyVaultParameters.java index f8716417d..b8dcae138 100644 --- a/commandline/src/main/java/tech/pegasys/web3signer/commandline/PicoCliAzureKeyVaultParameters.java +++ b/commandline/src/main/java/tech/pegasys/web3signer/commandline/PicoCliAzureKeyVaultParameters.java @@ -12,9 +12,11 @@ */ package tech.pegasys.web3signer.commandline; +import tech.pegasys.web3signer.keystorage.azure.BulkLoadOptions; import tech.pegasys.web3signer.signing.config.AzureAuthenticationMode; import tech.pegasys.web3signer.signing.config.AzureKeyVaultParameters; +import java.time.Duration; import java.util.LinkedHashMap; import java.util.Map; @@ -71,6 +73,22 @@ public class PicoCliAzureKeyVaultParameters implements AzureKeyVaultParameters { paramLabel = "") private long timeout = 60; + @Option( + names = {"--azure-bulk-load-max-concurrency"}, + description = + "Maximum number of concurrent requests to Azure Key Vault during bulk key loading " + + "(Default: ${DEFAULT-VALUE})", + paramLabel = "") + private int maxConcurrency = BulkLoadOptions.DEFAULT_MAX_CONCURRENCY; + + @Option( + names = {"--azure-bulk-load-timeout"}, + description = + "Overall time budget for bulk loading keys from Azure Key Vault (in seconds). Keys not " + + "loaded within it are reported as errors (Default: ${DEFAULT-VALUE})", + paramLabel = "") + private long bulkLoadTimeout = BulkLoadOptions.DEFAULT_DEADLINE.toSeconds(); + @CommandLine.Option( names = {"--azure-tags"}, mapFallbackValue = "", @@ -119,4 +137,9 @@ public long getTimeout() { public Map getTags() { return tags; } + + @Override + public BulkLoadOptions getBulkLoadOptions() { + return new BulkLoadOptions(maxConcurrency, Duration.ofSeconds(bulkLoadTimeout)); + } } diff --git a/core/src/main/java/tech/pegasys/web3signer/core/Eth1Runner.java b/core/src/main/java/tech/pegasys/web3signer/core/Eth1Runner.java index 03d6e1cd5..7c2e35e81 100644 --- a/core/src/main/java/tech/pegasys/web3signer/core/Eth1Runner.java +++ b/core/src/main/java/tech/pegasys/web3signer/core/Eth1Runner.java @@ -183,8 +183,10 @@ private MappedResults bulkLoadSigners( private MappedResults bulkLoadAzureKeys( AzureKeyVaultFactory azureKeyVaultFactory, AzureKeyVaultSignerFactory azureSignerFactory) { - LOG.info("Bulk loading keys from Azure key vault ... "); final AzureKeyVaultParameters azureKeyVaultConfig = eth1Config.getAzureKeyVaultConfig(); + LOG.info( + "Bulk loading keys from Azure key vault (max concurrency: {}) ... ", + azureKeyVaultConfig.getBulkLoadOptions().maxConcurrency()); final AzureKeyVault azureKeyVault = azureKeyVaultFactory.createAzureKeyVault( azureKeyVaultConfig.getClientId(), diff --git a/core/src/main/java/tech/pegasys/web3signer/core/Eth2Runner.java b/core/src/main/java/tech/pegasys/web3signer/core/Eth2Runner.java index a482d2550..2a28b1760 100644 --- a/core/src/main/java/tech/pegasys/web3signer/core/Eth2Runner.java +++ b/core/src/main/java/tech/pegasys/web3signer/core/Eth2Runner.java @@ -220,7 +220,9 @@ private MappedResults bulkLoadSigners( final AzureKeyVaultFactory azureKeyVaultFactory) { MappedResults results = MappedResults.newSetInstance(); if (azureKeyVaultParameters.isAzureKeyVaultEnabled()) { - LOG.info("Bulk loading keys from Azure key vault ... "); + LOG.info( + "Bulk loading keys from Azure key vault (max concurrency: {}) ... ", + azureKeyVaultParameters.getBulkLoadOptions().maxConcurrency()); /* Note: Azure supports 25K bytes per secret. https://learn.microsoft.com/en-us/azure/key-vault/secrets/about-secrets Each raw bls private key in hex format is approximately 100 bytes. We should store about 200 or fewer @@ -373,6 +375,7 @@ final MappedResults loadAzureSigners( return null; } }, - azureKeyVaultParameters.getTags()); + azureKeyVaultParameters.getTags(), + azureKeyVaultParameters.getBulkLoadOptions()); } } diff --git a/keystorage/src/main/java/tech/pegasys/web3signer/keystorage/azure/AzureKeyVault.java b/keystorage/src/main/java/tech/pegasys/web3signer/keystorage/azure/AzureKeyVault.java index e4e61915c..587c9d633 100644 --- a/keystorage/src/main/java/tech/pegasys/web3signer/keystorage/azure/AzureKeyVault.java +++ b/keystorage/src/main/java/tech/pegasys/web3signer/keystorage/azure/AzureKeyVault.java @@ -24,12 +24,11 @@ import java.util.Map; import java.util.Optional; import java.util.Set; -import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; -import java.util.concurrent.atomic.AtomicInteger; import java.util.function.BiFunction; import java.util.function.Function; import java.util.stream.Collectors; +import java.util.stream.Stream; import com.azure.core.credential.AccessToken; import com.azure.core.credential.TokenCredential; @@ -54,17 +53,15 @@ import com.azure.security.keyvault.secrets.models.SecretProperties; import com.google.common.annotations.VisibleForTesting; import io.vertx.core.json.JsonObject; -import org.apache.logging.log4j.LogManager; -import org.apache.logging.log4j.Logger; import org.apache.tuweni.bytes.Bytes; public class AzureKeyVault { - private static final Logger LOG = LogManager.getLogger(); private final TokenCredential tokenCredential; private final SecretClient secretClient; private final KeyClient keyClient; private static final List SCOPE = List.of("https://vault.azure.net/.default"); + private final TokenRequestContext tokenRequestContext = new TokenRequestContext().setScopes(SCOPE); @@ -182,45 +179,28 @@ public static String constructAzureKeyVaultUrl(final String keyVaultName) { * * @param mapper The mapper function to transform secret values to type R. * @param tags Map of tags. Only secrets which contains all the tags entries are processed. + * @param options Concurrency and retry parameters for the load. * @return Mapped results containing the converted secrets and error count. * @param The result type of mapper function. */ public MappedResults mapSecrets( - final BiFunction mapper, final Map tags) { - final Set result = ConcurrentHashMap.newKeySet(); - final AtomicInteger errorCount = new AtomicInteger(0); - try { - final PagedIterable secretsPagedIterable = - secretClient.listPropertiesOfSecrets(); - - secretsPagedIterable - .streamByPage() - .forEach( - keyPage -> - keyPage.getValue().parallelStream() - .filter(secretProperties -> secretPropertiesPredicate(tags, secretProperties)) - .forEach( - sp -> { - try { - final KeyVaultSecret secret = secretClient.getSecret(sp.getName()); - final MappedResults multiResult = - SecretValueMapperUtil.mapSecretValue( - mapper, sp.getName(), secret.getValue()); - result.addAll(multiResult.getValues()); - errorCount.addAndGet(multiResult.getErrorCount()); - } catch (final Exception e) { - LOG.warn( - "Failed to map secret '{}' to requested object type.", - sp.getName()); - errorCount.incrementAndGet(); - } - })); - - } catch (final Exception e) { - LOG.error("Unexpected error during Azure map-secrets", e); - errorCount.incrementAndGet(); - } - return MappedResults.newInstance(result, errorCount.intValue()); + final BiFunction mapper, + final Map tags, + final BulkLoadOptions options) { + // the listing is consumed lazily, so secrets are fetched while later pages are still listed + final Stream secrets = + secretClient.listPropertiesOfSecrets().stream() + .filter(sp -> secretPropertiesPredicate(tags, sp)); + + return new ConcurrentBulkLoader(options) + .load( + "Azure secrets bulk load", + secrets, + SecretProperties::getName, + sp -> { + final KeyVaultSecret secret = secretClient.getSecret(sp.getName()); + return SecretValueMapperUtil.mapSecretValue(mapper, sp.getName(), secret.getValue()); + }); } /** @@ -229,39 +209,24 @@ public MappedResults mapSecrets( * * @param mapper Mapper function to transform Azure KeyProperties to type R * @param tags Map of tags. Only keys which contains all the tags entries are processed. + * @param options Concurrency and retry parameters for the load. * @return Mapped results containing the converted keys and error count. * @param The result type of mapper function. */ public MappedResults mapKeyProperties( - final Function mapper, final Map tags) { - final Set result = ConcurrentHashMap.newKeySet(); - final AtomicInteger errorCount = new AtomicInteger(0); - try { - keyClient - .listPropertiesOfKeys() - .streamByPage() - .forEach( - keyPage -> - keyPage.getValue().parallelStream() - .filter(keyProperties -> keyPropertiesPredicate(tags, keyProperties)) - .forEach( - kp -> { - try { - final R value = mapper.apply(kp); - result.add(value); - } catch (final Exception e) { - LOG.warn( - "Failed to map keyProperties '{}' to requested object type.", - kp.getName()); - errorCount.incrementAndGet(); - } - })); - } catch (final Exception e) { - LOG.error("Unexpected error during Azure mapKeyProperties", e); - errorCount.incrementAndGet(); - } - - return MappedResults.newInstance(result, errorCount.intValue()); + final Function mapper, + final Map tags, + final BulkLoadOptions options) { + // the listing is consumed lazily, so keys are mapped while later pages are still listed + final Stream keys = + keyClient.listPropertiesOfKeys().stream().filter(kp -> keyPropertiesPredicate(tags, kp)); + + return new ConcurrentBulkLoader(options) + .load( + "Azure keys bulk load", + keys, + KeyProperties::getName, + kp -> MappedResults.newInstance(Set.of(mapper.apply(kp)), 0)); } /** diff --git a/keystorage/src/main/java/tech/pegasys/web3signer/keystorage/azure/BulkLoadOptions.java b/keystorage/src/main/java/tech/pegasys/web3signer/keystorage/azure/BulkLoadOptions.java new file mode 100644 index 000000000..ce265a702 --- /dev/null +++ b/keystorage/src/main/java/tech/pegasys/web3signer/keystorage/azure/BulkLoadOptions.java @@ -0,0 +1,41 @@ +/* + * Copyright 2026 ConsenSys AG. + * + * 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.web3signer.keystorage.azure; + +import java.time.Duration; + +import com.google.common.base.Preconditions; + +/** + * Tuning parameters for a bulk load performed by {@link ConcurrentBulkLoader}. + * + * @param maxConcurrency upper bound on concurrent requests. + * @param deadline overall time budget for the load. Items not loaded within it are reported as + * errors rather than silently dropped. + */ +public record BulkLoadOptions(int maxConcurrency, Duration deadline) { + + public static final int DEFAULT_MAX_CONCURRENCY = 20; + public static final Duration DEFAULT_DEADLINE = Duration.ofMinutes(15); + + /** Options applied when a caller does not configure bulk loading. */ + public static final BulkLoadOptions DEFAULT = + new BulkLoadOptions(DEFAULT_MAX_CONCURRENCY, DEFAULT_DEADLINE); + + /** Validates the parameters. */ + public BulkLoadOptions { + Preconditions.checkArgument(maxConcurrency > 0, "maxConcurrency must be positive"); + Preconditions.checkArgument( + !deadline.isNegative() && !deadline.isZero(), "deadline must be positive"); + } +} diff --git a/keystorage/src/main/java/tech/pegasys/web3signer/keystorage/azure/ConcurrentBulkLoader.java b/keystorage/src/main/java/tech/pegasys/web3signer/keystorage/azure/ConcurrentBulkLoader.java new file mode 100644 index 000000000..d6f93c027 --- /dev/null +++ b/keystorage/src/main/java/tech/pegasys/web3signer/keystorage/azure/ConcurrentBulkLoader.java @@ -0,0 +1,280 @@ +/* + * Copyright 2026 ConsenSys AG. + * + * 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.web3signer.keystorage.azure; + +import tech.pegasys.web3signer.keystorage.common.MappedResults; + +import java.time.Duration; +import java.util.Iterator; +import java.util.Map; +import java.util.Queue; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Semaphore; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Function; +import java.util.stream.Stream; + +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +/** + * Loads a stream of items from a remote vault concurrently. + * + *

Items are consumed lazily, so a vault which pages its listing continues listing while earlier + * items are being fetched. Concurrency is bounded by {@link BulkLoadOptions#maxConcurrency()}. + * Retries for transient failures, such as throttling, are left to the vault SDK's own retry policy; + * a failure reaching this loader is treated as final for that item. + * + *

Every item taken from the stream is accounted for: it contributes either a value or an error. + * Interrupts and an expired deadline both abandon the load with a non-zero error count rather than + * reporting a partial load as a complete one. + */ +public class ConcurrentBulkLoader { + + private static final Logger LOG = LogManager.getLogger(); + + private static final Duration PROGRESS_INTERVAL = Duration.ofSeconds(5); + private static final int MAX_LOGGED_FAILURES = 5; + + private static final String LABEL_DEADLINE = "abandoned: deadline exceeded"; + private static final String LABEL_INTERRUPTED = "abandoned: interrupted"; + private static final String LABEL_INCOMPLETE = "abandoned: did not complete"; + private static final String LABEL_LISTING = "listing failed"; + + private final BulkLoadOptions options; + + /** + * Creates a loader for a single vault. + * + * @param options concurrency and deadline parameters for the load. + */ + public ConcurrentBulkLoader(final BulkLoadOptions options) { + this.options = options; + } + + /** + * Applies {@code work} to every item of {@code items} concurrently. + * + * @param description used to identify this load in log messages. + * @param items the items to load. Consumed lazily and closed on completion. + * @param nameOf names an item for logging. + * @param work loads a single item. May return several values, and its own error count, for one + * item. + * @return the loaded values, and the number of errors encountered. + * @param the type of item being loaded. + * @param the type of value produced for an item. + */ + public MappedResults load( + final String description, + final Stream items, + final Function nameOf, + final Function> work) { + return new Run(description, nameOf, work).execute(items); + } + + private final class Run { + private final String description; + private final Function nameOf; + private final Function> work; + + private final Set values = ConcurrentHashMap.newKeySet(); + private final AtomicInteger listed = new AtomicInteger(); + private final AtomicInteger succeeded = new AtomicInteger(); + private final AtomicInteger failures = new AtomicInteger(); + private final AtomicInteger mappingErrors = new AtomicInteger(); + private final Map failuresByLabel = new ConcurrentHashMap<>(); + private final Queue loggedFailures = new ConcurrentLinkedQueue<>(); + private final AtomicBoolean incompleteListing = new AtomicBoolean(); + // fair, so a producer waiting to dispatch is not starved by workers releasing and re-acquiring + private final Semaphore limiter = new Semaphore(options.maxConcurrency(), true); + + private final long startNanos = System.nanoTime(); + private final long deadlineNanos = startNanos + options.deadline().toNanos(); + + private Run( + final String description, + final Function nameOf, + final Function> work) { + this.description = description; + this.nameOf = nameOf; + this.work = work; + } + + private MappedResults execute(final Stream items) { + final Thread progressReporter = startProgressReporter(); + try (final Stream stream = items; + final ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) { + try { + final Iterator iterator = stream.iterator(); + while (iterator.hasNext()) { + final T item = iterator.next(); + listed.incrementAndGet(); + if (!dispatch(executor, item)) { + incompleteListing.set(true); + break; + } + } + } catch (final Exception e) { + // a listing failure loses an unknown number of items, so the load cannot be trusted + incompleteListing.set(true); + recordFailure(LABEL_LISTING, "

", e); + LOG.error( + "{}: failed to list items, {} listed before the failure", + description, + listed.get(), + e); + } + LOG.debug("{}: listed {} items, awaiting in-flight work", description, listed.get()); + } finally { + progressReporter.interrupt(); + } + + reconcile(); + final int errorCount = failures.get() + mappingErrors.get(); + logSummary(errorCount); + return MappedResults.newInstance(values, errorCount); + } + + /** + * Acquires a permit and submits the item, blocking while the configured concurrency is + * saturated. Blocking here is what applies backpressure to the listing. + * + * @return false when the remaining items should not be dispatched. + */ + private boolean dispatch(final ExecutorService executor, final T item) { + if (deadlineExpired()) { + recordFailure(LABEL_DEADLINE, nameOf.apply(item), null); + LOG.error( + "{}: deadline of {} exceeded after {} items", + description, + options.deadline(), + listed.get()); + return false; + } + try { + limiter.acquire(); + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + recordFailure(LABEL_INTERRUPTED, nameOf.apply(item), e); + return false; + } + try { + executor.submit(() -> process(item)); + return true; + } catch (final RuntimeException e) { + limiter.release(); + recordFailure(e.getClass().getSimpleName(), nameOf.apply(item), e); + return false; + } + } + + private void process(final T item) { + try { + attempt(item); + } finally { + limiter.release(); + } + } + + private void attempt(final T item) { + final String name = nameOf.apply(item); + try { + final MappedResults mapped = work.apply(item); + values.addAll(mapped.getValues()); + mappingErrors.addAndGet(mapped.getErrorCount()); + succeeded.incrementAndGet(); + } catch (final Exception e) { + recordFailure(e.getClass().getSimpleName(), name, e); + } + } + + private boolean deadlineExpired() { + return System.nanoTime() - deadlineNanos >= 0; + } + + private void recordFailure(final String label, final String name, final Exception cause) { + failures.incrementAndGet(); + failuresByLabel.computeIfAbsent(label, _ -> new AtomicInteger()).incrementAndGet(); + if (loggedFailures.size() < MAX_LOGGED_FAILURES) { + loggedFailures.add(name + ": " + label); + LOG.warn("{}: failed to load '{}' - {}", description, name, label, cause); + } else { + LOG.debug("{}: failed to load '{}' - {}", description, name, label, cause); + } + } + + /** Ensures every listed item is accounted for, even if its task never ran to completion. */ + private void reconcile() { + final int unaccounted = listed.get() - (succeeded.get() + failures.get()); + if (unaccounted > 0) { + failures.addAndGet(unaccounted); + failuresByLabel + .computeIfAbsent(LABEL_INCOMPLETE, _ -> new AtomicInteger()) + .addAndGet(unaccounted); + } + } + + private void logSummary(final int errorCount) { + final Duration elapsed = Duration.ofNanos(System.nanoTime() - startNanos); + LOG.info( + "{}: loaded {} values from {} of the {} items {} in {}s", + description, + values.size(), + succeeded.get(), + listed.get(), + incompleteListing.get() ? "listed before the load was abandoned" : "listed", + elapsed.toSeconds()); + if (errorCount > 0) { + LOG.error( + "{}: {} items failed and {} values could not be mapped. Failures by cause: {}. First failures: {}", + description, + failures.get(), + mappingErrors.get(), + failuresByLabel, + loggedFailures); + } + if (incompleteListing.get()) { + LOG.error( + "{}: the load was abandoned before every item had been listed, so an unknown number of" + + " items were never attempted", + description); + } + } + + private Thread startProgressReporter() { + return Thread.ofVirtual() + .name("bulk-load-progress") + .start( + () -> { + try { + while (true) { + Thread.sleep(PROGRESS_INTERVAL); + LOG.info( + "{}: listed {}, loaded {}, failed {}", + description, + listed.get(), + succeeded.get(), + failures.get()); + } + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + } + }); + } + } +} diff --git a/keystorage/src/test/java/tech/pegasys/web3signer/keystorage/azure/AzureKeyVaultTest.java b/keystorage/src/test/java/tech/pegasys/web3signer/keystorage/azure/AzureKeyVaultTest.java index 3b406d0e8..fd37f1890 100644 --- a/keystorage/src/test/java/tech/pegasys/web3signer/keystorage/azure/AzureKeyVaultTest.java +++ b/keystorage/src/test/java/tech/pegasys/web3signer/keystorage/azure/AzureKeyVaultTest.java @@ -105,7 +105,7 @@ void secretsCanBeMappedUsingCustomMappingFunction() { // mapSecrets can convert multiple secrets under single key. final MappedResults> result = - azureKeyVault.mapSecrets(SimpleEntry::new, Collections.emptyMap()); + azureKeyVault.mapSecrets(SimpleEntry::new, Collections.emptyMap(), BulkLoadOptions.DEFAULT); final Collection> entries = result.getValues(); // the number of entries should match the available secret values count @@ -125,7 +125,8 @@ void keyPropertiesCanBeMappedUsingCustomMappingFunction() { assertThat(availableKeyNames).isNotEmpty(); final MappedResults result = - azureKeyVault.mapKeyProperties(KeyProperties::getName, Collections.emptyMap()); + azureKeyVault.mapKeyProperties( + KeyProperties::getName, Collections.emptyMap(), BulkLoadOptions.DEFAULT); final Collection entries = result.getValues(); assertThat(entries).containsExactlyInAnyOrderElementsOf(availableKeyNames); assertThat(result.getErrorCount()).isZero(); @@ -138,7 +139,7 @@ void mapSecretsUsingTags() { CLIENT_ID, CLIENT_SECRET, TENANT_ID, VAULT_NAME, azureExecutor, AZURE_DEFAULT_TIMEOUT); final MappedResults> result = - azureKeyVault.mapSecrets(SimpleEntry::new, Map.of("ENV", "TEST")); + azureKeyVault.mapSecrets(SimpleEntry::new, Map.of("ENV", "TEST"), BulkLoadOptions.DEFAULT); // The Azure key vault is set up with at least one Secret with above tag. assertThat(result.getValues()).isNotEmpty(); // we should not encounter any error count @@ -152,7 +153,8 @@ void mapKeyPropertiesUsingTags() { CLIENT_ID, CLIENT_SECRET, TENANT_ID, VAULT_NAME, azureExecutor, AZURE_DEFAULT_TIMEOUT); final MappedResults result = - azureKeyVault.mapKeyProperties(KeyProperties::getName, Map.of("ENV", "TEST")); + azureKeyVault.mapKeyProperties( + KeyProperties::getName, Map.of("ENV", "TEST"), BulkLoadOptions.DEFAULT); // The Azure key vault is set up with at least one Key with above tag. final Collection entries = result.getValues(); assertThat(entries).isNotEmpty(); @@ -167,7 +169,8 @@ void mapSecretsWhenTagsDoesNotExist() { CLIENT_ID, CLIENT_SECRET, TENANT_ID, VAULT_NAME, azureExecutor, AZURE_DEFAULT_TIMEOUT); final MappedResults> result = - azureKeyVault.mapSecrets(SimpleEntry::new, Map.of("INVALID_TAG", "INVALID_TEST")); + azureKeyVault.mapSecrets( + SimpleEntry::new, Map.of("INVALID_TAG", "INVALID_TEST"), BulkLoadOptions.DEFAULT); // The secret vault is not expected to have any secrets with above tags. assertThat(result.getValues()).isEmpty(); @@ -184,7 +187,7 @@ void mapKeyPropertiesWhenTagsDoesNotExist() { final MappedResults result = azureKeyVault.mapKeyProperties( - KeyProperties::getName, Map.of("INVALID_TAG", "INVALID_TEST")); + KeyProperties::getName, Map.of("INVALID_TAG", "INVALID_TEST"), BulkLoadOptions.DEFAULT); // The key vault is not expected to have any secrets with above tags. assertThat(result.getValues()).isEmpty(); @@ -211,7 +214,8 @@ void mapSecretsThrowsAwayObjectsWhichFailMapper() { } return new SimpleEntry<>(name, value); }, - Collections.emptyMap()); + Collections.emptyMap(), + BulkLoadOptions.DEFAULT); assertThat(result.getErrorCount()).isOne(); assertThat(result.getValues()).isNotEmpty(); } @@ -233,7 +237,8 @@ void mapKeyPropertiesThrowsAwayObjectsWhichFailMapper() { } return keyProperties.getName(); }, - Collections.emptyMap()); + Collections.emptyMap(), + BulkLoadOptions.DEFAULT); assertThat(result.getErrorCount()).isOne(); assertThat(result.getValues()).isNotEmpty(); @@ -247,7 +252,8 @@ void mapSecretsThrowsAwayObjectsWhichMapToNull() { // map all remote Secrets values to null - this is to simulate failure in mapping function final MappedResults> result = - azureKeyVault.mapSecrets((name, value) -> null, Collections.emptyMap()); + azureKeyVault.mapSecrets( + (name, value) -> null, Collections.emptyMap(), BulkLoadOptions.DEFAULT); assertThat(result.getErrorCount()).isNotZero(); assertThat(result.getValues()).isEmpty(); @@ -261,7 +267,8 @@ void mapKeyPropertiesThrowsAwayObjectsWhichMapToNull() { // map all remote Keys to null - this is to simulate failure in mapping function final MappedResults result = - azureKeyVault.mapKeyProperties(keyProperties -> null, Collections.emptyMap()); + azureKeyVault.mapKeyProperties( + keyProperties -> null, Collections.emptyMap(), BulkLoadOptions.DEFAULT); assertThat(result.getErrorCount()).isNotZero(); assertThat(result.getValues()).isEmpty(); diff --git a/keystorage/src/test/java/tech/pegasys/web3signer/keystorage/azure/ConcurrentBulkLoaderTest.java b/keystorage/src/test/java/tech/pegasys/web3signer/keystorage/azure/ConcurrentBulkLoaderTest.java new file mode 100644 index 000000000..3900a8a07 --- /dev/null +++ b/keystorage/src/test/java/tech/pegasys/web3signer/keystorage/azure/ConcurrentBulkLoaderTest.java @@ -0,0 +1,206 @@ +/* + * Copyright 2026 ConsenSys AG. + * + * 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.web3signer.keystorage.azure; + +import static org.assertj.core.api.Assertions.assertThat; + +import tech.pegasys.web3signer.keystorage.common.MappedResults; + +import java.time.Duration; +import java.util.List; +import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ThreadLocalRandom; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Function; +import java.util.stream.IntStream; +import java.util.stream.Stream; + +import org.junit.jupiter.api.Test; + +class ConcurrentBulkLoaderTest { + + private static final BulkLoadOptions FAST_OPTIONS = + new BulkLoadOptions(8, Duration.ofSeconds(60)); + + @Test + void loadsEveryItem() { + final MappedResults results = load(items(50), item -> value(item)); + + assertThat(results.getValues()).hasSize(50); + assertThat(results.getErrorCount()).isZero(); + } + + @Test + void keepsEveryValueReturnedForASingleItem() { + final MappedResults results = + load(items(3), item -> MappedResults.newInstance(Set.of(item + "-a", item + "-b"), 0)); + + assertThat(results.getValues()).hasSize(6); + assertThat(results.getErrorCount()).isZero(); + } + + @Test + void mappingErrorsReportedByTheWorkAreCounted() { + final MappedResults results = + load(items(4), item -> MappedResults.newInstance(Set.of(item), 2)); + + assertThat(results.getValues()).hasSize(4); + assertThat(results.getErrorCount()).isEqualTo(8); + } + + @Test + void aFailureIsNotRetried() { + final AtomicInteger attempts = new AtomicInteger(); + + final MappedResults results = + load( + items(1), + item -> { + attempts.incrementAndGet(); + throw new RuntimeException("failed"); + }); + + assertThat(results.getValues()).isEmpty(); + assertThat(results.getErrorCount()).isOne(); + assertThat(attempts).hasValue(1); + } + + @Test + void everyItemContributesEitherAValueOrAnError() { + final MappedResults results = + load( + items(300), + item -> { + if (ThreadLocalRandom.current().nextInt(4) == 0) { + throw new RuntimeException("unclassified"); + } + return value(item); + }); + + assertThat(results.getValues().size() + results.getErrorCount()).isEqualTo(300); + } + + @Test + void concurrencyNeverExceedsTheConfiguredMaximum() { + final AtomicInteger inFlight = new AtomicInteger(); + final AtomicInteger peak = new AtomicInteger(); + + final MappedResults results = + load( + items(200), + item -> { + peak.accumulateAndGet(inFlight.incrementAndGet(), Math::max); + try { + Thread.sleep(Duration.ofMillis(2)); + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + inFlight.decrementAndGet(); + } + return value(item); + }); + + assertThat(results.getErrorCount()).isZero(); + assertThat(peak.get()).isPositive().isLessThanOrEqualTo(FAST_OPTIONS.maxConcurrency()); + } + + @Test + void anExpiredDeadlineIsReportedRatherThanSilentlyDroppingItems() { + final BulkLoadOptions shortDeadline = new BulkLoadOptions(2, Duration.ofMillis(150)); + + final MappedResults results = + new ConcurrentBulkLoader(shortDeadline) + .load( + "test", + items(500), + Function.identity(), + item -> { + try { + Thread.sleep(Duration.ofMillis(5)); + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + } + return value(item); + }); + + assertThat(results.getValues()).hasSizeLessThan(500); + assertThat(results.getErrorCount()).isPositive(); + } + + @Test + void aListingFailureIsReportedAsAnError() { + final Stream failingListing = + IntStream.range(0, 100) + .mapToObj( + index -> { + if (index == 10) { + throw new RuntimeException("listing failed"); + } + return "item-" + index; + }); + + final MappedResults results = + new ConcurrentBulkLoader(FAST_OPTIONS) + .load("test", failingListing, Function.identity(), ConcurrentBulkLoaderTest::value); + + assertThat(results.getValues()).hasSize(10); + assertThat(results.getErrorCount()).isOne(); + } + + @Test + void anInterruptedLoadIsReportedRatherThanSilentlyDroppingItems() throws InterruptedException { + final CountDownLatch started = new CountDownLatch(1); + final AtomicInteger errorCount = new AtomicInteger(-1); + final AtomicInteger loadedCount = new AtomicInteger(-1); + + final Thread loader = + new Thread( + () -> { + final MappedResults results = + load( + items(500), + item -> { + started.countDown(); + try { + Thread.sleep(Duration.ofMillis(50)); + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + } + return value(item); + }); + loadedCount.set(results.getValues().size()); + errorCount.set(results.getErrorCount()); + }); + loader.start(); + assertThat(started.await(10, java.util.concurrent.TimeUnit.SECONDS)).isTrue(); + loader.interrupt(); + loader.join(Duration.ofSeconds(30)); + + assertThat(loadedCount.get()).isNotNegative().isLessThan(500); + assertThat(errorCount.get()).isPositive(); + } + + private static MappedResults load( + final Stream items, final Function> work) { + return new ConcurrentBulkLoader(FAST_OPTIONS).load("test", items, Function.identity(), work); + } + + private static Stream items(final int count) { + return IntStream.range(0, count).mapToObj(index -> "item-" + index); + } + + private static MappedResults value(final String item) { + return MappedResults.newInstance(List.of(item), 0); + } +} diff --git a/signing/src/main/java/tech/pegasys/web3signer/signing/bulkloading/SecpAzureBulkLoader.java b/signing/src/main/java/tech/pegasys/web3signer/signing/bulkloading/SecpAzureBulkLoader.java index b68498889..63e1b23f6 100644 --- a/signing/src/main/java/tech/pegasys/web3signer/signing/bulkloading/SecpAzureBulkLoader.java +++ b/signing/src/main/java/tech/pegasys/web3signer/signing/bulkloading/SecpAzureBulkLoader.java @@ -34,7 +34,8 @@ public SecpAzureBulkLoader( public MappedResults load(final AzureKeyVaultParameters azureKeyVaultParameters) { return azureKeyVault.mapKeyProperties( kp -> createSigner(kp.getName(), azureKeyVaultParameters), - azureKeyVaultParameters.getTags()); + azureKeyVaultParameters.getTags(), + azureKeyVaultParameters.getBulkLoadOptions()); } private EthSecpArtifactSigner createSigner( diff --git a/signing/src/main/java/tech/pegasys/web3signer/signing/config/AzureKeyVaultParameters.java b/signing/src/main/java/tech/pegasys/web3signer/signing/config/AzureKeyVaultParameters.java index 08d9a4f63..65440dd2a 100644 --- a/signing/src/main/java/tech/pegasys/web3signer/signing/config/AzureKeyVaultParameters.java +++ b/signing/src/main/java/tech/pegasys/web3signer/signing/config/AzureKeyVaultParameters.java @@ -12,6 +12,8 @@ */ package tech.pegasys.web3signer.signing.config; +import tech.pegasys.web3signer.keystorage.azure.BulkLoadOptions; + import java.util.Map; public interface AzureKeyVaultParameters { @@ -31,4 +33,14 @@ public interface AzureKeyVaultParameters { Map getTags(); long getTimeout(); + + /** + * Concurrency and retry parameters applied when bulk loading from the vault. Only meaningful for + * bulk loading, so single key configurations keep the defaults. + * + * @return the options to apply to a bulk load. + */ + default BulkLoadOptions getBulkLoadOptions() { + return BulkLoadOptions.DEFAULT; + } }