Stricter Check for Snapshot Restore Version Compatibility (#65580)

We can add a pre-flight check for version compatibility that uses the
version in `SnapshotInfo` so that we don't need to load the `INdexMetadata`
for this which may not be readable 2 major versions back to begin with.

Note: this still technically leaves the option where a snapshot of major N contains
indices of major `N - 1` but sine we don't have any way of reading the index meta version
other than reading the actual `IndexMetadata` there isn't much we can do about this without
investing non-trivial efforts -> I think this is a good enough enhancement to the checks
for now.

Closes #65567

server/src/internalClusterTest/java/org/elasticsearch/snapshots/RestoreSnapshotIT.java

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.snapshots;
 
+import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionFuture;
 import org.elasticsearch.action.admin.cluster.snapshots.restore.RestoreSnapshotResponse;
 import org.elasticsearch.action.admin.indices.settings.get.GetSettingsResponse;
@@ -34,6 +35,7 @@ import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.xcontent.XContentFactory;
 import org.elasticsearch.indices.InvalidIndexNameException;
 import org.elasticsearch.repositories.RepositoriesService;
+import org.elasticsearch.repositories.fs.FsRepository;
 import org.elasticsearch.rest.RestStatus;
 
 import java.nio.file.Path;
@@ -686,4 +688,17 @@ public class RestoreSnapshotIT extends AbstractSnapshotIntegTestCase {
                         .get());
         assertThat(restoreError.getMessage(), containsString("cannot disable setting [index.soft_deletes.enabled] on restore"));
     }
+
+    public void testFailOnAncientVersion() throws Exception {
+        final String repoName = "test-repo";
+        final Path repoPath = randomRepoPath();
+        createRepository(repoName, FsRepository.TYPE, repoPath);
+        final Version oldVersion = Version.CURRENT.previousMajor().previousMajor();
+        final String oldSnapshot = initWithSnapshotVersion(repoName, repoPath, oldVersion);
+        final SnapshotRestoreException snapshotRestoreException = expectThrows(SnapshotRestoreException.class,
+                () -> client().admin().cluster().prepareRestoreSnapshot(repoName, oldSnapshot).execute().actionGet());
+        assertThat(snapshotRestoreException.getMessage(), containsString( "the snapshot was created with Elasticsearch version ["
+                + oldVersion + "] which is below the current versions minimum index compatibility version [" +
+                Version.CURRENT.minimumIndexCompatibilityVersion() + "]"));
+    }
 }

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

@@ -63,7 +63,7 @@ import java.util.Map;
 public final class ChecksumBlobStoreFormat<T extends ToXContent> {
 
     // Serialization parameters to specify correct context for metadata serialization
-    private static final ToXContent.Params SNAPSHOT_ONLY_FORMAT_PARAMS;
+    public static final ToXContent.Params SNAPSHOT_ONLY_FORMAT_PARAMS;
 
     static {
         Map<String, String> snapshotOnlyParams = new HashMap<>();

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

@@ -927,6 +927,12 @@ public class RestoreService implements ClusterStateApplier {
                                                "the snapshot was created with Elasticsearch version [" + snapshotInfo.version() +
                                                    "] which is higher than the version of this node [" + Version.CURRENT + "]");
         }
+        if (snapshotInfo.version().before(Version.CURRENT.minimumIndexCompatibilityVersion())) {
+            throw new SnapshotRestoreException(new Snapshot(repository, snapshotInfo.snapshotId()),
+                    "the snapshot was created with Elasticsearch version [" + snapshotInfo.version() +
+                            "] which is below the current versions minimum index compatibility version [" +
+                            Version.CURRENT.minimumIndexCompatibilityVersion() + "]");
+        }
     }
 
     public static boolean failed(SnapshotInfo snapshot, String index) {

test/framework/src/main/java/org/elasticsearch/snapshots/AbstractSnapshotIntegTestCase.java

@@ -21,6 +21,7 @@ package org.elasticsearch.snapshots;
 import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionFuture;
 import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.ActionRunnable;
 import org.elasticsearch.action.admin.cluster.snapshots.create.CreateSnapshotResponse;
 import org.elasticsearch.action.index.IndexRequestBuilder;
 import org.elasticsearch.action.search.SearchRequest;
@@ -44,6 +45,7 @@ import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.ByteSizeUnit;
 import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.common.util.BigArrays;
 import org.elasticsearch.common.xcontent.DeprecationHandler;
 import org.elasticsearch.common.xcontent.NamedXContentRegistry;
 import org.elasticsearch.common.xcontent.XContentBuilder;
@@ -57,6 +59,7 @@ import org.elasticsearch.repositories.RepositoryData;
 import org.elasticsearch.repositories.ShardGenerations;
 import org.elasticsearch.repositories.blobstore.BlobStoreRepository;
 import org.elasticsearch.repositories.blobstore.BlobStoreTestUtil;
+import org.elasticsearch.repositories.blobstore.ChecksumBlobStoreFormat;
 import org.elasticsearch.search.builder.SearchSourceBuilder;
 import org.elasticsearch.snapshots.mockstore.MockRepository;
 import org.elasticsearch.test.ESIntegTestCase;
@@ -321,7 +324,8 @@ public abstract class AbstractSnapshotIntegTestCase extends ESIntegTestCase {
         final CreateSnapshotResponse createSnapshotResponse = clusterAdmin()
                 .prepareCreateSnapshot(repoName, oldVersionSnapshot).setIndices("does-not-exist-for-sure-*")
                 .setWaitForCompletion(true).get();
-        assertThat(createSnapshotResponse.getSnapshotInfo().totalShards(), is(0));
+        final SnapshotInfo snapshotInfo = createSnapshotResponse.getSnapshotInfo();
+        assertThat(snapshotInfo.totalShards(), is(0));
 
         logger.info("--> writing downgraded RepositoryData for repository metadata version [{}]", version);
         final RepositoryData repositoryData = getRepositoryData(repoName);
@@ -336,6 +340,17 @@ public abstract class AbstractSnapshotIntegTestCase extends ESIntegTestCase {
                 BytesReference.toBytes(BytesReference.bytes(
                         downgradedRepoData.snapshotsToXContent(XContentFactory.jsonBuilder(), version))),
                 StandardOpenOption.TRUNCATE_EXISTING);
+        final SnapshotInfo downgradedSnapshotInfo = SnapshotInfo.fromXContentInternal(
+                JsonXContent.jsonXContent.createParser(
+                        NamedXContentRegistry.EMPTY,
+                        DeprecationHandler.THROW_UNSUPPORTED_OPERATION,
+                        Strings.toString(snapshotInfo, ChecksumBlobStoreFormat.SNAPSHOT_ONLY_FORMAT_PARAMS)
+                                .replace(String.valueOf(Version.CURRENT.id), String.valueOf(version.id))));
+        final BlobStoreRepository blobStoreRepository = getRepositoryOnMaster(repoName);
+        PlainActionFuture.get(f -> blobStoreRepository.threadPool().generic().execute(ActionRunnable.run(f, () ->
+                BlobStoreRepository.SNAPSHOT_FORMAT.write(downgradedSnapshotInfo,
+                        blobStoreRepository.blobStore().blobContainer(blobStoreRepository.basePath()), snapshotInfo.snapshotId().getUUID(),
+                        randomBoolean(), internalCluster().getCurrentMasterNodeInstance(BigArrays.class)))));
         return oldVersionSnapshot;
     }