MB-34346: Allow pruning of compressed xattrs

The API which prunes documents of non-system xattrs can trigger an
exception when the incoming value is compressed. The  expects that the
final (pruned) value will fit into the input buffer, if not an
exception occurs.

This exception can be made to trigger when the incoming buffer
contains a snappy compressed value. The code decompresses the value
and prunes the xattrs, then fails because in some cases the
decompressed and pruned value is larger than the buffer it arrived in.

This is made safe by changing the API so that we don't re-use the
input buffer for output data, instead a new buffer is returned, which
is empty except in the case when a modified/pruned value is to be
returned.

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

Author: jimwwalker

daemon/doc_pre_expiry.cc

@@ -23,11 +23,11 @@
 
 #include <gsl/gsl>
 
-bool document_pre_expiry(item_info& itm_info) {
+std::string document_pre_expiry(const item_info& itm_info) {
     if (!mcbp::datatype::is_xattr(itm_info.datatype)) {
         // The object does not contain any XATTRs so we should remove
         // the entire content
-        return false;
+        return {};
     }
 
     cb::char_buffer payload{static_cast<char*>(itm_info.value[0].iov_base),
@@ -36,32 +36,11 @@ bool document_pre_expiry(item_info& itm_info) {
     cb::xattr::Blob blob(payload, mcbp::datatype::is_snappy(itm_info.datatype));
     blob.prune_user_keys();
     auto pruned = blob.finalize();
-    if (pruned.len == 0) {
+    if (pruned.size() == 0) {
         // The old payload only contained user xattrs and
         // we removed everything
-        return false;
+        return {};
     }
 
-    // Pruning the user keys should just repack the data internally
-    // without any allocations, but to be on the safe side we should
-    // just check to verify, and if it happened to have done any
-    // reallocations we could copy the data into our buffer if it fits.
-    // (or we could throw an exception here, but I'd hate to get that
-    // 2AM call if we can avoid that with a simple fallback check)
-    if (pruned.buf != payload.buf) {
-        if (pruned.len > payload.len) {
-           throw std::logic_error("pre_expiry_document: the pruned object "
-                                  "won't fit!");
-        }
-
-        std::copy(pruned.buf, pruned.buf + pruned.len, payload.buf);
-    }
-
-    // Update the length field of the item
-    itm_info.nbytes = gsl::narrow<uint32_t>(pruned.len);
-    itm_info.value[0].iov_len = pruned.len;
-    // Clear all other datatype flags (we've stripped off everything)
-    itm_info.datatype = PROTOCOL_BINARY_DATATYPE_XATTR;
-
-    return true;
+    return {pruned.data(), pruned.size()};
 }

daemon/doc_pre_expiry.h

@@ -23,4 +23,4 @@
  * Implementation of the pre_expiry hook in the `SERVER_DOCUMENT_API`. Check
  * the server API documentation for more information.
  */
-bool document_pre_expiry(item_info& itm_info);
+std::string document_pre_expiry(const item_info& itm_info);

engines/ep/src/kv_bucket.cc

@@ -542,11 +542,12 @@ void KVBucket::runPreExpiryHook(VBucket& vb, Item& it) {
     it.decompressValue(); // A no-op for already decompressed items
     auto info =
             it.toItemInfo(vb.failovers->getLatestUUID(), vb.getHLCEpochSeqno());
-    if (engine.getServerApi()->document->pre_expiry(info)) {
-        // The payload is modified and contains data we should use
-        it.replaceValue(Blob::New(static_cast<char*>(info.value[0].iov_base),
-                                  info.value[0].iov_len));
-        it.setDataType(info.datatype);
+    auto result = engine.getServerApi()->document->pre_expiry(info);
+    if (!result.empty()) {
+        // A modified value was returned, use it
+        it.replaceValue(Blob::New(result.data(), result.size()));
+        // The API states only uncompressed xattr values are returned
+        it.setDataType(PROTOCOL_BINARY_DATATYPE_XATTR);
     } else {
         // Make the document empty and raw
         it.replaceValue(Blob::New(0));

engines/ep/src/vbucket.cc

@@ -480,21 +480,19 @@ void VBucket::handlePreExpiry(const std::unique_lock<std::mutex>& hbl,
         EventuallyPersistentEngine* engine = ObjectRegistry::getCurrentEngine();
         itm_info =
                 itm->toItemInfo(failovers->getLatestUUID(), getHLCEpochSeqno());
-        value_t new_val(Blob::Copy(*value));
-        itm->replaceValue(new_val.get());
-        itm->setDataType(v.getDatatype());
 
         SERVER_HANDLE_V1* sapi = engine->getServerApi();
         /* TODO: In order to minimize allocations, the callback needs to
          * allocate an item whose value size will be exactly the size of the
          * value after pre-expiry is performed.
          */
-        if (sapi->document->pre_expiry(itm_info)) {
+        auto result = sapi->document->pre_expiry(itm_info);
+        if (!result.empty()) {
             Item new_item(v.getKey(),
                           v.getFlags(),
                           v.getExptime(),
-                          itm_info.value[0].iov_base,
-                          itm_info.value[0].iov_len,
+                          result.data(),
+                          result.size(),
                           itm_info.datatype,
                           v.getCas(),
                           v.getBySeqno(),

engines/ep/tests/module_tests/kv_bucket_test.cc

@@ -1252,6 +1252,50 @@ TEST_P(KVBucketParamTest, MB31495_GetRandomKey) {
     EXPECT_EQ(ENGINE_SUCCESS, gv.getStatus());
 }
 
+// Test that expiring a compressed xattr doesn't trigger any errors
+TEST_P(KVBucketParamTest, MB_34346) {
+    // Create an XTTR value with only a large system xattr, and compress the lot
+    // Note the large xattr should be highly compressible to make it easier to
+    // trigger the MB.
+    cb::xattr::Blob blob;
+    blob.set("_sync",
+             R"({"fffff":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"})");
+    auto xattr = blob.finalize();
+    cb::compression::Buffer output;
+    cb::compression::deflate(cb::compression::Algorithm::Snappy,
+                             {xattr.data(), xattr.size()},
+                             output);
+    EXPECT_LT(output.size(), xattr.size())
+            << "Expected the compressed buffer to be smaller than the input";
+
+    auto key = makeStoredDocKey("key_1");
+    store_item(
+            vbid,
+            key,
+            {output.data(), output.size()},
+            ep_abs_time(ep_current_time() + 10),
+            {cb::engine_errc::success},
+            PROTOCOL_BINARY_DATATYPE_XATTR | PROTOCOL_BINARY_DATATYPE_SNAPPY);
+
+    flushVBucketToDiskIfPersistent(vbid, 1);
+
+    EXPECT_EQ(1, engine->getVBucket(vbid)->getNumItems())
+            << "Should have 1 item after calling store()";
+
+    TimeTraveller docBrown(15);
+
+    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);
+    GetValue gv2 = store->get(key, vbid, cookie, options);
+    EXPECT_TRUE(gv2.item->isDeleted());
+
+    flushVBucketToDiskIfPersistent(vbid, 1);
+
+    EXPECT_EQ(0, engine->getVBucket(vbid)->getNumItems())
+            << "Should still have 0 items after time-travelling/expiry";
+}
+
 class StoreIfTest : public KVBucketTest {
 public:
     void SetUp() override {

include/memcached/server_api.h

@@ -317,17 +317,17 @@ struct SERVER_DOCUMENT_API {
 
     /**
      * This callback is called from the underlying engine right before
-     * a particular document expires. The callback is responsible for
-     * modifying the contents of the itm_info passed in. The updated
-     * total size is available in itm_info.nbytes.
+     * a particular document expires. The callback is responsible examining
+     * the value and possibly returning a new and modified value.
      *
      * @param itm_info info pertaining to the item that is to be expired.
-     * @return true indicating that the info has been modified in itm_info.
-     *         false indicating that there is no data available in itm_info.
+     * @return std::string empty if the value required no modification, not
+     *         empty then the string contains the modified value. When not empty
+     *         the datatype of the new value is datatype xattr only.
+     *
      * @throws std::bad_alloc in case of memory allocation failure
-     * @throws std::logic_error if the data has grown
      */
-    bool (*pre_expiry)(item_info& itm_info);
+    std::string (*pre_expiry)(const item_info& itm_info);
 };
 
 #ifdef WIN32

tests/doc_server_api/doc_pre_expiry_test.cc

@@ -24,7 +24,7 @@
 
 TEST(PreExpiry, EmptyDocument) {
     item_info info{};
-    EXPECT_FALSE(document_pre_expiry(info));
+    EXPECT_TRUE(document_pre_expiry(info).empty());
 }
 
 TEST(PreExpiry, DocumentWithoutXattr) {
@@ -33,8 +33,8 @@ TEST(PreExpiry, DocumentWithoutXattr) {
     info.value[0].iov_base = static_cast<void*>(blob);
     info.value[0].iov_len = sizeof(blob);
     info.nbytes = sizeof(blob);
-    EXPECT_FALSE(document_pre_expiry(info));
-    EXPECT_EQ(PROTOCOL_BINARY_RAW_BYTES, info.datatype);
+    auto rv = document_pre_expiry(info);
+    EXPECT_TRUE(rv.empty());
 }
 
 TEST(PreExpiry, DocumentWithUserXAttrOnly) {
@@ -46,8 +46,8 @@ TEST(PreExpiry, DocumentWithUserXAttrOnly) {
     info.value[0].iov_len = body.size();
     info.nbytes = gsl::narrow<uint32_t>(body.size());
     info.datatype = PROTOCOL_BINARY_DATATYPE_XATTR;
-    EXPECT_FALSE(document_pre_expiry(info));
-    EXPECT_EQ(PROTOCOL_BINARY_DATATYPE_XATTR, info.datatype);
+    auto rv = document_pre_expiry(info);
+    EXPECT_TRUE(rv.empty());
 }
 
 TEST(PreExpiry, DocumentWithSystemXattrOnly) {
@@ -59,8 +59,8 @@ TEST(PreExpiry, DocumentWithSystemXattrOnly) {
     info.value[0].iov_len = body.size();
     info.nbytes = gsl::narrow<uint32_t>(body.size());
     info.datatype = PROTOCOL_BINARY_DATATYPE_XATTR;
-    EXPECT_TRUE(document_pre_expiry(info));
-    EXPECT_EQ(PROTOCOL_BINARY_DATATYPE_XATTR, info.datatype);
+    auto rv = document_pre_expiry(info);
+    EXPECT_FALSE(rv.empty());
 }
 
 TEST(PreExpiry, DocumentWithUserAndSystemXattr) {
@@ -73,11 +73,9 @@ TEST(PreExpiry, DocumentWithUserAndSystemXattr) {
     info.value[0].iov_len = body.size();
     info.nbytes = gsl::narrow<uint32_t>(body.size());
     info.datatype = PROTOCOL_BINARY_DATATYPE_XATTR;
-    EXPECT_TRUE(document_pre_expiry(info));
-    EXPECT_EQ(PROTOCOL_BINARY_DATATYPE_XATTR, info.datatype);
-
-    // The body should have been modified (shrinked by stripping off user attr)
-    EXPECT_GT(body.size(), info.datatype);
+    auto rv = document_pre_expiry(info);
+    EXPECT_FALSE(rv.empty());
+    EXPECT_LT(rv.size(), body.size());
 }
 
 TEST(PreExpiry, DocumentWithJsonBodyAndXattrs) {
@@ -95,10 +93,7 @@ TEST(PreExpiry, DocumentWithJsonBodyAndXattrs) {
     info.nbytes = gsl::narrow<uint32_t>(body.size());
     info.datatype =
             PROTOCOL_BINARY_DATATYPE_XATTR | PROTOCOL_BINARY_DATATYPE_JSON;
-    EXPECT_TRUE(document_pre_expiry(info));
-    // The JSON datatype should be stripped off
-    EXPECT_EQ(PROTOCOL_BINARY_DATATYPE_XATTR, info.datatype);
-
-    // The body should have been modified (shrinked by stripping off user attr)
-    EXPECT_GT(body.size(), info.datatype);
+    auto rv = document_pre_expiry(info);
+    EXPECT_FALSE(rv.empty());
+    EXPECT_LT(rv.size(), body.size());
 }