summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--etc/backports_required_for_multiversion_tests.yml4
-rw-r--r--jstests/concurrency/fsm_workloads/collection_uuid.js4
-rw-r--r--jstests/sharding/multi_writes_with_shard_version_ignored_dont_bubble_up_critical_section.js150
-rw-r--r--src/mongo/db/exec/batched_delete_stage.cpp10
-rw-r--r--src/mongo/db/exec/delete_stage.cpp61
-rw-r--r--src/mongo/db/exec/update_stage.cpp65
-rw-r--r--src/mongo/db/query/plan_executor.cpp4
-rw-r--r--src/mongo/db/query/plan_executor.h9
-rw-r--r--src/mongo/db/query/plan_executor_impl.cpp20
9 files changed, 291 insertions, 36 deletions
diff --git a/etc/backports_required_for_multiversion_tests.yml b/etc/backports_required_for_multiversion_tests.yml
index f8045db1af4..b543687e1c4 100644
--- a/etc/backports_required_for_multiversion_tests.yml
+++ b/etc/backports_required_for_multiversion_tests.yml
@@ -294,6 +294,8 @@ last-continuous:
ticket: SERVER-76489
- test_file: jstests/sharding/transfer_mods_large_batches.js
ticket: SERVER-78414
+ - test_file: jstests/sharding/multi_writes_with_shard_version_ignored_dont_bubble_up_critical_section.js
+ ticket: SERVER-61127
suites: null
last-lts:
all:
@@ -665,4 +667,6 @@ last-lts:
ticket: SERVER-67699
- test_file: jstests/sharding/transfer_mods_large_batches.js
ticket: SERVER-78414
+ - test_file: jstests/sharding/multi_writes_with_shard_version_ignored_dont_bubble_up_critical_section.js
+ ticket: SERVER-61127
suites: null
diff --git a/jstests/concurrency/fsm_workloads/collection_uuid.js b/jstests/concurrency/fsm_workloads/collection_uuid.js
index 7e16eac0246..011d3423a5b 100644
--- a/jstests/concurrency/fsm_workloads/collection_uuid.js
+++ b/jstests/concurrency/fsm_workloads/collection_uuid.js
@@ -54,6 +54,10 @@ const runCommandInLoop = function(
// TODO (SERVER-64449): Get rid of this exception
ErrorCodes.OBSOLETE_StaleShardVersion,
ErrorCodes.QueryPlanKilled,
+ // StaleConfig is usually retried by the mongos, but in situations where multiple errors
+ // have ocurred on the same batch and MultipleErrorsOcurred is returned, one of the errors
+ // could be StaleConfig and the other could be one that mongos does not retry the batch on.
+ ErrorCodes.StaleConfig,
];
let iteration = 0;
diff --git a/jstests/sharding/multi_writes_with_shard_version_ignored_dont_bubble_up_critical_section.js b/jstests/sharding/multi_writes_with_shard_version_ignored_dont_bubble_up_critical_section.js
new file mode 100644
index 00000000000..82155211bef
--- /dev/null
+++ b/jstests/sharding/multi_writes_with_shard_version_ignored_dont_bubble_up_critical_section.js
@@ -0,0 +1,150 @@
+/*
+ * Tests that multi-writes where the router attaches 'shardVersion: IGNORED' (i.e. if they need to
+ * target several shards AND are not part of a txn) do not bubble up StaleConfig errors due to
+ * ongoing critical sections. Instead, the shard yields and waits for the critical section to finish
+ * and then continues the write plan.
+ */
+
+(function() {
+"use strict";
+
+load('jstests/libs/parallel_shell_helpers.js');
+load("jstests/libs/fail_point_util.js");
+
+// Configure 'internalQueryExecYieldIterations' on both shards such that operations will yield on
+// each 10th PlanExecuter iteration.
+var st = new ShardingTest({
+ shards: 2,
+ rs: {setParameter: {internalQueryExecYieldIterations: 10}},
+ other: {enableBalancer: false}
+});
+
+const dbName = "test";
+const collName = "foo";
+const ns = dbName + "." + collName;
+const numDocs = 100;
+let coll = st.s.getCollection(ns);
+
+assert.commandWorked(
+ st.s.adminCommand({enableSharding: dbName, primaryShard: st.shard0.shardName}));
+
+function setupTest() {
+ coll.drop();
+ assert.commandWorked(st.s.adminCommand({shardCollection: ns, key: {x: 1}}));
+
+ // Create three chunks:
+ // - [MinKey, 0) initially on shard0 and has no documents. This chunk will be migrated during
+ // the test execution.
+ // - [0, numDocs) on shard 0. Contains 'numDocs' documents.
+ // - [numDocs, MaxKey) shard 1. Contains no documents.
+ assert.commandWorked(st.s.adminCommand({split: ns, middle: {x: 0}}));
+ assert.commandWorked(st.s.adminCommand({split: ns, middle: {x: numDocs}}));
+ assert.commandWorked(st.s.adminCommand(
+ {moveChunk: ns, find: {x: numDocs}, to: st.shard1.shardName, waitForDelete: true}));
+
+ jsTest.log("Inserting initial data.");
+ const bulkOp = coll.initializeOrderedBulkOp();
+ for (let i = 0; i < numDocs; ++i) {
+ bulkOp.insert({x: i, c: 0});
+ }
+ assert.commandWorked(bulkOp.execute());
+ jsTest.log("Inserted initial data.");
+}
+
+function runMigration() {
+ const awaitResult = startParallelShell(
+ funWithArgs(function(ns, toShard) {
+ jsTest.log("Starting migration.");
+ assert.commandWorked(db.adminCommand({moveChunk: ns, find: {x: -1}, to: toShard}));
+ jsTest.log("Completed migration.");
+ }, ns, st.shard1.shardName), st.s.port);
+
+ return awaitResult;
+}
+
+function updateOperationFn(shardColl, numInitialDocsOnShard0) {
+ load('jstests/sharding/libs/shard_versioning_util.js'); // For kIgnoredShardVersion
+
+ jsTest.log("Begin multi-update.");
+
+ // Send a multi-update with 'shardVersion: IGNORED' directly to the shard, as if we were a
+ // router.
+ const result = assert.commandWorked(shardColl.runCommand({
+ update: shardColl.getName(),
+ updates: [{q: {}, u: {$inc: {c: 1}}, multi: true}],
+ shardVersion: ShardVersioningUtil.kIgnoredShardVersion
+ }));
+
+ jsTest.log("End multi-update. Result: " + tojson(result));
+
+ // Check that all documents got updates. Despite the weak guarantees of {multi: true} writes
+ // concurrent with migrations, this has to be the case in this test because the migrated chunk
+ // does not contain any document.
+ assert.eq(numInitialDocsOnShard0, shardColl.find({c: 1}).itcount());
+}
+
+function deleteOperationFn(shardColl, numInitialDocsOnShard0) {
+ load('jstests/sharding/libs/shard_versioning_util.js'); // For kIgnoredShardVersion
+
+ jsTest.log("Begin multi-delete");
+
+ // Send a multi-delete with 'shardVersion: IGNORED' directly to the shard, as if we were a
+ // router.
+ const result = assert.commandWorked(shardColl.runCommand({
+ delete: shardColl.getName(),
+ deletes: [{q: {}, limit: 0}],
+ shardVersion: ShardVersioningUtil.kIgnoredShardVersion
+ }));
+
+ jsTest.log("End multi-delete. Result: " + tojson(result));
+
+ // Check that all documents got deleted. Despite the weak guarantees of {multi: true} writes
+ // concurrent with migrations, this has to be the case in this test because the migrated chunk
+ // does not contain any document.
+ assert.eq(0, shardColl.find().itcount());
+}
+
+function runTest(writeOpFn) {
+ setupTest();
+
+ let fp1 = configureFailPoint(
+ st.rs0.getPrimary(), 'setYieldAllLocksHang', {namespace: coll.getFullName()});
+
+ const awaitWriteResult = startParallelShell(
+ funWithArgs(function(writeOpFn, dbName, collName, numDocs) {
+ const shardColl = db.getSiblingDB(dbName)[collName];
+ writeOpFn(shardColl, numDocs);
+ }, writeOpFn, coll.getDB().getName(), coll.getName(), numDocs), st.rs0.getPrimary().port);
+
+ // Wait for the write op to yield.
+ fp1.wait();
+ jsTest.log("Multi-write yielded");
+
+ // Start chunk migration and wait for it to enter the critical section.
+ let failpointHangMigrationWhileInCriticalSection =
+ configureFailPoint(st.rs0.getPrimary(), 'moveChunkHangAtStep5');
+ const awaitMigration = runMigration();
+ failpointHangMigrationWhileInCriticalSection.wait();
+
+ // Let the multi-write resume from the yield.
+ jsTest.log("Resuming yielded multi-write");
+ fp1.off();
+
+ // Let the multi-write run for a bit after the resuming from yield. It will encounter the
+ // critical section.
+ sleep(1000);
+
+ // Let the migration continue and release the critical section.
+ jsTest.log("Letting migration exit its critical section and complete");
+ failpointHangMigrationWhileInCriticalSection.off();
+ awaitMigration();
+
+ // Wait for the write op to finish. It should succeed.
+ awaitWriteResult();
+}
+
+runTest(updateOperationFn);
+runTest(deleteOperationFn);
+
+st.stop();
+})();
diff --git a/src/mongo/db/exec/batched_delete_stage.cpp b/src/mongo/db/exec/batched_delete_stage.cpp
index 78bfd05e352..fb8be63cd8a 100644
--- a/src/mongo/db/exec/batched_delete_stage.cpp
+++ b/src/mongo/db/exec/batched_delete_stage.cpp
@@ -242,6 +242,16 @@ PlanStage::StageState BatchedDeleteStage::_deleteBatch(WorkingSetID* out) {
wuow.commit();
} catch (const WriteConflictException&) {
return _prepareToRetryDrainAfterWCE(out, recordsThatNoLongerMatch);
+ } catch (const ExceptionFor<ErrorCodes::StaleConfig>& ex) {
+ if (ex->getVersionReceived() == ChunkVersion::IGNORED() && ex->getCriticalSectionSignal()) {
+ // If ChunkVersion is IGNORED and we encountered a critical section, then yield, wait
+ // for critical section to finish and then we'll resume the write from the point we had
+ // left. We do this to prevent large multi-writes from repeatedly failing due to
+ // StaleConfig and exhausting the mongos retry attempts.
+ planExecutorShardingCriticalSectionFuture(opCtx()) = ex->getCriticalSectionSignal();
+ return _prepareToRetryDrainAfterWCE(out, recordsThatNoLongerMatch);
+ }
+ throw;
}
incrementSSSMetricNoOverflow(batchedDeletesSSS.docs, docsDeleted);
diff --git a/src/mongo/db/exec/delete_stage.cpp b/src/mongo/db/exec/delete_stage.cpp
index 83941b91c4e..581ef2ef294 100644
--- a/src/mongo/db/exec/delete_stage.cpp
+++ b/src/mongo/db/exec/delete_stage.cpp
@@ -178,23 +178,38 @@ PlanStage::StageState DeleteStage::doWork(WorkingSetID* out) {
bool writeToOrphan = false;
if (!_params->isExplain && !_params->fromMigrate) {
- const auto action = _preWriteFilter.computeAction(member->doc.value());
- if (action == write_stage_common::PreWriteFilter::Action::kSkip) {
- LOGV2_DEBUG(5983201,
- 3,
- "Skipping delete operation to orphan document to prevent a wrong change "
- "stream event",
- "namespace"_attr = collection()->ns(),
- "record"_attr = member->doc.value());
- return PlanStage::NEED_TIME;
- } else if (action == write_stage_common::PreWriteFilter::Action::kWriteAsFromMigrate) {
- LOGV2_DEBUG(6184700,
- 3,
- "Marking delete operation to orphan document with the fromMigrate flag "
- "to prevent a wrong change stream event",
- "namespace"_attr = collection()->ns(),
- "record"_attr = member->doc.value());
- writeToOrphan = true;
+ try {
+ const auto action = _preWriteFilter.computeAction(member->doc.value());
+ if (action == write_stage_common::PreWriteFilter::Action::kSkip) {
+ LOGV2_DEBUG(
+ 5983201,
+ 3,
+ "Skipping delete operation to orphan document to prevent a wrong change "
+ "stream event",
+ "namespace"_attr = collection()->ns(),
+ "record"_attr = member->doc.value());
+ return PlanStage::NEED_TIME;
+ } else if (action == write_stage_common::PreWriteFilter::Action::kWriteAsFromMigrate) {
+ LOGV2_DEBUG(6184700,
+ 3,
+ "Marking delete operation to orphan document with the fromMigrate flag "
+ "to prevent a wrong change stream event",
+ "namespace"_attr = collection()->ns(),
+ "record"_attr = member->doc.value());
+ writeToOrphan = true;
+ }
+ } catch (const ExceptionFor<ErrorCodes::StaleConfig>& ex) {
+ if (ex->getVersionReceived() == ChunkVersion::IGNORED() &&
+ ex->getCriticalSectionSignal()) {
+ // If ChunkVersion is IGNORED and we encountered a critical section, then yield,
+ // wait for the critical section to finish and then we'll resume the write from the
+ // point we had left. We do this to prevent large multi-writes from repeatedly
+ // failing due to StaleConfig and exhausting the mongos retry attempts.
+ planExecutorShardingCriticalSectionFuture(opCtx()) = ex->getCriticalSectionSignal();
+ memberFreer.dismiss(); // Keep this member around so we can retry deleting it.
+ return prepareToRetryWSM(id, out);
+ }
+ throw;
}
}
@@ -235,6 +250,18 @@ PlanStage::StageState DeleteStage::doWork(WorkingSetID* out) {
} catch (const WriteConflictException&) {
memberFreer.dismiss(); // Keep this member around so we can retry deleting it.
return prepareToRetryWSM(id, out);
+ } catch (const ExceptionFor<ErrorCodes::StaleConfig>& ex) {
+ if (ex->getVersionReceived() == ChunkVersion::IGNORED() &&
+ ex->getCriticalSectionSignal()) {
+ // If ChunkVersion is IGNORED and we encountered a critical section, then yield,
+ // wait for the critical section to finish and then we'll resume the write from the
+ // point we had left. We do this to prevent large multi-writes from repeatedly
+ // failing due to StaleConfig and exhausting the mongos retry attempts.
+ planExecutorShardingCriticalSectionFuture(opCtx()) = ex->getCriticalSectionSignal();
+ memberFreer.dismiss(); // Keep this member around so we can retry deleting it.
+ return prepareToRetryWSM(id, out);
+ }
+ throw;
}
}
_specificStats.docsDeleted += _params->numStatsForDoc ? _params->numStatsForDoc(bsonObjDoc) : 1;
diff --git a/src/mongo/db/exec/update_stage.cpp b/src/mongo/db/exec/update_stage.cpp
index 5a273dc2d89..0ed308dc79c 100644
--- a/src/mongo/db/exec/update_stage.cpp
+++ b/src/mongo/db/exec/update_stage.cpp
@@ -459,24 +459,41 @@ PlanStage::StageState UpdateStage::doWork(WorkingSetID* out) {
bool writeToOrphan = false;
if (!_params.request->explain() && _isUserInitiatedWrite) {
- const auto action = _preWriteFilter.computeAction(member->doc.value());
- if (action == write_stage_common::PreWriteFilter::Action::kSkip) {
- LOGV2_DEBUG(
- 5983200,
- 3,
- "Skipping update operation to orphan document to prevent a wrong change "
- "stream event",
- "namespace"_attr = collection()->ns(),
- "record"_attr = member->doc.value());
- return PlanStage::NEED_TIME;
- } else if (action == write_stage_common::PreWriteFilter::Action::kWriteAsFromMigrate) {
- LOGV2_DEBUG(6184701,
- 3,
- "Marking update operation to orphan document with the fromMigrate flag "
- "to prevent a wrong change stream event",
- "namespace"_attr = collection()->ns(),
- "record"_attr = member->doc.value());
- writeToOrphan = true;
+ try {
+ const auto action = _preWriteFilter.computeAction(member->doc.value());
+ if (action == write_stage_common::PreWriteFilter::Action::kSkip) {
+ LOGV2_DEBUG(
+ 5983200,
+ 3,
+ "Skipping update operation to orphan document to prevent a wrong change "
+ "stream event",
+ "namespace"_attr = collection()->ns(),
+ "record"_attr = member->doc.value());
+ return PlanStage::NEED_TIME;
+ } else if (action ==
+ write_stage_common::PreWriteFilter::Action::kWriteAsFromMigrate) {
+ LOGV2_DEBUG(
+ 6184701,
+ 3,
+ "Marking update operation to orphan document with the fromMigrate flag "
+ "to prevent a wrong change stream event",
+ "namespace"_attr = collection()->ns(),
+ "record"_attr = member->doc.value());
+ writeToOrphan = true;
+ }
+ } catch (const ExceptionFor<ErrorCodes::StaleConfig>& ex) {
+ if (ex->getVersionReceived() == ChunkVersion::IGNORED() &&
+ ex->getCriticalSectionSignal()) {
+ // If ChunkVersion is IGNORED and we encountered a critical section, then yield,
+ // wait for critical section to finish and then we'll resume the write from the
+ // point we had left. We do this to prevent large multi-writes from repeatedly
+ // failing due to StaleConfig and exhausting the mongos retry attempts.
+ planExecutorShardingCriticalSectionFuture(opCtx()) =
+ ex->getCriticalSectionSignal();
+ memberFreer.dismiss(); // Keep this member around so we can retry deleting it.
+ return prepareToRetryWSM(id, out);
+ }
+ throw;
}
}
@@ -506,6 +523,18 @@ PlanStage::StageState UpdateStage::doWork(WorkingSetID* out) {
} catch (const WriteConflictException&) {
memberFreer.dismiss(); // Keep this member around so we can retry updating it.
return prepareToRetryWSM(id, out);
+ } catch (const ExceptionFor<ErrorCodes::StaleConfig>& ex) {
+ if (ex->getVersionReceived() == ChunkVersion::IGNORED() &&
+ ex->getCriticalSectionSignal()) {
+ // If ChunkVersion is IGNORED and we encountered a critical section, then yield,
+ // wait for critical section to finish and then we'll resume the write from the
+ // point we had left. We do this to prevent large multi-writes from repeatedly
+ // failing due to StaleConfig and exhausting the mongos retry attempts.
+ planExecutorShardingCriticalSectionFuture(opCtx()) = ex->getCriticalSectionSignal();
+ memberFreer.dismiss(); // Keep this member around so we can retry updating it.
+ return prepareToRetryWSM(id, out);
+ }
+ throw;
}
// Set member's obj to be the doc we want to return.
diff --git a/src/mongo/db/query/plan_executor.cpp b/src/mongo/db/query/plan_executor.cpp
index ee41d15d84c..99b2fd8fefa 100644
--- a/src/mongo/db/query/plan_executor.cpp
+++ b/src/mongo/db/query/plan_executor.cpp
@@ -38,6 +38,10 @@ namespace {
MONGO_FAIL_POINT_DEFINE(planExecutorAlwaysFails);
} // namespace
+const OperationContext::Decoration<boost::optional<SharedSemiFuture<void>>>
+ planExecutorShardingCriticalSectionFuture =
+ OperationContext::declareDecoration<boost::optional<SharedSemiFuture<void>>>();
+
std::string PlanExecutor::stateToStr(ExecState execState) {
switch (execState) {
case PlanExecutor::ADVANCED:
diff --git a/src/mongo/db/query/plan_executor.h b/src/mongo/db/query/plan_executor.h
index 33fbd075b93..069fb4b4608 100644
--- a/src/mongo/db/query/plan_executor.h
+++ b/src/mongo/db/query/plan_executor.h
@@ -56,6 +56,15 @@ class RecordId;
extern const OperationContext::Decoration<repl::OpTime> clientsLastKnownCommittedOpTime;
/**
+ * If a plan yielded because it encountered a sharding critical section,
+ * 'planExecutorShardingCriticalSectionFuture' will be set to a future that becomes ready when the
+ * critical section ends. This future can be waited on to hold off resuming the plan execution while
+ * the critical section is still active.
+ */
+extern const OperationContext::Decoration<boost::optional<SharedSemiFuture<void>>>
+ planExecutorShardingCriticalSectionFuture;
+
+/**
* A PlanExecutor is the abstraction that knows how to crank a tree of stages into execution.
* The executor is usually part of a larger abstraction that is interacting with the cache
* and/or the query optimizer.
diff --git a/src/mongo/db/query/plan_executor_impl.cpp b/src/mongo/db/query/plan_executor_impl.cpp
index c3ae2946a38..55253f6e02d 100644
--- a/src/mongo/db/query/plan_executor_impl.cpp
+++ b/src/mongo/db/query/plan_executor_impl.cpp
@@ -62,6 +62,7 @@
#include "mongo/db/query/plan_yield_policy_impl.h"
#include "mongo/db/query/yield_policy_callbacks_impl.h"
#include "mongo/db/repl/replication_coordinator.h"
+#include "mongo/db/s/operation_sharding_state.h"
#include "mongo/db/service_context.h"
#include "mongo/logv2/log.h"
#include "mongo/util/fail_point.h"
@@ -360,8 +361,25 @@ PlanExecutor::ExecState PlanExecutorImpl::_getNextImpl(Snapshotted<Document>* ob
// 2) some stage requested a yield, or
// 3) we need to yield and retry due to a WriteConflictException.
// In all cases, the actual yielding happens here.
+
+ const auto whileYieldingFn = [&]() {
+ // If we yielded because we encountered a sharding critical section, wait for the
+ // critical section to end before continuing. By waiting for the critical section to be
+ // exited we avoid busy spinning immediately and encountering the same critical section
+ // again. It is important that this wait happens after having released the lock
+ // hierarchy -- otherwise deadlocks could happen, or the very least, locks would be
+ // unnecessarily held while waiting.
+ const auto& shardingCriticalSection = planExecutorShardingCriticalSectionFuture(_opCtx);
+ if (shardingCriticalSection) {
+ OperationShardingState::waitForCriticalSectionToComplete(_opCtx,
+ *shardingCriticalSection)
+ .ignore();
+ planExecutorShardingCriticalSectionFuture(_opCtx).reset();
+ }
+ };
+
if (_yieldPolicy->shouldYieldOrInterrupt(_opCtx)) {
- uassertStatusOK(_yieldPolicy->yieldOrInterrupt(_opCtx));
+ uassertStatusOK(_yieldPolicy->yieldOrInterrupt(_opCtx, whileYieldingFn));
}
WorkingSetID id = WorkingSet::INVALID_ID;
generated by cgit v1.2.3 (git 2.39.1) at 2026-09-11 09:49:44 +0000