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

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 0673d0dec2a..220f5584f3c 100644
--- a/src/mongo/db/repl/session_update_tracker.cpp
+++ b/src/mongo/db/repl/session_update_tracker.cpp
@@ -26,6 +26,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,6 +37,7 @@
 #include "mongo/db/session.h"
 #include "mongo/db/session_txn_record_gen.h"
 #include "mongo/util/assert_util.h"
+#include "mongo/util/log.h"
 
 namespace mongo {
 namespace repl {
@@ -105,10 +108,11 @@ boost::optional<repl::OplogEntry> createMatchingTransactionTableUpdate(
 
 boost::optional<std::vector<OplogEntry>> SessionUpdateTracker::updateOrFlush(
     const OplogEntry& entry) {
-    auto ns = entry.getNss();
+    const auto& ns = entry.getNss();
+
     if (ns == NamespaceString::kSessionTransactionsTableNamespace ||
         (ns.isConfigDB() && ns.isCommand())) {
-        return flush(entry);
+        return _flush(entry);
     }
 
     _updateSessionInfo(entry);
@@ -116,14 +120,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()) {
@@ -131,12 +135,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 eab230aebdd..4e00a4ff01e 100644
--- a/src/mongo/db/repl/session_update_tracker.h
+++ b/src/mongo/db/repl/session_update_tracker.h
@@ -55,14 +55,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.
      */
@@ -70,6 +62,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 f67c3e005ec..929eb424a7f 100644
--- a/src/mongo/db/repl/sync_tail.cpp
+++ b/src/mongo/db/repl/sync_tail.cpp
@@ -580,9 +580,6 @@ void fillWriterVectors(OperationContext* opCtx,
     }
 }
 
-}  // namespace
-
-namespace {
 void tryToGoLiveAsASecondary(OperationContext* opCtx,
                              ReplicationCoordinator* replCoord,
                              OpTime minValid) {
@@ -629,7 +626,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 3dedea39286..45002a4b6f6 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
@@ -79,8 +79,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";