SERVER-53511: Make committed transactions aggregation pipeline

3 files changed, 182 insertions, 11 deletions

diff --git a/jstests/replsets/tenant_migration_commit_transaction_retry.js b/jstests/replsets/tenant_migration_commit_transaction_retry.js
index 113324fa5f3..16334c4221c 100644
--- a/jstests/replsets/tenant_migration_commit_transaction_retry.js
+++ b/jstests/replsets/tenant_migration_commit_transaction_retry.js
@@ -125,16 +125,6 @@ assert.eq(txnEntryOnDonor, aggRes.cursor.firstBatch[0]);
 // Test the client can retry commitTransaction for that transaction that committed prior to the
 // migration.
 
-// Insert the config.transactions entry on the recipient, but with a dummy lastWriteOpTime. The
-// recipient should not need a real lastWriteOpTime to support a commitTransaction retry.
-txnEntryOnDonor.lastWriteOpTime.ts = new Timestamp(0, 0);
-assert.commandWorked(
-    recipientPrimary.getCollection("config.transactions").insert([txnEntryOnDonor]));
-recipientRst.awaitLastOpCommitted();
-recipientRst.getSecondaries().forEach(node => {
-    assert.eq(1, node.getCollection("config.transactions").count(txnEntryOnDonor));
-});
-
 assert.commandWorked(recipientPrimary.adminCommand({
     commitTransaction: 1,
     lsid: txnEntryOnDonor._id,
diff --git a/jstests/replsets/tenant_migration_fetch_committed_transactions.js b/jstests/replsets/tenant_migration_fetch_committed_transactions.js
new file mode 100644
index 00000000000..821dd543106
--- /dev/null
+++ b/jstests/replsets/tenant_migration_fetch_committed_transactions.js
@@ -0,0 +1,142 @@
+/**
+ * Tests that the migration recipient will retrieve committed transactions on the donor with a
+ * 'lastWriteOpTime' before the stored 'startFetchingOpTime'. The recipient should store these
+ * committed transaction entries in its own 'config.transactions' collection.
+ *
+ * @tags: [requires_fcv_49, requires_majority_read_concern, incompatible_with_eft,
+ * incompatible_with_windows_tls]
+ */
+
+(function() {
+"use strict";
+
+load("jstests/core/txns/libs/prepare_helpers.js");
+load("jstests/replsets/libs/tenant_migration_test.js");
+load("jstests/replsets/libs/tenant_migration_util.js");
+load("jstests/replsets/rslib.js");
+load("jstests/libs/uuid_util.js");
+
+const tenantMigrationTest = new TenantMigrationTest({name: jsTestName()});
+if (!tenantMigrationTest.isFeatureFlagEnabled()) {
+    jsTestLog("Skipping test because the tenant migrations feature flag is disabled");
+    return;
+}
+
+const tenantId = "testTenantId";
+const tenantDB = tenantMigrationTest.tenantDB(tenantId, "testDB");
+const nonTenantDB = tenantMigrationTest.nonTenantDB(tenantId, "testDB");
+const collName = "testColl";
+const tenantNS = `${tenantDB}.${collName}`;
+const transactionsNS = "config.transactions";
+
+const donorPrimary = tenantMigrationTest.getDonorPrimary();
+const recipientPrimary = tenantMigrationTest.getRecipientPrimary();
+
+assert.commandWorked(donorPrimary.getCollection(tenantNS).insert([{_id: 0, x: 0}, {_id: 1, x: 1}],
+                                                                 {writeConcern: {w: "majority"}}));
+
+{
+    jsTestLog("Run and commit a transaction prior to the migration");
+    const session = donorPrimary.startSession({causalConsistency: false});
+    const sessionDb = session.getDatabase(tenantDB);
+    const sessionColl = sessionDb.getCollection(collName);
+
+    session.startTransaction({writeConcern: {w: "majority"}});
+    const findAndModifyRes0 = sessionColl.findAndModify({query: {x: 0}, remove: true});
+    assert.eq({_id: 0, x: 0}, findAndModifyRes0);
+    assert.commandWorked(session.commitTransaction_forTesting());
+    assert.sameMembers(sessionColl.find({}).toArray(), [{_id: 1, x: 1}]);
+    session.endSession();
+}
+
+// This should be the only transaction entry on the donor fetched by the recipient.
+const fetchedDonorTxnEntry = donorPrimary.getCollection(transactionsNS).find().toArray();
+
+{
+    jsTestLog("Run and abort a transaction prior to the migration");
+    const session = donorPrimary.startSession({causalConsistency: false});
+    const sessionDb = session.getDatabase(tenantDB);
+    const sessionColl = sessionDb.getCollection(collName);
+
+    session.startTransaction({writeConcern: {w: "majority"}});
+    const findAndModifyRes0 = sessionColl.findAndModify({query: {x: 1}, remove: true});
+    assert.eq({_id: 1, x: 1}, findAndModifyRes0);
+
+    // We prepare the transaction so that 'abortTransaction' will update the transactions table. We
+    // should later see that the recipient will not update its transactions table with this entry,
+    // since we only fetch committed transactions.
+    PrepareHelpers.prepareTransaction(session);
+
+    assert.commandWorked(session.abortTransaction_forTesting());
+    assert.sameMembers(sessionColl.find({}).toArray(), [{_id: 1, x: 1}]);
+    session.endSession();
+}
+
+{
+    jsTestLog("Run and commit a transaction that does not belong to the tenant");
+    const session = donorPrimary.startSession({causalConsistency: false});
+    const sessionDb = session.getDatabase(nonTenantDB);
+    const sessionColl = sessionDb.getCollection(collName);
+
+    session.startTransaction({writeConcern: {w: "majority"}});
+    assert.commandWorked(sessionColl.insert([{_id: 0, x: 0}, {_id: 1, x: 1}]));
+    assert.commandWorked(session.commitTransaction_forTesting());
+    session.endSession();
+}
+
+const fpAfterRetrievingStartOpTime = configureFailPoint(
+    recipientPrimary, "fpAfterRetrievingStartOpTimesMigrationRecipientInstance", {action: "hang"});
+
+jsTestLog("Starting a migration");
+const migrationId = UUID();
+const migrationOpts = {
+    migrationIdString: extractUUIDFromObject(migrationId),
+    tenantId,
+};
+assert.commandWorked(tenantMigrationTest.startMigration(migrationOpts));
+fpAfterRetrievingStartOpTime.wait();
+
+{
+    jsTestLog("Run and commit a transaction in the middle of the migration");
+    const session = donorPrimary.startSession({causalConsistency: false});
+    const sessionDb = session.getDatabase(tenantDB);
+    const sessionColl = sessionDb.getCollection(collName);
+
+    session.startTransaction({writeConcern: {w: "majority"}});
+    assert.commandWorked(sessionColl.insert([{_id: 2, x: 2}, {_id: 3, x: 3}]));
+    assert.commandWorked(session.commitTransaction_forTesting());
+    session.endSession();
+}
+
+jsTestLog("Waiting for migration to complete");
+fpAfterRetrievingStartOpTime.off();
+assert.commandWorked(tenantMigrationTest.waitForMigrationToComplete(migrationOpts));
+
+const donorTxnEntries = donorPrimary.getCollection(transactionsNS).find().toArray();
+jsTestLog(`All donor entries: ${tojson(donorTxnEntries)}`);
+assert.eq(4, donorTxnEntries.length, `donor transaction entries: ${tojson(donorTxnEntries)}`);
+
+const recipientTxnEntries = recipientPrimary.getCollection(transactionsNS).find().toArray();
+
+// Verify that the recipient has fetched and written only the first committed transaction entry from
+// the donor.
+assert.eq(
+    1, recipientTxnEntries.length, `recipient transaction entries: ${tojson(recipientTxnEntries)}`);
+assert.eq(
+    fetchedDonorTxnEntry,
+    recipientTxnEntries,
+    `fetched donor transaction entries: ${
+        tojson(
+            fetchedDonorTxnEntry)}; recipient transaction entries: ${tojson(recipientTxnEntries)}`);
+
+// Test that the client can retry 'commitTransaction' on the recipient.
+const recipientTxnEntry = recipientTxnEntries[0];
+assert.commandWorked(recipientPrimary.adminCommand({
+    commitTransaction: 1,
+    lsid: recipientTxnEntry._id,
+    txnNumber: recipientTxnEntry.txnNum,
+    autocommit: false,
+}));
+
+tenantMigrationTest.stop();
+})();
diff --git a/jstests/replsets/tenant_migration_oplog_view.js b/jstests/replsets/tenant_migration_oplog_view.js
index 18f9018f1e3..5803b97e836 100644
--- a/jstests/replsets/tenant_migration_oplog_view.js
+++ b/jstests/replsets/tenant_migration_oplog_view.js
@@ -45,12 +45,15 @@ if (!tenantMigrationTest.isFeatureFlagEnabled()) {
     return;
 }
 
+const dbName = "test";
+const collName = "collection";
+
 const donorPrimary = tenantMigrationTest.getDonorPrimary();
 const rsConn = new Mongo(donorRst.getURL());
 const oplog = donorPrimary.getDB("local")["oplog.rs"];
 const migrationOplogView = donorPrimary.getDB("local")["system.tenantMigration.oplogView"];
 const session = rsConn.startSession({retryWrites: true});
-const collection = session.getDatabase("test")["collection"];
+const collection = session.getDatabase(dbName)[collName];
 
 {
     // Assert an oplog entry representing a retryable write only projects fields defined in the
@@ -118,6 +121,42 @@ const collection = session.getDatabase("test")["collection"];
     assert.eq(viewEntry["postImageOpTime"]["ts"], resultOplogEntry["ts"]);
 }
 
+{
+    // Assert that an oplog entry that belongs to a transaction will project its 'o.applyOps.ns'
+    // field. This is used to filter transactions that belong to the tenant.
+    const txnSession = rsConn.startSession();
+    const txnDb = txnSession.getDatabase(dbName);
+    const txnColl = txnDb.getCollection(collName);
+    assert.commandWorked(txnColl.insert({_id: 'insertDoc'}));
+
+    txnSession.startTransaction({writeConcern: {w: "majority"}});
+    assert.commandWorked(txnColl.insert({_id: 'transaction0'}));
+    assert.commandWorked(txnColl.insert({_id: 'transaction1'}));
+    assert.commandWorked(txnSession.commitTransaction_forTesting());
+
+    const txnEntryOnDonor =
+        donorPrimary.getCollection("config.transactions").find({state: "committed"}).toArray()[0];
+    jsTestLog(`Txn entry on donor: ${tojson(txnEntryOnDonor)}`);
+
+    const viewEntry = migrationOplogView.find({ts: txnEntryOnDonor.lastWriteOpTime.ts}).next();
+    jsTestLog(`Transaction view entry: ${tojson(viewEntry)}`);
+
+    // The following fields are filtered out of the view.
+    assert(!viewEntry.hasOwnProperty("txnNumber"));
+    assert(!viewEntry.hasOwnProperty("state"));
+    assert(!viewEntry.hasOwnProperty("preImageOpTime"));
+    assert(!viewEntry.hasOwnProperty("postImageOpTime"));
+    assert(!viewEntry.hasOwnProperty("stmtId"));
+
+    // Verify that the view entry has the following fields.
+    assert(viewEntry.hasOwnProperty("ns"));
+    assert(viewEntry.hasOwnProperty("ts"));
+    assert(viewEntry.hasOwnProperty("applyOpsNs"));
+
+    // Assert that 'applyOpsNs' contains the namespace of the inserts.
+    assert.eq(viewEntry.applyOpsNs, `${dbName}.${collName}`);
+}
+
 donorRst.stopSet();
 tenantMigrationTest.stop();
 })();