Forbid the removal of the write block if the index is read-only (#120648)

Ensure that a `write` block cannot be removed on a read-only compatible index in version N-2, while allowing to change a `read_only` block into a `write` block if needed as well as closing/reopening such indices.

Requires #120647 to be merged on `8.x`.

Relates ES-10320

qa/lucene-index-compatibility/src/javaRestTest/java/org/elasticsearch/lucene/AbstractIndexCompatibilityTestCase.java

@@ -12,8 +12,12 @@ package org.elasticsearch.lucene;
 import org.apache.http.entity.ContentType;
 import org.apache.http.entity.InputStreamEntity;
 import org.elasticsearch.client.Request;
+import org.elasticsearch.client.ResponseException;
+import org.elasticsearch.cluster.block.ClusterBlock;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
+import org.elasticsearch.cluster.metadata.MetadataIndexStateService;
 import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.mapper.MapperService;
@@ -24,6 +28,7 @@ import org.elasticsearch.test.cluster.local.distribution.DistributionType;
 import org.elasticsearch.test.cluster.util.Version;
 import org.elasticsearch.test.rest.ESRestTestCase;
 import org.elasticsearch.xcontent.XContentType;
+import org.hamcrest.Matcher;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.ClassRule;
@@ -32,16 +37,18 @@ import org.junit.rules.TemporaryFolder;
 import org.junit.rules.TestRule;
 
 import java.io.IOException;
+import java.util.Comparator;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import java.util.stream.IntStream;
 
-import static org.elasticsearch.cluster.metadata.MetadataIndexStateService.VERIFIED_READ_ONLY_SETTING;
 import static org.elasticsearch.test.cluster.util.Version.CURRENT;
 import static org.elasticsearch.test.cluster.util.Version.fromString;
 import static org.elasticsearch.test.rest.ObjectPath.createFromResponse;
 import static org.hamcrest.Matchers.allOf;
+import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
 import static org.hamcrest.Matchers.is;
@@ -272,9 +279,51 @@ public abstract class AbstractIndexCompatibilityTestCase extends ESRestTestCase
         assertAcknowledged(client().performRequest(request));
     }
 
-    protected void assertThatIndexBlock(String indexName, IndexMetadata.APIBlock apiBlock) throws Exception {
+    private static ClusterBlock toIndexBlock(String blockId) {
+        int block = Integer.parseInt(blockId);
+        for (var indexBlock : List.of(
+            IndexMetadata.INDEX_READ_ONLY_BLOCK,
+            IndexMetadata.INDEX_READ_BLOCK,
+            IndexMetadata.INDEX_WRITE_BLOCK,
+            IndexMetadata.INDEX_METADATA_BLOCK,
+            IndexMetadata.INDEX_READ_ONLY_ALLOW_DELETE_BLOCK,
+            IndexMetadata.INDEX_REFRESH_BLOCK,
+            MetadataIndexStateService.INDEX_CLOSED_BLOCK
+        )) {
+            if (block == indexBlock.id()) {
+                return indexBlock;
+            }
+        }
+        throw new AssertionError("No index block found with id [" + blockId + ']');
+    }
+
+    @SuppressWarnings("unchecked")
+    protected static List<ClusterBlock> indexBlocks(String indexName) throws Exception {
+        var responseBody = createFromResponse(client().performRequest(new Request("GET", "_cluster/state/blocks/" + indexName)));
+        var blocks = (Map<String, ?>) responseBody.evaluate("blocks.indices." + indexName);
+        if (blocks == null || blocks.isEmpty()) {
+            return List.of();
+        }
+        return blocks.keySet()
+            .stream()
+            .map(AbstractIndexCompatibilityTestCase::toIndexBlock)
+            .sorted(Comparator.comparing(ClusterBlock::id))
+            .toList();
+    }
+
+    @SuppressWarnings("unchecked")
+    protected static void assertIndexSetting(String indexName, Setting<?> setting, Matcher<Boolean> matcher) throws Exception {
         var indexSettings = getIndexSettingsAsMap(indexName);
-        assertThat(indexSettings.get(VERIFIED_READ_ONLY_SETTING.getKey()), equalTo(Boolean.TRUE.toString()));
-        assertThat(indexSettings.get(apiBlock.settingName()), equalTo(Boolean.TRUE.toString()));
+        assertThat(Boolean.parseBoolean((String) indexSettings.get(setting.getKey())), matcher);
+    }
+
+    protected static ResponseException expectUpdateIndexSettingsThrows(String indexName, Settings.Builder settings) {
+        var exception = expectThrows(ResponseException.class, () -> updateIndexSettings(indexName, settings));
+        assertThat(exception.getResponse().getStatusLine().getStatusCode(), equalTo(400));
+        return exception;
+    }
+
+    protected static Matcher<String> containsStringCannotRemoveBlockOnReadOnlyIndex(String indexName) {
+        return allOf(containsString("Can't remove the write block on read-only compatible index"), containsString(indexName));
     }
 }

qa/lucene-index-compatibility/src/javaRestTest/java/org/elasticsearch/lucene/FullClusterRestartLuceneIndexCompatibilityIT.java

@@ -9,13 +9,23 @@
 
 package org.elasticsearch.lucene;
 
+import org.elasticsearch.client.ResponseException;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.repositories.fs.FsRepository;
 import org.elasticsearch.test.cluster.util.Version;
 
+import static org.elasticsearch.cluster.metadata.IndexMetadata.INDEX_READ_ONLY_BLOCK;
+import static org.elasticsearch.cluster.metadata.IndexMetadata.INDEX_WRITE_BLOCK;
+import static org.elasticsearch.cluster.metadata.MetadataIndexStateService.INDEX_CLOSED_BLOCK;
+import static org.elasticsearch.cluster.metadata.MetadataIndexStateService.VERIFIED_BEFORE_CLOSE_SETTING;
+import static org.elasticsearch.cluster.metadata.MetadataIndexStateService.VERIFIED_READ_ONLY_SETTING;
+import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.either;
 import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.is;
 
 public class FullClusterRestartLuceneIndexCompatibilityIT extends FullClusterRestartIndexCompatibilityTestCase {
 
@@ -28,14 +38,13 @@ public class FullClusterRestartLuceneIndexCompatibilityIT extends FullClusterRes
     }
 
     /**
-     * Creates an index on N-2, upgrades to N -1 and marks as read-only, then upgrades to N.
+     * Creates an index on N-2, upgrades to N-1 and marks as read-only, then upgrades to N.
      */
     public void testIndexUpgrade() throws Exception {
         final String index = suffix("index");
         final int numDocs = 2431;
 
         if (isFullyUpgradedTo(VERSION_MINUS_2)) {
-            logger.debug("--> creating index [{}]", index);
             createIndex(
                 client(),
                 index,
@@ -45,29 +54,85 @@ public class FullClusterRestartLuceneIndexCompatibilityIT extends FullClusterRes
                     .put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), true)
                     .build()
             );
-
-            logger.debug("--> indexing [{}] docs in [{}]", numDocs, index);
             indexDocs(index, numDocs);
             return;
         }
 
-        if (isFullyUpgradedTo(VERSION_MINUS_1)) {
-            ensureGreen(index);
+        assertThat(indexVersion(index), equalTo(VERSION_MINUS_2));
+        ensureGreen(index);
 
-            assertThat(indexVersion(index), equalTo(VERSION_MINUS_2));
+        if (isIndexClosed(index) == false) {
             assertDocCount(client(), index, numDocs);
+        }
 
-            addIndexBlock(index, IndexMetadata.APIBlock.WRITE);
+        if (isFullyUpgradedTo(VERSION_MINUS_1)) {
+            final boolean maybeClose = randomBoolean();
+            if (maybeClose) {
+                logger.debug("--> closing index [{}] before upgrade", index);
+                closeIndex(index);
+            }
+
+            final var block = randomFrom(IndexMetadata.APIBlock.WRITE, IndexMetadata.APIBlock.READ_ONLY);
+            addIndexBlock(index, block);
+
+            assertThat(indexBlocks(index), maybeClose ? contains(INDEX_CLOSED_BLOCK, block.getBlock()) : contains(block.getBlock()));
+            assertIndexSetting(index, VERIFIED_BEFORE_CLOSE_SETTING, is(maybeClose));
+            assertIndexSetting(index, VERIFIED_READ_ONLY_SETTING, is(true));
             return;
         }
 
         if (isFullyUpgradedTo(VERSION_CURRENT)) {
-            ensureGreen(index);
+            final var isClosed = isIndexClosed(index);
+            logger.debug("--> upgraded index [{}] is in [{}] state", index, isClosed ? "closed" : "open");
+            assertThat(
+                indexBlocks(index),
+                isClosed
+                    ? either(contains(INDEX_CLOSED_BLOCK, INDEX_WRITE_BLOCK)).or(contains(INDEX_CLOSED_BLOCK, INDEX_READ_ONLY_BLOCK))
+                    : either(contains(INDEX_WRITE_BLOCK)).or(contains(INDEX_READ_ONLY_BLOCK))
+            );
+            assertIndexSetting(index, VERIFIED_BEFORE_CLOSE_SETTING, is(isClosed));
+            assertIndexSetting(index, VERIFIED_READ_ONLY_SETTING, is(true));
 
-            assertThat(indexVersion(index), equalTo(VERSION_MINUS_2));
-            assertDocCount(client(), index, numDocs);
+            if (isClosed == false) {
+                logger.debug("--> write/read_only API blocks cannot be removed on an opened index");
+                var ex = expectUpdateIndexSettingsThrows(
+                    index,
+                    Settings.builder()
+                        .putNull(IndexMetadata.APIBlock.WRITE.settingName())
+                        .putNull(IndexMetadata.APIBlock.READ_ONLY.settingName())
+                );
+                assertThat(ex.getMessage(), containsStringCannotRemoveBlockOnReadOnlyIndex(index));
 
-            assertThatIndexBlock(index, IndexMetadata.APIBlock.WRITE);
+            } else if (randomBoolean()) {
+                logger.debug("--> write/read_only API blocks can be removed on a closed index: INDEX_CLOSED_BLOCK already blocks writes");
+                updateIndexSettings(
+                    index,
+                    Settings.builder()
+                        .putNull(IndexMetadata.APIBlock.WRITE.settingName())
+                        .putNull(IndexMetadata.APIBlock.READ_ONLY.settingName())
+                );
+                logger.debug("--> but attempts to re-opening [{}] should fail due to the missing block", index);
+                var ex = expectThrows(ResponseException.class, () -> openIndex(index));
+                assertThat(ex.getMessage(), containsString("must be marked as read-only"));
+
+                // TODO this could be randomized once we support recovering verified-before-close closed indices with no write/ro block
+                addIndexBlock(index, IndexMetadata.APIBlock.WRITE);
+            }
+
+            var block = indexBlocks(index).stream().filter(c -> c.equals(INDEX_WRITE_BLOCK) || c.equals(INDEX_READ_ONLY_BLOCK)).findFirst();
+            if (block.isPresent() && block.get().equals(INDEX_READ_ONLY_BLOCK)) {
+                logger.debug("--> read_only API block can be replaced by a write block (required for the remaining tests)");
+                updateIndexSettings(
+                    index,
+                    Settings.builder()
+                        .putNull(IndexMetadata.APIBlock.READ_ONLY.settingName())
+                        .put(IndexMetadata.APIBlock.WRITE.settingName(), true)
+                );
+            }
+
+            assertThat(indexBlocks(index), isClosed ? contains(INDEX_CLOSED_BLOCK, INDEX_WRITE_BLOCK) : contains(INDEX_WRITE_BLOCK));
+            assertIndexSetting(index, VERIFIED_BEFORE_CLOSE_SETTING, is(isClosed));
+            assertIndexSetting(index, VERIFIED_READ_ONLY_SETTING, is(true));
 
             var numberOfReplicas = getNumberOfReplicas(index);
             if (0 < numberOfReplicas) {
@@ -82,66 +147,29 @@ public class FullClusterRestartLuceneIndexCompatibilityIT extends FullClusterRes
             updateIndexSettings(index, Settings.builder().put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 1));
             ensureGreen(index);
 
-            logger.debug("--> closing restored index [{}]", index);
-            closeIndex(index);
-            ensureGreen(index);
+            if (isClosed) {
+                logger.debug("--> re-opening index [{}]", index);
+                openIndex(index);
+                ensureGreen(index);
 
-            logger.debug("--> adding replica to test peer-recovery for closed shards");
+                assertDocCount(client(), index, numDocs);
+            } else {
+                logger.debug("--> closing index [{}]", index);
+                closeIndex(index);
+                ensureGreen(index);
+            }
+
+            logger.debug("--> adding more replicas to test peer-recovery");
             updateIndexSettings(index, Settings.builder().put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 2));
             ensureGreen(index);
 
-            logger.debug("--> re-opening restored index [{}]", index);
-            openIndex(index);
-            ensureGreen(index);
-
-            assertDocCount(client(), index, numDocs);
-
-            logger.debug("--> deleting index [{}]", index);
-            deleteIndex(index);
-        }
-    }
-
-    /**
-     * Similar to {@link #testIndexUpgrade()} but with a read_only block.
-     */
-    public void testIndexUpgradeReadOnlyBlock() throws Exception {
-        final String index = suffix("index");
-        final int numDocs = 2531;
-
-        if (isFullyUpgradedTo(VERSION_MINUS_2)) {
-            logger.debug("--> creating index [{}]", index);
-            createIndex(
-                client(),
-                index,
-                Settings.builder()
-                    .put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, 1)
-                    .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, randomInt(2))
-                    .put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), true)
-                    .build()
+            assertIndexSetting(index, VERIFIED_READ_ONLY_SETTING, is(true));
+            assertThat(
+                indexBlocks(index),
+                isIndexClosed(index) ? contains(INDEX_CLOSED_BLOCK, INDEX_WRITE_BLOCK) : contains(INDEX_WRITE_BLOCK)
             );
 
-            logger.debug("--> indexing [{}] docs in [{}]", numDocs, index);
-            indexDocs(index, numDocs);
-            return;
-        }
-
-        if (isFullyUpgradedTo(VERSION_MINUS_1)) {
-            ensureGreen(index);
-
-            assertThat(indexVersion(index), equalTo(VERSION_MINUS_2));
-            assertDocCount(client(), index, numDocs);
-
-            addIndexBlock(index, IndexMetadata.APIBlock.READ_ONLY);
-            return;
-        }
-
-        if (isFullyUpgradedTo(VERSION_CURRENT)) {
-            ensureGreen(index);
-
-            assertThat(indexVersion(index), equalTo(VERSION_MINUS_2));
-            assertDocCount(client(), index, numDocs);
-
-            assertThatIndexBlock(index, IndexMetadata.APIBlock.READ_ONLY);
+            deleteIndex(index);
         }
     }
 
@@ -196,7 +224,8 @@ public class FullClusterRestartLuceneIndexCompatibilityIT extends FullClusterRes
             restoreIndex(repository, snapshot, index, restoredIndex);
             ensureGreen(restoredIndex);
 
-            assertThatIndexBlock(restoredIndex, IndexMetadata.APIBlock.WRITE);
+            assertIndexSetting(restoredIndex, VERIFIED_READ_ONLY_SETTING, is(true));
+            assertThat(indexBlocks(restoredIndex), contains(INDEX_WRITE_BLOCK));
             assertThat(indexVersion(restoredIndex), equalTo(VERSION_MINUS_2));
             assertDocCount(client(), restoredIndex, numDocs);
 
@@ -277,7 +306,8 @@ public class FullClusterRestartLuceneIndexCompatibilityIT extends FullClusterRes
 
         if (isFullyUpgradedTo(VERSION_CURRENT)) {
             assertThat(isIndexClosed(index), equalTo(true));
-            assertThatIndexBlock(index, IndexMetadata.APIBlock.WRITE);
+            assertThat(indexBlocks(index), contains(INDEX_CLOSED_BLOCK, INDEX_WRITE_BLOCK));
+            assertIndexSetting(index, VERIFIED_READ_ONLY_SETTING, is(true));
 
             logger.debug("--> restoring index [{}] over existing closed index", index);
             restoreIndex(repository, snapshot, index, index);

qa/lucene-index-compatibility/src/javaRestTest/java/org/elasticsearch/lucene/RollingUpgradeLuceneIndexCompatibilityTestCase.java

@@ -19,7 +19,19 @@ import org.elasticsearch.test.cluster.util.Version;
 
 import java.util.List;
 
+import static org.elasticsearch.cluster.metadata.IndexMetadata.INDEX_READ_ONLY_BLOCK;
+import static org.elasticsearch.cluster.metadata.IndexMetadata.INDEX_WRITE_BLOCK;
+import static org.elasticsearch.cluster.metadata.MetadataIndexStateService.INDEX_CLOSED_BLOCK;
+import static org.elasticsearch.cluster.metadata.MetadataIndexStateService.VERIFIED_BEFORE_CLOSE_SETTING;
+import static org.elasticsearch.cluster.metadata.MetadataIndexStateService.VERIFIED_READ_ONLY_SETTING;
+import static org.hamcrest.Matchers.allOf;
+import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.either;
 import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.hasItem;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.not;
 
 public class RollingUpgradeLuceneIndexCompatibilityTestCase extends RollingUpgradeIndexCompatibilityTestCase {
 
@@ -39,7 +51,6 @@ public class RollingUpgradeLuceneIndexCompatibilityTestCase extends RollingUpgra
         final int numDocs = 2543;
 
         if (isFullyUpgradedTo(VERSION_MINUS_2)) {
-            logger.debug("--> creating index [{}]", index);
             createIndex(
                 client(),
                 index,
@@ -49,26 +60,108 @@ public class RollingUpgradeLuceneIndexCompatibilityTestCase extends RollingUpgra
                     .put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), true)
                     .build()
             );
-
-            logger.debug("--> indexing [{}] docs in [{}]", numDocs, index);
             indexDocs(index, numDocs);
             return;
         }
 
+        assertThat(indexVersion(index), equalTo(VERSION_MINUS_2));
         ensureGreen(index);
 
-        if (isFullyUpgradedTo(VERSION_MINUS_1)) {
-            assertThat(indexVersion(index), equalTo(VERSION_MINUS_2));
+        if (isIndexClosed(index) == false) {
             assertDocCount(client(), index, numDocs);
+        }
 
-            addIndexBlock(index, IndexMetadata.APIBlock.WRITE);
+        if (isFullyUpgradedTo(VERSION_MINUS_1)) {
+            final var maybeClose = randomBoolean();
+            if (maybeClose) {
+                logger.debug("--> closing index [{}] before upgrade", index);
+                closeIndex(index);
+            }
+
+            final var randomBlocks = randomFrom(
+                List.of(IndexMetadata.APIBlock.WRITE, IndexMetadata.APIBlock.READ_ONLY),
+                List.of(IndexMetadata.APIBlock.READ_ONLY),
+                List.of(IndexMetadata.APIBlock.WRITE)
+            );
+            for (var randomBlock : randomBlocks) {
+                addIndexBlock(index, randomBlock);
+                assertThat(indexBlocks(index), hasItem(randomBlock.getBlock()));
+            }
+
+            assertThat(indexBlocks(index), maybeClose ? hasItem(INDEX_CLOSED_BLOCK) : not(hasItem(INDEX_CLOSED_BLOCK)));
+            assertIndexSetting(index, VERIFIED_BEFORE_CLOSE_SETTING, is(maybeClose));
+            assertIndexSetting(index, VERIFIED_READ_ONLY_SETTING, is(true));
             return;
         }
 
         if (nodesVersions().values().stream().anyMatch(v -> v.onOrAfter(VERSION_CURRENT))) {
-            assertThatIndexBlock(index, IndexMetadata.APIBlock.WRITE);
+            final var isClosed = isIndexClosed(index);
+            logger.debug("--> upgraded index [{}] is now in [{}] state", index, isClosed ? "closed" : "open");
+            assertThat(
+                indexBlocks(index),
+                allOf(
+                    either(hasItem(INDEX_READ_ONLY_BLOCK)).or(hasItem(INDEX_WRITE_BLOCK)),
+                    isClosed ? hasItem(INDEX_CLOSED_BLOCK) : not(hasItem(INDEX_CLOSED_BLOCK))
+                )
+            );
+            assertIndexSetting(index, VERIFIED_BEFORE_CLOSE_SETTING, is(isClosed));
+            assertIndexSetting(index, VERIFIED_READ_ONLY_SETTING, is(true));
 
-            if (isIndexClosed(index)) {
+            var blocks = indexBlocks(index).stream().filter(c -> c.equals(INDEX_WRITE_BLOCK) || c.equals(INDEX_READ_ONLY_BLOCK)).toList();
+            if (blocks.size() == 2) {
+                switch (randomInt(2)) {
+                    case 0:
+                        updateIndexSettings(
+                            index,
+                            Settings.builder()
+                                .putNull(IndexMetadata.APIBlock.WRITE.settingName())
+                                .put(IndexMetadata.APIBlock.READ_ONLY.settingName(), true)
+                        );
+                        assertThat(
+                            indexBlocks(index),
+                            isClosed ? contains(INDEX_CLOSED_BLOCK, INDEX_READ_ONLY_BLOCK) : contains(INDEX_READ_ONLY_BLOCK)
+                        );
+                        break;
+                    case 1:
+                        updateIndexSettings(
+                            index,
+                            Settings.builder()
+                                .putNull(IndexMetadata.APIBlock.READ_ONLY.settingName())
+                                .put(IndexMetadata.APIBlock.WRITE.settingName(), true)
+                        );
+                        assertThat(
+                            indexBlocks(index),
+                            isClosed ? contains(INDEX_CLOSED_BLOCK, INDEX_WRITE_BLOCK) : contains(INDEX_WRITE_BLOCK)
+                        );
+                        break;
+                    case 2:
+                        updateIndexSettings(index, Settings.builder().put(IndexMetadata.APIBlock.READ_ONLY.settingName(), false));
+                        assertThat(
+                            indexBlocks(index),
+                            isClosed ? contains(INDEX_CLOSED_BLOCK, INDEX_WRITE_BLOCK) : contains(INDEX_WRITE_BLOCK)
+                        );
+                        break;
+                    default:
+                        throw new AssertionError();
+                }
+            }
+
+            blocks = indexBlocks(index).stream().filter(c -> c.equals(INDEX_WRITE_BLOCK) || c.equals(INDEX_READ_ONLY_BLOCK)).toList();
+            if (blocks.contains(INDEX_READ_ONLY_BLOCK)) {
+                logger.debug("--> read_only API block can be replaced by a write block (required for the remaining tests)");
+                updateIndexSettings(
+                    index,
+                    Settings.builder()
+                        .putNull(IndexMetadata.APIBlock.READ_ONLY.settingName())
+                        .put(IndexMetadata.APIBlock.WRITE.settingName(), true)
+                );
+            }
+
+            assertIndexSetting(index, VERIFIED_READ_ONLY_SETTING, is(true));
+            assertIndexSetting(index, VERIFIED_BEFORE_CLOSE_SETTING, is(isClosed));
+            assertThat(indexBlocks(index), isClosed ? contains(INDEX_CLOSED_BLOCK, INDEX_WRITE_BLOCK) : contains(INDEX_WRITE_BLOCK));
+
+            if (isClosed) {
                 logger.debug("--> re-opening index [{}] after upgrade", index);
                 openIndex(index);
                 ensureGreen(index);
@@ -88,48 +181,6 @@ public class RollingUpgradeLuceneIndexCompatibilityTestCase extends RollingUpgra
         }
     }
 
-    /**
-     * Similar to {@link #testIndexUpgrade()} but with a read_only block.
-     */
-    public void testIndexUpgradeReadOnlyBlock() throws Exception {
-        final String index = suffix("index-");
-        final int numDocs = 2573;
-
-        if (isFullyUpgradedTo(VERSION_MINUS_2)) {
-            logger.debug("--> creating index [{}]", index);
-            createIndex(
-                client(),
-                index,
-                Settings.builder()
-                    .put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, 1)
-                    .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0)
-                    .put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), true)
-                    .build()
-            );
-
-            logger.debug("--> indexing [{}] docs in [{}]", numDocs, index);
-            indexDocs(index, numDocs);
-            return;
-        }
-
-        ensureGreen(index);
-
-        if (isFullyUpgradedTo(VERSION_MINUS_1)) {
-            assertThat(indexVersion(index), equalTo(VERSION_MINUS_2));
-            assertDocCount(client(), index, numDocs);
-
-            addIndexBlock(index, IndexMetadata.APIBlock.READ_ONLY);
-            return;
-        }
-
-        if (nodesVersions().values().stream().anyMatch(v -> v.onOrAfter(VERSION_CURRENT))) {
-            assertThatIndexBlock(index, IndexMetadata.APIBlock.READ_ONLY);
-
-            assertThat(indexVersion(index), equalTo(VERSION_MINUS_2));
-            assertDocCount(client(), index, numDocs);
-        }
-    }
-
     /**
      * Creates an index on N-2, marks as read-only on N-1 and creates a snapshot, then restores the snapshot during rolling upgrades to N.
      */
@@ -174,16 +225,24 @@ public class RollingUpgradeLuceneIndexCompatibilityTestCase extends RollingUpgra
             deleteIndex(index);
             return;
         }
+
         if (nodesVersions().values().stream().anyMatch(v -> v.onOrAfter(VERSION_CURRENT))) {
             var restoredIndex = suffix("index-restored-rolling");
             boolean success = false;
             try {
-
                 logger.debug("--> restoring index [{}] as [{}]", index, restoredIndex);
                 restoreIndex(repository, snapshot, index, restoredIndex);
                 ensureGreen(restoredIndex);
 
-                assertThatIndexBlock(restoredIndex, IndexMetadata.APIBlock.WRITE);
+                assertThat(indexBlocks(restoredIndex), contains(INDEX_WRITE_BLOCK));
+                assertIndexSetting(restoredIndex, VERIFIED_READ_ONLY_SETTING, is(true));
+
+                var ex = expectUpdateIndexSettingsThrows(
+                    restoredIndex,
+                    Settings.builder().putNull(IndexMetadata.APIBlock.WRITE.settingName())
+                );
+                assertThat(ex.getMessage(), containsStringCannotRemoveBlockOnReadOnlyIndex(restoredIndex));
+
                 assertThat(indexVersion(restoredIndex), equalTo(VERSION_MINUS_2));
                 assertDocCount(client(), restoredIndex, numDocs);
 
@@ -194,6 +253,15 @@ public class RollingUpgradeLuceneIndexCompatibilityTestCase extends RollingUpgra
                 closeIndex(restoredIndex);
                 ensureGreen(restoredIndex);
 
+                logger.debug("--> write API block can be removed on a closed index: INDEX_CLOSED_BLOCK already blocks writes");
+                updateIndexSettings(restoredIndex, Settings.builder().putNull(IndexMetadata.APIBlock.WRITE.settingName()));
+
+                logger.debug("--> but attempts to re-opening [{}] should fail due to the missing block", restoredIndex);
+                ex = expectThrows(ResponseException.class, () -> openIndex(restoredIndex));
+                assertThat(ex.getMessage(), containsString("must be marked as read-only"));
+
+                addIndexBlock(restoredIndex, IndexMetadata.APIBlock.WRITE);
+
                 logger.debug("--> re-opening restored index [{}]", restoredIndex);
                 openIndex(restoredIndex);
                 ensureGreen(restoredIndex);
@@ -214,5 +282,20 @@ public class RollingUpgradeLuceneIndexCompatibilityTestCase extends RollingUpgra
                 }
             }
         }
+
+        if (isFullyUpgradedTo(VERSION_CURRENT)) {
+            var exception = expectThrows(
+                ResponseException.class,
+                () -> restoreIndex(
+                    repository,
+                    snapshot,
+                    index,
+                    suffix("unrestorable"),
+                    Settings.builder().put(IndexMetadata.APIBlock.WRITE.settingName(), false).build()
+                )
+            );
+            assertThat(exception.getResponse().getStatusLine().getStatusCode(), equalTo(500));
+            assertThat(exception.getMessage(), containsString("must be marked as read-only using the setting"));
+        }
     }
 }

server/src/internalClusterTest/java/org/elasticsearch/indices/settings/UpdateSettingsIT.java

@@ -13,6 +13,7 @@ import org.elasticsearch.action.RequestBuilder;
 import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
 import org.elasticsearch.action.admin.indices.settings.get.GetSettingsResponse;
 import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.block.ClusterBlockException;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Priority;
@@ -23,6 +24,7 @@ import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.index.IndexModule;
 import org.elasticsearch.index.IndexService;
+import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.VersionType;
 import org.elasticsearch.index.engine.VersionConflictEngineException;
 import org.elasticsearch.indices.IndicesService;
@@ -720,4 +722,35 @@ public class UpdateSettingsIT extends ESIntegTestCase {
         }
     }
 
+    public void testMultipleSettingsUpdateWithMetadataWriteBlock() {
+        final var indexName = randomIdentifier();
+        createIndex(indexName, Settings.builder().put(IndexMetadata.APIBlock.READ_ONLY.settingName(), true).build());
+
+        // Metadata writes are blocked by the READ_ONLY block
+        expectThrows(
+            ClusterBlockException.class,
+            () -> updateIndexSettings(Settings.builder().put(IndexSettings.INDEX_REFRESH_INTERVAL_SETTING.getKey(), "12s"), indexName)
+        );
+
+        var randomSetting = randomFrom(IndexMetadata.APIBlock.READ_ONLY, IndexMetadata.APIBlock.READ_ONLY_ALLOW_DELETE).settingName();
+        updateIndexSettings(
+            Settings.builder()
+                .put(randomSetting, true) // still has the metadata write block...
+                .put(IndexMetadata.APIBlock.WRITE.settingName(), true)
+                .put(IndexSettings.INDEX_REFRESH_INTERVAL_SETTING.getKey(), "12s"), // should not be allowed
+            indexName
+        );
+
+        assertThat(
+            indicesAdmin().prepareGetSettings(indexName)
+                .get()
+                .getIndexToSettings()
+                .get(indexName)
+                .get(IndexSettings.INDEX_REFRESH_INTERVAL_SETTING.getKey()),
+            equalTo("12s")
+        );
+
+        // Updating the setting alone should always work
+        updateIndexSettings(Settings.builder().put(IndexMetadata.APIBlock.READ_ONLY.settingName(), false));
+    }
 }

server/src/main/java/org/elasticsearch/cluster/block/ClusterBlocks.java

@@ -146,6 +146,10 @@ public class ClusterBlocks implements SimpleDiffable<ClusterBlocks> {
         return indicesBlocks.containsKey(index) && indicesBlocks.get(index).contains(block);
     }
 
+    public boolean hasIndexBlockLevel(String index, ClusterBlockLevel level) {
+        return blocksForIndex(level, index).isEmpty() == false;
+    }
+
     public boolean hasIndexBlockWithId(String index, int blockId) {
         final Set<ClusterBlock> clusterBlocks = indicesBlocks.get(index);
         if (clusterBlocks != null) {
@@ -398,6 +402,10 @@ public class ClusterBlocks implements SimpleDiffable<ClusterBlocks> {
             return indices.getOrDefault(index, Set.of()).contains(block);
         }
 
+        public boolean hasIndexBlockLevel(String index, ClusterBlockLevel level) {
+            return indices.getOrDefault(index, Set.of()).stream().anyMatch(clusterBlock -> clusterBlock.contains(level));
+        }
+
         public Builder removeIndexBlock(String index, ClusterBlock block) {
             if (indices.containsKey(index) == false) {
                 return this;

server/src/main/java/org/elasticsearch/cluster/metadata/MetadataUpdateSettingsService.java

@@ -36,6 +36,7 @@ import org.elasticsearch.common.regex.Regex;
 import org.elasticsearch.common.settings.IndexScopedSettings;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.index.Index;
 import org.elasticsearch.index.IndexSettings;
@@ -51,7 +52,9 @@ import java.util.Locale;
 import java.util.Objects;
 import java.util.Set;
 import java.util.function.BiFunction;
+import java.util.function.Function;
 
+import static org.elasticsearch.cluster.metadata.MetadataIndexStateService.VERIFIED_READ_ONLY_SETTING;
 import static org.elasticsearch.index.IndexSettings.same;
 
 /**
@@ -181,11 +184,14 @@ public class MetadataUpdateSettingsService {
 
             RoutingTable.Builder routingTableBuilder = null;
             Metadata.Builder metadataBuilder = Metadata.builder(currentState.metadata());
+            final var minSupportedIndexVersion = currentState.nodes().getMinSupportedIndexVersion();
 
             // allow to change any settings to a closed index, and only allow dynamic settings to be changed
             // on an open index
             Set<Index> openIndices = new HashSet<>();
             Set<Index> closedIndices = new HashSet<>();
+            Set<Index> readOnlyIndices = null;
+
             final String[] actualIndices = new String[request.indices().length];
             for (int i = 0; i < request.indices().length; i++) {
                 Index index = request.indices()[i];
@@ -197,6 +203,12 @@ public class MetadataUpdateSettingsService {
                 } else {
                     closedIndices.add(index);
                 }
+                if (metadata.getCompatibilityVersion().before(minSupportedIndexVersion)) {
+                    if (readOnlyIndices == null) {
+                        readOnlyIndices = new HashSet<>();
+                    }
+                    readOnlyIndices.add(index);
+                }
             }
 
             if (skippedSettings.isEmpty() == false && openIndices.isEmpty() == false) {
@@ -327,10 +339,21 @@ public class MetadataUpdateSettingsService {
                 }
             }
 
+            final Function<String, Boolean> verifiedReadOnly = indexName -> VERIFIED_READ_ONLY_SETTING.get(
+                currentState.metadata().index(indexName).getSettings()
+            );
             final ClusterBlocks.Builder blocks = ClusterBlocks.builder().blocks(currentState.blocks());
             boolean changedBlocks = false;
             for (IndexMetadata.APIBlock block : IndexMetadata.APIBlock.values()) {
-                changedBlocks |= maybeUpdateClusterBlock(actualIndices, blocks, block.block, block.setting, openSettings, metadataBuilder);
+                changedBlocks |= maybeUpdateClusterBlock(
+                    actualIndices,
+                    blocks,
+                    block.block,
+                    block.setting,
+                    openSettings,
+                    metadataBuilder,
+                    verifiedReadOnly
+                );
             }
             changed |= changedBlocks;
 
@@ -359,6 +382,7 @@ public class MetadataUpdateSettingsService {
                     // This step is mandatory since we allow to update non-dynamic settings on closed indices.
                     indicesService.verifyIndexMetadata(updatedMetadata, updatedMetadata);
                 }
+                verifyReadOnlyIndices(readOnlyIndices, updatedState.blocks());
             } catch (IOException ex) {
                 throw ExceptionsHelper.convertToElastic(ex);
             }
@@ -417,6 +441,24 @@ public class MetadataUpdateSettingsService {
         }
     }
 
+    /**
+     * Verifies that read-only compatible indices always have a write block.
+     *
+     * @param readOnlyIndices the read-only compatible indices
+     * @param blocks the updated cluster state blocks
+     */
+    private static void verifyReadOnlyIndices(@Nullable Set<Index> readOnlyIndices, ClusterBlocks blocks) {
+        if (readOnlyIndices != null) {
+            for (Index readOnlyIndex : readOnlyIndices) {
+                if (blocks.hasIndexBlockLevel(readOnlyIndex.getName(), ClusterBlockLevel.WRITE) == false) {
+                    throw new IllegalArgumentException(
+                        String.format(Locale.ROOT, "Can't remove the write block on read-only compatible index %s", readOnlyIndex)
+                    );
+                }
+            }
+        }
+    }
+
     /**
      * Updates the cluster block only iff the setting exists in the given settings
      */
@@ -426,7 +468,8 @@ public class MetadataUpdateSettingsService {
         ClusterBlock block,
         Setting<Boolean> setting,
         Settings openSettings,
-        Metadata.Builder metadataBuilder
+        Metadata.Builder metadataBuilder,
+        Function<String, Boolean> verifiedReadOnlyBeforeBlockChanges
     ) {
         boolean changed = false;
         if (setting.exists(openSettings)) {
@@ -436,16 +479,32 @@ public class MetadataUpdateSettingsService {
                     if (blocks.hasIndexBlock(index, block) == false) {
                         blocks.addIndexBlock(index, block);
                         changed = true;
+                        if (block.contains(ClusterBlockLevel.WRITE)) {
+                            var isVerifiedReadOnly = verifiedReadOnlyBeforeBlockChanges.apply(index);
+                            if (isVerifiedReadOnly) {
+                                var indexMetadata = metadataBuilder.get(index);
+                                metadataBuilder.put(
+                                    IndexMetadata.builder(indexMetadata)
+                                        .settings(
+                                            Settings.builder()
+                                                .put(indexMetadata.getSettings())
+                                                .put(VERIFIED_READ_ONLY_SETTING.getKey(), true)
+                                        )
+                                );
+                            }
+                        }
                     }
                 } else {
                     if (blocks.hasIndexBlock(index, block)) {
                         blocks.removeIndexBlock(index, block);
                         changed = true;
                         if (block.contains(ClusterBlockLevel.WRITE)) {
-                            IndexMetadata indexMetadata = metadataBuilder.get(index);
-                            Settings.Builder indexSettings = Settings.builder().put(indexMetadata.getSettings());
-                            indexSettings.remove(MetadataIndexStateService.VERIFIED_READ_ONLY_SETTING.getKey());
-                            metadataBuilder.put(IndexMetadata.builder(indexMetadata).settings(indexSettings));
+                            if (blocks.hasIndexBlockLevel(index, ClusterBlockLevel.WRITE) == false) {
+                                var indexMetadata = metadataBuilder.get(index);
+                                var indexSettings = Settings.builder().put(indexMetadata.getSettings());
+                                indexSettings.remove(VERIFIED_READ_ONLY_SETTING.getKey());
+                                metadataBuilder.put(IndexMetadata.builder(indexMetadata).settings(indexSettings));
+                            }
                         }
                     }
                 }

test/framework/src/main/java/org/elasticsearch/test/rest/ESRestTestCase.java

@@ -1929,7 +1929,7 @@ public abstract class ESRestTestCase extends ESTestCase {
     }
 
     @SuppressWarnings("unchecked")
-    protected Map<String, Object> getIndexSettingsAsMap(String index) throws IOException {
+    protected static Map<String, Object> getIndexSettingsAsMap(String index) throws IOException {
         Map<String, Object> indexSettings = getIndexSettings(index);
         return (Map<String, Object>) ((Map<String, Object>) indexSettings.get(index)).get("settings");
     }