Fix deletions

* Treat queued with gen as started that blocks deletion to run
* Aborted queued with gen is failed with a separate cluster state update
  to simulate how abort for init works.

Author: ywangd

server/src/main/java/org/elasticsearch/cluster/SnapshotsInProgress.java

@@ -58,6 +58,7 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
+import java.util.function.BiConsumer;
 import java.util.stream.Stream;
 
 import static org.elasticsearch.repositories.ProjectRepo.PROJECT_REPO_SERIALIZER;
@@ -1229,7 +1230,7 @@ public Entry withClones(Map<RepositoryShardId, ShardSnapshotStatus> updatedClone
          * @return aborted snapshot entry or {@code null} if entry can be removed from the cluster state directly
          */
         @Nullable
-        public Entry abort() {
+        public Entry abort(BiConsumer<ShardId, ShardSnapshotStatus> abortedQueuedWithGenerationShardConsumer) {
             final Map<ShardId, ShardSnapshotStatus> shardsBuilder = new HashMap<>();
             boolean completed = true;
             boolean allQueued = true;
@@ -1238,12 +1239,23 @@ public Entry abort() {
                 allQueued &= status.state() == ShardState.QUEUED;
                 if (status.state().completed() == false) {
                     final String nodeId = status.nodeId();
+                    final var isQueuedWithGeneration = status.isQueuedWithGeneration();
                     status = new ShardSnapshotStatus(
                         nodeId,
-                        nodeId == null ? ShardState.FAILED : ShardState.ABORTED, // TODO: Aborted if assigned queued?
+                        // QUEUED with generation transitioned to ABORTED (incomplete) and is completed by a separate cluster state update
+                        (nodeId == null && isQueuedWithGeneration == false) ? ShardState.FAILED : ShardState.ABORTED,
                         status.generation(),
                         "aborted by snapshot deletion"
                     );
+                    if (isQueuedWithGeneration) {
+                        // Accumulate the updates needed to complete the aborted QUEUED with generation shard snapshots
+                        assert isClone() == false
+                            : "The state queued with generation should not be possible for a clone entry [" + this + "]";
+                        abortedQueuedWithGenerationShardConsumer.accept(
+                            shardEntry.getKey(),
+                            new ShardSnapshotStatus(null, ShardState.FAILED, status.generation, "aborted by snapshot deletion")
+                        );
+                    }
                 }
                 completed &= status.state().completed();
                 shardsBuilder.put(shardEntry.getKey(), status);

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

@@ -1874,7 +1874,9 @@ private static Tuple<ClusterState, List<SnapshotDeletionsInProgress.Entry>> read
             final var projectRepo = new ProjectRepo(entry.projectId(), entry.repository());
             if (repositoriesSeen.add(projectRepo)
                 && entry.state() == SnapshotDeletionsInProgress.State.WAITING
-                && snapshotsInProgress.forRepo(projectRepo).stream().noneMatch(SnapshotsService::isWritingToRepository)) {
+                && snapshotsInProgress.forRepo(projectRepo)
+                    .stream()
+                    .noneMatch(SnapshotsService::isWritingToRepositoryOrQueueWithGeneration)) {
                 changed = true;
                 final SnapshotDeletionsInProgress.Entry newEntry = entry.started();
                 readyDeletions.add(newEntry);
@@ -2293,14 +2295,26 @@ public ClusterState execute(ClusterState currentState) {
                 }
                 // Snapshot ids that will have to be physically deleted from the repository
                 final Set<SnapshotId> snapshotIdsRequiringCleanup = new HashSet<>(snapshotIds);
+
+                final List<Runnable> completeAbortedQueuedWithGenerationRunnables = new ArrayList<>();
                 final SnapshotsInProgress updatedSnapshots = snapshotsInProgress.createCopyWithUpdatedEntriesForRepo(
                     projectId,
                     repositoryName,
                     snapshotsInProgress.forRepo(projectId, repositoryName).stream().map(existing -> {
                         if (existing.state() == SnapshotsInProgress.State.STARTED
                             && snapshotIdsRequiringCleanup.contains(existing.snapshot().getSnapshotId())) {
                             // snapshot is started - mark every non completed shard as aborted
-                            final SnapshotsInProgress.Entry abortedEntry = existing.abort();
+                            final SnapshotsInProgress.Entry abortedEntry = existing.abort(
+                                ((shardId, shardSnapshotStatus) -> completeAbortedQueuedWithGenerationRunnables.add(
+                                    () -> innerUpdateSnapshotState(
+                                        existing.snapshot(),
+                                        shardId,
+                                        null,
+                                        shardSnapshotStatus,
+                                        ActionListener.noop()
+                                    )
+                                ))
+                            );
                             if (abortedEntry == null) {
                                 // No work has been done for this snapshot yet so we remove it from the cluster state directly
                                 final Snapshot existingNotYetStartedSnapshot = existing.snapshot();
@@ -2318,6 +2332,9 @@ public ClusterState execute(ClusterState currentState) {
                         return existing;
                     }).filter(Objects::nonNull).toList()
                 );
+                for (var runnable : completeAbortedQueuedWithGenerationRunnables) {
+                    runnable.run();
+                }
                 if (snapshotIdsRequiringCleanup.isEmpty()) {
                     // We only saw snapshots that could be removed from the cluster state right away, no need to update the deletions
                     return updateWithSnapshots(currentState, updatedSnapshots, null);
@@ -2350,7 +2367,9 @@ public ClusterState execute(ClusterState currentState) {
                         List.copyOf(snapshotIdsRequiringCleanup),
                         threadPool.absoluteTimeInMillis(),
                         repositoryData.getGenId(),
-                        updatedSnapshots.forRepo(projectId, repositoryName).stream().noneMatch(SnapshotsService::isWritingToRepository)
+                        updatedSnapshots.forRepo(projectId, repositoryName)
+                            .stream()
+                            .noneMatch(SnapshotsService::isWritingToRepositoryOrQueueWithGeneration)
                             && deletionsInProgress.hasExecutingDeletion(projectId, repositoryName) == false
                                 ? SnapshotDeletionsInProgress.State.STARTED
                                 : SnapshotDeletionsInProgress.State.WAITING
@@ -2432,13 +2451,13 @@ public String toString() {
      * @param entry snapshot entry
      * @return true if entry is currently writing to the repository
      */
-    private static boolean isWritingToRepository(SnapshotsInProgress.Entry entry) {
+    private static boolean isWritingToRepositoryOrQueueWithGeneration(SnapshotsInProgress.Entry entry) {
         if (entry.state().completed()) {
             // Entry is writing to the repo because it's finalizing on master
             return true;
         }
         for (ShardSnapshotStatus value : entry.shardSnapshotStatusByRepoShardId().values()) {
-            if (value.isActive()) {
+            if (value.isActiveOrQueuedWithGeneration()) {
                 // Entry is writing to the repo because it's writing to a shard on a data node or waiting to do so for a concrete shard
                 return true;
             }