SERVER-37657 Report the offending oplog entries if a batch contains non-increasing transaction numbers

(cherry picked from commit 16e139c29350840b1b026164c71e998b71ada2be)

4 files changed, 32 insertions, 23 deletions

diff --git a/src/mongo/db/repl/session_update_tracker.cpp b/src/mongo/db/repl/session_update_tracker.cpp
index c0baf92db3a..9988a3832b9 100644
--- a/src/mongo/db/repl/session_update_tracker.cpp
+++ b/src/mongo/db/repl/session_update_tracker.cpp
@@ -28,6 +28,8 @@
  *    it in the license file.
  */
 
+#define MONGO_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kReplication
+
 #include "mongo/platform/basic.h"
 
 #include "mongo/db/repl/session_update_tracker.h"
@@ -35,16 +37,18 @@
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/session.h"
 #include "mongo/util/assert_util.h"
+#include "mongo/util/log.h"
 
 namespace mongo {
 namespace repl {
 
 boost::optional<std::vector<OplogEntry>> SessionUpdateTracker::updateOrFlush(
     const OplogEntry& entry) {
-    auto ns = entry.getNamespace();
+    const auto& ns = entry.getNamespace();
+
     if (ns == NamespaceString::kSessionTransactionsTableNamespace ||
         (ns.isConfigDB() && ns.isCommand())) {
-        return flush(entry);
+        return _flush(entry);
     }
 
     _updateSessionInfo(entry);
@@ -52,14 +56,14 @@ boost::optional<std::vector<OplogEntry>> SessionUpdateTracker::updateOrFlush(
 }
 
 void SessionUpdateTracker::_updateSessionInfo(const OplogEntry& entry) {
-    auto sessionInfo = entry.getOperationSessionInfo();
+    const auto& sessionInfo = entry.getOperationSessionInfo();
 
     if (!sessionInfo.getTxnNumber()) {
         return;
     }
 
-    auto lsid = sessionInfo.getSessionId();
-    fassert(50842, lsid.is_initialized());
+    const auto& lsid = sessionInfo.getSessionId();
+    invariant(lsid);
 
     auto iter = _sessionsToUpdate.find(lsid->getId());
     if (iter == _sessionsToUpdate.end()) {
@@ -67,12 +71,21 @@ void SessionUpdateTracker::_updateSessionInfo(const OplogEntry& entry) {
         return;
     }
 
-    auto existingSessionInfo = iter->second.getOperationSessionInfo();
-    fassert(50843, *sessionInfo.getTxnNumber() >= *existingSessionInfo.getTxnNumber());
-    iter->second = entry;
+    const auto& existingSessionInfo = iter->second.getOperationSessionInfo();
+    if (*sessionInfo.getTxnNumber() >= *existingSessionInfo.getTxnNumber()) {
+        iter->second = entry;
+        return;
+    }
+
+    severe() << "Entry for session " << lsid->getId() << " has txnNumber "
+             << *sessionInfo.getTxnNumber() << " < " << *existingSessionInfo.getTxnNumber();
+    severe() << "New oplog entry: " << redact(entry.toString());
+    severe() << "Existing oplog entry: " << redact(iter->second.toString());
+
+    fassertFailedNoTrace(50843);
 }
 
-std::vector<OplogEntry> SessionUpdateTracker::flush(const OplogEntry& entry) {
+std::vector<OplogEntry> SessionUpdateTracker::_flush(const OplogEntry& entry) {
     switch (entry.getOpType()) {
         case OpTypeEnum::kInsert:
         case OpTypeEnum::kNoop:
diff --git a/src/mongo/db/repl/session_update_tracker.h b/src/mongo/db/repl/session_update_tracker.h
index de89d22bc1c..e4288324337 100644
--- a/src/mongo/db/repl/session_update_tracker.h
+++ b/src/mongo/db/repl/session_update_tracker.h
@@ -57,14 +57,6 @@ public:
     boost::optional<std::vector<OplogEntry>> updateOrFlush(const OplogEntry& entry);
 
     /**
-     * Analyzes the given oplog entry and determines which transactions stored so far needs to be
-     * converted to oplog writes.
-     *
-     * Note: should only be called when oplog entry's ns target config.transactions or config.$cmd.
-     */
-    std::vector<OplogEntry> flush(const OplogEntry& entry);
-
-    /**
      * Converts all stored transaction infos to oplog writes to config.transactions.
      * Can return an empty vector if there is nothing to flush.
      */
@@ -72,6 +64,14 @@ public:
 
 private:
     /**
+     * Analyzes the given oplog entry and determines which transactions stored so far needs to be
+     * converted to oplog writes.
+     *
+     * Note: should only be called when oplog entry's ns target config.transactions or config.$cmd.
+     */
+    std::vector<OplogEntry> _flush(const OplogEntry& entry);
+
+    /**
      * Converts stored transaction infos that has a matching transcation id with the given
      * query predicate. Can return an empty vector if there is nothing to flush.
      */
diff --git a/src/mongo/db/repl/sync_tail.cpp b/src/mongo/db/repl/sync_tail.cpp
index e121cd6b1b4..0356a80e1d9 100644
--- a/src/mongo/db/repl/sync_tail.cpp
+++ b/src/mongo/db/repl/sync_tail.cpp
@@ -686,9 +686,6 @@ void fillWriterVectors(OperationContext* opCtx,
     }
 }
 
-}  // namespace
-
-namespace {
 void tryToGoLiveAsASecondary(OperationContext* opCtx,
                              ReplicationCoordinator* replCoord,
                              OpTime minValid) {
@@ -735,7 +732,8 @@ void tryToGoLiveAsASecondary(OperationContext* opCtx,
                   << ". Current state: " << replCoord->getMemberState() << causedBy(status);
     }
 }
-}
+
+}  // namespace
 
 class SyncTail::OpQueueBatcher {
     MONGO_DISALLOW_COPYING(OpQueueBatcher);
diff --git a/src/mongo/db/s/config/sharding_catalog_manager_collection_operations.cpp b/src/mongo/db/s/config/sharding_catalog_manager_collection_operations.cpp
index 4ee4310d769..a8d0bc99b2e 100644
--- a/src/mongo/db/s/config/sharding_catalog_manager_collection_operations.cpp
+++ b/src/mongo/db/s/config/sharding_catalog_manager_collection_operations.cpp
@@ -81,8 +81,6 @@ using std::set;
 
 namespace {
 
-const Seconds kDefaultFindHostMaxWaitTime(20);
-
 const ReadPreferenceSetting kConfigReadSelector(ReadPreference::Nearest, TagSet{});
 const WriteConcernOptions kNoWaitWriteConcern(1, WriteConcernOptions::SyncMode::UNSET, Seconds(0));
 const char kWriteConcernField[] = "writeConcern";