MB-31141: Don't reject snappy|raw DCP deletes

A related bug means that is possible for 5.x to create
deletes with a non-zero raw value. When 5.5x backfills such
an item for transmission to another 5.5x node (and snappy
is enabled), the delete gets sent with a snappy datatype
and rejected by the target node, which triggers a rebalance
failure.

Change-Id: Ib91453f96882ef4e01ee0e2a3e5e51ed0786a325
Reviewed-on: http://review.couchbase.org/99414
Well-Formed: Build Bot <[email protected]>
Reviewed-by: Dave Rigby <[email protected]>
Tested-by: Dave Rigby <[email protected]>

Author: jimwwalker

daemon/mcbp_validators.cc

@@ -389,8 +389,12 @@ static bool valid_dcp_delete_datatype(protocol_binary_datatype_t datatype) {
     // it may send XATTR|JSON (with snappy possible). These are now allowed
     // so rebalance won't be failed and the consumer will sanitise the faulty
     // documents.
-    std::array<const protocol_binary_datatype_t, 5> valid = {
+    // MB-31141: Allowing RAW+Snappy. A bug in delWithMeta has allowed us to
+    // create deletes with a non-zero value tagged as RAW, which when snappy
+    // is enabled gets DCP shipped as RAW+Snappy.
+    std::array<const protocol_binary_datatype_t, 6> valid = {
             {PROTOCOL_BINARY_RAW_BYTES,
+             PROTOCOL_BINARY_RAW_BYTES | PROTOCOL_BINARY_DATATYPE_SNAPPY,
              PROTOCOL_BINARY_DATATYPE_XATTR,
              PROTOCOL_BINARY_DATATYPE_XATTR | PROTOCOL_BINARY_DATATYPE_SNAPPY,
              PROTOCOL_BINARY_DATATYPE_XATTR | PROTOCOL_BINARY_DATATYPE_JSON,

engines/ep/src/dcp/consumer.cc

@@ -574,10 +574,16 @@ ENGINE_ERROR_CODE DcpConsumer::deletion(uint32_t opaque,
         // 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
-        if (mcbp::datatype::is_xattr(datatype) && value.size()) {
-            auto vb = engine_.getVBucket(vbucket);
-            if (vb) {
-                engine_.getKVBucket()->runPreExpiryHook(*vb, *item);
+        if (value.size()) {
+            if (mcbp::datatype::is_xattr(datatype)) {
+                auto vb = engine_.getVBucket(vbucket);
+                if (vb) {
+                    engine_.getKVBucket()->runPreExpiryHook(*vb, *item);
+                }
+            } else {
+                // MB-31141: Deletes cannot have a value
+                item->replaceValue(Blob::New(0));
+                item->setDataType(PROTOCOL_BINARY_RAW_BYTES);
             }
         }
 

engines/ep/tests/module_tests/dcp_test.cc

@@ -2966,28 +2966,25 @@ TEST_P(ConnectionTest, test_mb24424_deleteResponse) {
     ASSERT_TRUE(stream->isActive());
 
     std::string key = "key";
-    std::string data = R"({"json":"yes"})";
     const DocKey docKey{ reinterpret_cast<const uint8_t*>(key.data()),
         key.size(),
         DocNamespace::DefaultCollection};
-    cb::const_byte_buffer value{reinterpret_cast<const uint8_t*>(data.data()),
-        data.size()};
     uint8_t extMeta[1] = {uint8_t(PROTOCOL_BINARY_DATATYPE_JSON)};
     cb::const_byte_buffer meta{extMeta, sizeof(uint8_t)};
 
-    consumer->deletion(/*opaque*/1,
-                       /*key*/docKey,
-                       /*values*/value,
-                       /*priv_bytes*/0,
-                       /*datatype*/PROTOCOL_BINARY_DATATYPE_JSON,
-                       /*cas*/0,
-                       /*vbucket*/vbid,
-                       /*bySeqno*/1,
-                       /*revSeqno*/0,
-                       /*meta*/meta);
-
-    auto messageSize = MutationResponse::deletionBaseMsgBytes +
-            key.size() + data.size() + sizeof(extMeta);
+    consumer->deletion(/*opaque*/ 1,
+                       /*key*/ docKey,
+                       /*value*/ {},
+                       /*priv_bytes*/ 0,
+                       /*datatype*/ PROTOCOL_BINARY_RAW_BYTES,
+                       /*cas*/ 0,
+                       /*vbucket*/ vbid,
+                       /*bySeqno*/ 1,
+                       /*revSeqno*/ 0,
+                       /*meta*/ meta);
+
+    auto messageSize = MutationResponse::deletionBaseMsgBytes + key.size() +
+                       sizeof(extMeta);
 
     EXPECT_EQ(messageSize, stream->responseMessageSize);
 

engines/ep/tests/module_tests/evp_store_single_threaded_test.cc

@@ -1613,16 +1613,13 @@ TEST_F(SingleThreadedEPBucketTest, MB_29861) {
                              /*startseq*/ 0,
                              /*endseq*/ 2,
                              /*flags*/ MARKER_FLAG_DISK);
-    std::string data = R"({"json":"yes"})";
-    cb::const_byte_buffer value{reinterpret_cast<const uint8_t*>(data.data()),
-                                data.size()};
 
     // 2. Now add a deletion.
     consumer->deletion(/*opaque*/ 1,
                        {"key1", DocNamespace::DefaultCollection},
-                       /*values*/ value,
+                       /*value*/ {},
                        /*priv_bytes*/ 0,
-                       /*datatype*/ PROTOCOL_BINARY_DATATYPE_JSON,
+                       /*datatype*/ PROTOCOL_BINARY_RAW_BYTES,
                        /*cas*/ 0,
                        /*vbucket*/ vbid,
                        /*bySeqno*/ 1,
@@ -1659,7 +1656,7 @@ TEST_F(SingleThreadedEPBucketTest, MB_29861) {
                                  deleted,
                                  datatype));
     EXPECT_EQ(1, deleted);
-    EXPECT_EQ(PROTOCOL_BINARY_DATATYPE_JSON, datatype);
+    EXPECT_EQ(PROTOCOL_BINARY_RAW_BYTES, datatype);
     EXPECT_NE(0, metadata.exptime); // A locally created deleteTime
 }
 
@@ -1684,15 +1681,12 @@ TEST_F(SingleThreadedEPBucketTest, MB_27457) {
                              /*startseq*/ 0,
                              /*endseq*/ 2,
                              /*flags*/ 0);
-    std::string data = R"({"json":"yes"})";
-    cb::const_byte_buffer value{reinterpret_cast<const uint8_t*>(data.data()),
-                                data.size()};
     // 2. Now add two deletions, one without deleteTime, one with
     consumer->deletionV2(/*opaque*/ 1,
                          {"key1", DocNamespace::DefaultCollection},
-                         /*values*/ value,
+                         /*values*/ {},
                          /*priv_bytes*/ 0,
-                         /*datatype*/ PROTOCOL_BINARY_DATATYPE_JSON,
+                         /*datatype*/ PROTOCOL_BINARY_RAW_BYTES,
                          /*cas*/ 0,
                          /*vbucket*/ vbid,
                          /*bySeqno*/ 1,
@@ -1702,9 +1696,9 @@ TEST_F(SingleThreadedEPBucketTest, MB_27457) {
     const uint32_t deleteTime = 10;
     consumer->deletionV2(/*opaque*/ 1,
                          {"key2", DocNamespace::DefaultCollection},
-                         /*values*/ value,
+                         /*value*/ {},
                          /*priv_bytes*/ 0,
-                         /*datatype*/ PROTOCOL_BINARY_DATATYPE_JSON,
+                         /*datatype*/ PROTOCOL_BINARY_RAW_BYTES,
                          /*cas*/ 0,
                          /*vbucket*/ vbid,
                          /*bySeqno*/ 2,
@@ -1741,7 +1735,7 @@ TEST_F(SingleThreadedEPBucketTest, MB_27457) {
                                  deleted,
                                  datatype));
     EXPECT_EQ(1, deleted);
-    EXPECT_EQ(PROTOCOL_BINARY_DATATYPE_JSON, datatype);
+    EXPECT_EQ(PROTOCOL_BINARY_RAW_BYTES, datatype);
     EXPECT_NE(0, metadata.exptime); // A locally created deleteTime
 
     deleted = 0;
@@ -1762,7 +1756,7 @@ TEST_F(SingleThreadedEPBucketTest, MB_27457) {
                                  deleted,
                                  datatype));
     EXPECT_EQ(1, deleted);
-    EXPECT_EQ(PROTOCOL_BINARY_DATATYPE_JSON, datatype);
+    EXPECT_EQ(PROTOCOL_BINARY_RAW_BYTES, datatype);
     EXPECT_EQ(deleteTime, metadata.exptime); // Our replicated deleteTime!
 }
 
@@ -3037,6 +3031,69 @@ TEST_P(XattrCompressedTest, MB_29040_sanitise_input) {
               gv.item->getDataType());
 }
 
+// Create a replica VB and consumer, then send it an delete with value which
+// should never of been created on the source.
+TEST_F(SingleThreadedEPBucketTest, MB_31141_sanitise_input) {
+    setVBucketStateAndRunPersistTask(vbid, vbucket_state_replica);
+
+    auto consumer = std::make_shared<MockDcpConsumer>(
+            *engine, cookie, "MB_31141_sanitise_input");
+    int opaque = 1;
+    ASSERT_EQ(ENGINE_SUCCESS, consumer->addStream(opaque, vbid, /*flags*/ 0));
+
+    std::string body = "value";
+
+    // Send deletion in a single seqno snapshot
+    int64_t bySeqno = 1;
+    EXPECT_EQ(ENGINE_SUCCESS,
+              consumer->snapshotMarker(
+                      opaque, vbid, bySeqno, bySeqno, MARKER_FLAG_CHK));
+
+    EXPECT_EQ(ENGINE_SUCCESS,
+              consumer->deletion(opaque,
+                                 {"key", DocNamespace::DefaultCollection},
+                                 {reinterpret_cast<const uint8_t*>(body.data()),
+                                  body.size()},
+                                 /*priv_bytes*/ 0,
+                                 PROTOCOL_BINARY_DATATYPE_SNAPPY |
+                                         PROTOCOL_BINARY_RAW_BYTES,
+                                 /*cas*/ 3,
+                                 vbid,
+                                 bySeqno,
+                                 /*revSeqno*/ 0,
+                                 /*meta*/ {}));
+
+    EXPECT_EQ(std::make_pair(false, size_t(1)),
+              getEPBucket().flushVBucket(vbid));
+
+    ASSERT_EQ(ENGINE_SUCCESS, consumer->closeStream(opaque, vbid));
+
+    // Switch to active
+    setVBucketStateAndRunPersistTask(vbid, vbucket_state_active);
+
+    get_options_t options = static_cast<get_options_t>(
+            QUEUE_BG_FETCH | HONOR_STATES | TRACK_REFERENCE | DELETE_TEMP |
+            HIDE_LOCKED_CAS | TRACK_STATISTICS | GET_DELETED_VALUE);
+    auto gv = store->get(
+            {"key", DocNamespace::DefaultCollection}, vbid, cookie, options);
+    EXPECT_EQ(ENGINE_EWOULDBLOCK, gv.getStatus());
+
+    // Manually run the bgfetch task.
+    MockGlobalTask mockTask(engine->getTaskable(), TaskId::MultiBGFetcherTask);
+    store->getVBucket(vbid)->getShard()->getBgFetcher()->run(&mockTask);
+    gv = store->get({"key", DocNamespace::DefaultCollection},
+                    vbid,
+                    cookie,
+                    GET_DELETED_VALUE);
+    ASSERT_EQ(ENGINE_SUCCESS, gv.getStatus());
+
+    EXPECT_TRUE(gv.item->isDeleted());
+    EXPECT_EQ(0, gv.item->getFlags());
+    EXPECT_EQ(3, gv.item->getCas());
+    EXPECT_EQ(0, gv.item->getValue()->valueSize());
+    EXPECT_EQ(PROTOCOL_BINARY_RAW_BYTES, gv.item->getDataType());
+}
+
 // Test highlighting MB_29480 - this is not demonstrating the issue is fixed.
 TEST_F(SingleThreadedEPBucketTest, MB_29480) {
     // Make vbucket active.

tests/mcbp/mcbp_test.cc

@@ -1942,8 +1942,9 @@ TEST_P(DcpDeletionValidatorTest, InvalidMagic) {
 
 TEST_P(DcpDeletionValidatorTest, ValidDatatype) {
     using cb::mcbp::Datatype;
-    const std::array<uint8_t, 3> datatypes = {
+    const std::array<uint8_t, 4> datatypes = {
             {uint8_t(Datatype::Raw),
+             uint8_t(Datatype::Raw) | uint8_t(Datatype::Snappy),
              uint8_t(Datatype::Xattr),
              uint8_t(Datatype::Xattr) | uint8_t(Datatype::Snappy)}};
 
@@ -1956,9 +1957,8 @@ TEST_P(DcpDeletionValidatorTest, ValidDatatype) {
 
 TEST_P(DcpDeletionValidatorTest, InvalidDatatype) {
     using cb::mcbp::Datatype;
-    const std::array<uint8_t, 3> datatypes = {
+    const std::array<uint8_t, 2> datatypes = {
             {uint8_t(Datatype::JSON),
-             uint8_t(Datatype::Snappy),
              uint8_t(Datatype::Snappy) | uint8_t(Datatype::JSON)}};
 
     for (auto invalid : datatypes) {