MB-37546: Don't infer the result of KVStore::commit by rejectQueue.size

The current logic checks the size of the rejectQueue for inferring if
the call to KVStore::commit succeeded or failed. Just use the result of
KVStore::commit instead.

Change-Id: Ifc612d87df8ddae46c1959959659ce3b10efea68
Reviewed-on: http://review.couchbase.org/121062
Well-Formed: Build Bot <build@couchbase.com>
Tested-by: Build Bot <build@couchbase.com>
Reviewed-by: James Harrison <james.harrison@couchbase.com>

engines/ep/src/ep_bucket.cc

@@ -695,8 +695,12 @@ std::pair<bool, size_t> EPBucket::flushVBucket(Vbid vbid) {
     }
 
     // Persist the flush-batch if not empty.
+    // The result of the flush is defaulted to true to comply the current logic
+    // that considers flush-success/no-flush equivalent.
+    auto flushSuccessOrNoFlush = true;
     if (items_flushed > 0) {
-        commit(vb->getId(), *rwUnderlying, collectionFlush);
+        flushSuccessOrNoFlush =
+                commit(vb->getId(), *rwUnderlying, collectionFlush);
 
         // @todo MB-37693: Commit could fail, do this only if flush-success.
         // Now the commit is complete, vBucket file must exist.
@@ -705,7 +709,9 @@ std::pair<bool, size_t> EPBucket::flushVBucket(Vbid vbid) {
         }
     }
 
-    if (vb->rejectQueue.empty()) {
+    if (flushSuccessOrNoFlush) {
+        Expects(vb->rejectQueue.empty());
+
         // only update the snapshot range if items were flushed, i.e.
         // don't appear to be in a snapshot when you have no data for it
         if (range) {
@@ -787,33 +793,38 @@ std::pair<bool, size_t> EPBucket::flushVBucket(Vbid vbid) {
     // just after getItemsToPersist(). See above for details.
 
     // Handle HighPriority requests
-    if (vb->rejectQueue.empty()) {
+    if (flushSuccessOrNoFlush) {
+        Expects(vb->rejectQueue.empty());
+
         handleCheckpointPersistence();
         vb->notifyHighPriorityRequests(
                 engine, vb->getPersistenceSeqno(), HighPriorityVBNotify::Seqno);
 
         return {toFlush.moreAvailable, items_flushed};
     }
 
-    // Reject queue not empty
+    // Flush has failed
+    Expects(!vb->rejectQueue.empty());
+
     return {true, items_flushed};
 }
 
 void EPBucket::setFlusherBatchSplitTrigger(size_t limit) {
     flusherBatchSplitTrigger = limit;
 }
 
-void EPBucket::commit(Vbid vbid,
+bool EPBucket::commit(Vbid vbid,
                       KVStore& kvstore,
                       Collections::VB::Flush& collectionsFlush) {
     BlockTimer timer(&stats.diskCommitHisto, "disk_commit", stats.timingLog);
     auto commit_start = std::chrono::steady_clock::now();
 
-    if (!kvstore.commit(collectionsFlush)) {
+    const auto res = kvstore.commit(collectionsFlush);
+    if (res) {
+        ++stats.flusherCommits;
+    } else {
         ++stats.commitFailed;
         EP_LOG_WARN("KVBucket::commit: kvstore.commit failed {}", vbid);
-    } else {
-        ++stats.flusherCommits;
     }
 
     auto commit_end = std::chrono::steady_clock::now();
@@ -822,6 +833,8 @@ void EPBucket::commit(Vbid vbid,
                                .count();
     stats.commit_time.store(commit_time);
     stats.cumulativeCommitTime.fetch_add(commit_time);
+
+    return res;
 }
 
 void EPBucket::startFlusher() {

engines/ep/src/ep_bucket.h

@@ -55,7 +55,15 @@ class EPBucket : public KVBucket {
      */
     void setFlusherBatchSplitTrigger(size_t limit);
 
-    void commit(Vbid vbid,
+    /**
+     * Persist whatever flush-batch previously queued into KVStore.
+     *
+     * @param vbid
+     * @param kvstore
+     * @param [out] collectionsFlush
+     * @return true if flush succeeds, false otherwise
+     */
+    bool commit(Vbid vbid,
                 KVStore& kvstore,
                 Collections::VB::Flush& collectionsFlush);