Abort early on finding duplicate snapshot name in internal structures (#29634)

Adds a check in BlobstoreRepository.snapshot(...) that prevents duplicate snapshot names and fails 
the snapshot before writing out the new index file. This ensures that you cannot end up in this
situation where the index file has duplicate names and cannot be read anymore .

Relates to #28906

Author: ywelsch

server/src/main/java/org/elasticsearch/repositories/blobstore/BlobStoreRepository.java

@@ -1130,6 +1130,11 @@ public void snapshot(final IndexCommit snapshotIndexCommit) {
             BlobStoreIndexShardSnapshots snapshots = tuple.v1();
             int fileListGeneration = tuple.v2();
 
+            if (snapshots.snapshots().stream().anyMatch(sf -> sf.snapshot().equals(snapshotId.getName()))) {
+                throw new IndexShardSnapshotFailedException(shardId,
+                    "Duplicate snapshot name [" + snapshotId.getName() + "] detected, aborting");
+            }
+
             final List<BlobStoreIndexShardSnapshot.FileInfo> indexCommitPointFiles = new ArrayList<>();
 
             store.incRef();

server/src/test/java/org/elasticsearch/repositories/blobstore/BlobStoreRepositoryRestoreTests.java

@@ -33,6 +33,7 @@
 import org.elasticsearch.index.shard.IndexShardState;
 import org.elasticsearch.index.shard.IndexShardTestCase;
 import org.elasticsearch.index.shard.ShardId;
+import org.elasticsearch.index.snapshots.IndexShardSnapshotFailedException;
 import org.elasticsearch.index.store.Store;
 import org.elasticsearch.index.store.StoreFileMetaData;
 import org.elasticsearch.repositories.IndexId;
@@ -48,6 +49,7 @@
 import java.util.List;
 
 import static org.elasticsearch.cluster.routing.RecoverySource.StoreRecoverySource.EXISTING_STORE_INSTANCE;
+import static org.hamcrest.Matchers.containsString;
 
 /**
  * This class tests the behavior of {@link BlobStoreRepository} when it
@@ -126,6 +128,43 @@ public void testRestoreSnapshotWithExistingFiles() throws IOException {
         }
     }
 
+    public void testSnapshotWithConflictingName() throws IOException {
+        final IndexId indexId = new IndexId(randomAlphaOfLength(10), UUIDs.randomBase64UUID());
+        final ShardId shardId = new ShardId(indexId.getName(), indexId.getId(), 0);
+
+        IndexShard shard = newShard(shardId, true);
+        try {
+            // index documents in the shards
+            final int numDocs = scaledRandomIntBetween(1, 500);
+            recoverShardFromStore(shard);
+            for (int i = 0; i < numDocs; i++) {
+                indexDoc(shard, "doc", Integer.toString(i));
+                if (rarely()) {
+                    flushShard(shard, false);
+                }
+            }
+            assertDocCount(shard, numDocs);
+
+            // snapshot the shard
+            final Repository repository = createRepository();
+            final Snapshot snapshot = new Snapshot(repository.getMetadata().name(), new SnapshotId(randomAlphaOfLength(10), "_uuid"));
+            snapshotShard(shard, snapshot, repository);
+            final Snapshot snapshotWithSameName = new Snapshot(repository.getMetadata().name(), new SnapshotId(
+                snapshot.getSnapshotId().getName(), "_uuid2"));
+            IndexShardSnapshotFailedException isfe = expectThrows(IndexShardSnapshotFailedException.class,
+                () -> snapshotShard(shard, snapshotWithSameName, repository));
+            assertThat(isfe.getMessage(), containsString("Duplicate snapshot name"));
+        } finally {
+            if (shard != null && shard.state() != IndexShardState.CLOSED) {
+                try {
+                    shard.close("test", false);
+                } finally {
+                    IOUtils.close(shard.store());
+                }
+            }
+        }
+    }
+
     /** Create a {@link Repository} with a random name **/
     private Repository createRepository() throws IOException {
         Settings settings = Settings.builder().put("location", randomAlphaOfLength(10)).build();