Add test for contents changed on resume

nicktindall · nicktindall · commit e94b246756c5 · 2025-05-02T15:58:18.000+10:00
diff --git a/modules/repository-gcs/src/main/java/org/elasticsearch/repositories/gcs/GoogleCloudStorageRetryingInputStream.java b/modules/repository-gcs/src/main/java/org/elasticsearch/repositories/gcs/GoogleCloudStorageRetryingInputStream.java
@@ -124,7 +124,7 @@ private InputStream openStream() throws IOException {
                                 + blobId.getName()
                                 + "] generation ["
                                 + lastGeneration
-                                + "] unavailable on resume: "
+                                + "] unavailable on resume (contents changed, or object deleted): "
                                 + storageException.getMessage()
                         )
                     );
@@ -150,12 +150,12 @@ private InputStream openStream() throws IOException {
     }
 
     private Long parseGenerationHeader(HttpResponse response) {
-        final Object generationHeader = response.getHeaders().get("x-goog-generation");
-        if (generationHeader instanceof String generationHeaderString) {
+        final String generationHeader = response.getHeaders().getFirstHeaderStringValue("x-goog-generation");
+        if (generationHeader != null) {
             try {
-                return Long.parseLong(generationHeaderString);
+                return Long.parseLong(generationHeader);
             } catch (NumberFormatException e) {
-                assert false : "Unexpected value for x-goog-generation header: " + generationHeaderString;
+                assert false : "Unexpected value for x-goog-generation header: " + generationHeader;
                 return null;
             }
         }
@@ -248,7 +248,6 @@ private void ensureOpen() {
         }
     }
 
-    // TODO: check that object did not change when stream is reopened (e.g. based on etag)
     private void reopenStreamOrFail(StorageException e) throws IOException {
         if (attempt >= maxAttempts) {
             throw addSuppressedExceptions(e);
diff --git a/modules/repository-gcs/src/test/java/org/elasticsearch/repositories/gcs/GoogleCloudStorageBlobContainerRetriesTests.java b/modules/repository-gcs/src/test/java/org/elasticsearch/repositories/gcs/GoogleCloudStorageBlobContainerRetriesTests.java
@@ -45,6 +45,7 @@
 import org.elasticsearch.core.SuppressForbidden;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.http.ResponseInjectingHttpHandler;
+import org.elasticsearch.mocksocket.MockHttpServer;
 import org.elasticsearch.repositories.blobstore.AbstractBlobContainerRetriesTestCase;
 import org.elasticsearch.repositories.blobstore.ESMockAPIBasedRepositoryIntegTestCase;
 import org.elasticsearch.rest.RestStatus;
@@ -56,6 +57,7 @@
 import java.io.InputStream;
 import java.net.InetSocketAddress;
 import java.net.SocketTimeoutException;
+import java.nio.file.NoSuchFileException;
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.Iterator;
@@ -71,6 +73,7 @@
 
 import static fixture.gcs.TestUtils.createServiceAccount;
 import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.elasticsearch.common.io.Streams.readFully;
 import static org.elasticsearch.repositories.blobstore.BlobStoreTestUtil.randomPurpose;
 import static org.elasticsearch.repositories.blobstore.ESBlobStoreRepositoryIntegTestCase.randomBytes;
 import static org.elasticsearch.repositories.gcs.GoogleCloudStorageBlobStore.MAX_DELETES_PER_BATCH;
@@ -86,6 +89,7 @@
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.lessThanOrEqualTo;
 import static org.hamcrest.Matchers.notNullValue;
+import static org.hamcrest.Matchers.startsWith;
 
 @SuppressForbidden(reason = "use a http server")
 public class GoogleCloudStorageBlobContainerRetriesTests extends AbstractBlobContainerRetriesTestCase {
@@ -570,6 +574,49 @@ public void testCompareAndExchangeWhenThrottled() throws IOException {
         container.delete(randomPurpose());
     }
 
+    public void testContentsChangeWhileStreaming() throws IOException {
+        GoogleCloudStorageHttpHandler handler = new GoogleCloudStorageHttpHandler("bucket");
+        httpServer.createContext("/", handler);
+        final int enoughBytesToTriggerChunkedDownload = Math.toIntExact(ByteSizeValue.ofMb(30).getBytes());
+
+        final BlobContainer container = createBlobContainer(1, null, null, null, null, null, null);
+
+        final String key = randomIdentifier();
+        byte[] initialValue = randomByteArrayOfLength(enoughBytesToTriggerChunkedDownload);
+        container.writeBlob(randomPurpose(), key, new BytesArray(initialValue), true);
+
+        BytesReference reference = readFully(container.readBlob(randomPurpose(), key));
+        assertEquals(new BytesArray(initialValue), reference);
+
+        try (InputStream inputStream = container.readBlob(randomPurpose(), key)) {
+            // Trigger the first chunk to load
+            int read = inputStream.read();
+            assert read != -1;
+
+            // Restart the server (this triggers a retry)
+            restartHttpServer();
+            httpServer.createContext("/", handler);
+
+            // Update the file
+            byte[] updatedValue = randomByteArrayOfLength(enoughBytesToTriggerChunkedDownload);
+            container.writeBlob(randomPurpose(), key, new BytesArray(updatedValue), false);
+
+            // Read the rest of the stream, it should throw because the contents changed
+            String message = assertThrows(NoSuchFileException.class, () -> readFully(inputStream)).getMessage();
+            assertThat(
+                message,
+                startsWith("Blob object [" + key + "] generation [1] unavailable on resume (contents changed, or object deleted):")
+            );
+        }
+    }
+
+    private void restartHttpServer() throws IOException {
+        InetSocketAddress currentAddress = httpServer.getAddress();
+        httpServer.stop(0);
+        httpServer = MockHttpServer.createHttp(currentAddress, 0);
+        httpServer.start();
+    }
+
     private HttpHandler safeHandler(HttpHandler handler) {
         final HttpHandler loggingHandler = ESMockAPIBasedRepositoryIntegTestCase.wrap(handler, logger);
         return exchange -> {
diff --git a/test/fixtures/gcs-fixture/src/main/java/fixture/gcs/GoogleCloudStorageHttpHandler.java b/test/fixtures/gcs-fixture/src/main/java/fixture/gcs/GoogleCloudStorageHttpHandler.java
@@ -11,8 +11,6 @@
 import com.sun.net.httpserver.HttpExchange;
 import com.sun.net.httpserver.HttpHandler;
 
-import org.apache.logging.log4j.LogManager;
-import org.apache.logging.log4j.Logger;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.io.Streams;
@@ -44,8 +42,8 @@
 @SuppressForbidden(reason = "Uses a HttpServer to emulate a Google Cloud Storage endpoint")
 public class GoogleCloudStorageHttpHandler implements HttpHandler {
 
-    private static final Logger logger = LogManager.getLogger(GoogleCloudStorageHttpHandler.class);
     private static final String IF_GENERATION_MATCH = "ifGenerationMatch";
+    private static final String GENERATION = "generation";
 
     private final AtomicInteger defaultPageLimit = new AtomicInteger(1_000);
     private final MockGcsBlobStore mockGcsBlobStore;
@@ -82,7 +80,8 @@ public void handle(final HttpExchange exchange) throws IOException {
             } else if (Regex.simpleMatch("GET /storage/v1/b/" + bucket + "/o/*", request)) {
                 final String key = exchange.getRequestURI().getPath().replace("/storage/v1/b/" + bucket + "/o/", "");
                 final Long ifGenerationMatch = parseOptionalLongParameter(exchange, IF_GENERATION_MATCH);
-                final MockGcsBlobStore.BlobVersion blob = mockGcsBlobStore.getBlob(key, ifGenerationMatch);
+                final Long generation = parseOptionalLongParameter(exchange, GENERATION);
+                final MockGcsBlobStore.BlobVersion blob = mockGcsBlobStore.getBlob(key, ifGenerationMatch, generation);
                 writeBlobVersionAsJson(exchange, blob);
             } else if (Regex.simpleMatch("GET /storage/v1/b/" + bucket + "/o*", request)) {
                 // List Objects https://cloud.google.com/storage/docs/json_api/v1/objects/list
@@ -116,7 +115,8 @@ public void handle(final HttpExchange exchange) throws IOException {
                 // Download Object https://cloud.google.com/storage/docs/request-body
                 final String path = exchange.getRequestURI().getPath().replace("/download/storage/v1/b/" + bucket + "/o/", "");
                 final Long ifGenerationMatch = parseOptionalLongParameter(exchange, IF_GENERATION_MATCH);
-                final MockGcsBlobStore.BlobVersion blob = mockGcsBlobStore.getBlob(path, ifGenerationMatch);
+                final Long generation = parseOptionalLongParameter(exchange, GENERATION);
+                final MockGcsBlobStore.BlobVersion blob = mockGcsBlobStore.getBlob(path, ifGenerationMatch, generation);
                 if (blob != null) {
                     final String rangeHeader = exchange.getRequestHeaders().getFirst("Range");
                     final BytesReference response;
@@ -144,6 +144,7 @@ public void handle(final HttpExchange exchange) throws IOException {
                     // we implement "metageneration", at that point we must incorporate both
                     // See: https://cloud.google.com/storage/docs/metadata#etags
                     exchange.getResponseHeaders().add("ETag", String.valueOf(blob.generation()));
+                    exchange.getResponseHeaders().add("X-Goog-Generation", String.valueOf(blob.generation()));
                     exchange.getResponseHeaders().add("Content-Type", "application/octet-stream");
                     exchange.sendResponseHeaders(statusCode, response.length());
                     response.writeTo(exchange.getResponseBody());
diff --git a/test/fixtures/gcs-fixture/src/main/java/fixture/gcs/MockGcsBlobStore.java b/test/fixtures/gcs-fixture/src/main/java/fixture/gcs/MockGcsBlobStore.java
@@ -88,18 +88,29 @@ public int length() {
         }
     }
 
-    BlobVersion getBlob(String path, Long ifGenerationMatch) {
+    /**
+     * Get the blob at the specified path
+     *
+     * @param path The path
+     * @param ifGenerationMatch The ifGenerationMatch parameter value (if present)
+     * @param generation The generation parameter value (if present)
+     * @return The blob if it exists
+     * @throws BlobNotFoundException if there is no blob at the path, or it's generation does not match the generation parameter
+     * @throws GcsRestException if the blob's generation does not match the ifGenerationMatch parameter
+     */
+    BlobVersion getBlob(String path, Long ifGenerationMatch, Long generation) {
         final BlobVersion blob = blobs.get(path);
         if (blob == null) {
             throw new BlobNotFoundException(path);
         } else {
-            if (ifGenerationMatch != null) {
-                if (blob.generation != ifGenerationMatch) {
-                    throw new GcsRestException(
-                        RestStatus.PRECONDITION_FAILED,
-                        "Generation mismatch, expected " + ifGenerationMatch + " but got " + blob.generation
-                    );
-                }
+            if (generation != null && generation != blob.generation) {
+                throw new BlobNotFoundException(blob.path, blob.generation);
+            }
+            if (ifGenerationMatch != null && blob.generation != ifGenerationMatch) {
+                throw new GcsRestException(
+                    RestStatus.PRECONDITION_FAILED,
+                    "Generation mismatch, expected " + ifGenerationMatch + " but got " + blob.generation
+                );
             }
             return blob;
         }
@@ -324,6 +335,10 @@ static class BlobNotFoundException extends GcsRestException {
         BlobNotFoundException(String path) {
             super(RestStatus.NOT_FOUND, "Blob not found: " + path);
         }
+
+        BlobNotFoundException(String path, long generation) {
+            super(RestStatus.NOT_FOUND, "Blob not found: " + path + ", generation " + generation);
+        }
     }
 
     static class GcsRestException extends RuntimeException {

Original file line number	Diff line number	Diff line change
`@@ -124,7 +124,7 @@ private InputStream openStream() throws IOException {`
`124`	`124`	`+ blobId.getName()`
`125`	`125`	`+ "] generation ["`
`126`	`126`	`+ lastGeneration`
`127`		`- + "] unavailable on resume: "`
	`127`	`+ + "] unavailable on resume (contents changed, or object deleted): "`
`128`	`128`	`+ storageException.getMessage()`
`129`	`129`	`)`
`130`	`130`	`);`
`@@ -150,12 +150,12 @@ private InputStream openStream() throws IOException {`
`150`	`150`	`}`
`151`	`151`
`152`	`152`	`private Long parseGenerationHeader(HttpResponse response) {`
`153`		`- final Object generationHeader = response.getHeaders().get("x-goog-generation");`
`154`		`- if (generationHeader instanceof String generationHeaderString) {`
	`153`	`+ final String generationHeader = response.getHeaders().getFirstHeaderStringValue("x-goog-generation");`
	`154`	`+ if (generationHeader != null) {`
`155`	`155`	`try {`
`156`		`- return Long.parseLong(generationHeaderString);`
	`156`	`+ return Long.parseLong(generationHeader);`
`157`	`157`	`} catch (NumberFormatException e) {`
`158`		`- assert false : "Unexpected value for x-goog-generation header: " + generationHeaderString;`
	`158`	`+ assert false : "Unexpected value for x-goog-generation header: " + generationHeader;`
`159`	`159`	`return null;`
`160`	`160`	`}`
`161`	`161`	`}`
`@@ -248,7 +248,6 @@ private void ensureOpen() {`
`248`	`248`	`}`
`249`	`249`	`}`
`250`	`250`
`251`		`- // TODO: check that object did not change when stream is reopened (e.g. based on etag)`
`252`	`251`	`private void reopenStreamOrFail(StorageException e) throws IOException {`
`253`	`252`	`if (attempt >= maxAttempts) {`
`254`	`253`	`throw addSuppressedExceptions(e);`