SERVER-16992 Add failpoints to throw WT_ROLLBACK

For the WT record store and indexes.

Also reverts commit 8e54b64c1d9b47579401411585ea1dfe209cddf9.

(cherry picked from commit 4f510d62cbf9201291910887ee733bde4d7877a1)

4 files changed, 64 insertions, 60 deletions

diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_index.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_index.cpp
index b4ee396f6ae..d728db942fc 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_index.cpp
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_index.cpp
@@ -46,8 +46,9 @@
 #include "mongo/db/storage/wiredtiger/wiredtiger_session_cache.h"
 #include "mongo/db/storage/wiredtiger/wiredtiger_util.h"
 #include "mongo/db/storage_options.h"
-#include "mongo/util/log.h"
 #include "mongo/util/assert_util.h"
+#include "mongo/util/fail_point.h"
+#include "mongo/util/log.h"
 #include "mongo/util/mongoutils/str.h"
 
 #if 0
@@ -330,7 +331,7 @@ namespace {
         WT_CURSOR *c = curwrap.get();
         if (!c)
             return true;
-        int ret = c->next(c);
+        int ret = WT_OP_CHECK(c->next(c));
         if (ret == WT_NOTFOUND)
             return true;
         invariantWTOK(ret);
@@ -349,9 +350,10 @@ namespace {
         KeyString data( key, _ordering );
         WiredTigerItem item( data.getBuffer(), data.getSize() );
         c->set_key( c, item.Get() );
-        int ret = c->search(c);
-        if ( ret == WT_NOTFOUND )
+        int ret = WT_OP_CHECK(c->search(c));
+        if (ret == WT_NOTFOUND) {
             return false;
+        }
         invariantWTOK( ret );
 
         // If the key exists, check if we already have this loc at this key. If so, we don't
@@ -449,7 +451,7 @@ namespace {
 
             _cursor->set_value(_cursor, valueItem.Get());
 
-            invariantWTOK(_cursor->insert(_cursor));
+            invariantWTOK(WT_OP_CHECK(_cursor->insert(_cursor)));
             invariantWTOK(_cursor->reset(_cursor));
 
             return Status::OK();
@@ -543,7 +545,7 @@ namespace {
             _cursor->set_key(_cursor, keyItem.Get());
             _cursor->set_value(_cursor, valueItem.Get());
 
-            invariantWTOK(_cursor->insert(_cursor));
+            invariantWTOK(WT_OP_CHECK(_cursor->insert(_cursor)));
             invariantWTOK(_cursor->reset(_cursor));
 
             _records.clear();
@@ -677,7 +679,7 @@ namespace {
         void advanceWTCursor() {
             invalidateCache();
             WT_CURSOR *c = _cursor.get();
-            int ret = _forward ? c->next(c) : c->prev(c);
+            int ret = WT_OP_CHECK(_forward ? c->next(c) : c->prev(c));
             if ( ret == WT_NOTFOUND ) {
                 _eof = true;
                 return;
@@ -695,7 +697,7 @@ namespace {
             const WiredTigerItem keyItem(_key.getBuffer(), _key.getSize());
             c->set_key(c, keyItem.Get());
 
-            int ret = c->search_near(c, &cmp);
+            int ret = WT_OP_CHECK(c->search_near(c, &cmp));
             if ( ret == WT_NOTFOUND ) {
                 _eof = true;
                 TRACE_CURSOR << "\t not found";
@@ -717,13 +719,13 @@ namespace {
             if (_forward) {
                 // We need to be >=
                 if (cmp < 0) {
-                    ret = c->next(c);
+                    ret = WT_OP_CHECK(c->next(c));
                 }
             }
             else {
                 // We need to be <=
                 if (cmp > 0) {
-                    ret = c->prev(c);
+                    ret = WT_OP_CHECK(c->prev(c));
                 }
             }
 
@@ -1003,7 +1005,7 @@ namespace {
         WiredTigerItem valueItem(value.getBuffer(), value.getSize());
         c->set_key( c, keyItem.Get() );
         c->set_value( c, valueItem.Get() );
-        int ret = c->insert( c );
+        int ret = WT_OP_CHECK(c->insert(c));
 
         if ( ret == WT_ROLLBACK && !dupsAllowed ) {
             // if there is a conflict on a unique key, it means there is a dup key
@@ -1019,7 +1021,7 @@ namespace {
         // we put them all in the "list"
         // Note that we can't omit AllZeros when there are multiple locs for a value. When we remove
         // down to a single value, it will be cleaned up.
-        ret = c->search(c);
+        ret = WT_OP_CHECK(c->search(c));
         invariantWTOK( ret );
 
         WT_ITEM old;
@@ -1069,7 +1071,7 @@ namespace {
 
         if ( !dupsAllowed ) {
             // nice and clear
-            int ret = c->remove(c);
+            int ret = WT_OP_CHECK(c->remove(c));
             if (ret == WT_NOTFOUND) {
                 return;
             }
@@ -1079,7 +1081,7 @@ namespace {
 
         // dups are allowed, so we have to deal with a vector of RecordIds.
 
-        int ret = c->search(c);
+        int ret = WT_OP_CHECK(c->search(c));
         if ( ret == WT_NOTFOUND )
             return;
         invariantWTOK( ret );
@@ -1099,7 +1101,7 @@ namespace {
                 if (records.empty() && !br.remaining()) {
                     // This is the common case: we are removing the only loc for this key.
                     // Remove the whole entry.
-                    invariantWTOK(c->remove(c));
+                    invariantWTOK(WT_OP_CHECK(c->remove(c)));
                     return;
                 }
 
@@ -1171,7 +1173,7 @@ namespace {
 
         c->set_key(c, keyItem.Get());
         c->set_value(c, valueItem.Get());
-        int ret = c->insert( c );
+        int ret = WT_OP_CHECK(c->insert(c));
 
         if ( ret != WT_DUPLICATE_KEY )
             return wtRCToStatus( ret );
@@ -1189,7 +1191,7 @@ namespace {
         KeyString data( key, _ordering, loc );
         WiredTigerItem item( data.getBuffer(), data.getSize() );
         c->set_key(c, item.Get() );
-        int ret = c->remove(c);
+        int ret = WT_OP_CHECK(c->remove(c));
         if (ret != WT_NOTFOUND) {
             invariantWTOK(ret);
         }
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_record_store.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_record_store.cpp
index 2fd16e21cb6..28f6328b22a 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_record_store.cpp
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_record_store.cpp
@@ -51,7 +51,6 @@
 #include "mongo/db/storage/wiredtiger/wiredtiger_util.h"
 #include "mongo/util/assert_util.h"
 #include "mongo/util/fail_point.h"
-#include "mongo/util/fail_point_service.h"
 #include "mongo/util/log.h"
 #include "mongo/util/mongoutils/str.h"
 #include "mongo/util/scopeguard.h"
@@ -65,6 +64,7 @@ namespace mongo {
     using std::string;
 
 namespace {
+
     static const int kMinimumRecordStoreVersion = 1;
     static const int kCurrentRecordStoreVersion = 1; // New record stores use this by default.
     static const int kMaximumRecordStoreVersion = 1;
@@ -80,17 +80,10 @@ namespace {
         return (appMetadata.getValue().getIntField("oplogKeyExtractionVersion") == 1);
     }
 
-    // Throws a WriteConflictException to ensure the calling code can handle it
-    MONGO_FP_DECLARE(wtWriteConflictException);
-
-    void checkWriteConflictExceptionFPEnabled() {
-        if (MONGO_FAIL_POINT(wtWriteConflictException)) {
-            throw WriteConflictException();
-        }
-    }
-
 } // namespace
 
+    MONGO_FP_DECLARE(WTWriteConflictException);
+
     const std::string kWiredTigerEngineName = "wiredTiger";
 
     const long long WiredTigerRecordStore::kCollectionScanOnCreationThreshold = 10000;
@@ -316,14 +309,12 @@ namespace {
     }
 
     RecordData WiredTigerRecordStore::dataFor(OperationContext* txn, const RecordId& loc) const {
-        checkWriteConflictExceptionFPEnabled();
-
         // ownership passes to the shared_array created below
         WiredTigerCursor curwrap( _uri, _instanceId, true, txn);
         WT_CURSOR *c = curwrap.get();
         invariant( c );
         c->set_key(c, _makeKey(loc));
-        int ret = c->search(c);
+        int ret = WT_OP_CHECK(c->search(c));
         massert(28556, "Didn't find RecordId in WiredTigerRecordStore", ret != WT_NOTFOUND);
         invariantWTOK(ret);
         return _getData(curwrap);
@@ -331,13 +322,11 @@ namespace {
 
     bool WiredTigerRecordStore::findRecord( OperationContext* txn,
                                             const RecordId& loc, RecordData* out ) const {
-        checkWriteConflictExceptionFPEnabled();
-
         WiredTigerCursor curwrap( _uri, _instanceId, true, txn);
         WT_CURSOR *c = curwrap.get();
         invariant( c );
         c->set_key(c, _makeKey(loc));
-        int ret = c->search(c);
+        int ret = WT_OP_CHECK(c->search(c));
         if (ret == WT_NOTFOUND) {
             return false;
         }
@@ -347,13 +336,11 @@ namespace {
     }
 
     void WiredTigerRecordStore::deleteRecord( OperationContext* txn, const RecordId& loc ) {
-        checkWriteConflictExceptionFPEnabled();
-
         WiredTigerCursor cursor( _uri, _instanceId, true, txn );
         cursor.assertInActiveTxn();
         WT_CURSOR *c = cursor.get();
         c->set_key(c, _makeKey(loc));
-        int ret = c->search(c);
+        int ret = WT_OP_CHECK(c->search(c));
         invariantWTOK(ret);
 
         WT_ITEM old_value;
@@ -362,7 +349,7 @@ namespace {
 
         int old_length = old_value.size;
 
-        ret = c->remove(c);
+        ret = WT_OP_CHECK(c->remove(c));
         invariantWTOK(ret);
 
         _changeNumRecords(txn, -1);
@@ -438,9 +425,10 @@ namespace {
             WT_CURSOR *c = curwrap.get();
             RecordId newestOld;
             int ret = 0;
-            while (( sizeSaved < sizeOverCap || docsRemoved < docsOverCap ) &&
-                   docsRemoved < 1000 &&
-                   (ret = c->next(c)) == 0 ) {
+            while ((sizeSaved < sizeOverCap || docsRemoved < docsOverCap) &&
+                   (docsRemoved < 1000) &&
+                   (ret = WT_OP_CHECK(c->next(c))) == 0) {
+
                 int64_t key;
                 ret = c->get_key(c, &key);
                 invariantWTOK(ret);
@@ -465,18 +453,21 @@ namespace {
                 }
             }
 
-            if (ret != WT_NOTFOUND) invariantWTOK(ret);
+            if (ret != WT_NOTFOUND) {
+                invariantWTOK(ret);
+            }
 
             if (docsRemoved > 0) {
                 // if we scanned to the end of the collection or past our insert, go back one
-                if ( ret == WT_NOTFOUND || newestOld >= justInserted ) {
-                    ret = c->prev(c);
+                if (ret == WT_NOTFOUND || newestOld >= justInserted) {
+                    ret = WT_OP_CHECK(c->prev(c));
                 }
                 invariantWTOK(ret);
 
                 WiredTigerCursor startWrap( _uri, _instanceId, true, txn);
                 WT_CURSOR* start = startWrap.get();
-                start->next(start);
+                ret = WT_OP_CHECK(start->next(start));
+                invariantWTOK(ret);
 
                 invariantWTOK(session->truncate(session, NULL, start, c, NULL));
                 _changeNumRecords(txn, -docsRemoved);
@@ -509,8 +500,6 @@ namespace {
                                                               const char* data,
                                                               int len,
                                                               bool enforceQuota ) {
-        checkWriteConflictExceptionFPEnabled();
-
         if ( _isCapped && len > _cappedMaxSize ) {
             return StatusWith<RecordId>( ErrorCodes::BadValue,
                                          "object to insert exceeds cappedMaxSize" );
@@ -546,7 +535,7 @@ namespace {
         c->set_key(c, _makeKey(loc));
         WiredTigerItem value(data, len);
         c->set_value(c, value.Get());
-        int ret = c->insert(c);
+        int ret = WT_OP_CHECK(c->insert(c));
         if (ret) {
             return StatusWith<RecordId>(wtRCToStatus(ret, "WiredTigerRecordStore::insertRecord"));
         }
@@ -598,7 +587,7 @@ namespace {
         WT_CURSOR *c = curwrap.get();
         invariant( c );
         c->set_key(c, _makeKey(loc));
-        int ret = c->search(c);
+        int ret = WT_OP_CHECK(c->search(c));
         invariantWTOK(ret);
 
         WT_ITEM old_value;
@@ -610,7 +599,7 @@ namespace {
         c->set_key(c, _makeKey(loc));
         WiredTigerItem value(data, len);
         c->set_value(c, value.Get());
-        ret = c->insert(c);
+        ret = WT_OP_CHECK(c->insert(c));
         invariantWTOK(ret);
 
         _increaseDataSize(txn, len - old_length);
@@ -873,11 +862,8 @@ namespace {
             OperationContext* txn,
             const RecordId& startingPosition) const {
 
-        if (!_useOplogHack) {
+        if (!_useOplogHack)
             return boost::none;
-        }
-
-        checkWriteConflictExceptionFPEnabled();
 
         {
             WiredTigerRecoveryUnit* wru = WiredTigerRecoveryUnit::get(txn);
@@ -889,7 +875,7 @@ namespace {
 
         int cmp;
         c->set_key(c, _makeKey(startingPosition));
-        int ret = c->search_near(c, &cmp);
+        int ret = WT_OP_CHECK(c->search_near(c, &cmp));
         if (ret == 0 && cmp > 0) ret = c->prev(c); // landed one higher than startingPosition
         if (ret == WT_NOTFOUND) return RecordId(); // nothing <= startingPosition
         invariantWTOK(ret);
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_record_store.h b/src/mongo/db/storage/wiredtiger/wiredtiger_record_store.h
index ed5d3dffcb7..b7fd421365f 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_record_store.h
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_record_store.h
@@ -41,6 +41,13 @@
 #include "mongo/db/storage/record_store.h"
 #include "mongo/db/storage/capped_callback.h"
 #include "mongo/platform/atomic_word.h"
+#include "mongo/util/fail_point_service.h"
+
+/**
+ * Either executes the specified operation and returns it's value or randomly throws a write
+ * conflict exception if the WTWriteConflictException failpoint is enabled.
+ */
+#define WT_OP_CHECK(x) (((MONGO_FAIL_POINT(WTWriteConflictException))) ? (WT_ROLLBACK) : (x))
 
 namespace mongo {
 
@@ -282,4 +289,8 @@ namespace mongo {
         WiredTigerSizeStorer* _sizeStorer; // not owned, can be NULL
         int _sizeStorerCounter;
     };
+
+    // WT failpoint to throw write conflict exceptions randomly
+    MONGO_FP_FORWARD_DECLARE(WTWriteConflictException);
+
 }
diff --git a/src/mongo/util/fail_point_service.h b/src/mongo/util/fail_point_service.h
index abc2aacb468..754bb759fb0 100644
--- a/src/mongo/util/fail_point_service.h
+++ b/src/mongo/util/fail_point_service.h
@@ -26,28 +26,33 @@
  *    then also delete it in the license file.
  */
 
-/**
- * Should NOT be included by other header files.  Include only in source files.
- */
-
 #pragma once
 
 #include "mongo/base/init.h"
 #include "mongo/util/fail_point_registry.h"
 
 namespace mongo {
+
     /**
      * @return the global fail point registry.
      */
     FailPointRegistry* getGlobalFailPointRegistry();
 
     /**
-     * Convenience macro for declaring a fail point. Must be used in global scope and
-     * never in a block with limited scope (ie, inside functions, loops, etc.)
+     * Convenience macro for declaring a fail point. Must be used in global scope and never in a
+     * block with limited scope (ie, inside functions, loops, etc.).
+     *
+     * NOTE: Never use in header files, only sources.
      */
     #define MONGO_FP_DECLARE(fp) FailPoint fp; \
         MONGO_INITIALIZER_GENERAL(fp, ("FailPointRegistry"), ("AllFailPointsRegistered")) \
                 (::mongo::InitializerContext* context) { \
             return getGlobalFailPointRegistry()->addFailPoint(#fp, &fp); \
         }
+
+    /**
+     * Convenience macro for defining a fail point in a header scope.
+     */
+    #define MONGO_FP_FORWARD_DECLARE(fp) extern FailPoint fp;
+
 }