SERVER-76807 Avoid adding opTimes for non-retryable internal transactions to the session migration new opTime buffer

25 files changed, 53 insertions, 9 deletions

diff --git a/src/mongo/db/auth/auth_op_observer.h b/src/mongo/db/auth/auth_op_observer.h
index c3ef4a6ecb1..60ee2a133e8 100644
--- a/src/mongo/db/auth/auth_op_observer.h
+++ b/src/mongo/db/auth/auth_op_observer.h
@@ -231,6 +231,7 @@ public:
         Date_t wallClockTime) final {}
 
     void onTransactionPrepareNonPrimary(OperationContext* opCtx,
+                                        const LogicalSessionId& lsid,
                                         const std::vector<repl::OplogEntry>& statements,
                                         const repl::OpTime& prepareOpTime) final {}
 
diff --git a/src/mongo/db/free_mon/free_mon_op_observer.h b/src/mongo/db/free_mon/free_mon_op_observer.h
index f734cb77696..8658d769840 100644
--- a/src/mongo/db/free_mon/free_mon_op_observer.h
+++ b/src/mongo/db/free_mon/free_mon_op_observer.h
@@ -231,6 +231,7 @@ public:
         Date_t wallClockTime) final {}
 
     void onTransactionPrepareNonPrimary(OperationContext* opCtx,
+                                        const LogicalSessionId& lsid,
                                         const std::vector<repl::OplogEntry>& statements,
                                         const repl::OpTime& prepareOpTime) final {}
 
diff --git a/src/mongo/db/op_observer/fcv_op_observer.h b/src/mongo/db/op_observer/fcv_op_observer.h
index 152b5b8466f..42288989cde 100644
--- a/src/mongo/db/op_observer/fcv_op_observer.h
+++ b/src/mongo/db/op_observer/fcv_op_observer.h
@@ -226,6 +226,7 @@ public:
         Date_t wallClockTime) final{};
 
     void onTransactionPrepareNonPrimary(OperationContext* opCtx,
+                                        const LogicalSessionId& lsid,
                                         const std::vector<repl::OplogEntry>& statements,
                                         const repl::OpTime& prepareOpTime) final {}
 
diff --git a/src/mongo/db/op_observer/op_observer.h b/src/mongo/db/op_observer/op_observer.h
index 4aa505712ec..8b28135a90f 100644
--- a/src/mongo/db/op_observer/op_observer.h
+++ b/src/mongo/db/op_observer/op_observer.h
@@ -524,11 +524,20 @@ public:
         Date_t wallClockTime) = 0;
 
     /**
-     * This is called when a transaction transitions into prepare while it is not primary. Example
-     * case can include secondary oplog application or when node was restared and tries to
-     * recover prepared transactions from the oplog.
+     * This method is called when a transaction transitions into prepare while it is not primary,
+     * e.g. during secondary oplog application or recoverying prepared transactions from the
+     * oplog after restart. The method explicitly requires a session id (i.e. does not use the
+     * session id attached to the opCtx) because transaction oplog application currently applies the
+     * oplog entries for each prepared transaction in multiple internal sessions acquired from the
+     * InternalSessionPool. Currently, those internal sessions are completely unrelated to the
+     * session for the transaction itself. For a non-retryable internal transaction, not using the
+     * transaction session id in the codepath here can cause the opTime for the transaction to
+     * show up in the chunk migration opTime buffer although the writes they correspond to are not
+     * retryable and therefore are discarded anyway.
+     *
      */
     virtual void onTransactionPrepareNonPrimary(OperationContext* opCtx,
+                                                const LogicalSessionId& lsid,
                                                 const std::vector<repl::OplogEntry>& statements,
                                                 const repl::OpTime& prepareOpTime) = 0;
 
diff --git a/src/mongo/db/op_observer/op_observer_impl.cpp b/src/mongo/db/op_observer/op_observer_impl.cpp
index e2d257b29e2..c6fd11fa947 100644
--- a/src/mongo/db/op_observer/op_observer_impl.cpp
+++ b/src/mongo/db/op_observer/op_observer_impl.cpp
@@ -2179,9 +2179,10 @@ void OpObserverImpl::onTransactionPrepare(
 }
 
 void OpObserverImpl::onTransactionPrepareNonPrimary(OperationContext* opCtx,
+                                                    const LogicalSessionId& lsid,
                                                     const std::vector<repl::OplogEntry>& statements,
                                                     const repl::OpTime& prepareOpTime) {
-    shardObserveNonPrimaryTransactionPrepare(opCtx, statements, prepareOpTime);
+    shardObserveNonPrimaryTransactionPrepare(opCtx, lsid, statements, prepareOpTime);
 }
 
 void OpObserverImpl::onTransactionAbort(OperationContext* opCtx,
diff --git a/src/mongo/db/op_observer/op_observer_impl.h b/src/mongo/db/op_observer/op_observer_impl.h
index cb31794a88c..57b2ef98b58 100644
--- a/src/mongo/db/op_observer/op_observer_impl.h
+++ b/src/mongo/db/op_observer/op_observer_impl.h
@@ -240,6 +240,7 @@ public:
         Date_t wallClockTime) final;
 
     void onTransactionPrepareNonPrimary(OperationContext* opCtx,
+                                        const LogicalSessionId& lsid,
                                         const std::vector<repl::OplogEntry>& statements,
                                         const repl::OpTime& prepareOpTime) final;
 
@@ -281,6 +282,7 @@ private:
         const repl::OpTime& prepareOrCommitOptime) {}
     virtual void shardObserveNonPrimaryTransactionPrepare(
         OperationContext* opCtx,
+        const LogicalSessionId& lsid,
         const std::vector<repl::OplogEntry>& stmts,
         const repl::OpTime& prepareOrCommitOptime) {}
     void _onReplicationRollback(OperationContext* opCtx, const RollbackObserverInfo& rbInfo) final;
diff --git a/src/mongo/db/op_observer/op_observer_noop.h b/src/mongo/db/op_observer/op_observer_noop.h
index 3ba66e80f79..b7a7c9707b4 100644
--- a/src/mongo/db/op_observer/op_observer_noop.h
+++ b/src/mongo/db/op_observer/op_observer_noop.h
@@ -207,6 +207,7 @@ public:
         size_t numberOfPrePostImagesToWrite,
         Date_t wallClockTime) override{};
     void onTransactionPrepareNonPrimary(OperationContext* opCtx,
+                                        const LogicalSessionId& lsid,
                                         const std::vector<repl::OplogEntry>& statements,
                                         const repl::OpTime& prepareOpTime) override {}
     void onTransactionAbort(OperationContext* opCtx,
diff --git a/src/mongo/db/op_observer/op_observer_registry.h b/src/mongo/db/op_observer/op_observer_registry.h
index afdedf8b582..49614a75f62 100644
--- a/src/mongo/db/op_observer/op_observer_registry.h
+++ b/src/mongo/db/op_observer/op_observer_registry.h
@@ -476,11 +476,12 @@ public:
     }
 
     void onTransactionPrepareNonPrimary(OperationContext* opCtx,
+                                        const LogicalSessionId& lsid,
                                         const std::vector<repl::OplogEntry>& statements,
                                         const repl::OpTime& prepareOpTime) override {
         ReservedTimes times{opCtx};
         for (auto& observer : _observers) {
-            observer->onTransactionPrepareNonPrimary(opCtx, statements, prepareOpTime);
+            observer->onTransactionPrepareNonPrimary(opCtx, lsid, statements, prepareOpTime);
         }
     }
 
diff --git a/src/mongo/db/op_observer/user_write_block_mode_op_observer.h b/src/mongo/db/op_observer/user_write_block_mode_op_observer.h
index e4cc611da70..75cff472d5a 100644
--- a/src/mongo/db/op_observer/user_write_block_mode_op_observer.h
+++ b/src/mongo/db/op_observer/user_write_block_mode_op_observer.h
@@ -255,6 +255,7 @@ public:
         Date_t wallClockTime) final {}
 
     void onTransactionPrepareNonPrimary(OperationContext* opCtx,
+                                        const LogicalSessionId& lsid,
                                         const std::vector<repl::OplogEntry>& statements,
                                         const repl::OpTime& prepareOpTime) final {}
 
diff --git a/src/mongo/db/repl/primary_only_service_op_observer.h b/src/mongo/db/repl/primary_only_service_op_observer.h
index ef4490c0099..b82a0ee58a9 100644
--- a/src/mongo/db/repl/primary_only_service_op_observer.h
+++ b/src/mongo/db/repl/primary_only_service_op_observer.h
@@ -233,6 +233,7 @@ public:
         Date_t wallClockTime) final {}
 
     void onTransactionPrepareNonPrimary(OperationContext* opCtx,
+                                        const LogicalSessionId& lsid,
                                         const std::vector<repl::OplogEntry>& statements,
                                         const repl::OpTime& prepareOpTime) final {}
 
diff --git a/src/mongo/db/repl/shard_merge_recipient_op_observer.h b/src/mongo/db/repl/shard_merge_recipient_op_observer.h
index 57b4cd668ed..53ee1c26120 100644
--- a/src/mongo/db/repl/shard_merge_recipient_op_observer.h
+++ b/src/mongo/db/repl/shard_merge_recipient_op_observer.h
@@ -230,6 +230,7 @@ public:
         Date_t wallClockTime) final {}
 
     void onTransactionPrepareNonPrimary(OperationContext* opCtx,
+                                        const LogicalSessionId& lsid,
                                         const std::vector<repl::OplogEntry>& statements,
                                         const repl::OpTime& prepareOpTime) final {}
 
diff --git a/src/mongo/db/repl/tenant_migration_donor_op_observer.h b/src/mongo/db/repl/tenant_migration_donor_op_observer.h
index f7ded62694c..b914baf2f7b 100644
--- a/src/mongo/db/repl/tenant_migration_donor_op_observer.h
+++ b/src/mongo/db/repl/tenant_migration_donor_op_observer.h
@@ -214,6 +214,7 @@ public:
         const std::vector<repl::ReplOperation>& statements) noexcept final {}
 
     void onTransactionPrepareNonPrimary(OperationContext* opCtx,
+                                        const LogicalSessionId& lsid,
                                         const std::vector<repl::OplogEntry>& statements,
                                         const repl::OpTime& prepareOpTime) final {}
 
diff --git a/src/mongo/db/repl/tenant_migration_recipient_op_observer.h b/src/mongo/db/repl/tenant_migration_recipient_op_observer.h
index 3ff69365c7e..d5e6377ae22 100644
--- a/src/mongo/db/repl/tenant_migration_recipient_op_observer.h
+++ b/src/mongo/db/repl/tenant_migration_recipient_op_observer.h
@@ -232,6 +232,7 @@ public:
         Date_t wallClockTime) final {}
 
     void onTransactionPrepareNonPrimary(OperationContext* opCtx,
+                                        const LogicalSessionId& lsid,
                                         const std::vector<repl::OplogEntry>& statements,
                                         const repl::OpTime& prepareOpTime) final {}
 
diff --git a/src/mongo/db/repl/transaction_oplog_application.cpp b/src/mongo/db/repl/transaction_oplog_application.cpp
index cf8bcbe63eb..e6f59b39f1b 100644
--- a/src/mongo/db/repl/transaction_oplog_application.cpp
+++ b/src/mongo/db/repl/transaction_oplog_application.cpp
@@ -639,7 +639,8 @@ Status _applyPrepareTransaction(OperationContext* opCtx,
 
         auto opObserver = opCtx->getServiceContext()->getOpObserver();
         invariant(opObserver);
-        opObserver->onTransactionPrepareNonPrimary(opCtx, txnOps, prepareOp.getOpTime());
+        opObserver->onTransactionPrepareNonPrimary(
+            opCtx, *prepareOp.getSessionId(), txnOps, prepareOp.getOpTime());
 
         // Prepare transaction success.
         abortOnError.dismiss();
diff --git a/src/mongo/db/s/config_server_op_observer.h b/src/mongo/db/s/config_server_op_observer.h
index f551d57bc55..bd1478c5005 100644
--- a/src/mongo/db/s/config_server_op_observer.h
+++ b/src/mongo/db/s/config_server_op_observer.h
@@ -233,6 +233,7 @@ public:
         Date_t wallClockTime) override {}
 
     void onTransactionPrepareNonPrimary(OperationContext* opCtx,
+                                        const LogicalSessionId& lsid,
                                         const std::vector<repl::OplogEntry>& statements,
                                         const repl::OpTime& prepareOpTime) override {}
 
diff --git a/src/mongo/db/s/op_observer_sharding_impl.cpp b/src/mongo/db/s/op_observer_sharding_impl.cpp
index 931497f2644..51cb98b6c9c 100644
--- a/src/mongo/db/s/op_observer_sharding_impl.cpp
+++ b/src/mongo/db/s/op_observer_sharding_impl.cpp
@@ -238,12 +238,13 @@ void OpObserverShardingImpl::shardObserveTransactionPrepareOrUnpreparedCommit(
 
 void OpObserverShardingImpl::shardObserveNonPrimaryTransactionPrepare(
     OperationContext* opCtx,
+    const LogicalSessionId& lsid,
     const std::vector<repl::OplogEntry>& stmts,
     const repl::OpTime& prepareOrCommitOptime) {
 
     opCtx->recoveryUnit()->registerChange(
         std::make_unique<LogTransactionOperationsForShardingHandler>(
-            *opCtx->getLogicalSessionId(), stmts, prepareOrCommitOptime));
+            lsid, stmts, prepareOrCommitOptime));
 }
 
 }  // namespace mongo
diff --git a/src/mongo/db/s/op_observer_sharding_impl.h b/src/mongo/db/s/op_observer_sharding_impl.h
index 2858f5b92d8..4974acc7bfe 100644
--- a/src/mongo/db/s/op_observer_sharding_impl.h
+++ b/src/mongo/db/s/op_observer_sharding_impl.h
@@ -77,6 +77,7 @@ protected:
         const repl::OpTime& prepareOrCommitOptime) override;
     void shardObserveNonPrimaryTransactionPrepare(
         OperationContext* opCtx,
+        const LogicalSessionId& lsid,
         const std::vector<repl::OplogEntry>& stmts,
         const repl::OpTime& prepareOrCommitOptime) override;
 };
diff --git a/src/mongo/db/s/query_analysis_op_observer.h b/src/mongo/db/s/query_analysis_op_observer.h
index 1a13e7e6134..5c9065a11eb 100644
--- a/src/mongo/db/s/query_analysis_op_observer.h
+++ b/src/mongo/db/s/query_analysis_op_observer.h
@@ -232,6 +232,7 @@ public:
         Date_t wallClockTime) final {}
 
     void onTransactionPrepareNonPrimary(OperationContext* opCtx,
+                                        const LogicalSessionId& lsid,
                                         const std::vector<repl::OplogEntry>& statements,
                                         const repl::OpTime& prepareOpTime) final {}
 
diff --git a/src/mongo/db/s/range_deleter_service_op_observer.h b/src/mongo/db/s/range_deleter_service_op_observer.h
index 4cfec3eeaae..efb612e8045 100644
--- a/src/mongo/db/s/range_deleter_service_op_observer.h
+++ b/src/mongo/db/s/range_deleter_service_op_observer.h
@@ -237,6 +237,7 @@ private:
         Date_t wallClockTime) override {}
 
     void onTransactionPrepareNonPrimary(OperationContext* opCtx,
+                                        const LogicalSessionId& lsid,
                                         const std::vector<repl::OplogEntry>& statements,
                                         const repl::OpTime& prepareOpTime) override {}
 
diff --git a/src/mongo/db/s/resharding/resharding_op_observer.h b/src/mongo/db/s/resharding/resharding_op_observer.h
index 12a6d5d83f3..46509864cb4 100644
--- a/src/mongo/db/s/resharding/resharding_op_observer.h
+++ b/src/mongo/db/s/resharding/resharding_op_observer.h
@@ -253,6 +253,7 @@ public:
         Date_t wallClockTime) override {}
 
     void onTransactionPrepareNonPrimary(OperationContext* opCtx,
+                                        const LogicalSessionId& lsid,
                                         const std::vector<repl::OplogEntry>& statements,
                                         const repl::OpTime& prepareOpTime) override {}
 
diff --git a/src/mongo/db/s/session_catalog_migration_source.cpp b/src/mongo/db/s/session_catalog_migration_source.cpp
index 162d7d91aae..4a75fc7192b 100644
--- a/src/mongo/db/s/session_catalog_migration_source.cpp
+++ b/src/mongo/db/s/session_catalog_migration_source.cpp
@@ -685,6 +685,13 @@ bool SessionCatalogMigrationSource::_fetchNextNewWriteOplog(OperationContext* op
                 const auto sessionId = *nextNewWriteOplog.getSessionId();
 
                 if (isInternalSessionForNonRetryableWrite(sessionId)) {
+                    dassert(0,
+                            str::stream() << "Cannot add op time for a non-retryable "
+                                             "internal transaction to the "
+                                             "session migration op time queue - "
+                                          << "session id:" << sessionId << " oplog entry: "
+                                          << redact(nextNewWriteOplog.toBSONForLogging()));
+
                     // Transactions inside internal sessions for non-retryable writes are not
                     // retryable so there is no need to transfer their write history to the
                     // recipient.
diff --git a/src/mongo/db/s/session_catalog_migration_source_test.cpp b/src/mongo/db/s/session_catalog_migration_source_test.cpp
index 3cca3b085b8..fc489def1bb 100644
--- a/src/mongo/db/s/session_catalog_migration_source_test.cpp
+++ b/src/mongo/db/s/session_catalog_migration_source_test.cpp
@@ -996,8 +996,9 @@ TEST_F(SessionCatalogMigrationSourceTest,
     ASSERT_EQ(migrationSource.getSessionOplogEntriesSkippedSoFarLowerBound(), 0);
 }
 
-TEST_F(SessionCatalogMigrationSourceTest,
-       DiscardOplogEntriesForNewCommittedInternalTransactionForNonRetryableWrite) {
+DEATH_TEST_F(SessionCatalogMigrationSourceTest,
+             DiscardOplogEntriesForNewCommittedInternalTransactionForNonRetryableWrite,
+             "invariant") {
     SessionCatalogMigrationSource migrationSource(opCtx(), kNs, kChunkRange, kShardKey);
     migrationSource.init(opCtx(), kMigrationLsid);
     ASSERT_FALSE(migrationSource.fetchNextOplog(opCtx()));
@@ -1021,6 +1022,10 @@ TEST_F(SessionCatalogMigrationSourceTest,
     ASSERT_TRUE(migrationSource.hasMoreOplog());
     ASSERT_FALSE(migrationSource.fetchNextOplog(opCtx()));
     ASSERT_EQ(migrationSource.getSessionOplogEntriesToBeMigratedSoFar(), 0);
+
+    // notifyNewWriteOpTime() uses dassert, so it will only invariant in debug mode. Deliberately
+    // crash here in non-debug mode to make the test work in both modes.
+    invariant(kDebugBuild);
 }
 
 TEST_F(SessionCatalogMigrationSourceTest,
diff --git a/src/mongo/db/s/shard_server_op_observer.h b/src/mongo/db/s/shard_server_op_observer.h
index 9b48d35c372..50417bee969 100644
--- a/src/mongo/db/s/shard_server_op_observer.h
+++ b/src/mongo/db/s/shard_server_op_observer.h
@@ -232,6 +232,7 @@ public:
         Date_t wallClockTime) override {}
 
     void onTransactionPrepareNonPrimary(OperationContext* opCtx,
+                                        const LogicalSessionId& lsid,
                                         const std::vector<repl::OplogEntry>& statements,
                                         const repl::OpTime& prepareOpTime) override {}
 
diff --git a/src/mongo/db/serverless/shard_split_donor_op_observer.h b/src/mongo/db/serverless/shard_split_donor_op_observer.h
index c8e65f7b294..f6cabb8ccde 100644
--- a/src/mongo/db/serverless/shard_split_donor_op_observer.h
+++ b/src/mongo/db/serverless/shard_split_donor_op_observer.h
@@ -229,6 +229,7 @@ public:
         Date_t wallClockTime) final {}
 
     void onTransactionPrepareNonPrimary(OperationContext* opCtx,
+                                        const LogicalSessionId& lsid,
                                         const std::vector<repl::OplogEntry>& statements,
                                         const repl::OpTime& prepareOpTime) final {}
 
diff --git a/src/mongo/idl/cluster_server_parameter_op_observer.h b/src/mongo/idl/cluster_server_parameter_op_observer.h
index a06f4dbde17..adff6757022 100644
--- a/src/mongo/idl/cluster_server_parameter_op_observer.h
+++ b/src/mongo/idl/cluster_server_parameter_op_observer.h
@@ -238,6 +238,7 @@ public:
         Date_t wallClockTime) final {}
 
     void onTransactionPrepareNonPrimary(OperationContext* opCtx,
+                                        const LogicalSessionId& lsid,
                                         const std::vector<repl::OplogEntry>& statements,
                                         const repl::OpTime& prepareOpTime) final {}