Allow inhibiting all default bucket space merges if global merges pending

If the cluster controller publishes a cluster state where at least
one node is marked as maintenance in the default space but is up
in the global space, global bucket merges are pending in the cluster.
We want these merges to complete ASAP, and to avoid starvation caused
by distributors that finish their global merges quicker than others
(or whose global bucket ops get preempted), allow for _completely_
inhibiting merges to buckets in the default space when global buckets
are pending.

Requires a live config flag to be set to actually be taken into account
in the bucket synchronization state checker.

Flag is tracked per distributor bucket space (per stripe) and is
automatically cleared once a cluster state is received without pending
global merges.

14 files changed, 205 insertions, 10 deletions

diff --git a/storage/src/tests/distributor/CMakeLists.txt b/storage/src/tests/distributor/CMakeLists.txt
index bee7650aebd..501f9a18c47 100644
--- a/storage/src/tests/distributor/CMakeLists.txt
+++ b/storage/src/tests/distributor/CMakeLists.txt
@@ -9,6 +9,7 @@ vespa_add_executable(storage_distributor_gtest_runner_app TEST
     bucketdbmetricupdatertest.cpp
     bucketgctimecalculatortest.cpp
     bucketstateoperationtest.cpp
+    distributor_bucket_space_repo_test.cpp
     distributor_bucket_space_test.cpp
     distributor_host_info_reporter_test.cpp
     distributor_message_sender_stub.cpp
diff --git a/storage/src/tests/distributor/distributor_bucket_space_repo_test.cpp b/storage/src/tests/distributor/distributor_bucket_space_repo_test.cpp
new file mode 100644
index 00000000000..151dcff3d10
--- /dev/null
+++ b/storage/src/tests/distributor/distributor_bucket_space_repo_test.cpp
@@ -0,0 +1,72 @@
+// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#include <vespa/document/bucket/fixed_bucket_spaces.h>
+#include <vespa/storage/distributor/distributor_bucket_space.h>
+#include <vespa/storage/distributor/distributor_bucket_space_repo.h>
+#include <vespa/vdslib/state/cluster_state_bundle.h>
+#include <vespa/vdslib/state/clusterstate.h>
+#include <vespa/vespalib/gtest/gtest.h>
+#include <memory>
+
+namespace storage::distributor {
+
+using document::FixedBucketSpaces;
+using namespace ::testing;
+
+struct DistributorBucketSpaceRepoTest : Test {
+    DistributorBucketSpaceRepo _repo;
+
+    DistributorBucketSpaceRepoTest() : _repo(123) {}
+};
+
+namespace {
+
+lib::ClusterStateBundle bundle_with_global_merges() {
+    auto global_state   = std::make_shared<lib::ClusterState>("distributor:1 storage:2");
+    auto default_state  = std::make_shared<lib::ClusterState>("distributor:1 storage:2 .1.s:m");
+    return lib::ClusterStateBundle(*global_state, {{FixedBucketSpaces::default_space(), default_state},
+                                                   {FixedBucketSpaces::global_space(), global_state}});
+}
+
+lib::ClusterStateBundle bundle_without_global_merges() {
+    auto global_state   = std::make_shared<lib::ClusterState>("distributor:1 storage:2");
+    auto default_state  = std::make_shared<lib::ClusterState>("distributor:1 storage:2");
+    return lib::ClusterStateBundle(*global_state, {{FixedBucketSpaces::default_space(), default_state},
+                                                   {FixedBucketSpaces::global_space(), global_state}});
+}
+
+}
+
+TEST_F(DistributorBucketSpaceRepoTest, bucket_spaces_are_initially_not_tagged_as_merge_inhibited) {
+    EXPECT_FALSE(_repo.get(FixedBucketSpaces::default_space()).merges_inhibited());
+    EXPECT_FALSE(_repo.get(FixedBucketSpaces::global_space()).merges_inhibited());
+}
+
+TEST_F(DistributorBucketSpaceRepoTest, enabled_bundle_with_pending_global_merges_tags_default_space_as_merge_inhibited) {
+    _repo.enable_cluster_state_bundle(bundle_with_global_merges());
+    EXPECT_TRUE(_repo.get(FixedBucketSpaces::default_space()).merges_inhibited());
+    EXPECT_FALSE(_repo.get(FixedBucketSpaces::global_space()).merges_inhibited());
+}
+
+TEST_F(DistributorBucketSpaceRepoTest, enabled_bundle_without_pending_global_merges_unsets_merge_inhibition) {
+    _repo.enable_cluster_state_bundle(bundle_with_global_merges());
+    _repo.enable_cluster_state_bundle(bundle_without_global_merges());
+    EXPECT_FALSE(_repo.get(FixedBucketSpaces::default_space()).merges_inhibited());
+    EXPECT_FALSE(_repo.get(FixedBucketSpaces::global_space()).merges_inhibited());
+}
+
+TEST_F(DistributorBucketSpaceRepoTest, pending_bundle_with_pending_global_merges_tags_default_space_as_merge_inhibited) {
+    _repo.enable_cluster_state_bundle(bundle_without_global_merges());
+    _repo.set_pending_cluster_state_bundle(bundle_with_global_merges());
+    EXPECT_TRUE(_repo.get(FixedBucketSpaces::default_space()).merges_inhibited());
+    EXPECT_FALSE(_repo.get(FixedBucketSpaces::global_space()).merges_inhibited());
+}
+
+TEST_F(DistributorBucketSpaceRepoTest, pending_bundle_without_pending_global_unsets_merge_inhibition) {
+    _repo.enable_cluster_state_bundle(bundle_with_global_merges());
+    _repo.set_pending_cluster_state_bundle(bundle_without_global_merges());
+    EXPECT_FALSE(_repo.get(FixedBucketSpaces::default_space()).merges_inhibited());
+    EXPECT_FALSE(_repo.get(FixedBucketSpaces::global_space()).merges_inhibited());
+}
+
+}
diff --git a/storage/src/tests/distributor/distributor_stripe_test.cpp b/storage/src/tests/distributor/distributor_stripe_test.cpp
index 8c2ebc983fa..709f2e6cdc5 100644
--- a/storage/src/tests/distributor/distributor_stripe_test.cpp
+++ b/storage/src/tests/distributor/distributor_stripe_test.cpp
@@ -629,6 +629,19 @@ TEST_F(DistributorStripeTest, max_clock_skew_config_is_propagated_to_distributor
     EXPECT_EQ(getConfig().getMaxClusterClockSkew(), std::chrono::seconds(5));
 }
 
+TEST_F(DistributorStripeTest, inhibit_default_merge_if_global_merges_pending_config_is_propagated)
+{
+    setup_stripe(Redundancy(2), NodeCount(2), "storage:2 distributor:1");
+    ConfigBuilder builder;
+    builder.inhibitDefaultMergesWhenGlobalMergesPending = true;
+    configure_stripe(builder);
+    EXPECT_TRUE(getConfig().inhibit_default_merges_when_global_merges_pending());
+
+    builder.inhibitDefaultMergesWhenGlobalMergesPending = false;
+    configure_stripe(builder);
+    EXPECT_FALSE(getConfig().inhibit_default_merges_when_global_merges_pending());
+}
+
 namespace {
 
 auto makeDummyRemoveCommand() {
diff --git a/storage/src/tests/distributor/statecheckerstest.cpp b/storage/src/tests/distributor/statecheckerstest.cpp
index d481370b2c1..94f913a3325 100644
--- a/storage/src/tests/distributor/statecheckerstest.cpp
+++ b/storage/src/tests/distributor/statecheckerstest.cpp
@@ -175,6 +175,8 @@ struct StateCheckersTest : Test, DistributorStripeTestUtil {
         bool _includeSchedulingPriority {false};
         bool _merge_operations_disabled {false};
         bool _prioritize_global_bucket_merges {true};
+        bool _config_enable_default_space_merge_inhibition {false};
+        bool _merges_inhibited_in_bucket_space {false};
         CheckerParams();
         ~CheckerParams();
 
@@ -222,6 +224,14 @@ struct StateCheckersTest : Test, DistributorStripeTestUtil {
             _bucket_space = bucket_space;
             return *this;
         }
+        CheckerParams& config_enable_default_space_merge_inhibition(bool enabled) noexcept {
+            _config_enable_default_space_merge_inhibition = enabled;
+            return *this;
+        }
+        CheckerParams& merges_inhibited_in_bucket_space(bool inhibited) noexcept {
+            _merges_inhibited_in_bucket_space = inhibited;
+            return *this;
+        }
     };
 
     template <typename CheckerImpl>
@@ -236,10 +246,12 @@ struct StateCheckersTest : Test, DistributorStripeTestUtil {
         vespa::config::content::core::StorDistributormanagerConfigBuilder config;
         config.mergeOperationsDisabled = params._merge_operations_disabled;
         config.prioritizeGlobalBucketMerges = params._prioritize_global_bucket_merges;
+        config.inhibitDefaultMergesWhenGlobalMergesPending = params._config_enable_default_space_merge_inhibition;
         configure_stripe(config);
         if (!params._pending_cluster_state.empty()) {
             simulate_set_pending_cluster_state(params._pending_cluster_state);
         }
+        getBucketSpaceRepo().get(params._bucket_space).set_merges_inhibited(params._merges_inhibited_in_bucket_space);
         NodeMaintenanceStatsTracker statsTracker;
         StateChecker::Context c(node_context(),
                                 operation_context(),
@@ -818,6 +830,32 @@ TEST_F(StateCheckersTest, no_merge_operation_generated_if_merges_explicitly_conf
             .merge_operations_disabled(true));
 }
 
+TEST_F(StateCheckersTest, no_merge_operation_generated_if_merges_inhibited_in_default_bucket_space_and_config_allowed) {
+    // Technically, the state checker doesn't look at global vs. non-global but instead defers
+    // to the distributor bucket space repo to set the inhibition flag on the correct bucket space.
+    // This particular logic is tested at a higher repo-level.
+    runAndVerify<SynchronizeAndMoveStateChecker>(
+            CheckerParams()
+                    .expect("NO OPERATIONS GENERATED") // Would normally generate a merge op
+                    .bucketInfo("0=1,2=2")
+                    .config_enable_default_space_merge_inhibition(true)
+                    .merges_inhibited_in_bucket_space(true)
+                    .clusterState("distributor:1 storage:3"));
+}
+
+TEST_F(StateCheckersTest, merge_operation_still_generated_if_merges_inhibited_in_default_bucket_space_but_config_disallowed) {
+    runAndVerify<SynchronizeAndMoveStateChecker>(
+            CheckerParams()
+                    .expect("[Moving bucket to ideal node 1]"
+                            "[Synchronizing buckets with different checksums "
+                            "node(idx=0,crc=0x1,docs=1/1,bytes=1/1,trusted=false,active=false,ready=false), "
+                            "node(idx=2,crc=0x2,docs=2/2,bytes=2/2,trusted=false,active=false,ready=false)]")
+                    .bucketInfo("0=1,2=2")
+                    .config_enable_default_space_merge_inhibition(false)
+                    .merges_inhibited_in_bucket_space(true)
+                    .clusterState("distributor:1 storage:3"));
+}
+
 std::string
 StateCheckersTest::testDeleteExtraCopies(
         const std::string& bucketInfo, uint32_t redundancy,
diff --git a/storage/src/vespa/storage/config/distributorconfiguration.cpp b/storage/src/vespa/storage/config/distributorconfiguration.cpp
index 8a40899165f..b4c23725493 100644
--- a/storage/src/vespa/storage/config/distributorconfiguration.cpp
+++ b/storage/src/vespa/storage/config/distributorconfiguration.cpp
@@ -51,6 +51,7 @@ DistributorConfiguration::DistributorConfiguration(StorageComponent& component)
       _enable_revert(true),
       _implicitly_clear_priority_on_schedule(false),
       _use_unordered_merge_chaining(false),
+      _inhibit_default_merges_when_global_merges_pending(false),
       _minimumReplicaCountingMode(ReplicaCountingMode::TRUSTED)
 {
 }
@@ -173,6 +174,7 @@ DistributorConfiguration::configure(const vespa::config::content::core::StorDist
     _enable_revert = config.enableRevert;
     _implicitly_clear_priority_on_schedule = config.implicitlyClearBucketPriorityOnSchedule;
     _use_unordered_merge_chaining = config.useUnorderedMergeChaining;
+    _inhibit_default_merges_when_global_merges_pending = config.inhibitDefaultMergesWhenGlobalMergesPending;
 
     _minimumReplicaCountingMode = config.minimumReplicaCountingMode;
 
diff --git a/storage/src/vespa/storage/config/distributorconfiguration.h b/storage/src/vespa/storage/config/distributorconfiguration.h
index ea1aca17116..b26f115827e 100644
--- a/storage/src/vespa/storage/config/distributorconfiguration.h
+++ b/storage/src/vespa/storage/config/distributorconfiguration.h
@@ -273,6 +273,12 @@ public:
     [[nodiscard]] bool use_unordered_merge_chaining() const noexcept {
         return _use_unordered_merge_chaining;
     }
+    void set_inhibit_default_merges_when_global_merges_pending(bool inhibit) noexcept {
+        _inhibit_default_merges_when_global_merges_pending = inhibit;
+    }
+    [[nodiscard]] bool inhibit_default_merges_when_global_merges_pending() const noexcept {
+        return _inhibit_default_merges_when_global_merges_pending;
+    }
 
     uint32_t num_distributor_stripes() const noexcept { return _num_distributor_stripes; }
 
@@ -331,6 +337,7 @@ private:
     bool _enable_revert;
     bool _implicitly_clear_priority_on_schedule;
     bool _use_unordered_merge_chaining;
+    bool _inhibit_default_merges_when_global_merges_pending;
 
     DistrConfig::MinimumReplicaCountingMode _minimumReplicaCountingMode;
 
diff --git a/storage/src/vespa/storage/distributor/distributor_bucket_space.cpp b/storage/src/vespa/storage/distributor/distributor_bucket_space.cpp
index 7293f9f7acc..5969ccad4cb 100644
--- a/storage/src/vespa/storage/distributor/distributor_bucket_space.cpp
+++ b/storage/src/vespa/storage/distributor/distributor_bucket_space.cpp
@@ -28,6 +28,7 @@ DistributorBucketSpace::DistributorBucketSpace(uint16_t node_index)
       _distribution(),
       _node_index(node_index),
       _distribution_bits(1u),
+      _merges_inhibited(false),
       _pending_cluster_state(),
       _available_nodes(),
       _ownerships(),
diff --git a/storage/src/vespa/storage/distributor/distributor_bucket_space.h b/storage/src/vespa/storage/distributor/distributor_bucket_space.h
index 32da78f9972..3dfd1e1ce30 100644
--- a/storage/src/vespa/storage/distributor/distributor_bucket_space.h
+++ b/storage/src/vespa/storage/distributor/distributor_bucket_space.h
@@ -37,6 +37,7 @@ class DistributorBucketSpace {
     std::shared_ptr<const lib::Distribution> _distribution;
     uint16_t                                 _node_index;
     uint16_t                                 _distribution_bits;
+    bool                                     _merges_inhibited;
     std::shared_ptr<const lib::ClusterState> _pending_cluster_state;
     std::vector<bool>                        _available_nodes;
     mutable vespalib::hash_map<document::BucketId, BucketOwnershipFlags, document::BucketId::hash>  _ownerships;
@@ -85,6 +86,13 @@ public:
     bool has_pending_cluster_state() const noexcept { return static_cast<bool>(_pending_cluster_state); }
     const lib::ClusterState& get_pending_cluster_state() const noexcept { return *_pending_cluster_state; }
 
+    void set_merges_inhibited(bool inhibited) noexcept {
+        _merges_inhibited = inhibited;
+    }
+    [[nodiscard]] bool merges_inhibited() const noexcept {
+        return _merges_inhibited;
+    }
+
     /**
      * Returns true if this distributor owns the given bucket in the
      * given cluster and current distribution config.
diff --git a/storage/src/vespa/storage/distributor/distributor_bucket_space_repo.cpp b/storage/src/vespa/storage/distributor/distributor_bucket_space_repo.cpp
index 09468b55430..c88abaf8373 100644
--- a/storage/src/vespa/storage/distributor/distributor_bucket_space_repo.cpp
+++ b/storage/src/vespa/storage/distributor/distributor_bucket_space_repo.cpp
@@ -3,6 +3,7 @@
 #include "distributor_bucket_space_repo.h"
 #include "distributor_bucket_space.h"
 #include <vespa/vdslib/state/cluster_state_bundle.h>
+#include <vespa/vdslib/state/clusterstate.h>
 #include <vespa/document/bucket/fixed_bucket_spaces.h>
 #include <cassert>
 
@@ -10,14 +11,15 @@
 LOG_SETUP(".distributor.distributor_bucket_space_repo");
 
 using document::BucketSpace;
+using document::FixedBucketSpaces;
 
 namespace storage::distributor {
 
 DistributorBucketSpaceRepo::DistributorBucketSpaceRepo(uint16_t node_index)
     : _map()
 {
-    add(document::FixedBucketSpaces::default_space(), std::make_unique<DistributorBucketSpace>(node_index));
-    add(document::FixedBucketSpaces::global_space(), std::make_unique<DistributorBucketSpace>(node_index));
+    add(FixedBucketSpaces::default_space(), std::make_unique<DistributorBucketSpace>(node_index));
+    add(FixedBucketSpaces::global_space(), std::make_unique<DistributorBucketSpace>(node_index));
 }
 
 DistributorBucketSpaceRepo::~DistributorBucketSpaceRepo() = default;
@@ -44,12 +46,54 @@ DistributorBucketSpaceRepo::get(BucketSpace bucketSpace) const
     return *itr->second;
 }
 
+namespace {
+
+bool content_node_is_up(const lib::ClusterState& state, uint16_t index) noexcept {
+    return (state.getNodeState(lib::Node(lib::NodeType::STORAGE, index)).getState() == lib::State::UP);
+}
+
+bool content_node_is_in_maintenance(const lib::ClusterState& state, uint16_t index) noexcept {
+    return (state.getNodeState(lib::Node(lib::NodeType::STORAGE, index)).getState() == lib::State::MAINTENANCE);
+}
+
+// Prioritized global bucket merging is taking place if at least one content node is
+// marked as Up in the global bucket space state, but Maintenance in the default
+// bucket space state.
+bool bundle_implies_global_merging_active(const lib::ClusterStateBundle& bundle) noexcept {
+    auto& default_cs = bundle.getDerivedClusterState(FixedBucketSpaces::default_space());
+    auto& global_cs  = bundle.getDerivedClusterState(FixedBucketSpaces::global_space());
+    if (default_cs.get() == global_cs.get()) {
+        return false;
+    }
+    uint16_t node_count = global_cs->getNodeCount(lib::NodeType::STORAGE);
+    for (uint16_t i = 0; i < node_count; ++i) {
+        if (content_node_is_up(*global_cs, i) && content_node_is_in_maintenance(*default_cs, i)) {
+            return true;
+        }
+    }
+    return false;
+}
+
+}
+
+void
+DistributorBucketSpaceRepo::enable_cluster_state_bundle(const lib::ClusterStateBundle& cluster_state_bundle)
+{
+    for (auto& entry : _map) {
+        entry.second->setClusterState(cluster_state_bundle.getDerivedClusterState(entry.first));
+    }
+    get(FixedBucketSpaces::default_space()).set_merges_inhibited(
+            bundle_implies_global_merging_active(cluster_state_bundle));
+}
+
 void
 DistributorBucketSpaceRepo::set_pending_cluster_state_bundle(const lib::ClusterStateBundle& cluster_state_bundle)
 {
     for (auto& entry : _map) {
         entry.second->set_pending_cluster_state(cluster_state_bundle.getDerivedClusterState(entry.first));
     }
+    get(FixedBucketSpaces::default_space()).set_merges_inhibited(
+            bundle_implies_global_merging_active(cluster_state_bundle));
 }
 
 void
diff --git a/storage/src/vespa/storage/distributor/distributor_bucket_space_repo.h b/storage/src/vespa/storage/distributor/distributor_bucket_space_repo.h
index ab1e235ae35..d77a9f37fb0 100644
--- a/storage/src/vespa/storage/distributor/distributor_bucket_space_repo.h
+++ b/storage/src/vespa/storage/distributor/distributor_bucket_space_repo.h
@@ -33,6 +33,7 @@ public:
     BucketSpaceMap::const_iterator begin() const { return _map.begin(); }
     BucketSpaceMap::const_iterator end() const { return _map.end(); }
     void add(document::BucketSpace bucketSpace, std::unique_ptr<DistributorBucketSpace> distributorBucketSpace);
+    void enable_cluster_state_bundle(const lib::ClusterStateBundle& cluster_state_bundle);
     void set_pending_cluster_state_bundle(const lib::ClusterStateBundle& cluster_state_bundle);
     void clear_pending_cluster_state_bundle();
 };
diff --git a/storage/src/vespa/storage/distributor/distributor_stripe.cpp b/storage/src/vespa/storage/distributor/distributor_stripe.cpp
index bcba976f2c3..1173a19d729 100644
--- a/storage/src/vespa/storage/distributor/distributor_stripe.cpp
+++ b/storage/src/vespa/storage/distributor/distributor_stripe.cpp
@@ -481,9 +481,7 @@ void
 DistributorStripe::propagateClusterStates()
 {
     for (auto* repo : {_bucketSpaceRepo.get(), _readOnlyBucketSpaceRepo.get()}) {
-        for (auto& iter : *repo) {
-            iter.second->setClusterState(_clusterStateBundle.getDerivedClusterState(iter.first));
-        }
+        repo->enable_cluster_state_bundle(_clusterStateBundle);
     }
 }
 
diff --git a/storage/src/vespa/storage/distributor/statechecker.cpp b/storage/src/vespa/storage/distributor/statechecker.cpp
index d1241f0bf66..e5a1822d455 100644
--- a/storage/src/vespa/storage/distributor/statechecker.cpp
+++ b/storage/src/vespa/storage/distributor/statechecker.cpp
@@ -63,9 +63,9 @@ StateChecker::Result::createStoredResult(
 
 StateChecker::Context::Context(const DistributorNodeContext& node_ctx_in,
                                const DistributorStripeOperationContext& op_ctx_in,
-                               const DistributorBucketSpace &distributorBucketSpace,
+                               const DistributorBucketSpace& distributorBucketSpace,
                                NodeMaintenanceStatsTracker& statsTracker,
-                               const document::Bucket &bucket_)
+                               const document::Bucket& bucket_)
     : bucket(bucket_),
       siblingBucket(op_ctx_in.get_sibling(bucket.getBucketId())),
       systemState(distributorBucketSpace.getClusterState()),
@@ -77,7 +77,8 @@ StateChecker::Context::Context(const DistributorNodeContext& node_ctx_in,
       node_ctx(node_ctx_in),
       op_ctx(op_ctx_in),
       db(distributorBucketSpace.getBucketDatabase()),
-      stats(statsTracker)
+      stats(statsTracker),
+      merges_inhibited_in_bucket_space(distributorBucketSpace.merges_inhibited())
 {
     idealState = distributorBucketSpace.get_ideal_service_layer_nodes_bundle(bucket.getBucketId()).get_available_nonretired_or_maintenance_nodes();
     unorderedIdealState.insert(idealState.begin(), idealState.end());
diff --git a/storage/src/vespa/storage/distributor/statechecker.h b/storage/src/vespa/storage/distributor/statechecker.h
index 3f938238ce7..b72898208e6 100644
--- a/storage/src/vespa/storage/distributor/statechecker.h
+++ b/storage/src/vespa/storage/distributor/statechecker.h
@@ -82,6 +82,7 @@ public:
         const DistributorStripeOperationContext& op_ctx;
         const BucketDatabase& db;
         NodeMaintenanceStatsTracker& stats;
+        const bool merges_inhibited_in_bucket_space;
 
         const BucketDatabase::Entry& getSiblingEntry() const {
             return siblingEntry;
@@ -97,7 +98,7 @@ public:
     class ResultImpl
     {
     public:
-        virtual ~ResultImpl() {}
+        virtual ~ResultImpl() = default;
         virtual IdealStateOperation::UP createOperation() = 0;
         virtual MaintenancePriority getPriority() const = 0;
         virtual MaintenanceOperation::Type getType() const = 0;
diff --git a/storage/src/vespa/storage/distributor/statecheckers.cpp b/storage/src/vespa/storage/distributor/statecheckers.cpp
index d400151b349..71b2da1359a 100644
--- a/storage/src/vespa/storage/distributor/statecheckers.cpp
+++ b/storage/src/vespa/storage/distributor/statecheckers.cpp
@@ -832,12 +832,20 @@ allCopiesAreInvalid(const StateChecker::Context& c)
     return true;
 }
 
+bool
+merging_effectively_disabled_for_state_checker(const StateChecker::Context& c) noexcept
+{
+    return (c.distributorConfig.merge_operations_disabled()
+            || (c.distributorConfig.inhibit_default_merges_when_global_merges_pending()
+                && c.merges_inhibited_in_bucket_space));
+}
+
 }
 
 StateChecker::Result
 SynchronizeAndMoveStateChecker::check(StateChecker::Context& c)
 {
-    if (c.distributorConfig.merge_operations_disabled()) {
+    if (merging_effectively_disabled_for_state_checker(c)) {
         return Result::noMaintenanceNeeded();
     }
     if (isInconsistentlySplit(c)) {