MB-43205: Consumer enforces allowSanitizeValueInDeletion at DCP_DELETE

At DCP_DELETE, if the sanitizer is enabled then the consumer removes any
invalid body in the payload. ENGINE_EINVAL is returned otherwise.

Change-Id: I6ac22952221b130d6b14a0acb072dc293f27d0db
Reviewed-on: http://review.couchbase.org/c/kv_engine/+/143712
Well-Formed: Build Bot <[email protected]>
Reviewed-by: Dave Rigby <[email protected]>
Tested-by: Build Bot <[email protected]>

Author: paolococchi

engines/ep/src/dcp/consumer.cc

@@ -590,9 +590,9 @@ ENGINE_ERROR_CODE DcpConsumer::deletion(uint32_t opaque,
                                            vbucket,
                                            revSeqno));
 
-    // MB-37374: 6.6 Producers may legally send user-xattrs in deletion.
-    // The existing validation for pre-6.6 Producers is still necessary.
     if (includeDeletedUserXattrs == IncludeDeletedUserXattrs::No) {
+        // Case pre-6.6 connection: Body and UserXattrs are invalid in deletion
+
         // MB-29040: Producer may send deleted doc with value that still has
         // the user xattrs and the body. Fix up that mistake by running the
         // expiry hook which will correctly process the document
@@ -609,16 +609,28 @@ ENGINE_ERROR_CODE DcpConsumer::deletion(uint32_t opaque,
             }
         }
     } else {
-        // Some validation seems reasonable here, given that in the past we
-        // already had issues in this area (see if-block above).
+        // Case 6.6 connection: UserXattrs in deletion is legal since MB-37374,
+        // Body still invalid.
+
         if (cb::xattr::get_body_size(
                     datatype,
                     {reinterpret_cast<const char*>(value.data()),
                      value.size()}) > 0) {
-            logger->warn(
-                    "DcpConsumer::deletion: ({}) Value cannot contain a body",
-                    vbucket);
-            return ENGINE_EINVAL;
+            // MB-43205 shows that we cannot unconditionally fail here as 6.6
+            // connections may still see Body in deletions (generated by pre-6.6
+            // nodes) after an offline upgrade.
+            // Note: (allowSanitizeValueInDeletion = true) by default, disabling
+            // the sanitizer is left only for test purpose.
+
+            if (!allowSanitizeValueInDeletion) {
+                logger->error(
+                        "DcpConsumer::deletion: ({}) Value cannot contain a "
+                        "body",
+                        vbucket);
+                return ENGINE_EINVAL;
+            }
+
+            item->removeBody();
         }
     }
 

engines/ep/src/dcp/consumer.h

@@ -332,6 +332,10 @@ class DcpConsumer : public ConnHandler,
         allowSanitizeValueInDeletion.store(value);
     }
 
+    bool isAllowSanitizeValueInDeletion() {
+        return allowSanitizeValueInDeletion.load();
+    }
+
 protected:
     /**
      * Records when the consumer last received a message from producer.

engines/ep/tests/module_tests/dcp_stream_test.cc

@@ -3447,8 +3447,14 @@ TEST_P(SingleThreadedActiveStreamTest, testExpirationRemovesBody_UserXa_SysXa) {
                               Xattrs::UserAndSys);
 }
 
-void SingleThreadedPassiveStreamTest::testConsumerRejectsBodyInDelete(
+void SingleThreadedPassiveStreamTest::testConsumerRejectsBodyInDeletion(
         const boost::optional<cb::durability::Requirements>& durReqs) {
+    auto& connMap = static_cast<MockDcpConnMap&>(engine->getDcpConnMap());
+    connMap.addConn(cookie, consumer);
+    ASSERT_TRUE(consumer->isAllowSanitizeValueInDeletion());
+    engine->getConfiguration().setAllowSanitizeValueInDeletion(false);
+    ASSERT_FALSE(consumer->isAllowSanitizeValueInDeletion());
+
     consumer->public_setIncludeDeletedUserXattrs(IncludeDeletedUserXattrs::Yes);
 
     // Send deletion in a single seqno snapshot
@@ -3462,7 +3468,8 @@ void SingleThreadedPassiveStreamTest::testConsumerRejectsBodyInDelete(
                                        {} /*maxVisibleSeqno*/));
 
     const auto verifyDCPFailure =
-            [this, &durReqs](const cb::const_byte_buffer& value) -> void {
+            [this, &durReqs](const cb::const_byte_buffer& value,
+                             protocol_binary_datatype_t datatype) -> void {
         const uint32_t opaque = 1;
         int64_t bySeqno = 1;
         if (durReqs) {
@@ -3471,7 +3478,7 @@ void SingleThreadedPassiveStreamTest::testConsumerRejectsBodyInDelete(
                                         {"key", DocKeyEncodesCollectionId::No},
                                         value,
                                         0 /*priv_bytes*/,
-                                        PROTOCOL_BINARY_RAW_BYTES,
+                                        datatype,
                                         0 /*cas*/,
                                         vbid,
                                         0 /*flags*/,
@@ -3488,7 +3495,7 @@ void SingleThreadedPassiveStreamTest::testConsumerRejectsBodyInDelete(
                                          {"key", DocKeyEncodesCollectionId::No},
                                          value,
                                          0 /*priv_bytes*/,
-                                         PROTOCOL_BINARY_RAW_BYTES,
+                                         datatype,
                                          0 /*cas*/,
                                          vbid,
                                          bySeqno,
@@ -3503,7 +3510,7 @@ void SingleThreadedPassiveStreamTest::testConsumerRejectsBodyInDelete(
                                 body.size()};
     {
         SCOPED_TRACE("");
-        verifyDCPFailure(value);
+        verifyDCPFailure(value, PROTOCOL_BINARY_RAW_BYTES);
     }
 
     // Verify the same for body + xattrs
@@ -3512,16 +3519,106 @@ void SingleThreadedPassiveStreamTest::testConsumerRejectsBodyInDelete(
              xattrValue.size()};
     {
         SCOPED_TRACE("");
-        verifyDCPFailure(value);
+        verifyDCPFailure(
+                value,
+                PROTOCOL_BINARY_RAW_BYTES | PROTOCOL_BINARY_DATATYPE_XATTR);
     }
+
+    connMap.removeConn(cookie);
+}
+
+TEST_P(SingleThreadedPassiveStreamTest, ConsumerRejectsBodyInDeletion) {
+    testConsumerRejectsBodyInDeletion({});
 }
 
-TEST_P(SingleThreadedPassiveStreamTest, ConsumerRejectsBodyInDelete) {
-    testConsumerRejectsBodyInDelete({});
+TEST_P(SingleThreadedPassiveStreamTest, ConsumerRejectsBodyInSyncDeletion) {
+    testConsumerRejectsBodyInDeletion(cb::durability::Requirements());
+}
+
+void SingleThreadedPassiveStreamTest::testConsumerSanitizesBodyInDeletion() {
+    ASSERT_TRUE(consumer->isAllowSanitizeValueInDeletion());
+    consumer->public_setIncludeDeletedUserXattrs(IncludeDeletedUserXattrs::Yes);
+
+    EXPECT_EQ(ENGINE_SUCCESS,
+              consumer->snapshotMarker(1 /*opaque*/,
+                                       vbid,
+                                       1 /*startSeqno*/,
+                                       10 /*endSeqno*/,
+                                       MARKER_FLAG_CHK,
+                                       {} /*HCS*/,
+                                       {} /*maxVisibleSeqno*/));
+
+    // Build up a value with just raw body and verify that DCP deletion succeeds
+    // and the Body has been removed from the payload.
+    const std::string body = "body";
+    cb::const_byte_buffer value{reinterpret_cast<const uint8_t*>(body.data()),
+                                body.size()};
+    const uint32_t opaque = 1;
+    const auto key = makeStoredDocKey("key");
+    EXPECT_EQ(ENGINE_SUCCESS,
+              consumer->deletion(opaque,
+                                 key,
+                                 value,
+                                 0 /*priv_bytes*/,
+                                 PROTOCOL_BINARY_RAW_BYTES,
+                                 0 /*cas*/,
+                                 vbid,
+                                 1 /*bySeqno*/,
+                                 0 /*revSeqno*/,
+                                 {} /*meta*/));
+    // Verify that the body has been removed
+    auto& vb = *store->getVBucket(vbid);
+    auto& ht = vb.ht;
+    {
+        const auto& res = ht.findForUpdate(key);
+        const auto* sv = res.committed;
+        EXPECT_TRUE(sv);
+        EXPECT_EQ(1, sv->getBySeqno());
+        EXPECT_FALSE(sv->getValue().get());
+    }
+
+    // Verify the same for body + user-xattrs + sys-xattrs
+    const auto xattrValue = createXattrValue(body);
+    value = {reinterpret_cast<const uint8_t*>(xattrValue.data()),
+             xattrValue.size()};
+    EXPECT_EQ(ENGINE_SUCCESS,
+              consumer->deletion(opaque,
+                                 key,
+                                 value,
+                                 0 /*priv_bytes*/,
+                                 PROTOCOL_BINARY_RAW_BYTES |
+                                         PROTOCOL_BINARY_DATATYPE_XATTR,
+                                 0 /*cas*/,
+                                 vbid,
+                                 2 /*bySeqno*/,
+                                 0 /*revSeqno*/,
+                                 {} /*meta*/));
+
+    {
+        const auto& res = ht.findForUpdate(key);
+        const auto* sv = res.committed;
+        EXPECT_TRUE(sv);
+        EXPECT_EQ(2, sv->getBySeqno());
+        EXPECT_TRUE(sv->getValue().get());
+        EXPECT_LT(sv->getValue()->valueSize(), xattrValue.size());
+
+        const auto finalValue =
+                cb::char_buffer(const_cast<char*>(sv->getValue()->getData()),
+                                sv->getValue()->valueSize());
+        // No body
+        EXPECT_EQ(0, cb::xattr::get_body_size(sv->getDatatype(), finalValue));
+        // Must have user/sys xattrs (created at createXattrValue())
+        cb::xattr::Blob blob(finalValue, false /*compressed*/);
+        for (uint8_t i = 1; i <= 6; ++i) {
+            EXPECT_FALSE(blob.get("ABCuser" + std::to_string(i)).empty());
+        }
+        EXPECT_FALSE(blob.get("meta").empty());
+        EXPECT_FALSE(blob.get("_sync").empty());
+    }
 }
 
-TEST_P(SingleThreadedPassiveStreamTest, ConsumerRejectsBodyInSyncDelete) {
-    testConsumerRejectsBodyInDelete(cb::durability::Requirements());
+TEST_P(SingleThreadedPassiveStreamTest, ConsumerSanitizesBodyInDeletion) {
+    testConsumerSanitizesBodyInDeletion();
 }
 
 void SingleThreadedPassiveStreamTest::testConsumerReceivesUserXattrsInDelete(

engines/ep/tests/module_tests/dcp_stream_test.h

@@ -151,13 +151,19 @@ class SingleThreadedPassiveStreamTest
             vbucket_state_t initialState);
 
     /**
-     * Test that the Consumer rejects body in deletion's value
+     * Test that the Consumer rejects body in deletion's value abd fails.
+     * That is the behaviour when (allowSanitizeValueInDeletion = false).
      *
      * @param durReqs (optional) The Dur Reqs, if we are testing SyncDelete
      */
-    void testConsumerRejectsBodyInDelete(
+    void testConsumerRejectsBodyInDeletion(
             const boost::optional<cb::durability::Requirements>& durReqs);
 
+    /**
+     * Test that the Consumer removes the body (if any) in deletion's value.
+     */
+    void testConsumerSanitizesBodyInDeletion();
+
     /**
      * Test that the Consumer accepts user-xattrs in deletion for enabled
      * connections.