diff options
Diffstat (limited to 'src/mongo/db/repl')
84 files changed, 1879 insertions, 504 deletions
diff --git a/src/mongo/db/repl/README.md b/src/mongo/db/repl/README.md index c0526e9d966..08af0e4073a 100644 --- a/src/mongo/db/repl/README.md +++ b/src/mongo/db/repl/README.md @@ -361,6 +361,8 @@ If the server does not support `hello`, the `helloOk` flag is ignored. A new dri not see "helloOk: true" in the response and continue to send `isMaster` on this connection. Old drivers will not specify this flag at all, so the behavior remains the same. +Communication between nodes in the cluster is always done using `hello`, never with `isMaster`. + ## Communication Each node has a copy of the **`ReplicaSetConfig`** in the `ReplicationCoordinator` that lists all @@ -508,7 +510,7 @@ assigns itself a priority takeover timeout proportional to its rank. After that node will check if it's eligible to run for election and if so will begin an election. The timeout is simply: `(election timeout) * (priority rank + 1)`. -Heartbeat threads belong to the +Heartbeat threads belong to the [`ReplCoordThreadPool`](https://github.com/mongodb/mongo/blob/674d57fc70d80dedbfd634ce00ca4b967ea89646/src/mongo/db/mongod_main.cpp#L944) connection pool started by the [`ReplicationCoordinator`](https://github.com/mongodb/mongo/blob/674d57fc70d80dedbfd634ce00ca4b967ea89646/src/mongo/db/mongod_main.cpp#L986). @@ -1640,15 +1642,15 @@ The `initialSyncTransientErrorRetryPeriodSeconds` is also used to control retrie fetcher and all network operations in initial sync which take place after the data cloning has started. -As of v4.4, initial syncing a node with [two-phase index builds](https://github.com/mongodb/mongo/blob/0a7641e69031fcfdf25a1780a3b62bca5f59d68f/src/mongo/db/catalog/README.md#replica-set-index-builds) -will immediately build all ready indexes from the sync source and setup the index builder threads -for any unfinished index builds. -[See here](https://github.com/mongodb/mongo/blob/85d75907fd12c2360cf16b97f941386f343ca6fc/src/mongo/db/repl/collection_cloner.cpp#L247-L301). +As of v4.4, initial syncing a node with [two-phase index builds](https://github.com/mongodb/mongo/blob/0a7641e69031fcfdf25a1780a3b62bca5f59d68f/src/mongo/db/catalog/README.md#replica-set-index-builds) +will immediately build all ready indexes from the sync source and setup the index builder threads +for any unfinished index builds. +[See here](https://github.com/mongodb/mongo/blob/85d75907fd12c2360cf16b97f941386f343ca6fc/src/mongo/db/repl/collection_cloner.cpp#L247-L301). -This is necessary to avoid a scenario where the primary node cannot satisfy the index builds commit -quorum if it depends on the initial syncing nodes vote. Prior to this, initial syncing nodes would -start the index build when they came across the `commitIndexBuild` oplog entry, which is only -observable once the index builds commit quorum has been satisfied. +This is necessary to avoid a scenario where the primary node cannot satisfy the index builds commit +quorum if it depends on the initial syncing nodes vote. Prior to this, initial syncing nodes would +start the index build when they came across the `commitIndexBuild` oplog entry, which is only +observable once the index builds commit quorum has been satisfied. [See this test for an example](https://github.com/mongodb/mongo/blob/f495bdead326a06a76f8a980e44092deb096a21d/jstests/noPassthrough/commit_quorum_does_not_hang_with_initial_sync.js). ## Oplog application phase @@ -1897,7 +1899,7 @@ Once the files are moved, we switch storage one more time, back to the original now has the downloaded files. We use the `InitialSyncFileMover` to delete the move marker and the entire .initialsync directory; a restart of the server at or after this point will not involve any more FCBIS work. Then we release the global lock, and retrieve the last applied -OpTime and WallTime from the top of the oplog (the latest entry in the oplog) in +OpTime and WallTime from the top of the oplog (the latest entry in the oplog) in `_updateLastAppliedOptime`. The initial sync attempt is now considered successful, and we call `_finishCallback`. This acts @@ -2208,7 +2210,7 @@ value has yet to be loaded from disk, the FCV is set to `kUnsetDefaultLastLTSBeh indicates that the server will be using the last-LTS feature set as to ensure compatibility with other nodes in the replica set. -As part of initial sync, the in-memory FCV value is always initially set to be +As part of initial sync, the in-memory FCV value is always initially set to be `kUnsetDefaultLastLTSBehavior`. This is to ensure compatibility between the sync source and sync target. If the sync source is actually in a different feature compatibility version, we will find out when we clone the `admin.system.version` collection. @@ -2219,7 +2221,7 @@ if it has not yet received the `replSetInitiate` command. ## setFeatureCompatibilityVersion The FCV can be set using the `setFeatureCompatibilityVersion` admin command to one of the following: -* The version of the last-LTS (Long Term Support) +* The version of the last-LTS (Long Term Support) * Indicates to the server to use the feature set compatible with the last LTS release version. * The version of the last-continuous release * Indicates to the server to use the feature set compatible with the last continuous release @@ -2263,8 +2265,8 @@ kUpgradingFrom_5_0_To_5_1: } kDowngradingFrom_5_1_To_5_0: -{ - version: 5.0, +{ + version: 5.0, targetVersion: 5.0, previousVersion: 5.1 } @@ -2327,11 +2329,11 @@ transaction. For a prepared transaction, we have the following guarantee: `prepa **`currentCommittedSnapshot`**: An optime maintained in `ReplicationCoordinator` that is used to serve majority reads and is always guaranteed to be <= `lastCommittedOpTime`. When `eMRC=true`, this -is currently [set to the stable optime](https://github.com/mongodb/mongo/blob/00fbc981646d9e6ebc391f45a31f4070d4466753/src/mongo/db/repl/replication_coordinator_impl.cpp#L4945). +is currently [set to the stable optime](https://github.com/mongodb/mongo/blob/00fbc981646d9e6ebc391f45a31f4070d4466753/src/mongo/db/repl/replication_coordinator_impl.cpp#L4945). Since it is reset every time we recalculate the stable optime, it will also be up to date. -When `eMRC=false`, this [is set](https://github.com/mongodb/mongo/blob/00fbc981646d9e6ebc391f45a31f4070d4466753/src/mongo/db/repl/replication_coordinator_impl.cpp#L4952-L4961) -to the minimum of the stable optime and the `lastCommittedOpTime`, even though it is not used to +When `eMRC=false`, this [is set](https://github.com/mongodb/mongo/blob/00fbc981646d9e6ebc391f45a31f4070d4466753/src/mongo/db/repl/replication_coordinator_impl.cpp#L4952-L4961) +to the minimum of the stable optime and the `lastCommittedOpTime`, even though it is not used to serve majority reads in that case. **`initialDataTimestamp`**: A timestamp used to indicate the timestamp at which history “begins”. @@ -2386,7 +2388,7 @@ populated internally from the `currentCommittedSnapshot` timestamp inside `Repli **`stable_timestamp`**: The newest timestamp at which the storage engine is allowed to take a checkpoint, which can be thought of as a consistent snapshot of the data. Replication informs the storage engine of where it is safe to take its next checkpoint. This timestamp is guaranteed to be -majority committed so that RTT rollback can use it. In the case when +majority committed so that RTT rollback can use it. In the case when [`eMRC=false`](#enableMajorityReadConcern-flag), the stable timestamp may not be majority committed, which is why we must use the Rollback via Refetch rollback algorithm. diff --git a/src/mongo/db/repl/SConscript b/src/mongo/db/repl/SConscript index 10783f244cd..21fecddeaf8 100644 --- a/src/mongo/db/repl/SConscript +++ b/src/mongo/db/repl/SConscript @@ -60,6 +60,7 @@ env.Library( '$BUILD_DIR/mongo/db/catalog/local_oplog_info', '$BUILD_DIR/mongo/db/catalog/multi_index_block', '$BUILD_DIR/mongo/db/commands/feature_compatibility_parsers', + '$BUILD_DIR/mongo/db/concurrency/exception_util', '$BUILD_DIR/mongo/db/db_raii', '$BUILD_DIR/mongo/db/dbdirectclient', '$BUILD_DIR/mongo/db/dbhelpers', @@ -117,7 +118,7 @@ env.Library( '$BUILD_DIR/mongo/db/concurrency/lock_manager', '$BUILD_DIR/mongo/db/db_raii', '$BUILD_DIR/mongo/db/dbhelpers', - '$BUILD_DIR/mongo/db/index/index_descriptor', + '$BUILD_DIR/mongo/db/index/index_access_method', '$BUILD_DIR/mongo/db/query_exec', '$BUILD_DIR/mongo/util/fail_point', 'oplog', @@ -137,7 +138,6 @@ env.Library( LIBDEPS=[ '$BUILD_DIR/mongo/client/connection_pool', '$BUILD_DIR/mongo/client/fetcher', - '$BUILD_DIR/mongo/db/concurrency/write_conflict_exception', '$BUILD_DIR/mongo/db/index_builds_coordinator_interface', '$BUILD_DIR/mongo/db/service_context', '$BUILD_DIR/mongo/util/concurrency/thread_pool', @@ -155,6 +155,7 @@ env.Library( 'sync_source_resolver', ], LIBDEPS_PRIVATE=[ + '$BUILD_DIR/mongo/db/concurrency/exception_util', '$BUILD_DIR/mongo/db/op_observer', '$BUILD_DIR/mongo/executor/thread_pool_task_executor', 'repl_server_parameters', @@ -246,6 +247,7 @@ env.Library( '$BUILD_DIR/mongo/db/catalog/database_holder', '$BUILD_DIR/mongo/db/catalog/multi_index_block', '$BUILD_DIR/mongo/db/common', + '$BUILD_DIR/mongo/db/concurrency/exception_util', '$BUILD_DIR/mongo/db/dbhelpers', '$BUILD_DIR/mongo/db/index_builds_coordinator_interface', '$BUILD_DIR/mongo/db/multitenancy', @@ -587,7 +589,6 @@ env.Library( LIBDEPS=[ '$BUILD_DIR/mongo/db/auth/authorization_manager_global', '$BUILD_DIR/mongo/db/concurrency/lock_manager', - '$BUILD_DIR/mongo/db/concurrency/write_conflict_exception', '$BUILD_DIR/mongo/db/curop', '$BUILD_DIR/mongo/db/query_exec', '$BUILD_DIR/mongo/db/stats/timer_stats', @@ -605,6 +606,8 @@ env.Library( ], LIBDEPS_PRIVATE=[ '$BUILD_DIR/mongo/db/commands/mongod_fsync', + '$BUILD_DIR/mongo/db/concurrency/exception_util', + '$BUILD_DIR/mongo/db/curop_metrics', '$BUILD_DIR/mongo/db/storage/storage_control', 'repl_server_parameters', 'replication_auth', @@ -701,7 +704,7 @@ env.Library( '$BUILD_DIR/mongo/db/commands/mongod_fcv', '$BUILD_DIR/mongo/db/common', '$BUILD_DIR/mongo/db/concurrency/lock_manager', - '$BUILD_DIR/mongo/db/index/index_descriptor', + '$BUILD_DIR/mongo/db/index/index_access_method', '$BUILD_DIR/mongo/db/kill_sessions_local', '$BUILD_DIR/mongo/db/mongod_options', '$BUILD_DIR/mongo/db/prepare_conflict_tracker', @@ -731,6 +734,7 @@ env.Library( '$BUILD_DIR/mongo/db/index_builds_coordinator_interface', '$BUILD_DIR/mongo/db/storage/journal_flusher', '$BUILD_DIR/mongo/idl/server_parameter', + 'delayable_timeout_callback', 'repl_server_parameters', 'replica_set_aware_service', 'split_horizon', @@ -1049,6 +1053,7 @@ env.Library( '$BUILD_DIR/mongo/db/catalog/document_validation', '$BUILD_DIR/mongo/db/commands/list_collections_filter', '$BUILD_DIR/mongo/db/ops/write_ops_exec', + '$BUILD_DIR/mongo/idl/cluster_parameter_synchronization_helpers', '$BUILD_DIR/mongo/rpc/metadata', '$BUILD_DIR/mongo/util/progress_meter', 'oplog', @@ -1341,6 +1346,7 @@ env.Library( "tenant_migration_recipient_entry_helpers.cpp", ], LIBDEPS=[ + '$BUILD_DIR/mongo/db/keys_collection_util', '$BUILD_DIR/mongo/util/future_util', 'repl_coordinator_interface', 'repl_server_parameters', @@ -1349,6 +1355,7 @@ env.Library( LIBDEPS_PRIVATE=[ "$BUILD_DIR/mongo/base", "$BUILD_DIR/mongo/db/catalog_raii", + "$BUILD_DIR/mongo/db/concurrency/exception_util", "$BUILD_DIR/mongo/db/dbhelpers", "$BUILD_DIR/mongo/db/namespace_string", '$BUILD_DIR/mongo/db/service_context', @@ -1378,12 +1385,14 @@ env.Library( ], LIBDEPS_PRIVATE=[ '$BUILD_DIR/mongo/client/clientdriver_network', + '$BUILD_DIR/mongo/db/concurrency/exception_util', '$BUILD_DIR/mongo/db/multitenancy', '$BUILD_DIR/mongo/db/ops/write_ops_exec', '$BUILD_DIR/mongo/db/pipeline/process_interface/mongo_process_interface', '$BUILD_DIR/mongo/db/storage/wiredtiger/storage_wiredtiger_import', '$BUILD_DIR/mongo/db/transaction', '$BUILD_DIR/mongo/executor/scoped_task_executor', + '$BUILD_DIR/mongo/idl/cluster_parameter_synchronization_helpers', 'cloner_utils', 'oplog', 'oplog_application_interface', @@ -1429,6 +1438,7 @@ env.Library( ], LIBDEPS_PRIVATE=[ '$BUILD_DIR/mongo/db/catalog/local_oplog_info', + '$BUILD_DIR/mongo/db/concurrency/exception_util', ], ) @@ -1447,6 +1457,7 @@ env.Library( ], LIBDEPS_PRIVATE=[ "$BUILD_DIR/mongo/db/catalog/local_oplog_info", + "$BUILD_DIR/mongo/db/concurrency/exception_util", "$BUILD_DIR/mongo/db/index_builds_coordinator_interface", ], ) @@ -1465,7 +1476,6 @@ env.Library( '$BUILD_DIR/mongo/db/cloner', '$BUILD_DIR/mongo/db/concurrency/lock_manager', '$BUILD_DIR/mongo/db/curop', - '$BUILD_DIR/mongo/db/free_mon/free_mon_mongod', '$BUILD_DIR/mongo/db/kill_sessions_local', '$BUILD_DIR/mongo/db/logical_time', '$BUILD_DIR/mongo/db/not_primary_error_tracker', @@ -1612,6 +1622,7 @@ if wiredtiger: 'abstract_async_component_test.cpp', 'apply_ops_test.cpp', 'check_quorum_for_config_change_test.cpp', + 'delayable_timeout_callback_test.cpp', 'drop_pending_collection_reaper_test.cpp', 'idempotency_document_structure_test.cpp', 'idempotency_update_sequence_test.cpp', @@ -1672,7 +1683,7 @@ if wiredtiger: '$BUILD_DIR/mongo/db/commands/mongod_fcv', '$BUILD_DIR/mongo/db/commands/txn_cmd_request', '$BUILD_DIR/mongo/db/dbdirectclient', - '$BUILD_DIR/mongo/db/index/index_access_methods', + '$BUILD_DIR/mongo/db/index/index_access_method', '$BUILD_DIR/mongo/db/index_build_entry_helpers', '$BUILD_DIR/mongo/db/index_builds_coordinator_mongod', '$BUILD_DIR/mongo/db/logical_session_id_helpers', @@ -1701,6 +1712,7 @@ if wiredtiger: '$BUILD_DIR/mongo/util/concurrency/thread_pool', 'abstract_async_component', 'data_replicator_external_state_mock', + 'delayable_timeout_callback', 'drop_pending_collection_reaper', 'idempotency_test_fixture', 'idempotency_test_util', @@ -1955,7 +1967,7 @@ env.Library( ], LIBDEPS_PRIVATE=[ '$BUILD_DIR/mongo/base', - '$BUILD_DIR/mongo/db/concurrency/write_conflict_exception', + '$BUILD_DIR/mongo/db/concurrency/exception_util', 'abstract_async_component', 'cloner_utils', 'oplog', @@ -1977,3 +1989,14 @@ env.Library( 'oplog_entry', ], ) + +env.Library( + target='delayable_timeout_callback', + source=[ + 'delayable_timeout_callback.cpp', + ], + LIBDEPS=[ + '$BUILD_DIR/mongo/base', + '$BUILD_DIR/mongo/executor/task_executor_interface', + ], +) diff --git a/src/mongo/db/repl/all_database_cloner.cpp b/src/mongo/db/repl/all_database_cloner.cpp index b5856b7aef0..6f6d73cefec 100644 --- a/src/mongo/db/repl/all_database_cloner.cpp +++ b/src/mongo/db/repl/all_database_cloner.cpp @@ -59,16 +59,17 @@ BaseCloner::ClonerStages AllDatabaseCloner::getStages() { } Status AllDatabaseCloner::ensurePrimaryOrSecondary( - const executor::RemoteCommandResponse& isMasterReply) { - if (!isMasterReply.isOK()) { - LOGV2(21054, "Cannot reconnect because isMaster command failed"); - return isMasterReply.status; + const executor::RemoteCommandResponse& helloReply) { + if (!helloReply.isOK()) { + LOGV2(21054, "Cannot reconnect because 'hello' command failed"); + return helloReply.status; } - if (isMasterReply.data["ismaster"].trueValue() || isMasterReply.data["secondary"].trueValue()) + if (helloReply.data["isWritablePrimary"].trueValue() || + helloReply.data["secondary"].trueValue()) return Status::OK(); // There is a window during startup where a node has an invalid configuration and will have - // an isMaster response the same as a removed node. So we must check to see if the node is + // an "hello" response the same as a removed node. So we must check to see if the node is // removed by checking local configuration. auto memberData = ReplicationCoordinator::get(getGlobalServiceContext())->getMemberData(); auto syncSourceIter = std::find_if( @@ -109,8 +110,8 @@ BaseCloner::AfterStageBehavior AllDatabaseCloner::connectStage() { // handle the reconnect itself. This is necessary to get correct backoff behavior. if (client->getServerHostAndPort() != getSource()) { client->setHandshakeValidationHook( - [this](const executor::RemoteCommandResponse& isMasterReply) { - return ensurePrimaryOrSecondary(isMasterReply); + [this](const executor::RemoteCommandResponse& helloReply) { + return ensurePrimaryOrSecondary(helloReply); }); uassertStatusOK(client->connect(getSource(), StringData(), boost::none)); } else { diff --git a/src/mongo/db/repl/all_database_cloner.h b/src/mongo/db/repl/all_database_cloner.h index d538af3869c..e61c09030b7 100644 --- a/src/mongo/db/repl/all_database_cloner.h +++ b/src/mongo/db/repl/all_database_cloner.h @@ -82,7 +82,7 @@ private: * Validation function to ensure we connect only to primary or secondary nodes. * * Because the cloner connection is separate from the usual inter-node connection pool and - * did not have the 'hangUpOnStepDown:false' flag set in the initial isMaster request, we + * did not have the 'hangUpOnStepDown:false' flag set in the initial "hello" request, we * will always disconnect if the sync source transitions to a state other than PRIMARY * or SECONDARY. It will not disconnect on a PRIMARY to SECONDARY or SECONDARY to PRIMARY * transition because we no longer do that (the flag name is anachronistic). After @@ -99,7 +99,7 @@ private: * would succeed and we would have an inconsistent node. If other data was added we would * invariant during oplog application with a NamespaceNotFound error. */ - Status ensurePrimaryOrSecondary(const executor::RemoteCommandResponse& isMasterReply); + Status ensurePrimaryOrSecondary(const executor::RemoteCommandResponse& helloReply); /** * Stage function that makes a connection to the sync source. diff --git a/src/mongo/db/repl/apply_ops.cpp b/src/mongo/db/repl/apply_ops.cpp index 5f71c23b3ab..e9ae0d4858c 100644 --- a/src/mongo/db/repl/apply_ops.cpp +++ b/src/mongo/db/repl/apply_ops.cpp @@ -39,8 +39,8 @@ #include "mongo/db/catalog/database_holder.h" #include "mongo/db/catalog/document_validation.h" #include "mongo/db/client.h" +#include "mongo/db/concurrency/exception_util.h" #include "mongo/db/concurrency/lock_state.h" -#include "mongo/db/concurrency/write_conflict_exception.h" #include "mongo/db/curop.h" #include "mongo/db/db_raii.h" #include "mongo/db/dbdirectclient.h" diff --git a/src/mongo/db/repl/bgsync.cpp b/src/mongo/db/repl/bgsync.cpp index 8217e38090a..17767d6ca09 100644 --- a/src/mongo/db/repl/bgsync.cpp +++ b/src/mongo/db/repl/bgsync.cpp @@ -41,8 +41,8 @@ #include "mongo/client/connection_pool.h" #include "mongo/db/auth/authorization_session.h" #include "mongo/db/client.h" +#include "mongo/db/concurrency/exception_util.h" #include "mongo/db/concurrency/replication_state_transition_lock_guard.h" -#include "mongo/db/concurrency/write_conflict_exception.h" #include "mongo/db/dbhelpers.h" #include "mongo/db/index_builds_coordinator.h" #include "mongo/db/repl/data_replicator_external_state_impl.h" diff --git a/src/mongo/db/repl/collection_bulk_loader_impl.cpp b/src/mongo/db/repl/collection_bulk_loader_impl.cpp index c0b3365621e..3e190cb7f6f 100644 --- a/src/mongo/db/repl/collection_bulk_loader_impl.cpp +++ b/src/mongo/db/repl/collection_bulk_loader_impl.cpp @@ -39,7 +39,7 @@ #include "mongo/db/catalog/index_catalog.h" #include "mongo/db/client.h" #include "mongo/db/concurrency/d_concurrency.h" -#include "mongo/db/concurrency/write_conflict_exception.h" +#include "mongo/db/concurrency/exception_util.h" #include "mongo/db/curop.h" #include "mongo/db/jsobj.h" #include "mongo/db/operation_context.h" diff --git a/src/mongo/db/repl/collection_cloner.cpp b/src/mongo/db/repl/collection_cloner.cpp index 0066240cc2f..fe1cc4b34a2 100644 --- a/src/mongo/db/repl/collection_cloner.cpp +++ b/src/mongo/db/repl/collection_cloner.cpp @@ -29,6 +29,8 @@ #define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kReplicationInitialSync +#include "mongo/db/index/index_descriptor_fwd.h" +#include "mongo/db/service_context.h" #include "mongo/platform/basic.h" #include "mongo/base/string_data.h" @@ -206,8 +208,20 @@ BaseCloner::AfterStageBehavior CollectionCloner::listIndexesStage() { "source"_attr = getSource()); } + const auto storageEngine = getGlobalServiceContext()->getStorageEngine(); // Parse the index specs into their respective state, ready or unfinished. for (auto&& spec : indexSpecs) { + // Sanitize storage engine options to remove options which might not apply to this node. See + // SERVER-68122. + if (auto storageEngineElem = spec.getField(IndexDescriptor::kStorageEngineFieldName)) { + auto sanitizedStorageEngineOpts = + storageEngine->getSanitizedStorageOptionsForSecondaryReplication( + storageEngineElem.embeddedObject()); + fassert(6812200, sanitizedStorageEngineOpts); + spec = spec.addFields(BSON(IndexDescriptor::kStorageEngineFieldName + << sanitizedStorageEngineOpts.getValue())); + } + if (spec.hasField("clustered")) { invariant(_collectionOptions.clusteredIndex); invariant(spec.getBoolField("clustered") == true); diff --git a/src/mongo/db/repl/data_replicator_external_state.h b/src/mongo/db/repl/data_replicator_external_state.h index 87826b0f199..d5be160d5cf 100644 --- a/src/mongo/db/repl/data_replicator_external_state.h +++ b/src/mongo/db/repl/data_replicator_external_state.h @@ -30,6 +30,7 @@ #pragma once #include "mongo/base/status_with.h" +#include "mongo/db/repl/last_vote.h" #include "mongo/db/repl/multiapplier.h" #include "mongo/db/repl/oplog_applier.h" #include "mongo/db/repl/oplog_buffer.h" @@ -144,6 +145,12 @@ public: virtual Status storeLocalConfigDocument(OperationContext* opCtx, const BSONObj& config) = 0; /** + * Returns the current stored replica set "last vote" if there is one, or an error why there + * isn't. + */ + virtual StatusWith<LastVote> loadLocalLastVoteDocument(OperationContext* opCtx) const = 0; + + /** * Returns the replication journal listener. */ virtual JournalListener* getReplicationJournalListener() = 0; diff --git a/src/mongo/db/repl/data_replicator_external_state_impl.cpp b/src/mongo/db/repl/data_replicator_external_state_impl.cpp index 00c924ff1ea..9bd60084aed 100644 --- a/src/mongo/db/repl/data_replicator_external_state_impl.cpp +++ b/src/mongo/db/repl/data_replicator_external_state_impl.cpp @@ -175,6 +175,11 @@ Status DataReplicatorExternalStateImpl::storeLocalConfigDocument(OperationContex opCtx, config, false /* write oplog entry */); } +StatusWith<LastVote> DataReplicatorExternalStateImpl::loadLocalLastVoteDocument( + OperationContext* opCtx) const { + return _replicationCoordinatorExternalState->loadLocalLastVoteDocument(opCtx); +} + JournalListener* DataReplicatorExternalStateImpl::getReplicationJournalListener() { return _replicationCoordinatorExternalState->getReplicationJournalListener(); } diff --git a/src/mongo/db/repl/data_replicator_external_state_impl.h b/src/mongo/db/repl/data_replicator_external_state_impl.h index c408c484dc9..9cd2364927e 100644 --- a/src/mongo/db/repl/data_replicator_external_state_impl.h +++ b/src/mongo/db/repl/data_replicator_external_state_impl.h @@ -87,6 +87,8 @@ public: Status storeLocalConfigDocument(OperationContext* opCtx, const BSONObj& config) override; + StatusWith<LastVote> loadLocalLastVoteDocument(OperationContext* opCtx) const override; + JournalListener* getReplicationJournalListener() override; protected: diff --git a/src/mongo/db/repl/data_replicator_external_state_mock.cpp b/src/mongo/db/repl/data_replicator_external_state_mock.cpp index ddcfc701ca6..617f4f24098 100644 --- a/src/mongo/db/repl/data_replicator_external_state_mock.cpp +++ b/src/mongo/db/repl/data_replicator_external_state_mock.cpp @@ -147,5 +147,10 @@ JournalListener* DataReplicatorExternalStateMock::getReplicationJournalListener( return nullptr; } +StatusWith<LastVote> DataReplicatorExternalStateMock::loadLocalLastVoteDocument( + OperationContext* opCtx) const { + return StatusWith<LastVote>(ErrorCodes::NoMatchingDocument, "mock"); +} + } // namespace repl } // namespace mongo diff --git a/src/mongo/db/repl/data_replicator_external_state_mock.h b/src/mongo/db/repl/data_replicator_external_state_mock.h index 535ee513102..beb7ecdc28d 100644 --- a/src/mongo/db/repl/data_replicator_external_state_mock.h +++ b/src/mongo/db/repl/data_replicator_external_state_mock.h @@ -77,6 +77,8 @@ public: Status storeLocalConfigDocument(OperationContext* opCtx, const BSONObj& config) override; + StatusWith<LastVote> loadLocalLastVoteDocument(OperationContext* opCtx) const override; + JournalListener* getReplicationJournalListener() override; // Task executor. diff --git a/src/mongo/db/repl/database_cloner.cpp b/src/mongo/db/repl/database_cloner.cpp index adeeeb0afb6..23da9db918a 100644 --- a/src/mongo/db/repl/database_cloner.cpp +++ b/src/mongo/db/repl/database_cloner.cpp @@ -29,6 +29,7 @@ #define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kReplicationInitialSync +#include "mongo/db/service_context.h" #include "mongo/platform/basic.h" #include "mongo/base/string_data.h" @@ -69,6 +70,7 @@ BaseCloner::AfterStageBehavior DatabaseCloner::listCollectionsStage() { BSONObj res; auto collectionInfos = getClient()->getCollectionInfos(_dbName, ListCollectionsFilter::makeTypeCollectionFilter()); + const auto storageEngine = getGlobalServiceContext()->getStorageEngine(); stdx::unordered_set<std::string> seen; for (auto&& info : collectionInfos) { @@ -104,6 +106,13 @@ BaseCloner::AfterStageBehavior DatabaseCloner::listCollectionsStage() { << "'" << result.getName() << "': " << info, isDuplicate); + // Sanitize storage engine options to remove options which might not apply to this node. See + // SERVER-68122. + auto sanitizedStorageOptions = + uassertStatusOK(storageEngine->getSanitizedStorageOptionsForSecondaryReplication( + result.getOptions().storageEngine)); + result.getOptions().storageEngine = sanitizedStorageOptions; + // While UUID is a member of CollectionOptions, listCollections does not return the // collectionUUID there as part of the options, but instead places it in the 'info' field. // We need to move it back to CollectionOptions to create the collection properly. diff --git a/src/mongo/db/repl/dbcheck.cpp b/src/mongo/db/repl/dbcheck.cpp index 6fb94689544..25651574898 100644 --- a/src/mongo/db/repl/dbcheck.cpp +++ b/src/mongo/db/repl/dbcheck.cpp @@ -35,7 +35,6 @@ #include "mongo/db/catalog/database_holder.h" #include "mongo/db/catalog/health_log.h" #include "mongo/db/catalog/index_catalog.h" -#include "mongo/db/concurrency/write_conflict_exception.h" #include "mongo/db/db_raii.h" #include "mongo/db/operation_context.h" #include "mongo/db/query/internal_plans.h" diff --git a/src/mongo/db/repl/delayable_timeout_callback.cpp b/src/mongo/db/repl/delayable_timeout_callback.cpp new file mode 100644 index 00000000000..d84d7099779 --- /dev/null +++ b/src/mongo/db/repl/delayable_timeout_callback.cpp @@ -0,0 +1,192 @@ +/** + * Copyright (C) 2022-present MongoDB, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * <http://www.mongodb.com/licensing/server-side-public-license>. + * + * As a special exception, the copyright holders give permission to link the + * code of portions of this program with the OpenSSL library under certain + * conditions as described in each individual source file and distribute + * linked combinations including the program with the OpenSSL library. You + * must comply with the Server Side Public License in all respects for + * all of the code used other than as permitted herein. If you modify file(s) + * with this exception, you may extend this exception to your version of the + * file(s), but you are not obligated to do so. If you do not wish to do so, + * delete this exception statement from your version. If you delete this + * exception statement from all source files in the program, then also delete + * it in the license file. + */ + +#define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kReplication + +#include "mongo/db/repl/delayable_timeout_callback.h" +#include "mongo/logv2/log.h" + +namespace mongo { +namespace repl { + +DelayableTimeoutCallback::~DelayableTimeoutCallback() { + cancel(); +} + +void DelayableTimeoutCallback::cancel() { + stdx::lock_guard lk(_mutex); + _cancel(lk); +} + +void DelayableTimeoutCallback::_cancel(WithLock) { + if (_cbHandle) { + _executor->cancel(_cbHandle); + _cbHandle = executor::TaskExecutor::CallbackHandle(); + _nextCall = Date_t(); + } + invariant(_nextCall == Date_t()); +} + +Date_t DelayableTimeoutCallback::getNextCall() const { + stdx::lock_guard lk(_mutex); + return _nextCall; +} + +bool DelayableTimeoutCallback::isActive() const { + return getNextCall() != Date_t(); +} + +Status DelayableTimeoutCallback::scheduleAt(Date_t when) { + stdx::lock_guard lk(_mutex); + return _scheduleAt(lk, when); +} + +Status DelayableTimeoutCallback::_scheduleAt(WithLock lk, Date_t when) { + if (_cbHandle && when < _nextCall) { + LOGV2_DEBUG(6602300, + 3, + "Moving a delayable timeout call backwards, which is inefficient", + "timerName"_attr = _timerName, + "when"_attr = when, + "nextCall"_attr = _nextCall); + _cancel(lk); + } + return _delayUntil(lk, when); +} + +Status DelayableTimeoutCallback::delayUntil(Date_t when) { + stdx::lock_guard lk(_mutex); + return _delayUntil(lk, when); +} + +Status DelayableTimeoutCallback::_delayUntil(WithLock lk, Date_t when) { + if (!_cbHandle) { + // No timeout is active; just schedule it + return _reschedule(lk, when); + } + if (when == _nextCall) { + LOGV2_DEBUG(6602301, + 5, + "'Rescheduling' to same time", + "timerName"_attr = _timerName, + "when"_attr = when, + "nextCall"_attr = _nextCall); + } + _nextCall = when; + return Status::OK(); +} + +void DelayableTimeoutCallback::_handleTimeout(const executor::TaskExecutor::CallbackArgs& args) { + { + stdx::lock_guard lk(_mutex); + if (args.myHandle != _cbHandle) { + // This is normal when scheduleAt() or cancel() is used. + LOGV2_DEBUG(6602302, + 5, + "DelayableTimeoutCallback::_handleTimeout got a timeout after a new handle " + "was scheduled", + "timerName"_attr = _timerName); + return; + } + Date_t now = _executor->now(); + if (args.status == ErrorCodes::CallbackCanceled) { + // If args.status is CallbackCanceled yet the handles matched, that means the + // executor canceled the callback itself, probably as part of shutdown. We + // do not want to reschedule in this case, nor call the callback. + _cbHandle = executor::TaskExecutor::CallbackHandle(); + _nextCall = Date_t(); + return; + } else if (_nextCall > now) { + Status status = _reschedule(lk, _nextCall); + if (!status.isOK()) { + LOGV2_DEBUG(6602303, + 2, + "DelayableTimeoutCallback::_handleTimeout unable to schedule", + "timerName"_attr = _timerName, + "error"_attr = status); + fassert(6602305, status == ErrorCodes::ShutdownInProgress); + } + return; + } + _cbHandle = executor::TaskExecutor::CallbackHandle(); + _nextCall = Date_t(); + } + _callback(args); +} + +Status DelayableTimeoutCallback::_reschedule(WithLock, Date_t when) { + // We clear _cbHandle and _nextCall in advance so if scheduleWorkAt fails for any reason + // (including by exception), the invariant that _cbHandle and _nextCall are clear when no + // callback is scheduled is maintained. + _cbHandle = executor::TaskExecutor::CallbackHandle(); + _nextCall = Date_t(); + auto cbh = _executor->scheduleWorkAt( + when, [this](const executor::TaskExecutor::CallbackArgs& args) { _handleTimeout(args); }); + if (cbh == ErrorCodes::ShutdownInProgress) { + return cbh.getStatus(); + } + _nextCall = when; + _cbHandle = fassert(6602304, cbh); + return Status::OK(); +} + +void DelayableTimeoutCallbackWithJitter::_resetRandomization(WithLock) { + _lastRandomizationTime = Date_t(); + _currentJitter = Milliseconds(0); +} + +Status DelayableTimeoutCallbackWithJitter::scheduleAt(Date_t when) { + stdx::lock_guard lk(_mutex); + _resetRandomization(lk); + return _scheduleAt(lk, when); +} + +Status DelayableTimeoutCallbackWithJitter::delayUntil(Date_t when) { + stdx::lock_guard lk(_mutex); + _resetRandomization(lk); + return _delayUntil(lk, when); +} + +Status DelayableTimeoutCallbackWithJitter::delayUntilWithJitter(Date_t when, + Milliseconds jitterUpperBound) { + if (jitterUpperBound == Milliseconds::zero()) + return delayUntil(when); + stdx::lock_guard lk(_mutex); + Date_t now = _getExecutor()->now(); + Milliseconds elapsed = now - _lastRandomizationTime; + if (_lastRandomizationTime == Date_t() || elapsed < Milliseconds::zero() || + elapsed >= jitterUpperBound || jitterUpperBound < _currentJitter) { + _lastRandomizationTime = now; + _currentJitter = Milliseconds(_randomSource(durationCount<Milliseconds>(jitterUpperBound))); + } + return _delayUntil(lk, when + _currentJitter); +} + +} // namespace repl +} // namespace mongo diff --git a/src/mongo/db/repl/delayable_timeout_callback.h b/src/mongo/db/repl/delayable_timeout_callback.h new file mode 100644 index 00000000000..d1cb8e29f95 --- /dev/null +++ b/src/mongo/db/repl/delayable_timeout_callback.h @@ -0,0 +1,159 @@ +/** + * Copyright (C) 2022-present MongoDB, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * <http://www.mongodb.com/licensing/server-side-public-license>. + * + * As a special exception, the copyright holders give permission to link the + * code of portions of this program with the OpenSSL library under certain + * conditions as described in each individual source file and distribute + * linked combinations including the program with the OpenSSL library. You + * must comply with the Server Side Public License in all respects for + * all of the code used other than as permitted herein. If you modify file(s) + * with this exception, you may extend this exception to your version of the + * file(s), but you are not obligated to do so. If you do not wish to do so, + * delete this exception statement from your version. If you delete this + * exception statement from all source files in the program, then also delete + * it in the license file. + */ +#pragma once + +#include <string> + +#include "mongo/base/status.h" +#include "mongo/executor/task_executor.h" +#include "mongo/util/time_support.h" + +namespace mongo { +namespace repl { + +/** + * The DelayableTimeoutCallback is a utility class which allows a callback to be scheduled on an + * executor at a given time, and then that time pushed back (later) arbitrarily often without + * rescheduling the call on the executor. The callback is never called with CallbackCanceled or + * ShutdownInProgress. + * + * All methods are thread safe, though isActive() and getNextCall() may return stale information + * if external synchronization is not used. The callback is called without any locks held. + */ +class DelayableTimeoutCallback { +public: + /** + * Creates a DelayableTimeoutCallback with the given executor and callback function. The + * DelayableTimeoutCallback is inactive (the callback is not scheduled) when constructed. + */ + DelayableTimeoutCallback(executor::TaskExecutor* executor, + executor::TaskExecutor::CallbackFn callback, + std::string timerName = std::string()) + : _executor(executor), _callback(std::move(callback)), _timerName(std::move(timerName)){}; + + ~DelayableTimeoutCallback(); + + /** + * If the timeout is scheduled, cancel it. The callback function will not be called. + */ + void cancel(); + + /** + * Schedule the timeout to occur at "when", regardless of if or when it is already scheduled. + * If it is already scheduled to occur after "when", it is canceled and rescheduled + * + * Returns status of the attempt to schedule on the executor. + */ + Status scheduleAt(Date_t when); + + /** + * Schedule the timeout to occur at "when" if it is not scheduled or scheduled to occur before + * "when". If it is already scheduled to occur before "when", this call has no effect. + * + * Returns status of the attempt to schedule on the executor. + */ + Status delayUntil(Date_t when); + + /** + * Returns whether the callback is scheduled at all. + */ + bool isActive() const; + + /** + * Returns when the next call to the passed-in callback will be made, or Date_t() if inactive. + */ + Date_t getNextCall() const; + +protected: + void _cancel(WithLock); + Status _scheduleAt(WithLock, Date_t when); + Status _delayUntil(WithLock, Date_t when); + executor::TaskExecutor* _getExecutor() { + return _executor; + } + + mutable Mutex _mutex = MONGO_MAKE_LATCH("DelayableTimeoutCallback"); + +private: + void _handleTimeout(const executor::TaskExecutor::CallbackArgs& cbData); + Status _reschedule(WithLock, Date_t when); + + executor::TaskExecutor* _executor; + executor::TaskExecutor::CallbackHandle _cbHandle; + const executor::TaskExecutor::CallbackFn _callback; + Date_t _nextCall; + + // Timer name is used only for logging. + const std::string _timerName; +}; + +/** + * DelayableTimeoutCallbackWithJitter is a slight variation on DelayableTimeoutCallback + * which adds some additional random time to delays. Since the callback may be delayed at + * intervals much shorter than the random time, this would naively result in the timeout + * either being moved backwards often, or if we forbid moving it backwards, ending up quickly + * moving to the maximum jitter (which isn't very random). To avoid that, we only recompute + * the jitter every maximum jitter interval -- e.g. if the max jitter is 10 seconds and we + * add 3 seconds jitter at time T, we will add 3 seconds jitter to every subsequent call until + * time T + 10. + * + * The typical purpose of the jitter is to prevent two timers receiving delay calls at the same + * times from firing at the same time. + * + * Synchronization of the randomSource is up to the caller; it is provided externally to + * avoid having a separate random number generator per timer. The randomSource function will + * be called with the maximum jitter value passed to delayUntilWithJitter; it should return + * a value in the range [0, maxJitter) or [0, maxJitter] depending on what you want the + * actual jitter range to be. + */ +class DelayableTimeoutCallbackWithJitter : public DelayableTimeoutCallback { +public: + using RandomSource = std::function<int64_t(int64_t)>; + + DelayableTimeoutCallbackWithJitter(executor::TaskExecutor* executor, + executor::TaskExecutor::CallbackFn callback, + RandomSource randomSource, + std::string timerName = std::string()) + : DelayableTimeoutCallback(executor, std::move(callback), timerName), + _randomSource(std::move(randomSource)) {} + + Status scheduleAt(Date_t when); + Status delayUntil(Date_t when); + Status delayUntilWithJitter(Date_t when, Milliseconds maxJitter); + +private: + void _resetRandomization(WithLock); + + RandomSource _randomSource; + Date_t _lastRandomizationTime; + Milliseconds _currentJitter; +}; + +} // namespace repl +} // namespace mongo diff --git a/src/mongo/db/repl/delayable_timeout_callback_test.cpp b/src/mongo/db/repl/delayable_timeout_callback_test.cpp new file mode 100644 index 00000000000..b293d2793bb --- /dev/null +++ b/src/mongo/db/repl/delayable_timeout_callback_test.cpp @@ -0,0 +1,360 @@ +/** + * Copyright (C) 2022-present MongoDB, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * <http://www.mongodb.com/licensing/server-side-public-license>. + * + * As a special exception, the copyright holders give permission to link the + * code of portions of this program with the OpenSSL library under certain + * conditions as described in each individual source file and distribute + * linked combinations including the program with the OpenSSL library. You + * must comply with the Server Side Public License in all respects for + * all of the code used other than as permitted herein. If you modify file(s) + * with this exception, you may extend this exception to your version of the + * file(s), but you are not obligated to do so. If you do not wish to do so, + * delete this exception statement from your version. If you delete this + * exception statement from all source files in the program, then also delete + * it in the license file. + */ + +#include "mongo/db/repl/delayable_timeout_callback.h" +#include "mongo/executor/thread_pool_task_executor_test_fixture.h" + +namespace mongo { +namespace repl { +template <typename T> +class DelayableTimeoutCallbackBaseTest : public unittest::Test { +protected: + void setUp() override { + auto network = std::make_unique<executor::NetworkInterfaceMock>(); + _net = network.get(); + _executor = makeSharedThreadPoolTestExecutor(std::move(network)); + _executor->startup(); + createDelayableTimeoutCallback(); + } + + void tearDown() override { + _delayableTimeoutCallback = boost::none; + _executor->shutdown(); + _executor->join(); + _executor.reset(); + } + + void callback(const mongo::executor::TaskExecutor::CallbackArgs& cbData) { + callbackRan++; + if (!cbData.status.isOK()) { + callbackRanWithError++; + } + } + + void createDelayableTimeoutCallback() { + MONGO_UNREACHABLE; + } + + +protected: + boost::optional<T> _delayableTimeoutCallback; + executor::NetworkInterfaceMock* _net; + std::shared_ptr<executor::TaskExecutor> _executor; + int callbackRan = 0; + int callbackRanWithError = 0; +}; + +template <> +void DelayableTimeoutCallbackBaseTest<DelayableTimeoutCallback>::createDelayableTimeoutCallback() { + _delayableTimeoutCallback.emplace( + _executor.get(), [this](const mongo::executor::TaskExecutor::CallbackArgs& cbData) { + this->callback(cbData); + }); +} + +template <> +void DelayableTimeoutCallbackBaseTest< + DelayableTimeoutCallbackWithJitter>::createDelayableTimeoutCallback() { + _delayableTimeoutCallback.emplace( + _executor.get(), + [this](const mongo::executor::TaskExecutor::CallbackArgs& cbData) { + this->callback(cbData); + }, + [](int64_t limit) { + static int64_t notVeryRandom = 0; + notVeryRandom += 10; + return notVeryRandom % limit; + }); +} + +typedef DelayableTimeoutCallbackBaseTest<DelayableTimeoutCallback> DelayableTimeoutCallbackTest; +typedef DelayableTimeoutCallbackBaseTest<DelayableTimeoutCallbackWithJitter> + DelayableTimeoutCallbackWithJitterTest; + +TEST_F(DelayableTimeoutCallbackTest, ScheduleAtSchedulesFirstCallback) { + executor::NetworkInterfaceMock::InNetworkGuard guard(_net); + ASSERT_FALSE(_delayableTimeoutCallback->isActive()); + + ASSERT_OK(_delayableTimeoutCallback->scheduleAt(_net->now() + Seconds(2))); + + ASSERT_TRUE(_delayableTimeoutCallback->isActive()); + ASSERT_EQ(0, callbackRan); + _net->runUntil(_net->now() + Seconds(1)); + ASSERT_TRUE(_delayableTimeoutCallback->isActive()); + ASSERT_EQ(0, callbackRan); + _net->runUntil(_net->now() + Seconds(1)); + ASSERT_EQ(1, callbackRan); + ASSERT_EQ(0, callbackRanWithError); + ASSERT_FALSE(_delayableTimeoutCallback->isActive()); +} + +TEST_F(DelayableTimeoutCallbackTest, DelayUntilSchedulesFirstCallback) { + executor::NetworkInterfaceMock::InNetworkGuard guard(_net); + ASSERT_FALSE(_delayableTimeoutCallback->isActive()); + + ASSERT_OK(_delayableTimeoutCallback->delayUntil(_net->now() + Seconds(2))); + + ASSERT_TRUE(_delayableTimeoutCallback->isActive()); + ASSERT_EQ(0, callbackRan); + _net->runUntil(_net->now() + Seconds(1)); + ASSERT_TRUE(_delayableTimeoutCallback->isActive()); + ASSERT_EQ(0, callbackRan); + _net->runUntil(_net->now() + Seconds(1)); + ASSERT_EQ(1, callbackRan); + ASSERT_EQ(0, callbackRanWithError); + ASSERT_FALSE(_delayableTimeoutCallback->isActive()); +} + +TEST_F(DelayableTimeoutCallbackTest, ScheduleAtMovesCallbackLater) { + executor::NetworkInterfaceMock::InNetworkGuard guard(_net); + ASSERT_FALSE(_delayableTimeoutCallback->isActive()); + + ASSERT_OK(_delayableTimeoutCallback->scheduleAt(_net->now() + Seconds(2))); + + ASSERT_TRUE(_delayableTimeoutCallback->isActive()); + ASSERT_EQ(0, callbackRan); + _net->runUntil(_net->now() + Seconds(1)); + ASSERT_TRUE(_delayableTimeoutCallback->isActive()); + ASSERT_EQ(0, callbackRan); + + ASSERT_OK(_delayableTimeoutCallback->scheduleAt(_net->now() + Seconds(2))); + + _net->runUntil(_net->now() + Seconds(1)); + ASSERT_TRUE(_delayableTimeoutCallback->isActive()); + ASSERT_EQ(0, callbackRan); + _net->runUntil(_net->now() + Seconds(1)); + ASSERT_EQ(1, callbackRan); + ASSERT_EQ(0, callbackRanWithError); + ASSERT_FALSE(_delayableTimeoutCallback->isActive()); +} + +TEST_F(DelayableTimeoutCallbackTest, DelayUntilMovesCallbackLater) { + executor::NetworkInterfaceMock::InNetworkGuard guard(_net); + ASSERT_FALSE(_delayableTimeoutCallback->isActive()); + + ASSERT_OK(_delayableTimeoutCallback->delayUntil(_net->now() + Seconds(2))); + + ASSERT_TRUE(_delayableTimeoutCallback->isActive()); + ASSERT_EQ(0, callbackRan); + _net->runUntil(_net->now() + Seconds(1)); + ASSERT_TRUE(_delayableTimeoutCallback->isActive()); + ASSERT_EQ(0, callbackRan); + + ASSERT_OK(_delayableTimeoutCallback->delayUntil(_net->now() + Seconds(2))); + + _net->runUntil(_net->now() + Seconds(1)); + ASSERT_TRUE(_delayableTimeoutCallback->isActive()); + ASSERT_EQ(0, callbackRan); + _net->runUntil(_net->now() + Seconds(1)); + ASSERT_EQ(1, callbackRan); + ASSERT_EQ(0, callbackRanWithError); + ASSERT_FALSE(_delayableTimeoutCallback->isActive()); +} + +TEST_F(DelayableTimeoutCallbackTest, ScheduleAtMovesCallbackEarlier) { + executor::NetworkInterfaceMock::InNetworkGuard guard(_net); + ASSERT_FALSE(_delayableTimeoutCallback->isActive()); + + ASSERT_OK(_delayableTimeoutCallback->scheduleAt(_net->now() + Seconds(3))); + + ASSERT_TRUE(_delayableTimeoutCallback->isActive()); + ASSERT_EQ(0, callbackRan); + _net->runUntil(_net->now() + Seconds(1)); + ASSERT_TRUE(_delayableTimeoutCallback->isActive()); + ASSERT_EQ(0, callbackRan); + + ASSERT_OK(_delayableTimeoutCallback->scheduleAt(_net->now() + Seconds(1))); + + _net->runUntil(_net->now() + Seconds(1)); + ASSERT_EQ(1, callbackRan); + ASSERT_EQ(0, callbackRanWithError); + ASSERT_FALSE(_delayableTimeoutCallback->isActive()); +} + +TEST_F(DelayableTimeoutCallbackTest, DelayUntilDoesNotMoveCallbackEarlier) { + executor::NetworkInterfaceMock::InNetworkGuard guard(_net); + ASSERT_FALSE(_delayableTimeoutCallback->isActive()); + + ASSERT_OK(_delayableTimeoutCallback->delayUntil(_net->now() + Seconds(3))); + + ASSERT_TRUE(_delayableTimeoutCallback->isActive()); + ASSERT_EQ(0, callbackRan); + _net->runUntil(_net->now() + Seconds(1)); + ASSERT_TRUE(_delayableTimeoutCallback->isActive()); + ASSERT_EQ(0, callbackRan); + + ASSERT_OK(_delayableTimeoutCallback->delayUntil(_net->now() + Seconds(1))); + + _net->runUntil(_net->now() + Seconds(1)); + ASSERT_TRUE(_delayableTimeoutCallback->isActive()); + ASSERT_EQ(0, callbackRan); + _net->runUntil(_net->now() + Seconds(1)); + ASSERT_EQ(1, callbackRan); + ASSERT_EQ(0, callbackRanWithError); + ASSERT_FALSE(_delayableTimeoutCallback->isActive()); +} + +TEST_F(DelayableTimeoutCallbackTest, ScheduleAtInPastRunsImmediately) { + executor::NetworkInterfaceMock::InNetworkGuard guard(_net); + // Make sure there's a past to schedule in. + _net->runUntil(_net->now() + Days(1)); + + ASSERT_FALSE(_delayableTimeoutCallback->isActive()); + ASSERT_OK(_delayableTimeoutCallback->scheduleAt(_net->now() - Seconds(1))); + + // Needed to trigger anything scheduled. + _net->runReadyNetworkOperations(); + + ASSERT_EQ(1, callbackRan); + ASSERT_EQ(0, callbackRanWithError); + ASSERT_FALSE(_delayableTimeoutCallback->isActive()); +} + +TEST_F(DelayableTimeoutCallbackTest, DelayUntilInPastRunsImmediately) { + executor::NetworkInterfaceMock::InNetworkGuard guard(_net); + // Make sure there's a past to schedule in. + _net->runUntil(_net->now() + Days(1)); + + ASSERT_FALSE(_delayableTimeoutCallback->isActive()); + ASSERT_OK(_delayableTimeoutCallback->delayUntil(_net->now() - Seconds(1))); + + // Needed to trigger anything scheduled. + _net->runReadyNetworkOperations(); + + ASSERT_EQ(1, callbackRan); + ASSERT_EQ(0, callbackRanWithError); + ASSERT_FALSE(_delayableTimeoutCallback->isActive()); +} + +TEST_F(DelayableTimeoutCallbackTest, Cancellation) { + executor::NetworkInterfaceMock::InNetworkGuard guard(_net); + ASSERT_FALSE(_delayableTimeoutCallback->isActive()); + + ASSERT_OK(_delayableTimeoutCallback->scheduleAt(_net->now() + Seconds(2))); + + ASSERT_TRUE(_delayableTimeoutCallback->isActive()); + ASSERT_EQ(0, callbackRan); + _net->runUntil(_net->now() + Seconds(1)); + ASSERT_TRUE(_delayableTimeoutCallback->isActive()); + _delayableTimeoutCallback->cancel(); + ASSERT_EQ(0, callbackRan); + ASSERT_FALSE(_delayableTimeoutCallback->isActive()); + _net->runUntil(_net->now() + Seconds(1)); + ASSERT_EQ(0, callbackRan); + ASSERT_EQ(0, callbackRanWithError); + ASSERT_FALSE(_delayableTimeoutCallback->isActive()); +} + +TEST_F(DelayableTimeoutCallbackTest, Shutdown) { + executor::NetworkInterfaceMock::InNetworkGuard guard(_net); + ASSERT_FALSE(_delayableTimeoutCallback->isActive()); + + ASSERT_OK(_delayableTimeoutCallback->scheduleAt(_net->now() + Seconds(2))); + + ASSERT_TRUE(_delayableTimeoutCallback->isActive()); + ASSERT_EQ(0, callbackRan); + _net->runUntil(_net->now() + Seconds(1)); + ASSERT_TRUE(_delayableTimeoutCallback->isActive()); + + _executor->shutdown(); + + // This makes sure the executor processes the shutdown. + _net->runReadyNetworkOperations(); + + ASSERT_EQ(0, callbackRan); + ASSERT_EQ(0, callbackRanWithError); + ASSERT_FALSE(_delayableTimeoutCallback->isActive()); +} + +TEST_F(DelayableTimeoutCallbackWithJitterTest, DelayUntilWithJitter) { + executor::NetworkInterfaceMock::InNetworkGuard guard(_net); + ASSERT_FALSE(_delayableTimeoutCallback->isActive()); + + ASSERT_OK(_delayableTimeoutCallback->delayUntilWithJitter(_net->now() + Seconds(10), + Milliseconds(100))); + + ASSERT_TRUE(_delayableTimeoutCallback->isActive()); + // Our "random" generator is just 10,20,30,... + ASSERT_EQ(_net->now() + Milliseconds(10010), _delayableTimeoutCallback->getNextCall()); + ASSERT_EQ(0, callbackRan); + + // Setting it again in the same time shouldn't change jitter. + ASSERT_OK(_delayableTimeoutCallback->delayUntilWithJitter(_net->now() + Seconds(10), + Milliseconds(100))); + ASSERT_EQ(_net->now() + Milliseconds(10010), _delayableTimeoutCallback->getNextCall()); + ASSERT_EQ(0, callbackRan); + + // Move forward less than the max jitter shouldn't change jitter. + for (int i = 0; i < 3; i++) { + _net->runUntil(_net->now() + Milliseconds(25)); + ASSERT_OK(_delayableTimeoutCallback->delayUntilWithJitter(_net->now() + Seconds(10), + Milliseconds(100))); + ASSERT_EQ(_net->now() + Milliseconds(10010), _delayableTimeoutCallback->getNextCall()); + ASSERT_EQ(0, callbackRan); + } + + // Move forward to the max jitter should recalculate jitter. + _net->runUntil(_net->now() + Milliseconds(25)); + ASSERT_OK(_delayableTimeoutCallback->delayUntilWithJitter(_net->now() + Seconds(10), + Milliseconds(100))); + ASSERT_EQ(_net->now() + Milliseconds(10020), _delayableTimeoutCallback->getNextCall()); + ASSERT_EQ(0, callbackRan); + + // Setting max jitter to less than actual jitter should recalculate jitter. + ASSERT_OK(_delayableTimeoutCallback->delayUntilWithJitter(_net->now() + Seconds(10), + Milliseconds(19))); + // Jitter value will be 30 % 19 = 11 + ASSERT_EQ(_net->now() + Milliseconds(10011), _delayableTimeoutCallback->getNextCall()); + ASSERT_EQ(0, callbackRan); + + _net->runUntil(_net->now() + Milliseconds(10011)); + ASSERT_EQ(1, callbackRan); + ASSERT_EQ(0, callbackRanWithError); + ASSERT_FALSE(_delayableTimeoutCallback->isActive()); +} + +TEST_F(DelayableTimeoutCallbackWithJitterTest, DelayUntilWithZeroJitter) { + executor::NetworkInterfaceMock::InNetworkGuard guard(_net); + ASSERT_FALSE(_delayableTimeoutCallback->isActive()); + + ASSERT_OK( + _delayableTimeoutCallback->delayUntilWithJitter(_net->now() + Seconds(10), Seconds(0))); + + ASSERT_TRUE(_delayableTimeoutCallback->isActive()); + ASSERT_EQ(_net->now() + Milliseconds(10000), _delayableTimeoutCallback->getNextCall()); + ASSERT_EQ(0, callbackRan); + + _net->runUntil(_net->now() + Milliseconds(10000)); + ASSERT_EQ(1, callbackRan); + ASSERT_EQ(0, callbackRanWithError); + ASSERT_FALSE(_delayableTimeoutCallback->isActive()); +} + +} // namespace repl +} // namespace mongo diff --git a/src/mongo/db/repl/idempotency_test_fixture.cpp b/src/mongo/db/repl/idempotency_test_fixture.cpp index 0043ff54e3e..3f80a0b1663 100644 --- a/src/mongo/db/repl/idempotency_test_fixture.cpp +++ b/src/mongo/db/repl/idempotency_test_fixture.cpp @@ -43,7 +43,6 @@ #include "mongo/db/catalog/index_catalog.h" #include "mongo/db/client.h" #include "mongo/db/concurrency/d_concurrency.h" -#include "mongo/db/concurrency/write_conflict_exception.h" #include "mongo/db/curop.h" #include "mongo/db/db_raii.h" #include "mongo/db/index_builds_coordinator.h" @@ -406,7 +405,8 @@ CollectionState IdempotencyTest::validate(const NamespaceString& nss) { CollectionValidation::ValidateMode::kForegroundFull, CollectionValidation::RepairMode::kNone, &validateResults, - &bob)); + &bob, + /*logDiagnostics=*/false)); ASSERT_TRUE(validateResults.valid); } diff --git a/src/mongo/db/repl/initial_syncer.cpp b/src/mongo/db/repl/initial_syncer.cpp index 02fb4af0be5..d8a2f5104d7 100644 --- a/src/mongo/db/repl/initial_syncer.cpp +++ b/src/mongo/db/repl/initial_syncer.cpp @@ -548,6 +548,7 @@ void InitialSyncer::_setUp_inlock(OperationContext* opCtx, std::uint32_t initial _stats.initialSyncStart = _exec->now(); _stats.maxFailedInitialSyncAttempts = initialSyncMaxAttempts; _stats.failedInitialSyncAttempts = 0; + _stats.exec = std::weak_ptr<executor::TaskExecutor>(_exec); _allowedOutageDuration = Seconds(initialSyncTransientErrorRetryPeriodSeconds.load()); } @@ -1393,53 +1394,63 @@ void InitialSyncer::_lastOplogEntryFetcherCallbackForStopTimestamp( std::shared_ptr<OnCompletionGuard> onCompletionGuard) { OpTimeAndWallTime resultOpTimeAndWallTime = {OpTime(), Date_t()}; { - stdx::lock_guard<Latch> lock(_mutex); - auto status = _checkForShutdownAndConvertStatus_inlock( - result.getStatus(), "error fetching last oplog entry for stop timestamp"); - if (_shouldRetryError(lock, status)) { - auto scheduleStatus = - (*_attemptExec) - ->scheduleWork([this, - onCompletionGuard](executor::TaskExecutor::CallbackArgs args) { - // It is not valid to schedule the retry from within this callback, - // hence we schedule a lambda to schedule the retry. - stdx::lock_guard<Latch> lock(_mutex); - // Since the stopTimestamp is retrieved after we have done all the work of - // retrieving collection data, we handle retries within this class by - // retrying for 'initialSyncTransientErrorRetryPeriodSeconds' (default 24 - // hours). This is the same retry strategy used when retrieving collection - // data, and avoids retrieving all the data and then throwing it away due to - // a transient network outage. - auto status = _scheduleLastOplogEntryFetcher_inlock( - [=](const StatusWith<mongo::Fetcher::QueryResponse>& status, - mongo::Fetcher::NextAction*, - mongo::BSONObjBuilder*) { - _lastOplogEntryFetcherCallbackForStopTimestamp(status, - onCompletionGuard); - }, - kInitialSyncerHandlesRetries); - if (!status.isOK()) { - onCompletionGuard->setResultAndCancelRemainingWork_inlock(lock, status); - } - }); - if (scheduleStatus.isOK()) + { + stdx::lock_guard<Latch> lock(_mutex); + auto status = _checkForShutdownAndConvertStatus_inlock( + result.getStatus(), "error fetching last oplog entry for stop timestamp"); + if (_shouldRetryError(lock, status)) { + auto scheduleStatus = + (*_attemptExec) + ->scheduleWork( + [this, onCompletionGuard](executor::TaskExecutor::CallbackArgs args) { + // It is not valid to schedule the retry from within this callback, + // hence we schedule a lambda to schedule the retry. + stdx::lock_guard<Latch> lock(_mutex); + // Since the stopTimestamp is retrieved after we have done all the + // work of retrieving collection data, we handle retries within this + // class by retrying for + // 'initialSyncTransientErrorRetryPeriodSeconds' (default 24 hours). + // This is the same retry strategy used when retrieving collection + // data, and avoids retrieving all the data and then throwing it + // away due to a transient network outage. + auto status = _scheduleLastOplogEntryFetcher_inlock( + [=](const StatusWith<mongo::Fetcher::QueryResponse>& status, + mongo::Fetcher::NextAction*, + mongo::BSONObjBuilder*) { + _lastOplogEntryFetcherCallbackForStopTimestamp( + status, onCompletionGuard); + }, + kInitialSyncerHandlesRetries); + if (!status.isOK()) { + onCompletionGuard->setResultAndCancelRemainingWork_inlock( + lock, status); + } + }); + if (scheduleStatus.isOK()) + return; + // If scheduling failed, we're shutting down and cannot retry. + // So just continue with the original failed status. + } + if (!status.isOK()) { + onCompletionGuard->setResultAndCancelRemainingWork_inlock(lock, status); return; - // If scheduling failed, we're shutting down and cannot retry. - // So just continue with the original failed status. - } - if (!status.isOK()) { - onCompletionGuard->setResultAndCancelRemainingWork_inlock(lock, status); - return; - } + } - auto&& optimeStatus = parseOpTimeAndWallTime(result); - if (!optimeStatus.isOK()) { - onCompletionGuard->setResultAndCancelRemainingWork_inlock(lock, - optimeStatus.getStatus()); - return; + auto&& optimeStatus = parseOpTimeAndWallTime(result); + if (!optimeStatus.isOK()) { + onCompletionGuard->setResultAndCancelRemainingWork_inlock(lock, + optimeStatus.getStatus()); + return; + } + resultOpTimeAndWallTime = optimeStatus.getValue(); } - resultOpTimeAndWallTime = optimeStatus.getValue(); + // Release the _mutex to write to disk. + auto opCtx = makeOpCtx(); + _replicationProcess->getConsistencyMarkers()->setMinValid( + opCtx.get(), resultOpTimeAndWallTime.opTime, true); + + stdx::lock_guard<Latch> lock(_mutex); _initialSyncState->stopTimestamp = resultOpTimeAndWallTime.opTime.getTimestamp(); // If the beginFetchingTimestamp is different from the stopTimestamp, it indicates that @@ -2186,9 +2197,10 @@ void InitialSyncer::Stats::append(BSONObjBuilder* builder) const { builder->appendNumber("maxFailedInitialSyncAttempts", static_cast<long long>(maxFailedInitialSyncAttempts)); + auto e = exec.lock(); if (initialSyncStart != Date_t()) { builder->appendDate("initialSyncStart", initialSyncStart); - auto elapsedDurationEnd = Date_t::now(); + auto elapsedDurationEnd = e ? e->now() : Date_t::now(); if (initialSyncEnd != Date_t()) { builder->appendDate("initialSyncEnd", initialSyncEnd); elapsedDurationEnd = initialSyncEnd; diff --git a/src/mongo/db/repl/initial_syncer.h b/src/mongo/db/repl/initial_syncer.h index 177122e12d9..3a124b6b02d 100644 --- a/src/mongo/db/repl/initial_syncer.h +++ b/src/mongo/db/repl/initial_syncer.h @@ -140,6 +140,7 @@ public: Date_t initialSyncStart; Date_t initialSyncEnd; std::vector<InitialSyncer::InitialSyncAttemptInfo> initialSyncAttemptInfos; + std::weak_ptr<executor::TaskExecutor> exec; std::string toString() const; BSONObj toBSON() const; diff --git a/src/mongo/db/repl/initial_syncer_test.cpp b/src/mongo/db/repl/initial_syncer_test.cpp index 147d0571ceb..82670cc53ba 100644 --- a/src/mongo/db/repl/initial_syncer_test.cpp +++ b/src/mongo/db/repl/initial_syncer_test.cpp @@ -4510,6 +4510,10 @@ TEST_F(InitialSyncerTest, TestRemainingInitialSyncEstimatedMillisMetric) { // Wait for the server to have reached the end of cloning collection 'a.a'. The size of this // collection is expected to equal 'dbSize'. hangDuringCloningFailPoint->waitForTimesEntered(timesEntered + 1); + { + executor::NetworkInterfaceMock::InNetworkGuard guard(net); + net->runUntil(Date_t::now() + Seconds(1)); + } auto progress = initialSyncer->getInitialSyncProgress(); LOGV2(5301701, "Progress in middle of cloning", "progress"_attr = progress); { @@ -4605,7 +4609,7 @@ TEST_F(InitialSyncerTest, GetInitialSyncProgressReturnsCorrectProgress) { ASSERT_FALSE(progress.hasField("InitialSyncEnd")); ASSERT_EQUALS(progress.getIntField("failedInitialSyncAttempts"), 0) << progress; ASSERT_EQUALS(progress.getIntField("maxFailedInitialSyncAttempts"), 2) << progress; - ASSERT_EQUALS(progress["totalInitialSyncElapsedMillis"].type(), NumberLong) << progress; + ASSERT_EQUALS(progress["totalInitialSyncElapsedMillis"].type(), NumberInt) << progress; ASSERT_EQUALS(progress.getIntField("approxTotalDataSize"), 0) << progress; ASSERT_EQUALS(progress.getIntField("approxTotalBytesCopied"), 0) << progress; ASSERT_EQUALS(progress["initialSyncStart"].type(), Date) << progress; @@ -4673,7 +4677,7 @@ TEST_F(InitialSyncerTest, GetInitialSyncProgressReturnsCorrectProgress) { ASSERT_FALSE(progress.hasField("InitialSyncEnd")); ASSERT_EQUALS(progress.getIntField("failedInitialSyncAttempts"), 1) << progress; ASSERT_EQUALS(progress.getIntField("maxFailedInitialSyncAttempts"), 2) << progress; - ASSERT_EQUALS(progress["totalInitialSyncElapsedMillis"].type(), NumberLong) << progress; + ASSERT_EQUALS(progress["totalInitialSyncElapsedMillis"].type(), NumberInt) << progress; ASSERT_EQUALS(progress.getIntField("approxTotalDataSize"), 0) << progress; ASSERT_EQUALS(progress.getIntField("approxTotalBytesCopied"), 0) << progress; ASSERT_EQUALS(progress["initialSyncStart"].type(), Date) << progress; @@ -4775,7 +4779,7 @@ TEST_F(InitialSyncerTest, GetInitialSyncProgressReturnsCorrectProgress) { ASSERT_EQUALS(progress.nFields(), 14) << progress; ASSERT_EQUALS(progress.getIntField("failedInitialSyncAttempts"), 1) << progress; ASSERT_EQUALS(progress.getIntField("maxFailedInitialSyncAttempts"), 2) << progress; - ASSERT_EQUALS(progress["totalInitialSyncElapsedMillis"].type(), NumberLong) << progress; + ASSERT_EQUALS(progress["totalInitialSyncElapsedMillis"].type(), NumberInt) << progress; ASSERT_EQUALS(progress.getIntField("approxTotalDataSize"), 10) << progress; ASSERT_EQUALS(progress.getIntField("approxTotalBytesCopied"), 10) << progress; ASSERT_EQUALS(progress["initialSyncOplogStart"].timestamp(), Timestamp(1, 1)) << progress; diff --git a/src/mongo/db/repl/isself.cpp b/src/mongo/db/repl/isself.cpp index 243a1c065f0..1e0ec7dfab7 100644 --- a/src/mongo/db/repl/isself.cpp +++ b/src/mongo/db/repl/isself.cpp @@ -163,10 +163,17 @@ std::vector<std::string> getAddrsForHost(const std::string& iporhost, } // namespace -bool isSelf(const HostAndPort& hostAndPort, ServiceContext* const ctx) { +bool isSelf(const HostAndPort& hostAndPort, ServiceContext* const ctx, Milliseconds timeout) { + if (isSelfFastPath(hostAndPort)) { + return true; + } + return isSelfSlowPath(hostAndPort, ctx, timeout); +} + +bool isSelfFastPath(const HostAndPort& hostAndPort) { if (MONGO_unlikely(failIsSelfCheck.shouldFail())) { LOGV2(356490, - "failIsSelfCheck failpoint activated, returning false from isSelf", + "failIsSelfCheck failpoint activated, returning false from isSelfFastPath", "hostAndPort"_attr = hostAndPort); return false; } @@ -222,18 +229,30 @@ bool isSelf(const HostAndPort& hostAndPort, ServiceContext* const ctx) { } } } + return false; +} +bool isSelfSlowPath(const HostAndPort& hostAndPort, + ServiceContext* const ctx, + Milliseconds timeout) { ctx->waitForStartupComplete(); + if (MONGO_unlikely(failIsSelfCheck.shouldFail())) { + LOGV2(6605000, + "failIsSelfCheck failpoint activated, returning false from isSelfSlowPath", + "hostAndPort"_attr = hostAndPort); + return false; + } try { DBClientConnection conn; - conn.setSoTimeout(30); // 30 second timeout - - // We need to avoid the isMaster call triggered by a normal connect, which would - // cause a deadlock. 'isSelf' is called by the Replication Coordinator when validating - // a replica set configuration document, but the 'isMaster' command requires a lock on the - // replication coordinator to execute. As such we call we call 'connectSocketOnly', which - // does not call 'isMaster'. + double timeoutSeconds = static_cast<double>(durationCount<Milliseconds>(timeout)) / 1000.0; + conn.setSoTimeout(timeoutSeconds); + + // We need to avoid the "hello" call triggered by a normal connect, which would cause a + // deadlock. 'isSelf' is called by the Replication Coordinator when validating a replica set + // configuration document, but the "hello" command requires a lock on the replication + // coordinator to execute. As such we call we call 'connectSocketOnly', which does not call + // "hello". auto connectSocketResult = conn.connectSocketOnly(hostAndPort, boost::none); if (!connectSocketResult.isOK()) { LOGV2(4834700, diff --git a/src/mongo/db/repl/isself.h b/src/mongo/db/repl/isself.h index 84264978720..fb61c8a6cb3 100644 --- a/src/mongo/db/repl/isself.h +++ b/src/mongo/db/repl/isself.h @@ -33,6 +33,7 @@ #include <vector> #include "mongo/bson/oid.h" +#include "mongo/util/duration.h" namespace mongo { struct HostAndPort; @@ -49,7 +50,21 @@ extern OID instanceId; /** * Returns true if "hostAndPort" identifies this instance. */ -bool isSelf(const HostAndPort& hostAndPort, ServiceContext* ctx); +bool isSelf(const HostAndPort& hostAndPort, + ServiceContext* ctx, + Milliseconds timeout = Seconds(30)); + +/** + * Returns true if "hostAndPort" identifies this instance by checking our bound IP addresses, + * without going out to the network and running the _isSelf command on the node. + */ +bool isSelfFastPath(const HostAndPort& hostAndPort); + +/** + * Returns true if "hostAndPort" identifies this instance by running the _isSelf command on the + * node. + */ +bool isSelfSlowPath(const HostAndPort& hostAndPort, ServiceContext* ctx, Milliseconds timeout); /** * Returns all the IP addresses bound to the network interfaces of this machine. diff --git a/src/mongo/db/repl/isself_test.cpp b/src/mongo/db/repl/isself_test.cpp index 511ef122ecf..91ad0ad3803 100644 --- a/src/mongo/db/repl/isself_test.cpp +++ b/src/mongo/db/repl/isself_test.cpp @@ -55,6 +55,7 @@ TEST_F(ServiceContextTest, DetectsSameHostIPv4) { // Fastpath should agree with the result of getBoundAddrs // since it uses it... for (std::vector<string>::const_iterator it = addrs.begin(); it != addrs.end(); ++it) { + ASSERT(isSelfFastPath(HostAndPort(*it, serverGlobalParams.port))); ASSERT(isSelf(HostAndPort(*it, serverGlobalParams.port), getGlobalServiceContext())); } #else @@ -72,6 +73,7 @@ TEST_F(ServiceContextTest, DetectsSameHostIPv6) { // Fastpath should agree with the result of getBoundAddrs // since it uses it... for (std::vector<string>::const_iterator it = addrs.begin(); it != addrs.end(); ++it) { + ASSERT(isSelfFastPath(HostAndPort(*it, serverGlobalParams.port))); ASSERT(isSelf(HostAndPort(*it, serverGlobalParams.port), getGlobalServiceContext())); } #else diff --git a/src/mongo/db/repl/member_config.h b/src/mongo/db/repl/member_config.h index e68c24b6f2d..cc668fef79c 100644 --- a/src/mongo/db/repl/member_config.h +++ b/src/mongo/db/repl/member_config.h @@ -115,7 +115,7 @@ public: } /** - * Gets the horizon name for which the parameters (captured during the first `isMaster`) + * Gets the horizon name for which the parameters (captured during the first `hello`) * correspond. */ StringData determineHorizon(const SplitHorizon::Parameters& params) const { @@ -194,7 +194,7 @@ public: } /** - * Returns true if this member is hidden (not reported by isMaster, not electable). + * Returns true if this member is hidden (not reported by "hello", not electable). */ bool isHidden() const { return getHidden(); diff --git a/src/mongo/db/repl/noop_writer.cpp b/src/mongo/db/repl/noop_writer.cpp index b3adb9da2cf..ad3fdd197ed 100644 --- a/src/mongo/db/repl/noop_writer.cpp +++ b/src/mongo/db/repl/noop_writer.cpp @@ -35,7 +35,7 @@ #include "mongo/db/commands.h" #include "mongo/db/concurrency/d_concurrency.h" -#include "mongo/db/concurrency/write_conflict_exception.h" +#include "mongo/db/concurrency/exception_util.h" #include "mongo/db/curop.h" #include "mongo/db/namespace_string.h" #include "mongo/db/op_observer.h" diff --git a/src/mongo/db/repl/oplog.cpp b/src/mongo/db/repl/oplog.cpp index 453c17c6688..fd99387d02f 100644 --- a/src/mongo/db/repl/oplog.cpp +++ b/src/mongo/db/repl/oplog.cpp @@ -62,8 +62,9 @@ #include "mongo/db/client.h" #include "mongo/db/coll_mod_gen.h" #include "mongo/db/commands.h" +#include "mongo/db/commands/create_gen.h" #include "mongo/db/commands/feature_compatibility_version_parser.h" -#include "mongo/db/concurrency/write_conflict_exception.h" +#include "mongo/db/concurrency/exception_util.h" #include "mongo/db/db_raii.h" #include "mongo/db/dbdirectclient.h" #include "mongo/db/dbhelpers.h" @@ -117,6 +118,7 @@ using std::string; using std::stringstream; using std::unique_ptr; using std::vector; +using namespace std::string_literals; using IndexVersion = IndexDescriptor::IndexVersion; @@ -289,23 +291,6 @@ void writeToImageCollection(OperationContext* opCtx, DisableDocumentValidation documentValidationDisabler( opCtx, DocumentValidationSettings::kDisableInternalValidation); - BSONObj existingImageEntryBson; - Helpers::findOne(opCtx, - autoColl.getCollection(), - BSON("_id" << imageEntry.get_id().toBSON() << "ts" << imageEntry.getTs()), - existingImageEntryBson); - if (!existingImageEntryBson.isEmpty()) { - auto existingImageEntry = repl::ImageEntry::parse( - IDLParserErrorContext("writeToImageCollection"), existingImageEntryBson); - uassert( - 6652600, - str::stream() - << "Found an existing findAndModify image entry with unexpected content. Found: " - << existingImageEntry.toBSON() << ". Expected: " << imageEntry.toBSON(), - existingImageEntry.toBSON().woCompare(imageEntry.toBSON()) == 0); - return; - } - UpdateRequest request; request.setNamespaceString(NamespaceString::kConfigImagesNamespace); request.setQuery( @@ -806,6 +791,26 @@ NamespaceString extractNsFromUUIDorNs(OperationContext* opCtx, return ui ? extractNsFromUUID(opCtx, ui.get()) : extractNs(ns.db(), cmd); } +StatusWith<BSONObj> getObjWithSanitizedStorageEngineOptions(OperationContext* opCtx, + const BSONObj& cmd) { + static_assert( + CreateCommand::kStorageEngineFieldName == IndexDescriptor::kStorageEngineFieldName, + "Expected storage engine options field to be the same for collections and indexes."); + + if (auto storageEngineElem = cmd[IndexDescriptor::kStorageEngineFieldName]) { + auto storageEngine = opCtx->getServiceContext()->getStorageEngine(); + auto engineObj = storageEngineElem.embeddedObject(); + auto sanitizedObj = + storageEngine->getSanitizedStorageOptionsForSecondaryReplication(engineObj); + if (!sanitizedObj.isOK()) { + return sanitizedObj.getStatus(); + } + return cmd.addFields( + BSON(IndexDescriptor::kStorageEngineFieldName << sanitizedObj.getValue())); + } + return cmd; +} + using OpApplyFn = std::function<Status( OperationContext* opCtx, const OplogEntry& entry, OplogApplication::Mode mode)>; @@ -830,7 +835,14 @@ const StringMap<ApplyOpMetadata> kOpsMap = { {"create", {[](OperationContext* opCtx, const OplogEntry& entry, OplogApplication::Mode mode) -> Status { const auto& ui = entry.getUuid(); - const auto& cmd = entry.getObject(); + // Sanitize storage engine options to remove options which might not apply to this node. + // See SERVER-68122. + const auto sanitizedCmdOrStatus = + getObjWithSanitizedStorageEngineOptions(opCtx, entry.getObject()); + if (!sanitizedCmdOrStatus.isOK()) { + return sanitizedCmdOrStatus.getStatus(); + } + const auto& cmd = sanitizedCmdOrStatus.getValue(); const NamespaceString nss(extractNs(entry.getNss().db(), cmd)); // Mode SECONDARY steady state replication should not allow create collection to rename an @@ -874,7 +886,15 @@ const StringMap<ApplyOpMetadata> kOpsMap = { {ErrorCodes::NamespaceExists}}}, {"createIndexes", {[](OperationContext* opCtx, const OplogEntry& entry, OplogApplication::Mode mode) -> Status { - const auto& cmd = entry.getObject(); + // Sanitize storage engine options to remove options which might not apply to this node. + // See SERVER-68122. + const auto sanitizedCmdOrStatus = + getObjWithSanitizedStorageEngineOptions(opCtx, entry.getObject()); + if (!sanitizedCmdOrStatus.isOK()) { + return sanitizedCmdOrStatus.getStatus(); + } + const auto& cmd = sanitizedCmdOrStatus.getValue(); + if (OplogApplication::Mode::kApplyOpsCmd == mode) { return {ErrorCodes::CommandNotSupported, "The createIndexes operation is not supported in applyOps mode"}; @@ -909,6 +929,16 @@ const StringMap<ApplyOpMetadata> kOpsMap = { "Error parsing 'startIndexBuild' oplog entry"); } + // Sanitize storage engine options to remove options which might not apply to this node. + // See SERVER-68122. + for (auto& spec : swOplogEntry.getValue().indexSpecs) { + auto sanitizedObj = getObjWithSanitizedStorageEngineOptions(opCtx, spec); + if (!sanitizedObj.isOK()) { + return swOplogEntry.getStatus(); + } + spec = sanitizedObj.getValue(); + } + IndexBuildsCoordinator::ApplicationMode applicationMode = IndexBuildsCoordinator::ApplicationMode::kNormal; if (mode == OplogApplication::Mode::kInitialSync) { @@ -1110,6 +1140,8 @@ void writeChangeStreamPreImage(OperationContext* opCtx, constexpr StringData OplogApplication::kInitialSyncOplogApplicationMode; constexpr StringData OplogApplication::kRecoveringOplogApplicationMode; +constexpr StringData OplogApplication::kStableRecoveringOplogApplicationMode; +constexpr StringData OplogApplication::kUnstableRecoveringOplogApplicationMode; constexpr StringData OplogApplication::kSecondaryOplogApplicationMode; constexpr StringData OplogApplication::kApplyOpsCmdOplogApplicationMode; @@ -1117,8 +1149,10 @@ StringData OplogApplication::modeToString(OplogApplication::Mode mode) { switch (mode) { case OplogApplication::Mode::kInitialSync: return OplogApplication::kInitialSyncOplogApplicationMode; - case OplogApplication::Mode::kRecovering: - return OplogApplication::kRecoveringOplogApplicationMode; + case OplogApplication::Mode::kUnstableRecovering: + return OplogApplication::kUnstableRecoveringOplogApplicationMode; + case OplogApplication::Mode::kStableRecovering: + return OplogApplication::kStableRecoveringOplogApplicationMode; case OplogApplication::Mode::kSecondary: return OplogApplication::kSecondaryOplogApplicationMode; case OplogApplication::Mode::kApplyOpsCmd: @@ -1131,7 +1165,9 @@ StatusWith<OplogApplication::Mode> OplogApplication::parseMode(const std::string if (mode == OplogApplication::kInitialSyncOplogApplicationMode) { return OplogApplication::Mode::kInitialSync; } else if (mode == OplogApplication::kRecoveringOplogApplicationMode) { - return OplogApplication::Mode::kRecovering; + // This only being used in applyOps command which is controlled by the client, so it should + // be unstable. + return OplogApplication::Mode::kUnstableRecovering; } else if (mode == OplogApplication::kSecondaryOplogApplicationMode) { return OplogApplication::Mode::kSecondary; } else if (mode == OplogApplication::kApplyOpsCmdOplogApplicationMode) { @@ -1143,6 +1179,46 @@ StatusWith<OplogApplication::Mode> OplogApplication::parseMode(const std::string MONGO_UNREACHABLE; } +void OplogApplication::checkOnOplogFailureForRecovery(OperationContext* opCtx, + const mongo::NamespaceString& nss, + const mongo::BSONObj& oplogEntry, + const std::string& errorMsg) { + const bool isReplicaSet = + repl::ReplicationCoordinator::get(opCtx->getServiceContext())->getReplicationMode() == + repl::ReplicationCoordinator::modeReplSet; + // Relax the constraints of oplog application if the node is not a replica set member or the + // node is in the middle of a backup and restore process. + if (!isReplicaSet || storageGlobalParams.restore) { + return; + } + + // During the recovery process, certain configuration collections such as + // 'config.image_collections' are handled differently, which may result in encountering oplog + // application failures in common scenarios, and therefore assert statements are not used. + if (nss.isConfigDB()) { + LOGV2_DEBUG( + 5415002, + 1, + "Error applying operation while recovering from stable checkpoint. This is related to " + "one of the configuration collections so this error might be benign.", + "oplogEntry"_attr = oplogEntry, + "error"_attr = errorMsg); + } else if (getTestCommandsEnabled()) { + // Only fassert in test environment. + LOGV2_FATAL(5415000, + "Error applying operation while recovering from stable " + "checkpoint. This can lead to data corruption.", + "oplogEntry"_attr = oplogEntry, + "error"_attr = errorMsg); + } else { + LOGV2_WARNING(5415001, + "Error applying operation while recovering from stable " + "checkpoint. This can lead to data corruption.", + "oplogEntry"_attr = oplogEntry, + "error"_attr = errorMsg); + } +} + // @return failure status if an update should have happened and the document DNE. // See replset initial sync code. Status applyOperation_inlock(OperationContext* opCtx, @@ -1179,11 +1255,22 @@ Status applyOperation_inlock(OperationContext* opCtx, return Status::OK(); } + const bool inStableRecovery = mode == OplogApplication::Mode::kStableRecovering; NamespaceString requestNss; CollectionPtr collection = nullptr; if (auto uuid = op.getUuid()) { auto catalog = CollectionCatalog::get(opCtx); collection = catalog->lookupCollectionByUUID(opCtx, uuid.get()); + if (!collection && inStableRecovery) { + repl::OplogApplication::checkOnOplogFailureForRecovery( + opCtx, + op.getNss(), + redact(op.toBSONForLogging()), + str::stream() + << "(NamespaceNotFound): Failed to apply operation due to missing collection (" + << uuid.value() << ")"); + } + uassert(ErrorCodes::NamespaceNotFound, str::stream() << "Failed to apply operation due to missing collection (" << uuid.get() << "): " << redact(opOrGroupedInserts.toBSON()), @@ -1251,7 +1338,7 @@ Status applyOperation_inlock(OperationContext* opCtx, case ReplicationCoordinator::modeNone: { // Only assign timestamps on standalones during replication recovery when // started with the 'recoverFromOplogAsStandalone' flag. - return mode == OplogApplication::Mode::kRecovering; + return OplogApplication::inRecovering(mode); } } } @@ -1270,8 +1357,7 @@ Status applyOperation_inlock(OperationContext* opCtx, // correct pre-image for them. return collection && collection->isChangeStreamPreAndPostImagesEnabled() && isDataConsistent && - (mode == OplogApplication::Mode::kRecovering || - mode == OplogApplication::Mode::kSecondary) && + (OplogApplication::inRecovering(mode) || mode == OplogApplication::Mode::kSecondary) && !op.getFromMigrate().get_value_or(false) && !requestNss.isTemporaryReshardingCollection(); }; @@ -1415,6 +1501,9 @@ Status applyOperation_inlock(OperationContext* opCtx, if (oplogApplicationEnforcesSteadyStateConstraints) { return status; } + } else if (inStableRecovery) { + repl::OplogApplication::checkOnOplogFailureForRecovery( + opCtx, op.getNss(), redact(op.toBSONForLogging()), redact(status)); } // Continue to the next block to retry the operation as an upsert. needToDoUpsert = true; @@ -1672,6 +1761,10 @@ Status applyOperation_inlock(OperationContext* opCtx, }); if (!status.isOK()) { + if (inStableRecovery) { + repl::OplogApplication::checkOnOplogFailureForRecovery( + opCtx, op.getNss(), redact(op.toBSONForLogging()), redact(status)); + } return status; } @@ -1753,6 +1846,17 @@ Status applyOperation_inlock(OperationContext* opCtx, writeChangeStreamPreImage(opCtx, collection, op, *(result.requestedPreImage)); } + if (result.nDeleted == 0 && inStableRecovery) { + repl::OplogApplication::checkOnOplogFailureForRecovery( + opCtx, + op.getNss(), + redact(op.toBSONForLogging()), + !collection ? str::stream() + << "(NamespaceNotFound): Failed to apply operation due " + "to missing collection (" + << requestNss << ")" + : "Applied a delete which did not delete anything."s); + } // It is legal for a delete operation on the pre-images collection to delete zero // documents - pre-image collections are not guaranteed to contain the same set of // documents at all times. @@ -1898,7 +2002,7 @@ Status applyCommand_inlock(OperationContext* opCtx, case ReplicationCoordinator::modeNone: { // Only assign timestamps on standalones during replication recovery when // started with 'recoverFromOplogAsStandalone'. - return mode == OplogApplication::Mode::kRecovering; + return OplogApplication::inRecovering(mode); } } MONGO_UNREACHABLE; diff --git a/src/mongo/db/repl/oplog.h b/src/mongo/db/repl/oplog.h index c3c9c2de13a..1752c2e64e6 100644 --- a/src/mongo/db/repl/oplog.h +++ b/src/mongo/db/repl/oplog.h @@ -171,7 +171,10 @@ using IncrementOpsAppliedStatsFn = std::function<void()>; class OplogApplication { public: static constexpr StringData kInitialSyncOplogApplicationMode = "InitialSync"_sd; + // This only being used in 'applyOps' command when sent by client. static constexpr StringData kRecoveringOplogApplicationMode = "Recovering"_sd; + static constexpr StringData kStableRecoveringOplogApplicationMode = "StableRecovering"_sd; + static constexpr StringData kUnstableRecoveringOplogApplicationMode = "UnstableRecovering"_sd; static constexpr StringData kSecondaryOplogApplicationMode = "Secondary"_sd; static constexpr StringData kApplyOpsCmdOplogApplicationMode = "ApplyOps"_sd; @@ -180,9 +183,12 @@ public: kInitialSync, // Used when we are applying oplog operations to recover the database state following an - // unclean shutdown, or when we are recovering from the oplog after we rollback to a + // clean/unclean shutdown, or when we are recovering from the oplog after we rollback to a // checkpoint. - kRecovering, + // If recovering from a unstable stable checkpoint. + kUnstableRecovering, + // If recovering from a stable checkpoint.~ + kStableRecovering, // Used when a secondary node is applying oplog operations from the primary during steady // state replication. @@ -193,9 +199,20 @@ public: kApplyOpsCmd }; + static bool inRecovering(Mode mode) { + return mode == Mode::kUnstableRecovering || mode == Mode::kStableRecovering; + } + static StringData modeToString(Mode mode); static StatusWith<Mode> parseMode(const std::string& mode); + + // Server will crash on oplog application failure during recovery from stable checkpoint in the + // test environment. + static void checkOnOplogFailureForRecovery(OperationContext* opCtx, + const mongo::NamespaceString& nss, + const mongo::BSONObj& oplogEntry, + const std::string& errorMsg); }; inline std::ostream& operator<<(std::ostream& s, OplogApplication::Mode mode) { diff --git a/src/mongo/db/repl/oplog_applier.h b/src/mongo/db/repl/oplog_applier.h index b582b282efd..252070077a2 100644 --- a/src/mongo/db/repl/oplog_applier.h +++ b/src/mongo/db/repl/oplog_applier.h @@ -67,10 +67,10 @@ public: Options() = delete; explicit Options(OplogApplication::Mode inputMode) : mode(inputMode), - allowNamespaceNotFoundErrorsOnCrudOps( - inputMode == OplogApplication::Mode::kInitialSync || - inputMode == OplogApplication::Mode::kRecovering), - skipWritesToOplog(inputMode == OplogApplication::Mode::kRecovering) {} + allowNamespaceNotFoundErrorsOnCrudOps(inputMode == + OplogApplication::Mode::kInitialSync || + OplogApplication::inRecovering(inputMode)), + skipWritesToOplog(OplogApplication::inRecovering(inputMode)) {} // Used to determine which operations should be applied. Only initial sync will set this to // be something other than the null optime. diff --git a/src/mongo/db/repl/oplog_applier_impl.cpp b/src/mongo/db/repl/oplog_applier_impl.cpp index c5ca80bb8d6..4f8c3f9427e 100644 --- a/src/mongo/db/repl/oplog_applier_impl.cpp +++ b/src/mongo/db/repl/oplog_applier_impl.cpp @@ -38,7 +38,6 @@ #include "mongo/db/catalog/document_validation.h" #include "mongo/db/catalog_raii.h" #include "mongo/db/client.h" -#include "mongo/db/concurrency/write_conflict_exception.h" #include "mongo/db/db_raii.h" #include "mongo/db/logical_session_id.h" #include "mongo/db/repl/apply_ops.h" @@ -147,6 +146,9 @@ protected: // We have to use setMyLastAppliedOpTimeAndWallTimeForward since this thread races with // ReplicationExternalStateImpl::onTransitionToPrimary. _replCoord->setMyLastAppliedOpTimeAndWallTimeForward(newOpTimeAndWallTime); + // We know we're at a no-holes point and we've already advanced visibility; we need + // to notify waiters since we changed the lastAppliedSnapshot. + signalOplogWaiters(); } void _recordDurable(const OpTimeAndWallTime& newOpTimeAndWallTime) { diff --git a/src/mongo/db/repl/oplog_applier_impl_test.cpp b/src/mongo/db/repl/oplog_applier_impl_test.cpp index f9c41a19877..072c2a24464 100644 --- a/src/mongo/db/repl/oplog_applier_impl_test.cpp +++ b/src/mongo/db/repl/oplog_applier_impl_test.cpp @@ -44,7 +44,6 @@ #include "mongo/db/client.h" #include "mongo/db/commands/feature_compatibility_version_parser.h" #include "mongo/db/concurrency/d_concurrency.h" -#include "mongo/db/concurrency/write_conflict_exception.h" #include "mongo/db/curop.h" #include "mongo/db/db_raii.h" #include "mongo/db/dbdirectclient.h" @@ -422,11 +421,14 @@ TEST_F(OplogApplierImplTest, applyOplogEntryToRecordChangeStreamPreImages) { } }; generateTestCasesForOperations(OplogApplication::Mode::kSecondary, {}, true); - generateTestCasesForOperations(OplogApplication::Mode::kRecovering, {}, true); + generateTestCasesForOperations(OplogApplication::Mode::kUnstableRecovering, {}, true); + generateTestCasesForOperations(OplogApplication::Mode::kStableRecovering, {}, true); generateTestCasesForOperations(OplogApplication::Mode::kInitialSync, {}, false); const auto kFromMigrate{true}; generateTestCasesForOperations(OplogApplication::Mode::kSecondary, kFromMigrate, false); - generateTestCasesForOperations(OplogApplication::Mode::kRecovering, kFromMigrate, false); + generateTestCasesForOperations( + OplogApplication::Mode::kUnstableRecovering, kFromMigrate, false); + generateTestCasesForOperations(OplogApplication::Mode::kStableRecovering, kFromMigrate, false); generateTestCasesForOperations(OplogApplication::Mode::kInitialSync, kFromMigrate, false); int docId{0}; @@ -860,7 +862,7 @@ TEST_F(MultiOplogEntryOplogApplierImplTest, MultiApplyUnpreparedTransactionAllAt ReplicationCoordinator::get(_opCtx.get()), getConsistencyMarkers(), getStorageInterface(), - repl::OplogApplier::Options(repl::OplogApplication::Mode::kRecovering), + repl::OplogApplier::Options(repl::OplogApplication::Mode::kStableRecovering), _writerPool.get()); // Apply both inserts and the commit in a single batch. We expect no oplog entries to @@ -1342,7 +1344,7 @@ TEST_F(MultiOplogEntryPreparedTransactionTest, MultiApplyPreparedTransactionReco ReplicationCoordinator::get(_opCtx.get()), getConsistencyMarkers(), getStorageInterface(), - repl::OplogApplier::Options(repl::OplogApplication::Mode::kRecovering), + repl::OplogApplier::Options(repl::OplogApplication::Mode::kStableRecovering), _writerPool.get()); // Apply a batch with the insert operations. This should have no effect, because this is @@ -1576,7 +1578,7 @@ TEST_F(MultiOplogEntryPreparedTransactionTest, ReplicationCoordinator::get(_opCtx.get()), getConsistencyMarkers(), getStorageInterface(), - repl::OplogApplier::Options(repl::OplogApplication::Mode::kRecovering), + repl::OplogApplier::Options(repl::OplogApplication::Mode::kStableRecovering), _writerPool.get()); const auto expectedStartOpTime = _singlePrepareApplyOp->getOpTime(); diff --git a/src/mongo/db/repl/oplog_applier_impl_test_fixture.cpp b/src/mongo/db/repl/oplog_applier_impl_test_fixture.cpp index 949a69c2a29..4b09e68e901 100644 --- a/src/mongo/db/repl/oplog_applier_impl_test_fixture.cpp +++ b/src/mongo/db/repl/oplog_applier_impl_test_fixture.cpp @@ -32,7 +32,7 @@ #include "mongo/db/repl/oplog_applier_impl_test_fixture.h" #include "mongo/db/catalog/document_validation.h" -#include "mongo/db/concurrency/write_conflict_exception.h" +#include "mongo/db/concurrency/exception_util.h" #include "mongo/db/curop.h" #include "mongo/db/db_raii.h" #include "mongo/db/dbdirectclient.h" diff --git a/src/mongo/db/repl/oplog_applier_utils.cpp b/src/mongo/db/repl/oplog_applier_utils.cpp index 65f8298af89..5029e0b57d6 100644 --- a/src/mongo/db/repl/oplog_applier_utils.cpp +++ b/src/mongo/db/repl/oplog_applier_utils.cpp @@ -34,8 +34,9 @@ #include "mongo/db/catalog/collection.h" #include "mongo/db/catalog/collection_catalog.h" #include "mongo/db/catalog/document_validation.h" -#include "mongo/db/concurrency/write_conflict_exception.h" +#include "mongo/db/concurrency/exception_util.h" #include "mongo/db/curop.h" +#include "mongo/db/curop_metrics.h" #include "mongo/db/db_raii.h" #include "mongo/db/repl/oplog_applier_utils.h" #include "mongo/db/repl/repl_server_parameters_gen.h" @@ -199,9 +200,11 @@ Status OplogApplierUtils::applyOplogEntryOrGroupedInsertsCommon( OpCounters* opCounters) { invariant(DocumentValidationSettings::get(opCtx).isSchemaValidationDisabled()); - auto op = entryOrGroupedInserts.getOp(); // Count each log op application as a separate operation, for reporting purposes CurOp individualOp(opCtx); + ON_BLOCK_EXIT([opCtx]() { recordCurOpMetricsOplogApplication(opCtx); }); + + auto op = entryOrGroupedInserts.getOp(); const NamespaceString nss(op.getNss()); auto opType = op.getOpType(); if (opType == OpTypeEnum::kNoop) { @@ -304,6 +307,7 @@ Status OplogApplierUtils::applyOplogBatchCommon( InsertGroup insertGroup( ops, opCtx, oplogApplicationMode, isDataConsistent, applyOplogEntryOrGroupedInserts); + const bool inStableRecovery = oplogApplicationMode == OplogApplication::Mode::kStableRecovering; for (auto it = ops->cbegin(); it != ops->cend(); ++it) { const OplogEntry& entry = **it; @@ -323,9 +327,18 @@ Status OplogApplierUtils::applyOplogBatchCommon( if (!status.isOK()) { // Tried to apply an update operation but the document is missing, there must be // a delete operation for the document later in the oplog. + // Server will crash on oplog application failure during recovery from stable + // checkpoint in the test environment. if (status == ErrorCodes::UpdateOperationFailed && (oplogApplicationMode == OplogApplication::Mode::kInitialSync || - oplogApplicationMode == OplogApplication::Mode::kRecovering)) { + OplogApplication::inRecovering(oplogApplicationMode))) { + if (inStableRecovery) { + repl::OplogApplication::checkOnOplogFailureForRecovery( + opCtx, + entry.getNss(), + redact(entry.toBSONForLogging()), + redact(status)); + } continue; } @@ -339,8 +352,14 @@ Status OplogApplierUtils::applyOplogBatchCommon( } catch (const DBException& e) { // SERVER-24927 If we have a NamespaceNotFound exception, then this document will be // dropped before initial sync or recovery ends anyways and we should ignore it. + // Server will crash on oplog application failure during recovery from stable checkpoint + // in the test environment. if (e.code() == ErrorCodes::NamespaceNotFound && entry.isCrudOpType() && allowNamespaceNotFoundErrorsOnCrudOps) { + if (inStableRecovery) { + repl::OplogApplication::checkOnOplogFailureForRecovery( + opCtx, entry.getNss(), redact(entry.toBSONForLogging()), redact(e)); + } continue; } diff --git a/src/mongo/db/repl/oplog_buffer_collection_test.cpp b/src/mongo/db/repl/oplog_buffer_collection_test.cpp index f93278f45e3..a320cc56acc 100644 --- a/src/mongo/db/repl/oplog_buffer_collection_test.cpp +++ b/src/mongo/db/repl/oplog_buffer_collection_test.cpp @@ -33,7 +33,6 @@ #include "mongo/db/catalog/database.h" #include "mongo/db/client.h" -#include "mongo/db/concurrency/write_conflict_exception.h" #include "mongo/db/db_raii.h" #include "mongo/db/dbhelpers.h" #include "mongo/db/json.h" diff --git a/src/mongo/db/repl/oplog_entry.h b/src/mongo/db/repl/oplog_entry.h index 40092e47f4e..576537ad9a1 100644 --- a/src/mongo/db/repl/oplog_entry.h +++ b/src/mongo/db/repl/oplog_entry.h @@ -87,11 +87,13 @@ public: o.parseProtected(ctxt, bsonObject); return o; } - const BSONObj& getPreImageDocumentKey() const { - return _preImageDocumentKey; + + const BSONObj& getPostImageDocumentKey() const { + return _postImageDocumentKey; } - void setPreImageDocumentKey(BSONObj value) { - _preImageDocumentKey = std::move(value); + + void setPostImageDocumentKey(BSONObj value) { + _postImageDocumentKey = std::move(value); } const BSONObj& getPreImage() const { @@ -215,7 +217,8 @@ public: } private: - BSONObj _preImageDocumentKey; + // Stores the post image _id + shard key values. + BSONObj _postImageDocumentKey; // Used for storing the pre-image and post-image for the operation in-memory regardless of where // the images should be persisted. diff --git a/src/mongo/db/repl/oplog_fetcher.cpp b/src/mongo/db/repl/oplog_fetcher.cpp index 816b5f19fa6..4747e392542 100644 --- a/src/mongo/db/repl/oplog_fetcher.cpp +++ b/src/mongo/db/repl/oplog_fetcher.cpp @@ -719,8 +719,21 @@ StatusWith<OplogFetcher::Documents> OplogFetcher::_getNextBatch() { auto lastCommittedWithCurrentTerm = _dataReplicatorExternalState->getCurrentTermAndLastCommittedOpTime(); if (lastCommittedWithCurrentTerm.value != OpTime::kUninitializedTerm) { - _cursor->setCurrentTermAndLastCommittedOpTime(lastCommittedWithCurrentTerm.value, - lastCommittedWithCurrentTerm.opTime); + if (!_cursor->isExhaust() && lastCommittedWithCurrentTerm.opTime.isNull()) { + // For non-exhaust cursors, only set the lastKnownCommittedOpTime when it is not + // a null opTime. This is to avoid sending null opTime again and again and + // triggering oplog empty batches every single time in case we can't advance our + // commit point (e.g. during initial sync). + _cursor->setCurrentTermAndLastCommittedOpTime( + lastCommittedWithCurrentTerm.value, boost::none); + } else { + // For exhaust cursors, it is safe to set a null lastKnownCommittedOpTime in the + // initial getMore because the sync source will update the exhaust cursor's + // lastKnownCommittedOpTime to the commit point sent in the last response after + // each oplog batch. + _cursor->setCurrentTermAndLastCommittedOpTime( + lastCommittedWithCurrentTerm.value, lastCommittedWithCurrentTerm.opTime); + } } _cursor->more(); } diff --git a/src/mongo/db/repl/oplog_fetcher_test.cpp b/src/mongo/db/repl/oplog_fetcher_test.cpp index 96003452793..ebd897c7ed1 100644 --- a/src/mongo/db/repl/oplog_fetcher_test.cpp +++ b/src/mongo/db/repl/oplog_fetcher_test.cpp @@ -202,16 +202,21 @@ void validateGetMoreCommand(Message m, ASSERT_EQ(cursorId, msg.body.getIntField("getMore")); ASSERT_EQUALS(timeout, msg.body.getIntField("maxTimeMS")); - // In unittests, lastCommittedWithCurrentTerm should always be default to valid and non-null. + // In unittests, lastCommittedWithCurrentTerm.value should always be a valid term. // The case when currentTerm is kUninitializedTerm is tested separately in // GetMoreQueryDoesNotContainTermIfGetCurrentTermAndLastCommittedOpTimeReturnsUninitializedTerm. invariant(lastCommittedWithCurrentTerm.value != OpTime::kUninitializedTerm); - invariant(!lastCommittedWithCurrentTerm.opTime.isNull()); ASSERT_EQUALS(lastCommittedWithCurrentTerm.value, msg.body["term"].numberLong()); - ASSERT_EQUALS(lastCommittedWithCurrentTerm.opTime.getTimestamp(), - msg.body["lastKnownCommittedOpTime"]["ts"].timestamp()); - ASSERT_EQUALS(lastCommittedWithCurrentTerm.opTime.getTerm(), - msg.body["lastKnownCommittedOpTime"]["t"].numberLong()); + if (!exhaustSupported && lastCommittedWithCurrentTerm.opTime.isNull()) { + // Test that we don't attach the lastKnownCommittedOpTime field for non-exhaust cursors when + // the lastCommittedOpTime is null. + ASSERT_FALSE(msg.body.hasField("lastKnownCommittedOpTime")); + } else { + ASSERT_EQUALS(lastCommittedWithCurrentTerm.opTime.getTimestamp(), + msg.body["lastKnownCommittedOpTime"]["ts"].timestamp()); + ASSERT_EQUALS(lastCommittedWithCurrentTerm.opTime.getTerm(), + msg.body["lastKnownCommittedOpTime"]["t"].numberLong()); + } if (exhaustSupported) { ASSERT_TRUE(OpMsg::isFlagSet(m, OpMsg::kExhaustSupported)); @@ -1580,6 +1585,14 @@ TEST_F(OplogFetcherTest, OplogFetcherWorksWithoutExhaust) { // Update lastFetched before it is updated by getting the next batch. lastFetched = oplogFetcher->getLastOpTimeFetched_forTest(); + // Set a null lastCommittedOpTime to test that non-exhaust cursors don't attach a null + // lastKnownCommittedOpTime. This must be done before we issue the response to the find request + // so that the first getMore request (made immediately after processSingleRequestResponse) can + // pick this up. + dataReplicatorExternalState->lastCommittedOpTime = OpTime(); + auto firstGetMoreTermAndLastCommittedOpTime = + dataReplicatorExternalState->getCurrentTermAndLastCommittedOpTime(); + // Creating the cursor will succeed. After this, the cursor will be blocked on call() for the // getMore command. auto m = processSingleRequestResponse(oplogFetcher->getDBClientConnection_forTest(), @@ -1599,6 +1612,15 @@ TEST_F(OplogFetcherTest, OplogFetcherWorksWithoutExhaust) { auto fourthEntry = makeNoopOplogEntry({{Seconds(458), 0}, lastFetched.getTerm()}); auto secondBatch = {thirdEntry, fourthEntry}; + + // Reset the lastCommittedOpTime to non-null. This must be done before we issue the response to + // the first getMore request so that the second getMore request (made immediately after + // processSingleRequestResponse) can pick this up. + dataReplicatorExternalState->lastCommittedOpTime = {{9999, 0}, + dataReplicatorExternalState->currentTerm}; + auto secondGetMoreTermAndLastCommittedOpTime = + dataReplicatorExternalState->getCurrentTermAndLastCommittedOpTime(); + // moreToCome would be set to false if oplogFetcherUsesExhaust was set to false. After this, // the cursor will be blocked on call() for the next getMore command. m = processSingleRequestResponse( @@ -1609,7 +1631,7 @@ TEST_F(OplogFetcherTest, OplogFetcherWorksWithoutExhaust) { validateGetMoreCommand(m, cursorId, durationCount<Milliseconds>(oplogFetcher->getAwaitDataTimeout_forTest()), - dataReplicatorExternalState->getCurrentTermAndLastCommittedOpTime(), + firstGetMoreTermAndLastCommittedOpTime, false /* exhaustSupported */); // Update lastFetched since it should have been updated after getting the last batch. @@ -1629,7 +1651,7 @@ TEST_F(OplogFetcherTest, OplogFetcherWorksWithoutExhaust) { validateGetMoreCommand(m, cursorId, durationCount<Milliseconds>(oplogFetcher->getAwaitDataTimeout_forTest()), - dataReplicatorExternalState->getCurrentTermAndLastCommittedOpTime(), + secondGetMoreTermAndLastCommittedOpTime, false /* exhaustSupported */); // Update lastFetched since it should have been updated after getting the last batch. diff --git a/src/mongo/db/repl/primary_only_service.cpp b/src/mongo/db/repl/primary_only_service.cpp index 6a093ac6490..778376c9485 100644 --- a/src/mongo/db/repl/primary_only_service.cpp +++ b/src/mongo/db/repl/primary_only_service.cpp @@ -38,7 +38,6 @@ #include "mongo/db/auth/authorization_session.h" #include "mongo/db/catalog_raii.h" #include "mongo/db/client.h" -#include "mongo/db/concurrency/write_conflict_exception.h" #include "mongo/db/dbdirectclient.h" #include "mongo/db/ops/write_ops.h" #include "mongo/db/repl/repl_client_info.h" diff --git a/src/mongo/db/repl/primary_only_service_op_observer.h b/src/mongo/db/repl/primary_only_service_op_observer.h index 5f552149f98..76c225a483c 100644 --- a/src/mongo/db/repl/primary_only_service_op_observer.h +++ b/src/mongo/db/repl/primary_only_service_op_observer.h @@ -208,6 +208,10 @@ public: size_t numberOfPrePostImagesToWrite, Date_t wallClockTime) final {} + void onTransactionPrepareNonPrimary(OperationContext* opCtx, + const std::vector<repl::OplogEntry>& statements, + const repl::OpTime& prepareOpTime) final {} + void onTransactionAbort(OperationContext* opCtx, boost::optional<OplogSlot> abortOplogEntryOpTime) final {} diff --git a/src/mongo/db/repl/repl_server_parameters.idl b/src/mongo/db/repl/repl_server_parameters.idl index 227685978ee..3f4f5aac440 100644 --- a/src/mongo/db/repl/repl_server_parameters.idl +++ b/src/mongo/db/repl/repl_server_parameters.idl @@ -654,6 +654,17 @@ server_parameters: default: expr: 15 * 60 * 1000 + unsupportedSyncSource: + description: >- + **Not a supported feature**. Specifies the host/port for a node to use as a sync source. + It is a fatal error to specify a node that is not a part of the replica set config or to + specify the node itself. + set_at: startup + cpp_vartype: std::string + cpp_varname: unsupportedSyncSource + default: "" + validator: { callback: 'validateHostAndPort' } + feature_flags: featureFlagRetryableFindAndModify: description: >- diff --git a/src/mongo/db/repl/repl_set_commands.cpp b/src/mongo/db/repl/repl_set_commands.cpp index 2724219e764..23ca5952c0f 100644 --- a/src/mongo/db/repl/repl_set_commands.cpp +++ b/src/mongo/db/repl/repl_set_commands.cpp @@ -47,7 +47,6 @@ #include "mongo/db/auth/authorization_session.h" #include "mongo/db/commands.h" #include "mongo/db/commands/server_status_metric.h" -#include "mongo/db/concurrency/write_conflict_exception.h" #include "mongo/db/dbhelpers.h" #include "mongo/db/op_observer.h" #include "mongo/db/repl/drop_pending_collection_reaper.h" diff --git a/src/mongo/db/repl/repl_set_config_checks.cpp b/src/mongo/db/repl/repl_set_config_checks.cpp index 3be0e88e5d8..cb769f382f8 100644 --- a/src/mongo/db/repl/repl_set_config_checks.cpp +++ b/src/mongo/db/repl/repl_set_config_checks.cpp @@ -321,11 +321,21 @@ StatusWith<int> findSelfInConfig(ReplicationCoordinatorExternalState* externalSt for (ReplSetConfig::MemberIterator iter = newConfig.membersBegin(); iter != newConfig.membersEnd(); ++iter) { - if (externalState->isSelf(iter->getHostAndPort(), ctx)) { + if (externalState->isSelfFastPath(iter->getHostAndPort())) { meConfigs.push_back(iter); } } if (meConfigs.empty()) { + // No self-hosts were found using the fastpath; check with the slow path. + for (ReplSetConfig::MemberIterator iter = newConfig.membersBegin(); + iter != newConfig.membersEnd(); + ++iter) { + if (externalState->isSelfSlowPath(iter->getHostAndPort(), ctx, Seconds(30))) { + meConfigs.push_back(iter); + } + } + } + if (meConfigs.empty()) { return StatusWith<int>(ErrorCodes::NodeNotFound, str::stream() << "No host described in new configuration with " << newConfig.getConfigVersionAndTerm().toString() diff --git a/src/mongo/db/repl/repl_set_config_checks_test.cpp b/src/mongo/db/repl/repl_set_config_checks_test.cpp index 21cb645cf10..d491bdd9131 100644 --- a/src/mongo/db/repl/repl_set_config_checks_test.cpp +++ b/src/mongo/db/repl/repl_set_config_checks_test.cpp @@ -1288,6 +1288,60 @@ TEST_F(ServiceContextTest, FindSelfInConfig) { .getStatus()); } +TEST_F(ServiceContextTest, FindSelfInConfigFastAndSlow) { + ReplSetConfig newConfig; + newConfig = ReplSetConfig::parse(BSON("_id" + << "rs0" + << "version" << 2 << "protocolVersion" << 1 << "members" + << BSON_ARRAY(BSON("_id" << 1 << "host" + << "h1") + << BSON("_id" << 2 << "host" + << "h2") + << BSON("_id" << 3 << "host" + << "h3")))); + + { + // Present once only on the slow path, but fast enough to be found + ReplicationCoordinatorExternalStateMock presentOnceExternalState; + presentOnceExternalState.addSelfSlow(HostAndPort("h2"), Seconds(29)); + ASSERT_EQUALS(1, + unittest::assertGet(findSelfInConfig( + &presentOnceExternalState, newConfig, getServiceContext()))); + } + + { + // Present twice only on the slow path, but fast enough to be found both times. + ReplicationCoordinatorExternalStateMock presentTwiceExternalState; + presentTwiceExternalState.addSelfSlow(HostAndPort("h2"), Seconds(29)); + presentTwiceExternalState.addSelfSlow(HostAndPort("h3"), Seconds(29)); + ASSERT_EQUALS(ErrorCodes::InvalidReplicaSetConfig, + findSelfInConfig(&presentTwiceExternalState, newConfig, getServiceContext()) + .getStatus()); + } + + { + // Present once on the fast path, once on the slow path. This is expected to erroneously + // succeed, because we should not check the slow path if we got a unique result on the fast + // path. + ReplicationCoordinatorExternalStateMock presentFastAndSlowExternalState; + presentFastAndSlowExternalState.addSelf(HostAndPort("h2")); + presentFastAndSlowExternalState.addSelfSlow(HostAndPort("h3"), Seconds(29)); + ASSERT_EQUALS(1, + unittest::assertGet(findSelfInConfig( + &presentFastAndSlowExternalState, newConfig, getServiceContext()))); + } + + { + // Present only on the slow path, with a long timeout. This will fail. + ReplicationCoordinatorExternalStateMock presentLongTimeoutExternalState; + presentLongTimeoutExternalState.addSelfSlow(HostAndPort("h2"), Seconds(31)); + ASSERT_EQUALS( + ErrorCodes::NodeNotFound, + findSelfInConfig(&presentLongTimeoutExternalState, newConfig, getServiceContext()) + .getStatus()); + } +} + } // namespace } // namespace repl } // namespace mongo diff --git a/src/mongo/db/repl/replication_consistency_markers_impl.cpp b/src/mongo/db/repl/replication_consistency_markers_impl.cpp index dde621e310d..8e7994669de 100644 --- a/src/mongo/db/repl/replication_consistency_markers_impl.cpp +++ b/src/mongo/db/repl/replication_consistency_markers_impl.cpp @@ -37,7 +37,6 @@ #include "mongo/db/catalog_raii.h" #include "mongo/db/concurrency/d_concurrency.h" #include "mongo/db/concurrency/lock_state.h" -#include "mongo/db/concurrency/write_conflict_exception.h" #include "mongo/db/repl/optime.h" #include "mongo/db/repl/replication_coordinator.h" #include "mongo/db/repl/storage_interface.h" diff --git a/src/mongo/db/repl/replication_consistency_markers_impl_test.cpp b/src/mongo/db/repl/replication_consistency_markers_impl_test.cpp index 3d58ef476e4..27339c1e53c 100644 --- a/src/mongo/db/repl/replication_consistency_markers_impl_test.cpp +++ b/src/mongo/db/repl/replication_consistency_markers_impl_test.cpp @@ -35,7 +35,7 @@ #include "mongo/db/client.h" #include "mongo/db/concurrency/d_concurrency.h" -#include "mongo/db/concurrency/write_conflict_exception.h" +#include "mongo/db/concurrency/exception_util.h" #include "mongo/db/db_raii.h" #include "mongo/db/dbhelpers.h" #include "mongo/db/namespace_string.h" diff --git a/src/mongo/db/repl/replication_coordinator.h b/src/mongo/db/repl/replication_coordinator.h index 01966a7722a..a057bf5ad7b 100644 --- a/src/mongo/db/repl/replication_coordinator.h +++ b/src/mongo/db/repl/replication_coordinator.h @@ -380,6 +380,11 @@ public: * necessarily commit in sequential order. It is also used when we finish oplog batch * application on secondaries, to avoid any potential race conditions around setting the * applied optime from more than one thread. + * + * Since the last applied op time and wall time might not be visible (i.e. there may be + * "oplog holes" from oplog entries with earlier timestamps which commit after this one) + * this method does not notify oplog waiters. Callers which know the new lastApplied is at + * a no-holes point should call signalOplogWaiters after calling this method. */ virtual void setMyLastAppliedOpTimeAndWallTimeForward( const OpTimeAndWallTime& opTimeAndWallTime) = 0; diff --git a/src/mongo/db/repl/replication_coordinator_external_state.h b/src/mongo/db/repl/replication_coordinator_external_state.h index 9bf96582627..eec7564971c 100644 --- a/src/mongo/db/repl/replication_coordinator_external_state.h +++ b/src/mongo/db/repl/replication_coordinator_external_state.h @@ -146,6 +146,20 @@ public: virtual bool isSelf(const HostAndPort& host, ServiceContext* service) = 0; /** + * Returns true if "host" is one of the network identities of this node, without actually + * going out to the network and checking. + */ + virtual bool isSelfFastPath(const HostAndPort& host) = 0; + + /** + * Returns true if "host" is one of the network identities of this node, without + * checking the fast path first. + */ + virtual bool isSelfSlowPath(const HostAndPort& host, + ServiceContext* service, + Milliseconds timeout) = 0; + + /** * Gets the replica set config document from local storage, or returns an error. */ virtual StatusWith<BSONObj> loadLocalConfigDocument(OperationContext* opCtx) = 0; diff --git a/src/mongo/db/repl/replication_coordinator_external_state_impl.cpp b/src/mongo/db/repl/replication_coordinator_external_state_impl.cpp index 22cff587223..70c674500be 100644 --- a/src/mongo/db/repl/replication_coordinator_external_state_impl.cpp +++ b/src/mongo/db/repl/replication_coordinator_external_state_impl.cpp @@ -51,11 +51,10 @@ #include "mongo/db/commands/rwc_defaults_commands_gen.h" #include "mongo/db/commands/server_status_metric.h" #include "mongo/db/concurrency/d_concurrency.h" -#include "mongo/db/concurrency/write_conflict_exception.h" +#include "mongo/db/concurrency/exception_util.h" #include "mongo/db/db_raii.h" #include "mongo/db/dbdirectclient.h" #include "mongo/db/dbhelpers.h" -#include "mongo/db/free_mon/free_mon_mongod.h" #include "mongo/db/index_builds_coordinator.h" #include "mongo/db/jsobj.h" #include "mongo/db/kill_sessions_local.h" @@ -537,8 +536,6 @@ OpTime ReplicationCoordinatorExternalStateImpl::onTransitionToPrimary(OperationC IndexBuildsCoordinator::get(opCtx)->onStepUp(opCtx); - notifyFreeMonitoringOnTransitionToPrimary(); - // It is only necessary to check the system indexes on the first transition to primary. // On subsequent transitions to primary the indexes will have already been created. static std::once_flag verifySystemIndexesOnce; @@ -701,7 +698,16 @@ Status ReplicationCoordinatorExternalStateImpl::storeLocalLastVoteDocument( // don't want to have this process interrupted due to us stepping down, since we // want to be able to cast our vote for a new primary right away. Both the write's lock // acquisition and the "waitUntilDurable" lock acquisition must be uninterruptible. - UninterruptibleLockGuard noInterrupt(opCtx->lockState()); + // + // It is not safe to take an uninterruptible lock during STARTUP2, so we only take this lock + // if we are primary or secondary. We do not have the RSTL but that is OK because we never + // move in to STARTUP2 from PRIMARY or SECONDARY, so the consequence of a stale state is + // only that we don't take an uninterruptible lock when we should. + auto* replCoord = ReplicationCoordinator::get(opCtx); + + boost::optional<UninterruptibleLockGuard> noInterrupt; + if (replCoord->isInPrimaryOrSecondaryState_UNSAFE()) + noInterrupt.emplace(opCtx->lockState()); Status status = writeConflictRetry( opCtx, @@ -796,6 +802,16 @@ bool ReplicationCoordinatorExternalStateImpl::isSelf(const HostAndPort& host, Se return repl::isSelf(host, ctx); } +bool ReplicationCoordinatorExternalStateImpl::isSelfFastPath(const HostAndPort& host) { + return repl::isSelfFastPath(host); +} + +bool ReplicationCoordinatorExternalStateImpl::isSelfSlowPath(const HostAndPort& host, + ServiceContext* ctx, + Milliseconds timeout) { + return repl::isSelfSlowPath(host, ctx, timeout); +} + HostAndPort ReplicationCoordinatorExternalStateImpl::getClientHostAndPort( const OperationContext* opCtx) { return HostAndPort(opCtx->getClient()->clientAddress(true)); @@ -919,30 +935,43 @@ void ReplicationCoordinatorExternalStateImpl::_shardingOnTransitionToPrimaryHook PeriodicShardedIndexConsistencyChecker::get(_service).onStepUp(_service); TransactionCoordinatorService::get(_service)->onStepUp(opCtx); - } else if (ShardingState::get(opCtx)->enabled()) { - Status status = ShardingStateRecovery::recover(opCtx); - VectorClockMutable::get(opCtx)->recoverDirect(opCtx); - - // If the node is shutting down or it lost quorum just as it was becoming primary, don't - // run the sharding onStepUp machinery. The onStepDown counterpart to these methods is - // already idempotent, so the machinery will remain in the stepped down state. - if (ErrorCodes::isShutdownError(status.code()) || - ErrorCodes::isNotPrimaryError(status.code())) { - return; + } else if (serverGlobalParams.clusterRole == ClusterRole::ShardServer) { + if (ShardingState::get(opCtx)->enabled()) { + Status status = ShardingStateRecovery::recover(opCtx); + VectorClockMutable::get(opCtx)->recoverDirect(opCtx); + + // If the node is shutting down or it lost quorum just as it was becoming primary, don't + // run the sharding onStepUp machinery. The onStepDown counterpart to these methods is + // already idempotent, so the machinery will remain in the stepped down state. + if (ErrorCodes::isShutdownError(status.code()) || + ErrorCodes::isNotPrimaryError(status.code())) { + return; + } + fassert(40107, status); + + const auto configsvrConnStr = + Grid::get(opCtx)->shardRegistry()->getConfigShard()->getConnString(); + ShardingInitializationMongoD::get(opCtx)->updateShardIdentityConfigString( + opCtx, configsvrConnStr); + + CatalogCacheLoader::get(_service).onStepUp(); + ChunkSplitter::get(_service).onStepUp(); + PeriodicBalancerConfigRefresher::get(_service).onStepUp(_service); + TransactionCoordinatorService::get(_service)->onStepUp(opCtx); + + // Note, these must be done after the configOpTime is recovered via + // ShardingStateRecovery::recover above, because they may trigger filtering metadata + // refreshes which should use the recovered configOpTime. + migrationutil::resubmitRangeDeletionsOnStepUp(_service); + migrationutil::resumeMigrationCoordinationsOnStepUp(opCtx); + migrationutil::resumeMigrationRecipientsOnStepUp(opCtx); + + const bool scheduleAsyncRefresh = true; + resharding::clearFilteringMetadata(opCtx, scheduleAsyncRefresh); } - fassert(40107, status); - - const auto configsvrConnStr = - Grid::get(opCtx)->shardRegistry()->getConfigShard()->getConnString(); - ShardingInitializationMongoD::get(opCtx)->updateShardIdentityConfigString(opCtx, - configsvrConnStr); - - CatalogCacheLoader::get(_service).onStepUp(); - ChunkSplitter::get(_service).onStepUp(); - PeriodicBalancerConfigRefresher::get(_service).onStepUp(_service); - TransactionCoordinatorService::get(_service)->onStepUp(opCtx); - - // Create uuid index on config.rangeDeletions if needed + // The code above will only be executed after a stepdown happens, however the code below + // needs to be executed also on startup, and the enabled check might fail in shards during + // startup. Create uuid index on config.rangeDeletions if needed auto minKeyFieldName = RangeDeletionTask::kRangeFieldName + "." + ChunkRange::kMinKey; auto maxKeyFieldName = RangeDeletionTask::kRangeFieldName + "." + ChunkRange::kMaxKey; Status indexStatus = createIndexOnConfigCollection( @@ -965,16 +994,6 @@ void ReplicationCoordinatorExternalStateImpl::_shardingOnTransitionToPrimaryHook indexStatus.withContext("Failed to create index on config.rangeDeletions on " "shard's first transition to primary")); } - - // Note, these must be done after the configOpTime is recovered via - // ShardingStateRecovery::recover above, because they may trigger filtering metadata - // refreshes which should use the recovered configOpTime. - migrationutil::resubmitRangeDeletionsOnStepUp(_service); - migrationutil::resumeMigrationCoordinationsOnStepUp(opCtx); - migrationutil::resumeMigrationRecipientsOnStepUp(opCtx); - - const bool scheduleAsyncRefresh = true; - resharding::clearFilteringMetadata(opCtx, scheduleAsyncRefresh); } else { // unsharded if (auto validator = LogicalTimeValidator::get(_service)) { validator->enableKeyGenerator(opCtx, true); diff --git a/src/mongo/db/repl/replication_coordinator_external_state_impl.h b/src/mongo/db/repl/replication_coordinator_external_state_impl.h index 9a1e448f636..20eb8c61984 100644 --- a/src/mongo/db/repl/replication_coordinator_external_state_impl.h +++ b/src/mongo/db/repl/replication_coordinator_external_state_impl.h @@ -81,6 +81,10 @@ public: OpTime onTransitionToPrimary(OperationContext* opCtx) override; virtual void forwardSecondaryProgress(); virtual bool isSelf(const HostAndPort& host, ServiceContext* service); + bool isSelfFastPath(const HostAndPort& host) final; + bool isSelfSlowPath(const HostAndPort& host, + ServiceContext* service, + Milliseconds timeout) final; Status createLocalLastVoteCollection(OperationContext* opCtx) final; virtual StatusWith<BSONObj> loadLocalConfigDocument(OperationContext* opCtx); virtual Status storeLocalConfigDocument(OperationContext* opCtx, diff --git a/src/mongo/db/repl/replication_coordinator_external_state_mock.cpp b/src/mongo/db/repl/replication_coordinator_external_state_mock.cpp index 86aaa76c00f..5f449c1ec51 100644 --- a/src/mongo/db/repl/replication_coordinator_external_state_mock.cpp +++ b/src/mongo/db/repl/replication_coordinator_external_state_mock.cpp @@ -96,9 +96,24 @@ void ReplicationCoordinatorExternalStateMock::forwardSecondaryProgress() {} bool ReplicationCoordinatorExternalStateMock::isSelf(const HostAndPort& host, ServiceContext* const service) { + return sequenceContains(_selfHosts, host) || _selfHostsSlow.find(host) != _selfHostsSlow.end(); +} + +bool ReplicationCoordinatorExternalStateMock::isSelfFastPath(const HostAndPort& host) { return sequenceContains(_selfHosts, host); } +bool ReplicationCoordinatorExternalStateMock::isSelfSlowPath(const HostAndPort& host, + ServiceContext* const service, + Milliseconds timeout) { + if (sequenceContains(_selfHosts, host)) + return true; + auto iter = _selfHostsSlow.find(host); + if (iter == _selfHostsSlow.end()) + return false; + return iter->second <= timeout; +} + void ReplicationCoordinatorExternalStateMock::addSelf(const HostAndPort& host) { _selfHosts.push_back(host); } @@ -107,6 +122,11 @@ void ReplicationCoordinatorExternalStateMock::clearSelfHosts() { _selfHosts.clear(); } +void ReplicationCoordinatorExternalStateMock::addSelfSlow(const HostAndPort& host, + Milliseconds timeout) { + _selfHostsSlow.emplace(host, timeout); +} + HostAndPort ReplicationCoordinatorExternalStateMock::getClientHostAndPort( const OperationContext* opCtx) { return _clientHostAndPort; diff --git a/src/mongo/db/repl/replication_coordinator_external_state_mock.h b/src/mongo/db/repl/replication_coordinator_external_state_mock.h index ecd0f072fed..fd8327f4fee 100644 --- a/src/mongo/db/repl/replication_coordinator_external_state_mock.h +++ b/src/mongo/db/repl/replication_coordinator_external_state_mock.h @@ -73,6 +73,10 @@ public: OpTime onTransitionToPrimary(OperationContext* opCtx) override; virtual void forwardSecondaryProgress(); virtual bool isSelf(const HostAndPort& host, ServiceContext* service); + bool isSelfFastPath(const HostAndPort& host) final; + bool isSelfSlowPath(const HostAndPort& host, + ServiceContext* service, + Milliseconds timeout) final; virtual HostAndPort getClientHostAndPort(const OperationContext* opCtx); virtual StatusWith<BSONObj> loadLocalConfigDocument(OperationContext* opCtx); virtual Status storeLocalConfigDocument(OperationContext* opCtx, @@ -105,13 +109,20 @@ public: /** * Adds "host" to the list of hosts that this mock will match when responding to "isSelf" - * messages. + * messages, including "isSelfFastPath" and "isSelfSlowPath". */ void addSelf(const HostAndPort& host); /** + * Adds "host" to the list of hosts that this mock will match when responding to + * "isSelfSlowPath" messages with a timeout less than or equal to that given, + * but not "isSelfFastPath" messages. + */ + void addSelfSlow(const HostAndPort& host, Milliseconds timeout); + + /** * Remove all hosts from the list of hosts that this mock will match when responding to "isSelf" - * messages. + * messages. Clears both regular and slow hosts. */ void clearSelfHosts(); @@ -208,6 +219,7 @@ private: StatusWith<OpTime> _lastOpTime; StatusWith<Date_t> _lastWallTime; std::vector<HostAndPort> _selfHosts; + stdx::unordered_map<HostAndPort, Milliseconds> _selfHostsSlow; bool _canAcquireGlobalSharedLock; Status _storeLocalConfigDocumentStatus; Status _storeLocalLastVoteDocumentStatus; diff --git a/src/mongo/db/repl/replication_coordinator_impl.cpp b/src/mongo/db/repl/replication_coordinator_impl.cpp index 5112a683f48..fb39e3b97ae 100644 --- a/src/mongo/db/repl/replication_coordinator_impl.cpp +++ b/src/mongo/db/repl/replication_coordinator_impl.cpp @@ -292,6 +292,7 @@ InitialSyncerInterface::Options createInitialSyncerOptions( externalState](const OpTimeAndWallTime& opTimeAndWallTime) { // Note that setting the last applied opTime forward also advances the global timestamp. replCoord->setMyLastAppliedOpTimeAndWallTimeForward(opTimeAndWallTime); + signalOplogWaiters(); // The oplog application phase of initial sync starts timestamping writes, causing // WiredTiger to pin this data in memory. Advancing the oldest timestamp in step with the // last applied optime here will permit WiredTiger to evict this data as it sees fit. @@ -336,6 +337,16 @@ ReplicationCoordinatorImpl::ReplicationCoordinatorImpl( _readWriteAbility(std::make_unique<ReadWriteAbility>(!settings.usingReplSets())), _replicationProcess(replicationProcess), _storage(storage), + _handleLivenessTimeoutCallback(_replExecutor.get(), + [this](const executor::TaskExecutor::CallbackArgs& args) { + _handleLivenessTimeout(args); + }), + _handleElectionTimeoutCallback( + _replExecutor.get(), + [this](const executor::TaskExecutor::CallbackArgs&) { + _startElectSelfIfEligibleV1(StartElectionReasonEnum::kElectionTimeout); + }, + [this](int64_t limit) { return _nextRandomInt64_inlock(limit); }), _random(prngSeed) { _termShadow.store(OpTime::kUninitializedTerm); @@ -382,10 +393,7 @@ ReplSetConfig ReplicationCoordinatorImpl::getReplicaSetConfig_forTest() { Date_t ReplicationCoordinatorImpl::getElectionTimeout_forTest() const { stdx::lock_guard<Latch> lk(_mutex); - if (!_handleElectionTimeoutCbh.isValid()) { - return Date_t(); - } - return _handleElectionTimeoutWhen; + return _handleElectionTimeoutCallback.getNextCall(); } Milliseconds ReplicationCoordinatorImpl::getRandomizedElectionOffset_forTest() { @@ -522,8 +530,8 @@ bool ReplicationCoordinatorImpl::_startLoadLocalConfig( LOGV2(4280504, "Cleaning up any partially applied oplog batches & reading last op from oplog"); // Read the last op from the oplog after cleaning up any partially applied batches. - const auto stableTimestamp = boost::none; - _replicationProcess->getReplicationRecovery()->recoverFromOplog(opCtx, stableTimestamp); + auto stableTimestamp = + _replicationProcess->getReplicationRecovery()->recoverFromOplog(opCtx, boost::none); LOGV2(4280505, "Creating any necessary TenantMigrationAccessBlockers for unfinished migrations"); @@ -535,7 +543,9 @@ bool ReplicationCoordinatorImpl::_startLoadLocalConfig( tenant_migration_access_blocker::recoverTenantMigrationAccessBlockers(opCtx); LOGV2(4280506, "Reconstructing prepared transactions"); - reconstructPreparedTransactions(opCtx, OplogApplication::Mode::kRecovering); + reconstructPreparedTransactions(opCtx, + stableTimestamp ? OplogApplication::Mode::kStableRecovering + : OplogApplication::Mode::kUnstableRecovering); const auto lastOpTimeAndWallTimeResult = _externalState->loadLastOpTimeAndWallTime(opCtx); @@ -828,6 +838,7 @@ void ReplicationCoordinatorImpl::_initialSyncerCompletionFunction( const auto lastApplied = opTimeStatus.getValue(); _setMyLastAppliedOpTimeAndWallTime(lock, lastApplied, false); + signalOplogWaiters(); _topCoord->resetMaintenanceCount(); } @@ -1006,6 +1017,10 @@ bool ReplicationCoordinatorImpl::enterQuiesceModeIfSecondary(Milliseconds quiesc return false; } + // Cancel any ongoing election so that the node cannot become primary once in quiesce mode, + // and do not wait for cancellation to complete. + _cancelElectionIfNeeded(lk); + _inQuiesceMode = true; _quiesceDeadline = _replExecutor->now() + quiesceTime; @@ -1399,6 +1414,7 @@ void ReplicationCoordinatorImpl::setMyLastAppliedOpTimeAndWallTime( stdx::unique_lock<Latch> lock(_mutex); // The optime passed to this function is required to represent a consistent database state. _setMyLastAppliedOpTimeAndWallTime(lock, opTimeAndWallTime, false); + signalOplogWaiters(); _reportUpstream_inlock(std::move(lock)); } @@ -1472,9 +1488,6 @@ void ReplicationCoordinatorImpl::_setMyLastAppliedOpTimeAndWallTime( }, opTime); - // Notify the oplog waiters after updating the local snapshot. - signalOplogWaiters(); - if (opTime.isNull()) { return; } @@ -1858,7 +1871,7 @@ Status ReplicationCoordinatorImpl::_setLastOptime(WithLock lk, _wakeReadyWaiters(lk, std::max(args.appliedOpTime, args.durableOpTime)); } - _cancelAndRescheduleLivenessUpdate_inlock(args.memberId); + _rescheduleLivenessUpdate_inlock(args.memberId); return Status::OK(); } @@ -2569,33 +2582,28 @@ ReplicationCoordinatorImpl::AutoGetRstlForStepUpStepDown::AutoGetRstlForStepUpSt deadline = start + Seconds(rstlTimeout); // cap deadline } - try { - // Enqueues RSTL in X mode. - _rstlLock.emplace(_opCtx, MODE_X, ReplicationStateTransitionLockGuard::EnqueueOnly()); - - ON_BLOCK_EXIT([&] { _stopAndWaitForKillOpThread(); }); - _startKillOpThread(); - - // Wait for RSTL to be acquired. - _rstlLock->waitForLockUntil(deadline); - - } catch (const ExceptionFor<ErrorCodes::LockTimeout>&) { - if (rstlTimeout > 0 && Date_t::now() - start >= Seconds(rstlTimeout)) { - // Dump all locks to identify which thread(s) are holding RSTL. - getGlobalLockManager()->dump(); - - auto lockerInfo = - opCtx->lockState()->getLockerInfo(CurOp::get(opCtx)->getLockStatsBase()); - BSONObjBuilder lockRep; - lockerInfo->stats.report(&lockRep); - LOGV2_FATAL(5675600, - "Time out exceeded waiting for RSTL, stepUp/stepDown is not possible " - "thus calling abort() to allow cluster to progress.", - "lockRep"_attr = lockRep.obj()); + _rstlLock.emplace(_opCtx, MODE_X, ReplicationStateTransitionLockGuard::EnqueueOnly()); + + ON_BLOCK_EXIT([&] { _stopAndWaitForKillOpThread(); }); + _startKillOpThread(); + + // Wait for RSTL to be acquired. + _rstlLock->waitForLockUntil(deadline, [opCtx, rstlTimeout, start] { + if (rstlTimeout <= 0 || Date_t::now() - start < Seconds{rstlTimeout}) { + return; } - // Rethrow to keep processing as before at a higher layer. - throw; - } + + // Dump all locks to identify which thread(s) are holding RSTL. + getGlobalLockManager()->dump(); + + auto lockerInfo = opCtx->lockState()->getLockerInfo(CurOp::get(opCtx)->getLockStatsBase()); + BSONObjBuilder lockRep; + lockerInfo->stats.report(&lockRep); + LOGV2_FATAL(5675600, + "Time out exceeded waiting for RSTL, stepUp/stepDown is not possible thus " + "calling abort() to allow cluster to progress", + "lockRep"_attr = lockRep.obj()); + }); }; void ReplicationCoordinatorImpl::AutoGetRstlForStepUpStepDown::_startKillOpThread() { @@ -5647,6 +5655,8 @@ Status ReplicationCoordinatorImpl::processReplSetRequestVotes( LastVote lastVote{args.getTerm(), args.getCandidateIndex()}; Status status = _externalState->storeLocalLastVoteDocument(opCtx, lastVote); if (!status.isOK()) { + // Note the topology coordinator has already advanced its last vote at this point, + // so this node will not be able to vote in this election; this is a "spoiled" vote. LOGV2_ERROR(21428, "replSetRequestVotes failed to store LastVote document", "error"_attr = status); diff --git a/src/mongo/db/repl/replication_coordinator_impl.h b/src/mongo/db/repl/replication_coordinator_impl.h index fafe33221b9..2f62b14204a 100644 --- a/src/mongo/db/repl/replication_coordinator_impl.h +++ b/src/mongo/db/repl/replication_coordinator_impl.h @@ -38,6 +38,7 @@ #include "mongo/bson/timestamp.h" #include "mongo/db/concurrency/d_concurrency.h" #include "mongo/db/concurrency/replication_state_transition_lock_guard.h" +#include "mongo/db/repl/delayable_timeout_callback.h" #include "mongo/db/repl/initial_syncer.h" #include "mongo/db/repl/initial_syncer_interface.h" #include "mongo/db/repl/member_state.h" @@ -1146,6 +1147,13 @@ private: */ Milliseconds _getRandomizedElectionOffset_inlock(); + /* + * Return the upper bound of the offset amount returned by _getRandomizedElectionOffset + * This is actually off by one, that is, the election offset is in the half-open range + * [0, electionOffsetUpperBound) + */ + long long _getElectionOffsetUpperBound_inlock(); + /** * Starts a heartbeat for each member in the current config. Called while holding _mutex. */ @@ -1486,8 +1494,10 @@ private: /** * Bottom half of _scheduleNextLivenessUpdate. * Must be called with _mutex held. + * If reschedule is true, will recompute the liveness update even if a timeout is + * already pending. */ - void _scheduleNextLivenessUpdate_inlock(); + void _scheduleNextLivenessUpdate_inlock(bool reschedule); /** * Callback which marks downed nodes as down, triggers a stepdown if a majority of nodes are no @@ -1496,11 +1506,11 @@ private: void _handleLivenessTimeout(const executor::TaskExecutor::CallbackArgs& cbData); /** - * If "updatedMemberId" is the current _earliestMemberId, cancels the current - * _handleLivenessTimeout callback and calls _scheduleNextLivenessUpdate to schedule a new one. + * If "updatedMemberId" is the current _earliestMemberId, calls _scheduleNextLivenessUpdate to + * schedule a new one. * Returns immediately otherwise. */ - void _cancelAndRescheduleLivenessUpdate_inlock(int updatedMemberId); + void _rescheduleLivenessUpdate_inlock(int updatedMemberId); /** * Cancels all outstanding _priorityTakeover callbacks. @@ -1752,15 +1762,10 @@ private: stdx::condition_variable _currentCommittedSnapshotCond; // (M) // Callback Handle used to cancel a scheduled LivenessTimeout callback. - executor::TaskExecutor::CallbackHandle _handleLivenessTimeoutCbh; // (M) + DelayableTimeoutCallback _handleLivenessTimeoutCallback; // (S) - // Callback Handle used to cancel a scheduled ElectionTimeout callback. - executor::TaskExecutor::CallbackHandle _handleElectionTimeoutCbh; // (M) - - // Election timeout callback will not run before this time. - // If this date is Date_t(), the callback is either unscheduled or canceled. - // Used for testing only. - Date_t _handleElectionTimeoutWhen; // (M) + // Used to manage scheduling and canceling election timeouts. + DelayableTimeoutCallbackWithJitter _handleElectionTimeoutCallback; // (M) // Callback Handle used to cancel a scheduled PriorityTakeover callback. executor::TaskExecutor::CallbackHandle _priorityTakeoverCbh; // (M) diff --git a/src/mongo/db/repl/replication_coordinator_impl_elect_v1.cpp b/src/mongo/db/repl/replication_coordinator_impl_elect_v1.cpp index 0c05b9f4a0c..e87cf7f7837 100644 --- a/src/mongo/db/repl/replication_coordinator_impl_elect_v1.cpp +++ b/src/mongo/db/repl/replication_coordinator_impl_elect_v1.cpp @@ -46,6 +46,8 @@ namespace mongo { namespace repl { MONGO_FAIL_POINT_DEFINE(hangInWritingLastVoteForDryRun); +MONGO_FAIL_POINT_DEFINE(electionHangsBeforeUpdateMemberState); +MONGO_FAIL_POINT_DEFINE(hangBeforeOnVoteRequestCompleteCallback); class ReplicationCoordinatorImpl::ElectionState::LoseElectionGuardV1 { LoseElectionGuardV1(const LoseElectionGuardV1&) = delete; @@ -134,7 +136,12 @@ ReplicationCoordinatorImpl::ElectionState::getElectionDryRunFinishedEvent(WithLo void ReplicationCoordinatorImpl::ElectionState::cancel(WithLock) { _isCanceled = true; - _voteRequester->cancel(); + // This check is necessary because _voteRequester is only initialized in _startVoteRequester. + // Since we don't hold mutex during the entire election process, it is possible to get here + // before _startVoteRequester is ever called. + if (_voteRequester) { + _voteRequester->cancel(); + } } void ReplicationCoordinatorImpl::ElectionState::start(WithLock lk, StartElectionReasonEnum reason) { @@ -390,13 +397,16 @@ void ReplicationCoordinatorImpl::ElectionState::_requestVotesForRealElection( _replExecutor ->onEvent(nextPhaseEvh.getValue(), [=](const executor::TaskExecutor::CallbackArgs&) { + if (MONGO_unlikely(hangBeforeOnVoteRequestCompleteCallback.shouldFail())) { + LOGV2(7277400, + "Hang due to hangBeforeOnVoteRequestCompleteCallback failpoint"); + hangBeforeOnVoteRequestCompleteCallback.pauseWhileSet(); + } _onVoteRequestComplete(newTerm, reason); }) .status_with_transitional_ignore(); } -MONGO_FAIL_POINT_DEFINE(electionHangsBeforeUpdateMemberState); - void ReplicationCoordinatorImpl::ElectionState::_onVoteRequestComplete( long long newTerm, StartElectionReasonEnum reason) { stdx::lock_guard<Latch> lk(_repl->_mutex); diff --git a/src/mongo/db/repl/replication_coordinator_impl_elect_v1_test.cpp b/src/mongo/db/repl/replication_coordinator_impl_elect_v1_test.cpp index 84b45c40fa5..d71d606c656 100644 --- a/src/mongo/db/repl/replication_coordinator_impl_elect_v1_test.cpp +++ b/src/mongo/db/repl/replication_coordinator_impl_elect_v1_test.cpp @@ -227,7 +227,8 @@ TEST_F(ReplCoordTest, ElectionSucceedsWhenNodeIsTheOnlyElectableNode) { const auto opCtxPtr = makeOperationContext(); auto& opCtx = *opCtxPtr; - // Since we're still in drain mode, expect that we report ismaster: false, issecondary:true. + // Since we're still in drain mode, expect that we report isWritablePrimary:false, + // issecondary:true. auto helloResponse = getReplCoord()->awaitHelloResponse(opCtxPtr.get(), {}, boost::none, boost::none); ASSERT_FALSE(helloResponse->isWritablePrimary()) << helloResponse->toBSON().toString(); @@ -284,7 +285,8 @@ TEST_F(ReplCoordTest, ElectionSucceedsWhenNodeIsTheOnlyNode) { const auto opCtxPtr = makeOperationContext(); auto& opCtx = *opCtxPtr; - // Since we're still in drain mode, expect that we report ismaster: false, issecondary:true. + // Since we're still in drain mode, expect that we report isWritablePrimary:false, + // issecondary:true. auto helloResponse = getReplCoord()->awaitHelloResponse(opCtxPtr.get(), {}, boost::none, boost::none); ASSERT_FALSE(helloResponse->isWritablePrimary()) << helloResponse->toBSON().toString(); diff --git a/src/mongo/db/repl/replication_coordinator_impl_heartbeat.cpp b/src/mongo/db/repl/replication_coordinator_impl_heartbeat.cpp index 060f4d6fc9b..98a6f722a73 100644 --- a/src/mongo/db/repl/replication_coordinator_impl_heartbeat.cpp +++ b/src/mongo/db/repl/replication_coordinator_impl_heartbeat.cpp @@ -82,11 +82,13 @@ MONGO_FAIL_POINT_DEFINE(waitForPostActionCompleteInHbReconfig); using executor::RemoteCommandRequest; -Milliseconds ReplicationCoordinatorImpl::_getRandomizedElectionOffset_inlock() { +long long ReplicationCoordinatorImpl::_getElectionOffsetUpperBound_inlock() { long long electionTimeout = durationCount<Milliseconds>(_rsConfig.getElectionTimeoutPeriod()); - long long randomOffsetUpperBound = - electionTimeout * _externalState->getElectionTimeoutOffsetLimitFraction(); + return electionTimeout * _externalState->getElectionTimeoutOffsetLimitFraction(); +} +Milliseconds ReplicationCoordinatorImpl::_getRandomizedElectionOffset_inlock() { + long long randomOffsetUpperBound = _getElectionOffsetUpperBound_inlock(); // Avoid divide by zero error in random number generator. if (randomOffsetUpperBound == 0) { return Milliseconds(0); @@ -1044,9 +1046,7 @@ void ReplicationCoordinatorImpl::_cancelHeartbeats_inlock() { // Heartbeat callbacks will remove themselves from _heartbeatHandles when they execute with // CallbackCanceled status, so it's better to leave the handles in the list, for now. - if (_handleLivenessTimeoutCbh.isValid()) { - _replExecutor->cancel(_handleLivenessTimeoutCbh); - } + _handleLivenessTimeoutCallback.cancel(); } void ReplicationCoordinatorImpl::restartScheduledHeartbeats_forTest() { @@ -1095,16 +1095,12 @@ void ReplicationCoordinatorImpl::_startHeartbeats_inlock() { _topCoord->restartHeartbeat(now, target); } - _scheduleNextLivenessUpdate_inlock(); + _scheduleNextLivenessUpdate_inlock(/* reschedule = */ false); } void ReplicationCoordinatorImpl::_handleLivenessTimeout( const executor::TaskExecutor::CallbackArgs& cbData) { stdx::unique_lock<Latch> lk(_mutex); - // Only reset the callback handle if it matches, otherwise more will be coming through - if (cbData.myHandle == _handleLivenessTimeoutCbh) { - _handleLivenessTimeoutCbh = CallbackHandle(); - } if (!cbData.status.isOK()) { return; } @@ -1116,10 +1112,10 @@ void ReplicationCoordinatorImpl::_handleLivenessTimeout( lk = _handleHeartbeatResponseAction_inlock( action, StatusWith(ReplSetHeartbeatResponse()), std::move(lk)); - _scheduleNextLivenessUpdate_inlock(); + _scheduleNextLivenessUpdate_inlock(/* reschedule = */ false); } -void ReplicationCoordinatorImpl::_scheduleNextLivenessUpdate_inlock() { +void ReplicationCoordinatorImpl::_scheduleNextLivenessUpdate_inlock(bool reschedule) { // Scan liveness table for earliest date; schedule a run at (that date plus election // timeout). Date_t earliestDate; @@ -1132,7 +1128,7 @@ void ReplicationCoordinatorImpl::_scheduleNextLivenessUpdate_inlock() { return; } - if (_handleLivenessTimeoutCbh.isValid() && !_handleLivenessTimeoutCbh.isCanceled()) { + if (!reschedule && _handleLivenessTimeoutCallback.isActive()) { // don't bother to schedule; one is already scheduled and pending. return; } @@ -1145,31 +1141,21 @@ void ReplicationCoordinatorImpl::_scheduleNextLivenessUpdate_inlock() { "nextTimeout"_attr = nextTimeout); // It is possible we will schedule the next timeout in the past. - // ThreadPoolTaskExecutor::_scheduleWorkAt() schedules its work immediately if it's given a - // time <= now(). + // DelayableTimeoutCallback schedules its work immediately if it's given a time <= now(). // If we missed the timeout, it means that on our last check the earliest live member was // just barely fresh and it has become stale since then. We must schedule another liveness // check to continue conducting liveness checks and be able to step down from primary if we // lose contact with a majority of nodes. - auto cbh = - _scheduleWorkAt(nextTimeout, [=](const executor::TaskExecutor::CallbackArgs& cbData) { - _handleLivenessTimeout(cbData); - }); - if (!cbh) { - return; - } - _handleLivenessTimeoutCbh = cbh; + // We ignore shutdown errors; any other error triggers an fassert. + _handleLivenessTimeoutCallback.delayUntil(nextTimeout).ignore(); _earliestMemberId = earliestMemberId.getData(); } -void ReplicationCoordinatorImpl::_cancelAndRescheduleLivenessUpdate_inlock(int updatedMemberId) { +void ReplicationCoordinatorImpl::_rescheduleLivenessUpdate_inlock(int updatedMemberId) { if ((_earliestMemberId != -1) && (_earliestMemberId != updatedMemberId)) { return; } - if (_handleLivenessTimeoutCbh.isValid()) { - _replExecutor->cancel(_handleLivenessTimeoutCbh); - } - _scheduleNextLivenessUpdate_inlock(); + _scheduleNextLivenessUpdate_inlock(/* reschedule = */ true); } void ReplicationCoordinatorImpl::_cancelPriorityTakeover_inlock() { @@ -1203,7 +1189,8 @@ void ReplicationCoordinatorImpl::_cancelAndRescheduleElectionTimeout_inlock() { // the logs. int cancelAndRescheduleLogLevel = 5; static auto logThrottleTime = _replExecutor->now(); - const bool wasActive = _handleElectionTimeoutCbh.isValid(); + auto oldWhen = _handleElectionTimeoutCallback.getNextCall(); + const bool wasActive = oldWhen != Date_t(); auto now = _replExecutor->now(); const bool doNotReschedule = _inShutdown || !_memberState.secondary() || _selfIndex < 0 || !_rsConfig.getMemberAt(_selfIndex).isElectable(); @@ -1212,48 +1199,40 @@ void ReplicationCoordinatorImpl::_cancelAndRescheduleElectionTimeout_inlock() { cancelAndRescheduleLogLevel = 4; logThrottleTime = now; } - if (wasActive) { + if (wasActive && doNotReschedule) { LOGV2_FOR_ELECTION(4615649, cancelAndRescheduleLogLevel, "Canceling election timeout callback at {when}", "Canceling election timeout callback", - "when"_attr = _handleElectionTimeoutWhen); - _replExecutor->cancel(_handleElectionTimeoutCbh); - _handleElectionTimeoutCbh = CallbackHandle(); - _handleElectionTimeoutWhen = Date_t(); + "when"_attr = oldWhen); + _handleElectionTimeoutCallback.cancel(); } if (doNotReschedule) return; - Milliseconds randomOffset = _getRandomizedElectionOffset_inlock(); - auto when = now + _rsConfig.getElectionTimeoutPeriod() + randomOffset; - invariant(when > now); + Milliseconds upperBound = Milliseconds(_getElectionOffsetUpperBound_inlock()); + auto requestedWhen = now + _rsConfig.getElectionTimeoutPeriod(); + invariant(requestedWhen > now); + Status delayStatus = + _handleElectionTimeoutCallback.delayUntilWithJitter(requestedWhen, upperBound); + Date_t when = _handleElectionTimeoutCallback.getNextCall(); if (wasActive) { // The log level here is 4 once per second, otherwise 5. LOGV2_FOR_ELECTION(4615650, cancelAndRescheduleLogLevel, - "Rescheduling election timeout callback at {when}", - "Rescheduling election timeout callback", - "when"_attr = when); + "Rescheduled election timeout callback", + "when"_attr = when, + "requestedWhen"_attr = requestedWhen, + "error"_attr = delayStatus); } else { LOGV2_FOR_ELECTION(4615651, 4, - "Scheduling election timeout callback at {when}", - "Scheduling election timeout callback", - "when"_attr = when); + "Scheduled election timeout callback", + "when"_attr = when, + "requestedWhen"_attr = requestedWhen, + "error"_attr = delayStatus); } - _handleElectionTimeoutWhen = when; - _handleElectionTimeoutCbh = - _scheduleWorkAt(when, [=](const mongo::executor::TaskExecutor::CallbackArgs& cbData) { - stdx::lock_guard<Latch> lk(_mutex); - if (_handleElectionTimeoutCbh == cbData.myHandle) { - // This lets _cancelAndRescheduleElectionTimeout_inlock know the callback - // has happened. - _handleElectionTimeoutCbh = CallbackHandle(); - } - _startElectSelfIfEligibleV1(lk, StartElectionReasonEnum::kElectionTimeout); - }); } void ReplicationCoordinatorImpl::_startElectSelfIfEligibleV1(StartElectionReasonEnum reason) { @@ -1277,7 +1256,7 @@ void ReplicationCoordinatorImpl::_startElectSelfIfEligibleV1(WithLock lk, _cancelCatchupTakeover_inlock(); _cancelPriorityTakeover_inlock(); _cancelAndRescheduleElectionTimeout_inlock(); - if (_inShutdown) { + if (_inShutdown || _inQuiesceMode) { LOGV2_FOR_ELECTION(4615654, 0, "Not starting an election, since we are shutting down"); return; } diff --git a/src/mongo/db/repl/replication_coordinator_impl_test.cpp b/src/mongo/db/repl/replication_coordinator_impl_test.cpp index 2a106fca9c0..6eb7e7c4458 100644 --- a/src/mongo/db/repl/replication_coordinator_impl_test.cpp +++ b/src/mongo/db/repl/replication_coordinator_impl_test.cpp @@ -136,7 +136,7 @@ std::shared_ptr<const repl::HelloResponse> awaitHelloWithNewOpCtx( return replCoord->awaitHelloResponse(newOpCtx.get(), horizonParams, topologyVersion, deadline); } -TEST_F(ReplCoordTest, IsMasterIsFalseDuringStepdown) { +TEST_F(ReplCoordTest, IsWritablePrimaryFalseDuringStepdown) { BSONObj configObj = BSON("_id" << "mySet" << "version" << 1 << "members" @@ -161,13 +161,13 @@ TEST_F(ReplCoordTest, IsMasterIsFalseDuringStepdown) { replCoord->updateTerm_forTest(replCoord->getTerm() + 1, &updateTermResult); ASSERT(TopologyCoordinator::UpdateTermResult::kTriggerStepDown == updateTermResult); - // Test that "ismaster" is immediately false, although "secondary" is not yet true. + // Test that "isWritablePrimary" is immediately false, although "secondary" is not yet true. auto opCtx = makeOperationContext(); const auto response = getReplCoord()->awaitHelloResponse(opCtx.get(), {}, boost::none, boost::none); ASSERT_TRUE(response->isConfigSet()); - BSONObj responseObj = response->toBSON(); - ASSERT_FALSE(responseObj["ismaster"].Bool()); + BSONObj responseObj = response->toBSON(false /*useLegacyResponseFields*/); + ASSERT_FALSE(responseObj["isWritablePrimary"].Bool()); ASSERT_FALSE(responseObj["secondary"].Bool()); ASSERT_FALSE(responseObj.hasField("isreplicaset")); @@ -3517,10 +3517,10 @@ TEST_F(ReplCoordTest, AwaitHelloResponseReturnsOnStepDown) { responseAfterDisablingWrites->getTopologyVersion(); ASSERT_EQUALS(topologyVersionAfterDisablingWrites->getCounter(), expectedCounter); ASSERT_EQUALS(topologyVersionAfterDisablingWrites->getProcessId(), expectedProcessId); - // We expect the server to increment the TopologyVersion and respond to waiting hellos - // once we disable writes on the node that is stepping down from primary. At this time, - // the 'ismaster' response field will be false but the node will have yet to transition to - // secondary. + // We expect the server to increment the TopologyVersion and respond to waiting hellos once + // we disable writes on the node that is stepping down from primary. At this time, the + // 'isWritablePrimary' response field will be false but the node will have yet to transition + // to secondary. ASSERT_FALSE(responseAfterDisablingWrites->isWritablePrimary()); ASSERT_FALSE(responseAfterDisablingWrites->isSecondary()); ASSERT_EQUALS(responseAfterDisablingWrites->getPrimary().host(), "node1"); @@ -5036,8 +5036,8 @@ TEST_F(ReplCoordTest, HelloResponseMentionsLackOfReplicaSetConfig) { const auto response = getReplCoord()->awaitHelloResponse(opCtx.get(), {}, boost::none, boost::none); ASSERT_FALSE(response->isConfigSet()); - BSONObj responseObj = response->toBSON(); - ASSERT_FALSE(responseObj["ismaster"].Bool()); + BSONObj responseObj = response->toBSON(false /*useLegacyResponseFields*/); + ASSERT_FALSE(responseObj["isWritablePrimary"].Bool()); ASSERT_FALSE(responseObj["secondary"].Bool()); ASSERT_TRUE(responseObj["isreplicaset"].Bool()); ASSERT_EQUALS("Does not have a valid replica set config", responseObj["info"].String()); @@ -6924,15 +6924,14 @@ TEST_F(ReplCoordTest, CancelAndRescheduleElectionTimeoutLogging) { // Setting mode to secondary should schedule the election timeout. ReplicationCoordinatorImpl* replCoord = getReplCoord(); ASSERT_OK(replCoord->setFollowerMode(MemberState::RS_SECONDARY)); - ASSERT_EQ(1, countTextFormatLogLinesContaining("Scheduling election timeout callback")); - ASSERT_EQ(0, countTextFormatLogLinesContaining("Rescheduling election timeout callback")); + ASSERT_EQ(1, countTextFormatLogLinesContaining("Scheduled election timeout callback")); + ASSERT_EQ(0, countTextFormatLogLinesContaining("Rescheduled election timeout callback")); ASSERT_EQ(0, countTextFormatLogLinesContaining("Canceling election timeout callback")); // Scheduling again should produce the "rescheduled", not the "scheduled", message . replCoord->cancelAndRescheduleElectionTimeout(); - ASSERT_EQ(1, countTextFormatLogLinesContaining("Scheduling election timeout callback")); - ASSERT_EQ(1, countTextFormatLogLinesContaining("Rescheduling election timeout callback")); - ASSERT_EQ(1, countTextFormatLogLinesContaining("Canceling election timeout callback")); + ASSERT_EQ(1, countTextFormatLogLinesContaining("Scheduled election timeout callback")); + ASSERT_EQ(1, countTextFormatLogLinesContaining("Rescheduled election timeout callback")); auto net = getNet(); net->enterNetwork(); @@ -6960,9 +6959,8 @@ TEST_F(ReplCoordTest, CancelAndRescheduleElectionTimeoutLogging) { net->exitNetwork(); // The election should have scheduled (not rescheduled) another timeout. - ASSERT_EQ(2, countTextFormatLogLinesContaining("Scheduling election timeout callback")); - ASSERT_EQ(1, countTextFormatLogLinesContaining("Rescheduling election timeout callback")); - ASSERT_EQ(1, countTextFormatLogLinesContaining("Canceling election timeout callback")); + ASSERT_EQ(2, countTextFormatLogLinesContaining("Scheduled election timeout callback")); + ASSERT_EQ(1, countTextFormatLogLinesContaining("Rescheduled election timeout callback")); auto replElectionReducedSeverityGuard = unittest::MinimumLoggedSeverityGuard{ logv2::LogComponent::kReplicationElection, logv2::LogSeverity::Debug(4)}; @@ -6973,9 +6971,8 @@ TEST_F(ReplCoordTest, CancelAndRescheduleElectionTimeoutLogging) { replCoord->cancelAndRescheduleElectionTimeout(); // We should not see this reschedule because it should be at log level 5. - ASSERT_EQ(2, countTextFormatLogLinesContaining("Scheduling election timeout callback")); - ASSERT_EQ(1, countTextFormatLogLinesContaining("Rescheduling election timeout callback")); - ASSERT_EQ(1, countTextFormatLogLinesContaining("Canceling election timeout callback")); + ASSERT_EQ(2, countTextFormatLogLinesContaining("Scheduled election timeout callback")); + ASSERT_EQ(1, countTextFormatLogLinesContaining("Rescheduled election timeout callback")); net->enterNetwork(); until = electionTimeoutWhen + Milliseconds(1001); @@ -6986,9 +6983,8 @@ TEST_F(ReplCoordTest, CancelAndRescheduleElectionTimeoutLogging) { stopCapturingLogMessages(); // We should see this reschedule at level 4 because it has been over 1 sec since we logged // at level 4. - ASSERT_EQ(2, countTextFormatLogLinesContaining("Scheduling election timeout callback")); - ASSERT_EQ(2, countTextFormatLogLinesContaining("Rescheduling election timeout callback")); - ASSERT_EQ(2, countTextFormatLogLinesContaining("Canceling election timeout callback")); + ASSERT_EQ(2, countTextFormatLogLinesContaining("Scheduled election timeout callback")); + ASSERT_EQ(2, countTextFormatLogLinesContaining("Rescheduled election timeout callback")); } TEST_F(ReplCoordTest, ZeroCommittedSnapshotAfterClearingCommittedSnapshot) { diff --git a/src/mongo/db/repl/replication_info.cpp b/src/mongo/db/repl/replication_info.cpp index 2172307d791..b9ed271ff77 100644 --- a/src/mongo/db/repl/replication_info.cpp +++ b/src/mongo/db/repl/replication_info.cpp @@ -248,11 +248,11 @@ public: BSONObjBuilder result; result.append("latestOptime", replCoord->getMyLastAppliedOpTime().getTimestamp()); - auto earliestOplogTimestampFetch = [&]() -> StatusWith<Timestamp> { + auto earliestOplogTimestampFetch = [&]() -> Timestamp { auto oplog = CollectionCatalog::get(opCtx)->lookupCollectionByNamespaceForRead( opCtx, NamespaceString::kRsOplogNamespace); if (!oplog) { - return StatusWith<Timestamp>(ErrorCodes::NamespaceNotFound, "oplog doesn't exist"); + return Timestamp(); } // Try to get the lock. If it's already locked, immediately return null timestamp. @@ -282,13 +282,13 @@ public: return o["ts"].timestamp(); } } - - return swEarliestOplogTimestamp; + if (!swEarliestOplogTimestamp.isOK()) { + return Timestamp(); + } + return swEarliestOplogTimestamp.getValue(); }(); - uassert( - 17347, "Problem reading earliest entry from oplog", earliestOplogTimestampFetch.isOK()); - result.append("earliestOptime", earliestOplogTimestampFetch.getValue()); + result.append("earliestOptime", earliestOplogTimestampFetch); return result.obj(); } diff --git a/src/mongo/db/repl/replication_recovery.cpp b/src/mongo/db/repl/replication_recovery.cpp index 80388fd2f3e..276d877ca67 100644 --- a/src/mongo/db/repl/replication_recovery.cpp +++ b/src/mongo/db/repl/replication_recovery.cpp @@ -334,7 +334,7 @@ void ReplicationRecoveryImpl::recoverFromOplogAsStandalone(OperationContext* opC // Initialize the cached pointer to the oplog collection. acquireOplogCollectionForLogging(opCtx); - + boost::optional<Timestamp> stableTimestamp = boost::none; if (recoveryTS || startupRecoveryForRestore) { if (startupRecoveryForRestore && !recoveryTS) { LOGV2_WARNING(5576601, @@ -345,8 +345,7 @@ void ReplicationRecoveryImpl::recoverFromOplogAsStandalone(OperationContext* opC // We pass in "none" for the stable timestamp so that recoverFromOplog asks storage // for the recoveryTimestamp just like on replica set recovery. - const auto stableTimestamp = boost::none; - recoverFromOplog(opCtx, stableTimestamp); + stableTimestamp = recoverFromOplog(opCtx, boost::none); } else { if (gTakeUnstableCheckpointOnShutdown) { // Ensure 'recoverFromOplogAsStandalone' with 'takeUnstableCheckpointOnShutdown' @@ -366,7 +365,10 @@ void ReplicationRecoveryImpl::recoverFromOplogAsStandalone(OperationContext* opC if (!_duringInitialSync) { // Initial sync will reconstruct prepared transactions when it is completely done. - reconstructPreparedTransactions(opCtx, OplogApplication::Mode::kRecovering); + reconstructPreparedTransactions(opCtx, + stableTimestamp + ? OplogApplication::Mode::kStableRecovering + : OplogApplication::Mode::kUnstableRecovering); // Two-phase index builds are built in the background, which may still be in-progress after // recovering from the oplog. To prevent crashing the server, skip enabling read-only mode. @@ -438,14 +440,14 @@ void ReplicationRecoveryImpl::recoverFromOplogUpTo(OperationContext* opCtx, Time invariant(appliedUpTo <= endPoint); } - reconstructPreparedTransactions(opCtx, OplogApplication::Mode::kRecovering); + reconstructPreparedTransactions(opCtx, OplogApplication::Mode::kStableRecovering); } -void ReplicationRecoveryImpl::recoverFromOplog(OperationContext* opCtx, - boost::optional<Timestamp> stableTimestamp) try { +boost::optional<Timestamp> ReplicationRecoveryImpl::recoverFromOplog( + OperationContext* opCtx, boost::optional<Timestamp> stableTimestamp) try { if (_consistencyMarkers->getInitialSyncFlag(opCtx)) { LOGV2(21542, "No recovery needed. Initial sync flag set"); - return; // Initial Sync will take over so no cleanup is needed. + return stableTimestamp; // Initial Sync will take over so no cleanup is needed. } const auto serviceCtx = getGlobalServiceContext(); @@ -487,7 +489,7 @@ void ReplicationRecoveryImpl::recoverFromOplog(OperationContext* opCtx, // Oplog is empty. There are no oplog entries to apply, so we exit recovery and go into // initial sync. LOGV2(21543, "No oplog entries to apply for recovery. Oplog is empty"); - return; + return stableTimestamp; } fassert(40290, topOfOplogSW); const auto topOfOplog = topOfOplogSW.getValue(); @@ -501,6 +503,7 @@ void ReplicationRecoveryImpl::recoverFromOplog(OperationContext* opCtx, _recoverFromUnstableCheckpoint( opCtx, _consistencyMarkers->getAppliedThrough(opCtx), topOfOplog); } + return stableTimestamp; } catch (...) { LOGV2_FATAL_CONTINUE(21570, "Caught exception during replication recovery: {error}", @@ -713,6 +716,10 @@ Timestamp ReplicationRecoveryImpl::_applyOplogOperations(OperationContext* opCtx RecoveryOplogApplierStats stats; + auto oplogApplicationMode = (recoveryMode == RecoveryMode::kStartupFromStableTimestamp || + recoveryMode == RecoveryMode::kRollbackFromStableTimestamp) + ? OplogApplication::Mode::kStableRecovering + : OplogApplication::Mode::kUnstableRecovering; auto writerPool = makeReplWriterPool(); auto* replCoord = ReplicationCoordinator::get(opCtx); OplogApplierImpl oplogApplier(nullptr, @@ -721,7 +728,7 @@ Timestamp ReplicationRecoveryImpl::_applyOplogOperations(OperationContext* opCtx replCoord, _consistencyMarkers, _storageInterface, - OplogApplier::Options(OplogApplication::Mode::kRecovering), + OplogApplier::Options(oplogApplicationMode), writerPool.get()); OplogApplier::BatchLimits batchLimits; diff --git a/src/mongo/db/repl/replication_recovery.h b/src/mongo/db/repl/replication_recovery.h index c6c4a000918..6eb7782de00 100644 --- a/src/mongo/db/repl/replication_recovery.h +++ b/src/mongo/db/repl/replication_recovery.h @@ -53,9 +53,12 @@ public: /** * Recovers the data on disk from the oplog. If the provided stable timestamp is not "none", * this function assumes the data reflects that timestamp. + * Returns the provided stable timestamp. If the provided stable timestamp is "none", this + * function might try to ask storage for the last stable timestamp if it exists before doing + * recovery which will be returned after performing successful recovery. */ - virtual void recoverFromOplog(OperationContext* opCtx, - boost::optional<Timestamp> stableTimestamp) = 0; + virtual boost::optional<Timestamp> recoverFromOplog( + OperationContext* opCtx, boost::optional<Timestamp> stableTimestamp) = 0; /** * Recovers the data on disk from the oplog and puts the node in readOnly mode. If @@ -79,8 +82,8 @@ public: ReplicationRecoveryImpl(StorageInterface* storageInterface, ReplicationConsistencyMarkers* consistencyMarkers); - void recoverFromOplog(OperationContext* opCtx, - boost::optional<Timestamp> stableTimestamp) override; + boost::optional<Timestamp> recoverFromOplog( + OperationContext* opCtx, boost::optional<Timestamp> stableTimestamp) override; void recoverFromOplogAsStandalone(OperationContext* opCtx, bool duringInitialSync = false) override; diff --git a/src/mongo/db/repl/replication_recovery_mock.h b/src/mongo/db/repl/replication_recovery_mock.h index 97072a4fdd9..a34ce54500a 100644 --- a/src/mongo/db/repl/replication_recovery_mock.h +++ b/src/mongo/db/repl/replication_recovery_mock.h @@ -42,8 +42,10 @@ class ReplicationRecoveryMock : public ReplicationRecovery { public: ReplicationRecoveryMock() = default; - void recoverFromOplog(OperationContext* opCtx, - boost::optional<Timestamp> stableTimestamp) override {} + boost::optional<Timestamp> recoverFromOplog( + OperationContext* opCtx, boost::optional<Timestamp> stableTimestamp) override { + return stableTimestamp; + } void recoverFromOplogAsStandalone(OperationContext* opCtx, bool duringInitialSync = false) override {} diff --git a/src/mongo/db/repl/replication_recovery_test.cpp b/src/mongo/db/repl/replication_recovery_test.cpp index 00bd8b1c35b..2306994463a 100644 --- a/src/mongo/db/repl/replication_recovery_test.cpp +++ b/src/mongo/db/repl/replication_recovery_test.cpp @@ -702,7 +702,7 @@ TEST_F(ReplicationRecoveryTest, testRecoveryToStableAppliesDocumentsWithNoAppliedThrough(false); } -TEST_F(ReplicationRecoveryTest, RecoveryIgnoresDroppedCollections) { +TEST_F(ReplicationRecoveryTest, UnstableRecoveryIgnoresDroppedCollections) { ReplicationRecoveryImpl recovery(getStorageInterface(), getConsistencyMarkers()); auto opCtx = getOperationContext(); @@ -714,7 +714,7 @@ TEST_F(ReplicationRecoveryTest, RecoveryIgnoresDroppedCollections) { ASSERT_FALSE(autoColl.getCollection()); } - getStorageInterfaceRecovery()->setRecoveryTimestamp(Timestamp(2, 2)); + // Not setting a stable timestamp in order to perform unstable recovery, recovery.recoverFromOplog(opCtx, boost::none); _assertDocsInOplog(opCtx, {1, 2, 3, 4, 5}); @@ -725,6 +725,24 @@ TEST_F(ReplicationRecoveryTest, RecoveryIgnoresDroppedCollections) { ASSERT_EQ(getConsistencyMarkers()->getOplogTruncateAfterPoint(opCtx), Timestamp()); } +DEATH_TEST_REGEX_F(ReplicationRecoveryTest, + StableRecoveryCrashOnDroppedCollectionsInTests, + "Fatal assertion.*5415000") { + ReplicationRecoveryImpl recovery(getStorageInterface(), getConsistencyMarkers()); + auto opCtx = getOperationContext(); + + _setUpOplog(opCtx, getStorageInterface(), {1, 2, 3, 4, 5}); + + ASSERT_OK(getStorageInterface()->dropCollection(opCtx, testNs)); + { + AutoGetCollectionForReadCommand autoColl(opCtx, testNs); + ASSERT_FALSE(autoColl.getCollection()); + } + + getStorageInterfaceRecovery()->setRecoveryTimestamp(Timestamp(2, 2)); + recovery.recoverFromOplog(opCtx, boost::none); +} + TEST_F(ReplicationRecoveryTest, RecoveryAppliesDocumentsWhenAppliedThroughIsBehindAfterTruncation) { ReplicationRecoveryImpl recovery(getStorageInterface(), getConsistencyMarkers()); auto opCtx = getOperationContext(); @@ -1212,7 +1230,7 @@ TEST_F(ReplicationRecoveryTest, ASSERT_EQ(getConsistencyMarkers()->getOplogTruncateAfterPoint(opCtx), Timestamp()); } -TEST_F(ReplicationRecoveryTest, RecoverFromOplogUpToBeforeEndOfOplog) { +TEST_F(ReplicationRecoveryTest, RecoverFromOplogUpTo) { ReplicationRecoveryImpl recovery(getStorageInterface(), getConsistencyMarkers()); auto opCtx = getOperationContext(); @@ -1222,8 +1240,16 @@ TEST_F(ReplicationRecoveryTest, RecoverFromOplogUpToBeforeEndOfOplog) { // Recovers operations with timestamps: 3, 4, 5. recovery.recoverFromOplogUpTo(opCtx, Timestamp(5, 5)); _assertDocsInTestCollection(opCtx, {3, 4, 5}); +} + +TEST_F(ReplicationRecoveryTest, RecoverFromOplogUpToBeforeEndOfOplog) { + ReplicationRecoveryImpl recovery(getStorageInterface(), getConsistencyMarkers()); + auto opCtx = getOperationContext(); - // Recovers operations with timestamps: 6, 7, 8, 9. + _setUpOplog(opCtx, getStorageInterface(), {2, 3, 4, 5, 6, 7, 8, 9, 10}); + getStorageInterfaceRecovery()->setRecoveryTimestamp(Timestamp(2, 2)); + + // Recovers operations with timestamps: 3, 4, 5, 6, 7, 8, 9. recovery.recoverFromOplogUpTo(opCtx, Timestamp(9, 9)); _assertDocsInTestCollection(opCtx, {3, 4, 5, 6, 7, 8, 9}); } @@ -1289,8 +1315,6 @@ TEST_F(ReplicationRecoveryTest, RecoverFromOplogUpToDoesNotExceedEndPoint) { _setUpOplog(opCtx, getStorageInterface(), {2, 5, 10}); getStorageInterfaceRecovery()->setRecoveryTimestamp(Timestamp(2, 2)); - recovery.recoverFromOplogUpTo(opCtx, Timestamp(9, 9)); - recovery.recoverFromOplogUpTo(opCtx, Timestamp(15, 15)); } diff --git a/src/mongo/db/repl/rollback_impl.cpp b/src/mongo/db/repl/rollback_impl.cpp index 1cec7be6b30..5ea811dff36 100644 --- a/src/mongo/db/repl/rollback_impl.cpp +++ b/src/mongo/db/repl/rollback_impl.cpp @@ -42,8 +42,8 @@ #include "mongo/db/catalog/import_collection_oplog_entry_gen.h" #include "mongo/db/commands.h" #include "mongo/db/concurrency/d_concurrency.h" +#include "mongo/db/concurrency/exception_util.h" #include "mongo/db/concurrency/replication_state_transition_lock_guard.h" -#include "mongo/db/concurrency/write_conflict_exception.h" #include "mongo/db/db_raii.h" #include "mongo/db/dbdirectclient.h" #include "mongo/db/dbhelpers.h" @@ -655,7 +655,7 @@ void RollbackImpl::_runPhaseFromAbortToReconstructPreparedTxns( // transactions were aborted (i.e. the in-memory counts were rolled-back) before computing // collection counts, reconstruct the prepared transactions now, adding on any additional counts // to the now corrected record store. - reconstructPreparedTransactions(opCtx, OplogApplication::Mode::kRecovering); + reconstructPreparedTransactions(opCtx, OplogApplication::Mode::kStableRecovering); } void RollbackImpl::_correctRecordStoreCounts(OperationContext* opCtx) { diff --git a/src/mongo/db/repl/rs_rollback.cpp b/src/mongo/db/repl/rs_rollback.cpp index 35951278ee1..d1d460dcd83 100644 --- a/src/mongo/db/repl/rs_rollback.cpp +++ b/src/mongo/db/repl/rs_rollback.cpp @@ -49,8 +49,8 @@ #include "mongo/db/client.h" #include "mongo/db/commands.h" #include "mongo/db/commands/txn_cmds_gen.h" +#include "mongo/db/concurrency/exception_util.h" #include "mongo/db/concurrency/replication_state_transition_lock_guard.h" -#include "mongo/db/concurrency/write_conflict_exception.h" #include "mongo/db/db_raii.h" #include "mongo/db/dbhelpers.h" #include "mongo/db/exec/working_set_common.h" @@ -860,9 +860,10 @@ void dropIndex(OperationContext* opCtx, const string& indexName, NamespaceString& nss) { IndexCatalog* indexCatalog = collection->getIndexCatalog(); - bool includeUnfinishedIndexes = true; - auto indexDescriptor = - indexCatalog->findIndexByName(opCtx, indexName, includeUnfinishedIndexes); + auto indexDescriptor = indexCatalog->findIndexByName( + opCtx, + indexName, + IndexCatalog::InclusionPolicy::kReady | IndexCatalog::InclusionPolicy::kUnfinished); if (!indexDescriptor) { LOGV2_WARNING(21725, "Rollback failed to drop index {indexName} in {namespace}: index not found.", @@ -998,7 +999,7 @@ void rollbackDropIndexes(OperationContext* opCtx, "uuid"_attr = uuid, "indexName"_attr = indexName); - createIndexForApplyOps(opCtx, indexSpec, *nss, OplogApplication::Mode::kRecovering); + createIndexForApplyOps(opCtx, indexSpec, *nss, OplogApplication::Mode::kStableRecovering); LOGV2_DEBUG(21676, 1, diff --git a/src/mongo/db/repl/session_update_tracker.cpp b/src/mongo/db/repl/session_update_tracker.cpp index 87c12cf5463..bdfa670385a 100644 --- a/src/mongo/db/repl/session_update_tracker.cpp +++ b/src/mongo/db/repl/session_update_tracker.cpp @@ -212,7 +212,7 @@ boost::optional<std::vector<OplogEntry>> SessionUpdateTracker::_updateSessionInf return {}; } - if (!entry.getObject2()) { + if (!entry.getObject2() || entry.getObject2()->isEmpty()) { return {}; } } diff --git a/src/mongo/db/repl/storage_interface_impl.cpp b/src/mongo/db/repl/storage_interface_impl.cpp index 4aedc7c2473..0c91ba3b592 100644 --- a/src/mongo/db/repl/storage_interface_impl.cpp +++ b/src/mongo/db/repl/storage_interface_impl.cpp @@ -53,8 +53,8 @@ #include "mongo/db/catalog/index_catalog.h" #include "mongo/db/client.h" #include "mongo/db/concurrency/d_concurrency.h" +#include "mongo/db/concurrency/exception_util.h" #include "mongo/db/concurrency/lock_state.h" -#include "mongo/db/concurrency/write_conflict_exception.h" #include "mongo/db/curop.h" #include "mongo/db/db_raii.h" #include "mongo/db/dbhelpers.h" @@ -638,7 +638,9 @@ Status StorageInterfaceImpl::setIndexIsMultikey(OperationContext* opCtx, } auto idx = collection->getIndexCatalog()->findIndexByName( - opCtx, indexName, true /* includeUnfinishedIndexes */); + opCtx, + indexName, + IndexCatalog::InclusionPolicy::kReady | IndexCatalog::InclusionPolicy::kUnfinished); if (!idx) { return Status(ErrorCodes::IndexNotFound, str::stream() @@ -786,9 +788,8 @@ StatusWith<std::vector<BSONObj>> _findOrDeleteDocuments( // Use index scan. auto indexCatalog = collection->getIndexCatalog(); invariant(indexCatalog); - bool includeUnfinishedIndexes = false; - const IndexDescriptor* indexDescriptor = - indexCatalog->findIndexByName(opCtx, *indexName, includeUnfinishedIndexes); + const IndexDescriptor* indexDescriptor = indexCatalog->findIndexByName( + opCtx, *indexName, IndexCatalog::InclusionPolicy::kReady); if (!indexDescriptor) { return Result(ErrorCodes::IndexNotFound, str::stream() << "Index not found, ns:" << nsOrUUID.toString() diff --git a/src/mongo/db/repl/storage_interface_impl_test.cpp b/src/mongo/db/repl/storage_interface_impl_test.cpp index de039c3705c..3c942ed7361 100644 --- a/src/mongo/db/repl/storage_interface_impl_test.cpp +++ b/src/mongo/db/repl/storage_interface_impl_test.cpp @@ -42,7 +42,7 @@ #include "mongo/db/catalog/validate_results.h" #include "mongo/db/client.h" #include "mongo/db/concurrency/d_concurrency.h" -#include "mongo/db/concurrency/write_conflict_exception.h" +#include "mongo/db/concurrency/exception_util.h" #include "mongo/db/db_raii.h" #include "mongo/db/index/index_access_method.h" #include "mongo/db/index/index_descriptor.h" diff --git a/src/mongo/db/repl/storage_timestamp_test.cpp b/src/mongo/db/repl/storage_timestamp_test.cpp index 7919675a946..acb142f339e 100644 --- a/src/mongo/db/repl/storage_timestamp_test.cpp +++ b/src/mongo/db/repl/storage_timestamp_test.cpp @@ -46,7 +46,7 @@ #include "mongo/db/catalog/index_catalog.h" #include "mongo/db/catalog/multi_index_block.h" #include "mongo/db/client.h" -#include "mongo/db/concurrency/write_conflict_exception.h" +#include "mongo/db/concurrency/exception_util.h" #include "mongo/db/db_raii.h" #include "mongo/db/dbdirectclient.h" #include "mongo/db/dbhelpers.h" @@ -1474,7 +1474,7 @@ TEST_F(StorageTimestampTest, SecondarySetWildcardIndexMultikeyOnInsert) { _coordinatorMock, _consistencyMarkers, storageInterface, - repl::OplogApplier::Options(repl::OplogApplication::Mode::kRecovering), + repl::OplogApplier::Options(repl::OplogApplication::Mode::kStableRecovering), writerPool.get()); uassertStatusOK(oplogApplier.applyOplogBatch(_opCtx, ops)); @@ -1572,7 +1572,7 @@ TEST_F(StorageTimestampTest, SecondarySetWildcardIndexMultikeyOnUpdate) { _coordinatorMock, _consistencyMarkers, storageInterface, - repl::OplogApplier::Options(repl::OplogApplication::Mode::kRecovering), + repl::OplogApplier::Options(repl::OplogApplication::Mode::kStableRecovering), writerPool.get()); uassertStatusOK(oplogApplier.applyOplogBatch(_opCtx, ops)); @@ -2704,8 +2704,10 @@ TEST_F(StorageTimestampTest, IndexBuildsResolveErrorsDuringStateChangeToPrimary) } auto indexCatalog = collection->getIndexCatalog(); - buildingIndex = indexCatalog->getEntry( - indexCatalog->findIndexByName(_opCtx, "a_1_b_1", /* includeUnfinished */ true)); + buildingIndex = indexCatalog->getEntry(indexCatalog->findIndexByName( + _opCtx, + "a_1_b_1", + IndexCatalog::InclusionPolicy::kReady | IndexCatalog::InclusionPolicy::kUnfinished)); ASSERT(buildingIndex); ASSERT_OK(indexer.insertAllDocumentsInCollection(_opCtx, collection.get())); diff --git a/src/mongo/db/repl/tenant_migration_access_blocker_util.cpp b/src/mongo/db/repl/tenant_migration_access_blocker_util.cpp index e1790384202..954e88d044a 100644 --- a/src/mongo/db/repl/tenant_migration_access_blocker_util.cpp +++ b/src/mongo/db/repl/tenant_migration_access_blocker_util.cpp @@ -35,7 +35,7 @@ #include "mongo/db/repl/tenant_migration_access_blocker_util.h" #include "mongo/db/catalog_raii.h" -#include "mongo/db/concurrency/write_conflict_exception.h" +#include "mongo/db/concurrency/exception_util.h" #include "mongo/db/op_observer.h" #include "mongo/db/persistent_task_store.h" #include "mongo/db/repl/tenant_migration_access_blocker_registry.h" diff --git a/src/mongo/db/repl/tenant_migration_donor_op_observer.h b/src/mongo/db/repl/tenant_migration_donor_op_observer.h index 43992f5e040..403d2b6b486 100644 --- a/src/mongo/db/repl/tenant_migration_donor_op_observer.h +++ b/src/mongo/db/repl/tenant_migration_donor_op_observer.h @@ -188,6 +188,10 @@ public: Timestamp commitTimestamp, const std::vector<repl::ReplOperation>& statements) noexcept final {} + void onTransactionPrepareNonPrimary(OperationContext* opCtx, + const std::vector<repl::OplogEntry>& statements, + const repl::OpTime& prepareOpTime) final {} + std::unique_ptr<ApplyOpsOplogSlotAndOperationAssignment> preTransactionPrepare( OperationContext* opCtx, const std::vector<OplogSlot>& reservedSlots, diff --git a/src/mongo/db/repl/tenant_migration_donor_service.cpp b/src/mongo/db/repl/tenant_migration_donor_service.cpp index e7912da9577..5fa0c560284 100644 --- a/src/mongo/db/repl/tenant_migration_donor_service.cpp +++ b/src/mongo/db/repl/tenant_migration_donor_service.cpp @@ -36,10 +36,11 @@ #include "mongo/config.h" #include "mongo/db/commands/tenant_migration_donor_cmds_gen.h" #include "mongo/db/commands/tenant_migration_recipient_cmds_gen.h" -#include "mongo/db/concurrency/write_conflict_exception.h" +#include "mongo/db/concurrency/exception_util.h" #include "mongo/db/db_raii.h" #include "mongo/db/dbhelpers.h" #include "mongo/db/index_builds_coordinator.h" +#include "mongo/db/keys_collection_util.h" #include "mongo/db/persistent_task_store.h" #include "mongo/db/query/find_command_gen.h" #include "mongo/db/repl/repl_server_parameters_gen.h" @@ -67,6 +68,7 @@ MONGO_FAIL_POINT_DEFINE(pauseTenantMigrationBeforeLeavingAbortingIndexBuildsStat MONGO_FAIL_POINT_DEFINE(pauseTenantMigrationBeforeLeavingBlockingState); MONGO_FAIL_POINT_DEFINE(pauseTenantMigrationBeforeLeavingDataSyncState); MONGO_FAIL_POINT_DEFINE(pauseTenantMigrationBeforeFetchingKeys); +MONGO_FAIL_POINT_DEFINE(pauseTenantMigrationDonorBeforeStoringExternalClusterTimeKeyDocs); MONGO_FAIL_POINT_DEFINE(pauseTenantMigrationDonorBeforeWaitingForKeysToReplicate); MONGO_FAIL_POINT_DEFINE(pauseTenantMigrationDonorBeforeMarkingStateGarbageCollectable); MONGO_FAIL_POINT_DEFINE(pauseTenantMigrationDonorAfterMarkingStateGarbageCollectable); @@ -77,6 +79,7 @@ MONGO_FAIL_POINT_DEFINE(pauseTenantMigrationBeforeInsertingDonorStateDoc); MONGO_FAIL_POINT_DEFINE(pauseTenantMigrationBeforeCreatingStateDocumentTTLIndex); MONGO_FAIL_POINT_DEFINE(pauseTenantMigrationBeforeCreatingExternalKeysTTLIndex); MONGO_FAIL_POINT_DEFINE(pauseTenantMigrationBeforeLeavingCommittedState); +MONGO_FAIL_POINT_DEFINE(pauseTenantMigrationAfterUpdatingToCommittedState); const std::string kTTLIndexName = "TenantMigrationDonorTTLIndex"; const std::string kExternalKeysTTLIndexName = "ExternalKeysTTLIndex"; @@ -617,6 +620,10 @@ ExecutorFuture<repl::OpTime> TenantMigrationDonorService::Instance::_updateState wuow.commit(); + if (nextState == TenantMigrationDonorStateEnum::kCommitted) { + pauseTenantMigrationAfterUpdatingToCommittedState.pauseWhileSet(); + } + updateOpTime = oplogSlot; }); @@ -932,13 +939,13 @@ SemiFuture<void> TenantMigrationDonorService::Instance::run( return _waitForRecipientToBecomeConsistentAndEnterBlockingState( executor, recipientTargeterRS, abortToken); }) - .then([this, self = shared_from_this(), executor, recipientTargeterRS, abortToken] { + .then([this, self = shared_from_this(), executor, recipientTargeterRS, abortToken, token] { LOGV2(6104905, "Waiting for recipient to reach the block timestamp.", "migrationId"_attr = _migrationUuid, "tenantId"_attr = _tenantId); return _waitForRecipientToReachBlockTimestampAndEnterCommittedState( - executor, recipientTargeterRS, abortToken); + executor, recipientTargeterRS, abortToken, token); }) // Note from here on the migration cannot be aborted, so only the token from the primary // only service should be used. @@ -1064,32 +1071,34 @@ TenantMigrationDonorService::Instance::_fetchAndStoreRecipientClusterTimeKeyDocs std::make_shared<std::vector<ExternalKeysCollectionDocument>>(); auto fetchStatus = std::make_shared<boost::optional<Status>>(); - auto fetcherCallback = - [this, self = shared_from_this(), fetchStatus, keyDocs]( - const Fetcher::QueryResponseStatus& dataStatus, - Fetcher::NextAction* nextAction, - BSONObjBuilder* getMoreBob) { - // Throw out any accumulated results on error - if (!dataStatus.isOK()) { - *fetchStatus = dataStatus.getStatus(); - keyDocs->clear(); - return; - } + auto fetcherCallback = [this, + self = shared_from_this(), + fetchStatus, + keyDocs]( + const Fetcher::QueryResponseStatus& dataStatus, + Fetcher::NextAction* nextAction, + BSONObjBuilder* getMoreBob) { + // Throw out any accumulated results on error + if (!dataStatus.isOK()) { + *fetchStatus = dataStatus.getStatus(); + keyDocs->clear(); + return; + } - const auto& data = dataStatus.getValue(); - for (const BSONObj& doc : data.documents) { - keyDocs->push_back( - tenant_migration_util::makeExternalClusterTimeKeyDoc( - _migrationUuid, doc.getOwned())); - } - *fetchStatus = Status::OK(); + const auto& data = dataStatus.getValue(); + for (const BSONObj& doc : data.documents) { + keyDocs->push_back( + keys_collection_util::makeExternalClusterTimeKeyDoc( + doc.getOwned(), _migrationUuid, boost::none /* expireAt */)); + } + *fetchStatus = Status::OK(); - if (!getMoreBob) { - return; - } - getMoreBob->append("getMore", data.cursorId); - getMoreBob->append("collection", data.nss.coll()); - }; + if (!getMoreBob) { + return; + } + getMoreBob->append("getMore", data.cursorId); + getMoreBob->append("collection", data.nss.coll()); + }; auto fetcher = std::make_shared<Fetcher>( _recipientCmdExecutor.get(), @@ -1144,8 +1153,11 @@ TenantMigrationDonorService::Instance::_fetchAndStoreRecipientClusterTimeKeyDocs .then([this, self = shared_from_this(), executor, token](auto keyDocs) { checkForTokenInterrupt(token); - return tenant_migration_util::storeExternalClusterTimeKeyDocs( - std::move(keyDocs)); + auto opCtx = cc().makeOperationContext(); + pauseTenantMigrationDonorBeforeStoringExternalClusterTimeKeyDocs + .pauseWhileSet(opCtx.get()); + return keys_collection_util::storeExternalClusterTimeKeyDocs( + opCtx.get(), std::move(keyDocs)); }) .then([this, self = shared_from_this(), token](repl::OpTime lastKeyOpTime) { pauseTenantMigrationDonorBeforeWaitingForKeysToReplicate.pauseWhileSet(); @@ -1219,6 +1231,7 @@ ExecutorFuture<void> TenantMigrationDonorService::Instance::_waitForRecipientToReachBlockTimestampAndEnterCommittedState( const std::shared_ptr<executor::ScopedTaskExecutor>& executor, std::shared_ptr<RemoteCommandTargeter> recipientTargeterRS, + const CancellationToken& abortToken, const CancellationToken& token) { { stdx::lock_guard<Latch> lg(_mutex); @@ -1228,7 +1241,6 @@ TenantMigrationDonorService::Instance::_waitForRecipientToReachBlockTimestampAnd invariant(_stateDoc.getBlockTimestamp()); } - // Source to cancel the timeout if the operation completed in time. CancellationSource cancelTimeoutSource; CancellationSource recipientSyncDataSource(token); @@ -1284,18 +1296,21 @@ TenantMigrationDonorService::Instance::_waitForRecipientToReachBlockTimestampAnd uasserted(ErrorCodes::InternalError, "simulate a tenant migration error"); } }) - .then([this, self = shared_from_this(), executor, token] { + .then([this, self = shared_from_this(), executor, abortToken, token] { + // Last chance to abort + checkForTokenInterrupt(abortToken); + // Enter "commit" state. LOGV2(6104908, "Entering 'committed' state.", "migrationId"_attr = _migrationUuid, "tenantId"_attr = _tenantId); + // Ignore the abort token once we've entered the committed state return _updateStateDoc(executor, TenantMigrationDonorStateEnum::kCommitted, token) .then([this, self = shared_from_this(), executor, token](repl::OpTime opTime) { return _waitForMajorityWriteConcern(executor, std::move(opTime), token) .then([this, self = shared_from_this()] { pauseTenantMigrationBeforeLeavingCommittedState.pauseWhileSet(); - stdx::lock_guard<Latch> lg(_mutex); // If interrupt is called at some point during execution, it is // possible that interrupt() will fulfill the promise before we diff --git a/src/mongo/db/repl/tenant_migration_donor_service.h b/src/mongo/db/repl/tenant_migration_donor_service.h index 87931e03343..11fe2d6ffab 100644 --- a/src/mongo/db/repl/tenant_migration_donor_service.h +++ b/src/mongo/db/repl/tenant_migration_donor_service.h @@ -188,6 +188,7 @@ public: ExecutorFuture<void> _waitForRecipientToReachBlockTimestampAndEnterCommittedState( const std::shared_ptr<executor::ScopedTaskExecutor>& executor, std::shared_ptr<RemoteCommandTargeter> recipientTargeterRS, + const CancellationToken& abortToken, const CancellationToken& token); ExecutorFuture<void> _handleErrorOrEnterAbortedState( diff --git a/src/mongo/db/repl/tenant_migration_recipient_entry_helpers.cpp b/src/mongo/db/repl/tenant_migration_recipient_entry_helpers.cpp index ce994c1581e..bc798479097 100644 --- a/src/mongo/db/repl/tenant_migration_recipient_entry_helpers.cpp +++ b/src/mongo/db/repl/tenant_migration_recipient_entry_helpers.cpp @@ -33,7 +33,7 @@ #include "mongo/db/catalog/database.h" #include "mongo/db/catalog_raii.h" -#include "mongo/db/concurrency/write_conflict_exception.h" +#include "mongo/db/concurrency/exception_util.h" #include "mongo/db/db_raii.h" #include "mongo/db/dbhelpers.h" #include "mongo/db/namespace_string.h" diff --git a/src/mongo/db/repl/tenant_migration_recipient_op_observer.h b/src/mongo/db/repl/tenant_migration_recipient_op_observer.h index dd42ff6581f..7f5181a6d4a 100644 --- a/src/mongo/db/repl/tenant_migration_recipient_op_observer.h +++ b/src/mongo/db/repl/tenant_migration_recipient_op_observer.h @@ -207,6 +207,10 @@ public: size_t numberOfPrePostImagesToWrite, Date_t wallClockTime) final {} + void onTransactionPrepareNonPrimary(OperationContext* opCtx, + const std::vector<repl::OplogEntry>& statements, + const repl::OpTime& prepareOpTime) final {} + void onTransactionAbort(OperationContext* opCtx, boost::optional<OplogSlot> abortOplogEntryOpTime) final {} diff --git a/src/mongo/db/repl/tenant_migration_recipient_service.cpp b/src/mongo/db/repl/tenant_migration_recipient_service.cpp index 3f32411c555..2fcfd6cbffa 100644 --- a/src/mongo/db/repl/tenant_migration_recipient_service.cpp +++ b/src/mongo/db/repl/tenant_migration_recipient_service.cpp @@ -43,9 +43,10 @@ #include "mongo/db/client.h" #include "mongo/db/commands/tenant_migration_donor_cmds_gen.h" #include "mongo/db/commands/test_commands_enabled.h" -#include "mongo/db/concurrency/write_conflict_exception.h" +#include "mongo/db/concurrency/exception_util.h" #include "mongo/db/db_raii.h" #include "mongo/db/dbdirectclient.h" +#include "mongo/db/keys_collection_util.h" #include "mongo/db/namespace_string.h" #include "mongo/db/op_observer.h" #include "mongo/db/ops/write_ops_exec.h" @@ -256,6 +257,10 @@ public: MONGO_UNREACHABLE; } + StatusWith<LastVote> loadLocalLastVoteDocument(OperationContext* opCtx) const final { + MONGO_UNREACHABLE; + } + JournalListener* getReplicationJournalListener() final { MONGO_UNREACHABLE; } @@ -2281,11 +2286,12 @@ void TenantMigrationRecipientService::Instance::_fetchAndStoreDonorClusterTimeKe auto cursor = _client->find(std::move(findRequest), _readPreference); while (cursor->more()) { const auto doc = cursor->nextSafe().getOwned(); - keyDocs.push_back( - tenant_migration_util::makeExternalClusterTimeKeyDoc(_migrationUuid, doc)); + keyDocs.push_back(keys_collection_util::makeExternalClusterTimeKeyDoc( + doc, _migrationUuid, boost::none /* expireAt */)); } - tenant_migration_util::storeExternalClusterTimeKeyDocs(std::move(keyDocs)); + auto opCtx = cc().makeOperationContext(); + keys_collection_util::storeExternalClusterTimeKeyDocs(opCtx.get(), std::move(keyDocs)); } void TenantMigrationRecipientService::Instance::_compareRecipientAndDonorFCV() const { diff --git a/src/mongo/db/repl/tenant_migration_shard_merge_util.cpp b/src/mongo/db/repl/tenant_migration_shard_merge_util.cpp index 05779a48618..2ef76d6839f 100644 --- a/src/mongo/db/repl/tenant_migration_shard_merge_util.cpp +++ b/src/mongo/db/repl/tenant_migration_shard_merge_util.cpp @@ -41,7 +41,7 @@ #include "mongo/db/catalog/create_collection.h" #include "mongo/db/catalog/uncommitted_catalog_updates.h" #include "mongo/db/concurrency/d_concurrency.h" -#include "mongo/db/concurrency/write_conflict_exception.h" +#include "mongo/db/concurrency/exception_util.h" #include "mongo/db/cursor_server_params_gen.h" #include "mongo/db/db_raii.h" #include "mongo/db/multitenancy.h" diff --git a/src/mongo/db/repl/tenant_migration_util.cpp b/src/mongo/db/repl/tenant_migration_util.cpp index 4386cdf1cce..48e21715a4c 100644 --- a/src/mongo/db/repl/tenant_migration_util.cpp +++ b/src/mongo/db/repl/tenant_migration_util.cpp @@ -32,7 +32,7 @@ #include "mongo/bson/json.h" #include "mongo/bson/mutable/algorithm.h" #include "mongo/bson/mutable/document.h" -#include "mongo/db/concurrency/write_conflict_exception.h" +#include "mongo/db/concurrency/exception_util.h" #include "mongo/db/db_raii.h" #include "mongo/db/dbdirectclient.h" #include "mongo/db/dbhelpers.h" @@ -62,50 +62,11 @@ const std::set<std::string> kSensitiveFieldNames{"donorCertificateForRecipient", "recipientCertificateForDonor"}; MONGO_FAIL_POINT_DEFINE(pauseTenantMigrationBeforeMarkingExternalKeysGarbageCollectable); -MONGO_FAIL_POINT_DEFINE(pauseTenantMigrationBeforeStoringExternalClusterTimeKeyDocs); } // namespace const Backoff kExponentialBackoff(Seconds(1), Milliseconds::max()); -ExternalKeysCollectionDocument makeExternalClusterTimeKeyDoc(UUID migrationId, BSONObj keyDoc) { - auto originalKeyDoc = KeysCollectionDocument::parse(IDLParserErrorContext("keyDoc"), keyDoc); - - ExternalKeysCollectionDocument externalKeyDoc( - OID::gen(), originalKeyDoc.getKeyId(), migrationId); - externalKeyDoc.setKeysCollectionDocumentBase(originalKeyDoc.getKeysCollectionDocumentBase()); - - return externalKeyDoc; -} - -repl::OpTime storeExternalClusterTimeKeyDocs(std::vector<ExternalKeysCollectionDocument> keyDocs) { - auto opCtxHolder = cc().makeOperationContext(); - auto opCtx = opCtxHolder.get(); - auto nss = NamespaceString::kExternalKeysCollectionNamespace; - - pauseTenantMigrationBeforeStoringExternalClusterTimeKeyDocs.pauseWhileSet(opCtx); - - for (auto& keyDoc : keyDocs) { - AutoGetCollection collection(opCtx, nss, MODE_IX); - - writeConflictRetry(opCtx, "CloneExternalKeyDocs", nss.ns(), [&] { - // Note that each external key's _id is generated by the migration, so this upsert can - // only insert. - const auto filter = - BSON(ExternalKeysCollectionDocument::kIdFieldName << keyDoc.getId()); - const auto updateMod = keyDoc.toBSON(); - - Helpers::upsert(opCtx, - nss.ns(), - filter, - updateMod, - /*fromMigrate=*/false); - }); - } - - return repl::ReplClientInfo::forClient(opCtx->getClient()).getLastOp(); -} - void createOplogViewForTenantMigrations(OperationContext* opCtx, Database* db) { writeConflictRetry( opCtx, "createDonorOplogView", NamespaceString::kTenantMigrationOplogView.ns(), [&] { diff --git a/src/mongo/db/repl/tenant_migration_util.h b/src/mongo/db/repl/tenant_migration_util.h index 44e2ac67bbf..40421caa5df 100644 --- a/src/mongo/db/repl/tenant_migration_util.h +++ b/src/mongo/db/repl/tenant_migration_util.h @@ -202,19 +202,6 @@ inline void protocolStorageOptionsCompatibilityCheck(OperationContext* opCtx, !opCtx->getServiceContext()->getStorageEngine()->isUsingDirectoryForIndexes()); } -/* - * Creates an ExternalKeysCollectionDocument representing an config.external_validation_keys - * document from the given the admin.system.keys document BSONObj. - */ -ExternalKeysCollectionDocument makeExternalClusterTimeKeyDoc(UUID migrationId, BSONObj keyDoc); - -/* - * For each given ExternalKeysCollectionDocument, inserts it if there is not an existing document in - * config.external_validation_keys for it with the same keyId and replicaSetName. Otherwise, - * updates the ttlExpiresAt of the existing document if it is less than the new ttlExpiresAt. - */ -repl::OpTime storeExternalClusterTimeKeyDocs(std::vector<ExternalKeysCollectionDocument> keyDocs); - /** * Sets the "ttlExpiresAt" field for the external keys so they can be garbage collected by the ttl * monitor. diff --git a/src/mongo/db/repl/tenant_oplog_applier.cpp b/src/mongo/db/repl/tenant_oplog_applier.cpp index 38eb4edfd71..731f83d6929 100644 --- a/src/mongo/db/repl/tenant_oplog_applier.cpp +++ b/src/mongo/db/repl/tenant_oplog_applier.cpp @@ -38,7 +38,7 @@ #include "mongo/db/auth/authorization_session.h" #include "mongo/db/catalog/document_validation.h" #include "mongo/db/catalog_raii.h" -#include "mongo/db/concurrency/write_conflict_exception.h" +#include "mongo/db/concurrency/exception_util.h" #include "mongo/db/db_raii.h" #include "mongo/db/dbhelpers.h" #include "mongo/db/op_observer.h" diff --git a/src/mongo/db/repl/topology_coordinator.cpp b/src/mongo/db/repl/topology_coordinator.cpp index 47a043e093e..630fefa541c 100644 --- a/src/mongo/db/repl/topology_coordinator.cpp +++ b/src/mongo/db/repl/topology_coordinator.cpp @@ -267,8 +267,23 @@ void TopologyCoordinator::_setSyncSource(HostAndPort newSyncSource, HostAndPort TopologyCoordinator::chooseNewSyncSource(Date_t now, const OpTime& lastOpTimeFetched, ReadPreference readPreference) { + // If we are not a member of the current replica set configuration, no sync source is valid. + if (_selfIndex == -1) { + LOGV2_DEBUG( + 21778, 1, "Cannot sync from any members because we are not in the replica set config"); + return HostAndPort(); + } + + // Check to see if we should choose a sync source because 'unsupportedSyncSource' was + // set. + auto maybeSyncSource = _chooseSyncSourceUnsupportedSyncSourceParameter(now); + if (maybeSyncSource) { + _setSyncSource(*maybeSyncSource, now, false /* fromReplSetSyncFrom */); + return _syncSource; + } + // Check to see if we should choose a sync source because the 'replSetSyncFrom' command was set. - auto maybeSyncSource = _chooseSyncSourceReplSetSyncFrom(now); + maybeSyncSource = _chooseSyncSourceReplSetSyncFrom(now); if (maybeSyncSource) { // If we have a forced sync source via 'replSetSyncFrom', set the _replSetSyncFromSet flag // to true. @@ -526,9 +541,7 @@ bool TopologyCoordinator::_isEligibleSyncSource(int candidateIndex, } boost::optional<HostAndPort> TopologyCoordinator::_chooseSyncSourceReplSetSyncFrom(Date_t now) { - if (_selfIndex == -1) { - return boost::none; - } + invariant(_selfIndex != -1, "Unexpectedly not in the replica set config"); if (_forceSyncSourceIndex == -1) { return boost::none; @@ -544,14 +557,44 @@ boost::optional<HostAndPort> TopologyCoordinator::_chooseSyncSourceReplSetSyncFr return syncSource; } -boost::optional<HostAndPort> TopologyCoordinator::_chooseSyncSourceInitialChecks(Date_t now) { - // If we are not a member of the current replica set configuration, no sync source is valid. - if (_selfIndex == -1) { - LOGV2_DEBUG( - 21778, 1, "Cannot sync from any members because we are not in the replica set config"); - return HostAndPort(); +boost::optional<HostAndPort> TopologyCoordinator::_chooseSyncSourceUnsupportedSyncSourceParameter( + Date_t now) { + invariant(_selfIndex != -1, "Unexpectedly not in the replica set config"); + + auto syncSourceStr = repl::unsupportedSyncSource; + if (syncSourceStr.empty()) { + return boost::none; + } + auto syncSource = HostAndPort(syncSourceStr); + const int syncSourceIndex = _rsConfig.findMemberIndexByHostAndPort(syncSource); + if (syncSourceIndex < 0) { + LOGV2_FATAL( + 7785600, + "Selecting node specified in 'unsupportedSyncSource' parameter failed due to host " + "and port not in replica set config.", + "unsupportedSyncSource"_attr = syncSourceStr); + } + + if (_selfIndex == syncSourceIndex) { + LOGV2_FATAL( + 7785601, + "Node specified in 'unsupportedSyncSource' parameter is self: cannot select self as " + "a sync source", + "unsupportedSyncSource"_attr = syncSourceStr); } + LOGV2(7785602, + "Choosing sync source candidate specified by 'unsupportedSyncSource' parameter", + "syncSource"_attr = syncSourceStr, + "syncsourceobj"_attr = syncSource); + std::string msg(str::stream() << "syncing from: " << syncSourceStr << " by request"); + setMyHeartbeatMessage(now, msg); + return syncSource; +} + +boost::optional<HostAndPort> TopologyCoordinator::_chooseSyncSourceInitialChecks(Date_t now) { + invariant(_selfIndex != -1, "Unexpectedly not in the replica set config"); + if (auto sfp = forceSyncSourceCandidate.scoped(); MONGO_unlikely(sfp.isActive())) { const auto& data = sfp.getData(); const auto hostAndPortElem = data["hostAndPort"]; @@ -705,6 +748,13 @@ void TopologyCoordinator::prepareSyncFromResponse(const HostAndPort& target, return; } + if (!repl::unsupportedSyncSource.empty()) { + *result = + Status(ErrorCodes::IllegalOperation, + "replSetSyncFrom may not be used when 'unsupportedSyncSource' parameter is set"); + return; + } + ReplSetConfig::MemberIterator targetConfig = _rsConfig.membersEnd(); int targetIndex = 0; for (ReplSetConfig::MemberIterator it = _rsConfig.membersBegin(); it != _rsConfig.membersEnd(); @@ -3111,6 +3161,13 @@ TopologyCoordinator::_shouldChangeSyncSourceInitialChecks(const HostAndPort& cur return {ChangeSyncSourceDecision::kNo, -1}; } + if (!repl::unsupportedSyncSource.empty()) { + LOGV2(7785604, + "Not choosing new sync source because sync source is forced via " + "'unsupportedSyncSource' parameter"); + return {ChangeSyncSourceDecision::kNo, -1}; + } + // If the user requested a sync source change, return kYes. if (_forceSyncSourceIndex != -1) { LOGV2(21829, @@ -3304,10 +3361,11 @@ bool TopologyCoordinator::shouldChangeSyncSourceDueToPingTime(const HostAndPort& // If we find an eligible sync source that is significantly closer than our current sync source, // return true. - // Do not re-evaluate our sync source if it was set via the replSetSyncFrom command or the - // forceSyncSourceCandidate failpoint. + // Do not re-evaluate our sync source if it was set via the replSetSyncFrom command, the + // forceSyncSourceCandidate failpoint, or the 'unsupportedSyncSource' server parameter. auto sfp = forceSyncSourceCandidate.scoped(); - if (_replSetSyncFromSet || MONGO_unlikely(sfp.isActive())) { + if (_replSetSyncFromSet || MONGO_unlikely(sfp.isActive()) || + !repl::unsupportedSyncSource.empty()) { return false; } @@ -3485,7 +3543,7 @@ void TopologyCoordinator::processReplSetRequestVotes(const ReplSetRequestVotesAr if (!args.isADryRun()) { _lastVote.setTerm(args.getTerm()); _lastVote.setCandidateIndex(args.getCandidateIndex()); - LOGV2_DEBUG(5972100, 0, "Voting yes in election"); + LOGV2_DEBUG(5972100, 1, "Voting yes in election"); } response->setVoteGranted(true); } diff --git a/src/mongo/db/repl/topology_coordinator.h b/src/mongo/db/repl/topology_coordinator.h index 3d3084ddf76..a9b8cd8ab16 100644 --- a/src/mongo/db/repl/topology_coordinator.h +++ b/src/mongo/db/repl/topology_coordinator.h @@ -906,6 +906,9 @@ private: // Returns a HostAndPort if one is forced via the 'replSetSyncFrom' command. boost::optional<HostAndPort> _chooseSyncSourceReplSetSyncFrom(Date_t now); + // Returns a HostAndPort if one is forced via the 'unsupportedSyncSource' startup parameter. + boost::optional<HostAndPort> _chooseSyncSourceUnsupportedSyncSourceParameter(Date_t now); + // Does preliminary checks involved in choosing sync source // * Do we have a valid configuration? // * Is the 'forceSyncSourceCandidate' failpoint enabled? diff --git a/src/mongo/db/repl/topology_coordinator_v1_test.cpp b/src/mongo/db/repl/topology_coordinator_v1_test.cpp index 3898035b08c..e29bf1edc1e 100644 --- a/src/mongo/db/repl/topology_coordinator_v1_test.cpp +++ b/src/mongo/db/repl/topology_coordinator_v1_test.cpp @@ -4888,6 +4888,107 @@ TEST_F(ReevalSyncSourceTest, NoChangeWhenSyncSourceForcedByFailPoint) { ReadPreference::Nearest)); } +// Test that we will select the node specified by the 'unsupportedSyncSource' parameter as a sync +// source even if it is farther away. +TEST_F(ReevalSyncSourceTest, ChooseSyncSourceForcedByStartupParameterEvenIfFarther) { + RAIIServerParameterControllerForTest syncSourceParamGuard{"unsupportedSyncSource", + "host2:27017"}; + + // Make the desired host much farther away. + getTopoCoord().setPing_forTest(HostAndPort("host2"), pingTime); + getTopoCoord().setPing_forTest(HostAndPort("host3"), significantlyCloserPingTime); + + // Select a sync source. + auto syncSource = + getTopoCoord().chooseNewSyncSource(now()++, OpTime(), ReadPreference::Nearest); + ASSERT_EQ(syncSource, HostAndPort("host2:27017")); +} + +// Test that we will not change from the node specified by the 'unsupportedSyncSource' parameter +// due to ping time. +TEST_F(ReevalSyncSourceTest, NoChangeDueToPingTimeWhenSyncSourceForcedByStartupParameter) { + RAIIServerParameterControllerForTest syncSourceParamGuard{"unsupportedSyncSource", + "host2:27017"}; + // Select a sync source. + auto syncSource = + getTopoCoord().chooseNewSyncSource(now()++, OpTime(), ReadPreference::Nearest); + ASSERT_EQ(syncSource, HostAndPort("host2:27017")); + + // Set up so that without forcing the sync source, the node otherwise would have changed sync + // sources. + getTopoCoord().setPing_forTest(HostAndPort("host2"), pingTime); + getTopoCoord().setPing_forTest(HostAndPort("host3"), significantlyCloserPingTime); + + ASSERT_FALSE(getTopoCoord().shouldChangeSyncSourceDueToPingTime(HostAndPort("host2"), + MemberState::RS_SECONDARY, + lastOpTimeFetched, + now(), + ReadPreference::Nearest)); +} + +// Test that we will not change sync sources due to the replSetSyncFrom command being run when +// the 'unsupportedSyncSource' startup parameter is set - that is, the parameter should always +// take priority. +TEST_F(ReevalSyncSourceTest, NoChangeDueToReplSetSyncFromWhenSyncSourceForcedByStartupParameter) { + RAIIServerParameterControllerForTest syncSourceParamGuard{"unsupportedSyncSource", + "host2:27017"}; + // Select a sync source. + auto syncSource = + getTopoCoord().chooseNewSyncSource(now()++, OpTime(), ReadPreference::Nearest); + ASSERT_EQ(syncSource, HostAndPort("host2:27017")); + + // Simulate calling replSetSyncFrom and selecting host3. + BSONObjBuilder response; + auto result = Status::OK(); + getTopoCoord().prepareSyncFromResponse(HostAndPort("host3:27017"), &response, &result); + + // Assert the command failed. + ASSERT_EQ(result.code(), ErrorCodes::IllegalOperation); + + // Reselect a sync source, and confirm it hasn't changed. + syncSource = getTopoCoord().chooseNewSyncSource(now()++, OpTime(), ReadPreference::Nearest); + ASSERT_EQ(syncSource, HostAndPort("host2:27017")); +} + +// Test that if a node is REMOVED but has 'unsupportedSyncSource' specified, we select no sync +// source. +TEST_F(ReevalSyncSourceTest, RemovedNodeSpecifiesSyncSourceStartupParameter) { + RAIIServerParameterControllerForTest syncSourceParamGuard{"unsupportedSyncSource", + "host2:27017"}; + // Remove ourselves from the config. + updateConfig(BSON("_id" + << "rs0" + << "version" << 2 << "members" + << BSON_ARRAY(BSON("_id" << 1 << "host" + << "host2:27017") + << BSON("_id" << 2 << "host" + << "host3:27017"))), + -1); + // Confirm we were actually removed. + ASSERT_EQUALS(MemberState::RS_REMOVED, getTopoCoord().getMemberState().s); + // Confirm we select no sync source. + auto syncSource = + getTopoCoord().chooseNewSyncSource(now()++, OpTime(), ReadPreference::Nearest); + ASSERT_EQUALS(syncSource, HostAndPort()); +} + +// Test that we crash if the 'unsupportedSyncSource' parameter specifies a node that is not in the +// replica set config. +DEATH_TEST_F(ReevalSyncSourceTest, CrashOnSyncSourceParameterNotInReplSet, "7785600") { + RAIIServerParameterControllerForTest syncSourceParamGuard{"unsupportedSyncSource", + "host4:27017"}; + auto syncSource = + getTopoCoord().chooseNewSyncSource(now()++, OpTime(), ReadPreference::Nearest); +} + +// Test that we crash if the 'unsupportedSyncSource' parameter specifies ourself as a node. +DEATH_TEST_F(ReevalSyncSourceTest, CrashOnSyncSourceParameterIsSelf, "7785601") { + RAIIServerParameterControllerForTest syncSourceParamGuard{"unsupportedSyncSource", + "host1:27017"}; + auto syncSource = + getTopoCoord().chooseNewSyncSource(now()++, OpTime(), ReadPreference::Nearest); +} + class HeartbeatResponseReconfigTestV1 : public TopoCoordTest { public: virtual void setUp() { diff --git a/src/mongo/db/repl/transaction_oplog_application.cpp b/src/mongo/db/repl/transaction_oplog_application.cpp index af8fa188671..a71f8470212 100644 --- a/src/mongo/db/repl/transaction_oplog_application.cpp +++ b/src/mongo/db/repl/transaction_oplog_application.cpp @@ -35,9 +35,10 @@ #include "mongo/db/catalog_raii.h" #include "mongo/db/commands/txn_cmds_gen.h" -#include "mongo/db/concurrency/write_conflict_exception.h" +#include "mongo/db/concurrency/exception_util.h" #include "mongo/db/dbdirectclient.h" #include "mongo/db/index_builds_coordinator.h" +#include "mongo/db/op_observer.h" #include "mongo/db/repl/apply_ops.h" #include "mongo/db/repl/storage_interface_impl.h" #include "mongo/db/repl/timestamp_block.h" @@ -88,9 +89,26 @@ Status _applyOperationsForTransaction(OperationContext* opCtx, } } catch (const DBException& ex) { // Ignore NamespaceNotFound errors if we are in initial sync or recovering mode. + // During recovery we reconsutuct prepared transactions at the end after applying all + // the oplogs, so 'NamespaceNotFound' error shouldn't be hit whether it is a stable or + // unstable recovery. However we have some scenarios when this error should be skipped: + // 1- This code path can be called while applying commit oplog during unstable recovery + // when 'startupRecoveryForRestore' is set. + // 2- During selective backup: + // - During restore when 'recoverFromOplogAsStandalone' is set which is usually be + // done in a stable recovery mode. + // - After the restore finished as the standalone node started with the flag + // 'takeUnstableCheckpointOnShutdown' so after restarting the node as a replica + // set member it will go through unstable recovery. const bool ignoreException = ex.code() == ErrorCodes::NamespaceNotFound && (oplogApplicationMode == repl::OplogApplication::Mode::kInitialSync || - oplogApplicationMode == repl::OplogApplication::Mode::kRecovering); + repl::OplogApplication::inRecovering(oplogApplicationMode)); + + if (ex.code() == ErrorCodes::NamespaceNotFound && + oplogApplicationMode == repl::OplogApplication::Mode::kStableRecovering) { + repl::OplogApplication::checkOnOplogFailureForRecovery( + opCtx, op.getNss(), redact(op.toBSONForLogging()), redact(ex)); + } if (!ignoreException) { LOGV2_DEBUG( @@ -130,7 +148,7 @@ Status _applyTransactionFromOplogChain(OperationContext* opCtx, repl::OplogApplication::Mode mode, Timestamp commitTimestamp, Timestamp durableTimestamp) { - invariant(mode == repl::OplogApplication::Mode::kRecovering); + invariant(repl::OplogApplication::inRecovering(mode)); auto ops = readTransactionOperationsFromOplogChain(opCtx, entry, {}); @@ -189,7 +207,8 @@ Status applyCommitTransaction(OperationContext* opCtx, invariant(commitCommand.getCommitTimestamp()); switch (mode) { - case repl::OplogApplication::Mode::kRecovering: { + case repl::OplogApplication::Mode::kUnstableRecovering: + case repl::OplogApplication::Mode::kStableRecovering: { return _applyTransactionFromOplogChain(opCtx, entry, mode, @@ -236,7 +255,8 @@ Status applyAbortTransaction(OperationContext* opCtx, const OplogEntry& entry, repl::OplogApplication::Mode mode) { switch (mode) { - case repl::OplogApplication::Mode::kRecovering: { + case repl::OplogApplication::Mode::kUnstableRecovering: + case repl::OplogApplication::Mode::kStableRecovering: { // We don't put transactions into the prepare state until the end of recovery, // so there is no transaction to abort. return Status::OK(); @@ -388,7 +408,7 @@ Status _applyPrepareTransaction(OperationContext* opCtx, // The prepare time of the transaction is set explicitly below. auto ops = readTransactionOperationsFromOplogChain(opCtx, entry, {}); - if (mode == repl::OplogApplication::Mode::kRecovering || + if (repl::OplogApplication::inRecovering(mode) || mode == repl::OplogApplication::Mode::kInitialSync) { // We might replay a prepared transaction behind oldest timestamp. Note that since this is // scoped to the storage transaction, and readTransactionOperationsFromOplogChain implicitly @@ -465,7 +485,7 @@ Status _applyPrepareTransaction(OperationContext* opCtx, // Set this in case the application of any ops need to use the prepare timestamp of this // transaction. It should be cleared automatically when the transaction finishes. - if (mode == repl::OplogApplication::Mode::kRecovering || + if (repl::OplogApplication::inRecovering(mode) || mode == repl::OplogApplication::Mode::kInitialSync) { txnParticipant.setPrepareOpTimeForRecovery(opCtx, entry.getOpTime()); } @@ -498,6 +518,11 @@ Status _applyPrepareTransaction(OperationContext* opCtx, } txnParticipant.prepareTransaction(opCtx, entry.getOpTime()); + + auto opObserver = opCtx->getServiceContext()->getOpObserver(); + invariant(opObserver); + opObserver->onTransactionPrepareNonPrimary(opCtx, ops, entry.getOpTime()); + // Prepare transaction success. abortOnError.dismiss(); @@ -543,7 +568,8 @@ Status applyPrepareTransaction(OperationContext* opCtx, const OplogEntry& entry, repl::OplogApplication::Mode mode) { switch (mode) { - case repl::OplogApplication::Mode::kRecovering: { + case repl::OplogApplication::Mode::kUnstableRecovering: + case repl::OplogApplication::Mode::kStableRecovering: { if (!serverGlobalParams.enableMajorityReadConcern) { LOGV2_ERROR( 21850, @@ -613,7 +639,10 @@ void reconstructPreparedTransactions(OperationContext* opCtx, repl::OplogApplica AlternativeClientRegion acr(newClient); const auto newOpCtx = cc().makeOperationContext(); - _reconstructPreparedTransaction(newOpCtx.get(), prepareOplogEntry, mode); + // Ignore interruptions while reconstructing prepared transactions, so that we do not + // fassert and crash due to interruptions inside this call. + newOpCtx->runWithoutInterruptionExceptAtGlobalShutdown( + [&] { _reconstructPreparedTransaction(newOpCtx.get(), prepareOplogEntry, mode); }); } } } |