diff options
Diffstat (limited to 'src/mongo/db')
-rw-r--r-- | src/mongo/db/repl/oplog_buffer_collection.cpp | 52 | ||||
-rw-r--r-- | src/mongo/db/repl/oplog_buffer_collection.h | 20 | ||||
-rw-r--r-- | src/mongo/db/repl/oplog_buffer_collection_test.cpp | 16 | ||||
-rw-r--r-- | src/mongo/db/repl/session_update_tracker.cpp | 23 | ||||
-rw-r--r-- | src/mongo/db/repl/session_update_tracker.h | 6 | ||||
-rw-r--r-- | src/mongo/db/repl/tenant_migration_recipient_service.cpp | 2 | ||||
-rw-r--r-- | src/mongo/db/repl/tenant_oplog_applier.cpp | 101 | ||||
-rw-r--r-- | src/mongo/db/session_catalog_mongod.cpp | 12 | ||||
-rw-r--r-- | src/mongo/db/session_catalog_mongod.h | 17 | ||||
-rw-r--r-- | src/mongo/db/transaction_participant.cpp | 19 | ||||
-rw-r--r-- | src/mongo/db/transaction_participant.h | 9 |
11 files changed, 248 insertions, 29 deletions
diff --git a/src/mongo/db/repl/oplog_buffer_collection.cpp b/src/mongo/db/repl/oplog_buffer_collection.cpp index fed20b125c1..caa7cbd3803 100644 --- a/src/mongo/db/repl/oplog_buffer_collection.cpp +++ b/src/mongo/db/repl/oplog_buffer_collection.cpp @@ -109,10 +109,6 @@ void OplogBufferCollection::startup(OperationContext* opCtx) { _size = sizeResult.getValue(); _sizeIsValid = true; - auto countResult = _storageInterface->getCollectionCount(opCtx, _nss); - fassert(40404, countResult); - _count = countResult.getValue(); - // We always start from the beginning, with _lastPoppedKey being empty. This is safe because // it is always safe to replay old oplog entries in order. We explicitly reset all fields // since nothing prevents reusing an OplogBufferCollection, and the underlying collection may @@ -120,6 +116,15 @@ void OplogBufferCollection::startup(OperationContext* opCtx) { _lastPoppedKey = {}; _peekCache = std::queue<BSONObj>(); + _updateLastPushedTimestampFromCollection(lk, opCtx); +} + +void OplogBufferCollection::_updateLastPushedTimestampFromCollection(WithLock, + OperationContext* opCtx) { + auto countResult = _storageInterface->getCollectionCount(opCtx, _nss); + fassert(40404, countResult); + _count = countResult.getValue(); + if (_count == 0) { _lastPushedTimestamp = {}; return; @@ -154,18 +159,43 @@ void OplogBufferCollection::push(OperationContext* opCtx, if (begin == end) { return; } - size_t numDocs = std::distance(begin, end); - std::vector<BSONObj> docsToInsert(numDocs); stdx::lock_guard<Latch> lk(_mutex); + // Make sure timestamp order is correct. auto ts = _lastPushedTimestamp; - std::transform(begin, end, docsToInsert.begin(), [&ts](const Value& value) { - BSONObj doc; + std::for_each(begin, end, [&ts](const Value& value) { auto previousTimestamp = ts; - std::tie(doc, ts) = addIdToDocument(value); - invariant(!value.isEmpty()); + ts = value[kTimestampFieldName].timestamp(); + invariant(!ts.isNull()); invariant(ts > previousTimestamp, str::stream() << "ts: " << ts.toString() << ", previous: " << previousTimestamp.toString()); + }); + + _push(lk, opCtx, begin, end); + _lastPushedTimestamp = ts; +} + +void OplogBufferCollection::preload(OperationContext* opCtx, + Batch::const_iterator begin, + Batch::const_iterator end) { + if (begin == end) { + return; + } + stdx::lock_guard<Latch> lk(_mutex); + invariant(_lastPoppedKey.isEmpty()); + _push(lk, opCtx, begin, end); + _updateLastPushedTimestampFromCollection(lk, opCtx); +} + +void OplogBufferCollection::_push(WithLock, + OperationContext* opCtx, + Batch::const_iterator begin, + Batch::const_iterator end) { + size_t numDocs = std::distance(begin, end); + std::vector<BSONObj> docsToInsert(numDocs); + std::transform(begin, end, docsToInsert.begin(), [](const Value& value) { + auto [doc, ts] = addIdToDocument(value); + invariant(!value.isEmpty()); return doc; }); @@ -191,8 +221,6 @@ void OplogBufferCollection::push(OperationContext* opCtx, // Since the writes are ordered, it's ok to check just the last writeOp result. uassertStatusOK(writeResult.results.back()); - - _lastPushedTimestamp = ts; _count += numDocs; if (_sizeIsValid) { _size += std::accumulate(begin, end, 0U, [](const size_t& docSize, const Value& value) { diff --git a/src/mongo/db/repl/oplog_buffer_collection.h b/src/mongo/db/repl/oplog_buffer_collection.h index 6c5597efe1c..00c65d5436b 100644 --- a/src/mongo/db/repl/oplog_buffer_collection.h +++ b/src/mongo/db/repl/oplog_buffer_collection.h @@ -126,6 +126,13 @@ public: // Only currently used by the TenantMigrationRecipientService, so not part of a parent API. Timestamp getLastPushedTimestamp() const; + /** + * Like push(), but allows the operations in the batch to be out of order with + * respect to themselves and to the buffer. Legal to be called only before reading anything, + * or immediately after a clear(). + */ + void preload(OperationContext* opCtx, Batch::const_iterator begin, Batch::const_iterator end); + // ---- Testing API ---- Timestamp getLastPoppedTimestamp_forTest() const; std::queue<BSONObj> getPeekCache_forTest() const; @@ -157,12 +164,25 @@ private: bool _pop_inlock(OperationContext* opCtx, Value* value); /** + * Puts documents in collection without checking for order and without updating + * _lastPushedTimestamp. + */ + void _push(WithLock, + OperationContext* opCtx, + Batch::const_iterator begin, + Batch::const_iterator end); + /** * Returns the last document pushed onto the collection. This does not remove the `_id` field * of the document. If the collection is empty, this returns boost::none. */ boost::optional<Value> _lastDocumentPushed_inlock(OperationContext* opCtx) const; /** + * Updates '_lastPushedTimestamp' based on the last document in the collection. + */ + void _updateLastPushedTimestampFromCollection(WithLock, OperationContext* opCtx); + + /** * Returns the document with the given timestamp, or ErrorCodes::NoSuchKey if not found. */ StatusWith<BSONObj> _getDocumentWithTimestamp(OperationContext* opCtx, const Timestamp& ts); diff --git a/src/mongo/db/repl/oplog_buffer_collection_test.cpp b/src/mongo/db/repl/oplog_buffer_collection_test.cpp index 35cdb2d89a8..f93278f45e3 100644 --- a/src/mongo/db/repl/oplog_buffer_collection_test.cpp +++ b/src/mongo/db/repl/oplog_buffer_collection_test.cpp @@ -839,6 +839,22 @@ DEATH_TEST_REGEX_F(OplogBufferCollectionTest, oplogBuffer.push(_opCtx.get(), oplog.begin(), oplog.end()); } +TEST_F(OplogBufferCollectionTest, PreloadAllNonBlockingSucceeds) { + auto nss = makeNamespace(_agent); + OplogBufferCollection oplogBuffer(_storageInterface, nss); + + oplogBuffer.startup(_opCtx.get()); + const std::vector<BSONObj> oplog = { + makeOplogEntry(2), + makeOplogEntry(1), + }; + ASSERT_EQUALS(oplogBuffer.getCount(), 0UL); + oplogBuffer.preload(_opCtx.get(), oplog.begin(), oplog.end()); + ASSERT_EQUALS(oplogBuffer.getCount(), 2UL); + ASSERT_NOT_EQUALS(oplogBuffer.getSize(), 0UL); + ASSERT_EQUALS(Timestamp(2, 2), oplogBuffer.getLastPushedTimestamp()); +} + OplogBufferCollection::Options _makeOptions(std::size_t peekCacheSize) { OplogBufferCollection::Options options; diff --git a/src/mongo/db/repl/session_update_tracker.cpp b/src/mongo/db/repl/session_update_tracker.cpp index bbd25a8b47d..99f1779c8c2 100644 --- a/src/mongo/db/repl/session_update_tracker.cpp +++ b/src/mongo/db/repl/session_update_tracker.cpp @@ -113,7 +113,7 @@ boost::optional<repl::OplogEntry> createMatchingTransactionTableUpdate( * 2) Be a no-op entry * 3) Have sessionId and txnNumber */ -bool isTransactionEntryFromTenantMigrations(OplogEntry& entry) { +bool isTransactionEntryFromTenantMigrations(const OplogEntry& entry) { if (!entry.getFromTenantMigration()) { return false; } @@ -122,6 +122,18 @@ bool isTransactionEntryFromTenantMigrations(OplogEntry& entry) { return false; } + // Transaction no-op entries will have an o2 with a command, or an o2 with no optype + // field (for entries generated from config.transactions). Retryable writes will have + // entries with optypes other than command. + if (entry.getObject2()) { + auto innerOpTypeStr = + (*entry.getObject2())[OplogEntry::kOpTypeFieldName].valueStringDataSafe(); + if (!innerOpTypeStr.empty() && + OpType_parse(IDLParserErrorContext("isTransactionEntryFromTenantMigration"_sd), + innerOpTypeStr) != OpTypeEnum::kCommand) + return false; + } + if (!entry.getSessionId() || !entry.getTxnNumber()) { return false; } @@ -129,10 +141,9 @@ bool isTransactionEntryFromTenantMigrations(OplogEntry& entry) { return true; } -/** - * Returns true if the oplog entry represents an operation in a transaction and false otherwise. - */ -bool isTransactionEntry(OplogEntry entry) { +} // namespace + +bool SessionUpdateTracker::isTransactionEntry(const OplogEntry& entry) { if (isTransactionEntryFromTenantMigrations(entry)) { return true; } @@ -148,8 +159,6 @@ bool isTransactionEntry(OplogEntry entry) { entry.getCommandType() == repl::OplogEntry::CommandType::kApplyOps; } -} // namespace - boost::optional<std::vector<OplogEntry>> SessionUpdateTracker::_updateOrFlush( const OplogEntry& entry) { const auto& ns = entry.getNss(); diff --git a/src/mongo/db/repl/session_update_tracker.h b/src/mongo/db/repl/session_update_tracker.h index 4d84af6be96..2ef43795ff7 100644 --- a/src/mongo/db/repl/session_update_tracker.h +++ b/src/mongo/db/repl/session_update_tracker.h @@ -63,6 +63,12 @@ public: */ boost::optional<std::vector<OplogEntry>> updateSession(const OplogEntry& entry); + /** + * Returns true if the oplog entry represents an operation in a transaction and false otherwise. + * No-ops representing migrated transactions are considered transaction operations. + */ + static bool isTransactionEntry(const OplogEntry& entry); + private: /** * Analyzes the given oplog entry and determines which transactions stored so far needs to be diff --git a/src/mongo/db/repl/tenant_migration_recipient_service.cpp b/src/mongo/db/repl/tenant_migration_recipient_service.cpp index 801224c81ce..f1fa388fed3 100644 --- a/src/mongo/db/repl/tenant_migration_recipient_service.cpp +++ b/src/mongo/db/repl/tenant_migration_recipient_service.cpp @@ -1124,7 +1124,7 @@ TenantMigrationRecipientService::Instance::_fetchRetryableWritesOplogBeforeStart // Wait for enough space. _donorOplogBuffer->waitForSpace(opCtx.get(), toApplyDocumentBytes); // Buffer retryable writes entries. - _donorOplogBuffer->push( + _donorOplogBuffer->preload( opCtx.get(), retryableWritesEntries.begin(), retryableWritesEntries.end()); } diff --git a/src/mongo/db/repl/tenant_oplog_applier.cpp b/src/mongo/db/repl/tenant_oplog_applier.cpp index a4251de3a32..d27b03ced65 100644 --- a/src/mongo/db/repl/tenant_oplog_applier.cpp +++ b/src/mongo/db/repl/tenant_oplog_applier.cpp @@ -47,6 +47,7 @@ #include "mongo/db/repl/oplog_applier_utils.h" #include "mongo/db/repl/repl_client_info.h" #include "mongo/db/repl/repl_server_parameters_gen.h" +#include "mongo/db/repl/session_update_tracker.h" #include "mongo/db/repl/tenant_migration_decoration.h" #include "mongo/db/repl/tenant_migration_recipient_service.h" #include "mongo/db/repl/tenant_oplog_batcher.h" @@ -515,6 +516,20 @@ void TenantOplogApplier::_writeSessionNoOpsForRange( // forward. repl::ReplClientInfo::forClient(opCtx->getClient()).clearLastOp(); + // All the ops will have the same session, so we can retain the scopedSession throughout + // the loop, except when invalidated by multi-document transactions. This allows us to + // track the statements in a retryable write. + boost::optional<MongoDOperationContextSessionWithoutOplogRead> scopedSession; + + // Make sure a partial session doesn't escape. + ON_BLOCK_EXIT([this, &scopedSession, &opCtx] { + if (scopedSession) { + auto txnParticipant = TransactionParticipant::get(opCtx.get()); + invariant(txnParticipant); + txnParticipant.invalidate(opCtx.get()); + } + }); + for (auto iter = begin; iter != end; iter++) { const auto& entry = *iter->first; invariant(!isResumeTokenNoop(entry)); @@ -529,8 +544,9 @@ void TenantOplogApplier::_writeSessionNoOpsForRange( noopEntry.setOpTime(*iter->second); noopEntry.setWallClockTime(opCtx->getServiceContext()->getFastClockSource()->now()); - boost::optional<MongoDOperationContextSession> scopedSession; boost::optional<SessionTxnRecord> sessionTxnRecord; + std::vector<StmtId> stmtIds; + boost::optional<OpTime> prevWriteOpTime = boost::none; if (entry.getTxnNumber() && !entry.isPartialTransaction() && (entry.getCommandType() == repl::OplogEntry::CommandType::kCommitTransaction || entry.getCommandType() == repl::OplogEntry::CommandType::kApplyOps)) { @@ -550,7 +566,8 @@ void TenantOplogApplier::_writeSessionNoOpsForRange( "op"_attr = redact(entry.toBSONForLogging())); // Check out the session. - scopedSession.emplace(opCtx.get()); + if (!scopedSession) + scopedSession.emplace(opCtx.get()); auto txnParticipant = TransactionParticipant::get(opCtx.get()); uassert( 5351500, @@ -574,16 +591,80 @@ void TenantOplogApplier::_writeSessionNoOpsForRange( // Use the same wallclock time as the noop entry. sessionTxnRecord.emplace(sessionId, txnNumber, OpTime(), noopEntry.getWallClockTime()); sessionTxnRecord->setState(DurableTxnStateEnum::kCommitted); + } else if (entry.getStatementId() && !SessionUpdateTracker::isTransactionEntry(entry)) { + // If it has a statement id but isn't a transaction, it's a retryable write. + const auto& sessionId = *entry.getSessionId(); + const auto& txnNumber = *entry.getTxnNumber(); + const auto& stmtId = *entry.getStatementId(); + if (entry.getOpType() == repl::OpTypeEnum::kNoop) { + // TODO(SERVER-53510): handle pre and post image no-ops + LOGV2_DEBUG(5350903, + 2, + "Skipping retryable write no-op", + "entry"_attr = entry.getEntry(), + "sessionId"_attr = sessionId, + "txnNumber"_attr = txnNumber, + "statementId"_attr = stmtId, + "tenant"_attr = _tenantId, + "migrationUuid"_attr = _migrationUuid); + continue; + } + stmtIds.push_back(stmtId); + + LOGV2_DEBUG(5350901, + 2, + "Tenant Oplog Applier processing retryable write", + "sessionId"_attr = sessionId, + "txnNumber"_attr = txnNumber, + "statementId"_attr = stmtId, + "tenant"_attr = _tenantId, + "migrationUuid"_attr = _migrationUuid); + opCtx->setLogicalSessionId(sessionId); + opCtx->setTxnNumber(txnNumber); + if (!scopedSession) + scopedSession.emplace(opCtx.get()); + auto txnParticipant = TransactionParticipant::get(opCtx.get()); + uassert(5350900, + str::stream() << "Tenant oplog application failed to get retryable write " + "for transaction " + << txnNumber << " on session " << sessionId, + txnParticipant); + // beginOrContinue throws on failure, which will abort the migration. Failure should + // only result from out-of-order processing, which should not happen. + txnParticipant.beginOrContinue(opCtx.get(), + txnNumber, + boost::none /* autocommit */, + boost::none /* startTransaction */); + + // We should never process the same donor statement twice, except in failover + // cases where we'll also have "forgotten" the statement was executed. + uassert(5350902, + str::stream() << "Tenant oplog application processed same retryable write " + "twice for transaction " + << txnNumber << " statement " << stmtId << " on session " + << sessionId, + !txnParticipant.checkStatementExecutedNoOplogEntryFetch(stmtId)); + prevWriteOpTime = txnParticipant.getLastWriteOpTime(); + + // Set sessionId, txnNumber, and statementId for all ops in a retryable write. + noopEntry.setSessionId(sessionId); + noopEntry.setTxnNumber(txnNumber); + noopEntry.setStatementId(stmtId); + + // set fromMigrate on the no-op so the session update tracker recognizes it. + noopEntry.setFromMigrate(true); + + // Use the same wallclock time as the noop entry. The lastWriteOpTime will be filled + // in after the no-op is written. + sessionTxnRecord.emplace(sessionId, txnNumber, OpTime(), noopEntry.getWallClockTime()); } // TODO(SERVER-53510) Correctly fill in pre-image and post-image op times. const boost::optional<OpTime> preImageOpTime = boost::none; const boost::optional<OpTime> postImageOpTime = boost::none; - // TODO(SERVER-53509) Correctly fill in prevWriteOpTime for retryable writes. - const boost::optional<OpTime> prevWriteOpTimeInTransaction = boost::none; noopEntry.setPreImageOpTime(preImageOpTime); noopEntry.setPostImageOpTime(postImageOpTime); - noopEntry.setPrevWriteOpTimeInTransaction(prevWriteOpTimeInTransaction); + noopEntry.setPrevWriteOpTimeInTransaction(prevWriteOpTime); AutoGetOplog oplogWrite(opCtx.get(), OplogAccessMode::kWrite); writeConflictRetry( @@ -591,10 +672,15 @@ void TenantOplogApplier::_writeSessionNoOpsForRange( WriteUnitOfWork wuow(opCtx.get()); // Write the noop entry and update config.transactions. - repl::logOp(opCtx.get(), &noopEntry); + auto oplogOpTime = repl::logOp(opCtx.get(), &noopEntry); if (sessionTxnRecord) { + // We do not need to record the last write op time for migrated transactions, so + // we leave it null for consistency with transactions completed before the + // migration.. + if (!opCtx->inMultiDocumentTransaction()) + sessionTxnRecord->setLastWriteOpTime(oplogOpTime); TransactionParticipant::get(opCtx.get()) - .onWriteOpCompletedOnPrimary(opCtx.get(), {}, *sessionTxnRecord); + .onWriteOpCompletedOnPrimary(opCtx.get(), {stmtIds}, *sessionTxnRecord); } wuow.commit(); @@ -607,6 +693,7 @@ void TenantOplogApplier::_writeSessionNoOpsForRange( invariant(txnParticipant); txnParticipant.invalidate(opCtx.get()); opCtx->resetMultiDocumentTransactionState(); + scopedSession = boost::none; } } } diff --git a/src/mongo/db/session_catalog_mongod.cpp b/src/mongo/db/session_catalog_mongod.cpp index a8a8ff9d7d6..46ea3c29f05 100644 --- a/src/mongo/db/session_catalog_mongod.cpp +++ b/src/mongo/db/session_catalog_mongod.cpp @@ -443,4 +443,16 @@ MongoDOperationContextSessionWithoutRefresh::~MongoDOperationContextSessionWitho invariant(!txnParticipant.transactionIsInProgress()); } +MongoDOperationContextSessionWithoutOplogRead::MongoDOperationContextSessionWithoutOplogRead( + OperationContext* opCtx) + : _operationContextSession(opCtx), _opCtx(opCtx) { + invariant(!opCtx->getClient()->isInDirectClient()); + + auto txnParticipant = TransactionParticipant::get(opCtx); + txnParticipant.refreshFromStorageIfNeededNoOplogEntryFetch(opCtx); +} + +MongoDOperationContextSessionWithoutOplogRead::~MongoDOperationContextSessionWithoutOplogRead() = + default; + } // namespace mongo diff --git a/src/mongo/db/session_catalog_mongod.h b/src/mongo/db/session_catalog_mongod.h index cd379edb8ec..a373b9c218c 100644 --- a/src/mongo/db/session_catalog_mongod.h +++ b/src/mongo/db/session_catalog_mongod.h @@ -121,4 +121,21 @@ private: OperationContext* const _opCtx; }; +/** + * Similar to MongoDOperationContextSession, but marks the TransactionParticipant as valid without + * loading the retryable write oplog history. If the last operation was a multi-document + * transaction, is equivalent to MongoDOperationContextSession. + * + * NOTE: Should only be used when reading the oplog history is not possible. + */ +class MongoDOperationContextSessionWithoutOplogRead { +public: + MongoDOperationContextSessionWithoutOplogRead(OperationContext* opCtx); + ~MongoDOperationContextSessionWithoutOplogRead(); + +private: + OperationContextSession _operationContextSession; + OperationContext* const _opCtx; +}; + } // namespace mongo diff --git a/src/mongo/db/transaction_participant.cpp b/src/mongo/db/transaction_participant.cpp index 1d43b5c1679..17bbde43b7e 100644 --- a/src/mongo/db/transaction_participant.cpp +++ b/src/mongo/db/transaction_participant.cpp @@ -123,7 +123,8 @@ struct ActiveTransactionHistory { }; ActiveTransactionHistory fetchActiveTransactionHistory(OperationContext* opCtx, - const LogicalSessionId& lsid) { + const LogicalSessionId& lsid, + bool fetchOplogEntries) { // Storage engine operations require at least Global IS. Lock::GlobalLock lk(opCtx, MODE_IS); @@ -159,6 +160,10 @@ ActiveTransactionHistory fetchActiveTransactionHistory(OperationContext* opCtx, return result; } + if (!fetchOplogEntries) { + return result; + } + auto it = TransactionHistoryIterator(result.lastTxnRecord->getLastWriteOpTime()); while (it.hasNext()) { try { @@ -2182,13 +2187,23 @@ void TransactionParticipant::Participant::_setNewTxnNumber(OperationContext* opC } void TransactionParticipant::Participant::refreshFromStorageIfNeeded(OperationContext* opCtx) { + return _refreshFromStorageIfNeeded(opCtx, true); +} + +void TransactionParticipant::Participant::refreshFromStorageIfNeededNoOplogEntryFetch( + OperationContext* opCtx) { + return _refreshFromStorageIfNeeded(opCtx, false); +} + +void TransactionParticipant::Participant::_refreshFromStorageIfNeeded(OperationContext* opCtx, + bool fetchOplogEntries) { invariant(!opCtx->getClient()->isInDirectClient()); invariant(!opCtx->lockState()->isLocked()); if (p().isValid) return; - auto activeTxnHistory = fetchActiveTransactionHistory(opCtx, _sessionId()); + auto activeTxnHistory = fetchActiveTransactionHistory(opCtx, _sessionId(), fetchOplogEntries); const auto& lastTxnRecord = activeTxnHistory.lastTxnRecord; if (lastTxnRecord) { stdx::lock_guard<Client> lg(*opCtx->getClient()); diff --git a/src/mongo/db/transaction_participant.h b/src/mongo/db/transaction_participant.h index 26579b7bf99..a382c466e1a 100644 --- a/src/mongo/db/transaction_participant.h +++ b/src/mongo/db/transaction_participant.h @@ -383,6 +383,12 @@ public: */ void refreshFromStorageIfNeeded(OperationContext* opCtx); + /* + * Same as above, but does not retrieve full transaction history and should be called + * only when oplog reads are not possible. + */ + void refreshFromStorageIfNeededNoOplogEntryFetch(OperationContext* opCtx); + /** * Starts a new transaction (and if the txnNumber is newer aborts any in-progress * transaction on the session), or continues an already active transaction. @@ -772,6 +778,9 @@ public: // number. void _continueMultiDocumentTransaction(OperationContext* opCtx, TxnNumber txnNumber); + // Implementation of public refreshFromStorageIfNeeded methods. + void _refreshFromStorageIfNeeded(OperationContext* opCtx, bool fetchOplogEntries); + // Helper that invalidates the session state and activeTxnNumber. Also resets the single // transaction stats because the session is no longer valid. void _invalidate(WithLock); |