MB-35133: Rollback SyncWrites

Rollback SyncWrites by performing a normal Rollback and remove all
items in the prepare namespace. Do a "warmup" of the SyncWrites to
correct the HashTable and PDM state post-rollback.

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

Author: BenHuddleston

engines/ep/src/ep_bucket.cc

@@ -1206,8 +1206,20 @@ class EPDiskRollbackCB : public RollbackCB {
         // This is the item in its current state, after the rollback seqno
         // (i.e. the state that we are reverting)
         UniqueItemPtr postRbSeqnoItem(std::move(val.item));
+
         VBucketPtr vb = engine.getVBucket(postRbSeqnoItem->getVBucketId());
 
+        // Nuke anything in the prepare namespace, we'll do a "warmup" later
+        // which will restore everything to the way it should be and this is
+        // far easier than dealing with individual states.
+        if (postRbSeqnoItem->isPending() || postRbSeqnoItem->isAbort()) {
+            removeDeletedDoc(*vb, *postRbSeqnoItem);
+            return;
+        }
+
+        EP_LOG_DEBUG("EPDiskRollbackCB: Handling post rollback item: {}",
+                     *postRbSeqnoItem);
+
         // The get value of the item before the rollback seqno
         GetValue preRbSeqnoGetValue =
                 engine.getKVBucket()
@@ -1216,10 +1228,18 @@ class EPDiskRollbackCB : public RollbackCB {
                                         DiskDocKey{*postRbSeqnoItem},
                                         postRbSeqnoItem->getVBucketId(),
                                         GetMetaOnly::No);
+
+        // This is the item in the state it was before the rollback seqno
+        // (i.e. the desired state). null if there was no previous
+        // Item.
+        UniqueItemPtr preRbSeqnoItem(std::move(preRbSeqnoGetValue.item));
+
         if (preRbSeqnoGetValue.getStatus() == ENGINE_SUCCESS) {
-            // This is the item in the state it was before the rollback seqno
-            // (i.e. the desired state)
-            UniqueItemPtr preRbSeqnoItem(std::move(preRbSeqnoGetValue.item));
+            EP_LOG_DEBUG(
+                    "EPDiskRollbackCB: Item existed pre-rollback; restoring to "
+                    "pre-rollback state: {}",
+                    *preRbSeqnoItem);
+
             if (preRbSeqnoItem->isDeleted()) {
                 // If the item existed before, but had been deleted, we
                 // should delete it now
@@ -1228,9 +1248,33 @@ class EPDiskRollbackCB : public RollbackCB {
                 // The item existed before and was not deleted, we need to
                 // revert the items state to the preRollbackSeqno state
                 MutationStatus mtype = vb->setFromInternal(*preRbSeqnoItem);
-
-                if (mtype == MutationStatus::NoMem) {
+                switch (mtype) {
+                case MutationStatus::NotFound:
+                    // NotFound is valid - if the item has been deleted
+                    // in-memory, but that was not flushed to disk as of
+                    // post-rollback seqno.
+                    break;
+                case MutationStatus::WasClean:
+                    // Item hasn't been modified since it was persisted to disk
+                    // as of post-rollback seqno.
+                    break;
+                case MutationStatus::WasDirty:
+                    // Item was modifed since it was persisted to disk - this
+                    // is ok because it's just a mutation which has not yet
+                    // been persisted to disk as of post-rollback seqno.
+                    break;
+                case MutationStatus::NoMem:
                     setStatus(ENGINE_ENOMEM);
+                    break;
+                case MutationStatus::InvalidCas:
+                case MutationStatus::IsLocked:
+                case MutationStatus::NeedBgFetch:
+                case MutationStatus::IsPendingSyncWrite:
+                    std::stringstream ss;
+                    ss << "EPDiskRollbackCB: Unexpected status:"
+                       << to_string(mtype)
+                       << " after setFromInternal for item:" << *preRbSeqnoItem;
+                    throw std::logic_error(ss.str());
                 }
 
                 // If we are rolling back a deletion then we should increment
@@ -1244,50 +1288,17 @@ class EPDiskRollbackCB : public RollbackCB {
                             .lock(preRbSeqnoItem->getKey())
                             .incrementDiskCount();
                 }
-
-                if (postRbSeqnoItem->isCommitSyncWrite() ||
-                    postRbSeqnoItem->isAbort()) {
-                    rollbackSyncWrite(*vb, *postRbSeqnoItem);
-                }
             }
         } else if (preRbSeqnoGetValue.getStatus() == ENGINE_KEY_ENOENT) {
             // If the item did not exist before we should delete it now
             removeDeletedDoc(*vb, *postRbSeqnoItem);
-
-            // If we are rolling back a commit that did not exist on disk
-            // before then we need to requeue a prepare
-            if (postRbSeqnoItem->isCommitSyncWrite()) {
-                rollbackSyncWrite(*vb, *postRbSeqnoItem);
-            }
         } else {
             EP_LOG_WARN(
                     "EPDiskRollbackCB::callback:Unexpected Error Status: {}",
                     preRbSeqnoGetValue.getStatus());
         }
     }
 
-    void rollbackSyncWrite(VBucket& vb, const Item& postRbSeqnoItem) {
-        // Search for prepare in the pre-rollback header
-        GetValue preRbSeqnoPrepare =
-                engine.getKVBucket()
-                        ->getROUnderlying(postRbSeqnoItem.getVBucketId())
-                        ->getWithHeader(dbHandle,
-                                        DiskDocKey(postRbSeqnoItem.getKey(),
-                                                   true /*prepare*/),
-                                        postRbSeqnoItem.getVBucketId(),
-                                        GetMetaOnly::No);
-        if (preRbSeqnoPrepare.getStatus() == ENGINE_SUCCESS &&
-            !preRbSeqnoPrepare.item->isDeleted()) {
-            // Exists before so we are rolling back to prepare
-            vb.addSyncWriteForRollback(*preRbSeqnoPrepare.item);
-
-            // Store the prepare we need to add back in to the Passive DM for
-            // post-rollback processing.
-            queued_item qi(std::make_unique<Item>(*preRbSeqnoPrepare.item));
-            preparesToAdd.push_back(qi);
-        }
-    }
-
     /// Remove a deleted-on-disk document from the VBucket's hashtable.
     void removeDeletedDoc(VBucket& vb, const Item& item) {
         if (vb.removeItemFromMemory(item)) {
@@ -1318,7 +1329,6 @@ RollbackResult EPBucket::doRollback(Vbid vbid, uint64_t rollbackSeqno) {
     auto cb = std::make_shared<EPDiskRollbackCB>(engine);
     KVStore* rwUnderlying = vbMap.getShardByVbId(vbid)->getRWUnderlying();
     auto result = rwUnderlying->rollback(vbid, rollbackSeqno, cb);
-    result.preparesToAdd = cb->preparesToAdd;
     return result;
 }
 

engines/ep/src/hash_table.cc

@@ -897,14 +897,14 @@ void HashTable::dump() const {
 
 nlohmann::json HashTable::dumpStoredValuesAsJson() const {
     MultiLockHolder mlh(mutexes);
-    auto obj = nlohmann::json::object();
+    auto obj = nlohmann::json::array();
     for (const auto& chain : values) {
         if (chain) {
             for (StoredValue* sv = chain.get().get(); sv != nullptr;
                  sv = sv->getNext().get().get()) {
                 std::stringstream ss;
                 ss << sv->getKey();
-                obj[ss.str()] = *sv;
+                obj.push_back(*sv);
             }
         }
     }

engines/ep/src/kv_bucket.cc

@@ -2420,20 +2420,14 @@ TaskStatus KVBucket::rollback(Vbid vbid, uint64_t rollbackSeqno) {
         return TaskStatus::Abort;
     }
 
-    folly::SharedMutex::ReadHolder rlh(vb->getStateLock());
+    // Acquire the vb stateLock in exclusive mode as we will recreate the
+    // DurabilityMonitor in the vBucket as part of rollback and this could race
+    // with stats calls.
+    folly::SharedMutex::WriteHolder wlh(vb->getStateLock());
     if ((vb->getState() == vbucket_state_replica) ||
         (vb->getState() == vbucket_state_pending)) {
         uint64_t prevHighSeqno =
                 static_cast<uint64_t>(vb->checkpointManager->getHighSeqno());
-
-        // MB-35060: Cannot reliably rollback if there have been any prepared
-        // SyncWrites in all cases.
-        // Temporarily rollback to zero in this case.
-        // @todo Fix this - see MB-35133
-        if (vb->getSyncWriteAcceptedCount() > 0) {
-            rollbackSeqno = 0;
-        }
-
         if (rollbackSeqno != 0) {
             RollbackResult result = doRollback(vbid, rollbackSeqno);
 
@@ -2444,6 +2438,7 @@ TaskStatus KVBucket::rollback(Vbid vbid, uint64_t rollbackSeqno) {
                                           instead of deleting everything in it
                                         */) {
                 rollbackUnpersistedItems(*vb, result.highSeqno);
+                loadPreparedSyncWrites(wlh, *vb);
                 vb->postProcessRollback(result, prevHighSeqno);
 
                 // And update collections post rollback