MB-30019: Always close replica-checkpoint on memory-snapshot-end

Currently, when a Consumer receives the shapshot-end mutation of a
memory-snapshot we close the current open checkpoint only if
(mem_used > high_watermark).
As seen in MB-30019, we may end up with keeping big checkpoints in
memory when a Consumer has received the latest snapshot.
We this patch we unconditionally close a open replica-checkpoint when
the Consumer receives the snapshot-end mutation for a memory-sanpshot.

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

Author: paolococchi

engines/ep/CMakeLists.txt

@@ -280,6 +280,7 @@ if (COUCHBASE_KV_BUILD_UNIT_TESTS)
                    tests/module_tests/checkpoint_remover_test.cc
                    tests/module_tests/checkpoint_test.h
                    tests/module_tests/checkpoint_test.cc
+                   tests/module_tests/checkpoint_utils.h
                    tests/module_tests/collections/collection_dockey_test.cc
                    tests/module_tests/collections/evp_store_collections_dcp_test.cc
                    tests/module_tests/collections/evp_store_collections_eraser_test.cc

engines/ep/src/dcp/stream.cc

@@ -2715,23 +2715,17 @@ void PassiveStream::processSetVBucketState(SetVBucketState* state) {
 void PassiveStream::handleSnapshotEnd(VBucketPtr& vb, uint64_t byseqno) {
     if (byseqno == cur_snapshot_end.load()) {
         auto& ckptMgr = *vb->checkpointManager;
+
         if (cur_snapshot_type.load() == Snapshot::Disk &&
                 vb->isBackfillPhase()) {
             vb->setBackfillPhase(false);
-            const auto id = ckptMgr.getOpenCheckpointId() + 1;
-            ckptMgr.checkAndAddNewCheckpoint(id, *vb);
-        } else {
-            size_t mem_threshold = engine->getEpStats().mem_high_wat.load();
-            size_t mem_used =
-                    engine->getEpStats().getEstimatedTotalMemoryUsed();
-            /* We want to add a new replica checkpoint if the mem usage is above
-               high watermark (85%) */
-            if (mem_threshold < mem_used) {
-                const auto id = ckptMgr.getOpenCheckpointId() + 1;
-                ckptMgr.checkAndAddNewCheckpoint(id, *vb);
-            }
         }
 
+        // MB-30019: we always want to close the open checkpoint on replica
+        // vbuckets when the Consumer receives the snapshot-end mutation
+        ckptMgr.checkAndAddNewCheckpoint(ckptMgr.getOpenCheckpointId() + 1,
+                                         *vb);
+
         if (cur_snapshot_ack) {
             {
                 LockHolder lh(streamMutex);

engines/ep/src/dcp/stream.h

@@ -742,7 +742,7 @@ class PassiveStream : public Stream {
 
     bool transitionState(StreamState newState);
 
-    ENGINE_ERROR_CODE processMutation(MutationResponse* mutation);
+    virtual ENGINE_ERROR_CODE processMutation(MutationResponse* mutation);
 
     ENGINE_ERROR_CODE processDeletion(MutationResponse* deletion);
 
@@ -782,7 +782,7 @@ class PassiveStream : public Stream {
 
     void handleSnapshotEnd(VBucketPtr& vb, uint64_t byseqno);
 
-    void processMarker(SnapshotMarker* marker);
+    virtual void processMarker(SnapshotMarker* marker);
 
     void processSetVBucketState(SetVBucketState* state);
 

engines/ep/tests/mock/mock_stream.h

@@ -272,6 +272,14 @@ class MockPassiveStream : public PassiveStream {
         return PassiveStream::messageReceived(std::move(dcpResponse));
     }
 
+    void processMarker(SnapshotMarker* marker) override {
+        PassiveStream::processMarker(marker);
+    }
+
+    ENGINE_ERROR_CODE processMutation(MutationResponse* mutation) override {
+        return PassiveStream::processMutation(mutation);
+    }
+
     size_t getNumBufferItems() const {
         LockHolder lh(buffer.bufMutex);
         return buffer.messages.size();

engines/ep/tests/module_tests/checkpoint_remover_test.cc

@@ -16,6 +16,9 @@
  */
 
 #include "checkpoint_remover_test.h"
+
+#include "checkpoint_utils.h"
+
 #include <engines/ep/src/checkpoint_remover.h>
 
 #include "../mock/mock_dcp.h"
@@ -28,12 +31,6 @@ size_t CheckpointRemoverTest::getMaxCheckpointItems(VBucket& vb) {
     return vb.checkpointManager->getCheckpointConfig().getCheckpointMaxItems();
 }
 
-const CheckpointList&
-CheckpointManagerTestIntrospector::public_getCheckpointList(
-        CheckpointManager& checkpointManager) {
-    return checkpointManager.checkpointList;
-}
-
 /**
  * Check that the VBucketMap.getActiveVBucketsSortedByChkMgrMem() returns the
  * correct ordering of vBuckets, sorted from largest memory usage to smallest.
@@ -207,4 +204,4 @@ TEST_F(CheckpointRemoverEPTest, CursorDropMemoryFreed) {
 
     // There should only be the one checkpoint cursor now for persistence
     ASSERT_EQ(1, checkpointManager->getNumOfCursors());
-}
+}

engines/ep/tests/module_tests/checkpoint_remover_test.h

@@ -50,13 +50,3 @@ class CheckpointRemoverEPTest : public CheckpointRemoverTest {
         return dynamic_cast<EPBucket&>(*store);
     }
 };
-
-/**
- * Stateless class used to gain privileged access into CheckpointManager for
- * testing purposes. This is a friend class of CheckpointManager.
- */
-class CheckpointManagerTestIntrospector {
-public:
-    static const CheckpointList& public_getCheckpointList(
-            CheckpointManager& checkpointManager);
-};

engines/ep/tests/module_tests/checkpoint_test.cc

@@ -15,6 +15,8 @@
  *   limitations under the License.
  */
 
+#include "checkpoint_test.h"
+
 #include "config.h"
 
 #include <algorithm>
@@ -23,7 +25,7 @@
 #include <vector>
 
 #include "checkpoint.h"
-#include "checkpoint_test.h"
+#include "checkpoint_utils.h"
 #include "configuration.h"
 #include "ep_vb.h"
 #include "failover-table.h"
@@ -32,6 +34,8 @@
 #include "tests/module_tests/test_helpers.h"
 #include "thread_gate.h"
 
+#include "../mock/mock_dcp_consumer.h"
+
 #include <engines/ep/src/ep_types.h>
 #include <gmock/gmock.h>
 #include <gtest/gtest.h>
@@ -1235,3 +1239,113 @@ TYPED_TEST(CheckpointTest,
     // Test - second item (duplicate key) should return false.
     EXPECT_FALSE(this->queueNewItem("key"));
 }
+
+/*
+ * We always want to close the current open checkpoint on replica-vbuckets
+ * when the Consumer receives the snapshotEnd mutation of a memory-snapshot.
+ */
+TEST_F(SingleThreadedCheckpointTest,
+       MB30019_CloseReplicaCheckpointOnMemorySnapshotEnd) {
+    setVBucketStateAndRunPersistTask(vbid, vbucket_state_replica);
+    auto vb = store->getVBuckets().getBucket(vbid);
+    auto* ckptMgr = vb->checkpointManager.get();
+    ASSERT_NE(nullptr, ckptMgr);
+
+    // We must have only 1 open checkpoint
+    ASSERT_EQ(1, ckptMgr->getNumCheckpoints());
+    // We must have only one cursor (the persistence cursor), as there is no
+    // DCP producer for vbid
+    ASSERT_EQ(1, ckptMgr->getNumOfCursors());
+    // We must have only the checkpoint-open and the vbucket-state meta-items
+    // in the open checkpoint
+    ASSERT_EQ(2, ckptMgr->getNumItems());
+    ASSERT_EQ(0, ckptMgr->getNumOpenChkItems());
+
+    auto consumer =
+            std::make_shared<MockDcpConsumer>(*engine, cookie, "test-consumer");
+    auto passiveStream = std::static_pointer_cast<MockPassiveStream>(
+            consumer->makePassiveStream(
+                    *engine,
+                    consumer,
+                    "test-passive-stream",
+                    0 /* flags */,
+                    0 /* opaque */,
+                    vbid,
+                    0 /* startSeqno */,
+                    std::numeric_limits<uint64_t>::max() /* endSeqno */,
+                    0 /* vbUuid */,
+                    0 /* snapStartSeqno */,
+                    0 /* snapEndSeqno */,
+                    0 /* vb_high_seqno */));
+
+    const size_t snapshotEnd = 3;
+    // 1) the consumer receives the snapshot-marker
+    SnapshotMarker snapshotMarker(
+            0 /* opaque */,
+            vbid,
+            0 /* startSeqno */,
+            snapshotEnd /* endSeqno */,
+            dcp_marker_flag_t::MARKER_FLAG_MEMORY /* flags */);
+    passiveStream->processMarker(&snapshotMarker);
+
+    // 2) the consumer receives the mutations until (snapshotEnd -1)
+    size_t i = 1;
+    for (; i < snapshotEnd; i++) {
+        // Queue item
+        queued_item qi(new Item(makeStoredDocKey("key_" + std::to_string(i)),
+                                0 /*flags*/,
+                                0 /*expiry*/,
+                                "value",
+                                5 /*valueSize*/,
+                                PROTOCOL_BINARY_RAW_BYTES,
+                                0 /*cas*/,
+                                i /*bySeqno*/,
+                                vb->getId()));
+
+        MutationResponse mutation(std::move(qi), 0 /* opaque */);
+
+        // PassiveStream::processMutation does 2 things:
+        //     1) setWithMeta (which enqueues the item into the checkpoint)
+        //     2) calls PassiveStream::handleSnapshotEnd (which must close the
+        //             open checkpoint if the current mutation is the
+        //             snapshot-end)
+        passiveStream->processMutation(&mutation);
+    }
+    // We must have 2 items in the checkpoint now
+    ASSERT_EQ(snapshotEnd - 1, ckptMgr->getNumOpenChkItems());
+    // We still must have only 1 open checkpoint, as the consumer has not
+    // received the snapshot-end mutation
+    ASSERT_EQ(1, ckptMgr->getNumCheckpoints());
+
+    // 3) the consumer receives the snapshotEnd mutation
+    queued_item qi(
+            new Item(makeStoredDocKey("key_" + std::to_string(snapshotEnd)),
+                     0 /*flags*/,
+                     0 /*expiry*/,
+                     "value",
+                     5 /*valueSize*/,
+                     PROTOCOL_BINARY_RAW_BYTES,
+                     0 /*cas*/,
+                     i /*bySeqno*/,
+                     vb->getId()));
+    MutationResponse mutation(std::move(qi), 0 /* opaque */);
+    passiveStream->processMutation(&mutation);
+
+    // The consumer has received the snapshotEnd mutation, now we expect
+    // that a new (empty) open checkpoint has been created. So we must have
+    // 2 checkpoints in total (the closed and the new open one).
+    ASSERT_EQ(2, ckptMgr->getNumCheckpoints());
+
+    // Also, the new open checkpoint must be empty (all mutations are in the
+    // closed one)
+    const auto& ckptList =
+            CheckpointManagerTestIntrospector::public_getCheckpointList(
+                    *ckptMgr);
+    ASSERT_EQ(ckptList.back()->getId(), ckptList.front()->getId() + 1);
+    ASSERT_EQ(checkpoint_state::CHECKPOINT_CLOSED,
+              ckptList.front()->getState_UNLOCKED());
+    ASSERT_EQ(snapshotEnd, ckptList.front()->getNumItems());
+    ASSERT_EQ(checkpoint_state::CHECKPOINT_OPEN,
+              ckptList.back()->getState_UNLOCKED());
+    ASSERT_EQ(0, ckptList.back()->getNumItems());
+}

engines/ep/tests/module_tests/checkpoint_test.h

@@ -21,6 +21,7 @@
 
 #include "checkpoint_config.h"
 #include "configuration.h"
+#include "evp_store_single_threaded_test.h"
 
 #include <gtest/gtest.h>
 
@@ -60,4 +61,9 @@ class CheckpointTest : public ::testing::Test {
     std::shared_ptr<Callback<uint16_t> > callback;
     std::unique_ptr<V> vbucket;
     std::unique_ptr<CheckpointManager> manager;
-};
+};
+
+/*
+ * Test fixture for single-threaded Checkpoint tests
+ */
+class SingleThreadedCheckpointTest : public SingleThreadedKVBucketTest {};

engines/ep/tests/module_tests/checkpoint_utils.h

@@ -0,0 +1,30 @@
+/* -*- Mode: C++; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */
+/*
+ *     Copyright 2018 Couchbase, Inc
+ *
+ *   Licensed under the Apache License, Version 2.0 (the "License");
+ *   you may not use this file except in compliance with the License.
+ *   You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *   Unless required by applicable law or agreed to in writing, software
+ *   distributed under the License is distributed on an "AS IS" BASIS,
+ *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *   See the License for the specific language governing permissions and
+ *   limitations under the License.
+ */
+
+#include "checkpoint.h"
+
+/**
+ * Stateless class used to gain privileged access into CheckpointManager for
+ * testing purposes. This is a friend class of CheckpointManager.
+ */
+class CheckpointManagerTestIntrospector {
+public:
+    static const CheckpointList& public_getCheckpointList(
+            CheckpointManager& checkpointManager) {
+        return checkpointManager.checkpointList;
+    }
+};