Fix S3RepositoryAnalysisRestIT (#126593)

- Translate a 404 during a multipart copy into a `FileNotFoundException`

- Use multiple threads in `S3HttpHandler` to avoid `CopyObject`/`PutObject` deadlock

Closes #126576

Author: DaveCTurner

modules/repository-s3/src/main/java/org/elasticsearch/repositories/s3/S3BlobContainer.java

@@ -10,6 +10,7 @@
 package org.elasticsearch.repositories.s3;
 
 import com.amazonaws.AmazonClientException;
+import com.amazonaws.AmazonServiceException;
 import com.amazonaws.services.s3.AmazonS3;
 import com.amazonaws.services.s3.model.AbortMultipartUploadRequest;
 import com.amazonaws.services.s3.model.AmazonS3Exception;
@@ -376,12 +377,11 @@ public void copyBlob(
                     SocketAccess.doPrivilegedVoid(() -> { clientReference.client().copyObject(copyRequest); });
                 }
             }
-        } catch (final AmazonClientException e) {
-            if (e instanceof AmazonS3Exception amazonS3Exception) {
-                if (amazonS3Exception.getStatusCode() == RestStatus.NOT_FOUND.getStatus()) {
-                    final var sourceKey = s3SourceBlobContainer.buildKey(sourceBlobName);
-                    throw new NoSuchFileException("Copy source [" + sourceKey + "] not found: " + amazonS3Exception.getMessage());
-                }
+        } catch (final Exception e) {
+            if (e instanceof AmazonServiceException ase && ase.getStatusCode() == RestStatus.NOT_FOUND.getStatus()) {
+                throw new NoSuchFileException(
+                    "Copy source [" + s3SourceBlobContainer.buildKey(sourceBlobName) + "] not found: " + ase.getMessage()
+                );
             }
             throw new IOException("Unable to copy object [" + blobName + "] from [" + sourceBlobContainer + "][" + sourceBlobName + "]", e);
         }
@@ -618,8 +618,10 @@ private void executeMultipart(
                 SocketAccess.doPrivilegedVoid(() -> clientReference.client().completeMultipartUpload(complRequest));
             }
             success = true;
-
-        } catch (final AmazonClientException e) {
+        } catch (final Exception e) {
+            if (e instanceof AmazonServiceException ase && ase.getStatusCode() == RestStatus.NOT_FOUND.getStatus()) {
+                throw new NoSuchFileException(blobName, null, e.getMessage());
+            }
             throw new IOException("Unable to upload or copy object [" + blobName + "] using multipart upload", e);
         } finally {
             if ((success == false) && Strings.hasLength(uploadId.get())) {

muted-tests.yml

@@ -420,9 +420,6 @@ tests:
 - class: org.elasticsearch.repositories.blobstore.testkit.rest.SnapshotRepoTestKitClientYamlTestSuiteIT
   method: test {p0=/10_analyze/Analysis without details}
   issue: https://github.com/elastic/elasticsearch/issues/126569
-- class: org.elasticsearch.repositories.blobstore.testkit.analyze.S3RepositoryAnalysisRestIT
-  method: testRepositoryAnalysis
-  issue: https://github.com/elastic/elasticsearch/issues/126576
 - class: org.elasticsearch.xpack.esql.action.ForkIT
   method: testWithStatsSimple
   issue: https://github.com/elastic/elasticsearch/issues/126607

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

@@ -13,10 +13,18 @@
 import com.sun.net.httpserver.HttpServer;
 
 import org.elasticsearch.ExceptionsHelper;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
+import org.elasticsearch.common.util.concurrent.ThreadContext;
+import org.elasticsearch.logging.LogManager;
+import org.elasticsearch.logging.Logger;
+import org.elasticsearch.threadpool.ThreadPool;
 import org.junit.rules.ExternalResource;
 
 import java.io.IOException;
 import java.util.Objects;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
 import java.util.function.BiPredicate;
 
 import static fixture.aws.AwsCredentialsUtils.ANY_REGION;
@@ -26,7 +34,10 @@
 
 public class S3HttpFixture extends ExternalResource {
 
+    private static final Logger logger = LogManager.getLogger(S3HttpFixture.class);
+
     private HttpServer server;
+    private ExecutorService executorService;
 
     private final boolean enabled;
     private final String bucket;
@@ -71,16 +82,30 @@ public void stop(int delay) {
 
     protected void before() throws Throwable {
         if (enabled) {
+            this.executorService = EsExecutors.newScaling(
+                "s3-http-fixture",
+                1,
+                100,
+                30,
+                TimeUnit.SECONDS,
+                true,
+                EsExecutors.daemonThreadFactory("s3-http-fixture"),
+                new ThreadContext(Settings.EMPTY)
+            );
+
             this.server = HttpServer.create(getLocalFixtureAddress(), 0);
             this.server.createContext("/", Objects.requireNonNull(createHandler()));
+            this.server.setExecutor(executorService);
             server.start();
+            logger.info("running S3HttpFixture at " + getAddress());
         }
     }
 
     @Override
     protected void after() {
         if (enabled) {
             stop(0);
+            ThreadPool.terminate(executorService, 10, TimeUnit.SECONDS);
         }
     }
 }

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

@@ -52,6 +52,7 @@
 import static org.elasticsearch.test.fixture.HttpHeaderParser.parseRangeHeader;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
 import static org.w3c.dom.Node.ELEMENT_NODE;
 
 /**
@@ -121,9 +122,11 @@ public void handle(final HttpExchange exchange) throws IOException {
                 uploadsList.append("<MaxUploads>10000</MaxUploads>");
                 uploadsList.append("<IsTruncated>false</IsTruncated>");
 
-                for (final var multipartUpload : uploads.values()) {
-                    if (multipartUpload.getPath().startsWith(prefix)) {
-                        multipartUpload.appendXml(uploadsList);
+                synchronized (uploads) {
+                    for (final var multipartUpload : uploads.values()) {
+                        if (multipartUpload.getPath().startsWith(prefix)) {
+                            multipartUpload.appendXml(uploadsList);
+                        }
                     }
                 }
 
@@ -135,9 +138,7 @@ public void handle(final HttpExchange exchange) throws IOException {
                 exchange.getResponseBody().write(response);
 
             } else if (request.isInitiateMultipartUploadRequest()) {
-                final var upload = new MultipartUpload(UUIDs.randomBase64UUID(), request.path().substring(bucket.length() + 2));
-                uploads.put(upload.getUploadId(), upload);
-
+                final var upload = putUpload(request.path().substring(bucket.length() + 2));
                 final var uploadResult = new StringBuilder();
                 uploadResult.append("<?xml version='1.0' encoding='UTF-8'?>");
                 uploadResult.append("<InitiateMultipartUploadResult>");
@@ -152,7 +153,7 @@ public void handle(final HttpExchange exchange) throws IOException {
                 exchange.getResponseBody().write(response);
 
             } else if (request.isUploadPartRequest()) {
-                final var upload = uploads.get(request.getQueryParamOnce("uploadId"));
+                final var upload = getUpload(request.getQueryParamOnce("uploadId"));
                 if (upload == null) {
                     exchange.sendResponseHeaders(RestStatus.NOT_FOUND.getStatus(), -1);
                 } else {
@@ -187,42 +188,45 @@ public void handle(final HttpExchange exchange) throws IOException {
                 }
 
             } else if (request.isCompleteMultipartUploadRequest()) {
-                final var upload = uploads.remove(request.getQueryParamOnce("uploadId"));
-                if (upload == null) {
-                    if (Randomness.get().nextBoolean()) {
-                        exchange.sendResponseHeaders(RestStatus.NOT_FOUND.getStatus(), -1);
+                final byte[] responseBody;
+                synchronized (uploads) {
+                    final var upload = removeUpload(request.getQueryParamOnce("uploadId"));
+                    if (upload == null) {
+                        if (Randomness.get().nextBoolean()) {
+                            responseBody = null;
+                        } else {
+                            responseBody = """
+                                <?xml version="1.0" encoding="UTF-8"?>
+                                <Error>
+                                <Code>NoSuchUpload</Code>
+                                <Message>No such upload</Message>
+                                <RequestId>test-request-id</RequestId>
+                                <HostId>test-host-id</HostId>
+                                </Error>""".getBytes(StandardCharsets.UTF_8);
+                        }
                     } else {
-                        byte[] response = ("""
-                            <?xml version="1.0" encoding="UTF-8"?>
-                            <Error>
-                            <Code>NoSuchUpload</Code>
-                            <Message>No such upload</Message>
-                            <RequestId>test-request-id</RequestId>
-                            <HostId>test-host-id</HostId>
-                            </Error>""").getBytes(StandardCharsets.UTF_8);
-                        exchange.getResponseHeaders().add("Content-Type", "application/xml");
-                        exchange.sendResponseHeaders(RestStatus.OK.getStatus(), response.length);
-                        exchange.getResponseBody().write(response);
+                        final var blobContents = upload.complete(extractPartEtags(Streams.readFully(exchange.getRequestBody())));
+                        blobs.put(request.path(), blobContents);
+                        responseBody = ("<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n"
+                            + "<CompleteMultipartUploadResult>\n"
+                            + "<Bucket>"
+                            + bucket
+                            + "</Bucket>\n"
+                            + "<Key>"
+                            + request.path()
+                            + "</Key>\n"
+                            + "</CompleteMultipartUploadResult>").getBytes(StandardCharsets.UTF_8);
                     }
+                }
+                if (responseBody == null) {
+                    exchange.sendResponseHeaders(RestStatus.NOT_FOUND.getStatus(), -1);
                 } else {
-                    final var blobContents = upload.complete(extractPartEtags(Streams.readFully(exchange.getRequestBody())));
-                    blobs.put(request.path(), blobContents);
-
-                    byte[] response = ("<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n"
-                        + "<CompleteMultipartUploadResult>\n"
-                        + "<Bucket>"
-                        + bucket
-                        + "</Bucket>\n"
-                        + "<Key>"
-                        + request.path()
-                        + "</Key>\n"
-                        + "</CompleteMultipartUploadResult>").getBytes(StandardCharsets.UTF_8);
                     exchange.getResponseHeaders().add("Content-Type", "application/xml");
-                    exchange.sendResponseHeaders(RestStatus.OK.getStatus(), response.length);
-                    exchange.getResponseBody().write(response);
+                    exchange.sendResponseHeaders(RestStatus.OK.getStatus(), responseBody.length);
+                    exchange.getResponseBody().write(responseBody);
                 }
             } else if (request.isAbortMultipartUploadRequest()) {
-                final var upload = uploads.remove(request.getQueryParamOnce("uploadId"));
+                final var upload = removeUpload(request.getQueryParamOnce("uploadId"));
                 exchange.sendResponseHeaders((upload == null ? RestStatus.NOT_FOUND : RestStatus.NO_CONTENT).getStatus(), -1);
 
             } else if (request.isPutObjectRequest()) {
@@ -521,8 +525,24 @@ private static HttpHeaderParser.Range parsePartRange(final HttpExchange exchange
         return parseRangeHeader(sourceRangeHeaders.getFirst());
     }
 
+    MultipartUpload putUpload(String path) {
+        final var upload = new MultipartUpload(UUIDs.randomBase64UUID(), path);
+        synchronized (uploads) {
+            assertNull("upload " + upload.getUploadId() + " should not exist", uploads.put(upload.getUploadId(), upload));
+            return upload;
+        }
+    }
+
     MultipartUpload getUpload(String uploadId) {
-        return uploads.get(uploadId);
+        synchronized (uploads) {
+            return uploads.get(uploadId);
+        }
+    }
+
+    MultipartUpload removeUpload(String uploadId) {
+        synchronized (uploads) {
+            return uploads.remove(uploadId);
+        }
     }
 
     public S3Request parseRequest(HttpExchange exchange) {