Merge remote-tracking branch 'couchbase/alice'

* couchbase/alice:
  MB-31495: Fix bug in getRandomKey
  MB-31175: Ephemeral HTTombstonePurger underflows age check
  MB-31327: Ephemeral backfill does not respect purge seqno.
  MB-31141: Don't reject snappy|raw DCP deletes
  MB-31141: Account for nmeta in deleteWithMeta

Change-Id: I15638b53f28908581011a852798494c4e76c54d2

Author: daverigby

daemon/mcbp_validators.cc

@@ -420,8 +420,12 @@ static bool valid_dcp_delete_datatype(protocol_binary_datatype_t datatype) {
     // it may send XATTR|JSON (with snappy possible). These are now allowed
     // so rebalance won't be failed and the consumer will sanitise the faulty
     // documents.
-    std::array<const protocol_binary_datatype_t, 5> valid = {
+    // MB-31141: Allowing RAW+Snappy. A bug in delWithMeta has allowed us to
+    // create deletes with a non-zero value tagged as RAW, which when snappy
+    // is enabled gets DCP shipped as RAW+Snappy.
+    std::array<const protocol_binary_datatype_t, 6> valid = {
             {PROTOCOL_BINARY_RAW_BYTES,
+             PROTOCOL_BINARY_RAW_BYTES | PROTOCOL_BINARY_DATATYPE_SNAPPY,
              PROTOCOL_BINARY_DATATYPE_XATTR,
              PROTOCOL_BINARY_DATATYPE_XATTR | PROTOCOL_BINARY_DATATYPE_SNAPPY,
              PROTOCOL_BINARY_DATATYPE_XATTR | PROTOCOL_BINARY_DATATYPE_JSON,

engines/ep/src/dcp/backfill_memory.cc

@@ -210,6 +210,21 @@ backfill_status_t DCPBackfillMemoryBuffered::create() {
         return backfill_snooze;
     }
 
+    /* Check startSeqno against the purge-seqno of the vb.
+     * If the startSeqno != 1 (a 0 to n request) then startSeqno must be
+     * greater than purgeSeqno. */
+    if (startSeqno != 1 && (startSeqno <= evb->getPurgeSeqno())) {
+        EP_LOG_WARN(
+                "DCPBackfillMemoryBuffered::create(): "
+                "({}) running backfill failed because the startSeqno:{} is < "
+                "purgeSeqno:{}",
+                getVBucketId(),
+                startSeqno,
+                evb->getPurgeSeqno());
+        stream->setDead(END_STREAM_ROLLBACK);
+        return backfill_finished;
+    }
+
     /* Advance the cursor till start, mark snapshot and update backfill
        remaining count */
     while (rangeItr.curr() != rangeItr.end()) {

engines/ep/src/dcp/consumer.cc

@@ -572,10 +572,16 @@ ENGINE_ERROR_CODE DcpConsumer::deletion(uint32_t opaque,
         // MB-29040: Producer may send deleted doc with value that still has
         // the user xattrs and the body. Fix up that mistake by running the
         // expiry hook which will correctly process the document
-        if (mcbp::datatype::is_xattr(datatype) && value.size()) {
-            auto vb = engine_.getVBucket(vbucket);
-            if (vb) {
-                engine_.getKVBucket()->runPreExpiryHook(*vb, *item);
+        if (value.size()) {
+            if (mcbp::datatype::is_xattr(datatype)) {
+                auto vb = engine_.getVBucket(vbucket);
+                if (vb) {
+                    engine_.getKVBucket()->runPreExpiryHook(*vb, *item);
+                }
+            } else {
+                // MB-31141: Deletes cannot have a value
+                item->replaceValue(Blob::New(0));
+                item->setDataType(PROTOCOL_BINARY_RAW_BYTES);
             }
         }
 

engines/ep/src/ep_engine.cc

@@ -5135,9 +5135,12 @@ ENGINE_ERROR_CODE EventuallyPersistentEngine::deleteWithMeta(
         keyOffset += 2; // 2 bytes for nmeta
         nmeta = ntohs(nmeta);
         if (nmeta > 0) {
-            emd = cb::const_byte_buffer(
-                    request->bytes + sizeof(request->bytes) + nkey + keyOffset,
-                    nmeta);
+            // Correct the vallen
+            vallen -= nmeta;
+            emd = cb::const_byte_buffer(request->bytes +
+                                                sizeof(request->bytes) + nkey +
+                                                keyOffset + vallen,
+                                        nmeta);
         }
     }
 

engines/ep/src/ephemeral_tombstone_purger.cc

@@ -44,7 +44,13 @@ bool EphemeralVBucket::HTTombstonePurger::visit(
         const HashTable::HashBucketLock& hbl, StoredValue& v) {
     auto* osv = v.toOrderedStoredValue();
 
-    if (osv->isDeleted() && (now - osv->getDeletedTime() >= purgeAge)) {
+    // MB-31175: Item must have been deleted before this task starts to ensure
+    // that we do not get a -ve value when we check if the time difference
+    // is >= purgeAge. This is preferable to updating the task start time for
+    // every visit and has little impact as this task runs frequently.
+    if (osv->isDeleted() &&
+        (now >= osv->getDeletedTime()) &&
+        (now - osv->getDeletedTime() >= purgeAge)) {
         // This item should be purged. Remove from the HashTable and move over
         // to being owned by the sequence list.
         auto ownedSV = vbucket->ht.unlocked_release(hbl, v.getKey());

engines/ep/src/kv_bucket.cc

@@ -1407,24 +1407,24 @@ GetValue KVBucket::getInternal(const DocKey& key,
 GetValue KVBucket::getRandomKey() {
     size_t max = vbMap.getSize();
 
-    const Vbid::id_type start = labs(random()) % max;
+    const Vbid::id_type start = labs(getRandom()) % max;
     Vbid::id_type curr = start;
     std::unique_ptr<Item> itm;
 
     while (itm == NULL) {
         VBucketPtr vb = getVBucket(Vbid(curr++));
         while (!vb || vb->getState() != vbucket_state_active) {
-            if (curr == start) {
-                return GetValue(NULL, ENGINE_KEY_ENOENT);
-            }
             if (curr == max) {
                 curr = 0;
             }
+            if (curr == start) {
+                return GetValue(NULL, ENGINE_KEY_ENOENT);
+            }
 
             vb = getVBucket(Vbid(curr++));
         }
 
-        if ((itm = vb->ht.getRandomKey(random()))) {
+        if ((itm = vb->ht.getRandomKey(getRandom()))) {
             GetValue rv(std::move(itm), ENGINE_SUCCESS);
             return rv;
         }

engines/ep/src/kv_bucket.h

@@ -177,6 +177,11 @@ class KVBucket : public KVBucketIface {
                            options);
     }
 
+    /**
+     * Retrieve a value randomly from the store.
+     *
+     * @return a GetValue representing the value retrieved
+     */
     GetValue getRandomKey(void);
 
     /**
@@ -1015,6 +1020,12 @@ class KVBucket : public KVBucketIface {
 
     std::atomic<size_t> maxTtl;
 
+    /**
+     * Allows us to override the random function.  This is used for testing
+     * purposes where we want a constant number as opposed to a random one.
+     */
+    std::function<long()> getRandom = random;
+
     friend class KVBucketTest;
 
     DISALLOW_COPY_AND_ASSIGN(KVBucket);

engines/ep/src/stored-value.cc

@@ -421,6 +421,7 @@ std::ostream& operator<<(std::ostream& os, const StoredValue& sv) {
 
     // seqno, revid, expiry / purge time
     os << "seq:" << sv.getBySeqno() << " rev:" << sv.getRevSeqno();
+    os << " cas:" << sv.getCas();
     os << " key:\"" << sv.getKey() << "\"";
     if (sv.isOrdered() && sv.isDeleted()) {
         os << " del_time:" << sv.lock_expiry_or_delete_time;

engines/ep/tests/mock/mock_stream.h

@@ -171,6 +171,8 @@ class MockActiveStream : public ActiveStream {
     void public_registerCursor(CheckpointManager& manager,
                                const std::string& name,
                                int64_t seqno);
+
+    bool isDead() { return ActiveStream::getState() == StreamState::Dead; };
 };
 
 /**

engines/ep/tests/module_tests/dcp_test.cc

@@ -1248,6 +1248,23 @@ TEST_P(StreamTest, backfillGetsNoItems) {
     }
 }
 
+TEST_P(StreamTest, bufferedMemoryBackfillPurgeGreaterThanStart) {
+    if (engine->getConfiguration().getBucketType() == "ephemeral") {
+        setup_dcp_stream(0, IncludeValue::No, IncludeXattrs::No);
+        auto evb = std::shared_ptr<EphemeralVBucket>(
+                std::dynamic_pointer_cast<EphemeralVBucket>(vb0));
+
+        // Force the purgeSeqno because it's easier than creating and
+        // deleting items
+        evb->setPurgeSeqno(3);
+
+        // Backfill with start != 1 and start != end and start < purge
+        DCPBackfillMemoryBuffered dcpbfm (evb, stream, 2, 4);
+        dcpbfm.run();
+        EXPECT_TRUE(stream->isDead());
+    }
+}
+
 /* Regression test for MB-17766 - ensure that when an ActiveStream is preparing
  * queued items to be sent out via a DCP consumer, that nextCheckpointItem()
  * doesn't incorrectly return false (meaning that there are no more checkpoint
@@ -3086,28 +3103,25 @@ TEST_P(ConnectionTest, test_mb24424_deleteResponse) {
     ASSERT_TRUE(stream->isActive());
 
     std::string key = "key";
-    std::string data = R"({"json":"yes"})";
     const DocKey docKey{reinterpret_cast<const uint8_t*>(key.data()),
                         key.size(),
                         DocKeyEncodesCollectionId::No};
-    cb::const_byte_buffer value{reinterpret_cast<const uint8_t*>(data.data()),
-        data.size()};
     uint8_t extMeta[1] = {uint8_t(PROTOCOL_BINARY_DATATYPE_JSON)};
     cb::const_byte_buffer meta{extMeta, sizeof(uint8_t)};
 
-    consumer->deletion(/*opaque*/1,
-                       /*key*/docKey,
-                       /*values*/value,
-                       /*priv_bytes*/0,
-                       /*datatype*/PROTOCOL_BINARY_DATATYPE_JSON,
-                       /*cas*/0,
-                       /*vbucket*/vbid,
-                       /*bySeqno*/1,
-                       /*revSeqno*/0,
-                       /*meta*/meta);
-
-    auto messageSize = MutationResponse::deletionBaseMsgBytes +
-            key.size() + data.size() + sizeof(extMeta);
+    consumer->deletion(/*opaque*/ 1,
+                       /*key*/ docKey,
+                       /*value*/ {},
+                       /*priv_bytes*/ 0,
+                       /*datatype*/ PROTOCOL_BINARY_RAW_BYTES,
+                       /*cas*/ 0,
+                       /*vbucket*/ vbid,
+                       /*bySeqno*/ 1,
+                       /*revSeqno*/ 0,
+                       /*meta*/ meta);
+
+    auto messageSize = MutationResponse::deletionBaseMsgBytes + key.size() +
+                       sizeof(extMeta);
 
     EXPECT_EQ(messageSize, stream->responseMessageSize);