Handle create-flag during Put write repair

2 files changed, 47 insertions, 4 deletions

diff --git a/storage/src/tests/distributor/putoperationtest.cpp b/storage/src/tests/distributor/putoperationtest.cpp
index 70e986d0316..405f25f9359 100644
--- a/storage/src/tests/distributor/putoperationtest.cpp
+++ b/storage/src/tests/distributor/putoperationtest.cpp
@@ -118,7 +118,7 @@ public:
         return reply;
     }
 
-    void set_up_tas_put_with_2_inconsistent_replica_nodes();
+    void set_up_tas_put_with_2_inconsistent_replica_nodes(bool create = false);
 
 };
 
@@ -677,7 +677,7 @@ TEST_F(PutOperationTest, minority_failure_override_not_in_effect_for_non_tas_err
               _sender.getLastReply());
 }
 
-void PutOperationTest::set_up_tas_put_with_2_inconsistent_replica_nodes() {
+void PutOperationTest::set_up_tas_put_with_2_inconsistent_replica_nodes(bool create) {
     setup_stripe(Redundancy(2), NodeCount(2), "version:1 storage:2 distributor:1");
     config_enable_condition_probing(true);
     tag_content_node_supports_condition_probing(0, true);
@@ -689,6 +689,7 @@ void PutOperationTest::set_up_tas_put_with_2_inconsistent_replica_nodes() {
 
     auto put = createPut(doc);
     put->setCondition(TestAndSetCondition("test.foo"));
+    put->set_create_if_non_existent(create);
     sendPut(std::move(put));
     ASSERT_EQ("Get => 1,Get => 0", _sender.getCommands(true));
 }
@@ -779,4 +780,45 @@ TEST_F(PutOperationTest, create_flag_in_parent_put_is_propagated_to_sent_puts) {
     EXPECT_TRUE(put_n0->get_create_if_non_existent());
 }
 
+TEST_F(PutOperationTest, not_found_condition_probe_with_create_set_acts_as_if_matched) {
+    ASSERT_NO_FATAL_FAILURE(set_up_tas_put_with_2_inconsistent_replica_nodes(true));
+
+    op->receive(_sender, make_get_reply(*sent_get_command(0), 0, false, false));
+    op->receive(_sender, make_get_reply(*sent_get_command(1), 0, false, false));
+
+    EXPECT_EQ(_sender.replies().size(), 0);
+    ASSERT_EQ("Get => 1,Get => 0,Put => 1,Put => 0", _sender.getCommands(true));
+
+    auto put_n1 = sent_put_command(2);
+    EXPECT_FALSE(put_n1->hasTestAndSetCondition());
+    auto put_n0 = sent_put_command(3);
+    EXPECT_FALSE(put_n0->hasTestAndSetCondition());
+}
+
+TEST_F(PutOperationTest, conditional_put_no_replicas_case_with_create_set_acts_as_if_matched) {
+    setup_stripe(Redundancy(2), NodeCount(2), "version:1 storage:2 distributor:1");
+    config_enable_condition_probing(true);
+    tag_content_node_supports_condition_probing(0, true);
+    tag_content_node_supports_condition_probing(1, true);
+
+    // Don't init the DB with any replicas for the put's target bucket
+    auto put = createPut(createDummyDocument("test", "test"));
+    put->setCondition(TestAndSetCondition("test.foo"));
+    put->set_create_if_non_existent(true);
+    sendPut(std::move(put));
+
+    EXPECT_EQ(_sender.replies().size(), 0);
+    // Pipelined create + put
+    ASSERT_EQ("Create bucket => 1,Create bucket => 0,Put => 1,Put => 0", _sender.getCommands(true));
+
+    // In this case we can preserve both the condition + create-flag.
+    // The content node will deal with them appropriately.
+    auto put_n1 = sent_put_command(2);
+    EXPECT_TRUE(put_n1->hasTestAndSetCondition());
+    EXPECT_TRUE(put_n1->get_create_if_non_existent());
+    auto put_n0 = sent_put_command(3);
+    EXPECT_TRUE(put_n0->hasTestAndSetCondition());
+    EXPECT_TRUE(put_n1->get_create_if_non_existent());
+}
+
 }
diff --git a/storage/src/vespa/storage/distributor/operations/external/putoperation.cpp b/storage/src/vespa/storage/distributor/operations/external/putoperation.cpp
index 9c8602781e6..96252c37844 100644
--- a/storage/src/vespa/storage/distributor/operations/external/putoperation.cpp
+++ b/storage/src/vespa/storage/distributor/operations/external/putoperation.cpp
@@ -277,11 +277,12 @@ void
 PutOperation::on_completed_check_condition(const CheckCondition::Outcome& outcome,
                                            DistributorStripeMessageSender& sender)
 {
-    if (outcome.matched_condition()) {
+    const bool effectively_matched = (outcome.matched_condition()
+                                      || (outcome.not_found() && _msg->get_create_if_non_existent()));
+    if (effectively_matched) {
         _msg->clear_condition(); // Transform to unconditional Put
         start_direct_put_dispatch(sender);
     } else if (outcome.not_found()) {
-        // TODO create:true combined with not found
         // TODO "not found" not a TaS error...
         _tracker.fail(sender, api::ReturnCode(api::ReturnCode::TEST_AND_SET_CONDITION_FAILED,
                                               "Document does not exist"));