MB-29287: Test that takeover finds all items

jimwwalker · daverigby · commit 3102e467432d · 2018-04-25T16:29:45.000Z
Using the single threaded test harness we can demonstrate the window in which a close-stream and create stream will lose items. By triggering the close/re-create from within the snapshot processor after it has a handle on the stream which will be closed and before it access the items, we will transfer items into the dead stream and prevent the new takeover stream from obtaining them. Change-Id: I896a7cbd3d32419576294ea88c288b097759a362 Reviewed-on: http://review.couchbase.org/93123 Well-Formed: Build Bot <build@couchbase.com> Reviewed-by: Dave Rigby <daver@couchbase.com> Tested-by: Build Bot <build@couchbase.com>
diff --git a/CMakeLists.txt b/CMakeLists.txt
@@ -185,6 +185,7 @@ TARGET_LINK_LIBRARIES(ep cJSON JSON_checker couchstore forestdb
 # so simpler / quicker just to link them into a single executable).
 ADD_EXECUTABLE(ep-engine_ep_unit_tests
   tests/mock/mock_dcp.cc
+  tests/mock/mock_dcp_producer.cc
   tests/module_tests/checkpoint_test.cc
   tests/module_tests/defragmenter_test.cc
   tests/module_tests/ep_unit_tests_main.cc
diff --git a/src/dcp/producer.h b/src/dcp/producer.h
@@ -237,7 +237,7 @@ class DcpProducer : public Producer {
 
     ExTask checkpointCreatorTask;
 
-private:
+protected:
 
 
     DcpResponse* getNextItem();
diff --git a/src/dcp/stream.h b/src/dcp/stream.h
@@ -264,7 +264,7 @@ class ActiveStream : public Stream {
 
 protected:
     // Returns the outstanding items for the stream's checkpoint cursor.
-    void getOutstandingItems(RCPtr<VBucket> &vb, std::vector<queued_item> &items);
+    virtual void getOutstandingItems(RCPtr<VBucket> &vb, std::vector<queued_item> &items);
 
     // Given a set of queued items, create mutation responses for each item,
     // and pass onto the producer associated with this stream.
diff --git a/tests/mock/mock_dcp_producer.cc b/tests/mock/mock_dcp_producer.cc
@@ -0,0 +1,49 @@
+/* -*- 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 "mock_dcp_producer.h"
+
+#include "mock_stream.h"
+
+void MockDcpProducer::mockActiveStreamRequest(uint32_t flags,
+                                              uint32_t opaque,
+                                              uint16_t vbucket,
+                                              uint64_t start_seqno,
+                                              uint64_t end_seqno,
+                                              uint64_t vbucket_uuid,
+                                              uint64_t snap_start_seqno,
+                                              uint64_t snap_end_seqno) {
+    stream_t stream = new MockActiveStream(
+            static_cast<EventuallyPersistentEngine*>(&engine_),
+            this,
+            getName(),
+            flags,
+            opaque,
+            vbucket,
+            start_seqno,
+            end_seqno,
+            vbucket_uuid,
+            snap_start_seqno,
+            snap_end_seqno);
+    stream->setActive();
+    auto insertResult = streams.insert(std::make_pair(vbucket, stream));
+    if (!insertResult.second) {
+        throw std::logic_error("MockDcpProducer::mockActiveStreamRequest "
+                               "failed to insert requested stream");
+    }
+    notifyStreamReady(vbucket);
+}
diff --git a/tests/mock/mock_dcp_producer.h b/tests/mock/mock_dcp_producer.h
@@ -60,6 +60,30 @@ class MockDcpProducer: public DcpProducer {
         return *static_cast<ActiveStreamCheckpointProcessorTask*>(
                         checkpointCreatorTask.get());
     }
+
+    /**
+     * Finds the stream for a given vbucket
+     */
+    SingleThreadedRCPtr<Stream> findStream(uint16_t vbid) {
+        auto it = streams.find(vbid);
+        if (it != streams.end()) {
+            return it->second;
+        } else {
+            return SingleThreadedRCPtr<Stream>();
+        }
+    }
+
+    /**
+     * Place a mock active stream into the producer
+     */
+    void mockActiveStreamRequest(uint32_t flags,
+                                 uint32_t opaque,
+                                 uint16_t vbucket,
+                                 uint64_t start_seqno,
+                                 uint64_t end_seqno,
+                                 uint64_t vbucket_uuid,
+                                 uint64_t snap_start_seqno,
+                                 uint64_t snap_end_seqno);
 };
 
 using mock_dcp_producer_t = SingleThreadedRCPtr<MockDcpProducer>;
diff --git a/tests/mock/mock_stream.h b/tests/mock/mock_stream.h
@@ -17,6 +17,8 @@
 
 #pragma once
 
+#include "dcp/consumer.h"
+#include "dcp/producer.h"
 #include "dcp/stream.h"
 
 /*
@@ -90,4 +92,13 @@ class MockActiveStream : public ActiveStream {
     void transitionStateToBackfilling() {
         transitionState(STREAM_BACKFILLING);
     }
+
+    virtual void getOutstandingItems(RCPtr<VBucket>& vb,
+                                     std::vector<queued_item>& items) override {
+        preGetOutstandingItemsCallback();
+        ActiveStream::getOutstandingItems(vb, items);
+    }
+
+    /// A callback to allow tests to inject code before we access the checkpoint
+    std::function<void()> preGetOutstandingItemsCallback = [] { return; };
 };
diff --git a/tests/module_tests/dcp_test.cc b/tests/module_tests/dcp_test.cc
@@ -973,21 +973,8 @@ class ActiveStreamChkptProcessorTaskTest : public SingleThreadedEPStoreTest {
     }
 
     void notifyAndStepToCheckpoint() {
-        auto vb = store->getVBucket(vbid);
-        ASSERT_NE(nullptr, vb.get());
-
-        producer->notifySeqnoAvailable(vbid, vb->getHighSeqno());
-
-        /* Step which will notify the checkpoint processor task */
-        EXPECT_EQ(ENGINE_SUCCESS, producer->step(producers.get()));
-        EXPECT_EQ(1, producer->getCheckpointSnapshotTask().queueSize());
-
-        /* Run the task */
-        producer->getCheckpointSnapshotTask().run();
-
-        /* This time the step should return something that is read from the
-         checkpoint processor */
-        EXPECT_EQ(ENGINE_WANT_MORE, producer->step(producers.get()));
+        SingleThreadedEPStoreTest::notifyAndStepToCheckpoint(*producer,
+                                                              producers.get());
     }
 
     const void* cookie;
diff --git a/tests/module_tests/evp_store_single_threaded_test.cc b/tests/module_tests/evp_store_single_threaded_test.cc
@@ -17,11 +17,34 @@
 
 #include "evp_store_single_threaded_test.h"
 
+#include "../mock/mock_dcp.h"
 #include "../mock/mock_dcp_producer.h"
 #include "../mock/mock_dcp_consumer.h"
 #include "../mock/mock_stream.h"
+
+#include "failover-table.h"
 #include "programs/engine_testapp/mock_server.h"
 
+void SingleThreadedEPStoreTest::notifyAndStepToCheckpoint(
+        MockDcpProducer& producer, dcp_message_producers* producers) {
+
+    auto vb = store->getVBucket(vbid);
+    ASSERT_NE(nullptr, vb.get());
+
+    producer.notifySeqnoAvailable(vbid, vb->getHighSeqno());
+
+    /* Step which will notify the checkpoint processor task */
+    EXPECT_EQ(ENGINE_SUCCESS, producer.step(producers));
+    EXPECT_EQ(1, producer.getCheckpointSnapshotTask().queueSize());
+
+    /* Run the task */
+    producer.getCheckpointSnapshotTask().run();
+
+    /* This time the step should return something that is read from the
+       checkpoint processor */
+    EXPECT_EQ(ENGINE_WANT_MORE, producer.step(producers));
+}
+
 /*
  * The following test checks to see if we call handleSlowStream when in a
  * backfilling state, but the backfillTask is not running, we
@@ -674,3 +697,194 @@ TEST_F(SingleThreadedEPStoreTest, MB18953_taskWake) {
     EXPECT_EQ(1, task_executor->getNumReadyTasks(NONIO_TASK_IDX));
     runNextTask(lpNonioQ, "TestTask DefragmenterTask"); // lptask goes second
 }
+
+extern uint8_t dcp_last_op;
+extern std::string dcp_last_key;
+
+class MB_29287 : public SingleThreadedEPStoreTest {
+public:
+    void SetUp() override {
+        SingleThreadedEPStoreTest::SetUp();
+        cookie = create_mock_cookie();
+        setVBucketStateAndRunPersistTask(vbid, vbucket_state_active);
+        // 1. Mock producer
+        producer = new MockDcpProducer(*engine,
+                                       cookie,
+                                       "test_producer",
+                                       /*notifyOnly*/ false);
+
+        producers = get_dcp_producers(
+                reinterpret_cast<ENGINE_HANDLE*>(engine.get()),
+                reinterpret_cast<ENGINE_HANDLE_V1*>(engine.get()));
+
+        // 2. Mock active stream
+        producer->mockActiveStreamRequest(0, // flags
+                                          1, // opaque
+                                          vbid,
+                                          0, // start_seqno
+                                          ~0, // end_seqno
+                                          0, // vbucket_uuid,
+                                          0, // snap_start_seqno,
+                                          0); // snap_end_seqno,
+
+        store_item(vbid, "1", "value1");
+        store_item(vbid, "2", "value2");
+        store_item(vbid, "3", "value3");
+        flush_vbucket_to_disk(vbid, 3);
+        notifyAndStepToCheckpoint(*producer, producers.get());
+
+        for (int i = 0; i < 3; i++) { // 1, 2 and 3
+            EXPECT_EQ(ENGINE_WANT_MORE, producer->step(producers.get()));
+            EXPECT_EQ(PROTOCOL_BINARY_CMD_DCP_MUTATION, dcp_last_op);
+        }
+
+        store_item(vbid, "4", "value4");
+
+        auto stream = producer->findStream(vbid);
+        auto* mockStream = static_cast<MockActiveStream*>(stream.get());
+        mockStream->preGetOutstandingItemsCallback =
+                std::bind(&MB_29287::closeAndRecreateStream, this);
+
+        // call next - get success (nothing ready, but task has been scheduled)
+        EXPECT_EQ(ENGINE_SUCCESS, producer->step(producers.get()));
+
+        // Run the snapshot task and step (triggering
+        // preGetOutstandingItemsCallback)
+        notifyAndStepToCheckpoint(*producer, producers.get());
+    }
+
+    void TearDown() override {
+        destroy_mock_cookie(cookie);
+        producer->closeAllStreams();
+        producer->cancelCheckpointCreatorTask();
+        producer.reset();
+        SingleThreadedEPStoreTest::TearDown();
+    }
+
+    void closeAndRecreateStream() {
+        // Without the fix, 5 will be lost
+        store_item(vbid, "5", "don't lose me");
+        producer->closeStream(1, 0);
+        auto vb = store->getVBuckets().getBucket(vbid);
+        ASSERT_NE(nullptr, vb.get());
+        producer->mockActiveStreamRequest(DCP_ADD_STREAM_FLAG_TAKEOVER,
+                                          1, // opaque
+                                          vbid,
+                                          3, // start_seqno
+                                          ~0, // end_seqno
+                                          vb->failovers->getLatestUUID(),
+                                          3, // snap_start_seqno
+                                          ~0); // snap_end_seqno
+    }
+
+    const void* cookie = nullptr;
+    mock_dcp_producer_t producer;
+    std::unique_ptr<dcp_message_producers> producers;
+};
+
+// Stream takeover with no more writes
+TEST_F(MB_29287, dataloss_end) {
+    auto stream = producer->findStream(vbid);
+    auto* as = static_cast<ActiveStream*>(stream.get());
+
+    EXPECT_EQ(STREAM_TAKEOVER_SEND, stream->getState());
+    EXPECT_EQ(ENGINE_WANT_MORE, producer->step(producers.get()));
+    EXPECT_EQ(PROTOCOL_BINARY_CMD_DCP_MUTATION, dcp_last_op);
+    dcp_last_op = 0;
+    EXPECT_EQ("4", dcp_last_key);
+
+    EXPECT_EQ(ENGINE_WANT_MORE, producer->step(producers.get()));
+    EXPECT_EQ(PROTOCOL_BINARY_CMD_DCP_MUTATION, dcp_last_op);
+    dcp_last_op = 0;
+    EXPECT_EQ("5", dcp_last_key);
+
+    // Snapshot received
+    as->snapshotMarkerAckReceived();
+
+    // set-vb-state now underway
+    EXPECT_EQ(ENGINE_WANT_MORE, producer->step(producers.get()));
+    EXPECT_EQ(PROTOCOL_BINARY_CMD_DCP_SET_VBUCKET_STATE, dcp_last_op);
+
+    // Move stream to pending and vb to dead
+    as->setVBucketStateAckRecieved();
+
+    // Cannot store anymore items
+    store_item(vbid,
+               "K6",
+               "value6",
+               ENGINE_NOT_MY_VBUCKET);
+
+    EXPECT_EQ(ENGINE_WANT_MORE, producer->step(producers.get()));
+    EXPECT_EQ(PROTOCOL_BINARY_CMD_DCP_SET_VBUCKET_STATE, dcp_last_op);
+    as->setVBucketStateAckRecieved();
+    EXPECT_TRUE(!stream->isActive());
+
+    auto vb = store->getVBuckets().getBucket(vbid);
+    ASSERT_NE(nullptr, vb.get());
+    // Have persistence cursor only (dcp now closed down)
+    EXPECT_EQ(1, vb->checkpointManager.getNumOfCursors());
+}
+
+// takeover when more writes occur
+TEST_F(MB_29287, dataloss_hole) {
+    auto stream = producer->findStream(vbid);
+    auto* as = static_cast<ActiveStream*>(stream.get());
+
+    store_item(vbid, "6", "value6");
+
+    EXPECT_EQ(STREAM_TAKEOVER_SEND, stream->getState());
+    EXPECT_EQ(ENGINE_WANT_MORE, producer->step(producers.get()));
+    EXPECT_EQ(PROTOCOL_BINARY_CMD_DCP_MUTATION, dcp_last_op);
+    dcp_last_op = 0;
+    EXPECT_EQ("4", dcp_last_key);
+
+    EXPECT_EQ(ENGINE_WANT_MORE, producer->step(producers.get()));
+    EXPECT_EQ(PROTOCOL_BINARY_CMD_DCP_MUTATION, dcp_last_op);
+    dcp_last_op = 0;
+    EXPECT_EQ("5", dcp_last_key);
+
+    // Snapshot received
+    as->snapshotMarkerAckReceived();
+
+    // More data in the checkpoint (key 6)
+
+    // call next - get success (nothing ready, but task has been scheduled)
+    EXPECT_EQ(ENGINE_SUCCESS, producer->step(producers.get()));
+
+    // Run the snapshot task and step
+    notifyAndStepToCheckpoint(*producer, producers.get());
+
+    EXPECT_EQ(ENGINE_WANT_MORE, producer->step(producers.get()));
+    EXPECT_EQ(PROTOCOL_BINARY_CMD_DCP_MUTATION, dcp_last_op);
+    EXPECT_EQ("6", dcp_last_key);
+
+    // Snapshot received
+    as->snapshotMarkerAckReceived();
+
+    // Now send
+    EXPECT_EQ(STREAM_TAKEOVER_SEND, stream->getState());
+
+    // set-vb-state now underway
+    EXPECT_EQ(ENGINE_WANT_MORE, producer->step(producers.get()));
+    EXPECT_EQ(PROTOCOL_BINARY_CMD_DCP_SET_VBUCKET_STATE, dcp_last_op);
+    dcp_last_op = 0;
+
+    // Move stream to pending and vb to dead
+    as->setVBucketStateAckRecieved();
+
+    // Cannot store anymore items
+    store_item(vbid,
+               "K6",
+               "value6",
+               ENGINE_NOT_MY_VBUCKET);
+
+    EXPECT_EQ(ENGINE_WANT_MORE, producer->step(producers.get()));
+    EXPECT_EQ(PROTOCOL_BINARY_CMD_DCP_SET_VBUCKET_STATE, dcp_last_op);
+    as->setVBucketStateAckRecieved();
+    EXPECT_TRUE(!stream->isActive());
+
+    auto vb = store->getVBuckets().getBucket(vbid);
+    ASSERT_NE(nullptr, vb.get());
+    // Have persistence cursor only (dcp now closed down)
+    EXPECT_EQ(1, vb->checkpointManager.getNumOfCursors());
+}
diff --git a/tests/module_tests/evp_store_single_threaded_test.h b/tests/module_tests/evp_store_single_threaded_test.h
diff --git a/tests/module_tests/evp_store_test.cc b/tests/module_tests/evp_store_test.cc
diff --git a/tests/module_tests/evp_store_test.h b/tests/module_tests/evp_store_test.h
