Merge pull request #16104 from vespa-engine/geirst/reject-two-phase-update-op-in-distributor

Reject two phase update operation in safe path if feed in cluster is …

3 files changed, 37 insertions, 5 deletions

diff --git a/storage/src/tests/distributor/twophaseupdateoperationtest.cpp b/storage/src/tests/distributor/twophaseupdateoperationtest.cpp
index 1518c8594aa..924678a6cd0 100644
--- a/storage/src/tests/distributor/twophaseupdateoperationtest.cpp
+++ b/storage/src/tests/distributor/twophaseupdateoperationtest.cpp
@@ -151,6 +151,12 @@ struct TwoPhaseUpdateOperationTest : Test, DistributorTestUtil {
         return cb;
     }
 
+    void set_up_distributor_with_feed_blocked_state() {
+        setup_distributor(2, 2,
+                          lib::ClusterStateBundle(lib::ClusterState("distributor:1 storage:2"),
+                                                  {}, {true, "full disk"}, false));
+    }
+
 };
 
 TwoPhaseUpdateOperationTest::TwoPhaseUpdateOperationTest() = default;
@@ -1091,6 +1097,17 @@ TEST_F(TwoPhaseUpdateOperationTest, update_gets_are_sent_with_strong_consistency
     EXPECT_EQ(get_cmd.internal_read_consistency(), api::InternalReadConsistency::Strong);
 }
 
+TEST_F(TwoPhaseUpdateOperationTest, operation_is_rejected_in_safe_path_if_feed_is_blocked) {
+    set_up_distributor_with_feed_blocked_state();
+    auto cb = sendUpdate("0=1/2/3,1=2/3/4"); // Inconsistent replicas to trigger safe path
+    cb->start(_sender, framework::MilliSecTime(0));
+
+    EXPECT_EQ("UpdateReply(id:ns:testdoctype1::1, BucketId(0x0000000000000000), "
+              "timestamp 0, timestamp of updated doc: 0) "
+              "ReturnCode(NO_SPACE, External feed is blocked due to resource exhaustion: full disk)",
+              _sender.getLastReply(true));
+}
+
 struct ThreePhaseUpdateTest : TwoPhaseUpdateOperationTest {};
 
 TEST_F(ThreePhaseUpdateTest, metadata_only_gets_are_sent_if_3phase_update_enabled) {
diff --git a/storage/src/vespa/storage/distributor/operations/external/twophaseupdateoperation.cpp b/storage/src/vespa/storage/distributor/operations/external/twophaseupdateoperation.cpp
index 80e7942c68e..362acdf18ec 100644
--- a/storage/src/vespa/storage/distributor/operations/external/twophaseupdateoperation.cpp
+++ b/storage/src/vespa/storage/distributor/operations/external/twophaseupdateoperation.cpp
@@ -1,16 +1,17 @@
 // Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 
-#include "twophaseupdateoperation.h"
 #include "getoperation.h"
 #include "putoperation.h"
+#include "twophaseupdateoperation.h"
 #include "updateoperation.h"
-#include <vespa/storage/distributor/distributor_bucket_space.h>
-#include <vespa/storage/distributor/distributor_bucket_space_repo.h>
-#include <vespa/storageapi/message/persistence.h>
 #include <vespa/document/datatype/documenttype.h>
+#include <vespa/document/fieldset/fieldsets.h>
 #include <vespa/document/fieldvalue/document.h>
 #include <vespa/document/select/parser.h>
-#include <vespa/document/fieldset/fieldsets.h>
+#include <vespa/storage/distributor/distributor_bucket_space.h>
+#include <vespa/storage/distributor/distributor_bucket_space_repo.h>
+#include <vespa/storageapi/message/persistence.h>
+#include <vespa/vdslib/state/cluster_state_bundle.h>
 #include <vespa/vespalib/stllike/hash_map.hpp>
 
 #include <vespa/log/log.h>
@@ -197,6 +198,10 @@ TwoPhaseUpdateOperation::startFastPathUpdate(DistributorMessageSender& sender, s
 void
 TwoPhaseUpdateOperation::startSafePathUpdate(DistributorMessageSender& sender)
 {
+    if (_op_ctx.cluster_state_bundle().block_feed_in_cluster()) {
+        send_feed_blocked_error_reply(sender);
+        return;
+    }
     _mode = Mode::SLOW_PATH;
     auto get_operation = create_initial_safe_path_get_operation();
     GetOperation& op = *get_operation;
@@ -279,6 +284,15 @@ TwoPhaseUpdateOperation::sendLostOwnershipTransientErrorReply(DistributorMessage
 }
 
 void
+TwoPhaseUpdateOperation::send_feed_blocked_error_reply(DistributorMessageSender& sender)
+{
+    sendReplyWithResult(sender,
+                        api::ReturnCode(api::ReturnCode::NO_SPACE,
+                                        "External feed is blocked due to resource exhaustion: " +
+                                        _op_ctx.cluster_state_bundle().feed_block()->description()));
+}
+
+void
 TwoPhaseUpdateOperation::schedulePutsWithUpdatedDocument(std::shared_ptr<document::Document> doc,
                                                          api::Timestamp putTimestamp, DistributorMessageSender& sender)
 {
diff --git a/storage/src/vespa/storage/distributor/operations/external/twophaseupdateoperation.h b/storage/src/vespa/storage/distributor/operations/external/twophaseupdateoperation.h
index af45932b530..d353498c8e3 100644
--- a/storage/src/vespa/storage/distributor/operations/external/twophaseupdateoperation.h
+++ b/storage/src/vespa/storage/distributor/operations/external/twophaseupdateoperation.h
@@ -104,6 +104,7 @@ private:
     void startSafePathUpdate(DistributorMessageSender&);
     bool lostBucketOwnershipBetweenPhases() const;
     void sendLostOwnershipTransientErrorReply(DistributorMessageSender&);
+    void send_feed_blocked_error_reply(DistributorMessageSender& sender);
     void schedulePutsWithUpdatedDocument(
             std::shared_ptr<document::Document>,
             api::Timestamp,