Throw better exception if verifying empty repo (#131685)

Today if you attempt to verify the integrity of a brand-new repository
(no `index-${N}` blob) then it will fail because the repository
generation is `-1` which cannot be sent over the wire. But it makes no
sense to verify the integrity of such a repository anyway, so with this
commit we fail such requests up-front with a more helpful error message.

Backport of #131677 to `8.19`

docs/changelog/131677.yaml

@@ -0,0 +1,5 @@
+pr: 131677
+summary: Throw better exception if verifying empty repo
+area: Snapshot/Restore
+type: bug
+issues: []

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

@@ -14,6 +14,7 @@ import org.elasticsearch.action.support.ActiveShardCount;
 import org.elasticsearch.action.support.SubscribableListener;
 import org.elasticsearch.client.Request;
 import org.elasticsearch.client.Response;
+import org.elasticsearch.client.ResponseException;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.ByteSizeValue;
@@ -35,6 +36,7 @@ import org.elasticsearch.repositories.blobstore.BlobStoreRepository;
 import org.elasticsearch.repositories.blobstore.RepositoryFileType;
 import org.elasticsearch.repositories.blobstore.testkit.SnapshotRepositoryTestKit;
 import org.elasticsearch.repositories.fs.FsRepository;
+import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.snapshots.AbstractSnapshotIntegTestCase;
 import org.elasticsearch.snapshots.SnapshotId;
 import org.elasticsearch.snapshots.SnapshotInfo;
@@ -73,6 +75,7 @@ import static org.elasticsearch.repositories.blobstore.BlobStoreRepository.INDEX
 import static org.elasticsearch.repositories.blobstore.BlobStoreRepository.SNAPSHOT_FORMAT;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
 import static org.hamcrest.Matchers.allOf;
+import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.empty;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
@@ -713,6 +716,21 @@ public class RepositoryVerifyIntegrityIT extends AbstractSnapshotIntegTestCase {
         }, "blob in snapshot but not shard generation");
     }
 
+    public void testFreshRepository() {
+        final var repositoryName = randomIdentifier();
+        final var repositoryRootPath = randomRepoPath();
+
+        createRepository(repositoryName, FsRepository.TYPE, repositoryRootPath);
+        try {
+            final var request = new Request("POST", "/_snapshot/" + repositoryName + "/_verify_integrity");
+            final var responseException = expectThrows(ResponseException.class, () -> getRestClient().performRequest(request));
+            assertEquals(RestStatus.BAD_REQUEST.getStatus(), responseException.getResponse().getStatusLine().getStatusCode());
+            assertThat(responseException.getMessage(), containsString("repository is empty, cannot verify its integrity"));
+        } finally {
+            deleteRepository(repositoryName);
+        }
+    }
+
     private void runInconsistentShardGenerationBlobTest(
         TestContext testContext,
         UnaryOperator<BlobStoreIndexShardSnapshots> shardGenerationUpdater,

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

@@ -132,6 +132,7 @@ class TransportRepositoryVerifyIntegrityAction extends HandledTransportAction<
 
             .<RepositoryData>newForked(l -> repository.getRepositoryData(executor, l))
             .andThenApply(repositoryData -> {
+                ensureValidGenId(repositoryData.getGenId());
                 final var cancellableThreads = new CancellableThreads();
                 task.addListener(() -> cancellableThreads.cancel("task cancelled"));
                 final var verifier = new RepositoryIntegrityVerifier(
@@ -155,4 +156,17 @@ class TransportRepositoryVerifyIntegrityAction extends HandledTransportAction<
             .<RepositoryVerifyIntegrityResponse>andThen((l, repositoryIntegrityVerifier) -> repositoryIntegrityVerifier.start(l))
             .addListener(listener);
     }
+
+    static void ensureValidGenId(long repositoryGenId) {
+        if (repositoryGenId == RepositoryData.EMPTY_REPO_GEN) {
+            throw new IllegalArgumentException("repository is empty, cannot verify its integrity");
+        }
+        if (repositoryGenId < 0) {
+            final var exception = new IllegalStateException(
+                "repository is in an unexpected state [" + repositoryGenId + "], cannot verify its integrity"
+            );
+            assert false : exception; // cannot be unknown, and if corrupt we throw a corruptedStateException from getRepositoryData
+            throw exception;
+        }
+    }
 }

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

@@ -0,0 +1,35 @@
+/*
+ * 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.integrity;
+
+import org.elasticsearch.repositories.RepositoryData;
+import org.elasticsearch.test.ESTestCase;
+
+import static org.hamcrest.Matchers.equalTo;
+
+public class TransportRepositoryVerifyIntegrityActionTests extends ESTestCase {
+    public void testEnsureValidGenId() {
+        TransportRepositoryVerifyIntegrityAction.ensureValidGenId(0);
+        TransportRepositoryVerifyIntegrityAction.ensureValidGenId(randomNonNegativeLong());
+        assertThat(
+            expectThrows(
+                IllegalArgumentException.class,
+                () -> TransportRepositoryVerifyIntegrityAction.ensureValidGenId(RepositoryData.EMPTY_REPO_GEN)
+            ).getMessage(),
+            equalTo("repository is empty, cannot verify its integrity")
+        );
+        assertThat(expectThrows(IllegalStateException.class, () -> {
+            try {
+                TransportRepositoryVerifyIntegrityAction.ensureValidGenId(RepositoryData.CORRUPTED_REPO_GEN);
+            } catch (AssertionError e) {
+                // if assertions disabled, we throw the cause directly
+                throw e.getCause();
+            }
+        }).getMessage(), equalTo("repository is in an unexpected state [-3], cannot verify its integrity"));
+    }
+}