SERVER-99185 Handle transactionally replicated vectored inserts when restoring config.transactions during rollback (#31750)

GitOrigin-RevId: aee7790

Author: moustafamaher

etc/backports_required_for_multiversion_tests.yml

@@ -374,6 +374,8 @@ last-continuous:
       ticket: SERVER-89883
     - test_file: jstests/sharding/metadata_refresh_on_shard_removal.js
       ticket: SERVER-99277
+    - test_file: jstests/replsets/rollback_with_coalesced_txn_table_updates_during_oplog_application_inserts.js
+      ticket: SERVER-99185
   suites: null
 last-lts:
   all:
@@ -795,4 +797,6 @@ last-lts:
       ticket: SERVER-89883
     - test_file: jstests/sharding/metadata_refresh_on_shard_removal.js
       ticket: SERVER-99277
+    - test_file: jstests/replsets/rollback_with_coalesced_txn_table_updates_during_oplog_application_inserts.js
+      ticket: SERVER-99185
   suites: null

jstests/replsets/rollback_with_coalesced_txn_table_updates_during_oplog_application.js renamed to jstests/replsets/libs/rollback_with_coalesced_txn_table_updates_during_oplog_application_helper.js

@@ -5,17 +5,19 @@
  * application.
  * We also test that if a node crashes after oplog truncation during rollback, the update made to
  * the 'config.transactions' table is persisted on startup.
- *
- * @tags: [requires_persistence]
  */
 
 import {configureFailPoint} from "jstests/libs/fail_point_util.js";
 import {ReplSetTest} from "jstests/libs/replsettest.js";
 
-const oplogApplierBatchSize = 100;
-
-function runTest(crashAfterRollbackTruncation) {
+let runTest = function(crashAfterRollbackTruncation,
+                       initFunc,
+                       stopReplProducerOnDocumentFunc,
+                       opsFunc,
+                       stmtMajorityCommittedFunc,
+                       validateFunc) {
     jsTestLog(`Running test with crashAfterRollbackTruncation = ${crashAfterRollbackTruncation}`);
+    const oplogApplierBatchSize = 100;
     const rst = new ReplSetTest({
         nodes: {
             n0: {},
@@ -41,8 +43,9 @@ function runTest(crashAfterRollbackTruncation) {
     const lsid = ({id: UUID()});
     const primary = rst.getPrimary();
     const ns = "test.retryable_write_partial_rollback";
-    assert.commandWorked(
-        primary.getCollection(ns).insert({_id: 0, counter: 0}, {writeConcern: {w: 5}}));
+    const counterTotal = oplogApplierBatchSize;
+    initFunc(primary, ns, counterTotal);
+
     // SERVER-65971: Do a write with `lsid` to add an entry to config.transactions. This write will
     // persist after rollback and be updated when the rollback code corrects for omitted writes to
     // the document.
@@ -63,25 +66,25 @@ function runTest(crashAfterRollbackTruncation) {
     const stopReplProducerFailpoints = [secondary1, secondary2, secondary3, secondary4].map(
         conn => configureFailPoint(conn, 'stopReplProducer'));
 
+    // Using an odd number ensures that when the 'ReplicateVectoredInsertsTransactionally' feature
+    // flag is enabled, counterMajorityCommitted + 1 will be in a different oplog entry. This is
+    // because we're using an internal batching size of 2, resulting in pairs like [0, 1], [2, 3],
+    // etc.
+    const counterMajorityCommitted = counterTotal - 5;
+
     // While replication is still entirely disabled, additionally disable replication partway into
     // the retryable write on all but the first secondary. The idea is that while secondary1 will
     // apply all of the oplog entries in a single batch, the other secondaries will only apply up to
     // counterMajorityCommitted oplog entries.
-    const counterTotal = oplogApplierBatchSize;
-    const counterMajorityCommitted = counterTotal - 2;
     const stopReplProducerOnDocumentFailpoints = [secondary2, secondary3, secondary4].map(
         conn => configureFailPoint(conn,
                                    'stopReplProducerOnDocument',
-                                   {document: {"diff.u.counter": counterMajorityCommitted + 1}}));
-
-    assert.commandWorked(primary.getCollection(ns).runCommand("update", {
-        updates: Array.from({length: counterTotal}, () => ({q: {_id: 0}, u: {$inc: {counter: 1}}})),
-        lsid,
-        txnNumber: NumberLong(2),
-    }));
+                                   stopReplProducerOnDocumentFunc(counterMajorityCommitted)));
 
-    const stmtMajorityCommitted = primary.getCollection("local.oplog.rs")
-                                      .findOne({ns, "o.diff.u.counter": counterMajorityCommitted});
+    opsFunc(primary, ns, counterTotal, lsid);
+    const stmtMajorityCommitted =
+        primary.getCollection("local.oplog.rs")
+            .findOne(stmtMajorityCommittedFunc(primary, ns, counterMajorityCommitted));
     assert.neq(null, stmtMajorityCommitted);
 
     for (const fp of stopReplProducerFailpoints) {
@@ -170,25 +173,30 @@ function runTest(crashAfterRollbackTruncation) {
     assert.commandWorked(secondary1.adminCommand({replSetFreeze: 0}));
     rst.stepUp(secondary1);
 
-    const docBeforeRetry = secondary1.getCollection(ns).findOne({_id: 0});
-    assert.eq(docBeforeRetry, {_id: 0, counter: counterMajorityCommitted});
-
-    assert.commandWorked(secondary1.getCollection(ns).runCommand("update", {
-        updates: Array.from({length: counterTotal}, () => ({q: {_id: 0}, u: {$inc: {counter: 1}}})),
-        lsid,
-        txnNumber: NumberLong(2),
-        writeConcern: {w: 5},
-    }));
-
-    const docAfterRetry = secondary1.getCollection(ns).findOne({_id: 0});
-    assert.eq(docAfterRetry, {_id: 0, counter: counterTotal});
+    validateFunc(secondary1, ns, counterMajorityCommitted, counterTotal, lsid);
 
     rst.stopSet();
-}
-
-// Test the general scenario where we perform the appropriate update to the 'config.transactions'
-// table during rollback.
-runTest(false);
-// Extends the test to crash the secondary in the middle of rollback right after oplog truncation.
-// We assert that the update made to the 'config.transactions' table persisted on startup.
-runTest(true);
+};
+
+export var runTests = function(
+    initFunc, stopReplProducerOnDocumentFunc, opsFunc, stmtMajorityCommittedFunc, validateFunc) {
+    // Test the general scenario where we perform the appropriate update to the
+    // 'config.transactions'
+    // table during rollback.
+    runTest(false,
+            initFunc,
+            stopReplProducerOnDocumentFunc,
+            opsFunc,
+            stmtMajorityCommittedFunc,
+            validateFunc);
+
+    // Extends the test to crash the secondary in the middle of rollback right after oplog
+    // truncation. We assert that the update made to the 'config.transactions' table persisted on
+    // startup.
+    runTest(true,
+            initFunc,
+            stopReplProducerOnDocumentFunc,
+            opsFunc,
+            stmtMajorityCommittedFunc,
+            validateFunc);
+};

jstests/replsets/rollback_with_coalesced_txn_table_updates_during_oplog_application_deletes.js

@@ -0,0 +1,76 @@
+/**
+ * Tests that the rollback procedure will update the 'config.transactions' table to be consistent
+ * with the node data at the 'stableTimestamp', specifically in the case where multiple delete ops
+ * to the 'config.transactions' table were coalesced into a single operation during secondary oplog
+ * application.
+ *
+ * @tags: [requires_persistence]
+ */
+
+import {
+    runTests
+} from
+    "jstests/replsets/libs/rollback_with_coalesced_txn_table_updates_during_oplog_application_helper.js";
+
+const initFunc = (primary, ns, counterTotal) => {
+    assert.commandWorked(primary.getCollection(ns).runCommand("insert", {
+        documents: Array.from({length: counterTotal}, (_, i) => ({_id: i})),
+        writeConcern: {w: 5}
+    }));
+};
+
+const stopReplProducerOnDocumentFunc = (counterMajorityCommitted) => {
+    return {document: {"_id": counterMajorityCommitted + 1}};
+};
+
+const opsFunc = (primary, ns, counterTotal, lsid) => {
+    assert.commandWorked(primary.getCollection(ns).runCommand("delete", {
+        deletes: Array.from({length: counterTotal}, (_, i) => ({q: {_id: i}, limit: 1})),
+        lsid,
+        txnNumber: NumberLong(2),
+    }));
+};
+
+const stmtMajorityCommittedFunc = (primary, ns, counterMajorityCommitted) => {
+    return {ns, "o._id": counterMajorityCommitted, "op": "d"};
+};
+
+const validateFunc = (secondary1, ns, counterMajorityCommitted, counterTotal, lsid) => {
+    // Insert doc in the range [0, counterMajorityCommitted] which should have been deleted.
+    assert.commandWorked(secondary1.getCollection(ns).runCommand("insert", {
+        documents: Array.from({length: counterMajorityCommitted + 1}, (_, i) => ({_id: i})),
+        writeConcern: {w: 5}
+    }));
+
+    // Docs in the range [counterMajorityCommitted + 1, counterTotal - 1] should exist because the
+    // delete statements were rolled back.
+    for (var i = counterMajorityCommitted + 1; i < counterTotal; i++) {
+        const docBeforeRetry = secondary1.getCollection(ns).findOne({_id: i});
+        assert.eq(docBeforeRetry, {_id: i});
+    }
+
+    // Retry the operation which should only delete the range
+    // [counterMajorityCommitted + 1, counterTotal - 1].
+    assert.commandWorked(secondary1.getCollection(ns).runCommand("delete", {
+        deletes: Array.from({length: counterTotal}, (_, i) => ({q: {_id: i}, limit: 1})),
+        lsid,
+        txnNumber: NumberLong(2),
+        writeConcern: {w: 5},
+    }));
+
+    // We should still find the documents in the range [0, counterMajorityCommitted].
+    for (var i = 0; i <= counterMajorityCommitted; i++) {
+        const docAfterRetry = secondary1.getCollection(ns).findOne({_id: i});
+        assert.eq(docAfterRetry, {_id: i});
+    }
+
+    // Docs in the range [counterMajorityCommitted + 1, counterTotal - 1] should be deleted by the
+    // retry.
+    for (var i = counterMajorityCommitted + 1; i < counterTotal; i++) {
+        const docAfterRetry = secondary1.getCollection(ns).findOne({_id: i});
+        assert.eq(docAfterRetry, null);
+    }
+};
+
+runTests(
+    initFunc, stopReplProducerOnDocumentFunc, opsFunc, stmtMajorityCommittedFunc, validateFunc);

jstests/replsets/rollback_with_coalesced_txn_table_updates_during_oplog_application_inserts.js

@@ -0,0 +1,57 @@
+/**
+ * Tests that the rollback procedure will update the 'config.transactions' table to be consistent
+ * with the node data at the 'stableTimestamp', specifically in the case where multiple insert ops
+ * to the 'config.transactions' table were coalesced into a single operation during secondary oplog
+ * application.
+ *
+ * @tags: [requires_persistence]
+ */
+
+import {FeatureFlagUtil} from "jstests/libs/feature_flag_util.js";
+import {
+    runTests
+} from
+    "jstests/replsets/libs/rollback_with_coalesced_txn_table_updates_during_oplog_application_helper.js";
+
+const initFunc = (primary, ns, counterTotal) => {
+    if (FeatureFlagUtil.isPresentAndEnabled(primary, "ReplicateVectoredInsertsTransactionally")) {
+        // Set the batch size to 2 so we're testing batching but don't have to insert a huge number
+        // of documents
+        assert.commandWorked(
+            primary.adminCommand({setParameter: 1, internalInsertMaxBatchSize: 2}));
+    }
+};
+
+const stopReplProducerOnDocumentFunc = (counterMajorityCommitted) => {
+    return {document: {"_id": counterMajorityCommitted + 1}};
+};
+
+const opsFunc = (primary, ns, counterTotal, lsid) => {
+    assert.commandWorked(primary.getCollection(ns).runCommand("insert", {
+        documents: Array.from({length: counterTotal}, (_, i) => ({_id: i})),
+        lsid,
+        txnNumber: NumberLong(2),
+    }));
+};
+
+const stmtMajorityCommittedFunc = (primary, ns, counterMajorityCommitted) => {
+    if (FeatureFlagUtil.isPresentAndEnabled(primary, "ReplicateVectoredInsertsTransactionally")) {
+        return {"o.applyOps.ns": ns, "o.applyOps.o._id": counterMajorityCommitted};
+    } else {
+        return {ns: ns, "o._id": counterMajorityCommitted};
+    }
+};
+
+const validateFunc = (secondary1, ns, counterMajorityCommitted, counterTotal, lsid) => {
+    // Make sure we don't re-execute operations that have already been inserted by making sure we
+    // we don't get a 'DuplicateKeyError'.
+    assert.commandWorked(secondary1.getCollection(ns).runCommand("insert", {
+        documents: Array.from({length: counterTotal}, (_, i) => ({_id: i})),
+        lsid,
+        txnNumber: NumberLong(2),
+        writeConcern: {w: 5},
+    }));
+};
+
+runTests(
+    initFunc, stopReplProducerOnDocumentFunc, opsFunc, stmtMajorityCommittedFunc, validateFunc);

jstests/replsets/rollback_with_coalesced_txn_table_updates_during_oplog_application_updates.js

@@ -0,0 +1,54 @@
+/**
+ * Tests that the rollback procedure will update the 'config.transactions' table to be consistent
+ * with the node data at the 'stableTimestamp', specifically in the case where multiple update ops
+ * to the 'config.transactions' table were coalesced into a single operation during secondary oplog
+ * application.
+ *
+ * @tags: [requires_persistence]
+ */
+
+import {
+    runTests
+} from
+    "jstests/replsets/libs/rollback_with_coalesced_txn_table_updates_during_oplog_application_helper.js";
+
+const initFunc = (primary, ns, counterTotal) => {
+    assert.commandWorked(
+        primary.getCollection(ns).insert({_id: 0, counter: 0}, {writeConcern: {w: 5}}));
+};
+
+const stopReplProducerOnDocumentFunc = (counterMajorityCommitted) => {
+    return {document: {"diff.u.counter": counterMajorityCommitted + 1}};
+};
+
+const opsFunc = (primary, ns, counterTotal, lsid) => {
+    assert.commandWorked(primary.getCollection(ns).runCommand("update", {
+        updates: Array.from({length: counterTotal}, () => ({q: {_id: 0}, u: {$inc: {counter: 1}}})),
+        lsid,
+        txnNumber: NumberLong(2),
+    }));
+};
+
+const stmtMajorityCommittedFunc = (primary, ns, counterMajorityCommitted) => {
+    return {ns, "o.diff.u.counter": counterMajorityCommitted};
+};
+
+const validateFunc = (secondary1, ns, counterMajorityCommitted, counterTotal, lsid) => {
+    const docBeforeRetry = secondary1.getCollection(ns).findOne({_id: 0});
+    assert.eq(docBeforeRetry, {_id: 0, counter: counterMajorityCommitted});
+
+    assert.commandWorked(secondary1.getCollection(ns).runCommand("update", {
+        updates: Array.from({length: counterTotal}, () => ({q: {_id: 0}, u: {$inc: {counter: 1}}})),
+        lsid,
+        txnNumber: NumberLong(2),
+        writeConcern: {w: 5},
+    }));
+
+    // Make sure we don't re-execute operations that have already been updated by making sure that
+    // counter equals exactly the counterTotal after the retry.
+    const docAfterRetry = secondary1.getCollection(ns).findOne({_id: 0});
+    assert.eq(docAfterRetry, {_id: 0, counter: counterTotal});
+};
+
+runTests(
+    initFunc, stopReplProducerOnDocumentFunc, opsFunc, stmtMajorityCommittedFunc, validateFunc);

jstests/replsets/rollback_with_coalesced_txn_table_updates_from_vectored_inserts.js

@@ -3,7 +3,9 @@
  * with the node data at the 'stableTimestamp', specifically in the case where multiple derived ops
  * to the 'config.transactions' table were coalesced into a single operation when performing
  * vectored inserts on the primary.
- *
+ * Note that when the 'ReplicateVectoredInsertsTransactionally' feature flag is enabled, updates to
+ * the 'config.transactions' table are not coalesced on the primary. However, we are still testing
+ * this as a sanity check in case the behavior changes in the future.
  * @tags: [requires_persistence]
  */
 
@@ -42,9 +44,14 @@ const [secondary1, secondary2] = rst.getSecondaries();
 // while the primary will apply all of the writes in a single storage transaction, the secondaries
 // will only apply up to insertBatchMajorityCommitted oplog entries.
 
-let insertBatchTotal;
-let insertBatchMajorityCommitted;
-let stopReplProducerOnDocumentFailpoints;
+let insertBatchTotal = 40;
+// Using an odd number ensures that when the 'ReplicateVectoredInsertsTransactionally' feature flag
+// is enabled, insertBatchMajorityCommitted + 1 will be in a different oplog entry. This is because
+// we're using an internal batching size of 2, resulting in pairs like [0, 1], [2, 3], etc.
+let insertBatchMajorityCommitted = insertBatchTotal - 5;
+let stopReplProducerOnDocumentFailpoints = [secondary1, secondary2].map(
+    conn => configureFailPoint(
+        conn, 'stopReplProducerOnDocument', {document: {"_id": insertBatchMajorityCommitted + 1}}));
 let oplogFilterForMajority;
 
 // When ReplicateVectoredInsertsTransactionally is enabled, inserts are batched into applyOps
@@ -53,25 +60,11 @@ if (FeatureFlagUtil.isPresentAndEnabled(primary, "ReplicateVectoredInsertsTransa
     // Set the batch size to 2 so we're testing batching but don't have to insert a huge number
     // of documents
     assert.commandWorked(primary.adminCommand({setParameter: 1, internalInsertMaxBatchSize: 2}));
-    // Using an odd number tests that the short batch (which should be an 'i', not an applyOps)
-    // works.
-    insertBatchTotal = 41;
-    insertBatchMajorityCommitted = insertBatchTotal - 3;
-    stopReplProducerOnDocumentFailpoints = [secondary1, secondary2].map(
-        conn => configureFailPoint(conn, 'stopReplProducerOnDocument', {
-            document: {"o.applyOps.o._id": insertBatchMajorityCommitted + 1}
-        }));
     oplogFilterForMajority = {
         "o.applyOps.ns": ns,
         "o.applyOps.o._id": insertBatchMajorityCommitted
     };
 } else {
-    insertBatchTotal = 20;
-    insertBatchMajorityCommitted = insertBatchTotal - 2;
-    stopReplProducerOnDocumentFailpoints = [secondary1, secondary2].map(
-        conn => configureFailPoint(conn,
-                                   'stopReplProducerOnDocument',
-                                   {document: {"_id": insertBatchMajorityCommitted + 1}}));
     oplogFilterForMajority = {ns: ns, "o._id": insertBatchMajorityCommitted};
 }