MB-35496: Remove completed prepares at compaction

When we compact a database file we can remove all completed
prepares to save space. A prepare is only needed for the lifetime
of a SyncWrite to ensure that we still have it should we shutdown
and restart. We don't need to worry about the metadata purge
interval when it comes to prepares because we stream Mutations
instead of Commits when streaming from disk.

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

Author: BenHuddleston

engines/ep/src/couch-kvstore/couch-kvstore-metadata.h

@@ -332,6 +332,10 @@ class MetaData {
             return operation == Operation::Commit;
         }
 
+        bool isPrepare() const {
+            return operation == Operation::Pending;
+        }
+
     private:
         // Assigning a whole byte to this (see MetaDataV3 class comment)
         // although only currently need 2 bits.
@@ -536,6 +540,11 @@ class MetaData {
                allMeta.v3.isCommit();
     }
 
+    bool isPrepare() const {
+        return getVersionInitialisedFrom() == MetaData::Version::V3 &&
+               allMeta.v3.isPrepare();
+    }
+
     cb::uint48_t getPrepareSeqno() const {
         return allMeta.v3.getPrepareSeqno();
     }

engines/ep/src/couch-kvstore/couch-kvstore.cc

@@ -953,6 +953,16 @@ static int time_purge_hook(Db* d, DocInfo* info, sized_buf item, void* ctx_p) {
                 }
             }
         } else {
+            // We can remove any prepares that have been completed. This works
+            // because we send Mutations intead of Commits when streaming from
+            // Disk so we do not need to send a Prepare message to keep things
+            // consistent on a replica.
+            if (metadata->isPrepare()) {
+                if (info->db_seq <= ctx->highCompletedSeqno) {
+                    return COUCHSTORE_COMPACT_DROP_ITEM;
+                }
+            }
+
             time_t currtime = ep_real_time();
             if (exptime && exptime < currtime && metadata->isCommit()) {
                 int ret;
@@ -1085,6 +1095,20 @@ bool CouchKVStore::compactDBInternal(compaction_ctx* hook_ctx,
         flags |= couchstore_encode_periodic_sync_flags(periodicSyncBytes);
     }
 
+    // It would seem logical to grab the state from disk her (readVBState(...))
+    // but we cannot do that. If we do, we may race with a concurrent scan as
+    // readVBState will overwrite the cached vbucket_state. As such, just use
+    // the cached vbucket_state.
+    vbucket_state* vbState = getVBucketState(vbid);
+    if (!vbState) {
+        EP_LOG_WARN(
+                "CouchKVStore::compactDBInternal ({}) Failed to obtain vbState "
+                "for the highCompletedSeqno",
+                vbid);
+        return false;
+    }
+    hook_ctx->highCompletedSeqno = vbState->highCompletedSeqno;
+
     // Perform COMPACTION of vbucket.couch.rev into vbucket.couch.rev.compact
     errCode = couchstore_compact_db_ex(compactdb,
                                        compact_file.c_str(),

engines/ep/src/kvstore.h

@@ -129,6 +129,9 @@ struct compaction_ctx {
     /// pointer as context cannot be constructed until deeper inside storage
     std::unique_ptr<Collections::VB::EraserContext> eraserContext;
     Collections::KVStore::DroppedCb droppedKeyCb;
+
+    /// The SyncRepl HCS, can purge any prepares before the HCS.
+    uint64_t highCompletedSeqno = 0;
 };
 
 struct kvstats_ctx {

engines/ep/tests/module_tests/evp_store_durability_test.cc

@@ -173,6 +173,11 @@ class DurabilityEPBucketTest : public STParameterizedBucketTest {
                                    uint64_t expectedValue);
 };
 
+/**
+ * Test fixtures for persistent bucket tests that only run under couchstore
+ */
+class DurabilityCouchstoreBucketTest : public DurabilityEPBucketTest {};
+
 /**
  * Test fixture for Durability-related tests applicable to ephemeral and
  * persistent buckets with either eviction modes.
@@ -1801,6 +1806,95 @@ TEST_P(DurabilityEPBucketTest, DoNotExpirePendingItem) {
     EXPECT_TRUE(gv.item->isDeleted());
 }
 
+// @TODO Rocksdb when we have manual compaction/compaction filtering this test
+// should be made to pass.
+TEST_P(DurabilityCouchstoreBucketTest, RemoveCommittedPreparesAtCompaction) {
+    setVBucketToActiveWithValidTopology();
+    using namespace cb::durability;
+
+    auto key = makeStoredDocKey("key");
+    auto req = Requirements(Level::Majority, Timeout(1000));
+    auto pending = makePendingItem(key, "value", req);
+    EXPECT_EQ(ENGINE_EWOULDBLOCK, store->set(*pending, cookie));
+
+    auto vb = store->getVBucket(vbid);
+    vb->commit(key,
+               1 /*prepareSeqno*/,
+               {} /*commitSeqno*/,
+               vb->lockCollections(key));
+
+    flushVBucketToDiskIfPersistent(vbid, 2);
+
+    CompactionConfig config;
+    compaction_ctx cctx(config, 0);
+    cctx.expiryCallback = std::make_shared<FailOnExpiryCallback>();
+
+    auto* kvstore = store->getOneRWUnderlying();
+
+    // Sanity - prepare exists before compaction
+    DiskDocKey prefixedKey(key, true /*prepare*/);
+    auto gv = kvstore->get(prefixedKey, Vbid(0));
+    EXPECT_EQ(ENGINE_SUCCESS, gv.getStatus());
+
+    EXPECT_TRUE(kvstore->compactDB(&cctx));
+
+    // Check the committed item on disk.
+    gv = kvstore->get(DiskDocKey(key), Vbid(0));
+    EXPECT_EQ(ENGINE_SUCCESS, gv.getStatus());
+    EXPECT_EQ("value", gv.item->getValue()->to_s());
+
+    // Check the Prepare on disk
+    gv = kvstore->get(prefixedKey, Vbid(0));
+    EXPECT_EQ(ENGINE_KEY_ENOENT, gv.getStatus());
+}
+
+TEST_P(DurabilityCouchstoreBucketTest, RemoveAbortedPreparedAtCompaction) {
+    setVBucketToActiveWithValidTopology();
+    using namespace cb::durability;
+
+    auto key = makeStoredDocKey("key");
+    auto req = Requirements(Level::Majority, Timeout(1000));
+    auto pending = makePendingItem(key, "value", req);
+    EXPECT_EQ(ENGINE_EWOULDBLOCK, store->set(*pending, cookie));
+
+    // Flush prepare
+    flushVBucketToDiskIfPersistent(vbid, 1);
+
+    auto vb = store->getVBucket(vbid);
+    vb->abort(key,
+              1 /*prepareSeqno*/,
+              {} /*commitSeqno*/,
+              vb->lockCollections(key));
+
+    // We can't purge the last item so write a dummy
+    auto dummyKey = makeStoredDocKey("dummy");
+    auto dummyItem = makeCommittedItem(dummyKey, "dummyValue");
+    EXPECT_EQ(ENGINE_SUCCESS, store->set(*dummyItem, cookie));
+
+    // Flush Abort and dummy
+    flushVBucketToDiskIfPersistent(vbid, 2);
+
+    CompactionConfig config;
+    compaction_ctx cctx(config, 0);
+    cctx.expiryCallback = std::make_shared<FailOnExpiryCallback>();
+    auto* kvstore = store->getOneRWUnderlying();
+    EXPECT_TRUE(kvstore->compactDB(&cctx));
+
+    // Check the Abort on disk. We won't remove it until the purge interval has
+    // passed because we need it to ensure we can resume a replica that had an
+    // outstanding prepare within the purge interval.
+    DiskDocKey prefixedKey(key, true /*prepare*/);
+    auto gv = kvstore->get(prefixedKey, Vbid(0));
+    EXPECT_EQ(ENGINE_SUCCESS, gv.getStatus());
+
+    cctx.compactConfig.purge_before_ts = std::numeric_limits<uint64_t>::max();
+    EXPECT_TRUE(kvstore->compactDB(&cctx));
+
+    // Now the Abort should be gone
+    gv = kvstore->get(prefixedKey, Vbid(0));
+    EXPECT_EQ(ENGINE_KEY_ENOENT, gv.getStatus());
+}
+
 template <typename F>
 void DurabilityEphemeralBucketTest::testPurgeCompletedPrepare(F& func) {
     setVBucketStateAndRunPersistTask(
@@ -1961,6 +2055,12 @@ TEST_P(DurabilityBucketTest, ActiveToReplicaAndCommit) {
     ASSERT_EQ(ENGINE_SUCCESS, vb.commit(key, 1, 4, vb.lockCollections(key)));
 }
 
+// Test cases which run against couchstore
+INSTANTIATE_TEST_CASE_P(AllBackends,
+                        DurabilityCouchstoreBucketTest,
+                        STParameterizedBucketTest::persistentConfigValues(),
+                        STParameterizedBucketTest::PrintToStringParamName);
+
 // Test cases which run against all persistent storage backends.
 INSTANTIATE_TEST_CASE_P(
         AllBackends,