Repo analysis of uncontended register behaviour (#101185)

Today repository analysis verifies that a register behaves correctly
under contention, retrying until successful, but it turns out that some
repository implementations cannot even perform uncontended register
writes correctly which may cause endless retries in the contended case.
This commit adds another repository analyser which verifies that
uncontended register writes work correctly on the first attempt.

docs/changelog/101185.yaml

@@ -0,0 +1,5 @@
+pr: 101185
+summary: Repo analysis of uncontended register behaviour
+area: Snapshot/Restore
+type: enhancement
+issues: []

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -144,7 +144,7 @@ public class TransportVersions {
     public static final TransportVersion PIPELINES_IN_BULK_RESPONSE_ADDED = def(8_519_00_0);
     public static final TransportVersion PLUGIN_DESCRIPTOR_STRING_VERSION = def(8_520_00_0);
     public static final TransportVersion TOO_MANY_SCROLL_CONTEXTS_EXCEPTION_ADDED = def(8_521_00_0);
-
+    public static final TransportVersion UNCONTENDED_REGISTER_ANALYSIS_ADDED = def(8_522_00_0);
     /*
      * STOP! READ THIS FIRST! No, really,
      *        ____ _____ ___  ____  _        ____  _____    _    ____    _____ _   _ ___ ____    _____ ___ ____  ____ _____ _

test/fixtures/s3-fixture/src/main/java/fixture/s3/S3HttpHandler.java

@@ -102,8 +102,10 @@ public class S3HttpHandler implements HttpHandler {
                 uploadsList.append("<MaxUploads>10000</MaxUploads>");
                 uploadsList.append("<IsTruncated>false</IsTruncated>");
 
-                for (MultipartUpload value : uploads.values()) {
-                    value.appendXml(uploadsList);
+                for (final var multipartUpload : uploads.values()) {
+                    if (multipartUpload.getPath().startsWith(prefix)) {
+                        multipartUpload.appendXml(uploadsList);
+                    }
                 }
 
                 uploadsList.append("</ListMultipartUploadsResult>");

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/operator/DefaultOperatorOnlyRegistry.java

@@ -46,6 +46,7 @@ public class DefaultOperatorOnlyRegistry implements OperatorOnlyRegistry {
         "cluster:admin/repository/analyze/blob",
         "cluster:admin/repository/analyze/blob/read",
         "cluster:admin/repository/analyze/register",
+        "cluster:admin/repository/analyze/register/uncontended",
         // Node shutdown APIs are operator only
         "cluster:admin/shutdown/create",
         "cluster:admin/shutdown/get",

x-pack/plugin/snapshot-repo-test-kit/src/internalClusterTest/java/org/elasticsearch/repositories/blobstore/testkit/RepositoryAnalysisFailureIT.java

@@ -63,6 +63,7 @@ import java.util.stream.Collectors;
 
 import static org.elasticsearch.repositories.blobstore.testkit.ContendedRegisterAnalyzeAction.bytesFromLong;
 import static org.elasticsearch.repositories.blobstore.testkit.ContendedRegisterAnalyzeAction.longFromBytes;
+import static org.hamcrest.Matchers.allOf;
 import static org.hamcrest.Matchers.anEmptyMap;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
@@ -304,7 +305,7 @@ public class RepositoryAnalysisFailureIT extends AbstractSnapshotIntegTestCase {
             private final AtomicBoolean registerWasCorrupted = new AtomicBoolean();
 
             @Override
-            public BytesReference onCompareAndExchange(BytesRegister register, BytesReference expected, BytesReference updated) {
+            public BytesReference onContendedCompareAndExchange(BytesRegister register, BytesReference expected, BytesReference updated) {
                 if (registerWasCorrupted.compareAndSet(false, true)) {
                     register.updateAndGet(bytes -> bytesFromLong(longFromBytes(bytes) + 1));
                 }
@@ -321,7 +322,7 @@ public class RepositoryAnalysisFailureIT extends AbstractSnapshotIntegTestCase {
         final long expectedMax = Math.max(request.getConcurrency(), internalCluster().getNodeNames().length);
         blobStore.setDisruption(new Disruption() {
             @Override
-            public BytesReference onCompareAndExchange(BytesRegister register, BytesReference expected, BytesReference updated) {
+            public BytesReference onContendedCompareAndExchange(BytesRegister register, BytesReference expected, BytesReference updated) {
                 if (randomBoolean() && sawSpuriousValue.compareAndSet(false, true)) {
                     final var currentValue = longFromBytes(register.get());
                     if (currentValue == expectedMax) {
@@ -357,8 +358,9 @@ public class RepositoryAnalysisFailureIT extends AbstractSnapshotIntegTestCase {
 
         blobStore.setDisruption(new Disruption() {
             @Override
-            public boolean compareAndExchangeReturnsWitness() {
-                return false;
+            public boolean compareAndExchangeReturnsWitness(String key) {
+                // let uncontended accesses succeed but all contended ones fail
+                return isContendedRegisterKey(key) == false;
             }
         });
         final var exception = expectThrows(RepositoryVerificationException.class, () -> analyseRepository(request));
@@ -369,6 +371,22 @@ public class RepositoryAnalysisFailureIT extends AbstractSnapshotIntegTestCase {
         );
     }
 
+    public void testFailsIfAllRegisterOperationsInconclusive() {
+        final RepositoryAnalyzeAction.Request request = new RepositoryAnalyzeAction.Request("test-repo");
+        blobStore.setDisruption(new Disruption() {
+            @Override
+            public boolean compareAndExchangeReturnsWitness(String key) {
+                return false;
+            }
+        });
+        final var exception = expectThrows(RepositoryVerificationException.class, () -> analyseRepository(request));
+        assertThat(exception.getMessage(), containsString("analysis failed"));
+        assertThat(
+            asInstanceOf(RepositoryVerificationException.class, ExceptionsHelper.unwrapCause(exception.getCause())).getMessage(),
+            allOf(containsString("uncontended register operation failed"), containsString("did not observe any value"))
+        );
+    }
+
     private void analyseRepository(RepositoryAnalyzeAction.Request request) {
         client().execute(RepositoryAnalyzeAction.INSTANCE, request).actionGet(30L, TimeUnit.SECONDS);
     }
@@ -486,11 +504,11 @@ public class RepositoryAnalysisFailureIT extends AbstractSnapshotIntegTestCase {
             return false;
         }
 
-        default boolean compareAndExchangeReturnsWitness() {
+        default boolean compareAndExchangeReturnsWitness(String key) {
             return true;
         }
 
-        default BytesReference onCompareAndExchange(BytesRegister register, BytesReference expected, BytesReference updated) {
+        default BytesReference onContendedCompareAndExchange(BytesRegister register, BytesReference expected, BytesReference updated) {
             return register.compareAndExchange(expected, updated);
         }
     }
@@ -663,13 +681,28 @@ public class RepositoryAnalysisFailureIT extends AbstractSnapshotIntegTestCase {
             ActionListener<OptionalBytesReference> listener
         ) {
             assertPurpose(purpose);
-            if (disruption.compareAndExchangeReturnsWitness()) {
+            final boolean isContendedRegister = isContendedRegisterKey(key); // validate key
+            if (disruption.compareAndExchangeReturnsWitness(key)) {
                 final var register = registers.computeIfAbsent(key, ignored -> new BytesRegister());
-                listener.onResponse(OptionalBytesReference.of(disruption.onCompareAndExchange(register, expected, updated)));
+                if (isContendedRegister) {
+                    listener.onResponse(OptionalBytesReference.of(disruption.onContendedCompareAndExchange(register, expected, updated)));
+                } else {
+                    listener.onResponse(OptionalBytesReference.of(register.compareAndExchange(expected, updated)));
+                }
             } else {
                 listener.onResponse(OptionalBytesReference.MISSING);
             }
         }
     }
 
+    static boolean isContendedRegisterKey(String key) {
+        if (key.startsWith(RepositoryAnalyzeAction.CONTENDED_REGISTER_NAME_PREFIX)) {
+            return true;
+        }
+        if (key.startsWith(RepositoryAnalyzeAction.UNCONTENDED_REGISTER_NAME_PREFIX)) {
+            return false;
+        }
+        return fail(null, "unknown register: %s", key);
+    }
+
 }

x-pack/plugin/snapshot-repo-test-kit/src/internalClusterTest/java/org/elasticsearch/repositories/blobstore/testkit/RepositoryAnalysisSuccessIT.java

@@ -57,6 +57,7 @@ import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
 
+import static org.elasticsearch.repositories.blobstore.testkit.RepositoryAnalysisFailureIT.isContendedRegisterKey;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.lessThanOrEqualTo;
@@ -442,12 +443,15 @@ public class RepositoryAnalysisSuccessIT extends AbstractSnapshotIntegTestCase {
         @Override
         public void getRegister(OperationPurpose purpose, String key, ActionListener<OptionalBytesReference> listener) {
             assertPurpose(purpose);
-            if (firstRegisterRead.compareAndSet(true, false) && randomBoolean() && randomBoolean()) {
+            if (isContendedRegisterKey(key) && firstRegisterRead.compareAndSet(true, false) && randomBoolean() && randomBoolean()) {
+                // it's ok if _contended_ register accesses are a little disrupted since they retry until success, however,
                 // only fail the first read, we must not fail the final check
                 listener.onResponse(OptionalBytesReference.EMPTY);
             } else if (randomBoolean()) {
+                // read the register directly
                 listener.onResponse(OptionalBytesReference.of(registers.computeIfAbsent(key, ignored -> new BytesRegister()).get()));
             } else {
+                // read using a compare-and-exchange that cannot succeed, but which returns the current value anyway
                 final var bogus = randomFrom(BytesArray.EMPTY, new BytesArray(new byte[] { randomByte() }));
                 compareAndExchangeRegister(purpose, key, bogus, bogus, listener);
             }
@@ -462,17 +466,22 @@ public class RepositoryAnalysisSuccessIT extends AbstractSnapshotIntegTestCase {
             ActionListener<OptionalBytesReference> listener
         ) {
             assertPurpose(purpose);
-            firstRegisterRead.set(false);
-            if (updated.length() > 1 && randomBoolean() && randomBoolean()) {
-                // updated.length() > 1 so we don't fail the final check because we know there can be no concurrent operations at that point
-                listener.onResponse(OptionalBytesReference.MISSING);
-            } else {
-                listener.onResponse(
-                    OptionalBytesReference.of(
-                        registers.computeIfAbsent(key, ignored -> new BytesRegister()).compareAndExchange(expected, updated)
-                    )
-                );
+            if (isContendedRegisterKey(key)) {
+                // it's ok if _contended_ register accesses are a little disrupted since they retry until success
+
+                firstRegisterRead.set(false);
+                if (updated.length() > 1 && randomBoolean() && randomBoolean()) {
+                    // updated.length() > 1 so the final check succeeds because we know there can be no concurrent operations at that point
+                    listener.onResponse(OptionalBytesReference.MISSING);
+                    return;
+                }
             }
+
+            listener.onResponse(
+                OptionalBytesReference.of(
+                    registers.computeIfAbsent(key, ignored -> new BytesRegister()).compareAndExchange(expected, updated)
+                )
+            );
         }
     }
 

x-pack/plugin/snapshot-repo-test-kit/src/main/java/org/elasticsearch/repositories/blobstore/testkit/RepositoryAnalyzeAction.java

@@ -11,8 +11,8 @@ import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.ElasticsearchTimeoutException;
 import org.elasticsearch.ExceptionsHelper;
+import org.elasticsearch.TransportVersion;
 import org.elasticsearch.TransportVersions;
-import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionListenerResponseHandler;
 import org.elasticsearch.action.ActionRequest;
@@ -74,6 +74,7 @@ import java.util.Queue;
 import java.util.Random;
 import java.util.Set;
 import java.util.concurrent.Semaphore;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
@@ -98,6 +99,7 @@ public class RepositoryAnalyzeAction extends ActionType<RepositoryAnalyzeAction.
     public static final RepositoryAnalyzeAction INSTANCE = new RepositoryAnalyzeAction();
     public static final String NAME = "cluster:admin/repository/analyze";
 
+    static final String UNCONTENDED_REGISTER_NAME_PREFIX = "test-register-uncontended-";
     static final String CONTENDED_REGISTER_NAME_PREFIX = "test-register-contended-";
 
     private RepositoryAnalyzeAction() {
@@ -147,6 +149,7 @@ public class RepositoryAnalyzeAction extends ActionType<RepositoryAnalyzeAction.
                     (CancellableTask) task,
                     request,
                     state.nodes(),
+                    state.getMinTransportVersion(),
                     threadPool::relativeTimeInMillis,
                     listener
                 ).run();
@@ -368,6 +371,7 @@ public class RepositoryAnalyzeAction extends ActionType<RepositoryAnalyzeAction.
         private final CancellableTask task;
         private final Request request;
         private final DiscoveryNodes discoveryNodes;
+        private final TransportVersion minClusterTransportVersion;
         private final LongSupplier currentTimeMillisSupplier;
         private final ActionListener<Response> listener;
         private final SubscribableListener<Void> cancellationListener;
@@ -391,6 +395,7 @@ public class RepositoryAnalyzeAction extends ActionType<RepositoryAnalyzeAction.
             CancellableTask task,
             Request request,
             DiscoveryNodes discoveryNodes,
+            TransportVersion minClusterTransportVersion,
             LongSupplier currentTimeMillisSupplier,
             ActionListener<Response> listener
         ) {
@@ -399,6 +404,7 @@ public class RepositoryAnalyzeAction extends ActionType<RepositoryAnalyzeAction.
             this.task = task;
             this.request = request;
             this.discoveryNodes = discoveryNodes;
+            this.minClusterTransportVersion = minClusterTransportVersion;
             this.currentTimeMillisSupplier = currentTimeMillisSupplier;
             this.timeoutTimeMillis = currentTimeMillisSupplier.getAsLong() + request.getTimeout().millis();
 
@@ -482,22 +488,35 @@ public class RepositoryAnalyzeAction extends ActionType<RepositoryAnalyzeAction.
             final Random random = new Random(request.getSeed());
             final List<DiscoveryNode> nodes = getSnapshotNodes(discoveryNodes);
 
-            final String contendedRegisterName = CONTENDED_REGISTER_NAME_PREFIX + UUIDs.randomBase64UUID(random);
-            try (
-                var registerRefs = new RefCountingRunnable(finalRegisterValueVerifier(contendedRegisterName, random, requestRefs.acquire()))
-            ) {
-                final int registerOperations = Math.max(nodes.size(), request.getConcurrency());
-                for (int i = 0; i < registerOperations; i++) {
-                    final ContendedRegisterAnalyzeAction.Request registerAnalyzeRequest = new ContendedRegisterAnalyzeAction.Request(
-                        request.getRepositoryName(),
-                        blobPath,
-                        contendedRegisterName,
-                        registerOperations,
-                        random.nextInt((registerOperations + 1) * 2)
-                    );
-                    final DiscoveryNode node = nodes.get(i < nodes.size() ? i : random.nextInt(nodes.size()));
-                    final Releasable registerRef = registerRefs.acquire();
-                    queue.add(ref -> runContendedRegisterAnalysis(Releasables.wrap(registerRef, ref), registerAnalyzeRequest, node));
+            if (minClusterTransportVersion.onOrAfter(TransportVersions.V_8_8_0)) {
+                final String contendedRegisterName = CONTENDED_REGISTER_NAME_PREFIX + UUIDs.randomBase64UUID(random);
+                final AtomicBoolean contendedRegisterAnalysisComplete = new AtomicBoolean();
+                try (
+                    var registerRefs = new RefCountingRunnable(
+                        finalRegisterValueVerifier(
+                            contendedRegisterName,
+                            random,
+                            Releasables.wrap(requestRefs.acquire(), () -> contendedRegisterAnalysisComplete.set(true))
+                        )
+                    )
+                ) {
+                    final int registerOperations = Math.max(nodes.size(), request.getConcurrency());
+                    for (int i = 0; i < registerOperations; i++) {
+                        final ContendedRegisterAnalyzeAction.Request registerAnalyzeRequest = new ContendedRegisterAnalyzeAction.Request(
+                            request.getRepositoryName(),
+                            blobPath,
+                            contendedRegisterName,
+                            registerOperations,
+                            random.nextInt((registerOperations + 1) * 2)
+                        );
+                        final DiscoveryNode node = nodes.get(i < nodes.size() ? i : random.nextInt(nodes.size()));
+                        final Releasable registerRef = registerRefs.acquire();
+                        queue.add(ref -> runContendedRegisterAnalysis(Releasables.wrap(registerRef, ref), registerAnalyzeRequest, node));
+                    }
+                }
+
+                if (minClusterTransportVersion.onOrAfter(TransportVersions.UNCONTENDED_REGISTER_ANALYSIS_ADDED)) {
+                    new UncontendedRegisterAnalysis(new Random(random.nextLong()), nodes, contendedRegisterAnalysisComplete).run();
                 }
             }
 
@@ -600,7 +619,7 @@ public class RepositoryAnalyzeAction extends ActionType<RepositoryAnalyzeAction.
         }
 
         private void runContendedRegisterAnalysis(Releasable ref, ContendedRegisterAnalyzeAction.Request request, DiscoveryNode node) {
-            if (node.getVersion().onOrAfter(Version.V_8_8_0) && isRunning()) {
+            if (isRunning()) {
                 transportService.sendChildRequest(
                     node,
                     ContendedRegisterAnalyzeAction.NAME,
@@ -690,6 +709,59 @@ public class RepositoryAnalyzeAction extends ActionType<RepositoryAnalyzeAction.
             };
         }
 
+        private class UncontendedRegisterAnalysis implements Runnable {
+            private final Random random;
+            private final String registerName;
+            private final List<DiscoveryNode> nodes;
+            private final AtomicBoolean otherAnalysisComplete;
+            private int currentValue; // actions run in strict sequence so no need for synchronization
+
+            UncontendedRegisterAnalysis(Random random, List<DiscoveryNode> nodes, AtomicBoolean otherAnalysisComplete) {
+                this.random = random;
+                this.registerName = UNCONTENDED_REGISTER_NAME_PREFIX + UUIDs.randomBase64UUID(random);
+                this.nodes = nodes;
+                this.otherAnalysisComplete = otherAnalysisComplete;
+            }
+
+            private final ActionListener<ActionResponse.Empty> stepListener = new ActionListener<>() {
+                @Override
+                public void onResponse(ActionResponse.Empty ignored) {
+                    currentValue += 1;
+                    run();
+                }
+
+                @Override
+                public void onFailure(Exception e) {
+                    fail(e);
+                }
+            };
+
+            @Override
+            public void run() {
+                if (isRunning() == false) {
+                    return;
+                }
+
+                // complete at least request.getConcurrency() steps, but we may as well keep running for longer too
+                if (currentValue > request.getConcurrency() && otherAnalysisComplete.get()) {
+                    return;
+                }
+
+                transportService.sendChildRequest(
+                    nodes.get(currentValue < nodes.size() ? currentValue : random.nextInt(nodes.size())),
+                    UncontendedRegisterAnalyzeAction.NAME,
+                    new UncontendedRegisterAnalyzeAction.Request(request.getRepositoryName(), blobPath, registerName, currentValue),
+                    task,
+                    TransportRequestOptions.EMPTY,
+                    new ActionListenerResponseHandler<>(
+                        ActionListener.releaseAfter(stepListener, requestRefs.acquire()),
+                        in -> ActionResponse.Empty.INSTANCE,
+                        TransportResponseHandler.TRANSPORT_WORKER
+                    )
+                );
+            }
+        }
+
         private void runCleanUp() {
             transportService.getThreadPool().executor(ThreadPool.Names.SNAPSHOT).execute(ActionRunnable.wrap(listener, l -> {
                 final long listingStartTimeNanos = System.nanoTime();

x-pack/plugin/snapshot-repo-test-kit/src/main/java/org/elasticsearch/repositories/blobstore/testkit/SnapshotRepositoryTestKit.java

@@ -34,7 +34,8 @@ public class SnapshotRepositoryTestKit extends Plugin implements ActionPlugin {
             new ActionHandler<>(RepositoryAnalyzeAction.INSTANCE, RepositoryAnalyzeAction.TransportAction.class),
             new ActionHandler<>(BlobAnalyzeAction.INSTANCE, BlobAnalyzeAction.TransportAction.class),
             new ActionHandler<>(GetBlobChecksumAction.INSTANCE, GetBlobChecksumAction.TransportAction.class),
-            new ActionHandler<>(ContendedRegisterAnalyzeAction.INSTANCE, ContendedRegisterAnalyzeAction.TransportAction.class)
+            new ActionHandler<>(ContendedRegisterAnalyzeAction.INSTANCE, ContendedRegisterAnalyzeAction.TransportAction.class),
+            new ActionHandler<>(UncontendedRegisterAnalyzeAction.INSTANCE, UncontendedRegisterAnalyzeAction.TransportAction.class)
         );
     }
 

x-pack/plugin/snapshot-repo-test-kit/src/main/java/org/elasticsearch/repositories/blobstore/testkit/UncontendedRegisterAnalyzeAction.java

@@ -0,0 +1,215 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.repositories.blobstore.testkit;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.elasticsearch.TransportVersions;
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.ActionRequest;
+import org.elasticsearch.action.ActionRequestValidationException;
+import org.elasticsearch.action.ActionResponse;
+import org.elasticsearch.action.ActionType;
+import org.elasticsearch.action.support.ActionFilters;
+import org.elasticsearch.action.support.HandledTransportAction;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.blobstore.BlobContainer;
+import org.elasticsearch.common.blobstore.BlobPath;
+import org.elasticsearch.common.blobstore.OperationPurpose;
+import org.elasticsearch.common.blobstore.OptionalBytesReference;
+import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.repositories.RepositoriesService;
+import org.elasticsearch.repositories.Repository;
+import org.elasticsearch.repositories.RepositoryVerificationException;
+import org.elasticsearch.repositories.blobstore.BlobStoreRepository;
+import org.elasticsearch.tasks.CancellableTask;
+import org.elasticsearch.tasks.Task;
+import org.elasticsearch.tasks.TaskId;
+import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.transport.TransportService;
+
+import java.io.IOException;
+import java.util.Map;
+
+import static org.elasticsearch.repositories.blobstore.testkit.ContendedRegisterAnalyzeAction.bytesFromLong;
+import static org.elasticsearch.repositories.blobstore.testkit.ContendedRegisterAnalyzeAction.longFromBytes;
+
+public class UncontendedRegisterAnalyzeAction extends ActionType<ActionResponse.Empty> {
+
+    private static final Logger logger = LogManager.getLogger(UncontendedRegisterAnalyzeAction.class);
+
+    public static final UncontendedRegisterAnalyzeAction INSTANCE = new UncontendedRegisterAnalyzeAction();
+    public static final String NAME = "cluster:admin/repository/analyze/register/uncontended";
+
+    private UncontendedRegisterAnalyzeAction() {
+        super(NAME, in -> ActionResponse.Empty.INSTANCE);
+    }
+
+    public static class TransportAction extends HandledTransportAction<Request, ActionResponse.Empty> {
+
+        private static final Logger logger = UncontendedRegisterAnalyzeAction.logger;
+
+        private final RepositoriesService repositoriesService;
+
+        @Inject
+        public TransportAction(TransportService transportService, ActionFilters actionFilters, RepositoriesService repositoriesService) {
+            super(
+                NAME,
+                transportService,
+                actionFilters,
+                Request::new,
+                transportService.getThreadPool().executor(ThreadPool.Names.SNAPSHOT)
+            );
+            this.repositoriesService = repositoriesService;
+        }
+
+        @Override
+        protected void doExecute(Task task, Request request, ActionListener<ActionResponse.Empty> outerListener) {
+            final ActionListener<Void> listener = ActionListener.assertOnce(outerListener.map(ignored -> ActionResponse.Empty.INSTANCE));
+            final Repository repository = repositoriesService.repository(request.getRepositoryName());
+            if (repository instanceof BlobStoreRepository == false) {
+                throw new IllegalArgumentException("repository [" + request.getRepositoryName() + "] is not a blob-store repository");
+            }
+            if (repository.isReadOnly()) {
+                throw new IllegalArgumentException("repository [" + request.getRepositoryName() + "] is read-only");
+            }
+            final BlobStoreRepository blobStoreRepository = (BlobStoreRepository) repository;
+            final BlobPath path = blobStoreRepository.basePath().add(request.getContainerPath());
+            final BlobContainer blobContainer = blobStoreRepository.blobStore().blobContainer(path);
+
+            logger.trace("handling [{}]", request);
+
+            assert task instanceof CancellableTask;
+            blobContainer.compareAndExchangeRegister(
+                OperationPurpose.REPOSITORY_ANALYSIS,
+                request.getRegisterName(),
+                bytesFromLong(request.getExpectedValue()),
+                bytesFromLong(request.getExpectedValue() + 1),
+                new ActionListener<>() {
+                    @Override
+                    public void onResponse(OptionalBytesReference optionalBytesReference) {
+                        ActionListener.completeWith(listener, () -> {
+                            if (optionalBytesReference.isPresent() == false) {
+                                throw new RepositoryVerificationException(
+                                    repository.getMetadata().name(),
+                                    Strings.format(
+                                        "uncontended register operation failed: expected [%d] but did not observe any value",
+                                        request.getExpectedValue()
+                                    )
+                                );
+                            }
+
+                            final var witness = longFromBytes(optionalBytesReference.bytesReference());
+                            if (witness != request.getExpectedValue()) {
+                                throw new RepositoryVerificationException(
+                                    repository.getMetadata().name(),
+                                    Strings.format(
+                                        "uncontended register operation failed: expected [%d] but observed [%d]",
+                                        request.getExpectedValue(),
+                                        witness
+                                    )
+                                );
+                            }
+
+                            return null;
+                        });
+                    }
+
+                    @Override
+                    public void onFailure(Exception e) {
+                        if (e instanceof UnsupportedOperationException) {
+                            // Registers are not supported on all repository types, and that's ok.
+                            listener.onResponse(null);
+                        } else {
+                            listener.onFailure(e);
+                        }
+                    }
+                }
+            );
+        }
+    }
+
+    public static class Request extends ActionRequest {
+        private final String repositoryName;
+        private final String containerPath;
+        private final String registerName;
+        private final long expectedValue;
+
+        public Request(String repositoryName, String containerPath, String registerName, long expectedValue) {
+            this.repositoryName = repositoryName;
+            this.containerPath = containerPath;
+            this.registerName = registerName;
+            this.expectedValue = expectedValue;
+        }
+
+        public Request(StreamInput in) throws IOException {
+            super(in);
+            assert in.getTransportVersion().onOrAfter(TransportVersions.UNCONTENDED_REGISTER_ANALYSIS_ADDED);
+            repositoryName = in.readString();
+            containerPath = in.readString();
+            registerName = in.readString();
+            expectedValue = in.readVLong();
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            assert out.getTransportVersion().onOrAfter(TransportVersions.UNCONTENDED_REGISTER_ANALYSIS_ADDED);
+            super.writeTo(out);
+            out.writeString(repositoryName);
+            out.writeString(containerPath);
+            out.writeString(registerName);
+            out.writeVLong(expectedValue);
+        }
+
+        @Override
+        public ActionRequestValidationException validate() {
+            return null;
+        }
+
+        public String getRepositoryName() {
+            return repositoryName;
+        }
+
+        public String getContainerPath() {
+            return containerPath;
+        }
+
+        public String getRegisterName() {
+            return registerName;
+        }
+
+        public long getExpectedValue() {
+            return expectedValue;
+        }
+
+        @Override
+        public Task createTask(long id, String type, String action, TaskId parentTaskId, Map<String, String> headers) {
+            return new CancellableTask(id, type, action, getDescription(), parentTaskId, headers);
+        }
+
+        @Override
+        public String toString() {
+            return getDescription();
+        }
+
+        @Override
+        public String getDescription() {
+            return Strings.format(
+                """
+                    UncontendedRegisterAnalyzeAction.Request{\
+                    repositoryName='%s', containerPath='%s', registerName='%s', expectedValue='%d'}""",
+                repositoryName,
+                containerPath,
+                registerName,
+                expectedValue
+            );
+        }
+    }
+}