Prevent duplicate snapshot names after failure

Today we prevent creating a shard snapshot whose name matches one that
already exists (elastic#29634) but it's possible to end up with duplicate
snapshot names anyway if the `index-N` file is unreadable and the
fallback which reads all the `snap-UUID.dat` files encounters two with
the same name, perhaps due to some earlier failure. This results in
writing a broken `index-N` file that contains duplicate keys, which
completely breaks the repository.

This commit fails the shard snapshot without writing the broken
`index-N` file in this situation, on the assumption that the
unreadability of the existing `index-N` file is only temporary.

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

@@ -106,6 +106,7 @@
 import java.nio.file.NoSuchFileException;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -1095,13 +1096,18 @@ protected Tuple<BlobStoreIndexShardSnapshots, Integer> buildBlobStoreIndexShardS
 
             // We couldn't load the index file - falling back to loading individual snapshots
             List<SnapshotFiles> snapshots = new ArrayList<>();
+            final HashSet<String> snapshotNames = new HashSet<>();
             for (String name : blobKeys) {
                 try {
                     BlobStoreIndexShardSnapshot snapshot = null;
                     if (name.startsWith(SNAPSHOT_PREFIX)) {
                         snapshot = indexShardSnapshotFormat.readBlob(blobContainer, name);
                     }
                     if (snapshot != null) {
+                        if (snapshotNames.add(snapshot.snapshot()) == false) {
+                            throw new IndexShardSnapshotFailedException(shardId,
+                                    "index-N file was not readable, and multiple snapshots with name [" + snapshot.snapshot() + "] found");
+                        }
                         snapshots.add(new SnapshotFiles(snapshot.snapshot(), snapshot.indexFiles()));
                     }
                 } catch (IOException e) {

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

@@ -26,6 +26,8 @@
 import org.elasticsearch.cluster.routing.ShardRouting;
 import org.elasticsearch.cluster.routing.ShardRoutingHelper;
 import org.elasticsearch.common.UUIDs;
+import org.elasticsearch.common.blobstore.BlobContainer;
+import org.elasticsearch.common.blobstore.BlobPath;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.core.internal.io.IOUtils;
 import org.elasticsearch.env.Environment;
@@ -46,11 +48,13 @@
 import org.elasticsearch.snapshots.SnapshotId;
 
 import java.io.IOException;
+import java.io.InputStream;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.Arrays;
 import java.util.List;
 
+import static org.hamcrest.Matchers.allOf;
 import static org.hamcrest.Matchers.containsString;
 
 /**
@@ -176,6 +180,62 @@ public void testSnapshotWithConflictingName() throws IOException {
         }
     }
 
+    public void testDuplicateSnapshotNameInIndexShardSnapshots() 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 {
+            recoverShardFromStore(shard);
+
+            // snapshot the shard
+            final Repository repository = createRepository();
+            final String snapshotName = randomAlphaOfLength(10);
+            final Snapshot snapshot = new Snapshot(
+                    repository.getMetadata().name(),
+                    new SnapshotId(snapshotName, UUIDs.randomBase64UUID()));
+            snapshotShard(shard, snapshot, repository);
+
+            final BlobStoreRepository blobStoreRepository = (BlobStoreRepository) repository;
+            final BlobPath shardPath = blobStoreRepository.basePath().add("indices").add(indexId.getId()).add("0");
+            final BlobContainer blobContainer = blobStoreRepository.getBlobStore().blobContainer(shardPath);
+            final String snapDatFilename = "snap-" + snapshot.getSnapshotId().getUUID() + ".dat";
+
+            // Duplicate the snap-UUID.dat file so there are two with the same snapshot name
+            try (InputStream inputStream = blobContainer.readBlob(snapDatFilename)) {
+                blobContainer.writeBlob(
+                        "snap-" + UUIDs.randomBase64UUID() + ".dat",
+                        inputStream,
+                        blobContainer.listBlobsByPrefix(snapDatFilename).get(snapDatFilename).length(),
+                        true);
+            }
+
+            // Make the index-N file unreadable
+            blobContainer.deleteBlob("index-0");
+
+            final Snapshot newSnapshot = new Snapshot(
+                    repository.getMetadata().name(),
+                    new SnapshotId(randomAlphaOfLength(10), UUIDs.randomBase64UUID())
+            );
+            assertThat(
+                expectThrows(
+                    IndexShardSnapshotFailedException.class,
+                    () -> snapshotShard(shard, newSnapshot, repository)).getMessage(),
+                allOf(
+                    containsString("index-N file was not readable"),
+                    containsString("multiple snapshots"),
+                    containsString("[" + snapshotName + "]")));
+        } 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() {
         Settings settings = Settings.builder().put("location", randomAlphaOfLength(10)).build();