MB-41321: Fix a regression of collection persisted high-seqno

The high-persisted seqno should only change for committed items,
part 1 of MB-41321 regressed that yet was undetected.

Update Flush to skip prepares in stat updates and add checks
in sync-write collection tests.

Change-Id: I4fc91ecbf5b93358686fcf553f01f5d9f896826b
Reviewed-on: http://review.couchbase.org/c/kv_engine/+/137967
Reviewed-by: Dave Rigby <[email protected]>
Tested-by: Build Bot <[email protected]>

Author: jimwwalker

engines/ep/src/collections/flush.cc

@@ -58,36 +58,25 @@ void Flush::StatisticsUpdate::updateDiskSize(ssize_t delta) {
 }
 
 void Flush::StatisticsUpdate::insert(bool isSystem,
-                                     bool isCommitted,
                                      bool isDelete,
                                      ssize_t diskSizeDelta) {
-    if (!isSystem && !isDelete && isCommitted) {
+    if (!isSystem && !isDelete) {
         incrementItemCount();
-    } // else inserting a tombstone or it's a prepare
+    } // else inserting a tombstone - no item increment
 
-    if (isSystem || isCommitted) {
-        updateDiskSize(diskSizeDelta);
-    }
+    updateDiskSize(diskSizeDelta);
 }
 
-void Flush::StatisticsUpdate::update(bool isSystem,
-                                     bool isCommitted,
-                                     ssize_t diskSizeDelta) {
-    if (isSystem || isCommitted) {
-        updateDiskSize(diskSizeDelta);
-    }
+void Flush::StatisticsUpdate::update(ssize_t diskSizeDelta) {
+    updateDiskSize(diskSizeDelta);
 }
 
 void Flush::StatisticsUpdate::remove(bool isSystem,
-                                     bool isCommitted,
                                      ssize_t diskSizeDelta) {
-    if (isCommitted) {
+    if (!isSystem) {
         decrementItemCount();
-    } // else inserting a tombstone or it's a prepare
-
-    if (isSystem || isCommitted) {
-        updateDiskSize(diskSizeDelta);
     }
+    updateDiskSize(diskSizeDelta);
 }
 
 // Called from KVStore during the flush process and before we consider the
@@ -195,6 +184,10 @@ void Flush::updateStats(const DocKey& key,
                         bool isCommitted,
                         bool isDelete,
                         size_t size) {
+    // Prepares don't change the stats
+    if (!isCommitted) {
+        return;
+    }
     auto [isSystemEvent, cid] = getCollectionID(key);
 
     if (!cid || isLogicallyDeleted(cid.value(), seqno)) {
@@ -209,7 +202,7 @@ void Flush::updateStats(const DocKey& key,
     }
 
     getStatsAndMaybeSetPersistedHighSeqno(cid.value(), seqno)
-            .insert(isSystemEvent, isCommitted, isDelete, size);
+            .insert(isSystemEvent, isDelete, size);
 }
 
 void Flush::updateStats(const DocKey& key,
@@ -220,6 +213,10 @@ void Flush::updateStats(const DocKey& key,
                         uint64_t oldSeqno,
                         bool oldIsDelete,
                         size_t oldSize) {
+    // Prepares don't change the stats
+    if (!isCommitted) {
+        return;
+    }
     // Same logic and comment as updateStats above.
     auto [isSystemEvent, cid] = getCollectionID(key);
     if (!cid || isLogicallyDeleted(cid.value(), seqno)) {
@@ -239,11 +236,11 @@ void Flush::updateStats(const DocKey& key,
     auto& stats = getStatsAndMaybeSetPersistedHighSeqno(cid.value(), seqno);
 
     if (oldIsDelete) {
-        stats.insert(isSystemEvent, isCommitted, isDelete, size);
+        stats.insert(isSystemEvent, isDelete, size);
     } else if (!oldIsDelete && isDelete) {
-        stats.remove(isSystemEvent, isCommitted, size - oldSize);
+        stats.remove(isSystemEvent, size - oldSize);
     } else {
-        stats.update(isSystemEvent, isCommitted, size - oldSize);
+        stats.update(size - oldSize);
     }
 }
 

engines/ep/src/collections/flush.h

@@ -257,40 +257,32 @@ class Flush {
         /**
          * Process an insert into the collection
          * @param isSystem true if a system event is inserted
-         * @param isCommitted true if a committed item is inserted, false for
-         **       prepare/abort
          * @param isDelete true if a deleted item is inserted (tombstone
          *        creation)
          * @param diskSize size in bytes 'inserted' into disk. Should be
          *        representative of the bytes used by each document, but does
          *        not need to be exact.
          */
         void insert(bool isSystem,
-                    bool isCommitted,
                     bool isDelete,
                     ssize_t diskSize);
 
         /**
          * Process an update into the collection
-         * @param isSystem true if a system event is updated
-         * @param isCommitted true if a committed item is updated, false for
-         **       prepare/abort
          * @param diskSizeDelta size in bytes difference. Should be
          *        representative of the difference between existing and new
          *        documents, but does not need to be exact.
          */
-        void update(bool isSystem, bool isCommitted, ssize_t diskSizeDelta);
+        void update(ssize_t diskSizeDelta);
 
         /**
          * Process a remove from the collection (store of a delete)
          * @param isSystem true if a system event is removed
-         * @param isCommitted true if a committed item is removed, false for
-         **       prepare/abort
          * @param diskSizeDelta size in bytes difference. Should be
          *        representative of the difference between existing and new
          *        documents, but does not need to be exact.
          */
-        void remove(bool isSystem, bool isCommitted, ssize_t diskSizeDelta);
+        void remove(bool isSystem, ssize_t diskSizeDelta);
 
         /// @returns the highest persisted seqno recorded by the Flush object
         uint64_t getPersistedHighSeqno() const {

engines/ep/tests/module_tests/collections/evp_store_collections_eraser_test.cc

@@ -952,10 +952,16 @@ class CollectionsEraserSyncWriteTest : public CollectionsEraserTest {
     }
 
     void createPendingWrite(bool deleted = false) {
+        auto persistedHighSeqno = vb->lockCollections().getPersistedHighSeqno(
+                key.getCollectionID());
+
         if (deleted) {
             auto item = makeCommittedItem(key, "value");
             EXPECT_EQ(ENGINE_SUCCESS, store->set(*item, cookie));
             flushVBucketToDiskIfPersistent(vbid, 1);
+            // Delete changed the stats
+            persistedHighSeqno = vb->lockCollections().getPersistedHighSeqno(
+                    key.getCollectionID());
 
             mutation_descr_t delInfo;
             uint64_t cas = item->getCas();
@@ -973,6 +979,12 @@ class CollectionsEraserSyncWriteTest : public CollectionsEraserTest {
             EXPECT_EQ(ENGINE_SYNC_WRITE_PENDING, store->set(*item, cookie));
         }
         flushVBucketToDiskIfPersistent(vbid, 1);
+
+        // Validate the collection high-persisted-seqno didn't change for the
+        // pending item
+        EXPECT_EQ(persistedHighSeqno,
+                  vb->lockCollections().getPersistedHighSeqno(
+                          key.getCollectionID()));
     }
 
     void commit() {