Merge branch 'master' into geirst/config-defaults-for-improved-merge-performance

38 files changed, 465 insertions, 232 deletions

diff --git a/config-model-api/src/main/java/com/yahoo/config/model/api/ModelContext.java b/config-model-api/src/main/java/com/yahoo/config/model/api/ModelContext.java
index 03813d21ca8..6108c39f9d3 100644
--- a/config-model-api/src/main/java/com/yahoo/config/model/api/ModelContext.java
+++ b/config-model-api/src/main/java/com/yahoo/config/model/api/ModelContext.java
@@ -83,24 +83,22 @@ public interface ModelContext {
         @ModelFeatureFlag(owners = {"baldersheim"}) default boolean useAsyncMessageHandlingOnSchedule() { throw new UnsupportedOperationException("TODO specify default value"); }
         @ModelFeatureFlag(owners = {"baldersheim"}) default double feedConcurrency() { throw new UnsupportedOperationException("TODO specify default value"); }
         @ModelFeatureFlag(owners = {"baldersheim"}) default int metricsproxyNumThreads() { throw new UnsupportedOperationException("TODO specify default value"); }
-        @ModelFeatureFlag(owners = {"baldersheim"}, removeAfter = "7.526") default int largeRankExpressionLimit() { return 8192; }
+        @ModelFeatureFlag(owners = {"baldersheim"}, removeAfter = "7.527") default int largeRankExpressionLimit() { return 8192; }
         @ModelFeatureFlag(owners = {"baldersheim"}) default int maxUnCommittedMemory() { return 130000; }
         @ModelFeatureFlag(owners = {"baldersheim"}) default int maxConcurrentMergesPerNode() { return 16; }
         @ModelFeatureFlag(owners = {"baldersheim"}) default int maxMergeQueueSize() { return 100; }
         @ModelFeatureFlag(owners = {"vekterli", "geirst"}) default boolean ignoreMergeQueueLimit() { return true; }
         @ModelFeatureFlag(owners = {"baldersheim"}) default boolean containerDumpHeapOnShutdownTimeout() { throw new UnsupportedOperationException("TODO specify default value"); }
         @ModelFeatureFlag(owners = {"baldersheim"}) default double containerShutdownTimeout() { throw new UnsupportedOperationException("TODO specify default value"); }
-        @ModelFeatureFlag(owners = {"baldersheim"}) default double diskBloatFactor() { throw new UnsupportedOperationException("TODO specify default value"); }
-        @ModelFeatureFlag(owners = {"baldersheim"}) default int docstoreCompressionLevel() { throw new UnsupportedOperationException("TODO specify default value"); }
+        @ModelFeatureFlag(owners = {"baldersheim"}, removeAfter = "7.527") default double diskBloatFactor() { return 0.25; }
+        @ModelFeatureFlag(owners = {"baldersheim"}, removeAfter = "7.527") default int docstoreCompressionLevel() { return 3; }
         @ModelFeatureFlag(owners = {"geirst"}) default boolean enableFeedBlockInDistributor() { return true; }
         @ModelFeatureFlag(owners = {"bjorncs", "tokle"}) default List<String> allowedAthenzProxyIdentities() { return List.of(); }
         @ModelFeatureFlag(owners = {"vekterli"}) default int maxActivationInhibitedOutOfSyncGroups() { return 0; }
         @ModelFeatureFlag(owners = {"hmusum"}) default String jvmOmitStackTraceInFastThrowOption(ClusterSpec.Type type) { return ""; }
-        @ModelFeatureFlag(owners = {"arnej"}) default boolean requireConnectivityCheck() { return true; }
         @ModelFeatureFlag(owners = {"hmusum"}) default double resourceLimitDisk() { return 0.8; }
         @ModelFeatureFlag(owners = {"hmusum"}) default double resourceLimitMemory() { return 0.8; }
         @ModelFeatureFlag(owners = {"geirst", "vekterli"}) default double minNodeRatioPerGroup() { return 0.0; }
-        @ModelFeatureFlag(owners = {"arnej"}) default boolean newLocationBrokerLogic() { return true; }
         @ModelFeatureFlag(owners = {"geirst", "vekterli"}) default int distributorMergeBusyWait() { return 1; }
         @ModelFeatureFlag(owners = {"vekterli", "geirst"}) default boolean distributorEnhancedMaintenanceScheduling() { return true; }
         @ModelFeatureFlag(owners = {"arnej"}) default boolean forwardIssuesAsErrors() { return true; }
@@ -111,9 +109,10 @@ public interface ModelContext {
         @ModelFeatureFlag(owners = {"arnej", "baldersheim"}) default boolean useV8DocManagerCfg() { return false; }
         @ModelFeatureFlag(owners = {"baldersheim", "geirst", "toregge"}) default int maxCompactBuffers() { return 1; }
         @ModelFeatureFlag(owners = {"hmusum"}) default boolean failDeploymentWithInvalidJvmOptions() { return false; }
-        @ModelFeatureFlag(owners = {"baldersheim"}) default double tlsSizeFraction() { throw new UnsupportedOperationException("TODO specify default value"); }
+        @ModelFeatureFlag(owners = {"baldersheim"}, removeAfter = "7.527") default double tlsSizeFraction() { return 0.02; }
         @ModelFeatureFlag(owners = {"arnej", "andreer"}) default List<String> ignoredHttpUserAgents() { return List.of(); }
         @ModelFeatureFlag(owners = {"bjorncs"}) default boolean enableServerOcspStapling() { return false; }
+        @ModelFeatureFlag(owners = {"vekterli"}) default String persistenceAsyncThrottling() {  throw new UnsupportedOperationException("TODO specify default value"); }
     }
 
     /** Warning: As elsewhere in this package, do not make backwards incompatible changes that will break old config models! */
diff --git a/config-model/src/main/java/com/yahoo/config/model/deploy/TestProperties.java b/config-model/src/main/java/com/yahoo/config/model/deploy/TestProperties.java
index 3a3df4224b1..c148bb0e6e4 100644
--- a/config-model/src/main/java/com/yahoo/config/model/deploy/TestProperties.java
+++ b/config-model/src/main/java/com/yahoo/config/model/deploy/TestProperties.java
@@ -67,16 +67,14 @@ public class TestProperties implements ModelContext.Properties, ModelContext.Fea
     private boolean containerDumpHeapOnShutdownTimeout = false;
     private double containerShutdownTimeout = 50.0;
     private int distributorMergeBusyWait = 1;
-    private int docstoreCompressionLevel = 9;
     private int maxUnCommittedMemory = 123456;
-    private double diskBloatFactor = 0.25;
     private boolean distributorEnhancedMaintenanceScheduling = true;
     private boolean asyncApplyBucketDiff = true;
     private boolean unorderedMergeChaining = true;
     private List<String> zoneDnsSuffixes = List.of();
     private int maxCompactBuffers = 1;
     private boolean failDeploymentWithInvalidJvmOptions = false;
-    private double tlsSizeFraction = 0.02;
+    private String persistenceAsyncThrottling = "UNLIMITED";
 
     @Override public ModelContext.FeatureFlags featureFlags() { return this; }
     @Override public boolean multitenant() { return multitenant; }
@@ -124,8 +122,6 @@ public class TestProperties implements ModelContext.Properties, ModelContext.Fea
     @Override public double containerShutdownTimeout() { return containerShutdownTimeout; }
     @Override public boolean containerDumpHeapOnShutdownTimeout() { return containerDumpHeapOnShutdownTimeout; }
     @Override public int distributorMergeBusyWait() { return distributorMergeBusyWait; }
-    @Override public double diskBloatFactor() { return diskBloatFactor; }
-    @Override public int docstoreCompressionLevel() { return docstoreCompressionLevel; }
     @Override public boolean distributorEnhancedMaintenanceScheduling() { return distributorEnhancedMaintenanceScheduling; }
     @Override public int maxUnCommittedMemory() { return maxUnCommittedMemory; }
     @Override public boolean asyncApplyBucketDiff() { return asyncApplyBucketDiff; }
@@ -133,23 +129,13 @@ public class TestProperties implements ModelContext.Properties, ModelContext.Fea
     @Override public List<String> zoneDnsSuffixes() { return zoneDnsSuffixes; }
     @Override public int maxCompactBuffers() { return maxCompactBuffers; }
     @Override public boolean failDeploymentWithInvalidJvmOptions() { return failDeploymentWithInvalidJvmOptions; }
-    @Override public double tlsSizeFraction() { return tlsSizeFraction; }
+    @Override public String persistenceAsyncThrottling() { return persistenceAsyncThrottling; }
 
     public TestProperties maxUnCommittedMemory(int maxUnCommittedMemory) {
         this.maxUnCommittedMemory = maxUnCommittedMemory;
         return this;
     }
 
-    public TestProperties docstoreCompressionLevel(int docstoreCompressionLevel) {
-        this.docstoreCompressionLevel = docstoreCompressionLevel;
-        return this;
-    }
-
-    public TestProperties diskBloatFactor(double diskBloatFactor) {
-        this.diskBloatFactor = diskBloatFactor;
-        return this;
-    }
-
     public TestProperties containerDumpHeapOnShutdownTimeout(boolean value) {
         containerDumpHeapOnShutdownTimeout = value;
         return this;
@@ -351,8 +337,8 @@ public class TestProperties implements ModelContext.Properties, ModelContext.Fea
         return this;
     }
 
-    public TestProperties tlsSizeFraction(double tlsSizeFraction) {
-        this.tlsSizeFraction = tlsSizeFraction;
+    public TestProperties setPersistenceAsyncThrottling(String type) {
+        this.persistenceAsyncThrottling = type;
         return this;
     }
 
diff --git a/config-model/src/main/java/com/yahoo/vespa/model/content/ContentSearchCluster.java b/config-model/src/main/java/com/yahoo/vespa/model/content/ContentSearchCluster.java
index 54d09bacfa9..d0cba617cfc 100644
--- a/config-model/src/main/java/com/yahoo/vespa/model/content/ContentSearchCluster.java
+++ b/config-model/src/main/java/com/yahoo/vespa/model/content/ContentSearchCluster.java
@@ -43,6 +43,9 @@ import static java.util.stream.Collectors.toList;
  */
 public class ContentSearchCluster extends AbstractConfigProducer<SearchCluster> implements ProtonConfig.Producer, DispatchConfig.Producer {
 
+    private static final int DEFAULT_DOC_STORE_COMPRESSION_LEVEL = 3;
+    private static final double DEFAULT_DISK_BLOAT = 0.25;
+
     private final boolean flushOnShutdown;
     private final Boolean syncTransactionLog;
 
@@ -68,8 +71,6 @@ public class ContentSearchCluster extends AbstractConfigProducer<SearchCluster>
     private final int feedMasterTaskLimit;
     private final ProtonConfig.Feeding.Shared_field_writer_executor.Enum sharedFieldWriterExecutor;
     private final double defaultFeedConcurrency;
-    private final double defaultDiskBloatFactor;
-    private final int defaultDocStoreCompressionLevel;
     private final boolean forwardIssuesToQrs;
     private final int defaultMaxCompactBuffers;
 
@@ -223,8 +224,6 @@ public class ContentSearchCluster extends AbstractConfigProducer<SearchCluster>
         this.feedMasterTaskLimit = featureFlags.feedMasterTaskLimit();
         this.sharedFieldWriterExecutor = convertSharedFieldWriterExecutor(featureFlags.sharedFieldWriterExecutor());
         this.defaultFeedConcurrency = featureFlags.feedConcurrency();
-        this.defaultDiskBloatFactor = featureFlags.diskBloatFactor();
-        this.defaultDocStoreCompressionLevel = featureFlags.docstoreCompressionLevel();
         this.forwardIssuesToQrs = featureFlags.forwardIssuesAsErrors();
         this.defaultMaxCompactBuffers = featureFlags.maxCompactBuffers();
     }
@@ -291,7 +290,7 @@ public class ContentSearchCluster extends AbstractConfigProducer<SearchCluster>
         if (element == null) {
             searchNode = SearchNode.create(parent, "" + node.getDistributionKey(), node.getDistributionKey(), spec,
                                            clusterName, node, flushOnShutdown, tuning, resourceLimits, parentGroup.isHosted(),
-                                           fractionOfMemoryReserved, deployState.featureFlags().tlsSizeFraction());
+                                           fractionOfMemoryReserved);
             searchNode.setHostResource(node.getHostResource());
             searchNode.initService(deployState.getDeployLogger());
 
@@ -423,10 +422,10 @@ public class ContentSearchCluster extends AbstractConfigProducer<SearchCluster>
         } else {
             builder.feeding.concurrency(defaultFeedConcurrency);
         }
-        builder.flush.memory.diskbloatfactor(defaultDiskBloatFactor);
-        builder.flush.memory.each.diskbloatfactor(defaultDiskBloatFactor);
-        builder.summary.log.chunk.compression.level(defaultDocStoreCompressionLevel);
-        builder.summary.log.compact.compression.level(defaultDocStoreCompressionLevel);
+        builder.flush.memory.diskbloatfactor(DEFAULT_DISK_BLOAT);
+        builder.flush.memory.each.diskbloatfactor(DEFAULT_DISK_BLOAT);
+        builder.summary.log.chunk.compression.level(DEFAULT_DOC_STORE_COMPRESSION_LEVEL);
+        builder.summary.log.compact.compression.level(DEFAULT_DOC_STORE_COMPRESSION_LEVEL);
         builder.forward_issues(forwardIssuesToQrs);
 
         int numDocumentDbs = builder.documentdb.size();
diff --git a/config-model/src/main/java/com/yahoo/vespa/model/content/storagecluster/FileStorProducer.java b/config-model/src/main/java/com/yahoo/vespa/model/content/storagecluster/FileStorProducer.java
index d7f6fb6c581..1f3a76b766e 100644
--- a/config-model/src/main/java/com/yahoo/vespa/model/content/storagecluster/FileStorProducer.java
+++ b/config-model/src/main/java/com/yahoo/vespa/model/content/storagecluster/FileStorProducer.java
@@ -46,6 +46,7 @@ public class FileStorProducer implements StorFilestorConfig.Producer {
     private final ContentCluster cluster;
     private final int reponseNumThreads;
     private final StorFilestorConfig.Response_sequencer_type.Enum responseSequencerType;
+    private final StorFilestorConfig.Async_operation_throttler_type.Enum asyncOperationThrottlerType;
     private final boolean useAsyncMessageHandlingOnSchedule;
     private final boolean asyncApplyBucketDiff;
 
@@ -57,11 +58,20 @@ public class FileStorProducer implements StorFilestorConfig.Producer {
         }
     }
 
+    private static StorFilestorConfig.Async_operation_throttler_type.Enum toAsyncOperationThrottlerType(String throttlerType) {
+        try {
+            return StorFilestorConfig.Async_operation_throttler_type.Enum.valueOf(throttlerType);
+        } catch (Throwable t) {
+            return StorFilestorConfig.Async_operation_throttler_type.UNLIMITED;
+        }
+    }
+
     public FileStorProducer(ModelContext.FeatureFlags featureFlags, ContentCluster parent, Integer numThreads) {
         this.numThreads = numThreads;
         this.cluster = parent;
         this.reponseNumThreads = featureFlags.defaultNumResponseThreads();
         this.responseSequencerType = convertResponseSequencerType(featureFlags.responseSequencerType());
+        this.asyncOperationThrottlerType = toAsyncOperationThrottlerType(featureFlags.persistenceAsyncThrottling());
         useAsyncMessageHandlingOnSchedule = featureFlags.useAsyncMessageHandlingOnSchedule();
         asyncApplyBucketDiff = featureFlags.asyncApplyBucketDiff();
     }
@@ -76,6 +86,7 @@ public class FileStorProducer implements StorFilestorConfig.Producer {
         builder.response_sequencer_type(responseSequencerType);
         builder.use_async_message_handling_on_schedule(useAsyncMessageHandlingOnSchedule);
         builder.async_apply_bucket_diff(asyncApplyBucketDiff);
+        builder.async_operation_throttler_type(asyncOperationThrottlerType);
     }
 
 }
diff --git a/config-model/src/main/java/com/yahoo/vespa/model/search/NodeResourcesTuning.java b/config-model/src/main/java/com/yahoo/vespa/model/search/NodeResourcesTuning.java
index e321892b5ad..9b9a525ab29 100644
--- a/config-model/src/main/java/com/yahoo/vespa/model/search/NodeResourcesTuning.java
+++ b/config-model/src/main/java/com/yahoo/vespa/model/search/NodeResourcesTuning.java
@@ -17,6 +17,7 @@ public class NodeResourcesTuning implements ProtonConfig.Producer {
     private final static double SUMMARY_FILE_SIZE_AS_FRACTION_OF_MEMORY = 0.02;
     private final static double SUMMARY_CACHE_SIZE_AS_FRACTION_OF_MEMORY = 0.04;
     private final static double MEMORY_GAIN_AS_FRACTION_OF_MEMORY = 0.08;
+    private final static double TLS_SIZE_FRACTION = 0.02;
     final static long MB = 1024 * 1024;
     public final static long GB = MB * 1024;
     // This is an approximate number base on observation of a node using 33G memory with 765M docs
@@ -24,19 +25,16 @@ public class NodeResourcesTuning implements ProtonConfig.Producer {
     private final NodeResources resources;
     private final int threadsPerSearch;
     private final double fractionOfMemoryReserved;
-    private final double tlsSizeFraction;
 
     // "Reserve" 0.5GB of memory for other processes running on the content node (config-proxy, metrics-proxy).
     public static final double reservedMemoryGb = 0.5;
 
     public NodeResourcesTuning(NodeResources resources,
                                int threadsPerSearch,
-                               double fractionOfMemoryReserved,
-                               double tlsSizeFraction) {
+                               double fractionOfMemoryReserved) {
         this.resources = resources;
         this.threadsPerSearch = threadsPerSearch;
         this.fractionOfMemoryReserved = fractionOfMemoryReserved;
-        this.tlsSizeFraction = tlsSizeFraction;
     }
 
     @Override
@@ -93,7 +91,7 @@ public class NodeResourcesTuning implements ProtonConfig.Producer {
     }
 
     private void tuneFlushStrategyTlsSize(ProtonConfig.Flush.Memory.Builder builder) {
-        long tlsSizeBytes = (long) ((resources.diskGb() * tlsSizeFraction) * GB);
+        long tlsSizeBytes = (long) ((resources.diskGb() * TLS_SIZE_FRACTION) * GB);
         tlsSizeBytes = max(2*GB, min(tlsSizeBytes, 100 * GB));
         builder.maxtlssize(tlsSizeBytes);
     }
diff --git a/config-model/src/main/java/com/yahoo/vespa/model/search/SearchNode.java b/config-model/src/main/java/com/yahoo/vespa/model/search/SearchNode.java
index 31513a273b2..28d1fbe72ef 100644
--- a/config-model/src/main/java/com/yahoo/vespa/model/search/SearchNode.java
+++ b/config-model/src/main/java/com/yahoo/vespa/model/search/SearchNode.java
@@ -60,7 +60,7 @@ public class SearchNode extends AbstractService implements
 
     private final boolean isHostedVespa;
     private final boolean flushOnShutdown;
-    private NodeSpec nodeSpec;
+    private final NodeSpec nodeSpec;
     private int distributionKey;
     private final String clusterName;
     private TransactionLogServer tls;
@@ -68,7 +68,6 @@ public class SearchNode extends AbstractService implements
     private final Optional<Tuning> tuning;
     private final Optional<ResourceLimits> resourceLimits;
     private final double fractionOfMemoryReserved;
-    private final double tlsSizeFraction;
 
     public static class Builder extends VespaDomBuilder.DomConfigProducerBuilder<SearchNode> {
 
@@ -97,8 +96,7 @@ public class SearchNode extends AbstractService implements
         @Override
         protected SearchNode doBuild(DeployState deployState, AbstractConfigProducer ancestor, Element producerSpec) {
             return new SearchNode(ancestor, name, contentNode.getDistributionKey(), nodeSpec, clusterName, contentNode,
-                                  flushOnShutdown, tuning, resourceLimits, deployState.isHosted(), fractionOfMemoryReserved,
-                                  deployState.featureFlags().tlsSizeFraction());
+                                  flushOnShutdown, tuning, resourceLimits, deployState.isHosted(), fractionOfMemoryReserved);
         }
 
     }
@@ -106,16 +104,16 @@ public class SearchNode extends AbstractService implements
     public static SearchNode create(AbstractConfigProducer parent, String name, int distributionKey, NodeSpec nodeSpec,
                                     String clusterName, AbstractService serviceLayerService, boolean flushOnShutdown,
                                     Optional<Tuning> tuning, Optional<ResourceLimits> resourceLimits, boolean isHostedVespa,
-                                    double fractionOfMemoryReserved, double tlsSizeFraction) {
+                                    double fractionOfMemoryReserved) {
         return new SearchNode(parent, name, distributionKey, nodeSpec, clusterName, serviceLayerService, flushOnShutdown,
-                              tuning, resourceLimits, isHostedVespa, fractionOfMemoryReserved, tlsSizeFraction);
+                              tuning, resourceLimits, isHostedVespa, fractionOfMemoryReserved);
     }
 
     private SearchNode(AbstractConfigProducer parent, String name, int distributionKey, NodeSpec nodeSpec,
                        String clusterName, AbstractService serviceLayerService, boolean flushOnShutdown,
                        Optional<Tuning> tuning, Optional<ResourceLimits> resourceLimits, boolean isHostedVespa,
-                       double fractionOfMemoryReserved, double tlsSizeFraction) {
-        this(parent, name, nodeSpec, clusterName, flushOnShutdown, tuning, resourceLimits, isHostedVespa, fractionOfMemoryReserved, tlsSizeFraction);
+                       double fractionOfMemoryReserved) {
+        this(parent, name, nodeSpec, clusterName, flushOnShutdown, tuning, resourceLimits, isHostedVespa, fractionOfMemoryReserved);
         this.distributionKey = distributionKey;
         this.serviceLayerService = serviceLayerService;
         setPropertiesElastic(clusterName, distributionKey);
@@ -123,12 +121,11 @@ public class SearchNode extends AbstractService implements
 
     private SearchNode(AbstractConfigProducer parent, String name, NodeSpec nodeSpec, String clusterName,
                        boolean flushOnShutdown, Optional<Tuning> tuning, Optional<ResourceLimits> resourceLimits, boolean isHostedVespa,
-                       double fractionOfMemoryReserved, double tlsSizeFraction) {
+                       double fractionOfMemoryReserved) {
         super(parent, name);
         setOmpNumThreads(1);
         this.isHostedVespa = isHostedVespa;
         this.fractionOfMemoryReserved = fractionOfMemoryReserved;
-        this.tlsSizeFraction = tlsSizeFraction;
         this.nodeSpec = nodeSpec;
         this.clusterName = clusterName;
         this.flushOnShutdown = flushOnShutdown;
@@ -282,7 +279,7 @@ public class SearchNode extends AbstractService implements
         if (nodeResources.isPresent()) {
             var nodeResourcesTuning = new NodeResourcesTuning(nodeResources.get(),
                                                               tuning.map(Tuning::threadsPerSearch).orElse(1),
-                                                              fractionOfMemoryReserved, tlsSizeFraction);
+                                                              fractionOfMemoryReserved);
             nodeResourcesTuning.getConfig(builder);
 
             tuning.ifPresent(t -> t.getConfig(builder));
diff --git a/config-model/src/test/java/com/yahoo/vespa/model/content/ContentClusterTest.java b/config-model/src/test/java/com/yahoo/vespa/model/content/ContentClusterTest.java
index 4591726d1f9..87a962339e9 100644
--- a/config-model/src/test/java/com/yahoo/vespa/model/content/ContentClusterTest.java
+++ b/config-model/src/test/java/com/yahoo/vespa/model/content/ContentClusterTest.java
@@ -45,7 +45,6 @@ import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
-import java.util.OptionalDouble;
 import java.util.OptionalInt;
 
 import static org.junit.Assert.assertEquals;
@@ -1046,11 +1045,9 @@ public class ContentClusterTest extends ContentBaseTest {
         assertEquals(7, resolveMaxCompactBuffers(OptionalInt.of(7)));
     }
 
-    private long resolveMaxTLSSize(OptionalDouble tlsSizeFraction, Optional<Flavor> flavor) throws Exception {
+    private long resolveMaxTLSSize(Optional<Flavor> flavor) throws Exception {
         TestProperties testProperties = new TestProperties();
-        if (tlsSizeFraction.isPresent()) {
-            testProperties.tlsSizeFraction(tlsSizeFraction.getAsDouble());
-        }
+
         ContentCluster cc = createOneNodeCluster(testProperties, flavor);
         ProtonConfig.Builder protonBuilder = new ProtonConfig.Builder();
         cc.getSearch().getSearchNodes().get(0).getConfig(protonBuilder);
@@ -1058,13 +1055,10 @@ public class ContentClusterTest extends ContentBaseTest {
         return protonConfig.flush().memory().maxtlssize();
     }
     @Test
-    public void default_max_tls_size_controlled_by_properties() throws Exception {
+    public void verifyt_max_tls_size() throws Exception {
         var flavor = new Flavor(new FlavorsConfig.Flavor(new FlavorsConfig.Flavor.Builder().name("test").minDiskAvailableGb(100)));
-        assertEquals(21474836480L, resolveMaxTLSSize(OptionalDouble.empty(), Optional.empty()));
-        assertEquals(21474836480L, resolveMaxTLSSize(OptionalDouble.of(0.02), Optional.empty()));
-        assertEquals(2147483648L, resolveMaxTLSSize(OptionalDouble.empty(), Optional.of(flavor)));
-        assertEquals(2147483648L, resolveMaxTLSSize(OptionalDouble.of(0.02), Optional.of(flavor)));
-        assertEquals(3221225472L, resolveMaxTLSSize(OptionalDouble.of(0.03), Optional.of(flavor)));
+        assertEquals(21474836480L, resolveMaxTLSSize(Optional.empty()));
+        assertEquals(2147483648L, resolveMaxTLSSize(Optional.of(flavor)));
     }
 
     void assertZookeeperServerImplementation(String expectedClassName,
diff --git a/config-model/src/test/java/com/yahoo/vespa/model/content/ContentSchemaClusterTest.java b/config-model/src/test/java/com/yahoo/vespa/model/content/ContentSchemaClusterTest.java
index 51badae2746..68e722f45d3 100644
--- a/config-model/src/test/java/com/yahoo/vespa/model/content/ContentSchemaClusterTest.java
+++ b/config-model/src/test/java/com/yahoo/vespa/model/content/ContentSchemaClusterTest.java
@@ -33,7 +33,7 @@ import static org.junit.Assert.assertTrue;
  */
 public class ContentSchemaClusterTest {
 
-    private static double EPSILON = 0.000001;
+    private static final double EPSILON = 0.000001;
 
     private static ContentCluster createClusterWithOneDocumentType() throws Exception {
         return createCluster(new ContentClusterBuilder().getXml());
@@ -261,27 +261,16 @@ public class ContentSchemaClusterTest {
     }
 
     @Test
-    public void verifyControlOfDocStoreCompression() throws Exception {
+    public void verifyDefaultDocStoreCompression() throws Exception {
         ProtonConfig cfg = getProtonConfig(createCluster(new ContentClusterBuilder().getXml()));
-        assertEquals(9, cfg.summary().log().chunk().compression().level());
-        assertEquals(9, cfg.summary().log().compact().compression().level());
-
-        cfg = getProtonConfig(createCluster(new ContentClusterBuilder().getXml(),
-                new DeployState.Builder().properties(new TestProperties().docstoreCompressionLevel(3))));
         assertEquals(3, cfg.summary().log().chunk().compression().level());
         assertEquals(3, cfg.summary().log().compact().compression().level());
     }
 
     @Test
-    public void verifyControlOfDiskBloatFactor() throws Exception {
+    public void verifyDefaultDiskBloatFactor() throws Exception {
         var defaultCfg = getProtonConfig(createCluster(new ContentClusterBuilder().getXml()));
         assertEquals(0.25, defaultCfg.flush().memory().diskbloatfactor(), EPSILON);
         assertEquals(0.25, defaultCfg.flush().memory().each().diskbloatfactor(), EPSILON);
-
-        var controlledCfg = getProtonConfig(createCluster(new ContentClusterBuilder().getXml(),
-                new DeployState.Builder().properties(new TestProperties().diskBloatFactor(0.31))
-        ));
-        assertEquals(0.31, controlledCfg.flush().memory().diskbloatfactor(), EPSILON);
-        assertEquals(0.31, controlledCfg.flush().memory().each().diskbloatfactor(), EPSILON);
     }
 }
diff --git a/config-model/src/test/java/com/yahoo/vespa/model/content/StorageClusterTest.java b/config-model/src/test/java/com/yahoo/vespa/model/content/StorageClusterTest.java
index c521d24d22f..cf877d3bf88 100644
--- a/config-model/src/test/java/com/yahoo/vespa/model/content/StorageClusterTest.java
+++ b/config-model/src/test/java/com/yahoo/vespa/model/content/StorageClusterTest.java
@@ -306,6 +306,25 @@ public class StorageClusterTest {
     }
 
     @Test
+    public void persistence_async_throttle_config_defaults_to_unlimited() {
+        var config = filestorConfigFromProducer(simpleCluster(new TestProperties()));
+        assertEquals(StorFilestorConfig.Async_operation_throttler_type.UNLIMITED, config.async_operation_throttler_type());
+    }
+
+    @Test
+    public void persistence_async_throttle_config_is_derived_from_flag() {
+        var config = filestorConfigFromProducer(simpleCluster(new TestProperties().setPersistenceAsyncThrottling("UNLIMITED")));
+        assertEquals(StorFilestorConfig.Async_operation_throttler_type.UNLIMITED, config.async_operation_throttler_type());
+
+        config = filestorConfigFromProducer(simpleCluster(new TestProperties().setPersistenceAsyncThrottling("DYNAMIC")));
+        assertEquals(StorFilestorConfig.Async_operation_throttler_type.DYNAMIC, config.async_operation_throttler_type());
+
+        // Invalid enum values fall back to the default
+        config = filestorConfigFromProducer(simpleCluster(new TestProperties().setPersistenceAsyncThrottling("BANANAS")));
+        assertEquals(StorFilestorConfig.Async_operation_throttler_type.UNLIMITED, config.async_operation_throttler_type());
+    }
+
+    @Test
     public void integrity_checker_explicitly_disabled_when_not_running_with_vds_provider() {
         StorIntegritycheckerConfig.Builder builder = new StorIntegritycheckerConfig.Builder();
         parse(cluster("bees", "")).getConfig(builder);
diff --git a/config-model/src/test/java/com/yahoo/vespa/model/search/NodeResourcesTuningTest.java b/config-model/src/test/java/com/yahoo/vespa/model/search/NodeResourcesTuningTest.java
index 18ad9a4def9..5571ead11ce 100644
--- a/config-model/src/test/java/com/yahoo/vespa/model/search/NodeResourcesTuningTest.java
+++ b/config-model/src/test/java/com/yahoo/vespa/model/search/NodeResourcesTuningTest.java
@@ -42,7 +42,7 @@ public class NodeResourcesTuningTest {
         assertEquals(0.5, reservedMemoryGb, delta);
     }
 
-    private ProtonConfig getProtonMemoryConfig(List<Pair<String, String>> sdAndMode, double gb, int redundancy, int searchableCopies) {
+    private ProtonConfig getProtonMemoryConfig(List<Pair<String, String>> sdAndMode, double gb) {
         ProtonConfig.Builder builder = new ProtonConfig.Builder();
         for (Pair<String, String> sdMode : sdAndMode) {
             builder.documentdb.add(new ProtonConfig.Documentdb.Builder()
@@ -53,8 +53,8 @@ public class NodeResourcesTuningTest {
         return configFromMemorySetting(gb, builder);
     }
 
-    private void verify_that_initial_numdocs_is_dependent_of_mode(int redundancy, int searchablecopies) {
-        ProtonConfig cfg = getProtonMemoryConfig(Arrays.asList(new Pair<>("a", "INDEX"), new Pair<>("b", "STREAMING"), new Pair<>("c", "STORE_ONLY")), 24 + reservedMemoryGb, redundancy, searchablecopies);
+    private void verify_that_initial_numdocs_is_dependent_of_mode() {
+        ProtonConfig cfg = getProtonMemoryConfig(Arrays.asList(new Pair<>("a", "INDEX"), new Pair<>("b", "STREAMING"), new Pair<>("c", "STORE_ONLY")), 24 + reservedMemoryGb);
         assertEquals(3, cfg.documentdb().size());
         assertEquals(1024, cfg.documentdb(0).allocation().initialnumdocs());
         assertEquals("a", cfg.documentdb(0).inputdoctypename());
@@ -66,10 +66,8 @@ public class NodeResourcesTuningTest {
 
     @Test
     public void require_that_initial_numdocs_is_dependent_of_mode_and_searchablecopies() {
-        verify_that_initial_numdocs_is_dependent_of_mode(2,0);
-        verify_that_initial_numdocs_is_dependent_of_mode(1,1);
-        verify_that_initial_numdocs_is_dependent_of_mode(3, 2);
-        verify_that_initial_numdocs_is_dependent_of_mode(3, 3);
+        verify_that_initial_numdocs_is_dependent_of_mode();
+
     }
 
     @Test
@@ -134,13 +132,11 @@ public class NodeResourcesTuningTest {
 
     @Test
     public void require_that_flush_strategy_tls_size_is_set_based_on_available_disk() {
-        assertFlushStrategyTlsSize(2 * GB, 10, 0.05);
-        assertFlushStrategyTlsSize(7 * GB, 100, 0.07);
-        assertFlushStrategyTlsSize(5 * GB, 100, 0.05);
-        assertFlushStrategyTlsSize(35 * GB, 500, 0.07);
-        assertFlushStrategyTlsSize(84 * GB, 1200, 0.07);
-        assertFlushStrategyTlsSize(100 * GB, 1720, 0.07);
-        assertFlushStrategyTlsSize(100 * GB, 24000, 0.07);
+        assertFlushStrategyTlsSize(2 * GB, 10);
+        assertFlushStrategyTlsSize(2 * GB, 100);
+        assertFlushStrategyTlsSize(10 * GB, 500);
+        assertFlushStrategyTlsSize(24 * GB, 1200);
+        assertFlushStrategyTlsSize(100 * GB, 24000);
     }
 
     @Test
@@ -181,8 +177,8 @@ public class NodeResourcesTuningTest {
         assertEquals(expMemoryBytes, configFromMemorySetting(wantedMemoryGb + reservedMemoryGb, 0).flush().memory().each().maxmemory());
     }
 
-    private static void assertFlushStrategyTlsSize(long expTlsSizeBytes, int diskGb, double tlsSizeFraction) {
-        assertEquals(expTlsSizeBytes, configFromDiskSetting(diskGb, tlsSizeFraction).flush().memory().maxtlssize());
+    private static void assertFlushStrategyTlsSize(long expTlsSizeBytes, int diskGb) {
+        assertEquals(expTlsSizeBytes, configFromDiskSetting(diskGb).flush().memory().maxtlssize());
     }
 
     private static void assertSummaryReadIo(ProtonConfig.Summary.Read.Io.Enum expValue, boolean fastDisk) {
@@ -197,23 +193,16 @@ public class NodeResourcesTuningTest {
         assertEquals(sharedDisk, configFromEnvironmentType(docker).hwinfo().disk().shared());
     }
 
-    private static void assertWriteFilter(double expMemoryLimit, int memoryGb) {
-        assertEquals(expMemoryLimit, configFromMemorySetting(memoryGb, 0).writefilter().memorylimit(), delta);
-    }
-
     private static ProtonConfig configFromDiskSetting(boolean fastDisk) {
         return getConfig(new FlavorsConfig.Flavor.Builder().fastDisk(fastDisk));
     }
 
     private static ProtonConfig configFromDiskSetting(int diskGb) {
-        return configFromDiskSetting(diskGb, 0.07);
-    }
-    private static ProtonConfig configFromDiskSetting(int diskGb, double tlsSizeFraction) {
-        return getConfig(new FlavorsConfig.Flavor.Builder().minDiskAvailableGb(diskGb), 0, tlsSizeFraction);
+        return getConfig(new FlavorsConfig.Flavor.Builder().minDiskAvailableGb(diskGb), 0);
     }
 
     private static ProtonConfig configFromMemorySetting(double memoryGb, double fractionOfMemoryReserved) {
-        return getConfig(new FlavorsConfig.Flavor.Builder().minMainMemoryAvailableGb(memoryGb), fractionOfMemoryReserved, 0.07);
+        return getConfig(new FlavorsConfig.Flavor.Builder().minMainMemoryAvailableGb(memoryGb), fractionOfMemoryReserved);
     }
 
     private static ProtonConfig configFromMemorySetting(double memoryGb, ProtonConfig.Builder builder) {
@@ -239,26 +228,26 @@ public class NodeResourcesTuningTest {
         return getConfig(flavorBuilder, new ProtonConfig.Builder());
     }
 
-    private static ProtonConfig getConfig(FlavorsConfig.Flavor.Builder flavorBuilder, double fractionOfMemoryReserved, double tlsSizeFraction) {
-        return getConfig(flavorBuilder, new ProtonConfig.Builder(), fractionOfMemoryReserved, tlsSizeFraction);
+    private static ProtonConfig getConfig(FlavorsConfig.Flavor.Builder flavorBuilder, double fractionOfMemoryReserved) {
+        return getConfig(flavorBuilder, new ProtonConfig.Builder(), fractionOfMemoryReserved);
     }
 
     private static ProtonConfig getConfig(FlavorsConfig.Flavor.Builder flavorBuilder, ProtonConfig.Builder protonBuilder) {
         return getConfig(flavorBuilder, protonBuilder,1);
     }
-    private static ProtonConfig getConfig(FlavorsConfig.Flavor.Builder flavorBuilder, ProtonConfig.Builder protonBuilder, double fractionOfMemoryReserved, double tlsSizeFraction) {
-        return getConfig(flavorBuilder, protonBuilder, 1, fractionOfMemoryReserved, tlsSizeFraction);
+    private static ProtonConfig getConfig(FlavorsConfig.Flavor.Builder flavorBuilder, ProtonConfig.Builder protonBuilder, double fractionOfMemoryReserved) {
+        return getConfig(flavorBuilder, protonBuilder, 1, fractionOfMemoryReserved);
     }
 
     private static ProtonConfig getConfig(FlavorsConfig.Flavor.Builder flavorBuilder, ProtonConfig.Builder protonBuilder,
                                           int numThreadsPerSearch) {
-        return getConfig(flavorBuilder, protonBuilder, numThreadsPerSearch, 0, 0.07);
+        return getConfig(flavorBuilder, protonBuilder, numThreadsPerSearch, 0);
     }
 
     private static ProtonConfig getConfig(FlavorsConfig.Flavor.Builder flavorBuilder, ProtonConfig.Builder protonBuilder,
-                                          int numThreadsPerSearch, double fractionOfMemoryReserved, double tlsSizeFraction) {
+                                          int numThreadsPerSearch, double fractionOfMemoryReserved) {
         flavorBuilder.name("my_flavor");
-        NodeResourcesTuning tuning = new NodeResourcesTuning(new Flavor(new FlavorsConfig.Flavor(flavorBuilder)).resources(), numThreadsPerSearch, fractionOfMemoryReserved, tlsSizeFraction);
+        NodeResourcesTuning tuning = new NodeResourcesTuning(new Flavor(new FlavorsConfig.Flavor(flavorBuilder)).resources(), numThreadsPerSearch, fractionOfMemoryReserved);
         tuning.getConfig(protonBuilder);
         return new ProtonConfig(protonBuilder);
     }
diff --git a/config-model/src/test/java/com/yahoo/vespa/model/search/test/SchemaNodeTest.java b/config-model/src/test/java/com/yahoo/vespa/model/search/test/SchemaNodeTest.java
index d000f83d6cd..226045f4d8a 100644
--- a/config-model/src/test/java/com/yahoo/vespa/model/search/test/SchemaNodeTest.java
+++ b/config-model/src/test/java/com/yahoo/vespa/model/search/test/SchemaNodeTest.java
@@ -51,8 +51,7 @@ public class SchemaNodeTest {
     private static SearchNode createSearchNode(MockRoot root, String name, int distributionKey,
                                                NodeSpec nodeSpec, boolean flushOnShutDown, boolean isHosted) {
         return SearchNode.create(root, name, distributionKey, nodeSpec, "mycluster", null, flushOnShutDown,
-                Optional.empty(), Optional.empty(), isHosted, 0.0,
-                root.getDeployState().featureFlags().tlsSizeFraction());
+                Optional.empty(), Optional.empty(), isHosted, 0.0);
     }
 
     private static SearchNode createSearchNode(MockRoot root) {
diff --git a/config-provisioning/src/main/resources/configdefinitions/config.provisioning.node-repository.def b/config-provisioning/src/main/resources/configdefinitions/config.provisioning.node-repository.def
index 6181efc7184..b054f434322 100644
--- a/config-provisioning/src/main/resources/configdefinitions/config.provisioning.node-repository.def
+++ b/config-provisioning/src/main/resources/configdefinitions/config.provisioning.node-repository.def
@@ -11,4 +11,4 @@ tenantContainerImage string default=""
 useCuratorClientCache bool default=false
 
 # The number of Node objects to cache in-memory.
-nodeCacheSize long default=2000
+nodeCacheSize long default=3000
diff --git a/configserver/src/main/java/com/yahoo/vespa/config/server/deploy/ModelContextImpl.java b/configserver/src/main/java/com/yahoo/vespa/config/server/deploy/ModelContextImpl.java
index 7d4ac948194..5d944df2f30 100644
--- a/configserver/src/main/java/com/yahoo/vespa/config/server/deploy/ModelContextImpl.java
+++ b/configserver/src/main/java/com/yahoo/vespa/config/server/deploy/ModelContextImpl.java
@@ -191,8 +191,6 @@ public class ModelContextImpl implements ModelContext {
         private final boolean containerDumpHeapOnShutdownTimeout;
         private final double containerShutdownTimeout;
         private final int distributorMergeBusyWait;
-        private final int docstoreCompressionLevel;
-        private final double diskBloatFactor;
         private final boolean distributorEnhancedMaintenanceScheduling;
         private final int maxUnCommittedMemory;
         private final boolean forwardIssuesAsErrors;
@@ -203,9 +201,9 @@ public class ModelContextImpl implements ModelContext {
         private final boolean useV8DocManagerCfg;
         private final int maxCompactBuffers;
         private final boolean failDeploymentWithInvalidJvmOptions;
-        private final double tlsSizeFraction;
         private final List<String> ignoredHttpUserAgents;
         private final boolean enableServerOcspStapling;
+        private final String persistenceAsyncThrottling;
 
         public FeatureFlags(FlagSource source, ApplicationId appId) {
             this.defaultTermwiseLimit = flagValue(source, appId, Flags.DEFAULT_TERM_WISE_LIMIT);
@@ -235,8 +233,6 @@ public class ModelContextImpl implements ModelContext {
             this.containerDumpHeapOnShutdownTimeout = flagValue(source, appId, Flags.CONTAINER_DUMP_HEAP_ON_SHUTDOWN_TIMEOUT);
             this.containerShutdownTimeout = flagValue(source, appId,Flags.CONTAINER_SHUTDOWN_TIMEOUT);
             this.distributorMergeBusyWait = flagValue(source, appId, Flags.DISTRIBUTOR_MERGE_BUSY_WAIT);
-            this.docstoreCompressionLevel = flagValue(source, appId, Flags.DOCSTORE_COMPRESSION_LEVEL);
-            this.diskBloatFactor = flagValue(source, appId, Flags.DISK_BLOAT_FACTOR);
             this.distributorEnhancedMaintenanceScheduling = flagValue(source, appId, Flags.DISTRIBUTOR_ENHANCED_MAINTENANCE_SCHEDULING);
             this.maxUnCommittedMemory = flagValue(source, appId, Flags.MAX_UNCOMMITTED_MEMORY);;
             this.forwardIssuesAsErrors = flagValue(source, appId, PermanentFlags.FORWARD_ISSUES_AS_ERRORS);
@@ -247,9 +243,9 @@ public class ModelContextImpl implements ModelContext {
             this.useV8DocManagerCfg = flagValue(source, appId, Flags.USE_V8_DOC_MANAGER_CFG);
             this.maxCompactBuffers = flagValue(source, appId, Flags.MAX_COMPACT_BUFFERS);
             this.failDeploymentWithInvalidJvmOptions = flagValue(source, appId, Flags.FAIL_DEPLOYMENT_WITH_INVALID_JVM_OPTIONS);
-            this.tlsSizeFraction = flagValue(source, appId, Flags.TLS_SIZE_FRACTION);
             this.ignoredHttpUserAgents = flagValue(source, appId, PermanentFlags.IGNORED_HTTP_USER_AGENTS);
             this.enableServerOcspStapling = flagValue(source, appId, Flags.ENABLE_SERVER_OCSP_STAPLING);
+            this.persistenceAsyncThrottling = flagValue(source, appId, Flags.PERSISTENCE_ASYNC_THROTTLING);
         }
 
         @Override public double defaultTermwiseLimit() { return defaultTermwiseLimit; }
@@ -281,8 +277,6 @@ public class ModelContextImpl implements ModelContext {
         @Override public double containerShutdownTimeout() { return containerShutdownTimeout; }
         @Override public boolean containerDumpHeapOnShutdownTimeout() { return containerDumpHeapOnShutdownTimeout; }
         @Override public int distributorMergeBusyWait() { return distributorMergeBusyWait; }
-        @Override public double diskBloatFactor() { return diskBloatFactor; }
-        @Override public int docstoreCompressionLevel() { return docstoreCompressionLevel; }
         @Override public boolean distributorEnhancedMaintenanceScheduling() { return distributorEnhancedMaintenanceScheduling; }
         @Override public int maxUnCommittedMemory() { return maxUnCommittedMemory; }
         @Override public boolean forwardIssuesAsErrors() { return forwardIssuesAsErrors; }
@@ -293,9 +287,9 @@ public class ModelContextImpl implements ModelContext {
         @Override public boolean useV8DocManagerCfg() { return useV8DocManagerCfg; }
         @Override public boolean failDeploymentWithInvalidJvmOptions() { return failDeploymentWithInvalidJvmOptions; }
         @Override public int maxCompactBuffers() { return maxCompactBuffers; }
-        @Override public double tlsSizeFraction() { return tlsSizeFraction; }
         @Override public List<String> ignoredHttpUserAgents() { return ignoredHttpUserAgents; }
         @Override public boolean enableServerOcspStapling() { return enableServerOcspStapling; }
+        @Override public String persistenceAsyncThrottling() { return persistenceAsyncThrottling; }
 
         private static <V> V flagValue(FlagSource source, ApplicationId appId, UnboundFlag<? extends V, ?, ?> flag) {
             return flag.bindTo(source)
diff --git a/configserver/src/main/java/com/yahoo/vespa/config/server/filedistribution/FileServer.java b/configserver/src/main/java/com/yahoo/vespa/config/server/filedistribution/FileServer.java
index 81cd1dd9738..8e5eee2104c 100644
--- a/configserver/src/main/java/com/yahoo/vespa/config/server/filedistribution/FileServer.java
+++ b/configserver/src/main/java/com/yahoo/vespa/config/server/filedistribution/FileServer.java
@@ -27,6 +27,7 @@ import java.io.File;
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.time.Duration;
 import java.util.List;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -217,7 +218,8 @@ public class FileServer {
         return new FileDownloader(configServers.isEmpty()
                                           ? FileDownloader.emptyConnectionPool()
                                           : createConnectionPool(configServers, supervisor),
-                                  supervisor);
+                                  supervisor,
+                                  Duration.ofSeconds(10)); // set this low, to make sure we don't wait a for a long time in this thread
     }
 
     private static ConnectionPool createConnectionPool(List<String> configServers, Supervisor supervisor) {
diff --git a/filedistribution/src/main/java/com/yahoo/vespa/filedistribution/FileDownloader.java b/filedistribution/src/main/java/com/yahoo/vespa/filedistribution/FileDownloader.java
index 1821c8971e7..5941ed536a8 100644
--- a/filedistribution/src/main/java/com/yahoo/vespa/filedistribution/FileDownloader.java
+++ b/filedistribution/src/main/java/com/yahoo/vespa/filedistribution/FileDownloader.java
@@ -27,7 +27,6 @@ import java.util.logging.Logger;
 public class FileDownloader implements AutoCloseable {
 
     private static final Logger log = Logger.getLogger(FileDownloader.class.getName());
-    private static final Duration defaultTimeout = Duration.ofMinutes(3);
     private static final Duration defaultSleepBetweenRetries = Duration.ofSeconds(5);
     public static final File defaultDownloadDirectory = new File(Defaults.getDefaults().underVespaHome("var/db/vespa/filedistribution"));
 
@@ -38,10 +37,6 @@ public class FileDownloader implements AutoCloseable {
     private final FileReferenceDownloader fileReferenceDownloader;
     private final Downloads downloads = new Downloads();
 
-    public FileDownloader(ConnectionPool connectionPool, Supervisor supervisor) {
-        this(connectionPool, supervisor, defaultDownloadDirectory, defaultTimeout, defaultSleepBetweenRetries);
-    }
-
     public FileDownloader(ConnectionPool connectionPool, Supervisor supervisor, Duration timeout) {
         this(connectionPool, supervisor, defaultDownloadDirectory, timeout, defaultSleepBetweenRetries);
     }
diff --git a/flags/src/main/java/com/yahoo/vespa/flags/Flags.java b/flags/src/main/java/com/yahoo/vespa/flags/Flags.java
index 2a36c62239a..065636604d9 100644
--- a/flags/src/main/java/com/yahoo/vespa/flags/Flags.java
+++ b/flags/src/main/java/com/yahoo/vespa/flags/Flags.java
@@ -52,13 +52,6 @@ public class Flags {
             "Takes effect at redeployment",
             ZONE_ID, APPLICATION_ID);
 
-    public static final UnboundDoubleFlag TLS_SIZE_FRACTION = defineDoubleFlag(
-            "tls-size-fraction", 0.02,
-            List.of("baldersheim"), "2021-12-20", "2022-02-01",
-            "Fraction of disk available for transaction log",
-            "Takes effect at redeployment",
-            ZONE_ID, APPLICATION_ID);
-
     public static final UnboundStringFlag FEED_SEQUENCER_TYPE = defineStringFlag(
             "feed-sequencer-type", "THROUGHPUT",
             List.of("baldersheim"), "2020-12-02", "2022-02-01",
@@ -159,20 +152,6 @@ public class Flags {
             "Takes effect at redeployment",
             ZONE_ID, APPLICATION_ID);
 
-    public static final UnboundDoubleFlag DISK_BLOAT_FACTOR = defineDoubleFlag(
-            "disk-bloat-factor", 0.25,
-            List.of("baldersheim"), "2021-10-08", "2022-02-01",
-            "Amount of bloat allowed before compacting file",
-            "Takes effect at redeployment",
-            ZONE_ID, APPLICATION_ID);
-
-    public static final UnboundIntFlag DOCSTORE_COMPRESSION_LEVEL = defineIntFlag(
-            "docstore-compression-level", 3,
-            List.of("baldersheim"), "2021-10-08", "2022-02-01",
-            "Default compression level used for document store",
-            "Takes effect at redeployment",
-            ZONE_ID, APPLICATION_ID);
-
     public static final UnboundBooleanFlag ENABLE_FEED_BLOCK_IN_DISTRIBUTOR = defineFeatureFlag(
             "enable-feed-block-in-distributor", true,
             List.of("geirst"), "2021-01-27", "2022-01-31",
@@ -408,6 +387,14 @@ public class Flags {
             "Takes effect on Docker container restart",
             ZONE_ID, APPLICATION_ID, NODE_TYPE);
 
+    public static final UnboundStringFlag PERSISTENCE_ASYNC_THROTTLING = defineStringFlag(
+            "persistence-async-throttling", "UNLIMITED",
+            List.of("vekterli"), "2022-01-12", "2022-05-01",
+            "Sets the throttling policy used for async persistence operations on the content nodes. " +
+            "Valid values: UNLIMITED, DYNAMIC",
+            "Triggers restart, takes effect immediately",
+            ZONE_ID, APPLICATION_ID);
+
     /** WARNING: public for testing: All flags should be defined in {@link Flags}. */
     public static UnboundBooleanFlag defineFeatureFlag(String flagId, boolean defaultValue, List<String> owners,
                                                        String createdAt, String expiresAt, String description,
diff --git a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/autoscale/ClusterModel.java b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/autoscale/ClusterModel.java
index 3c26eef41d9..b7a5c1e7fe7 100644
--- a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/autoscale/ClusterModel.java
+++ b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/autoscale/ClusterModel.java
@@ -25,7 +25,7 @@ public class ClusterModel {
     private static final Logger log = Logger.getLogger(ClusterModel.class.getName());
 
     /** Containers typically use more cpu right after generation change, so discard those metrics */
-    public static final Duration warmupDuration = Duration.ofSeconds(90);
+    public static final Duration warmupDuration = Duration.ofMinutes(5);
 
     private static final Duration currentLoadDuration = Duration.ofMinutes(5);
 
diff --git a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/autoscale/ClusterNodesTimeseries.java b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/autoscale/ClusterNodesTimeseries.java
index dd35eb7b54d..5ad4ef2e263 100644
--- a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/autoscale/ClusterNodesTimeseries.java
+++ b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/autoscale/ClusterNodesTimeseries.java
@@ -31,7 +31,7 @@ public class ClusterNodesTimeseries {
         // If none can be detected we assume the node is new/was down.
         // If either this is the case, or there is a generation change, we ignore
         // the first warmupWindow metrics
-        var timeseries = db.getNodeTimeseries(period.plus(warmupDuration.multipliedBy(8)), clusterNodes);
+        var timeseries = db.getNodeTimeseries(period.plus(warmupDuration.multipliedBy(4)), clusterNodes);
         if (cluster.lastScalingEvent().isPresent()) {
             long currentGeneration = cluster.lastScalingEvent().get().generation();
             timeseries = keepCurrentGenerationAfterWarmup(timeseries, currentGeneration);
diff --git a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/autoscale/NodeTimeseries.java b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/autoscale/NodeTimeseries.java
index c58b08cb3b5..4a5f8972e11 100644
--- a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/autoscale/NodeTimeseries.java
+++ b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/autoscale/NodeTimeseries.java
@@ -85,7 +85,7 @@ public class NodeTimeseries {
         if (snapshot.generation() < 0) return true; // Content nodes do not yet send generation
         if (snapshot.generation() < currentGeneration) return false;
         if (generationChange.isEmpty()) return true;
-        return ! snapshot.at().isBefore(generationChange.get().plus(warmupDuration.multipliedBy(2)));
+        return ! snapshot.at().isBefore(generationChange.get().plus(warmupDuration));
     }
 
 }
diff --git a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/node/Nodes.java b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/node/Nodes.java
index 480fd72967e..7f57ec219ae 100644
--- a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/node/Nodes.java
+++ b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/node/Nodes.java
@@ -16,7 +16,7 @@ import com.yahoo.vespa.hosted.provision.Node;
 import com.yahoo.vespa.hosted.provision.NodeList;
 import com.yahoo.vespa.hosted.provision.NodeMutex;
 import com.yahoo.vespa.hosted.provision.maintenance.NodeFailer;
-import com.yahoo.vespa.hosted.provision.node.filter.StateFilter;
+import com.yahoo.vespa.hosted.provision.node.filter.NodeFilter;
 import com.yahoo.vespa.hosted.provision.persistence.CuratorDatabaseClient;
 
 import java.time.Clock;
@@ -591,7 +591,7 @@ public class Nodes {
      * @return the nodes in their new state
      */
     public List<Node> restartActive(Predicate<Node> filter) {
-        return restart(StateFilter.from(Node.State.active).and(filter));
+        return restart(NodeFilter.in(Set.of(Node.State.active)).and(filter));
     }
 
     /**
diff --git a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/node/filter/NodeFilter.java b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/node/filter/NodeFilter.java
new file mode 100644
index 00000000000..a65ec30264f
--- /dev/null
+++ b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/node/filter/NodeFilter.java
@@ -0,0 +1,63 @@
+// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+package com.yahoo.vespa.hosted.provision.node.filter;
+
+
+import com.yahoo.text.StringUtilities;
+import com.yahoo.vespa.hosted.provision.Node;
+
+import java.util.EnumSet;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+
+/**
+ * A filter for nodes, matching by state. This should be the top-most filter so that the node-repository can determine
+ * which node states to read before testing additional filters.
+ *
+ * @author mpolden
+ */
+public class NodeFilter implements Predicate<Node> {
+
+    private final Set<Node.State> states;
+    private final Predicate<Node> filter;
+
+    private NodeFilter(Set<Node.State> states, Predicate<Node> filter) {
+        this.states = Objects.requireNonNull(states);
+        this.filter = Objects.requireNonNull(filter);
+    }
+
+    @Override
+    public boolean test(Node node) {
+        return states.contains(node.state()) && filter.test(node);
+    }
+
+    /** The node states to match */
+    public Set<Node.State> states() {
+        return states;
+    }
+
+    /** Returns a copy of this that matches with given filter */
+    public NodeFilter matching(Predicate<Node> filter) {
+        return new NodeFilter(states, filter);
+    }
+
+    /** Returns a node filter which matches a comma or space-separated list of states */
+    public static NodeFilter in(String states, boolean includeDeprovisioned) {
+        if (states == null) {
+            return NodeFilter.in(includeDeprovisioned
+                                         ? EnumSet.allOf(Node.State.class)
+                                         : EnumSet.complementOf(EnumSet.of(Node.State.deprovisioned)));
+        }
+        return NodeFilter.in(StringUtilities.split(states).stream()
+                                            .map(Node.State::valueOf)
+                                            .collect(Collectors.toSet()));
+    }
+
+    /** Returns a node filter matching given states */
+    public static NodeFilter in(Set<Node.State> states) {
+        return new NodeFilter(states, (ignored) -> true);
+    }
+
+
+}
diff --git a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/node/filter/StateFilter.java b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/node/filter/StateFilter.java
deleted file mode 100644
index 9e3928ecbe5..00000000000
--- a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/node/filter/StateFilter.java
+++ /dev/null
@@ -1,43 +0,0 @@
-// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
-package com.yahoo.vespa.hosted.provision.node.filter;
-
-import com.yahoo.text.StringUtilities;
-import com.yahoo.vespa.hosted.provision.Node;
-
-import java.util.EnumSet;
-import java.util.Objects;
-import java.util.function.Predicate;
-import java.util.stream.Collectors;
-
-/**
- * A node filter which filters on node states.
- *
- * @author bratseth
- */
-public class StateFilter {
-
-    private StateFilter() {}
-
-    private static Predicate<Node> makePredicate(EnumSet<Node.State> states) {
-        Objects.requireNonNull(states, "state cannot be null, use an empty set");
-        return node -> states.contains(node.state());
-    }
-
-    /** Returns a copy of the given filter which only matches for the given state */
-    public static Predicate<Node> from(Node.State state) {
-        return makePredicate(EnumSet.of(state));
-    }
-
-    /** Returns a node filter which matches a comma or space-separated list of states */
-    public static Predicate<Node> from(String states, boolean includeDeprovisioned) {
-        if (states == null) {
-            return makePredicate(includeDeprovisioned ?
-                    EnumSet.allOf(Node.State.class) : EnumSet.complementOf(EnumSet.of(Node.State.deprovisioned)));
-        }
-
-        return makePredicate(StringUtilities.split(states).stream()
-                .map(Node.State::valueOf)
-                .collect(Collectors.toCollection(() -> EnumSet.noneOf(Node.State.class))));
-    }
-
-}
diff --git a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/persistence/NodeSerializer.java b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/persistence/NodeSerializer.java
index 7d03c14f172..543972a9cb3 100644
--- a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/persistence/NodeSerializer.java
+++ b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/persistence/NodeSerializer.java
@@ -254,10 +254,10 @@ public class NodeSerializer {
     // ---------------- Deserialization --------------------------------------------------
 
     public Node fromJson(Node.State state, byte[] data) {
-        var key = Hashing.sipHash24().newHasher()
-                         .putString(state.name(), StandardCharsets.UTF_8)
-                         .putBytes(data).hash()
-                         .asLong();
+        long key = Hashing.sipHash24().newHasher()
+                          .putString(state.name(), StandardCharsets.UTF_8)
+                          .putBytes(data).hash()
+                          .asLong();
         try {
             return cache.get(key, () -> nodeFromSlime(state, SlimeUtils.jsonToSlime(data).get()));
         } catch (ExecutionException e) {
diff --git a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/restapi/NodesResponse.java b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/restapi/NodesResponse.java
index 4852f0f8269..922c8bc8e20 100644
--- a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/restapi/NodesResponse.java
+++ b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/restapi/NodesResponse.java
@@ -5,27 +5,28 @@ import com.yahoo.config.provision.ApplicationId;
 import com.yahoo.config.provision.ClusterMembership;
 import com.yahoo.config.provision.DockerImage;
 import com.yahoo.config.provision.NodeResources;
-import com.yahoo.config.provision.NodeType;
 import com.yahoo.config.provision.serialization.NetworkPortsSerializer;
 import com.yahoo.container.jdisc.HttpRequest;
 import com.yahoo.restapi.SlimeJsonResponse;
 import com.yahoo.slime.Cursor;
 import com.yahoo.vespa.applicationmodel.HostName;
 import com.yahoo.vespa.hosted.provision.Node;
+import com.yahoo.vespa.hosted.provision.NodeList;
 import com.yahoo.vespa.hosted.provision.NodeRepository;
 import com.yahoo.vespa.hosted.provision.node.Address;
 import com.yahoo.vespa.hosted.provision.node.History;
 import com.yahoo.vespa.hosted.provision.node.TrustStoreItem;
+import com.yahoo.vespa.hosted.provision.node.filter.NodeFilter;
 import com.yahoo.vespa.orchestrator.Orchestrator;
 import com.yahoo.vespa.orchestrator.status.HostInfo;
 import com.yahoo.vespa.orchestrator.status.HostStatus;
 
 import java.net.URI;
+import java.util.Comparator;
 import java.util.List;
 import java.util.Optional;
 import java.util.Set;
 import java.util.function.Function;
-import java.util.function.Predicate;
 
 /**
 * @author bratseth
@@ -41,7 +42,7 @@ class NodesResponse extends SlimeJsonResponse {
     /** The parent url of nodes */
     private final String nodeParentUrl;
 
-    private final Predicate<Node> filter;
+    private final NodeFilter filter;
     private final boolean recursive;
     private final Function<HostName, Optional<HostInfo>> orchestrator;
     private final NodeRepository nodeRepository;
@@ -57,9 +58,9 @@ class NodesResponse extends SlimeJsonResponse {
 
         Cursor root = slime.setObject();
         switch (responseType) {
-            case nodeList: nodesToSlime(root); break;
+            case nodeList: nodesToSlime(filter.states(), root); break;
             case stateList : statesToSlime(root); break;
-            case nodesInStateList: nodesToSlime(NodeSerializer.stateFrom(lastElement(parentUrl)), root); break;
+            case nodesInStateList: nodesToSlime(Set.of(NodeSerializer.stateFrom(lastElement(parentUrl))), root); break;
             case singleNode : nodeToSlime(lastElement(parentUrl), root); break;
             default: throw new IllegalArgumentException();
         }
@@ -87,23 +88,23 @@ class NodesResponse extends SlimeJsonResponse {
     private void toSlime(Node.State state, Cursor object) {
         object.setString("url", parentUrl + NodeSerializer.toString(state));
         if (recursive)
-            nodesToSlime(state, object);
+            nodesToSlime(Set.of(state), object);
     }
 
-    /** Outputs the nodes in the given state to a node array */
-    private void nodesToSlime(Node.State state, Cursor parentObject) {
+    /** Outputs the nodes in the given states to a node array */
+    private void nodesToSlime(Set<Node.State> statesToRead, Cursor parentObject) {
         Cursor nodeArray = parentObject.setArray("nodes");
-        for (NodeType type : NodeType.values())
-            toSlime(nodeRepository.nodes().list(state).nodeType(type).asList(), nodeArray);
-    }
-
-    /** Outputs all the nodes to a node array */
-    private void nodesToSlime(Cursor parentObject) {
-        Cursor nodeArray = parentObject.setArray("nodes");
-        toSlime(nodeRepository.nodes().list().asList(), nodeArray);
+        boolean sortByNodeType = statesToRead.size() == 1;
+        statesToRead.stream().sorted().forEach(state -> {
+            NodeList nodes = nodeRepository.nodes().list(state);
+            if (sortByNodeType) {
+                nodes = nodes.sortedBy(Comparator.comparing(Node::type));
+            }
+            toSlime(nodes, nodeArray);
+        });
     }
 
-    private void toSlime(List<Node> nodes, Cursor array) {
+    private void toSlime(NodeList nodes, Cursor array) {
         for (Node node : nodes) {
             if ( ! filter.test(node)) continue;
             toSlime(node, recursive, array.addObject());
diff --git a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/restapi/NodesV2ApiHandler.java b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/restapi/NodesV2ApiHandler.java
index 67bb69b6191..15e1061f5e1 100644
--- a/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/restapi/NodesV2ApiHandler.java
+++ b/node-repository/src/main/java/com/yahoo/vespa/hosted/provision/restapi/NodesV2ApiHandler.java
@@ -35,11 +35,11 @@ import com.yahoo.vespa.hosted.provision.node.Address;
 import com.yahoo.vespa.hosted.provision.node.Agent;
 import com.yahoo.vespa.hosted.provision.node.IP;
 import com.yahoo.vespa.hosted.provision.node.filter.ApplicationFilter;
+import com.yahoo.vespa.hosted.provision.node.filter.NodeFilter;
 import com.yahoo.vespa.hosted.provision.node.filter.NodeHostFilter;
 import com.yahoo.vespa.hosted.provision.node.filter.NodeOsVersionFilter;
 import com.yahoo.vespa.hosted.provision.node.filter.NodeTypeFilter;
 import com.yahoo.vespa.hosted.provision.node.filter.ParentHostFilter;
-import com.yahoo.vespa.hosted.provision.node.filter.StateFilter;
 import com.yahoo.vespa.hosted.provision.restapi.NodesResponse.ResponseType;
 import com.yahoo.vespa.orchestrator.Orchestrator;
 import com.yahoo.yolean.Exceptions;
@@ -56,7 +56,6 @@ import java.util.List;
 import java.util.Optional;
 import java.util.Set;
 import java.util.function.Function;
-import java.util.function.Predicate;
 import java.util.logging.Level;
 import java.util.stream.Collectors;
 
@@ -322,16 +321,17 @@ public class NodesV2ApiHandler extends LoggingRequestHandler {
         return NodeSerializer.typeFrom(object.asString());
     }
 
-    public static Predicate<Node> toNodeFilter(HttpRequest request) {
-        return NodeHostFilter.from(HostFilter.from(request.getProperty("hostname"),
-                                                   request.getProperty("flavor"),
-                                                   request.getProperty("clusterType"),
-                                                   request.getProperty("clusterId")))
-                .and(ApplicationFilter.from(request.getProperty("application")))
-                .and(StateFilter.from(request.getProperty("state"), request.getBooleanProperty("includeDeprovisioned")))
-                .and(NodeTypeFilter.from(request.getProperty("type")))
-                .and(ParentHostFilter.from(request.getProperty("parentHost")))
-                .and(NodeOsVersionFilter.from(request.getProperty("osVersion")));
+    public static NodeFilter toNodeFilter(HttpRequest request) {
+        return NodeFilter.in(request.getProperty("state"),
+                             request.getBooleanProperty("includeDeprovisioned"))
+                         .matching(NodeHostFilter.from(HostFilter.from(request.getProperty("hostname"),
+                                                                       request.getProperty("flavor"),
+                                                                       request.getProperty("clusterType"),
+                                                                       request.getProperty("clusterId")))
+                                                 .and(ApplicationFilter.from(request.getProperty("application")))
+                                                 .and(NodeTypeFilter.from(request.getProperty("type")))
+                                                 .and(ParentHostFilter.from(request.getProperty("parentHost")))
+                                                 .and(NodeOsVersionFilter.from(request.getProperty("osVersion"))));
     }
 
     private static boolean isPatchOverride(HttpRequest request) {
diff --git a/searchcorespi/CMakeLists.txt b/searchcorespi/CMakeLists.txt
index 66ff3fff0f2..1029619150a 100644
--- a/searchcorespi/CMakeLists.txt
+++ b/searchcorespi/CMakeLists.txt
@@ -18,4 +18,8 @@ vespa_define_module(
     src/vespa/searchcorespi
     src/vespa/searchcorespi/flush
     src/vespa/searchcorespi/index
+
+    TESTS
+    src/tests/index/active_disk_indexes
+    src/tests/index/index_disk_layout
 )
diff --git a/searchcorespi/src/tests/index/active_disk_indexes/CMakeLists.txt b/searchcorespi/src/tests/index/active_disk_indexes/CMakeLists.txt
new file mode 100644
index 00000000000..e10ada381bf
--- /dev/null
+++ b/searchcorespi/src/tests/index/active_disk_indexes/CMakeLists.txt
@@ -0,0 +1,9 @@
+# Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+vespa_add_executable(searchcorespi_active_disk_indexes_test_app
+    SOURCES
+    active_disk_indexes_test.cpp
+    DEPENDS
+    searchcorespi
+    GTest::GTest
+)
+vespa_add_test(NAME searchcorespi_active_disk_indexes_test_app COMMAND searchcorespi_active_disk_indexes_test_app)
diff --git a/searchcorespi/src/tests/index/active_disk_indexes/active_disk_indexes_test.cpp b/searchcorespi/src/tests/index/active_disk_indexes/active_disk_indexes_test.cpp
new file mode 100644
index 00000000000..a6d412817d9
--- /dev/null
+++ b/searchcorespi/src/tests/index/active_disk_indexes/active_disk_indexes_test.cpp
@@ -0,0 +1,53 @@
+// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#include <vespa/searchcorespi/index/activediskindexes.h>
+#include <vespa/searchcorespi/index/index_disk_dir.h>
+#include <vespa/vespalib/gtest/gtest.h>
+
+namespace searchcorespi::index {
+
+class ActiveDiskIndexesTest : public ::testing::Test,
+                              public ActiveDiskIndexes
+{
+protected:
+    ActiveDiskIndexesTest();
+    ~ActiveDiskIndexesTest();
+};
+
+ActiveDiskIndexesTest::ActiveDiskIndexesTest()
+    : ::testing::Test(),
+      ActiveDiskIndexes()
+{
+}
+
+ActiveDiskIndexesTest::~ActiveDiskIndexesTest() = default;
+
+TEST_F(ActiveDiskIndexesTest, simple_set_active_works)
+{
+    EXPECT_FALSE(isActive("index.flush.1"));
+    setActive("index.flush.1");
+    EXPECT_TRUE(isActive("index.flush.1"));
+    notActive("index.flush.1");
+    EXPECT_FALSE(isActive("index.flush.1"));
+}
+
+TEST_F(ActiveDiskIndexesTest, nested_set_active_works)
+{
+    setActive("index.flush.1");
+    setActive("index.flush.1");
+    EXPECT_TRUE(isActive("index.flush.1"));
+    notActive("index.flush.1");
+    EXPECT_TRUE(isActive("index.flush.1"));
+    notActive("index.flush.1");
+    EXPECT_FALSE(isActive("index.flush.1"));
+}
+
+TEST_F(ActiveDiskIndexesTest, is_active_returns_false_for_bad_name)
+{
+    EXPECT_FALSE(isActive("foo/bar/baz"));
+    EXPECT_FALSE(isActive("index.flush.0"));
+}
+
+}
+
+GTEST_MAIN_RUN_ALL_TESTS()
diff --git a/searchcorespi/src/tests/index/index_disk_layout/CMakeLists.txt b/searchcorespi/src/tests/index/index_disk_layout/CMakeLists.txt
new file mode 100644
index 00000000000..4e82cf1b9d2
--- /dev/null
+++ b/searchcorespi/src/tests/index/index_disk_layout/CMakeLists.txt
@@ -0,0 +1,9 @@
+# Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+vespa_add_executable(searchcorespi_index_disk_layout_test_app
+    SOURCES
+    index_disk_layout_test.cpp
+    DEPENDS
+    searchcorespi
+    GTest::GTest
+)
+vespa_add_test(NAME searchcorespi_index_disk_layout_test_app COMMAND searchcorespi_index_disk_layout_test_app)
diff --git a/searchcorespi/src/tests/index/index_disk_layout/index_disk_layout_test.cpp b/searchcorespi/src/tests/index/index_disk_layout/index_disk_layout_test.cpp
new file mode 100644
index 00000000000..e35225b2745
--- /dev/null
+++ b/searchcorespi/src/tests/index/index_disk_layout/index_disk_layout_test.cpp
@@ -0,0 +1,60 @@
+// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#include <vespa/searchcorespi/index/indexdisklayout.h>
+#include <vespa/searchcorespi/index/index_disk_dir.h>
+#include <vespa/vespalib/gtest/gtest.h>
+
+namespace searchcorespi::index {
+
+namespace {
+
+void expect_index_disk_dir(IndexDiskDir exp, const vespalib::string& dir)
+{
+    auto act = IndexDiskLayout::get_index_disk_dir(dir);
+    ASSERT_TRUE(act.valid());
+    ASSERT_EQ(exp, act);
+}
+
+void expect_bad_index_disk_dir(const vespalib::string& dir)
+{
+    auto act = IndexDiskLayout::get_index_disk_dir(dir);
+    ASSERT_FALSE(act.valid());
+}
+
+}
+
+TEST(IndexDiskLayoutTest, get_index_disk_dir_works)
+{
+    {
+        SCOPED_TRACE("index.fusion.1");
+        expect_index_disk_dir(IndexDiskDir(1, true), "index.fusion.1");
+    }
+    {
+        SCOPED_TRACE("index.flush.2");
+        expect_index_disk_dir(IndexDiskDir(2, false), "index.flush.2");
+    }
+    {
+        SCOPED_TRACE("index.flush.3");
+        expect_index_disk_dir(IndexDiskDir(3, false), "index.flush.3");
+    }
+    {
+        SCOPED_TRACE("foo/bar/index.flush.4");
+        expect_index_disk_dir(IndexDiskDir(4, false), "foo/bar/index.flush.4");
+    }
+    {
+        SCOPED_TRACE("index.flush.");
+        expect_bad_index_disk_dir("index.flush.");
+    }
+    {
+        SCOPED_TRACE("index.flush.0");
+        expect_bad_index_disk_dir("index.flush.0");
+    }
+    {
+        SCOPED_TRACE("asdf");
+        expect_bad_index_disk_dir("asdf");
+    }
+}
+
+}
+
+GTEST_MAIN_RUN_ALL_TESTS()
diff --git a/searchcorespi/src/vespa/searchcorespi/index/CMakeLists.txt b/searchcorespi/src/vespa/searchcorespi/index/CMakeLists.txt
index c1e2b2f3dd1..1987304dc7e 100644
--- a/searchcorespi/src/vespa/searchcorespi/index/CMakeLists.txt
+++ b/searchcorespi/src/vespa/searchcorespi/index/CMakeLists.txt
@@ -8,6 +8,7 @@ vespa_add_library(searchcorespi_index OBJECT
     fusionrunner.cpp
     iindexmanager.cpp
     iindexcollection.cpp
+    index_disk_dir_active_state.cpp
     index_manager_explorer.cpp
     index_manager_stats.cpp
     indexcollection.cpp
diff --git a/searchcorespi/src/vespa/searchcorespi/index/activediskindexes.cpp b/searchcorespi/src/vespa/searchcorespi/index/activediskindexes.cpp
index c7891709801..fb9585fb58e 100644
--- a/searchcorespi/src/vespa/searchcorespi/index/activediskindexes.cpp
+++ b/searchcorespi/src/vespa/searchcorespi/index/activediskindexes.cpp
@@ -1,6 +1,9 @@
 // Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 
 #include "activediskindexes.h"
+#include "indexdisklayout.h"
+#include "index_disk_dir.h"
+#include "index_disk_dir_active_state.h"
 #include <cassert>
 
 using vespalib::string;
@@ -11,20 +14,34 @@ ActiveDiskIndexes::ActiveDiskIndexes() = default;
 ActiveDiskIndexes::~ActiveDiskIndexes() = default;
 
 void ActiveDiskIndexes::setActive(const string &index) {
+    auto index_disk_dir = IndexDiskLayout::get_index_disk_dir(index);
+    assert(index_disk_dir.valid());
     std::lock_guard lock(_lock);
-    _active.insert(index);
+    auto insres = _active.insert(std::make_pair(index_disk_dir, IndexDiskDirActiveState()));
+    insres.first->second.activate();
 }
 
 void ActiveDiskIndexes::notActive(const string & index) {
+    auto index_disk_dir = IndexDiskLayout::get_index_disk_dir(index);
+    assert(index_disk_dir.valid());
     std::lock_guard lock(_lock);
-    auto it = _active.find(index);
+    auto it = _active.find(index_disk_dir);
     assert(it != _active.end());
-    _active.erase(it);
+    assert(it->second.is_active());
+    it->second.deactivate();
+    if (!it->second.is_active()) {
+        _active.erase(it);
+    }
 }
 
 bool ActiveDiskIndexes::isActive(const string &index) const {
+    auto index_disk_dir = IndexDiskLayout::get_index_disk_dir(index);
+    if (!index_disk_dir.valid()) {
+        return false;
+    }
     std::lock_guard lock(_lock);
-    return _active.find(index) != _active.end();
+    auto it = _active.find(index_disk_dir);
+    return (it != _active.end()) && it->second.is_active();
 }
 
 }
diff --git a/searchcorespi/src/vespa/searchcorespi/index/activediskindexes.h b/searchcorespi/src/vespa/searchcorespi/index/activediskindexes.h
index a63dbb8b2a7..365025e7450 100644
--- a/searchcorespi/src/vespa/searchcorespi/index/activediskindexes.h
+++ b/searchcorespi/src/vespa/searchcorespi/index/activediskindexes.h
@@ -3,18 +3,21 @@
 #pragma once
 
 #include <vespa/vespalib/stllike/string.h>
-#include <set>
+#include <map>
 #include <mutex>
 #include <memory>
 
 namespace searchcorespi::index {
 
+class IndexDiskDir;
+class IndexDiskDirActiveState;
+
 /**
  * Class used to keep track of the set of active disk indexes in an index maintainer.
  * The index directories are used as identifiers.
  */
 class ActiveDiskIndexes {
-    std::multiset<vespalib::string> _active;
+    std::map<IndexDiskDir, IndexDiskDirActiveState> _active;
     mutable std::mutex _lock;
 
 public:
diff --git a/searchcorespi/src/vespa/searchcorespi/index/index_disk_dir.h b/searchcorespi/src/vespa/searchcorespi/index/index_disk_dir.h
new file mode 100644
index 00000000000..278fd73e555
--- /dev/null
+++ b/searchcorespi/src/vespa/searchcorespi/index/index_disk_dir.h
@@ -0,0 +1,34 @@
+// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+#pragma once
+
+namespace searchcorespi::index {
+
+/*
+ * Class naming a disk index for a document type.
+ */
+class IndexDiskDir {
+    uint32_t _id;
+    bool     _fusion;
+public:
+    IndexDiskDir(uint32_t id, bool fusion) noexcept
+        : _id(id),
+          _fusion(fusion)
+    {
+    }
+    IndexDiskDir() noexcept
+        : IndexDiskDir(0, false)
+    {
+    }
+    bool operator<(const IndexDiskDir& rhs) const noexcept {
+        if (_id != rhs._id) {
+            return _id < rhs._id;
+        }
+        return !_fusion && rhs._fusion;
+    }
+    bool operator==(const IndexDiskDir& rhs) const noexcept {
+        return (_id == rhs._id) && (_fusion == rhs._fusion);
+    }
+    bool valid() const noexcept { return _id != 0u; }
+};
+
+}
diff --git a/searchcorespi/src/vespa/searchcorespi/index/index_disk_dir_active_state.cpp b/searchcorespi/src/vespa/searchcorespi/index/index_disk_dir_active_state.cpp
new file mode 100644
index 00000000000..603971c866e
--- /dev/null
+++ b/searchcorespi/src/vespa/searchcorespi/index/index_disk_dir_active_state.cpp
@@ -0,0 +1,15 @@
+// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#include "index_disk_dir_active_state.h"
+#include <cassert>
+
+namespace searchcorespi::index {
+
+void
+IndexDiskDirActiveState::deactivate() noexcept
+{
+    assert(_active_count > 0u);
+    --_active_count;
+}
+
+}
diff --git a/searchcorespi/src/vespa/searchcorespi/index/index_disk_dir_active_state.h b/searchcorespi/src/vespa/searchcorespi/index/index_disk_dir_active_state.h
new file mode 100644
index 00000000000..ac84de5adee
--- /dev/null
+++ b/searchcorespi/src/vespa/searchcorespi/index/index_disk_dir_active_state.h
@@ -0,0 +1,25 @@
+// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#pragma once
+
+#include <cstdint>
+
+namespace searchcorespi::index {
+
+/*
+ * Class describing active state for a disk index directory.
+ */
+class IndexDiskDirActiveState {
+    uint32_t _active_count;
+public:
+    IndexDiskDirActiveState()
+        : _active_count(0)
+    {
+    }
+
+    void activate() noexcept { ++_active_count; }
+    void deactivate() noexcept;
+    bool is_active() const noexcept { return _active_count != 0; }
+};
+
+}
diff --git a/searchcorespi/src/vespa/searchcorespi/index/indexdisklayout.cpp b/searchcorespi/src/vespa/searchcorespi/index/indexdisklayout.cpp
index a13633751bb..c701d1dfb1d 100644
--- a/searchcorespi/src/vespa/searchcorespi/index/indexdisklayout.cpp
+++ b/searchcorespi/src/vespa/searchcorespi/index/indexdisklayout.cpp
@@ -1,6 +1,7 @@
 // Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 
 #include "indexdisklayout.h"
+#include "index_disk_dir.h"
 #include <sstream>
 
 namespace searchcorespi::index {
@@ -53,4 +54,24 @@ IndexDiskLayout::getSelectorFileName(const vespalib::string &dir)
     return dir + "/selector";
 }
 
+IndexDiskDir
+IndexDiskLayout::get_index_disk_dir(const vespalib::string& dir)
+{
+    auto name = dir.substr(dir.rfind('/') + 1);
+    const vespalib::string* prefix = nullptr;
+    bool fusion = false;
+    if (name.find(FlushDirPrefix) == 0) {
+        prefix = &FlushDirPrefix;
+    } else if (name.find(FusionDirPrefix) == 0) {
+        prefix = &FusionDirPrefix;
+        fusion = true;
+    } else {
+        return IndexDiskDir(); // invalid
+    }
+    std::istringstream ist(name.substr(prefix->size()));
+    uint32_t id = 0;
+    ist >> id;
+    return IndexDiskDir(id, fusion); // invalid if id == 0u
+}
+
 }
diff --git a/searchcorespi/src/vespa/searchcorespi/index/indexdisklayout.h b/searchcorespi/src/vespa/searchcorespi/index/indexdisklayout.h
index 0598ad17c8a..94b35936cc7 100644
--- a/searchcorespi/src/vespa/searchcorespi/index/indexdisklayout.h
+++ b/searchcorespi/src/vespa/searchcorespi/index/indexdisklayout.h
@@ -6,6 +6,8 @@
 namespace searchcorespi {
 namespace index {
 
+class IndexDiskDir;
+
 /**
  * Utility class used to get static aspects of the disk layout (i.e directory and file names)
  * needed by the index maintainer.
@@ -23,6 +25,7 @@ public:
     IndexDiskLayout(const vespalib::string &baseDir);
     vespalib::string getFlushDir(uint32_t sourceId) const;
     vespalib::string getFusionDir(uint32_t sourceId) const;
+    static IndexDiskDir get_index_disk_dir(const vespalib::string& dir);
 
     static vespalib::string getSerialNumFileName(const vespalib::string &dir);
     static vespalib::string getSchemaFileName(const vespalib::string &dir);