MB-35908: Compare cas instead of seqno when deduping setVBState

The flusher currently compares the seqnos of set_vbucket_state
items to determine which is the most recent vbucket state (i.e. the
one that should be persisted). This is an issue as set_vbucket_state
operations do not update the lastBySeqno of the vBucket. This means
that subsequent set_vbucket_state items may have the same seqno.
It's not valid to udpate the lastBySeqno of the vBucket as the
replica seqnos will be different, consider setting the failover
table on DCP stream acceptance, and DCP Consumers may expect to see
a seqno that will never be sent.

Compare the CAS of the set_vbucket_state items to determine which
is the most recent instead of the seqno as the CAS will be unique.

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

Author: BenHuddleston

engines/ep/src/checkpoint_manager.cc

@@ -787,6 +787,12 @@ void CheckpointManager::queueSetVBState(VBucket& vb) {
     queued_item item = createCheckpointItem(/*id*/0, vbucketId,
                                             queue_op::set_vbucket_state);
 
+    // We need to set the cas of the item as two subsequent set_vbucket_state
+    // items will have the same seqno and the flusher needs a way to determine
+    // which is the latest so that we persist the correct state.
+    // We do this 'atomically' as we are holding the ::queueLock.
+    item->setCas(vb.nextHLCCas());
+
     // Store a JSON version of the vbucket transition data in the value
     vbstate.toItem(*item);
 
@@ -1184,6 +1190,14 @@ queued_item CheckpointManager::createCheckpointItem(uint64_t id,
         bySeqno = lastBySeqno;
         break;
     case queue_op::set_vbucket_state:
+        // It's not valid to actually increment lastBySeqno for a
+        // set_vbucket_state for two reasons:
+        // 1) This may be called independently on the replica to the active
+        // (i.e. for a failover table change) so the seqnos would differ to
+        // those on the active.
+        // 2) DcpConsumer calling getAllVBucketSeqnos would expect to see a
+        // seqno that will never be sent to them if the last item queued is a
+        // set_vbucket_state.
         bySeqno = lastBySeqno + 1;
         break;
 

engines/ep/src/ep_bucket.cc

@@ -397,7 +397,7 @@ std::pair<bool, size_t> EPBucket::flushVBucket(Vbid vbid) {
             }
             // We need to set a few values from the in-memory state.
             uint64_t maxSeqno = 0;
-            uint64_t maxVbStateOperation = 0;
+            uint64_t maxVbStateOpCas = 0;
 
             auto minSeqno = std::numeric_limits<uint64_t>::max();
 
@@ -460,21 +460,31 @@ std::pair<bool, size_t> EPBucket::flushVBucket(Vbid vbid) {
                 }
 
                 if (op == queue_op::set_vbucket_state) {
-                    // Only process vbstate if it's sequenced higher
-                    if (static_cast<uint64_t>(item->getBySeqno()) >
-                        maxVbStateOperation) {
-                        maxVbStateOperation = item->getBySeqno();
+                    // Only process vbstate if it's sequenced higher (by cas).
+                    // We use the cas instead of the seqno here because a
+                    // set_vbucket_state does not increment the lastBySeqno in
+                    // the CheckpointManager when it is created. This means that
+                    // it is possible to have two set_vbucket_state items that
+                    // follow one another with the same seqno. The cas will be
+                    // bumped for every item so it can be used to distinguish
+                    // which item is the latest and should be flushed.
+                    if (item->getCas() > maxVbStateOpCas) {
+                        // Should only bump the stat once for the latest state
+                        // change that we want to flush
+                        if (maxVbStateOpCas == 0) {
+                            // There is at least a commit to be done, so
+                            // increase todo
+                            ++stats.flusher_todo;
+                        }
+
+                        maxVbStateOpCas = item->getCas();
 
                         // It could be the case that the set_vbucket_state is
                         // alone, i.e. no mutations are being flushed, we must
-                        // trigger an update of the vbstatem which will always
+                        // trigger an update of the vbstate, which will always
                         // happen when we set this.
                         mustCheckpointVBState = true;
 
-                        // There is at least a commit to be done, so increase
-                        // todo
-                        ++stats.flusher_todo;
-
                         // Process the Item's value into the transition struct
                         vbstate.transition.fromItem(*item);
                     }

engines/ep/tests/module_tests/evp_store_warmup_test.cc

@@ -26,6 +26,7 @@
 #include "ep_time.h"
 #include "evp_store_durability_test.h"
 #include "evp_store_single_threaded_test.h"
+#include "failover-table.h"
 #include "kvstore.h"
 #include "programs/engine_testapp/mock_server.h"
 #include "test_helpers.h"
@@ -551,6 +552,35 @@ TEST_F(WarmupTest, SetVBState) {
     EXPECT_EQ(vbucket_state_replica, store->getVBucket(vbid)->getState());
 }
 
+TEST_F(WarmupTest, TwoStateChangesAtSameSeqno) {
+    // 1) Do a normal state change to replica
+    EXPECT_EQ(ENGINE_SUCCESS,
+              store->setVBucketState(vbid, vbucket_state_replica));
+
+    // 2) Make a change to the failover table as we would if we had just created
+    // a stream to an active node
+    auto failoverTableLastSeqno = 999;
+    { // VBPtr scope (not valid after warmup)
+        auto vb = store->getVBucket(vbid);
+        vb->failovers->createEntry(failoverTableLastSeqno);
+        EXPECT_EQ(2, vb->failovers->getNumEntries());
+        EXPECT_EQ(failoverTableLastSeqno,
+                  vb->failovers->getLatestEntry().by_seqno);
+    }
+    store->scheduleVBStatePersist(vbid);
+
+    // 3) Flush the two state changes together and warmup so that we can more
+    // easily read what was persisted
+    flush_vbucket_to_disk(vbid, 0);
+    resetEngineAndWarmup();
+
+    // Test
+    auto vb = store->getVBucket(vbid);
+    auto state = store->getVBucket(vbid)->getTransitionState();
+    EXPECT_EQ(vbucket_state_replica, vb->getState());
+    EXPECT_EQ(failoverTableLastSeqno, vb->failovers->getLatestEntry().by_seqno);
+}
+
 // Test uses a sequence of operations that will mean delete is de-duplicated
 // resulting in the flusher missing the update to the on-disk max-deleted
 // revision.