MB-60066: Tolerate DCP deletions with value larger than 1MiB

Subdoc CreateWithDelete can legitimately create a user xattr that
exceeds 1MiB, yet the DCP validation expects deletes to only come
with <= 1MiB of value.

Change the incorrect validation, now matches dcp_mutation.cc checks

Change-Id: I9863495db104ca10f8d651f61ee68a844055a027
Reviewed-on: https://review.couchbase.org/c/kv_engine/+/202507
Well-Formed: Restriction Checker
Reviewed-by: Trond Norbye <[email protected]>
Tested-by: Jim Walker <[email protected]>

Author: jimwwalker

cluster_framework/dcp_replicator.cc

@@ -182,7 +182,8 @@ void DcpReplicatorImpl::createPipesForNode(const Cluster& cluster,
              cb::mcbp::Feature::XERROR,
              cb::mcbp::Feature::SELECT_BUCKET,
              cb::mcbp::Feature::SNAPPY,
-             cb::mcbp::Feature::JSON}};
+             cb::mcbp::Feature::JSON,
+             cb::mcbp::Feature::Collections}};
 
     for (size_t node = 0; node < vbids.size(); ++node) {
         if (vbids[node].empty()) {
@@ -216,7 +217,8 @@ void DcpReplicatorImpl::createPipeForNodes(const Cluster& cluster,
              cb::mcbp::Feature::XERROR,
              cb::mcbp::Feature::SELECT_BUCKET,
              cb::mcbp::Feature::SNAPPY,
-             cb::mcbp::Feature::JSON}};
+             cb::mcbp::Feature::JSON,
+             cb::mcbp::Feature::Collections}};
 
     if (vbids[specific.consumer].empty()) {
         // I don't have any connections towards this node
@@ -240,7 +242,12 @@ void DcpReplicatorImpl::createDcpPipe(const Cluster& cluster,
 
     // Create and send a DCP open
     auto rsp = connection->execute(BinprotDcpOpenCommand{
-            name, cb::mcbp::request::DcpOpenPayload::Producer});
+            name,
+            cb::mcbp::request::DcpOpenPayload::Producer |
+                    cb::mcbp::request::DcpOpenPayload::IncludeXattrs |
+                    cb::mcbp::request::DcpOpenPayload::
+                            IncludeDeletedUserXattrs |
+                    cb::mcbp::request::DcpOpenPayload::IncludeDeleteTimes});
     if (!rsp.isSuccess()) {
         throw std::runtime_error(
                 "DcpReplicatorImpl::start: Failed to set up "

daemon/protocol/mcbp/dcp_deletion.cc

@@ -16,6 +16,15 @@
 #include <mcbp/protocol/request.h>
 #include <memcached/limits.h>
 #include <memcached/protocol_binary.h>
+#include <xattr/blob.h>
+
+static bool invalidXattrSize(cb::const_byte_buffer value,
+                             protocol_binary_datatype_t datatype) {
+    const char* payload = reinterpret_cast<const char*>(value.data());
+    cb::xattr::Blob blob({const_cast<char*>(payload), value.size()},
+                         mcbp::datatype::is_snappy(datatype));
+    return blob.get_system_size() > cb::limits::PrivilegedBytes;
+}
 
 static cb::engine_errc dcp_deletion_v1_executor(Cookie& cookie) {
     auto& request = cookie.getHeader().getRequest();
@@ -41,24 +50,22 @@ static cb::engine_errc dcp_deletion_v1_executor(Cookie& cookie) {
     cb::const_byte_buffer meta{value.data() + value.size() - nmeta, nmeta};
     value = {value.data(), value.size() - nmeta};
 
-    uint32_t priv_bytes = 0;
-    if (mcbp::datatype::is_xattr(datatype)) {
-        priv_bytes = gsl::narrow<uint32_t>(value.size());
-    }
-    if (priv_bytes <= cb::limits::PrivilegedBytes) {
-        return dcpDeletion(cookie,
-                           opaque,
-                           key,
-                           value,
-                           priv_bytes,
-                           datatype,
-                           cas,
-                           vbucket,
-                           by_seqno,
-                           rev_seqno,
-                           meta);
+    if (mcbp::datatype::is_xattr(datatype) &&
+        invalidXattrSize(value, datatype)) {
+        return cb::engine_errc::too_big;
     }
-    return cb::engine_errc::too_big;
+
+    return dcpDeletion(cookie,
+                       opaque,
+                       key,
+                       value,
+                       0, // @todo: remove priv_bytes. unused.
+                       datatype,
+                       cas,
+                       vbucket,
+                       by_seqno,
+                       rev_seqno,
+                       meta);
 }
 
 // The updated deletion sends no extended meta, but does send a deletion time
@@ -86,25 +93,22 @@ static cb::engine_errc dcp_deletion_v2_executor(Cookie& cookie) {
     const uint32_t delete_time = payload.getDeleteTime();
 
     auto value = request.getValue();
-    uint32_t priv_bytes = 0;
-    if (mcbp::datatype::is_xattr(datatype)) {
-        priv_bytes = gsl::narrow<uint32_t>(value.size());
+    if (mcbp::datatype::is_xattr(datatype) &&
+        invalidXattrSize(value, datatype)) {
+        return cb::engine_errc::too_big;
     }
 
-    if (priv_bytes <= cb::limits::PrivilegedBytes) {
-        return dcpDeletionV2(cookie,
-                             opaque,
-                             key,
-                             value,
-                             priv_bytes,
-                             datatype,
-                             cas,
-                             vbucket,
-                             by_seqno,
-                             rev_seqno,
-                             delete_time);
-    }
-    return cb::engine_errc::too_big;
+    return dcpDeletionV2(cookie,
+                         opaque,
+                         key,
+                         value,
+                         0, // @todo: remove this unused parameter
+                         datatype,
+                         cas,
+                         vbucket,
+                         by_seqno,
+                         rev_seqno,
+                         delete_time);
 }
 
 void dcp_deletion_executor(Cookie& cookie) {

tests/testapp_cluster/durability_tests.cc

@@ -12,6 +12,7 @@
 
 #include <cluster_framework/bucket.h>
 #include <cluster_framework/cluster.h>
+#include <memcached/limits.h>
 #include <protocol/connection/client_connection.h>
 #include <protocol/connection/client_mcbp_commands.h>
 #include <protocol/connection/frameinfo.h>
@@ -246,6 +247,43 @@ TEST_F(DurabilityTest, SyncWriteReviveDeletedDocument) {
     EXPECT_EQ(cb::mcbp::Status::Success, resp.getStatus());
 }
 
+// Verify that we can create (and replicate) a large xattr, exceed the PrivBytes
+TEST_F(DurabilityTest, CreateAsDeletedLarge) {
+    BinprotSubdocMultiMutationCommand cmd;
+    std::string name = "CreateAsDeletedLarge";
+    // Ensure we exceed priv bytes. Just using this size is enough when we add
+    // the key
+    std::string value(cb::limits::PrivilegedBytes, 'v');
+    cmd.setKey(name);
+    cmd.addDocFlag(mcbp::subdoc::doc_flag::Add);
+    cmd.addDocFlag(mcbp::subdoc::doc_flag::CreateAsDeleted);
+    cmd.addMutation(cb::mcbp::ClientOpcode::SubdocDictUpsert,
+                    SUBDOC_FLAG_XATTR_PATH | SUBDOC_FLAG_MKDIR_P,
+                    "tnx.foo",
+                    "\"" + value + "\"");
+    auto conn = getConnection();
+    conn->sendCommand(cmd);
+
+    BinprotSubdocMultiMutationResponse resp;
+    conn->recvResponse(resp);
+    ASSERT_EQ(cb::mcbp::Status::SubdocSuccessDeleted, resp.getStatus());
+
+    // Do a sync-write after to ensure the test replicates the previous big
+    // xattr
+    cmd = {};
+    cmd.setKey(name);
+    cmd.addDocFlag(mcbp::subdoc::doc_flag::AccessDeleted);
+    cmd.addFrameInfo(DurabilityFrameInfo{cb::durability::Level::Majority});
+    cmd.addDocFlag(mcbp::subdoc::doc_flag::ReviveDocument);
+    cmd.addMutation(cb::mcbp::ClientOpcode::SubdocDictUpsert,
+                    SUBDOC_FLAG_XATTR_PATH,
+                    "tnx.bar",
+                    R"("This should succeed")");
+    conn->sendCommand(cmd);
+    conn->recvResponse(resp);
+    EXPECT_EQ(cb::mcbp::Status::Success, resp.getStatus());
+}
+
 /**
  * MB-34780 - Bucket delete fails if we've got pending sync writes
  *