Two Step Snapshot Delete

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

@@ -29,6 +29,8 @@
 import org.elasticsearch.repositories.RepositoryData;
 import org.elasticsearch.repositories.RepositoryOperation;
 import org.elasticsearch.snapshots.Snapshot;
+import org.elasticsearch.snapshots.SnapshotId;
+import org.elasticsearch.snapshots.SnapshotsService;
 
 import java.io.IOException;
 import java.util.ArrayList;
@@ -140,8 +142,8 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
         for (Entry entry : entries) {
             builder.startObject();
             {
-                builder.field("repository", entry.snapshot.getRepository());
-                builder.field("snapshot", entry.snapshot.getSnapshotId().getName());
+                builder.field("repository", entry.repository());
+                builder.field("snapshot", entry.getSnapshotName());
                 builder.humanReadableField("start_time_millis", "start_time", new TimeValue(entry.startTime));
                 builder.field("repository_state_id", entry.repositoryStateId);
             }
@@ -155,7 +157,7 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
     public String toString() {
         StringBuilder builder = new StringBuilder("SnapshotDeletionsInProgress[");
         for (int i = 0; i < entries.size(); i++) {
-            builder.append(entries.get(i).getSnapshot().getSnapshotId().getName());
+            builder.append(entries.get(i).getSnapshotName());
             if (i + 1 < entries.size()) {
                 builder.append(",");
             }
@@ -167,29 +169,77 @@ public String toString() {
      * A class representing a snapshot deletion request entry in the cluster state.
      */
     public static final class Entry implements Writeable, RepositoryOperation {
-        private final Snapshot snapshot;
+        private final List<SnapshotId> snapshotIds;
+        private final String repo;
+        private final String snapshotName;
         private final long startTime;
         private final long repositoryStateId;
 
+        public Entry(String repo, String snapshotName, long startTime) {
+            this(Collections.emptyList(), repo, snapshotName, startTime, RepositoryData.UNKNOWN_REPO_GEN);
+        }
+
         public Entry(Snapshot snapshot, long startTime, long repositoryStateId) {
-            this.snapshot = snapshot;
+            this(Collections.singletonList(snapshot.getSnapshotId()), snapshot.getRepository(), snapshot.getSnapshotId().getName(),
+                startTime, repositoryStateId);
+        }
+
+        public Entry(Entry entry, SnapshotId snapshotId, long repositoryStateId) {
+            this(Collections.singletonList(snapshotId), entry.repo, entry.snapshotName, entry.startTime, repositoryStateId);
+        }
+
+        private Entry(List<SnapshotId> snapshotIds, String repo, String snapshotName, long startTime, long repositoryStateId) {
+            this.snapshotIds = snapshotIds;
+            this.repo = repo;
+            this.snapshotName = snapshotName;
             this.startTime = startTime;
             this.repositoryStateId = repositoryStateId;
-            assert repositoryStateId > RepositoryData.EMPTY_REPO_GEN :
-                "Can't delete based on an empty or unknown repository generation but saw [" + repositoryStateId + "]";
+            assert assertConsistent();
         }
 
         public Entry(StreamInput in) throws IOException {
-            this.snapshot = new Snapshot(in);
+            if (in.getVersion().onOrAfter(SnapshotsService.TWO_STEP_DELETE_VERSION)) {
+                this.repo = in.readString();
+                this.snapshotName = in.readString();
+                this. snapshotIds = in.readList(SnapshotId::new);
+            } else {
+                final Snapshot snapshot = new Snapshot(in);
+                final SnapshotId snapshotId = snapshot.getSnapshotId();
+                this.snapshotIds = Collections.singletonList(snapshotId);
+                this.repo = snapshot.getRepository();
+                this.snapshotName = snapshotId.getName();
+            }
             this.startTime = in.readVLong();
             this.repositoryStateId = in.readLong();
+            assert assertConsistent();
         }
 
-        /**
-         * The snapshot to delete.
-         */
-        public Snapshot getSnapshot() {
-            return snapshot;
+        private boolean assertConsistent() {
+            if (repositoryStateId == RepositoryData.UNKNOWN_REPO_GEN) {
+                assert snapshotIds.isEmpty() : "Saw concrete snapshot ids for unknown repository generation in [" + this + "]";
+            } else {
+                assert snapshotIds.size() == 1
+                    : "Should have resolved one snapshot id for concrete repository generation but saw [" + this + "]";
+            }
+            return true;
+        }
+
+        public String getSnapshotName() {
+            return snapshotName;
+        }
+
+        public boolean matches(Snapshot snapshot) {
+            if (repo.equals(snapshot.getRepository()) == false) {
+                return false;
+            }
+            if (repositoryStateId == RepositoryData.UNKNOWN_REPO_GEN) {
+                return snapshotName.equals(snapshot.getSnapshotId().getName());
+            }
+            return snapshotIds.contains(snapshot.getSnapshotId());
+        }
+
+        public List<SnapshotId> getSnapshotIds() {
+            return snapshotIds;
         }
 
         /**
@@ -208,31 +258,46 @@ public boolean equals(Object o) {
                 return false;
             }
             Entry that = (Entry) o;
-            return snapshot.equals(that.snapshot)
+            return snapshotName.equals(that.snapshotName)
+                       && snapshotIds.equals(that.snapshotIds)
                        && startTime == that.startTime
                        && repositoryStateId == that.repositoryStateId;
         }
 
         @Override
         public int hashCode() {
-            return Objects.hash(snapshot, startTime, repositoryStateId);
+            return Objects.hash(snapshotName, snapshotIds, startTime, repositoryStateId);
         }
 
         @Override
         public void writeTo(StreamOutput out) throws IOException {
-            snapshot.writeTo(out);
+            if (out.getVersion().onOrAfter(SnapshotsService.TWO_STEP_DELETE_VERSION)) {
+                out.writeString(repo);
+                out.writeString(snapshotName);
+                out.writeList(snapshotIds);
+            } else {
+                assert snapshotIds.size() == 1 : "Saw multiple snapshot ids [" + snapshotIds +
+                    "] in delete but cluster contains node version [" + out.getVersion() + "]";
+                new Snapshot(repo, snapshotIds.get(0)).writeTo(out);
+            }
             out.writeVLong(startTime);
             out.writeLong(repositoryStateId);
         }
 
         @Override
         public String repository() {
-            return snapshot.getRepository();
+            return repo;
         }
 
         @Override
         public long repositoryStateId() {
             return repositoryStateId;
         }
+
+        @Override
+        public String toString() {
+            return "Entry[repository=" + repo + ", repositoryStateId=" + repositoryStateId + ", snapshotIds="
+                + snapshotIds + ", pattern=" + snapshotName + "]";
+        }
     }
 }

server/src/main/java/org/elasticsearch/repositories/RepositoriesService.java

@@ -487,7 +487,7 @@ private static boolean isRepositoryInUse(ClusterState clusterState, String repos
         SnapshotDeletionsInProgress deletionsInProgress = clusterState.custom(SnapshotDeletionsInProgress.TYPE);
         if (deletionsInProgress != null) {
             for (SnapshotDeletionsInProgress.Entry entry : deletionsInProgress.getEntries()) {
-                if (entry.getSnapshot().getRepository().equals(repository)) {
+                if (entry.repository().equals(repository)) {
                     return true;
                 }
             }

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

@@ -355,13 +355,14 @@ public void updateState(ClusterState state) {
             // Don't use generation from the delete task if we already found a generation for an in progress snapshot.
             // In this case, the generation points at the generation the repo will be in after the snapshot finishes so it may not yet
             // exist
-            if (bestGenerationFromCS == RepositoryData.EMPTY_REPO_GEN && deletionsInProgress != null) {
+            if (bestGenerationFromCS == RepositoryData.UNKNOWN_REPO_GEN && deletionsInProgress != null) {
                 bestGenerationFromCS = bestGeneration(deletionsInProgress.getEntries());
             }
             final RepositoryCleanupInProgress cleanupInProgress = state.custom(RepositoryCleanupInProgress.TYPE);
-            if (bestGenerationFromCS == RepositoryData.EMPTY_REPO_GEN && cleanupInProgress != null) {
+            if (bestGenerationFromCS == RepositoryData.UNKNOWN_REPO_GEN && cleanupInProgress != null) {
                 bestGenerationFromCS = bestGeneration(cleanupInProgress.entries());
             }
+            bestGenerationFromCS = Math.max(RepositoryData.EMPTY_REPO_GEN, bestGenerationFromCS);
             final long finalBestGen = Math.max(bestGenerationFromCS, metadata.generation());
             latestKnownRepoGen.updateAndGet(known -> Math.max(known, finalBestGen));
         } else {
@@ -378,7 +379,7 @@ private long bestGeneration(Collection<? extends RepositoryOperation> operations
         final String repoName = metadata.name();
         assert operations.size() <= 1 : "Assumed one or no operations but received " + operations;
         return operations.stream().filter(e -> e.repository().equals(repoName)).mapToLong(RepositoryOperation::repositoryStateId)
-            .max().orElse(RepositoryData.EMPTY_REPO_GEN);
+            .max().orElse(RepositoryData.UNKNOWN_REPO_GEN);
     }
 
     public ThreadPool threadPool() {

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

@@ -227,10 +227,10 @@ public ClusterState execute(ClusterState currentState) {
                         // Check if the snapshot to restore is currently being deleted
                         SnapshotDeletionsInProgress deletionsInProgress = currentState.custom(SnapshotDeletionsInProgress.TYPE);
                         if (deletionsInProgress != null
-                            && deletionsInProgress.getEntries().stream().anyMatch(entry -> entry.getSnapshot().equals(snapshot))) {
+                            && deletionsInProgress.getEntries().stream().anyMatch(entry -> entry.matches(snapshot))) {
                             throw new ConcurrentSnapshotExecutionException(snapshot,
                                 "cannot restore a snapshot while a snapshot deletion is in-progress [" +
-                                    deletionsInProgress.getEntries().get(0).getSnapshot() + "]");
+                                    deletionsInProgress.getEntries().get(0) + "]");
                         }
 
                         // Updating cluster state