MB-40493: ActiveStream doesn't try to inflate an inflated/empty value

paolococchi · paolococchi · commit 05db33e71b96 · 2020-09-28T10:57:56.000Z
An unnecessary warning is logged otherwise. Change-Id: I8416275ee71b1391616dea6bab053e47a2c0e090 Reviewed-on: http://review.couchbase.org/c/kv_engine/+/136710 Tested-by: Build Bot <build@couchbase.com> Reviewed-by: Jim Walker <jim@couchbase.com> Well-Formed: Build Bot <build@couchbase.com>
diff --git a/engines/ep/src/dcp/active_stream.cc b/engines/ep/src/dcp/active_stream.cc
@@ -1057,7 +1057,7 @@ std::unique_ptr<DcpResponse> ActiveStream::makeResponseFromItem(
                              isForceValueCompressionEnabled(),
                              isSnappyEnabled())) {
             auto finalItem = std::make_unique<Item>(*item);
-            finalItem->removeBodyAndOrXattrs(
+            const auto wasInflated = finalItem->removeBodyAndOrXattrs(
                     includeValue, includeXattributes, includeDeletedUserXattrs);
 
             if (isSnappyEnabled()) {
@@ -1072,12 +1072,23 @@ std::unique_ptr<DcpResponse> ActiveStream::makeResponseFromItem(
                     }
                 }
             } else {
-                if (mcbp::datatype::is_snappy(finalItem->getDataType())) {
+                // The purpose of this block is to uncompress compressed items
+                // as they are being streamed over a connection that doesn't
+                // support compression.
+                //
+                // MB-40493: IncludeValue::NoWithUnderlyingDatatype may reset
+                //  datatype to SNAPPY, even if the value has been already
+                //  decompressed (eg, the original value contained Body+Xattr
+                //  and Body have been removed) or if there is no value at all
+                //  (eg, the original value contained only a Body, now removed).
+                //  We need to avoid the call to Item::decompress in both cases,
+                //  we log an unnecessary warning otherwise.
+                if (mcbp::datatype::is_snappy(finalItem->getDataType()) &&
+                    (wasInflated == Item::WasValueInflated::No) &&
+                    (finalItem->getNBytes() > 0)) {
                     if (!finalItem->decompressValue()) {
                         log(spdlog::level::level_enum::warn,
-
-                            "{} Failed to snappy uncompress a compressed "
-                            "value",
+                            "{} Failed to snappy uncompress a compressed value",
                             logPrefix);
                     }
                 }
diff --git a/engines/ep/src/item.cc b/engines/ep/src/item.cc
@@ -405,42 +405,44 @@ item_info Item::toItemInfo(uint64_t vb_uuid, int64_t hlcEpoch) const {
     return info;
 }
 
-void Item::removeBody() {
+Item::WasValueInflated Item::removeBody() {
     if (!value) {
         // No value, nothing to do
-        return;
+        return WasValueInflated::No;
     }
 
     if (!mcbp::datatype::is_xattr(getDataType())) {
         // We don't want the body and there are no xattrs, just set empty value
         setData(nullptr, 0);
         setDataType(PROTOCOL_BINARY_RAW_BYTES);
-        return;
+        return WasValueInflated::No;
     }
 
     // No-op if already uncompressed
-    decompressValue();
+    const auto wasInflated = decompressValue();
 
     // We want only xattrs.
     // Note: The following is no-op if no Body present.
     const cb::const_char_buffer valBuffer{value->getData(), value->valueSize()};
     setData(valBuffer.data(), cb::xattr::get_body_offset(valBuffer));
     setDataType(PROTOCOL_BINARY_DATATYPE_XATTR);
+
+    return wasInflated ? WasValueInflated::Yes : WasValueInflated::No;
 }
 
-void Item::removeXattrs() {
+Item::WasValueInflated Item::removeXattrs() {
     if (!value) {
         // No value, nothing to do
-        return;
+        return WasValueInflated::No;
     }
 
     if (!mcbp::datatype::is_xattr(getDataType())) {
         // No Xattrs, nothing to do
-        return;
+        return WasValueInflated::No;
     }
 
     // No-op if already uncompressed
-    decompressValue();
+    const auto wasInflated = decompressValue();
 
     // We want only the body
     const cb::const_char_buffer valBuffer{value->getData(), value->valueSize()};
@@ -454,21 +456,23 @@ void Item::removeXattrs() {
         // Subdoc logic for details. Here we have to rectify.
         setDataType(getDataType() & ~PROTOCOL_BINARY_DATATYPE_JSON);
     }
+
+    return wasInflated ? WasValueInflated::Yes : WasValueInflated::No;
 }
 
-void Item::removeUserXattrs() {
+Item::WasValueInflated Item::removeUserXattrs() {
     if (!value) {
         // No value, nothing to do
-        return;
+        return WasValueInflated::No;
     }
 
     if (!mcbp::datatype::is_xattr(getDataType())) {
         // No Xattrs, nothing to do
-        return;
+        return WasValueInflated::No;
     }
 
     // No-op if already uncompressed
-    decompressValue();
+    const auto wasInflated = decompressValue();
 
     // The function currently does not support value with body.
     // That is fine for now as this is introduced for MB-37374, thus is supposed
@@ -494,33 +498,37 @@ void Item::removeUserXattrs() {
     // Note: Doing this unconditionally as we reach this line iff there is no
     // body. We would need to do this conditionally otherwise.
     setDataType(getDataType() & ~PROTOCOL_BINARY_DATATYPE_JSON);
+
+    return wasInflated ? WasValueInflated::Yes : WasValueInflated::No;
 }
 
-void Item::removeBodyAndOrXattrs(
+Item::WasValueInflated Item::removeBodyAndOrXattrs(
         IncludeValue includeVal,
         IncludeXattrs includeXattrs,
         IncludeDeletedUserXattrs includeDeletedUserXattrs) {
     if (!value) {
         // If no value (ie, no body and/or xattrs) then nothing to do
-        return;
+        return WasValueInflated::No;
     }
 
     // Take a copy of the original datatype before proceeding, any modification
     // to the value may change the datatype.
     const auto originalDatatype = getDataType();
 
+    auto wasInflated = WasValueInflated::No;
+
     // Note: IncludeValue acts like "include body"
     if (includeVal != IncludeValue::Yes) {
-        removeBody();
+        wasInflated = removeBody();
     }
 
     if (includeXattrs == IncludeXattrs::No) {
-        removeXattrs();
+        wasInflated = removeXattrs();
     }
 
     if (isDeleted() &&
         includeDeletedUserXattrs == IncludeDeletedUserXattrs::No) {
-        removeUserXattrs();
+        wasInflated = removeUserXattrs();
     }
 
     // Datatype for no-value must be RAW
@@ -532,6 +540,8 @@ void Item::removeBodyAndOrXattrs(
     if (includeVal == IncludeValue::NoWithUnderlyingDatatype) {
         setDataType(originalDatatype);
     }
+
+    return wasInflated;
 }
 
 item_info to_item_info(const ItemMetaData& itemMeta,
diff --git a/engines/ep/src/item.h b/engines/ep/src/item.h
@@ -494,26 +494,34 @@ class Item : public RCValue {
      */
     item_info toItemInfo(uint64_t vb_uuid, int64_t hlcEpoch) const;
 
+    enum class WasValueInflated : uint8_t { No, Yes };
+
     /**
      * Remove the Body of this item's value.
      * No-op if no Value or no Body present.
      * Keeps the Xattrs chunk intact, if any.
+     *
+     * @return whether the value has been decompressed for processing
      */
-    void removeBody();
+    WasValueInflated removeBody();
 
     /**
      * Remove the Xattrs chunk of this item's value.
      * No-op if no Value or no Xattr present.
      * Keeps the Body intact, if any.
+     *
+     * @return whether the value has been decompressed for processing
      */
-    void removeXattrs();
+    WasValueInflated removeXattrs();
 
     /**
      * Remove user-xattrs from the Xattrs chunk of this item's value.
      * No-op if no Value or no user-xattr present.
      * Keeps the Body and the sys-xattrs intact, if any.
+     *
+     * @return whether the value has been decompressed for processing
      */
-    void removeUserXattrs();
+    WasValueInflated removeUserXattrs();
 
     /**
      * Removes Body and/or Xattrs from the item depending on the given params
@@ -522,8 +530,10 @@ class Item : public RCValue {
      * @param includeXattrs states whether the item should include xattrs
      * @param includeDeletedUserXattrs states whether a delete item should
      *  include user-xattrs
+     *
+     * @return whether the value has been decompressed for processing
      */
-    void removeBodyAndOrXattrs(
+    WasValueInflated removeBodyAndOrXattrs(
             IncludeValue includeVal,
             IncludeXattrs includeXattrs,
             IncludeDeletedUserXattrs includeDeletedUserXattrs);
diff --git a/engines/ep/tests/mock/mock_dcp_producer.cc b/engines/ep/tests/mock/mock_dcp_producer.cc
@@ -20,13 +20,12 @@
 #include "dcp/active_stream_checkpoint_processor_task.h"
 #include "dcp/msg_producers_border_guard.h"
 #include "dcp/response.h"
+#include "mock_bucket_logger.h"
 #include "mock_dcp.h"
 #include "mock_dcp_backfill_mgr.h"
 #include "mock_stream.h"
 #include "vbucket.h"
 
-#include <folly/portability/GTest.h>
-
 extern cb::mcbp::ClientOpcode last_op;
 
 MockDcpProducer::MockDcpProducer(EventuallyPersistentEngine& theEngine,
@@ -154,3 +153,12 @@ BackfillScanBuffer& MockDcpProducer::public_getBackfillScanBuffer() {
 void MockDcpProducer::bytesForceRead(size_t bytes) {
     backfillMgr->bytesForceRead(bytes);
 }
+
+void MockDcpProducer::setupMockLogger() {
+    logger = std::make_shared<::testing::NiceMock<MockBucketLogger>>("prod");
+}
+
+MockBucketLogger& MockDcpProducer::public_getLogger() const {
+    EXPECT_TRUE(logger);
+    return dynamic_cast<MockBucketLogger&>(*logger);
+}
diff --git a/engines/ep/tests/mock/mock_dcp_producer.h b/engines/ep/tests/mock/mock_dcp_producer.h
@@ -24,6 +24,7 @@
 class ActiveStreamCheckpointProcessorTask;
 struct BackfillScanBuffer;
 class MockActiveStream;
+class MockBucketLogger;
 class MockDcpMessageProducers;
 
 /*
@@ -236,4 +237,8 @@ class MockDcpProducer : public DcpProducer {
     IncludeDeletedUserXattrs public_getIncludeDeletedUserXattrs() const {
         return includeDeletedUserXattrs;
     }
+
+    void setupMockLogger();
+
+    MockBucketLogger& public_getLogger() const;
 };
diff --git a/engines/ep/tests/module_tests/dcp_test.cc b/engines/ep/tests/module_tests/dcp_test.cc
@@ -23,6 +23,7 @@
  */
 
 #include "dcp_test.h"
+#include "../mock/mock_bucket_logger.h"
 #include "../mock/mock_checkpoint_manager.h"
 #include "../mock/mock_dcp.h"
 #include "../mock/mock_dcp_conn_map.h"
@@ -49,6 +50,7 @@
 #include "vbucket.h"
 #include "warmup.h"
 
+#include <folly/portability/GMock.h>
 #include <folly/portability/GTest.h>
 #include <memcached/server_cookie_iface.h>
 #include <platform/cbassert.h>
@@ -698,6 +700,61 @@ TEST_P(CompressionStreamTest, force_value_compression_enabled) {
     destroy_dcp_stream();
 }
 
+TEST_P(CompressionStreamTest,
+       NoWithUnderlyingDatatype_CompressionDisabled_ItemCompressed) {
+    setup_dcp_stream(
+            0, IncludeValue::NoWithUnderlyingDatatype, IncludeXattrs::Yes);
+    ASSERT_FALSE(producer->isCompressionEnabled());
+    ASSERT_EQ(IncludeValue::NoWithUnderlyingDatatype,
+              stream->public_getIncludeValue());
+    ASSERT_EQ(IncludeXattrs::Yes, stream->public_getIncludeXattrs());
+
+    // Create a compressed item
+    auto item = makeCompressibleItem(vbid,
+                                     makeStoredDocKey("key"),
+                                     "body000000000000000000000000000000000000",
+                                     isXattr() ? PROTOCOL_BINARY_DATATYPE_JSON
+                                               : PROTOCOL_BINARY_RAW_BYTES,
+                                     true, // compressed
+                                     isXattr());
+
+    // ActiveStream::makeResponseFromItem is where we modify the item value (if
+    // necessary) before pushing items into the Stream::readyQ. Here we just
+    // pass the item in input to the function and check that we get the expected
+    // DcpResponse.
+
+    // Core expectation of this test: we don't try to inflate an uncompressed or
+    // empty value, which leads to logging a warning. Before the fix, this
+    // expectation fails as GMock intercept 1 call to mlog.
+    producer->setupMockLogger();
+    using namespace testing;
+    EXPECT_CALL(producer->public_getLogger(), mlog(_, _)).Times(0);
+
+    queued_item originalItem(std::move(item));
+    const auto resp = stream->public_makeResponseFromItem(
+            originalItem, SendCommitSyncWriteAs::Commit);
+
+    const auto* mut = dynamic_cast<MutationResponse*>(resp.get());
+    ASSERT_TRUE(mut);
+
+    // Expecting a modified item, new allocation occurred.
+    ASSERT_NE(originalItem.get(), mut->getItem().get());
+
+    const auto originalValueSize = originalItem->getNBytes();
+    ASSERT_GT(originalValueSize, 0);
+    const auto onTheWireValueSize = mut->getItem()->getNBytes();
+
+    if (isXattr()) {
+        // Stream::makeResponseFromItem will have inflated the value for
+        // removing Xattrs, and then not re-compressed as passive compression
+        // is disabled.
+        EXPECT_GT(onTheWireValueSize, originalValueSize);
+    } else {
+        // Body only, which must have been removed.
+        EXPECT_EQ(0, onTheWireValueSize);
+    }
+}
+
 class ConnectionTest : public DCPTest,
                        public ::testing::WithParamInterface<
                                std::tuple<std::string, std::string>> {
diff --git a/engines/ep/tests/module_tests/evp_store_with_meta.cc b/engines/ep/tests/module_tests/evp_store_with_meta.cc
@@ -510,8 +510,11 @@ TEST_F(WithMetaTest, storeUncompressedInOffMode) {
          return;
     }
     std::string valueData{R"({"aaaaaaaaa":10000000000})"};
-    auto item = makeCompressibleItem(vbid, makeStoredDocKey("key"), valueData,
-                                     PROTOCOL_BINARY_RAW_BYTES, true);
+    auto item = makeCompressibleItem(vbid,
+                                     makeStoredDocKey("key"),
+                                     valueData,
+                                     PROTOCOL_BINARY_RAW_BYTES,
+                                     true);
 
     item->setCas();
 
diff --git a/engines/ep/tests/module_tests/test_helpers.cc b/engines/ep/tests/module_tests/test_helpers.cc
@@ -77,27 +77,35 @@ queued_item makePendingItem(StoredDocKey key,
 
 std::unique_ptr<Item> makeCompressibleItem(Vbid vbid,
                                            const DocKey& key,
-                                           const std::string& value,
+                                           const std::string& body,
                                            protocol_binary_datatype_t datatype,
                                            bool shouldCompress,
                                            bool makeXattrBody) {
     protocol_binary_datatype_t itemDataType = datatype;
-    std::string v = value;
+    std::string value = body;
     if (makeXattrBody) {
-        v = createXattrValue(value, true, false);
+        value = createXattrValue(body, true, false);
         itemDataType |= PROTOCOL_BINARY_DATATYPE_XATTR;
     }
     if (shouldCompress) {
         cb::compression::Buffer output;
-        cb::compression::deflate(cb::compression::Algorithm::Snappy, v, output);
+        cb::compression::deflate(
+                cb::compression::Algorithm::Snappy, value, output);
         itemDataType |= PROTOCOL_BINARY_DATATYPE_SNAPPY;
         return std::make_unique<Item>(key, /*flags*/0, /*exp*/0,
                                       output.data(), output.size(),
                                       itemDataType);
     }
 
-    return std::make_unique<Item>(
-            key, /*flags*/ 0, /*exp*/ 0, v.c_str(), v.length(), itemDataType);
+    return std::make_unique<Item>(key,
+                                  0 /*flags*/,
+                                  0 /*exp*/,
+                                  value.c_str(),
+                                  value.length(),
+                                  itemDataType,
+                                  0 /*cas*/,
+                                  -1 /*seqno*/,
+                                  vbid);
 }
 
 bool queueNewItem(VBucket& vbucket, const std::string& key) {
