deleteWithMeta to manage XATTR documents

When deleting a document with XATTRs, system XATTRs are retained.

Change-Id: Icde8ac48466359ee57b946b6aea39b66466990ad
Reviewed-on: http://review.couchbase.org/76478
Reviewed-by: Dave Rigby <daver@couchbase.com>
Tested-by: Build Bot <build@couchbase.com>

diff --git a/src/vbucket.cc b/src/vbucket.cc
index 306948a..cdac653 100644 (file)
--- a/src/vbucket.cc
+++ b/src/vbucket.cc
@@ -39,6 +39,9 @@
 
 #include "vbucket.h"
 
+#include <xattr/blob.h>
+#include <xattr/utils.h>
+
 /* Macros */
 const size_t MIN_CHK_FLUSH_TIMEOUT = 10; // 10 sec.
 const size_t MAX_CHK_FLUSH_TIMEOUT = 30; // 30 sec.
@@ -1246,13 +1249,22 @@ ENGINE_ERROR_CODE VBucket::deleteWithMeta(const DocKey& key,
                                    TrackCasDrift::Yes,
                                    backfill,
                                    nullptr /* No pre link step needed */);
-        std::tie(delrv, v, notifyCtx) = processSoftDelete(hbl,
-                                                          *v,
-                                                          cas,
-                                                          itemMeta,
-                                                          queueItmCtx,
-                                                          /*use_meta*/ true,
-                                                          bySeqno);
+
+        // system xattrs must remain
+        std::unique_ptr<Item> itm;
+        if (mcbp::datatype::is_xattr(v->getDatatype()) &&
+            (itm = pruneXattrDocument(*v, itemMeta))) {
+            std::tie(v, delrv, notifyCtx) =
+                    updateStoredValue(hbl, *v, *itm, &queueItmCtx);
+        } else {
+            std::tie(delrv, v, notifyCtx) = processSoftDelete(hbl,
+                                                              *v,
+                                                              cas,
+                                                              itemMeta,
+                                                              queueItmCtx,
+                                                              /*use_meta*/ true,
+                                                              bySeqno);
+        }
     }
     cas = v ? v->getCas() : 0;
 
@@ -2354,3 +2366,44 @@ void VBucket::adjustCheckpointFlushTimeout(size_t wall_time) {
 size_t VBucket::getCheckpointFlushTimeout() {
     return chkFlushTimeout;
 }
+
+std::unique_ptr<Item> VBucket::pruneXattrDocument(
+        StoredValue& v, const ItemMetaData& itemMeta) {
+    // Need to take a copy of the value, prune it, and add it back
+
+    // Create work-space document
+    std::vector<uint8_t> workspace(v.getValue()->vlength());
+    std::copy_n(v.getValue()->getData(),
+                v.getValue()->vlength(),
+                workspace.begin());
+
+    // Now attach to the XATTRs in the document
+    auto sz = cb::xattr::get_body_offset(
+            {reinterpret_cast<char*>(workspace.data()), workspace.size()});
+
+    cb::xattr::Blob xattr({workspace.data(), sz});
+    xattr.prune_user_keys();
+
+    auto prunedXattrs = xattr.finalize();
+
+    if (prunedXattrs.size()) {
+        // Something remains - Create a Blob and copy-in just the XATTRs
+        auto newValue =
+                Blob::New(reinterpret_cast<const char*>(prunedXattrs.data()),
+                          prunedXattrs.size(),
+                          const_cast<uint8_t*>(reinterpret_cast<const uint8_t*>(
+                                  v.getValue()->getExtMeta())),
+                          v.getValue()->getExtLen());
+
+        return std::make_unique<Item>(v.getKey(),
+                                      itemMeta.flags,
+                                      itemMeta.exptime,
+                                      newValue,
+                                      itemMeta.cas,
+                                      v.getBySeqno(),
+                                      getId(),
+                                      itemMeta.revSeqno);
+    } else {
+        return {};
+    }
+}

diff --git a/src/vbucket.h b/src/vbucket.h
index 8dba942..c755e7c 100644 (file)
--- a/src/vbucket.h
+++ b/src/vbucket.h
@@ -1520,6 +1520,20 @@ private:
 
     void adjustCheckpointFlushTimeout(size_t wall_time);
 
+    /**
+     * Given a StoredValue with XATTRs - prune the user keys so only system keys
+     * remain.
+     *
+     * @param v StoredValue with XATTR value
+     * @param itemMeta New ItemMetaData to use in item creation
+     * @return unique_ptr<Item> which matches the StoredValue's meta-data and
+     *         has the XATTR value with only the system-keys. If the pruning
+     *         removed all keys (because no system-keys exist) an empty
+     *         unique_ptr is returned.
+     */
+    std::unique_ptr<Item> pruneXattrDocument(StoredValue& v,
+                                             const ItemMetaData& itemMeta);
+
     id_type                         id;
     std::atomic<vbucket_state_t>    state;
     cb::RWLock                      stateLock;

diff --git a/tests/module_tests/evp_store_with_meta.cc b/tests/module_tests/evp_store_with_meta.cc
index ed8c1ed..de2e732 100644 (file)
--- a/tests/module_tests/evp_store_with_meta.cc
+++ b/tests/module_tests/evp_store_with_meta.cc
@@ -16,7 +16,13 @@
  *   limitations under the License.
  */
 
+#include "bgfetcher.h"
 #include "evp_store_single_threaded_test.h"
+#include "tests/mock/mock_global_task.h"
+
+#include <string_utilities.h>
+#include <xattr/blob.h>
+#include <xattr/utils.h>
 
 class WithMetaTest : public SingleThreadedEPBucketTest {
 public:
@@ -977,6 +983,114 @@ TEST_P(AllWithMetaTest, markJSON) {
     delete result.getValue();
 }
 
+// Test uses an XATTR body that has 1 system key (see createXattrValue)
+TEST_F(WithMetaTest, xattrPruneUserKeysOnDelete1) {
+    auto value = createXattrValue(R"({"json":"yesplease"})");
+    ItemMetaData itemMeta{1, 1, 0, expiry};
+    std::string mykey = "mykey";
+    DocKey key{mykey, DocNamespace::DefaultCollection};
+    auto swm = buildWithMetaPacket(PROTOCOL_BINARY_CMD_SET_WITH_META,
+                                   PROTOCOL_BINARY_DATATYPE_XATTR,
+                                   vbid /*vbucket*/,
+                                   0 /*opaque*/,
+                                   0 /*cas*/,
+                                   itemMeta,
+                                   mykey,
+                                   value);
+    EXPECT_EQ(ENGINE_SUCCESS,
+              callEngine(PROTOCOL_BINARY_CMD_SET_WITH_META, swm));
+    EXPECT_EQ(1, store->flushVBucket(vbid));
+
+    itemMeta.revSeqno++; // make delete succeed
+    auto dwm = buildWithMeta(
+            PROTOCOL_BINARY_CMD_DEL_WITH_META, itemMeta, mykey, {});
+    EXPECT_EQ(ENGINE_SUCCESS,
+              callEngine(PROTOCOL_BINARY_CMD_DEL_WITH_META, dwm));
+
+    EXPECT_EQ(1, store->flushVBucket(vbid));
+
+    auto options = get_options_t(QUEUE_BG_FETCH | GET_DELETED_VALUE);
+    auto result = store->get(key, vbid, nullptr, options);
+    ASSERT_EQ(ENGINE_SUCCESS, result.getStatus());
+
+    // Now reconstruct a XATTR Blob and validate the user keys are gone
+    // These code relies on knowing what createXattrValue generates.
+    auto sz = cb::xattr::get_body_offset(
+            {result.getValue()->getData(), result.getValue()->getNBytes()});
+
+    auto p = reinterpret_cast<const uint8_t*>(result.getValue()->getData());
+    cb::xattr::Blob blob({const_cast<uint8_t*>(p), sz});
+
+    EXPECT_EQ(0, blob.get("user").size());
+    EXPECT_EQ(0, blob.get("meta").size());
+    ASSERT_NE(0, blob.get("_sync").size());
+    EXPECT_STREQ("{\"cas\":\"0xdeadbeefcafefeed\"}",
+                 reinterpret_cast<char*>(blob.get("_sync").data()));
+
+    auto itm = result.getValue();
+    // The meta-data should match the delete_with_meta
+    EXPECT_EQ(itemMeta.cas, itm->getCas());
+    EXPECT_EQ(itemMeta.flags, itm->getFlags());
+    EXPECT_EQ(itemMeta.revSeqno, itm->getRevSeqno());
+    EXPECT_EQ(itemMeta.exptime, itm->getExptime());
+
+    delete result.getValue();
+}
+
+// Test uses an XATTR body that has no system keys
+TEST_F(WithMetaTest, xattrPruneUserKeysOnDelete2) {
+    cb::xattr::Blob blob;
+
+    // No _ prefixed keys
+    blob.set(to_const_byte_buffer("user"),
+             to_const_byte_buffer("{\"author\":\"bubba\"}"));
+    blob.set(to_const_byte_buffer("meta"),
+             to_const_byte_buffer("{\"content-type\":\"text\"}"));
+
+    auto xattrValue = blob.finalize();
+
+    // append body to the xattrs and store in data
+    std::string body = "document_body";
+    std::string data;
+    std::copy_n(xattrValue.buf, xattrValue.len, std::back_inserter(data));
+    std::copy_n(body.c_str(), body.size(), std::back_inserter(data));
+
+    ItemMetaData itemMeta{1, 1, 0, expiry};
+    std::string mykey = "mykey";
+    DocKey key{mykey, DocNamespace::DefaultCollection};
+    auto swm = buildWithMetaPacket(PROTOCOL_BINARY_CMD_SET_WITH_META,
+                                   PROTOCOL_BINARY_DATATYPE_XATTR,
+                                   vbid /*vbucket*/,
+                                   0 /*opaque*/,
+                                   0 /*cas*/,
+                                   itemMeta,
+                                   mykey,
+                                   data);
+    EXPECT_EQ(ENGINE_SUCCESS,
+              callEngine(PROTOCOL_BINARY_CMD_SET_WITH_META, swm));
+    EXPECT_EQ(1, store->flushVBucket(vbid));
+
+    itemMeta.revSeqno++; // make delete succeed
+    auto dwm = buildWithMeta(
+            PROTOCOL_BINARY_CMD_DEL_WITH_META, itemMeta, mykey, {});
+    EXPECT_EQ(ENGINE_SUCCESS,
+              callEngine(PROTOCOL_BINARY_CMD_DEL_WITH_META, dwm));
+
+    EXPECT_EQ(1, store->flushVBucket(vbid));
+
+    auto options = get_options_t(QUEUE_BG_FETCH | GET_DELETED_VALUE);
+    auto result = store->get(key, vbid, nullptr, options);
+    EXPECT_EQ(ENGINE_EWOULDBLOCK, result.getStatus());
+
+    // Run the BGFetcher task
+    MockGlobalTask mockTask(engine->getTaskable(), TaskId::MultiBGFetcherTask);
+    store->getVBucket(vbid)->getShard()->getBgFetcher()->run(&mockTask);
+
+    // K/V is gone
+    result = store->get(key, vbid, nullptr, options);
+    EXPECT_EQ(ENGINE_KEY_ENOENT, result.getStatus());
+}
+
 auto opcodeValues = ::testing::Values(PROTOCOL_BINARY_CMD_SET_WITH_META,
                                       PROTOCOL_BINARY_CMD_SETQ_WITH_META,
                                       PROTOCOL_BINARY_CMD_ADD_WITH_META,