Add full conditional write support to S3 test fixture

In elastic#133030 we added limited support for conditional writes in
`S3HttpHandler`, allowing callers to prevent overwriting an existing
blob with an `If-None-Match: *` precondition header. This commit extends
the implementation to include support for the `If-Match: <etag>`
precondition header allowing callers to perform atomic compare-and-set
operations which overwrite existing objects.

Author: DaveCTurner

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

@@ -44,6 +44,7 @@
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
@@ -248,6 +249,9 @@ public void handle(final HttpExchange exchange) throws IOException {
                     if (isProtectOverwrite(exchange)) {
                         throw new AssertionError("If-None-Match: * header is not supported here");
                     }
+                    if (getRequiredExistingETag(exchange) != null) {
+                        throw new AssertionError("If-Match: * header is not supported here");
+                    }
 
                     var sourceBlob = blobs.get(copySource);
                     if (sourceBlob == null) {
@@ -412,10 +416,24 @@ public void handle(final HttpExchange exchange) throws IOException {
     private boolean updateBlobContents(HttpExchange exchange, String path, BytesReference newContents) {
         if (isProtectOverwrite(exchange)) {
             return blobs.putIfAbsent(path, newContents) == null;
-        } else {
-            blobs.put(path, newContents);
-            return true;
         }
+
+        final var requireExistingETag = getRequiredExistingETag(exchange);
+        if (requireExistingETag != null) {
+            final var success = new AtomicBoolean(true);
+            blobs.compute(path, (ignoredPath, existingContents) -> {
+                if (existingContents != null && requireExistingETag.equals(getEtagFromContents(existingContents))) {
+                    return newContents;
+                }
+
+                success.set(false);
+                return existingContents;
+            });
+            return success.get();
+        }
+
+        blobs.put(path, newContents);
+        return true;
     }
 
     /**
@@ -594,6 +612,9 @@ private static boolean isProtectOverwrite(final HttpExchange exchange) {
             return false;
         }
 
+        if (exchange.getRequestHeaders().get("If-Match") != null) {
+            throw new AssertionError("Handling both If-None-Match and If-Match headers is not supported");
+        }
         if (ifNoneMatch.size() != 1) {
             throw new AssertionError("multiple If-None-Match headers found: " + ifNoneMatch);
         }
@@ -605,6 +626,29 @@ private static boolean isProtectOverwrite(final HttpExchange exchange) {
         throw new AssertionError("invalid If-None-Match header: " + ifNoneMatch);
     }
 
+    @Nullable // if no If-Match header found
+    private static String getRequiredExistingETag(final HttpExchange exchange) {
+        final var ifMatch = exchange.getRequestHeaders().get("If-Match");
+
+        if (ifMatch == null) {
+            return null;
+        }
+
+        if (exchange.getRequestHeaders().get("If-None-Match") != null) {
+            throw new AssertionError("Handling both If-None-Match and If-Match headers is not supported");
+        }
+
+        final var iterator = ifMatch.iterator();
+        if (iterator.hasNext()) {
+            final var result = iterator.next();
+            if (iterator.hasNext() == false) {
+                return result;
+            }
+        }
+
+        throw new AssertionError("multiple If-Match headers found: " + ifMatch);
+    }
+
     MultipartUpload putUpload(String path) {
         final var upload = new MultipartUpload(UUIDs.randomBase64UUID(), path);
         synchronized (uploads) {

test/fixtures/s3-fixture/src/test/java/fixture/s3/S3HttpHandlerTests.java

@@ -413,13 +413,29 @@ public void testExtractPartEtags() {
     }
 
     public void testPreventObjectOverwrite() throws InterruptedException {
+        ensureExactlyOneSuccess(new S3HttpHandler("bucket", "path"), null);
+    }
+
+    public void testConditionalOverwrite() throws InterruptedException {
         final var handler = new S3HttpHandler("bucket", "path");
 
+        final var originalBody = new BytesArray(randomAlphaOfLength(50).getBytes(StandardCharsets.UTF_8));
+        final var originalETag = S3HttpHandler.getEtagFromContents(originalBody);
+        assertEquals(RestStatus.OK, handleRequest(handler, "PUT", "/bucket/path/blob", originalBody).status());
+        assertEquals(
+            new TestHttpResponse(RestStatus.OK, originalBody, addETag(originalETag, TestHttpExchange.EMPTY_HEADERS)),
+            handleRequest(handler, "GET", "/bucket/path/blob")
+        );
+
+        ensureExactlyOneSuccess(handler, originalETag);
+    }
+
+    private static void ensureExactlyOneSuccess(S3HttpHandler handler, String originalETag) throws InterruptedException {
         var tasks = List.of(
-            createPutObjectTask(handler),
-            createPutObjectTask(handler),
-            createMultipartUploadTask(handler),
-            createMultipartUploadTask(handler)
+            createPutObjectTask(handler, originalETag),
+            createPutObjectTask(handler, originalETag),
+            createMultipartUploadTask(handler, originalETag),
+            createMultipartUploadTask(handler, originalETag)
         );
 
         try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
@@ -450,13 +466,14 @@ public void testPreventObjectOverwrite() throws InterruptedException {
         );
     }
 
-    private static TestWriteTask createPutObjectTask(S3HttpHandler handler) {
+    private static TestWriteTask createPutObjectTask(S3HttpHandler handler, @Nullable String originalETag) {
         return new TestWriteTask(
-            (task) -> task.status = handleRequest(handler, "PUT", "/bucket/path/blob", task.body, ifNoneMatchHeader()).status()
+            (task) -> task.status = handleRequest(handler, "PUT", "/bucket/path/blob", task.body, conditionalWriteHeader(originalETag))
+                .status()
         );
     }
 
-    private static TestWriteTask createMultipartUploadTask(S3HttpHandler handler) {
+    private static TestWriteTask createMultipartUploadTask(S3HttpHandler handler, @Nullable String originalETag) {
         final var multipartUploadTask = new TestWriteTask(
             (task) -> task.status = handleRequest(
                 handler,
@@ -470,7 +487,7 @@ private static TestWriteTask createMultipartUploadTask(S3HttpHandler handler) {
                           <PartNumber>1</PartNumber>
                        </Part>
                     </CompleteMultipartUpload>""", task.etag)),
-                ifNoneMatchHeader()
+                conditionalWriteHeader(originalETag)
             ).status()
         );
 
@@ -599,9 +616,13 @@ private static Headers contentRangeHeader(long start, long end, long length) {
         return headers;
     }
 
-    private static Headers ifNoneMatchHeader() {
+    private static Headers conditionalWriteHeader(@Nullable String originalEtag) {
         var headers = new Headers();
-        headers.put("If-None-Match", List.of("*"));
+        if (originalEtag == null) {
+            headers.put("If-None-Match", List.of("*"));
+        } else {
+            headers.put("If-Match", List.of(originalEtag));
+        }
         return headers;
     }