Replica should handle deduped commits if backfilling from disk (3/3) [ETA 2019/7/19]

libc++abi.dylib: terminating with uncaught exception of type std::logic_error: PassiveDurabilityMonitor::resolvePrepare: Pending resolution for 'SW @0x10b650ff0 cookie:0x0 qi:[key:'cid:0x0:key' seqno:1 reqs:{Majority, timeout=infinite}] maj:0 #1stChainAcks:0 1stChainAcks:[] #2ndChainAcks:0 2ndChainAcks:[]', but received unexpected commit for key cid:0x0:key_1

TEST_P(DurabilityPassiveStreamTest, ReceiveDcpPrepareX2) {

    // Consumer receives [1, 1] snapshot with just a prepare

    testReceiveDcpPrepare();

    // The consumer now "disconnects" then "re-connects" and misses a commit for

    // the given key at seqno 2. It instead receives the following snapshot

    // [3, 4] for the same key containing a prepare and a commit.

    uint32_t opaque = 0;

    // Fake disconnect and reconnect, importantly, this sets up the valid window

    // for ignoring DCPAborts.

    consumer->closeAllStreams();

    consumer->addStream(opaque, vbid, 0 /*flags*/);

    stream = static_cast<MockPassiveStream*>(

            (consumer->getVbucketStream(vbid)).get());

    stream->acceptStream(cb::mcbp::Status::Success, opaque);

    ASSERT_TRUE(stream->isActive());

    // At Replica we don't expect multiple Durability items (for the same key)

    // within the same snapshot. That is because the Active prevents that for

    // avoiding de-duplication.

    // So, we need to simulate a Producer sending another SnapshotMarker with

    // the MARKER_FLAG_CHK set before the Consumer receives the Commit. That

    // will force the Consumer closing the open checkpoint (which Contains the

    // Prepare) and creating a new open one for queueing the Commit.

    SnapshotMarker marker(

            opaque,

            vbid,

            3 /*snapStart*/,

            3 /*snapEnd*/,

            dcp_marker_flag_t::MARKER_FLAG_MEMORY | MARKER_FLAG_CHK,

            {} /*streamId*/);

    stream->processMarker(&marker);

    // The consumer receives s:1 durable

    const std::string value("value");

    const uint64_t prepareSeqno = 3;

    auto key = makeStoredDocKey("key_1");

    using namespace cb::durability;

    const uint64_t cas = 999;

    queued_item qi(new Item(key,

                            0 /*flags*/,

                            0 /*expiry*/,

                            value.c_str(),

                            value.size(),

                            PROTOCOL_BINARY_RAW_BYTES,

                            cas /*cas*/,

                            prepareSeqno,

                            vbid));

    qi->setPendingSyncWrite(Requirements(Level::Majority, Timeout::Infinity()));

    ASSERT_EQ(ENGINE_SUCCESS,

              stream->messageReceived(std::make_unique<MutationConsumerMessage>(

                      std::move(qi),

                      opaque,

                      IncludeValue::Yes,

                      IncludeXattrs::Yes,

                      IncludeDeleteTime::No,

                      DocKeyEncodesCollectionId::No,

                      nullptr,

                      cb::mcbp::DcpStreamId{})));

    marker = SnapshotMarker(

            opaque,

            vbid,

            4 /*snapStart*/,

            4 /*snapEnd*/,

            dcp_marker_flag_t::MARKER_FLAG_MEMORY | MARKER_FLAG_CHK,

            {} /*streamId*/);

    stream->processMarker(&marker);

    auto commitSeqno = prepareSeqno + 1;

    ASSERT_EQ(ENGINE_SUCCESS,

              stream->messageReceived(std::make_unique<CommitSyncWrite>(

                      opaque, vbid, commitSeqno, key)));

}