Fix Concurrent Snapshot Ending And Stabilize Snapshot Finalization (#38368)

* The problem in #38226 is that in some corner cases multiple calls to `endSnapshot` were made concurrently, leading to non-deterministic behavior (`beginSnapshot` was triggering a repository finalization while one that was triggered by a `deleteSnapshot` was already in progress)
   * Fixed by:
      * Making all `endSnapshot` calls originate from the cluster state being in a "completed" state (apart from on short-circuit on initializing an empty snapshot). This forced putting the failure string into `SnapshotsInProgress.Entry`.
      * Adding deduplication logic to `endSnapshot`
* Also:
  * Streamlined the init behavior to work the same way (keep state on the `SnapshotsService` to decide which snapshot entries are stale)
* closes #38226

Author: original-brownbear

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

@@ -24,6 +24,7 @@
 import com.carrotsearch.hppc.cursors.ObjectObjectCursor;
 import org.elasticsearch.Version;
 import org.elasticsearch.cluster.ClusterState.Custom;
+import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.collect.ImmutableOpenMap;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
@@ -87,9 +88,11 @@ public static class Entry {
         private final ImmutableOpenMap<String, List<ShardId>> waitingIndices;
         private final long startTime;
         private final long repositoryStateId;
+        @Nullable private final String failure;
 
         public Entry(Snapshot snapshot, boolean includeGlobalState, boolean partial, State state, List<IndexId> indices,
-                     long startTime, long repositoryStateId, ImmutableOpenMap<ShardId, ShardSnapshotStatus> shards) {
+                     long startTime, long repositoryStateId, ImmutableOpenMap<ShardId, ShardSnapshotStatus> shards,
+                     String failure) {
             this.state = state;
             this.snapshot = snapshot;
             this.includeGlobalState = includeGlobalState;
@@ -104,15 +107,26 @@ public Entry(Snapshot snapshot, boolean includeGlobalState, boolean partial, Sta
                 this.waitingIndices = findWaitingIndices(shards);
             }
             this.repositoryStateId = repositoryStateId;
+            this.failure = failure;
+        }
+
+        public Entry(Snapshot snapshot, boolean includeGlobalState, boolean partial, State state, List<IndexId> indices,
+            long startTime, long repositoryStateId, ImmutableOpenMap<ShardId, ShardSnapshotStatus> shards) {
+            this(snapshot, includeGlobalState, partial, state, indices, startTime, repositoryStateId, shards, null);
         }
 
         public Entry(Entry entry, State state, ImmutableOpenMap<ShardId, ShardSnapshotStatus> shards) {
             this(entry.snapshot, entry.includeGlobalState, entry.partial, state, entry.indices, entry.startTime,
-                 entry.repositoryStateId, shards);
+                entry.repositoryStateId, shards, entry.failure);
+        }
+
+        public Entry(Entry entry, State state, ImmutableOpenMap<ShardId, ShardSnapshotStatus> shards, String failure) {
+            this(entry.snapshot, entry.includeGlobalState, entry.partial, state, entry.indices, entry.startTime,
+                 entry.repositoryStateId, shards, failure);
         }
 
         public Entry(Entry entry, ImmutableOpenMap<ShardId, ShardSnapshotStatus> shards) {
-            this(entry, entry.state, shards);
+            this(entry, entry.state, shards, entry.failure);
         }
 
         public Snapshot snapshot() {
@@ -151,6 +165,10 @@ public long getRepositoryStateId() {
             return repositoryStateId;
         }
 
+        public String failure() {
+            return failure;
+        }
+
         @Override
         public boolean equals(Object o) {
             if (this == o) return true;
@@ -427,14 +445,21 @@ public SnapshotsInProgress(StreamInput in) throws IOException {
                 }
             }
             long repositoryStateId = in.readLong();
+            final String failure;
+            if (in.getVersion().onOrAfter(Version.V_7_0_0)) {
+                failure = in.readOptionalString();
+            } else {
+                failure = null;
+            }
             entries[i] = new Entry(snapshot,
                                    includeGlobalState,
                                    partial,
                                    state,
                                    Collections.unmodifiableList(indexBuilder),
                                    startTime,
                                    repositoryStateId,
-                                   builder.build());
+                                   builder.build(),
+                                   failure);
         }
         this.entries = Arrays.asList(entries);
     }
@@ -463,6 +488,9 @@ public void writeTo(StreamOutput out) throws IOException {
                 }
             }
             out.writeLong(entry.repositoryStateId);
+            if (out.getVersion().onOrAfter(Version.V_7_0_0)) {
+                out.writeOptionalString(entry.failure);
+            }
         }
     }
 

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

@@ -591,8 +591,6 @@ private class SnapshotStateExecutor implements ClusterStateTaskExecutor<UpdateIn
                             // TODO: Add PARTIAL_SUCCESS status?
                             SnapshotsInProgress.Entry updatedEntry = new SnapshotsInProgress.Entry(entry, State.SUCCESS, shards.build());
                             entries.add(updatedEntry);
-                            // Finalize snapshot in the repository
-                            snapshotsService.endSnapshot(updatedEntry);
                         }
                     } else {
                         entries.add(entry);