MB-49742: Don't update droppedCollection if CompactKVStore fails

One CompactKVStore of many could fail and we should only update
droppedCollections for those CompactKVStore calls that passed.

Change-Id: I2903b54990bcdc0c6f8a4d5a4e345c8faf48b44f
Reviewed-on: https://review.couchbase.org/c/kv_engine/+/166003
Reviewed-by: Dave Rigby <daver@couchbase.com>
Tested-by: Build Bot <build@couchbase.com>

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

@@ -2166,6 +2166,7 @@ bool MagmaKVStore::compactDBInternal(std::unique_lock<std::mutex>& vbLock,
     LocalDbReqs localDbReqs;
 
     Status status;
+    std::unordered_set<CollectionID> purgedCollections;
     if (dropped.empty()) {
         // Compact the entire key range
         Slice nullKey;
@@ -2228,6 +2229,9 @@ bool MagmaKVStore::compactDBInternal(std::unique_lock<std::mutex>& vbLock,
 
             status = magma->CompactKVStore(
                     vbid.get(), keySlice, keySlice, compactionCB);
+
+            compactionStatusHook(status);
+
             if (!status) {
                 logger->warn(
                         "MagmaKVStore::compactDBInternal CompactKVStore {} "
@@ -2271,6 +2275,10 @@ bool MagmaKVStore::compactDBInternal(std::unique_lock<std::mutex>& vbLock,
             }
             ctx->eraserContext->processSystemEvent(key.getDocKey(),
                                                    SystemEvent::Collection);
+
+            // This collection purged successfully, save this for later so that
+            // we can update the droppedCollections local doc
+            purgedCollections.insert(dc.collectionId);
         }
     }
 
@@ -2312,9 +2320,8 @@ bool MagmaKVStore::compactDBInternal(std::unique_lock<std::mutex>& vbLock,
 
         // 2) Generate a new flatbuffer document to write back
         auto fbData = Collections::VB::Flush::
-                encodeRelativeComplementOfDroppedCollections(
-                        droppedCollections,
-                        ctx->eraserContext->getDroppedCollections());
+                encodeRelativeComplementOfDroppedCollections(droppedCollections,
+                                                             purgedCollections);
 
         // 3) If the function returned data, write it, else the document is
         // delete.

engines/ep/src/kvstore/magma-kvstore/magma-kvstore.h

@@ -744,6 +744,12 @@ class MagmaKVStore : public KVStore {
     folly::Synchronized<std::queue<std::tuple<Vbid, uint64_t>>>
             pendingVbucketDeletions;
 
+    /**
+     * Testing hook called with the result of CompactKVStore when dropping
+     * collections.
+     */
+    TestingHook<magma::Status&> compactionStatusHook;
+
 private:
     EventuallyPersistentEngine* currEngine;
 };

engines/ep/tests/mock/mock_magma_kvstore.h

@@ -47,6 +47,10 @@ class MockMagmaKVStore : public MagmaKVStore {
      */
     magma::Status newCheckpoint(Vbid vbid);
 
+    void setCompactionStatusHook(std::function<void(magma::Status&)> hook) {
+        compactionStatusHook = hook;
+    }
+
     TestingHook<> readVBStateFromDiskHook;
 
     std::function<int(VB::Commit&, kvstats_ctx&)> saveDocsErrorInjector;

engines/ep/tests/module_tests/magma_bucket_tests.cc

@@ -662,6 +662,69 @@ TEST_P(STParamMagmaBucketTest, MB_47566) {
     compactionThread.join();
 }
 
+/**
+ * Test that when we fail a CompactKVStore call we update stats appropriately
+ */
+TEST_P(STParamMagmaBucketTest, FailCompactKVStoreCall) {
+    const auto& config = store->getRWUnderlying(vbid)->getConfig();
+    auto& nonConstConfig = const_cast<KVStoreConfig&>(config);
+    replaceMagmaKVStore(dynamic_cast<MagmaKVStoreConfig&>(nonConstConfig));
+
+    setVBucketStateAndRunPersistTask(vbid, vbucket_state_active);
+
+    CollectionsManifest cm;
+    setCollections(cookie, cm.add(CollectionEntry::fruit));
+    setCollections(cookie, cm.add(CollectionEntry::meat));
+    flushVBucketToDiskIfPersistent(vbid, 2);
+
+    ASSERT_TRUE(store_items(
+            1, vbid, StoredDocKey{"f", CollectionEntry::fruit}, "value"));
+    ASSERT_TRUE(store_items(
+            1, vbid, StoredDocKey{"m", CollectionEntry::meat}, "value"));
+    flushVBucketToDiskIfPersistent(vbid, 2);
+
+    // 2 items, 1 in each collection
+    auto vb = store->getVBucket(vbid);
+    ASSERT_EQ(2, vb->getNumTotalItems());
+
+    setCollections(cookie, cm.remove(CollectionEntry::fruit));
+    setCollections(cookie, cm.remove(CollectionEntry::meat));
+    flushVBucketToDiskIfPersistent(vbid, 2);
+
+    // Still 2 items, waiting for purge
+    ASSERT_EQ(2, vb->getNumTotalItems());
+
+    auto* kvstore = store->getRWUnderlying(vbid);
+    ASSERT_TRUE(kvstore);
+    auto& magmaKVStore = static_cast<MockMagmaKVStore&>(*kvstore);
+
+    // "Fail" the second compaction. This is a sort of "soft" failure as the
+    // magma portion works but we're going to pretend that it doesn't and
+    // skip updating state based on that.
+    bool first = true;
+    magmaKVStore.setCompactionStatusHook([&first](magma::Status& status) {
+        if (first) {
+            first = false;
+        } else {
+            status = magma::Status(magma::Status::Code::IOError, "bad");
+        }
+    });
+
+    // Compaction for one KVStore passes and another "fails". Given we update
+    // stats with the dropped stats docs on success we'll check the vb item
+    // count to check how many items were "purged".
+    runCompaction(vbid);
+    EXPECT_EQ(1, vb->getNumTotalItems());
+
+    // Our hook wouldn't do anything now, but reset it anyway for simplicity
+    // and run the compaction again allowing the other collection to compact.
+    magmaKVStore.setCompactionStatusHook([](magma::Status&) {});
+    runCompaction(vbid);
+
+    // Items all gone, before the fix 1 would remain
+    EXPECT_EQ(0, vb->getNumTotalItems());
+}
+
 INSTANTIATE_TEST_SUITE_P(STParamMagmaBucketTest,
                          STParamMagmaBucketTest,
                          STParameterizedBucketTest::magmaConfigValues(),