diff options
| author | Lucas de Castro Borges <lucas@gnuabordo.com.br> | 2025-02-11 15:07:35 -0300 |
|---|---|---|
| committer | Lucas de Castro Borges <lucas@gnuabordo.com.br> | 2025-02-11 15:07:35 -0300 |
| commit | 4cb8841196d0625dfa3825aa326f071cd27c7b8b (patch) | |
| tree | 1682a647d4463397c119183369ae6f750d5fdcff /src/mongo/s/write_ops | |
| parent | aa03c6362cbaa767638e6eed9b031d86dd2643d1 (diff) | |
| parent | 8f0827553e09872941945a093b647a4211a9db7f (diff) | |
Update upstream source from tag 'upstream/6.0.0'master
Update to upstream version '6.0.0'
with Debian dir 5604a80ec1c96ca76f25f40d78e6ef855abec322
Diffstat (limited to 'src/mongo/s/write_ops')
| -rw-r--r-- | src/mongo/s/write_ops/batch_write_exec.cpp | 24 | ||||
| -rw-r--r-- | src/mongo/s/write_ops/batch_write_exec_test.cpp | 508 | ||||
| -rw-r--r-- | src/mongo/s/write_ops/batch_write_op.cpp | 78 | ||||
| -rw-r--r-- | src/mongo/s/write_ops/batch_write_op_test.cpp | 29 | ||||
| -rw-r--r-- | src/mongo/s/write_ops/batched_command_request.cpp | 27 | ||||
| -rw-r--r-- | src/mongo/s/write_ops/batched_command_request.h | 34 | ||||
| -rw-r--r-- | src/mongo/s/write_ops/batched_command_request_test.cpp | 2 | ||||
| -rw-r--r-- | src/mongo/s/write_ops/write_op.cpp | 51 |
8 files changed, 162 insertions, 591 deletions
diff --git a/src/mongo/s/write_ops/batch_write_exec.cpp b/src/mongo/s/write_ops/batch_write_exec.cpp index 980d376b607..45b89579dbb 100644 --- a/src/mongo/s/write_ops/batch_write_exec.cpp +++ b/src/mongo/s/write_ops/batch_write_exec.cpp @@ -104,25 +104,6 @@ bool hasTransientTransactionError(const BatchedCommandResponse& response) { // applies when no writes are occurring and metadata is not changing on reload. const int kMaxRoundsWithoutProgress(5); -/** - * Provides the write concern with which child batches have to be internally submitted. - */ -boost::optional<WriteConcernOptions> getWriteConcernForChildBatch(OperationContext* opCtx) { - // Per-operation write concern is not supported in transactions. - if (TransactionRouter::get(opCtx)) { - return boost::none; - } - - // Retrieve the WC specified by the remote client; in case of "fire and forget" request, the WC - // needs to be upgraded to "w: 1" for the sharding protocol to correctly handle internal - // writeErrors. - auto wc = opCtx->getWriteConcern(); - if (!wc.requiresWriteAcknowledgement()) { - wc.w = 1; - } - - return wc; -} } // namespace void BatchWriteExec::executeBatch(OperationContext* opCtx, @@ -214,7 +195,6 @@ void BatchWriteExec::executeBatch(OperationContext* opCtx, // std::vector<AsyncRequestsSender::Request> requests; - const auto wcSettingForChildBatch = getWriteConcernForChildBatch(opCtx); // Get as many batches as we can at once for (auto&& childBatch : childBatches) { @@ -237,10 +217,6 @@ void BatchWriteExec::executeBatch(OperationContext* opCtx, BSONObjBuilder requestBuilder; shardBatchRequest.serialize(&requestBuilder); logical_session_id_helpers::serializeLsidAndTxnNumber(opCtx, &requestBuilder); - if (wcSettingForChildBatch) { - requestBuilder.append(WriteConcernOptions::kWriteConcernField, - wcSettingForChildBatch->toBSON()); - } return requestBuilder.obj(); }(); diff --git a/src/mongo/s/write_ops/batch_write_exec_test.cpp b/src/mongo/s/write_ops/batch_write_exec_test.cpp index 3e90e5b65d3..3b4079b4c84 100644 --- a/src/mongo/s/write_ops/batch_write_exec_test.cpp +++ b/src/mongo/s/write_ops/batch_write_exec_test.cpp @@ -33,14 +33,12 @@ #include "mongo/client/remote_command_targeter_factory_mock.h" #include "mongo/client/remote_command_targeter_mock.h" #include "mongo/db/commands.h" -#include "mongo/db/concurrency/write_conflict_exception.h" #include "mongo/db/logical_session_id.h" #include "mongo/db/vector_clock.h" #include "mongo/s/catalog/type_shard.h" #include "mongo/s/client/shard_registry.h" #include "mongo/s/mock_ns_targeter.h" #include "mongo/s/session_catalog_router.h" -#include "mongo/s/shard_cannot_refresh_due_to_locks_held_exception.h" #include "mongo/s/sharding_router_test_fixture.h" #include "mongo/s/stale_exception.h" #include "mongo/s/transaction_router.h" @@ -202,40 +200,6 @@ BSONObj expectInsertsReturnTenantMigrationAbortedErrorsBase( return tenantMigrationAbortedResponse.obj(); } -BSONObj expectInsertsReturnCannotRefreshErrorsBase(const NamespaceString& nss, - const std::vector<BSONObj>& expected, - const executor::RemoteCommandRequest& request) { - ASSERT_EQUALS(nss.db(), request.dbname); - - const auto opMsgRequest(OpMsgRequest::fromDBAndBody(request.dbname, request.cmdObj)); - const auto actualBatchedInsert(BatchedCommandRequest::parseInsert(opMsgRequest)); - ASSERT_EQUALS(nss.toString(), actualBatchedInsert.getNS().ns()); - - const auto& inserted = actualBatchedInsert.getInsertRequest().getDocuments(); - ASSERT_EQUALS(expected.size(), inserted.size()); - - auto itInserted = inserted.begin(); - auto itExpected = expected.begin(); - - for (; itInserted != inserted.end(); itInserted++, itExpected++) { - ASSERT_BSONOBJ_EQ(*itExpected, *itInserted); - } - - BatchedCommandResponse cannotRefreshResponse; - cannotRefreshResponse.setStatus(Status::OK()); - cannotRefreshResponse.setN(0); - - // Report a ShardCannotRefreshDueToLocksHeld error for each write in the batch. - int i = 0; - for (itInserted = inserted.begin(); itInserted != inserted.end(); ++itInserted) { - cannotRefreshResponse.addToErrDetails(write_ops::WriteError( - i, Status(ShardCannotRefreshDueToLocksHeldInfo(nss), "Catalog cache busy in refresh"))); - ++i; - } - - return cannotRefreshResponse.toBSON(); -} - /** * Mimics a single shard backend for a particular collection which can be initialized with a * set of write command results to return. @@ -335,12 +299,6 @@ public: }); } - void expectInsertsReturnCannotRefreshErrors(const std::vector<BSONObj>& expected) { - onCommandForPoolExecutor([&](const executor::RemoteCommandRequest& request) { - return expectInsertsReturnCannotRefreshErrorsBase(nss, expected, request); - }); - } - void expectInsertsReturnError(const std::vector<BSONObj>& expected, const BatchedCommandResponse& errResponse) { onCommandForPoolExecutor([&](const executor::RemoteCommandRequest& request) { @@ -397,6 +355,7 @@ TEST_F(BatchWriteExecTest, SingleOpUnordered) { insertOp.setDocuments({BSON("x" << 1)}); return insertOp; }()); + request.setWriteConcern(BSONObj()); // Do single-target, single doc batch write op auto future = launchAsync([&] { @@ -435,6 +394,7 @@ TEST_F(BatchWriteExecTest, SingleUpdateTargetsShardWithLet) { << "100")))}); return updateOp; }()); + updateRequest.setWriteConcern(BSONObj()); const static auto epoch = OID::gen(); const static Timestamp timestamp(2); @@ -520,6 +480,7 @@ TEST_F(BatchWriteExecTest, SingleDeleteTargetsShardWithLet) { deleteOp.setDeletes(std::vector{write_ops::DeleteOpEntry(q, false)}); return deleteOp; }()); + deleteRequest.setWriteConcern(BSONObj()); const static auto epoch = OID::gen(); @@ -605,6 +566,7 @@ TEST_F(BatchWriteExecTest, MultiOpLargeOrdered) { insertOp.setDocuments(docsToInsert); return insertOp; }()); + request.setWriteConcern(BSONObj()); auto future = launchAsync([&] { BatchedCommandResponse response; @@ -637,6 +599,7 @@ TEST_F(BatchWriteExecTest, SingleOpUnorderedError) { insertOp.setDocuments({BSON("x" << 1)}); return insertOp; }()); + request.setWriteConcern(BSONObj()); auto future = launchAsync([&] { BatchedCommandResponse response; @@ -677,6 +640,7 @@ TEST_F(BatchWriteExecTest, MultiOpLargeUnorderedWithStaleShardVersionError) { insertOp.setDocuments(docsToInsert); return insertOp; }()); + request.setWriteConcern(BSONObj()); auto future = launchAsync([&] { BatchedCommandResponse response; @@ -708,6 +672,7 @@ TEST_F(BatchWriteExecTest, StaleShardVersionReturnedFromBatchWithSingleMultiWrit write_ops::UpdateModification::parseFromClassicUpdate(BSON("Key" << 100)))}); return updateOp; }()); + request.setWriteConcern(BSONObj()); const static auto epoch = OID::gen(); const static Timestamp timestamp(2); @@ -785,43 +750,6 @@ TEST_F(BatchWriteExecTest, StaleShardVersionReturnedFromBatchWithSingleMultiWrit ASSERT_EQ(3, response.getNModified()); } -TEST_F(BatchWriteExecTest, MultiOpLargeUnorderedWithCannotRefreshError) { - const int kNumDocsToInsert = 100'000; - - std::vector<BSONObj> docsToInsert; - docsToInsert.reserve(kNumDocsToInsert); - for (int i = 0; i < kNumDocsToInsert; i++) { - docsToInsert.push_back(BSON("_id" << i)); - } - - BatchedCommandRequest request([&] { - write_ops::InsertCommandRequest insertOp(nss); - insertOp.setWriteCommandRequestBase([] { - write_ops::WriteCommandRequestBase writeCommandBase; - writeCommandBase.setOrdered(false); - return writeCommandBase; - }()); - insertOp.setDocuments(docsToInsert); - return insertOp; - }()); - - auto future = launchAsync([&] { - BatchedCommandResponse response; - BatchWriteExecStats stats; - BatchWriteExec::executeBatch( - operationContext(), singleShardNSTargeter, request, &response, &stats); - - ASSERT(response.getOk()); - ASSERT_EQ(kNumDocsToInsert, response.getN()); - }); - - expectInsertsReturnCannotRefreshErrors({docsToInsert.begin(), docsToInsert.begin() + 60133}); - expectInsertsReturnSuccess({docsToInsert.begin(), docsToInsert.begin() + 60133}); - expectInsertsReturnSuccess({docsToInsert.begin() + 60133, docsToInsert.end()}); - - future.default_timed_get(); -} - TEST_F(BatchWriteExecTest, RetryableErrorReturnedFromMultiWriteWithShard1AllOKShard2AllStaleShardVersion) { BatchedCommandRequest request([&] { @@ -840,6 +768,7 @@ TEST_F(BatchWriteExecTest, write_ops::UpdateModification::parseFromClassicUpdate(BSON("y" << 2)))}); return updateOp; }()); + request.setWriteConcern(BSONObj()); const static auto epoch = OID::gen(); const static Timestamp timestamp(2); @@ -943,6 +872,7 @@ TEST_F(BatchWriteExecTest, RetryableErrorReturnedFromMultiWriteWithShard1Firs) { write_ops::UpdateModification::parseFromClassicUpdate(BSON("y" << 2)))}); return updateOp; }()); + request.setWriteConcern(BSONObj()); const static auto epoch = OID::gen(); const static Timestamp timestamp(2); @@ -1056,6 +986,7 @@ TEST_F(BatchWriteExecTest, RetryableErrorReturnedFromMultiWriteWithShard1FirstOK write_ops::UpdateModification::parseFromClassicUpdate(BSON("y" << 2)))}); return updateOp; }()); + request.setWriteConcern(BSONObj()); const static auto epoch = OID::gen(); const static Timestamp timestamp(2); @@ -1165,6 +1096,7 @@ TEST_F(BatchWriteExecTest, RetryableErrorReturnedFromWriteWithShard1SSVShard2OK) write_ops::UpdateModification::parseFromClassicUpdate(BSON("x" << 1)))}); return updateOp; }()); + request.setWriteConcern(BSONObj()); const static auto epoch = OID::gen(); const static Timestamp timestamp(2); @@ -1264,6 +1196,7 @@ TEST_F(BatchWriteExecTest, StaleShardOp) { insertOp.setDocuments({BSON("x" << 1)}); return insertOp; }()); + request.setWriteConcern(BSONObj()); // Execute request auto future = launchAsync([&] { @@ -1295,6 +1228,7 @@ TEST_F(BatchWriteExecTest, MultiStaleShardOp) { insertOp.setDocuments({BSON("x" << 1)}); return insertOp; }()); + request.setWriteConcern(BSONObj()); auto future = launchAsync([&] { BatchedCommandResponse response; @@ -1332,6 +1266,7 @@ TEST_F(BatchWriteExecTest, TooManyStaleShardOp) { insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)}); return insertOp; }()); + request.setWriteConcern(BSONObj()); auto future = launchAsync([&] { BatchedCommandResponse response; @@ -1366,6 +1301,7 @@ TEST_F(BatchWriteExecTest, StaleDbOp) { insertOp.setDocuments({BSON("x" << 1)}); return insertOp; }()); + request.setWriteConcern(BSONObj()); // Execute request auto future = launchAsync([&] { @@ -1397,6 +1333,7 @@ TEST_F(BatchWriteExecTest, MultiStaleDbOp) { insertOp.setDocuments({BSON("x" << 1)}); return insertOp; }()); + request.setWriteConcern(BSONObj()); auto future = launchAsync([&] { BatchedCommandResponse response; @@ -1434,6 +1371,7 @@ TEST_F(BatchWriteExecTest, TooManyStaleDbOp) { insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)}); return insertOp; }()); + request.setWriteConcern(BSONObj()); auto future = launchAsync([&] { BatchedCommandResponse response; @@ -1457,73 +1395,6 @@ TEST_F(BatchWriteExecTest, TooManyStaleDbOp) { future.default_timed_get(); } -TEST_F(BatchWriteExecTest, MultiCannotRefreshShardOp) { - BatchedCommandRequest request([&] { - write_ops::InsertCommandRequest insertOp(nss); - insertOp.setWriteCommandRequestBase([] { - write_ops::WriteCommandRequestBase writeCommandBase; - writeCommandBase.setOrdered(false); - return writeCommandBase; - }()); - insertOp.setDocuments({BSON("x" << 1)}); - return insertOp; - }()); - - auto future = launchAsync([&] { - BatchedCommandResponse response; - BatchWriteExecStats stats; - BatchWriteExec::executeBatch( - operationContext(), singleShardNSTargeter, request, &response, &stats); - ASSERT(response.getOk()); - }); - - const std::vector<BSONObj> expected{BSON("x" << 1)}; - - // Return multiple ShardCannotRefreshDueToLocksHeld errors, but less than the give-up number - for (int i = 0; i < 3; i++) { - expectInsertsReturnCannotRefreshErrors(expected); - } - - expectInsertsReturnSuccess(expected); - - future.default_timed_get(); -} - -TEST_F(BatchWriteExecTest, TooManyCannotRefreshShardOp) { - // Retry op in exec too many times b/c of busy catalog cache (the error is not expected to - // trigger a refresh on any implementation of NSTargeter). We should report a no progress error - // for everything in the batch. - BatchedCommandRequest request([&] { - write_ops::InsertCommandRequest insertOp(nss); - insertOp.setWriteCommandRequestBase([] { - write_ops::WriteCommandRequestBase writeCommandBase; - writeCommandBase.setOrdered(false); - return writeCommandBase; - }()); - insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)}); - return insertOp; - }()); - - auto future = launchAsync([&] { - BatchedCommandResponse response; - BatchWriteExecStats stats; - BatchWriteExec::executeBatch( - operationContext(), singleShardNSTargeter, request, &response, &stats); - ASSERT(response.getOk()); - ASSERT_EQ(0, response.getN()); - ASSERT(response.isErrDetailsSet()); - ASSERT_EQUALS(response.getErrDetailsAt(0).getStatus().code(), ErrorCodes::NoProgressMade); - ASSERT_EQUALS(response.getErrDetailsAt(1).getStatus().code(), ErrorCodes::NoProgressMade); - }); - - // Return multiple StaleShardVersion errors - for (int i = 0; i < (1 + kMaxRoundsWithoutProgress); i++) { - expectInsertsReturnCannotRefreshErrors({BSON("x" << 1), BSON("x" << 2)}); - } - - future.default_timed_get(); -} - TEST_F(BatchWriteExecTest, RetryableWritesLargeBatch) { // A retryable error without a txnNumber is not retried. @@ -1546,6 +1417,7 @@ TEST_F(BatchWriteExecTest, RetryableWritesLargeBatch) { insertOp.setDocuments(docsToInsert); return insertOp; }()); + request.setWriteConcern(BSONObj()); operationContext()->setLogicalSessionId(makeLogicalSessionIdForTest()); operationContext()->setTxnNumber(5); @@ -1580,6 +1452,7 @@ TEST_F(BatchWriteExecTest, RetryableErrorNoTxnNumber) { insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)}); return insertOp; }()); + request.setWriteConcern(BSONObj()); BatchedCommandResponse retryableErrResponse; retryableErrResponse.setStatus({ErrorCodes::NotWritablePrimary, "mock retryable error"}); @@ -1618,6 +1491,7 @@ TEST_F(BatchWriteExecTest, RetryableErrorTxnNumber) { insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)}); return insertOp; }()); + request.setWriteConcern(BSONObj()); operationContext()->setLogicalSessionId(makeLogicalSessionIdForTest()); operationContext()->setTxnNumber(5); @@ -1655,6 +1529,7 @@ TEST_F(BatchWriteExecTest, NonRetryableErrorTxnNumber) { insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)}); return insertOp; }()); + request.setWriteConcern(BSONObj()); operationContext()->setLogicalSessionId(makeLogicalSessionIdForTest()); operationContext()->setTxnNumber(5); @@ -1696,6 +1571,7 @@ TEST_F(BatchWriteExecTest, StaleEpochIsNotRetryable) { insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)}); return insertOp; }()); + request.setWriteConcern(BSONObj()); operationContext()->setLogicalSessionId(makeLogicalSessionIdForTest()); operationContext()->setTxnNumber(5); @@ -1723,249 +1599,6 @@ TEST_F(BatchWriteExecTest, StaleEpochIsNotRetryable) { future.default_timed_get(); } -TEST_F(BatchWriteExecTest, FireAndForgetBatchInsertGetsReplyWithOnlyOkStatus) { - const int kNumDocsToInsert = 5; - const std::string kDocValue("sample"); - - std::vector<BSONObj> docsToInsert; - docsToInsert.reserve(kNumDocsToInsert); - for (int i = 0; i < kNumDocsToInsert; i++) { - docsToInsert.push_back(BSON("_id" << i << "otherField" << kDocValue)); - } - - BatchedCommandRequest request([&] { - write_ops::InsertCommandRequest insertOp(nss); - insertOp.setWriteCommandRequestBase([] { - write_ops::WriteCommandRequestBase writeCommandBase; - writeCommandBase.setOrdered(true); - return writeCommandBase; - }()); - insertOp.setDocuments(docsToInsert); - return insertOp; - }()); - - auto future = launchAsync([&] { - BatchedCommandResponse response; - BatchWriteExecStats stats; - // Set Unacknowledged WC for a "fire & forget" request - auto opCtx = operationContext(); - opCtx->setWriteConcern( - WriteConcernOptions::parse(WriteConcernOptions::Unacknowledged).getValue()); - BatchWriteExec::executeBatch(opCtx, singleShardNSTargeter, request, &response, &stats); - - // The reply should only contain an OK status, without any further detail on - // the ops actually executed across the cluster. - BatchedCommandResponse expectedReplyToFireAndForgetRequest; - expectedReplyToFireAndForgetRequest.setStatus(Status::OK()); - ASSERT_EQUALS(response.toBSON().woCompare(expectedReplyToFireAndForgetRequest.toBSON()), 0); - }); - - expectInsertsReturnSuccess(docsToInsert.begin(), docsToInsert.end()); - - future.default_timed_get(); -} - -TEST_F(BatchWriteExecTest, FireAndForgetBatchUpdateGetsReplyWithOnlyOkStatus) { - BatchedCommandRequest request([&] { - write_ops::UpdateCommandRequest updateOp(nss); - updateOp.setWriteCommandRequestBase([] { - write_ops::WriteCommandRequestBase writeCommandBase; - writeCommandBase.setOrdered(false); - return writeCommandBase; - }()); - updateOp.setUpdates(std::vector{write_ops::UpdateOpEntry( - BSON("_id" << 100), - write_ops::UpdateModification::parseFromClassicUpdate(BSON("Key" << 100)))}); - return updateOp; - }()); - - const static auto epoch = OID::gen(); - const static Timestamp timestamp(2); - - // This allows the batch to target each write operation to perform this test - class MultiShardTargeter : public MockNSTargeter { - public: - using MockNSTargeter::MockNSTargeter; - - std::vector<ShardEndpoint> targetUpdate(OperationContext* opCtx, - const BatchItemRef& itemRef) const override { - if (targetAll) { - return std::vector{ - ShardEndpoint( - kShardName1, ChunkVersion(100, 200, epoch, timestamp), boost::none), - ShardEndpoint( - kShardName2, ChunkVersion(101, 200, epoch, timestamp), boost::none)}; - } else { - return std::vector{ShardEndpoint( - kShardName2, ChunkVersion(101, 200, epoch, timestamp), boost::none)}; - } - } - - bool targetAll = true; - }; - - MultiShardTargeter multiShardNSTargeter( - nss, - {MockRange( - ShardEndpoint(kShardName1, ChunkVersion(100, 200, epoch, timestamp), boost::none), - BSON("sk" << MINKEY), - BSON("sk" << 10)), - MockRange( - ShardEndpoint(kShardName2, ChunkVersion(101, 200, epoch, timestamp), boost::none), - BSON("sk" << 10), - BSON("sk" << MAXKEY))}); - auto future = launchAsync([&] { - // Set Unacknowledged WC for a "fire & forget" request - auto opCtx = operationContext(); - opCtx->setWriteConcern( - WriteConcernOptions::parse(WriteConcernOptions::Unacknowledged).getValue()); - - BatchedCommandResponse response; - BatchWriteExecStats stats; - BatchWriteExec::executeBatch(opCtx, multiShardNSTargeter, request, &response, &stats); - return response; - }); - - onCommandForPoolExecutor([&](const RemoteCommandRequest& request) { - ASSERT_EQ(kTestShardHost1, request.target); - - BatchedCommandResponse response; - response.setStatus(Status::OK()); - response.setNModified(1); - - return response.toBSON(); - }); - - onCommandForPoolExecutor([&](const RemoteCommandRequest& request) { - ASSERT_EQ(kTestShardHost2, request.target); - - BatchedCommandResponse response; - response.setStatus(Status::OK()); - response.setNModified(0); - response.addToErrDetails( - write_ops::WriteError(0, - Status(StaleConfigInfo(nss, - ChunkVersion(101, 200, epoch, timestamp), - ChunkVersion(105, 200, epoch, timestamp), - ShardId(kShardName2)), - "Stale error"))); - return response.toBSON(); - }); - - onCommandForPoolExecutor([&](const RemoteCommandRequest& request) { - ASSERT_EQ(kTestShardHost2, request.target); - - BatchedCommandResponse response; - response.setStatus(Status::OK()); - response.setNModified(2); - - return response.toBSON(); - }); - - // The reply should only contain an OK status, without any further detail on - // the ops actually executed across the cluster. - // (Despite of a "fire and forget" request, child batches still need to be internally processed - // before returning a reply). - auto response = future.default_timed_get(); - BatchedCommandResponse expectedReplyToFireAndForgetRequest; - expectedReplyToFireAndForgetRequest.setStatus(Status::OK()); - ASSERT_EQUALS(response.toBSON().woCompare(expectedReplyToFireAndForgetRequest.toBSON()), 0); -} - -TEST_F(BatchWriteExecTest, FireAndForgetBatchDeleteGetsReplyWithOnlyOkStatus) { - // Try to update the single doc where a let param is used in the shard key. - const auto let = BSON("y" << 100); - const auto rtc = LegacyRuntimeConstants{Date_t::now(), Timestamp(1, 1)}; - const auto q = BSON("x" - << "$$y"); - BatchedCommandRequest deleteRequest([&] { - write_ops::DeleteCommandRequest deleteOp(nss); - deleteOp.setWriteCommandRequestBase([] { - write_ops::WriteCommandRequestBase writeCommandBase; - writeCommandBase.setOrdered(false); - return writeCommandBase; - }()); - deleteOp.setLet(let); - deleteOp.setLegacyRuntimeConstants(rtc); - deleteOp.setDeletes(std::vector{write_ops::DeleteOpEntry(q, false)}); - return deleteOp; - }()); - - const static auto epoch = OID::gen(); - - class MultiShardTargeter : public MockNSTargeter { - public: - using MockNSTargeter::MockNSTargeter; - - protected: - std::vector<ShardEndpoint> targetDelete(OperationContext* opCtx, - const BatchItemRef& itemRef) const override { - return std::vector{ShardEndpoint( - kShardName2, ChunkVersion(101, 200, epoch, Timestamp(1, 1)), boost::none)}; - } - }; - - MultiShardTargeter multiShardNSTargeter( - nss, - {MockRange(ShardEndpoint( - kShardName1, ChunkVersion(100, 200, epoch, Timestamp(1, 1)), boost::none), - BSON("x" << MINKEY), - BSON("x" << 0)), - MockRange(ShardEndpoint( - kShardName2, ChunkVersion(101, 200, epoch, Timestamp(1, 1)), boost::none), - BSON("x" << 0), - BSON("x" << MAXKEY))}); - - auto future = launchAsync([&] { - BatchedCommandResponse response; - BatchWriteExecStats stats; - - // Set Unacknowledged WC for a "fire & forget" request - auto opCtx = operationContext(); - opCtx->setWriteConcern( - WriteConcernOptions::parse(WriteConcernOptions::Unacknowledged).getValue()); - - BatchWriteExec::executeBatch(opCtx, multiShardNSTargeter, deleteRequest, &response, &stats); - - return response; - }); - - // The update will hit the first shard. - onCommandForPoolExecutor( - [&](const RemoteCommandRequest& request) { - ASSERT_EQ(kTestShardHost2, request.target); - - BatchedCommandResponse response; - response.setStatus(Status::OK()); - - // Check that let params are propagated to shards. - const auto opMsgRequest(OpMsgRequest::fromDBAndBody(request.dbname, request.cmdObj)); - const auto actualBatchedUpdate(BatchedCommandRequest::parseDelete(opMsgRequest)); - ASSERT_BSONOBJ_EQ(let, actualBatchedUpdate.getLet().value_or(BSONObj())); - ASSERT_EQUALS(actualBatchedUpdate.getLegacyRuntimeConstants()->getLocalNow(), - rtc.getLocalNow()); - ASSERT_EQUALS(actualBatchedUpdate.getLegacyRuntimeConstants()->getClusterTime(), - rtc.getClusterTime()); - - // Check that let params are only forwarded and not evaluated. - auto expectedQ = BSON("x" - << "$$y"); - for (auto&& u : actualBatchedUpdate.getDeleteRequest().getDeletes()) - ASSERT_BSONOBJ_EQ(expectedQ, u.getQ()); - - return response.toBSON(); - }); - - // The reply should only contain an OK status, without any further detail on - // the ops actually executed across the cluster. - // (Despite of a "fire and forget" request, child batches still need to be internally processed - // before returning a reply). - auto response = future.default_timed_get(); - BatchedCommandResponse expectedReplyToFireAndForgetRequest; - expectedReplyToFireAndForgetRequest.setStatus(Status::OK()); - ASSERT_EQUALS(response.toBSON().woCompare(expectedReplyToFireAndForgetRequest.toBSON()), 0); -} - TEST_F(BatchWriteExecTest, TenantMigrationAbortedErrorOrderedOp) { const std::vector<BSONObj> expected{BSON("x" << 1), BSON("x" << 2), BSON("x" << 3)}; BatchedCommandRequest request([&] { @@ -1978,6 +1611,7 @@ TEST_F(BatchWriteExecTest, TenantMigrationAbortedErrorOrderedOp) { insertOp.setDocuments(expected); return insertOp; }()); + request.setWriteConcern(BSONObj()); // Execute request auto future = launchAsync([&] { @@ -2008,6 +1642,7 @@ TEST_F(BatchWriteExecTest, TenantMigrationAbortedErrorUnorderedOp) { insertOp.setDocuments(expected); return insertOp; }()); + request.setWriteConcern(BSONObj()); // Execute request auto future = launchAsync([&] { @@ -2038,6 +1673,7 @@ TEST_F(BatchWriteExecTest, MultipleTenantMigrationAbortedErrorUnorderedOp) { insertOp.setDocuments(expected); return insertOp; }()); + request.setWriteConcern(BSONObj()); const int numTenantMigrationAbortedErrors = 3; @@ -2072,6 +1708,7 @@ TEST_F(BatchWriteExecTest, MultipleTenantMigrationAbortedErrorOrderedOp) { insertOp.setDocuments(expected); return insertOp; }()); + request.setWriteConcern(BSONObj()); const int numTenantMigrationAbortedErrors = 3; @@ -2106,6 +1743,7 @@ TEST_F(BatchWriteExecTest, PartialTenantMigrationAbortedErrorOrderedOp) { insertOp.setDocuments(expected); return insertOp; }()); + request.setWriteConcern(BSONObj()); // Execute request auto future = launchAsync([&] { @@ -2138,6 +1776,7 @@ TEST_F(BatchWriteExecTest, PartialTenantMigrationErrorUnorderedOp) { insertOp.setDocuments(expected); return insertOp; }()); + request.setWriteConcern(BSONObj()); // Execute request auto future = launchAsync([&] { @@ -2222,6 +1861,7 @@ TEST_F(BatchWriteExecTargeterErrorTest, TargetedFailedAndErrorResponse) { write_ops::UpdateModification::parseFromClassicUpdate(BSON("Key" << 100)))}); return updateOp; }()); + request.setWriteConcern(BSONObj()); const static auto epoch = OID::gen(); const static Timestamp timestamp(2); @@ -2357,6 +1997,7 @@ TEST_F(BatchWriteExecTransactionTargeterErrorTest, TargetedFailedAndErrorRespons write_ops::UpdateModification::parseFromClassicUpdate(BSON("Key" << 100)))}); return updateOp; }()); + request.setWriteConcern(BSONObj()); const static auto epoch = OID::gen(); const static Timestamp timestamp(2); @@ -2500,6 +2141,7 @@ TEST_F(BatchWriteExecTransactionMultiShardTest, TargetedSucceededAndErrorRespons write_ops::UpdateModification::parseFromClassicUpdate(BSON("Key" << 100)))}); return updateOp; }()); + request.setWriteConcern(BSONObj()); const static auto epoch = OID::gen(); const static Timestamp timestamp(2); @@ -2625,22 +2267,6 @@ public: }); } - void expectInsertsReturnCannotRefreshErrors(const std::vector<BSONObj>& expected) { - onCommandForPoolExecutor([&](const executor::RemoteCommandRequest& request) { - BSONObjBuilder bob; - - bob.appendElementsUnique( - expectInsertsReturnCannotRefreshErrorsBase(nss, expected, request)); - - // Because this is the transaction-specific fixture, return transaction metadata in - // the response. - TxnResponseMetadata txnResponseMetadata(false /* readOnly */); - txnResponseMetadata.serialize(&bob); - - return bob.obj(); - }); - } - void expectInsertsReturnTransientTxnErrors(const std::vector<BSONObj>& expected) { onCommandForPoolExecutor([&](const executor::RemoteCommandRequest& request) { ASSERT_EQUALS(nss.db(), request.dbname); @@ -2690,6 +2316,7 @@ TEST_F(BatchWriteExecTransactionTest, ErrorInBatchThrows_CommandError) { insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)}); return insertOp; }()); + request.setWriteConcern(BSONObj()); auto future = launchAsync([&] { BatchedCommandResponse response; @@ -2721,6 +2348,7 @@ TEST_F(BatchWriteExecTransactionTest, ErrorInBatchSets_WriteError) { insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)}); return insertOp; }()); + request.setWriteConcern(BSONObj()); auto future = launchAsync([&] { BatchedCommandResponse response; @@ -2750,6 +2378,7 @@ TEST_F(BatchWriteExecTransactionTest, ErrorInBatchSets_WriteErrorOrdered) { insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)}); return insertOp; }()); + request.setWriteConcern(BSONObj()); auto future = launchAsync([&] { BatchedCommandResponse response; @@ -2768,64 +2397,6 @@ TEST_F(BatchWriteExecTransactionTest, ErrorInBatchSets_WriteErrorOrdered) { future.default_timed_get(); } -TEST_F(BatchWriteExecTransactionTest, ErrorInBatchSets_WriteErrorFromBusyCache) { - BatchedCommandRequest request([&] { - write_ops::InsertCommandRequest insertOp(nss); - insertOp.setWriteCommandRequestBase([] { - write_ops::WriteCommandRequestBase writeCommandBase; - writeCommandBase.setOrdered(false); - return writeCommandBase; - }()); - insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)}); - return insertOp; - }()); - - auto future = launchAsync([&] { - BatchedCommandResponse response; - BatchWriteExecStats stats; - BatchWriteExec::executeBatch( - operationContext(), singleShardNSTargeter, request, &response, &stats); - - ASSERT(response.isErrDetailsSet()); - ASSERT_GT(response.sizeErrDetails(), 0u); - ASSERT_EQ(ErrorCodes::ShardCannotRefreshDueToLocksHeld, - response.getErrDetailsAt(0).getStatus().code()); - }); - - expectInsertsReturnCannotRefreshErrors({BSON("x" << 1), BSON("x" << 2)}); - - future.default_timed_get(); -} - -TEST_F(BatchWriteExecTransactionTest, ErrorInBatchSets_WriteErrorOrderedFromBusyCache) { - BatchedCommandRequest request([&] { - write_ops::InsertCommandRequest insertOp(nss); - insertOp.setWriteCommandRequestBase([] { - write_ops::WriteCommandRequestBase writeCommandBase; - writeCommandBase.setOrdered(true); - return writeCommandBase; - }()); - insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)}); - return insertOp; - }()); - - auto future = launchAsync([&] { - BatchedCommandResponse response; - BatchWriteExecStats stats; - BatchWriteExec::executeBatch( - operationContext(), singleShardNSTargeter, request, &response, &stats); - - ASSERT(response.isErrDetailsSet()); - ASSERT_GT(response.sizeErrDetails(), 0u); - ASSERT_EQ(ErrorCodes::ShardCannotRefreshDueToLocksHeld, - response.getErrDetailsAt(0).getStatus().code()); - }); - - expectInsertsReturnCannotRefreshErrors({BSON("x" << 1), BSON("x" << 2)}); - - future.default_timed_get(); -} - TEST_F(BatchWriteExecTransactionTest, ErrorInBatchSets_TransientTxnError) { BatchedCommandRequest request([&] { write_ops::InsertCommandRequest insertOp(nss); @@ -2837,13 +2408,16 @@ TEST_F(BatchWriteExecTransactionTest, ErrorInBatchSets_TransientTxnError) { insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)}); return insertOp; }()); + request.setWriteConcern(BSONObj()); auto future = launchAsync([&] { BatchedCommandResponse response; BatchWriteExecStats stats; - ASSERT_THROWS(BatchWriteExec::executeBatch( - operationContext(), singleShardNSTargeter, request, &response, &stats), - WriteConflictException); + ASSERT_THROWS_CODE( + BatchWriteExec::executeBatch( + operationContext(), singleShardNSTargeter, request, &response, &stats), + AssertionException, + ErrorCodes::WriteConflict); }); expectInsertsReturnTransientTxnErrors({BSON("x" << 1), BSON("x" << 2)}); @@ -2862,6 +2436,7 @@ TEST_F(BatchWriteExecTransactionTest, ErrorInBatchSets_DispatchError) { insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)}); return insertOp; }()); + request.setWriteConcern(BSONObj()); auto future = launchAsync([&] { BatchedCommandResponse response; @@ -2893,6 +2468,7 @@ TEST_F(BatchWriteExecTransactionTest, ErrorInBatchSets_TransientDispatchError) { insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)}); return insertOp; }()); + request.setWriteConcern(BSONObj()); auto future = launchAsync([&] { BatchedCommandResponse response; diff --git a/src/mongo/s/write_ops/batch_write_op.cpp b/src/mongo/s/write_ops/batch_write_op.cpp index ed67f6c0088..a61ee3dd4bf 100644 --- a/src/mongo/s/write_ops/batch_write_op.cpp +++ b/src/mongo/s/write_ops/batch_write_op.cpp @@ -59,6 +59,7 @@ struct WriteErrorComp { // batches before serializing. // // TODO: Revisit when we revisit command limits in general +const int kEstUpdateOverheadBytes = (BSONObjMaxInternalSize - BSONObjMaxUserSize) / 100; const int kEstDeleteOverheadBytes = (BSONObjMaxInternalSize - BSONObjMaxUserSize) / 100; /** @@ -158,17 +159,51 @@ int getWriteSizeBytes(const WriteOp& writeOp) { return item.getDocument().objsize(); } else if (batchType == BatchedCommandRequest::BatchType_Update) { // Note: Be conservative here - it's okay if we send slightly too many batches. - const auto& update = item.getUpdate(); - auto estSize = write_ops::getUpdateSizeEstimate(update.getQ(), - update.getU(), - update.getC(), - update.getUpsertSupplied().has_value(), - update.getCollation(), - update.getArrayFilters(), - update.getHint()); + auto estSize = static_cast<int>(BSONObj::kMinBSONLength); + static const auto boolSize = 1; + + // Add the size of the 'collation' field, if present. + estSize += !item.getUpdate().getCollation() ? 0 + : (UpdateOpEntry::kCollationFieldName.size() + + item.getUpdate().getCollation()->objsize()); + + // Add the size of the 'arrayFilters' field, if present. + estSize += !item.getUpdate().getArrayFilters() ? 0 : ([&item]() { + auto size = BSONObj::kMinBSONLength + UpdateOpEntry::kArrayFiltersFieldName.size(); + for (auto&& filter : *item.getUpdate().getArrayFilters()) { + size += filter.objsize(); + } + return size; + })(); + + // Add the sizes of the 'multi' and 'upsert' fields. + estSize += UpdateOpEntry::kUpsertFieldName.size() + boolSize; + estSize += UpdateOpEntry::kMultiFieldName.size() + boolSize; + + // Add the size of 'upsertSupplied' field if present. + if (auto upsertSupplied = item.getUpdate().getUpsertSupplied()) { + estSize += UpdateOpEntry::kUpsertSuppliedFieldName.size() + boolSize; + } + + // Add the sizes of the 'q' and 'u' fields. + estSize += (UpdateOpEntry::kQFieldName.size() + item.getUpdate().getQ().objsize() + + UpdateOpEntry::kUFieldName.size() + item.getUpdate().getU().objsize()); + + // Add the size of the 'c' field if present. + if (auto constants = item.getUpdate().getC()) { + estSize += UpdateOpEntry::kCFieldName.size() + item.getUpdate().getC()->objsize(); + } + + // Add the size of 'hint' field if present. + if (auto hint = item.getUpdate().getHint(); !hint.isEmpty()) { + estSize += UpdateOpEntry::kHintFieldName.size() + hint.objsize(); + } + + // Finally, add the constant updateOp overhead size. + estSize += kEstUpdateOverheadBytes; // When running a debug build, verify that estSize is at least the BSON serialization size. - dassert(estSize >= update.toBSON().objsize()); + dassert(estSize >= item.getUpdate().toBSON().objsize()); return estSize; } else if (batchType == BatchedCommandRequest::BatchType_Delete) { // Note: Be conservative here - it's okay if we send slightly too many batches. @@ -568,9 +603,6 @@ BatchedCommandRequest BatchWriteOp::buildBatchRequest(const TargetedWriteBatch& wcb.setStmtIds(std::move(stmtIdsForOp)); } - wcb.setBypassEmptyTsReplacement( - _clientRequest.getWriteCommandRequestBase().getBypassEmptyTsReplacement()); - return wcb; }()); @@ -583,6 +615,19 @@ BatchedCommandRequest BatchWriteOp::buildBatchRequest(const TargetedWriteBatch& if (dbVersion) request.setDbVersion(*dbVersion); + if (_clientRequest.hasWriteConcern()) { + if (_clientRequest.isVerboseWC()) { + request.setWriteConcern(_clientRequest.getWriteConcern()); + } else { + // Mongos needs to send to the shard with w > 0 so it will be able to see the + // writeErrors + request.setWriteConcern(upgradeWriteConcern(_clientRequest.getWriteConcern())); + } + } else if (!TransactionRouter::get(_opCtx)) { + // Apply the WC from the opCtx (except if in a transaction). + request.setWriteConcern(_opCtx->getWriteConcern().toBSON()); + } + return request; } @@ -765,7 +810,7 @@ void BatchWriteOp::buildClientResponse(BatchedCommandResponse* batchResp) { batchResp->setStatus(Status::OK()); // For non-verbose, it's all we need. - if (!_opCtx->getWriteConcern().requiresWriteAcknowledgement()) { + if (!_clientRequest.isVerboseWC()) { return; } @@ -816,7 +861,12 @@ void BatchWriteOp::buildClientResponse(BatchedCommandResponse* batchResp) { } } - if (!_wcErrors.empty()) { + // Only return a write concern error if everything succeeded (unordered or ordered) + // OR if something succeeded and we're unordered + const bool orderedOps = _clientRequest.getWriteCommandRequestBase().getOrdered(); + const bool reportWCError = + errOps.empty() || (!orderedOps && errOps.size() < _clientRequest.sizeWriteOps()); + if (!_wcErrors.empty() && reportWCError) { WriteConcernErrorDetail* error = new WriteConcernErrorDetail; // Generate the multi-error message below diff --git a/src/mongo/s/write_ops/batch_write_op_test.cpp b/src/mongo/s/write_ops/batch_write_op_test.cpp index b21e64c6d53..bfda09f0814 100644 --- a/src/mongo/s/write_ops/batch_write_op_test.cpp +++ b/src/mongo/s/write_ops/batch_write_op_test.cpp @@ -32,7 +32,6 @@ #include "mongo/s/concurrency/locker_mongos_client_observer.h" #include "mongo/s/mock_ns_targeter.h" #include "mongo/s/session_catalog_router.h" -#include "mongo/s/shard_cannot_refresh_due_to_locks_held_exception.h" #include "mongo/s/sharding_router_test_fixture.h" #include "mongo/s/transaction_router.h" #include "mongo/s/write_ops/batch_write_op.h" @@ -235,6 +234,7 @@ TEST_F(BatchWriteOpTest, SingleWriteConcernErrorOrdered) { insertOp.setDocuments({BSON("x" << 1)}); return insertOp; }()); + request.setWriteConcern(BSON("w" << 3)); BatchWriteOp batchOp(_opCtx, request); @@ -246,6 +246,7 @@ TEST_F(BatchWriteOpTest, SingleWriteConcernErrorOrdered) { BatchedCommandRequest targetBatch = batchOp.buildBatchRequest(*targeted.begin()->second, targeter); + ASSERT(targetBatch.getWriteConcern().woCompare(request.getWriteConcern()) == 0); BatchedCommandResponse response; buildResponse(1, &response); @@ -304,22 +305,12 @@ TEST_F(BatchWriteOpTest, SingleStaleError) { batchOp.noteBatchResponse(*targeted.begin()->second, response, nullptr); ASSERT(!batchOp.isFinished()); - // Respond with a ShardCannotRefreshDueToLocksHeld error; the batch should still be retriable. - targeted.clear(); - ASSERT_OK(batchOp.targetBatch(targeter, false, &targeted)); - buildResponse(0, &response); - response.addToErrDetails(write_ops::WriteError( - 0, Status{ShardCannotRefreshDueToLocksHeldInfo(nss), "mock cache busy error"})); - - batchOp.noteBatchResponse(*targeted.begin()->second, response, nullptr); - ASSERT(!batchOp.isFinished()); - - // Respond with an 'ok' response targeted.clear(); ASSERT_OK(batchOp.targetBatch(targeter, false, &targeted)); buildResponse(1, &response); + // Respond with an 'ok' response batchOp.noteBatchResponse(*targeted.begin()->second, response, nullptr); ASSERT(batchOp.isFinished()); @@ -1010,7 +1001,9 @@ TEST_F(BatchWriteOpTest, MultiOpPartialSingleShardErrorOrdered) { // Tests of edge-case functionality, lifecycle is assumed to be behaving normally // -// Multi-op (unordered) error and write concern error test. +// Multi-op (unordered) error and write concern error test. We never report the write concern error +// for single-doc batches, since the error means there's no write concern applied. Don't suppress +// the error if ordered : false. TEST_F(BatchWriteOpTest, MultiOpErrorAndWriteConcernErrorUnordered) { NamespaceString nss("foo.bar"); ShardEndpoint endpoint(ShardId("shard"), ChunkVersion::IGNORED(), boost::none); @@ -1027,6 +1020,7 @@ TEST_F(BatchWriteOpTest, MultiOpErrorAndWriteConcernErrorUnordered) { insertOp.setDocuments({BSON("x" << 1), BSON("x" << 1)}); return insertOp; }()); + request.setWriteConcern(BSON("w" << 3)); BatchWriteOp batchOp(_opCtx, request); @@ -1051,7 +1045,8 @@ TEST_F(BatchWriteOpTest, MultiOpErrorAndWriteConcernErrorUnordered) { ASSERT(clientResponse.isWriteConcernErrorSet()); } -// Single-op (ordered) error and write concern error test. +// Single-op (ordered) error and write concern error test. Suppress the write concern error if +// ordered and we also have an error TEST_F(BatchWriteOpTest, SingleOpErrorAndWriteConcernErrorOrdered) { NamespaceString nss("foo.bar"); ShardEndpoint endpointA(ShardId("shardA"), ChunkVersion::IGNORED(), boost::none); @@ -1069,6 +1064,7 @@ TEST_F(BatchWriteOpTest, SingleOpErrorAndWriteConcernErrorOrdered) { updateOp.setUpdates({buildUpdate(BSON("x" << GTE << -1 << LT << 2), true)}); return updateOp; }()); + request.setWriteConcern(BSON("w" << 3)); BatchWriteOp batchOp(_opCtx, request); @@ -1095,14 +1091,14 @@ TEST_F(BatchWriteOpTest, SingleOpErrorAndWriteConcernErrorOrdered) { ASSERT(batchOp.isFinished()); ASSERT(++targetedIt == targeted.end()); - // Ordered reports write concern error. + // Ordered doesn't report write concern error BatchedCommandResponse clientResponse; batchOp.buildClientResponse(&clientResponse); ASSERT(clientResponse.getOk()); ASSERT_EQUALS(clientResponse.getN(), 1); ASSERT(clientResponse.isErrDetailsSet()); ASSERT_EQUALS(clientResponse.sizeErrDetails(), 1u); - ASSERT(clientResponse.isWriteConcernErrorSet()); + ASSERT(!clientResponse.isWriteConcernErrorSet()); } // Targeting failure on second op in batch op (ordered) @@ -1411,6 +1407,7 @@ TEST_F(BatchWriteOpTest, MultiOpTwoWCErrors) { insertOp.setDocuments({BSON("x" << -1), BSON("x" << 2)}); return insertOp; }()); + request.setWriteConcern(BSON("w" << 3)); BatchWriteOp batchOp(_opCtx, request); diff --git a/src/mongo/s/write_ops/batched_command_request.cpp b/src/mongo/s/write_ops/batched_command_request.cpp index 185857d6acc..107f1a49204 100644 --- a/src/mongo/s/write_ops/batched_command_request.cpp +++ b/src/mongo/s/write_ops/batched_command_request.cpp @@ -38,6 +38,8 @@ namespace mongo { namespace { +const auto kWriteConcern = "writeConcern"_sd; + template <class T> BatchedCommandRequest constructBatchedCommandRequest(const OpMsgRequest& request) { auto batchRequest = BatchedCommandRequest{T::parse(request)}; @@ -51,6 +53,11 @@ BatchedCommandRequest constructBatchedCommandRequest(const OpMsgRequest& request batchRequest.setShardVersion(shardVersion); } + auto writeConcernField = request.body[kWriteConcern]; + if (!writeConcernField.eoo()) { + batchRequest.setWriteConcern(writeConcernField.Obj()); + } + // The 'isTimeseriesNamespace' is an internal parameter used for communication between mongos // and mongod. auto isTimeseriesNamespace = @@ -166,9 +173,19 @@ const boost::optional<BSONObj>& BatchedCommandRequest::getLet() const { return _visit(Visitor{}); }; -const OptionalBool& BatchedCommandRequest::getBypassEmptyTsReplacement() const { - return _visit([](auto&& op) -> decltype(auto) { return op.getBypassEmptyTsReplacement(); }); -}; +bool BatchedCommandRequest::isVerboseWC() const { + if (!hasWriteConcern()) { + return true; + } + + BSONObj writeConcern = getWriteConcern(); + BSONElement wElem = writeConcern["w"]; + if (!wElem.isNumber() || wElem.Number() != 0) { + return true; + } + + return false; +} const write_ops::WriteCommandRequestBase& BatchedCommandRequest::getWriteCommandRequestBase() const { @@ -189,6 +206,10 @@ void BatchedCommandRequest::serialize(BSONObjBuilder* builder) const { if (_dbVersion) { builder->append("databaseVersion", _dbVersion->toBSON()); } + + if (_writeConcern) { + builder->append(kWriteConcern, *_writeConcern); + } } BSONObj BatchedCommandRequest::toBSON() const { diff --git a/src/mongo/s/write_ops/batched_command_request.h b/src/mongo/s/write_ops/batched_command_request.h index 0bcb51a3556..eea7f7bbe11 100644 --- a/src/mongo/s/write_ops/batched_command_request.h +++ b/src/mongo/s/write_ops/batched_command_request.h @@ -52,25 +52,15 @@ public: : _batchType(BatchType_Insert), _insertReq(std::make_unique<write_ops::InsertCommandRequest>(std::move(insertOp))) {} - BatchedCommandRequest(std::unique_ptr<write_ops::InsertCommandRequest> insertOp) - : _batchType(BatchType_Insert), _insertReq(std::move(insertOp)) {} - BatchedCommandRequest(write_ops::UpdateCommandRequest updateOp) : _batchType(BatchType_Update), _updateReq(std::make_unique<write_ops::UpdateCommandRequest>(std::move(updateOp))) {} - BatchedCommandRequest(std::unique_ptr<write_ops::UpdateCommandRequest> updateOp) - : _batchType(BatchType_Update), _updateReq(std::move(updateOp)) {} - BatchedCommandRequest(write_ops::DeleteCommandRequest deleteOp) : _batchType(BatchType_Delete), _deleteReq(std::make_unique<write_ops::DeleteCommandRequest>(std::move(deleteOp))) {} - BatchedCommandRequest(std::unique_ptr<write_ops::DeleteCommandRequest> deleteOp) - : _batchType(BatchType_Delete), _deleteReq(std::move(deleteOp)) {} - BatchedCommandRequest(BatchedCommandRequest&&) = default; - BatchedCommandRequest& operator=(BatchedCommandRequest&&) = default; static BatchedCommandRequest parseInsert(const OpMsgRequest& request); static BatchedCommandRequest parseUpdate(const OpMsgRequest& request); @@ -101,19 +91,26 @@ public: return *_deleteReq; } - std::unique_ptr<write_ops::InsertCommandRequest> extractInsertRequest() { - return std::move(_insertReq); + std::size_t sizeWriteOps() const; + + void setWriteConcern(const BSONObj& writeConcern) { + _writeConcern = writeConcern.getOwned(); } - std::unique_ptr<write_ops::UpdateCommandRequest> extractUpdateRequest() { - return std::move(_updateReq); + void unsetWriteConcern() { + _writeConcern = boost::none; } - std::unique_ptr<write_ops::DeleteCommandRequest> extractDeleteRequest() { - return std::move(_deleteReq); + bool hasWriteConcern() const { + return _writeConcern.is_initialized(); } - std::size_t sizeWriteOps() const; + const BSONObj& getWriteConcern() const { + invariant(_writeConcern); + return *_writeConcern; + } + + bool isVerboseWC() const; void setShardVersion(ChunkVersion shardVersion) { _shardVersion = std::move(shardVersion); @@ -149,7 +146,6 @@ public: const boost::optional<LegacyRuntimeConstants>& getLegacyRuntimeConstants() const; const boost::optional<BSONObj>& getLet() const; - const OptionalBool& getBypassEmptyTsReplacement() const; const write_ops::WriteCommandRequestBase& getWriteCommandRequestBase() const; void setWriteCommandRequestBase(write_ops::WriteCommandRequestBase writeCommandBase); @@ -232,6 +228,8 @@ private: boost::optional<ChunkVersion> _shardVersion; boost::optional<DatabaseVersion> _dbVersion; + + boost::optional<BSONObj> _writeConcern; }; /** diff --git a/src/mongo/s/write_ops/batched_command_request_test.cpp b/src/mongo/s/write_ops/batched_command_request_test.cpp index be0728b1533..9a5e968f10d 100644 --- a/src/mongo/s/write_ops/batched_command_request_test.cpp +++ b/src/mongo/s/write_ops/batched_command_request_test.cpp @@ -92,12 +92,14 @@ TEST(BatchedCommandRequest, InsertCloneWithIds) { insertOp.setDocuments({BSON("x" << 1), BSON("x" << 2)}); return insertOp; }()); + batchedRequest.setWriteConcern(BSON("w" << 2)); const auto clonedRequest(BatchedCommandRequest::cloneInsertWithIds(std::move(batchedRequest))); ASSERT_EQ("xyz.abc", clonedRequest.getNS().ns()); ASSERT(clonedRequest.getWriteCommandRequestBase().getOrdered()); ASSERT(clonedRequest.getWriteCommandRequestBase().getBypassDocumentValidation()); + ASSERT_BSONOBJ_EQ(BSON("w" << 2), clonedRequest.getWriteConcern()); const auto& insertDocs = clonedRequest.getInsertRequest().getDocuments(); ASSERT_EQ(2u, insertDocs.size()); diff --git a/src/mongo/s/write_ops/write_op.cpp b/src/mongo/s/write_ops/write_op.cpp index 6b33d3b9312..236c7efbc94 100644 --- a/src/mongo/s/write_ops/write_op.cpp +++ b/src/mongo/s/write_ops/write_op.cpp @@ -29,24 +29,7 @@ #include "mongo/s/write_ops/write_op.h" - -#include <absl/container/flat_hash_set.h> -#include <algorithm> -#include <boost/move/utility_core.hpp> -#include <boost/none.hpp> -#include <boost/optional/optional.hpp> -#include <ostream> -#include <string> - -#include "mongo/base/error_codes.h" -#include "mongo/base/status.h" -#include "mongo/bson/bsonobjbuilder.h" -#include "mongo/db/catalog/collection_uuid_mismatch_info.h" -#include "mongo/db/stats/counters.h" -#include "mongo/s/sharding_feature_flags_gen.h" #include "mongo/s/transaction_router.h" -#include "mongo/s/write_ops/batch_write_op.h" -#include "mongo/s/write_ops/batched_command_request.h" #include "mongo/util/assert_util.h" namespace mongo { @@ -96,16 +79,6 @@ write_ops::WriteError combineOpErrors(const std::vector<ChildWriteOp const*>& er Status(MultipleErrorsOccurredInfo(errB.arr()), msg.str())); } -bool isSafeToIgnoreErrorInPartiallyAppliedOp(write_ops::WriteError& error) { - // UUID mismatch errors are safe to ignore if the actualCollection is null in conjuntion with - // other successful operations. This is true because it means we wrongly targeted a non-owning - // shard with the operation and we wouldn't have applied any modifications anyway. - // - // Note this is only safe if we're using ShardVersion::IGNORED since we're ignoring any - // placement concern and broadcasting to all shards. - return error.getStatus().code() == ErrorCodes::CollectionUUIDMismatch && - !error.getStatus().extraInfo<CollectionUUIDMismatchInfo>()->actualCollection(); -} } // namespace const BatchItemRef& WriteOp::getWriteItem() const { @@ -209,29 +182,7 @@ void WriteOp::_updateOpState() { _state = WriteOpState_Ready; } else if (!childErrors.empty()) { _error = combineOpErrors(childErrors); - bool isTargetingAllShardsWithSVIgnored = - childErrors.front() - ->endpoint->shardVersion - .map([&](const auto& cv) { return ChunkVersion::isIgnoredVersion(cv); }) - .get_value_or(false); - // There are errors that are safe to ignore if they were correctly applied to other shards - // and we're using ShardVersion::IGNORED. They are safe to ignore as they can be interpreted - // as no-ops if the shard response had been instead a successful result since they wouldn't - // have modified any data. As a result, we can swallow the errors and treat them as a - // successful operation. - if (isTargetingAllShardsWithSVIgnored && isSafeToIgnoreErrorInPartiallyAppliedOp(*_error) && - !_successfulShardSet.empty()) { - if (!hasPendingChild) { - _error.reset(); - _state = WriteOpState_Completed; - } else { - // As this error is acceptable we wait until all other operations finish to take a - // decision. - return; - } - } else { - _state = WriteOpState_Error; - } + _state = WriteOpState_Error; } else if (hasPendingChild && _inTxn) { // Return early here since this means that there were no errors while in txn // but there are still ops that have not yet finished. |
