Do not block global merges to nodes tagged as busy

To avoid starvation of high priority global bucket merges, we do not consider
these for blocking due to a node being "busy" (usually caused by a full merge
throttler queue).

This is for two reasons:
 1. When an ideal state op is blocked, it is still removed from the internal
    maintenance priority queue. This means a blocked high pri operation will
    not be retried until the next DB pass (at which point the node is likely
    to still be marked as busy when there's heavy merge traffic).
 2. Global bucket merges have high priority and will most likely be allowed
    to enter the merge throttler queues, displacing lower priority merges.

3 files changed, 39 insertions, 5 deletions

diff --git a/storage/src/tests/distributor/mergeoperationtest.cpp b/storage/src/tests/distributor/mergeoperationtest.cpp
index 7a90299bfb3..1026ea2855e 100644
--- a/storage/src/tests/distributor/mergeoperationtest.cpp
+++ b/storage/src/tests/distributor/mergeoperationtest.cpp
@@ -315,7 +315,7 @@ TEST_F(MergeOperationTest, allow_deleting_active_source_only_replica) {
               _sender.getLastCommand(true));
 }
 
-TEST_F(MergeOperationTest, MarkRedundantTrustedCopiesAsSourceOnly) {
+TEST_F(MergeOperationTest, mark_redundant_trusted_copies_as_source_only) {
     // This test uses the same distribution as testGenerateNodeList(), i.e.
     // an ideal state sequence of [3, 5, 7, 6, 8, 0, 9, 2, 1, 4]
 
@@ -415,6 +415,21 @@ TEST_F(MergeOperationTest, merge_operation_is_blocked_by_any_busy_target_node) {
     EXPECT_TRUE(op.isBlocked(*_pendingTracker, _operation_sequencer));
 }
 
+
+TEST_F(MergeOperationTest, global_bucket_merges_are_not_blocked_by_busy_nodes) {
+    getClock().setAbsoluteTimeInSeconds(10);
+    document::BucketId bucket_id(16, 1);
+    addNodesToBucketDB(bucket_id, "0=10/1/1/t,1=20/1/1,2=10/1/1/t");
+    enableDistributorClusterState("distributor:1 storage:3");
+    document::Bucket global_bucket(document::FixedBucketSpaces::global_space(), bucket_id);
+    MergeOperation op(BucketAndNodes(global_bucket, toVector<uint16_t>(0, 1, 2)));
+    op.setIdealStateManager(&getIdealStateManager());
+
+    // Node 1 is included in operation node set but should not cause a block of global bucket merge
+    _pendingTracker->getNodeInfo().setBusy(0, std::chrono::seconds(10));
+    EXPECT_FALSE(op.isBlocked(*_pendingTracker, _operation_sequencer));
+}
+
 TEST_F(MergeOperationTest, merge_operation_is_blocked_by_locked_bucket) {
     getClock().setAbsoluteTimeInSeconds(10);
     addNodesToBucketDB(document::BucketId(16, 1), "0=10/1/1/t,1=20/1/1,2=10/1/1/t");
diff --git a/storage/src/vespa/storage/distributor/operations/idealstate/mergeoperation.cpp b/storage/src/vespa/storage/distributor/operations/idealstate/mergeoperation.cpp
index 481506096eb..27e203a9060 100644
--- a/storage/src/vespa/storage/distributor/operations/idealstate/mergeoperation.cpp
+++ b/storage/src/vespa/storage/distributor/operations/idealstate/mergeoperation.cpp
@@ -1,5 +1,6 @@
 // Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 #include "mergeoperation.h"
+#include <vespa/document/bucket/fixed_bucket_spaces.h>
 #include <vespa/storage/distributor/idealstatemanager.h>
 #include <vespa/storage/distributor/distributor_bucket_space.h>
 #include <vespa/storage/distributor/pendingmessagetracker.h>
@@ -325,13 +326,30 @@ bool MergeOperation::shouldBlockThisOperation(uint32_t messageType, uint8_t pri)
 
 bool MergeOperation::isBlocked(const PendingMessageTracker& pending_tracker,
                                const OperationSequencer& op_seq) const {
-    const auto& node_info = pending_tracker.getNodeInfo();
-    for (auto node : getNodes()) {
-        if (node_info.isBusy(node)) {
-            return true;
+    // To avoid starvation of high priority global bucket merges, we do not consider
+    // these for blocking due to a node being "busy" (usually caused by a full merge
+    // throttler queue).
+    //
+    // This is for two reasons:
+    //  1. When an ideal state op is blocked, it is still removed from the internal
+    //     maintenance priority queue. This means a blocked high pri operation will
+    //     not be retried until the next DB pass (at which point the node is likely
+    //     to still be marked as busy when there's heavy merge traffic).
+    //  2. Global bucket merges have high priority and will most likely be allowed
+    //     to enter the merge throttler queues, displacing lower priority merges.
+    if (!is_global_bucket_merge()) {
+        const auto& node_info = pending_tracker.getNodeInfo();
+        for (auto node : getNodes()) {
+            if (node_info.isBusy(node)) {
+                return true;
+            }
         }
     }
     return IdealStateOperation::isBlocked(pending_tracker, op_seq);
 }
 
+bool MergeOperation::is_global_bucket_merge() const noexcept {
+    return getBucket().getBucketSpace() == document::FixedBucketSpaces::global_space();
+}
+
 }
diff --git a/storage/src/vespa/storage/distributor/operations/idealstate/mergeoperation.h b/storage/src/vespa/storage/distributor/operations/idealstate/mergeoperation.h
index 5df9421e815..11b5494fd9b 100644
--- a/storage/src/vespa/storage/distributor/operations/idealstate/mergeoperation.h
+++ b/storage/src/vespa/storage/distributor/operations/idealstate/mergeoperation.h
@@ -61,6 +61,7 @@ private:
 
     void deleteSourceOnlyNodes(const BucketDatabase::Entry& currentState,
                                DistributorStripeMessageSender& sender);
+    bool is_global_bucket_merge() const noexcept;
 };
 
 }