MB-24034 [Ephemeral]: Fix incorrect NumDeletedItems after un-delete

In an Ephemeral bucket, if an item is created, deleted, and then
re-created, the numDeletedItem count in the SeqList is incorrect - we
fail to decrement the deleted item count when it's re-created.

Change-Id: Iba9b77be4814ebd81f252c37e4c934c65965532f
Reviewed-on: http://review.couchbase.org/77186
Tested-by: Build Bot <build@couchbase.com>
Reviewed-by: Jim Walker <jim@couchbase.com>

diff --git a/src/ephemeral_vb.cc b/src/ephemeral_vb.cc
index d2ce3ec..87da513 100644 (file)
--- a/src/ephemeral_vb.cc
+++ b/src/ephemeral_vb.cc
@@ -295,7 +295,8 @@ EphemeralVBucket::updateStoredValue(const HashTable::HashBucketLock& hbl,
                                     const VBQueueItemCtx* queueItmCtx) {
     std::lock_guard<std::mutex> lh(sequenceLock);
 
-    const bool recreatingDeletedItem = v.isDeleted();
+    const bool oldValueDeleted = v.isDeleted();
+    const bool recreatingDeletedItem = v.isDeleted() && !itm.isDeleted();
 
     /* Update the OrderedStoredValue in hash table + Ordered data structure
        (list) */
@@ -348,6 +349,8 @@ EphemeralVBucket::updateStoredValue(const HashTable::HashBucketLock& hbl,
         ++opsUpdate;
     }
 
+    seqList->updateNumDeletedItems(oldValueDeleted, itm.isDeleted());
+
     if (res == SequenceList::UpdateStatus::Append) {
         /* Mark the un-updated storedValue as stale. This must be done after
            the new storedvalue for the item is visible for range read in the
@@ -390,6 +393,8 @@ std::pair<StoredValue*, VBNotifyCtx> EphemeralVBucket::addNewStoredValue(
     seqList->updateHighSeqno(*(v->toOrderedStoredValue()));
     ++opsCreate;
 
+    seqList->updateNumDeletedItems(false, itm.isDeleted());
+
     return {v, notifyCtx};
 }
 
@@ -404,6 +409,8 @@ std::tuple<StoredValue*, VBNotifyCtx> EphemeralVBucket::softDeleteStoredValue(
     StoredValue* newSv = &v;
     StoredValue::UniquePtr ownedSv;
 
+    const bool oldValueDeleted = v.isDeleted();
+
     /* Update the OrderedStoredValue in hash table + Ordered data structure
        (list) */
     auto res = seqList->updateListElem(lh, *(v.toOrderedStoredValue()));
@@ -445,6 +452,8 @@ std::tuple<StoredValue*, VBNotifyCtx> EphemeralVBucket::softDeleteStoredValue(
     seqList->updateHighSeqno(*(newSv->toOrderedStoredValue()));
     ++opsDelete;
 
+    seqList->updateNumDeletedItems(oldValueDeleted, true);
+
     if (res == SequenceList::UpdateStatus::Append) {
         /* Mark the un-updated storedValue as stale. This must be done after
            the new storedvalue for the item is visible for range read in the

diff --git a/src/linked_list.cc b/src/linked_list.cc
index 3bff91c..0ae1e6a 100644 (file)
--- a/src/linked_list.cc
+++ b/src/linked_list.cc
@@ -169,9 +169,6 @@ BasicLinkedList::rangeRead(seqno_t start, seqno_t end) {
 void BasicLinkedList::updateHighSeqno(const OrderedStoredValue& v) {
     std::lock_guard<SpinLock> lh(rangeLock);
     highSeqno = v.getBySeqno();
-    if (v.isDeleted()) {
-        ++numDeletedItems;
-    }
 }
 
 void BasicLinkedList::markItemStale(StoredValue::UniquePtr ownedSv) {
@@ -292,6 +289,14 @@ size_t BasicLinkedList::purgeTombstones() {
     return purgedCount;
 }
 
+void BasicLinkedList::updateNumDeletedItems(bool oldDeleted, bool newDeleted) {
+    if (oldDeleted && !newDeleted) {
+        --numDeletedItems;
+    } else if (!oldDeleted && newDeleted) {
+        ++numDeletedItems;
+    }
+}
+
 uint64_t BasicLinkedList::getNumStaleItems() const {
     return numStaleItems;
 }

diff --git a/src/linked_list.h b/src/linked_list.h
index 6ecf898..9fbbcee 100644 (file)
--- a/src/linked_list.h
+++ b/src/linked_list.h
@@ -152,6 +152,8 @@ public:
 
     size_t purgeTombstones() override;
 
+    void updateNumDeletedItems(bool oldDeleted, bool newDeleted) override;
+
     uint64_t getNumStaleItems() const override;
 
     size_t getStaleValueBytes() const override;

diff --git a/src/seqlist.h b/src/seqlist.h
index 10174a9..8cefdbe 100644 (file)
--- a/src/seqlist.h
+++ b/src/seqlist.h
@@ -125,8 +125,6 @@ public:
      * Updates the highSeqno in the list. Since seqno is generated and managed
      * outside the list, the module managing it must update this after the seqno
      * is generated for the item already put in the list.
-     * If the last OrderedStoredValue added to the list is soft deleted then
-     * it also updates the deleted items count in the list
      *
      * @param v Ref to orderedStoredValue
      */
@@ -154,6 +152,15 @@ public:
     virtual size_t purgeTombstones() = 0;
 
     /**
+     * Updates the number of deleted items in the sequence list whenever
+     * an item is modified.
+     *
+     * @param oldDeleted Was the old item deleted?
+     * @param newDeleted Was the new item replacing it deleted?
+     */
+    virtual void updateNumDeletedItems(bool oldDeleted, bool newDeleted) = 0;
+
+    /**
      * Returns the number of stale items in the list.
      *
      * @return count of stale items

diff --git a/tests/module_tests/basic_ll_test.cc b/tests/module_tests/basic_ll_test.cc
index 57e3e45..ff93136 100644 (file)
--- a/tests/module_tests/basic_ll_test.cc
+++ b/tests/module_tests/basic_ll_test.cc
@@ -346,6 +346,7 @@ TEST_F(BasicLinkedListTest, DeletedItem) {
 
     /* Delete the item */
     softDeleteItem(numItems, keyPrefix + std::to_string(numItems));
+    basicLL->updateNumDeletedItems(false, true);
 
     /* Check if the delete is added correctly */
     std::vector<seqno_t> expectedSeqno = {numItems + 1};

diff --git a/tests/module_tests/ephemeral_vb_test.cc b/tests/module_tests/ephemeral_vb_test.cc
index 0b885f1..35d8488 100644 (file)
--- a/tests/module_tests/ephemeral_vb_test.cc
+++ b/tests/module_tests/ephemeral_vb_test.cc
@@ -111,6 +111,36 @@ TEST_F(EphemeralVBucketTest, PageOutAfterDeleteWithValue) {
     EXPECT_FALSE(storedVal->getValue());
 }
 
+// NRU: check the seqlist has correct statistics for a create, pageout,
+// and (re)create of the same key.
+TEST_F(EphemeralVBucketTest, CreatePageoutCreate) {
+    auto key = makeStoredDocKey("key");
+
+    // Add a key, then page out.
+    ASSERT_EQ(AddStatus::Success, addOne(key));
+    {
+        auto lock_sv = lockAndFind(key);
+        EXPECT_TRUE(vbucket->pageOut(lock_sv.first, lock_sv.second));
+    }
+    // Sanity check - should have just the one deleted item.
+    ASSERT_EQ(0, vbucket->getNumItems());
+    ASSERT_EQ(1, mockEpheVB->getLL()->getNumDeletedItems());
+
+    // Test: Set the key again.
+    ASSERT_EQ(MutationStatus::WasDirty, setOne(key));
+
+    EXPECT_EQ(1, vbucket->getNumItems());
+    EXPECT_EQ(0, mockEpheVB->getLL()->getNumDeletedItems());
+
+    // Finally for good measure, delete again and check the numbers are correct.
+    {
+        auto lock_sv = lockAndFind(key);
+        EXPECT_TRUE(vbucket->pageOut(lock_sv.first, lock_sv.second));
+    }
+    EXPECT_EQ(0, vbucket->getNumItems());
+    EXPECT_EQ(1, mockEpheVB->getLL()->getNumDeletedItems());
+}
+
 TEST_F(EphemeralVBucketTest, SetItems) {
     const int numItems = 3;