MB-24376: Stop rangeRead on invalid Seqno

An exception is thrown if the start of the readRange is updated with a
value x < 0 || x > endSeqno, as this should not be done.

When updating an item which is currently covered by a rangeRead, the
following states occur in the seqList:

 A₁ B₂ C₃        Initial items
[A₁ B₂ C₃]       rangeRead to highSeqno

{ Write Lock

[A₁ B₂ C₃] A₋₁   ** Update; A₁ marked stale, new OSV appended.
                 NB: this is before queueDirty updates seqNo

[A₁ B₂ C₃] A₄    Seqno updated in queueDirty

}

RangeReads do not hold the write lock, and stop only upon either
reaching the end of the seqList, or reading an item with
seqno > endSeqno.
There is a brief window at the state marked ** in which a rangeRead
may try to update the start of the readRange with -1:

- rangeRead (endSeqno 3) reads C₃ normally
- proceeds to next item in the seqList
- checks if seqno > endSeqno - !(-1 > 3), don't stop yet
- update the start of the readRange with -1

This patch ends the readRange if the current item has a seqno < 0.

Change-Id: Ia1e53e4c40705b673c1f01e1b15fecbed5dff2fc
Reviewed-on: http://review.couchbase.org/78353
Tested-by: Build Bot <build@couchbase.com>
Reviewed-by: Manu Dhundi <manu@couchbase.com>
Reviewed-by: Dave Rigby <daver@couchbase.com>

diff --git a/src/linked_list.cc b/src/linked_list.cc
index b7c8a9b..95402fe 100644 (file)
--- a/src/linked_list.cc
+++ b/src/linked_list.cc
@@ -123,19 +123,22 @@ BasicLinkedList::rangeRead(seqno_t start, seqno_t end) {
     std::vector<UniqueItemPtr> items;
 
     for (const auto& osv : seqList) {
     std::vector<UniqueItemPtr> items;
 
     for (const auto& osv : seqList) {

-        if (osv.getBySeqno() > end) {
-            /* we are done */
+        int64_t currSeqno(osv.getBySeqno());
+
+        if (currSeqno > end || currSeqno < 0) {
+            /* We have read all the items in the requested range, or the osv
+             * does not yet have a valid seqno; either way we are done */

             break;
         }
 
         {
             std::lock_guard<SpinLock> lh(rangeLock);
             break;
         }
 
         {
             std::lock_guard<SpinLock> lh(rangeLock);

-            readRange.setBegin(osv.getBySeqno()); /* [EPHE TODO]: should we
+            readRange.setBegin(currSeqno); /* [EPHE TODO]: should we

                                                      update the min every time ?
                                                    */
         }
 
                                                      update the min every time ?
                                                    */
         }
 

-        if (osv.getBySeqno() < start) {
+        if (currSeqno < start) {

             /* skip this item */
             continue;
         }
             /* skip this item */
             continue;
         }


@@ -166,7 +169,7 @@ BasicLinkedList::rangeRead(seqno_t start, seqno_t end) {
                 "(vb %d) ENOMEM while trying to copy "
                 "item with seqno %" PRIi64 "before streaming it",
                 vbid,
                 "(vb %d) ENOMEM while trying to copy "
                 "item with seqno %" PRIi64 "before streaming it",
                 vbid,

-                osv.getBySeqno());
+                currSeqno);

             return std::make_tuple(ENGINE_ENOMEM, std::move(empty), 0);
         }
     }
             return std::make_tuple(ENGINE_ENOMEM, std::move(empty), 0);
         }
     }

diff --git a/tests/module_tests/ephemeral_vb_test.cc b/tests/module_tests/ephemeral_vb_test.cc
index 46626ce..7359729 100644 (file)
--- a/tests/module_tests/ephemeral_vb_test.cc
+++ b/tests/module_tests/ephemeral_vb_test.cc
@@ -677,3 +677,31 @@ TEST_F(EphemeralVBucketTest, SnapshotHasNoDuplicatesWithMultipleStale) {
     EXPECT_EQ(numItems * (updateIterations + 1),
               std::get<2>(res)); // extended end of readRange
 }
     EXPECT_EQ(numItems * (updateIterations + 1),
               std::get<2>(res)); // extended end of readRange
 }

+
+TEST_F(EphemeralVBucketTest, RangeReadStopsOnInvalidSeqno) {
+    /* MB-24376: rangeRead has to stop if it encounters an OSV with a seqno of
+     * -1; this item is definitely past the end of the rangeRead, and has not
+     * yet had its seqno updated in queueDirty */
+    const int numItems = 2;
+
+    // store two items
+    auto keys = generateKeys(numItems);
+    setMany(keys, MutationStatus::WasClean);
+
+    auto lastKey = makeStoredDocKey(std::to_string(numItems + 1));
+    Item i(lastKey, 0, 0, lastKey.data(), lastKey.size());
+
+    // set item with no queueItemCtx - will not be queueDirty'd, and will
+    // keep seqno -1
+    EXPECT_EQ(MutationStatus::WasClean,
+              public_processSet(i, i.getCas(), false));
+
+    EXPECT_EQ(-1, mockEpheVB->getLL()->getSeqList().back().getBySeqno());
+
+    auto res = mockEpheVB->inMemoryBackfill(
+            1, std::numeric_limits<seqno_t>::max());
+
+    EXPECT_EQ(ENGINE_SUCCESS, std::get<0>(res));
+    EXPECT_EQ(numItems, std::get<1>(res).size());
+    EXPECT_EQ(numItems, std::get<2>(res));
+}
\ No newline at end of file

diff --git a/tests/module_tests/vbucket_test.cc b/tests/module_tests/vbucket_test.cc
index b54fb03..0b94377 100644 (file)
--- a/tests/module_tests/vbucket_test.cc
+++ b/tests/module_tests/vbucket_test.cc
@@ -122,7 +122,9 @@ std::pair<HashTable::HashBucketLock, StoredValue*> VBucketTest::lockAndFind(
     return std::make_pair(std::move(hbl), storedVal);
 }
 
     return std::make_pair(std::move(hbl), storedVal);
 }
 

-MutationStatus VBucketTest::public_processSet(Item& itm, const uint64_t cas) {
+MutationStatus VBucketTest::public_processSet(Item& itm,
+                                              const uint64_t cas,
+                                              bool withCtx) {

     auto hbl_sv = lockAndFind(itm.getKey());
     VBQueueItemCtx queueItmCtx(GenerateBySeqno::Yes,
                                GenerateCas::No,
     auto hbl_sv = lockAndFind(itm.getKey());
     VBQueueItemCtx queueItmCtx(GenerateBySeqno::Yes,
                                GenerateCas::No,


@@ -136,7 +138,7 @@ MutationStatus VBucketTest::public_processSet(Item& itm, const uint64_t cas) {
                          cas,
                          true,
                          false,
                          cas,
                          true,
                          false,

-                         &queueItmCtx)
+                         withCtx ? &queueItmCtx : nullptr)

             .first;
 }
 
             .first;
 }
 

diff --git a/tests/module_tests/vbucket_test.h b/tests/module_tests/vbucket_test.h
index 9f697a1..1ef4e56 100644 (file)
--- a/tests/module_tests/vbucket_test.h
+++ b/tests/module_tests/vbucket_test.h
@@ -76,7 +76,9 @@ protected:
     std::pair<HashTable::HashBucketLock, StoredValue*> lockAndFind(
             const StoredDocKey& key);
 
     std::pair<HashTable::HashBucketLock, StoredValue*> lockAndFind(
             const StoredDocKey& key);
 

-    MutationStatus public_processSet(Item& itm, const uint64_t cas);
+    MutationStatus public_processSet(Item& itm,
+                                     const uint64_t cas,
+                                     bool withCtx = true);

 
     AddStatus public_processAdd(Item& itm);
 
 
     AddStatus public_processAdd(Item& itm);