MB-47386: Don't allow limit=0 arg at VBucket::getItemsToPersist

That is the only code path in the function that makes a O(N) call into
the CheckpointManager.
Code path currently never executed in the production code, so safe to
remove.

Change-Id: I32519c5d633ad3baefbe7a25f3efe721bbacd260
Reviewed-on: http://review.couchbase.org/c/kv_engine/+/159598
Reviewed-by: Dave Rigby <daver@couchbase.com>
Tested-by: Build Bot <build@couchbase.com>

engines/ep/src/vbucket.cc

@@ -407,9 +407,7 @@ VBucket::ItemsToFlush VBucket::getItemsToPersist(size_t approxLimit) {
     ItemsToFlush result;
 
     if (approxLimit == 0) {
-        result.moreAvailable =
-                (checkpointManager->getNumItemsForPersistence() > 0);
-        return result;
+        throw std::invalid_argument("VBucket::getItemsToPersist: limit=0");
     }
 
     // Get up to approxLimit checkpoint items outstanding for the persistence

engines/ep/src/vbucket.h

@@ -576,10 +576,12 @@ class VBucket : public std::enable_shared_from_this<VBucket> {
      * Obtain the series of items to be flushed for this vBucket.
      *
      * @param vb VBucket to fetch items for.
-     * @param approxLimit Upper bound on how many items to fetch.
+     * @param approxLimit Upper bound on how many items to fetch. Must be a
+     *  positive integer
      * @return The items to flush; along with their seqno range and
      *         if more items are available for this vBucket (i.e. the
      *         limit was reached).
+     * @throw std::invalid_argument if the user passes approxLimit=0
      */
     ItemsToFlush getItemsToPersist(size_t approxLimit);
 

engines/ep/tests/module_tests/vbucket_test.cc

@@ -29,6 +29,7 @@
 
 #include "../mock/mock_ephemeral_vb.h"
 
+#include <folly/portability/GMock.h>
 #include <nlohmann/json.hpp>
 #include <platform/cb_malloc.h>
 #include <memory>
@@ -622,22 +623,44 @@ TEST_P(VBucketTest, DISABLED_GetItemsToPersist_Limit) {
     EXPECT_EQ(range.getEnd() + 5, result.ranges.back().getEnd());
 }
 
+TEST_P(VBucketTest, GetItemsToPersist_ZeroLimitThrows) {
+    if (!persistent()) {
+        return;
+    }
+
+    try {
+        vbucket->getItemsToPersist(0);
+    } catch (const std::invalid_argument& e) {
+        EXPECT_THAT(e.what(), testing::HasSubstr("limit=0"));
+        return;
+    }
+    FAIL();
+}
+
 // Check that getItemsToPersist() correctly returns `moreAvailable` if we
 // hit the CheckpointManager limit early.
 TEST_P(VBucketTest, GetItemsToPersist_LimitCkptMoreAvailable) {
     if (!persistent()) {
         return;
     }
 
-    // Setup - Add an item to checkpoint manager (in addition to initial
+    // Setup - Add 2 items to checkpoint manager (in addition to initial
     // checkpoint_start).
+    // The 2 items need to be in different checkpoints as the limit is an
+    // approximate limit.
+    auto& manager = *vbucket->checkpointManager;
+    ASSERT_EQ(1, manager.getOpenCheckpointId());
     ASSERT_EQ(MutationStatus::WasClean, setOne(makeStoredDocKey("1")));
+    manager.createNewCheckpoint();
+    ASSERT_EQ(2, manager.getOpenCheckpointId());
+    ASSERT_EQ(MutationStatus::WasClean, setOne(makeStoredDocKey("2")));
 
     // Test - fetch items such that we have a limit for CheckpointManager of
-    // zero. This should return moreAvailable=true.
-    auto result = this->vbucket->getItemsToPersist(0);
+    // 1. This should return moreAvailable=true.
+    auto result = vbucket->getItemsToPersist(1);
     EXPECT_TRUE(result.moreAvailable);
-    EXPECT_EQ(0, result.items.size());
+    // start + mutation + end
+    EXPECT_EQ(3, result.items.size());
 }
 
 class VBucketEvictionTest : public VBucketTest {};