Specify generation on resume

Author: nicktindall

modules/repository-gcs/src/main/java/org/elasticsearch/repositories/gcs/GoogleCloudStorageRetryingInputStream.java

@@ -52,6 +52,7 @@ class GoogleCloudStorageRetryingInputStream extends InputStream {
     private List<StorageException> failures = new ArrayList<>(MAX_SUPPRESSED_EXCEPTIONS);
     private long currentOffset;
     private boolean closed;
+    private Long lastGeneration;
 
     // Used for testing only
     GoogleCloudStorageRetryingInputStream(OperationPurpose purpose, MeteredStorage client, BlobId blobId) throws IOException {
@@ -84,13 +85,22 @@ private InputStream openStream() throws IOException {
                         return SocketAccess.doPrivilegedIOException(() -> {
                             final var meteredGet = client.meteredObjectsGet(purpose, blobId.getBucket(), blobId.getName());
                             meteredGet.setReturnRawInputStream(true);
+                            if (lastGeneration != null) {
+                                meteredGet.setGeneration(lastGeneration);
+                            }
 
                             if (currentOffset > 0 || start > 0 || end < Long.MAX_VALUE - 1) {
                                 if (meteredGet.getRequestHeaders() != null) {
                                     meteredGet.getRequestHeaders().setRange("bytes=" + Math.addExact(start, currentOffset) + "-" + end);
                                 }
                             }
                             final HttpResponse resp = meteredGet.executeMedia();
+                            // Store the generation of the first response we received, so we can detect
+                            // if the file has changed if we need to resume
+                            if (lastGeneration == null) {
+                                lastGeneration = parseGenerationHeader(resp);
+                            }
+
                             final Long contentLength = resp.getHeaders().getContentLength();
                             InputStream content = resp.getContent();
                             if (contentLength != null) {
@@ -107,9 +117,22 @@ private InputStream openStream() throws IOException {
             }
         } catch (StorageException storageException) {
             if (storageException.getCode() == RestStatus.NOT_FOUND.getStatus()) {
-                throw addSuppressedExceptions(
-                    new NoSuchFileException("Blob object [" + blobId.getName() + "] not found: " + storageException.getMessage())
-                );
+                if (lastGeneration != null) {
+                    throw addSuppressedExceptions(
+                        new NoSuchFileException(
+                            "Blob object ["
+                                + blobId.getName()
+                                + "] generation ["
+                                + lastGeneration
+                                + "] unavailable on resume: "
+                                + storageException.getMessage()
+                        )
+                    );
+                } else {
+                    throw addSuppressedExceptions(
+                        new NoSuchFileException("Blob object [" + blobId.getName() + "] not found: " + storageException.getMessage())
+                    );
+                }
             }
             if (storageException.getCode() == RestStatus.REQUESTED_RANGE_NOT_SATISFIED.getStatus()) {
                 long currentPosition = Math.addExact(start, currentOffset);
@@ -126,6 +149,19 @@ private InputStream openStream() throws IOException {
         }
     }
 
+    private Long parseGenerationHeader(HttpResponse response) {
+        final Object generationHeader = response.getHeaders().get("x-goog-generation");
+        if (generationHeader instanceof String generationHeaderString) {
+            try {
+                return Long.parseLong(generationHeaderString);
+            } catch (NumberFormatException e) {
+                assert false : "Unexpected value for x-goog-generation header: " + generationHeaderString;
+                return null;
+            }
+        }
+        return null;
+    }
+
     // Google's SDK ignores the Content-Length header when no bytes are sent, see NetHttpResponse.SizeValidatingInputStream
     // We have to implement our own validation logic here
     static final class ContentLengthValidatingInputStream extends FilterInputStream {

modules/repository-gcs/src/main/java/org/elasticsearch/repositories/gcs/MeteredStorage.java

@@ -144,6 +144,10 @@ public void setReturnRawInputStream(boolean b) {
             get.setReturnRawInputStream(b);
         }
 
+        public void setGeneration(Long generation) {
+            get.setGeneration(generation);
+        }
+
         public HttpHeaders getRequestHeaders() {
             return get.getRequestHeaders();
         }