Inhibit scheduling of merges towards nodes that are marked busy

Utilizes existing maintenance operation scheduler system which checks
if an operation is considered blocked and does not start it if this
is the case. We now block a merge operation if any of the nodes in its
node set are marked busy by the pending message tracker.

Duration for which nodes are marked busy is live-configurable.

1 files changed, 25 insertions, 0 deletions

diff --git a/storage/src/tests/distributor/mergeoperationtest.cpp b/storage/src/tests/distributor/mergeoperationtest.cpp
index b298bec3977..7e7ca52635e 100644
--- a/storage/src/tests/distributor/mergeoperationtest.cpp
+++ b/storage/src/tests/distributor/mergeoperationtest.cpp
@@ -28,6 +28,7 @@ class MergeOperationTest : public CppUnit::TestFixture,
     CPPUNIT_TEST(testMarkRedundantTrustedCopiesAsSourceOnly);
     CPPUNIT_TEST(onlyMarkRedundantRetiredReplicasAsSourceOnly);
     CPPUNIT_TEST(mark_post_merge_redundant_replicas_source_only);
+    CPPUNIT_TEST(merge_operation_is_blocked_by_any_busy_target_node);
     CPPUNIT_TEST_SUITE_END();
 
     std::unique_ptr<PendingMessageTracker> _pendingTracker;
@@ -41,6 +42,7 @@ protected:
     void testMarkRedundantTrustedCopiesAsSourceOnly();
     void onlyMarkRedundantRetiredReplicasAsSourceOnly();
     void mark_post_merge_redundant_replicas_source_only();
+    void merge_operation_is_blocked_by_any_busy_target_node();
 
 public:
     void setUp() override {
@@ -478,5 +480,28 @@ void MergeOperationTest::mark_post_merge_redundant_replicas_source_only() {
             getNodeList("storage:10", 4, "3,5,7,6"));
 }
 
+void MergeOperationTest::merge_operation_is_blocked_by_any_busy_target_node() {
+    getClock().setAbsoluteTimeInSeconds(10);
+    addNodesToBucketDB(document::BucketId(16, 1), "0=10/1/1/t,1=20/1/1,2=10/1/1/t");
+    _distributor->enableClusterState(lib::ClusterState("distributor:1 storage:3"));
+    MergeOperation op(BucketAndNodes(document::BucketId(16, 1), toVector<uint16_t>(0, 1, 2)));
+    op.setIdealStateManager(&getIdealStateManager());
+
+    // Should not block on nodes _not_ included in operation node set
+    _pendingTracker->getNodeInfo().setBusy(3, std::chrono::seconds(10));
+    CPPUNIT_ASSERT(!op.isBlocked(*_pendingTracker));
+
+    // Node 1 is included in operation node set and should cause a block
+    _pendingTracker->getNodeInfo().setBusy(0, std::chrono::seconds(10));
+    CPPUNIT_ASSERT(op.isBlocked(*_pendingTracker));
+
+    getClock().addSecondsToTime(11);
+    CPPUNIT_ASSERT(!op.isBlocked(*_pendingTracker)); // No longer busy
+
+    // Should block on other operation nodes than the first listed as well
+    _pendingTracker->getNodeInfo().setBusy(1, std::chrono::seconds(10));
+    CPPUNIT_ASSERT(op.isBlocked(*_pendingTracker));
+}
+
 } // distributor
 } // storage