MB-41989: document_pre_expiry() operates on a copy

Expiration removes the UserXattrs (if any) from the item's payload.
Currently we modify the shared blob pointed by the item in the HT
and CM, which is incorrect as changes may reflect on the pre-existing
mutation being persisted, replicated and exposed on the frontend.

Change-Id: I5fd2d3d039a4f0949f6e823592513098b457b09d
Reviewed-on: http://review.couchbase.org/c/kv_engine/+/137972
Tested-by: Build Bot <[email protected]>
Reviewed-by: Trond Norbye <[email protected]>
Well-Formed: Build Bot <[email protected]>

Author: paolococchi

daemon/doc_pre_expiry.cc

@@ -30,17 +30,20 @@ std::string document_pre_expiry(const item_info& itm_info) {
         return {};
     }
 
+    // Operate on a copy
     cb::char_buffer payload{static_cast<char*>(itm_info.value[0].iov_base),
                             itm_info.value[0].iov_len};
+    cb::xattr::Blob originalBlob(payload,
+                                 mcbp::datatype::is_snappy(itm_info.datatype));
+    auto copy = cb::xattr::Blob(originalBlob);
+    copy.prune_user_keys();
+    const auto final = copy.finalize();
 
-    cb::xattr::Blob blob(payload, mcbp::datatype::is_snappy(itm_info.datatype));
-    blob.prune_user_keys();
-    auto pruned = blob.finalize();
-    if (pruned.size() == 0) {
+    if (final.size() == 0) {
         // The old payload only contained user xattrs and
         // we removed everything
         return {};
     }
 
-    return {pruned.data(), pruned.size()};
+    return {final.data(), final.size()};
 }

engines/ep/tests/module_tests/dcp_stream_test.cc

@@ -3334,6 +3334,16 @@ void SingleThreadedActiveStreamTest::testExpirationRemovesBody(uint32_t flags,
     GetValue gv = store->get(docKey, vbid, nullptr, get_options_t::NONE);
     EXPECT_EQ(ENGINE_KEY_ENOENT, gv.getStatus());
 
+    // MB-41989: Expiration removes UserXattrs (if any), but it must do that on
+    // a copy of the payload that is then enqueued in the new expired item. So,
+    // the payload of the original mutation must be untouched here.
+    // Note: 'it' still points to the original mutation in the first checkpoint
+    // in CM.
+    EXPECT_EQ(queue_op::mutation, (*it)->getOperation());
+    EXPECT_FALSE((*it)->isDeleted());
+    EXPECT_EQ(cb::const_char_buffer(value.c_str(), value.size()),
+              cb::const_char_buffer((*it)->getData(), (*it)->getNBytes()));
+
     ASSERT_EQ(2, list.size());
     ckpt = list.back().get();
     ASSERT_EQ(checkpoint_state::CHECKPOINT_OPEN, ckpt->getState());