Fix Race in Concurrent Snapshot Delete and Create (#37612)

* The repo id was determined wrong when the delete picked up on an in progress snapshot
  * NOTE: This solution is still a best-effort fix and there's a slight chance of running into concurrency issues here
when multiple create and delete requests for the same snapshot name are happening concurrently, but these require a sequence
of multiple cluster state updates between the changed method reading the genId and submitting its cluster state update task
* Added test reproduced the issue reliably in about 50% of runs
* Closes #37581

Author: original-brownbear

server/src/main/java/org/elasticsearch/snapshots/SnapshotsService.java

@@ -1118,14 +1118,20 @@ public void deleteSnapshot(final String repositoryName, final String snapshotNam
                                                 .filter(s -> s.getName().equals(snapshotName))
                                                 .findFirst();
         // if nothing found by the same name, then look in the cluster state for current in progress snapshots
+        long repoGenId = repositoryData.getGenId();
         if (matchedEntry.isPresent() == false) {
-            matchedEntry = currentSnapshots(repositoryName, Collections.emptyList()).stream()
-                               .map(e -> e.snapshot().getSnapshotId()).filter(s -> s.getName().equals(snapshotName)).findFirst();
+            Optional<SnapshotsInProgress.Entry> matchedInProgress = currentSnapshots(repositoryName, Collections.emptyList()).stream()
+                               .filter(s -> s.snapshot().getSnapshotId().getName().equals(snapshotName)).findFirst();
+            if (matchedInProgress.isPresent()) {
+                matchedEntry = matchedInProgress.map(s -> s.snapshot().getSnapshotId());
+                // Derive repository generation if a snapshot is in progress because it will increment the generation when it finishes
+                repoGenId = matchedInProgress.get().getRepositoryStateId() + 1L;
+            }
         }
         if (matchedEntry.isPresent() == false) {
             throw new SnapshotMissingException(repositoryName, snapshotName);
         }
-        deleteSnapshot(new Snapshot(repositoryName, matchedEntry.get()), listener, repositoryData.getGenId(), immediatePriority);
+        deleteSnapshot(new Snapshot(repositoryName, matchedEntry.get()), listener, repoGenId, immediatePriority);
     }
 
     /**

server/src/test/java/org/elasticsearch/snapshots/SnapshotsServiceTests.java

@@ -29,6 +29,9 @@
 import org.elasticsearch.action.admin.cluster.repositories.put.TransportPutRepositoryAction;
 import org.elasticsearch.action.admin.cluster.snapshots.create.CreateSnapshotAction;
 import org.elasticsearch.action.admin.cluster.snapshots.create.TransportCreateSnapshotAction;
+import org.elasticsearch.action.admin.cluster.snapshots.delete.DeleteSnapshotAction;
+import org.elasticsearch.action.admin.cluster.snapshots.delete.DeleteSnapshotRequest;
+import org.elasticsearch.action.admin.cluster.snapshots.delete.TransportDeleteSnapshotAction;
 import org.elasticsearch.action.admin.indices.create.CreateIndexAction;
 import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
 import org.elasticsearch.action.admin.indices.create.TransportCreateIndexAction;
@@ -196,6 +199,53 @@ public void testSuccessfulSnapshot() {
         assertEquals(0, snapshotInfo.failedShards());
     }
 
+    public void testConcurrentSnapshotCreateAndDelete() {
+        setupTestCluster(randomFrom(1, 3, 5), randomIntBetween(2, 10));
+
+        String repoName = "repo";
+        String snapshotName = "snapshot";
+        final String index = "test";
+
+        final int shards = randomIntBetween(1, 10);
+
+        TestClusterNode masterNode =
+            testClusterNodes.currentMaster(testClusterNodes.nodes.values().iterator().next().clusterService.state());
+        final AtomicBoolean createdSnapshot = new AtomicBoolean();
+        masterNode.client.admin().cluster().preparePutRepository(repoName)
+            .setType(FsRepository.TYPE).setSettings(Settings.builder().put("location", randomAlphaOfLength(10)))
+            .execute(
+                assertNoFailureListener(
+                    () -> masterNode.client.admin().indices().create(
+                        new CreateIndexRequest(index).waitForActiveShards(ActiveShardCount.ALL).settings(
+                            Settings.builder()
+                                .put(IndexMetaData.INDEX_NUMBER_OF_SHARDS_SETTING.getKey(), shards)
+                                .put(IndexMetaData.INDEX_NUMBER_OF_REPLICAS_SETTING.getKey(), 0)),
+                        assertNoFailureListener(
+                            () -> masterNode.client.admin().cluster().prepareCreateSnapshot(repoName, snapshotName)
+                                .execute(assertNoFailureListener(
+                                    () -> masterNode.client.admin().cluster().deleteSnapshot(
+                                        new DeleteSnapshotRequest(repoName, snapshotName),
+                                        assertNoFailureListener(() -> masterNode.client.admin().cluster()
+                                            .prepareCreateSnapshot(repoName, snapshotName).execute(
+                                                assertNoFailureListener(() -> createdSnapshot.set(true))
+                                            )))))))));
+
+        deterministicTaskQueue.runAllRunnableTasks();
+
+        assertTrue(createdSnapshot.get());
+        SnapshotsInProgress finalSnapshotsInProgress = masterNode.clusterService.state().custom(SnapshotsInProgress.TYPE);
+        assertFalse(finalSnapshotsInProgress.entries().stream().anyMatch(entry -> entry.state().completed() == false));
+        final Repository repository = masterNode.repositoriesService.repository(repoName);
+        Collection<SnapshotId> snapshotIds = repository.getRepositoryData().getSnapshotIds();
+        assertThat(snapshotIds, hasSize(1));
+
+        final SnapshotInfo snapshotInfo = repository.getSnapshotInfo(snapshotIds.iterator().next());
+        assertEquals(SnapshotState.SUCCESS, snapshotInfo.state());
+        assertThat(snapshotInfo.indices(), containsInAnyOrder(index));
+        assertEquals(shards, snapshotInfo.successfulShards());
+        assertEquals(0, snapshotInfo.failedShards());
+    }
+
     private void startCluster() {
         final ClusterState initialClusterState =
             new ClusterState.Builder(ClusterName.DEFAULT).nodes(testClusterNodes.randomDiscoveryNodes()).build();
@@ -519,6 +569,11 @@ allocationService, new AliasValidator(), environment, indexScopedSettings,
                     transportService, clusterService, threadPool,
                     snapshotsService, actionFilters, indexNameExpressionResolver
                 ));
+            actions.put(DeleteSnapshotAction.INSTANCE,
+                new TransportDeleteSnapshotAction(
+                    transportService, clusterService, threadPool,
+                    snapshotsService, actionFilters, indexNameExpressionResolver
+                ));
             client.initialize(actions, () -> clusterService.localNode().getId(), transportService.getRemoteClusterService());
         }