SERVER-29127 Improve behavior of tailable awaitData cursors with readConcern:majority

Fixes a bug with awaitData cursors on readConcern:majority; they would not be awakened when a new
snapshot was set to "committed".  Also removed a related false comment.

Fixes a bug in planExecutor which would prevent awakening when oplog visibility changed with
a visibility-restricted cursor active.

Author: mtrussotto

buildscripts/resmokeconfig/suites/aggregation_read_concern_majority_passthrough.yml

@@ -17,12 +17,6 @@ selector:
                                                              # secondaryThrottle is not set.
   - jstests/aggregation/sources/facet/use_cases.js # Cannot specify write concern when
                                                    # secondaryThrottle is not set.
-
-  # TODO (SERVER-29127) The CappedInsertNotifier wakes up the change stream's collection scan before
-  # it can see the new insert, so it will immediately see EOF and go back to sleep. This will cause
-  # the test to fail, since it is expecting an insert to immediately wake up the change stream and
-  # return the new result.
-  - jstests/aggregation/sources/changeStream/only_wake_getmore_for_relevant_changes.js
   - jstests/aggregation/testSlave.js # Majority read on secondary requires afterOpTime.
   - jstests/aggregation/sources/out/*.js # Cannot specify write concern when secondaryThrottle is not set.
 

src/mongo/db/query/plan_executor.cpp

@@ -65,6 +65,11 @@ const OperationContext::Decoration<bool> shouldWaitForInserts =
 const OperationContext::Decoration<repl::OpTime> clientsLastKnownCommittedOpTime =
     OperationContext::declareDecoration<repl::OpTime>();
 
+struct CappedInsertNotifierData {
+    shared_ptr<CappedInsertNotifier> notifier;
+    uint64_t lastEOFVersion = ~0;
+};
+
 namespace {
 
 namespace {
@@ -407,30 +412,45 @@ bool PlanExecutor::shouldWaitForInserts() {
     return false;
 }
 
-bool PlanExecutor::waitForInserts() {
-    // If we cannot yield, we should retry immediately.
+std::shared_ptr<CappedInsertNotifier> PlanExecutor::getCappedInsertNotifier() {
+    // If we cannot yield, we should retry immediately when we hit EOF, so do not get
+    // a CappedInsertNotifier.
     if (!_yieldPolicy->canReleaseLocksDuringExecution())
-        return true;
+        return nullptr;
 
-    // We can only wait if we have a collection; otherwise retry immediately.
+    // We can only wait if we have a collection; otherwise we should retry immediately when
+    // we hit EOF.
     dassert(_opCtx->lockState()->isCollectionLockedForMode(_nss.ns(), MODE_IS));
     auto db = dbHolder().get(_opCtx, _nss.db());
     if (!db)
-        return true;
+        return nullptr;
     auto collection = db->getCollection(_opCtx, _nss);
     if (!collection)
+        return nullptr;
+
+    return collection->getCappedInsertNotifier();
+}
+
+bool PlanExecutor::waitForInserts(CappedInsertNotifierData* notifierData) {
+    // We tested to see if we could wait when getting the CappedInsertNotifier.
+    if (!notifierData->notifier)
         return true;
 
-    auto notifier = collection->getCappedInsertNotifier();
-    uint64_t notifierVersion = notifier->getVersion();
+    // The notifier wait() method will not wait unless the version passed to it matches the
+    // current version of the notifier.  Since the version passed to it is the current version
+    // of the notifier at the time of the previous EOF, we require two EOFs in a row with no
+    // notifier version change in order to wait.  This is sufficient to ensure we never wait
+    // when data is available.
     auto curOp = CurOp::get(_opCtx);
     curOp->pauseTimer();
     ON_BLOCK_EXIT([curOp] { curOp->resumeTimer(); });
     auto opCtx = _opCtx;
-    bool yieldResult = _yieldPolicy->yield(nullptr, [opCtx, notifier, notifierVersion] {
+    uint64_t currentNotifierVersion = notifierData->notifier->getVersion();
+    bool yieldResult = _yieldPolicy->yield(nullptr, [opCtx, notifierData] {
         const auto timeout = opCtx->getRemainingMaxTimeMicros();
-        notifier->wait(notifierVersion, timeout);
+        notifierData->notifier->wait(notifierData->lastEOFVersion, timeout);
     });
+    notifierData->lastEOFVersion = currentNotifierVersion;
     return yieldResult;
 }
 
@@ -471,6 +491,13 @@ PlanExecutor::ExecState PlanExecutor::getNextImpl(Snapshotted<BSONObj>* objOut,
     // Incremented on every writeConflict, reset to 0 on any successful call to _root->work.
     size_t writeConflictsInARow = 0;
 
+    // Capped insert data; declared outside the loop so we hold a shared pointer to the capped
+    // insert notifier the entire time we are in the loop.  Holding a shared pointer to the capped
+    // insert notifier is necessary for the notifierVersion to advance.
+    CappedInsertNotifierData cappedInsertNotifierData;
+    if (shouldWaitForInserts()) {
+        cappedInsertNotifierData.notifier = getCappedInsertNotifier();
+    }
     for (;;) {
         // These are the conditions which can cause us to yield:
         //   1) The yield policy's timer elapsed, or
@@ -563,7 +590,7 @@ PlanExecutor::ExecState PlanExecutor::getNextImpl(Snapshotted<BSONObj>* objOut,
             // Fall through to yield check at end of large conditional.
         } else if (PlanStage::IS_EOF == code) {
             if (shouldWaitForInserts()) {
-                const bool locksReacquiredAfterYield = waitForInserts();
+                const bool locksReacquiredAfterYield = waitForInserts(&cappedInsertNotifierData);
                 if (locksReacquiredAfterYield) {
                     // There may be more results, try to get more data.
                     continue;

src/mongo/db/query/plan_executor.h

@@ -41,6 +41,8 @@
 namespace mongo {
 
 class BSONObj;
+class CappedInsertNotifier;
+struct CappedInsertNotifierData;
 class Collection;
 class CursorManager;
 class PlanExecutor;
@@ -447,10 +449,14 @@ class PlanExecutor {
     // should be returned immediately.
     bool shouldWaitForInserts();
 
+    // Gets the CappedInsertNotifier for a capped collection.  Returns nullptr if this plan executor
+    // is not capable of yielding based on a notifier.
+    std::shared_ptr<CappedInsertNotifier> getCappedInsertNotifier();
+
     // Yields locks and waits for inserts to the collection.  Returns true if there may be new
     // inserts, false if there is a timeout or an interrupt.  If this planExecutor cannot yield,
     // returns true immediately.
-    bool waitForInserts();
+    bool waitForInserts(CappedInsertNotifierData* notifierData);
 
     ExecState getNextImpl(Snapshotted<BSONObj>* objOut, RecordId* dlOut);
 

src/mongo/db/repl/replication_coordinator_external_state_impl.cpp

@@ -799,6 +799,7 @@ void ReplicationCoordinatorExternalStateImpl::updateCommittedSnapshot(SnapshotIn
     auto manager = _service->getGlobalStorageEngine()->getSnapshotManager();
     invariant(manager);  // This should never be called if there is no SnapshotManager.
     manager->setCommittedSnapshot(newCommitPoint.name, newCommitPoint.opTime.getTimestamp());
+    notifyOplogMetadataWaiters(newCommitPoint.opTime);
 }
 
 void ReplicationCoordinatorExternalStateImpl::createSnapshot(OperationContext* opCtx,

src/mongo/db/repl/replication_coordinator_impl.cpp

@@ -3039,7 +3039,6 @@ void ReplicationCoordinatorImpl::_advanceCommitPoint_inlock(const OpTime& commit
 
 void ReplicationCoordinatorImpl::_updateCommitPoint_inlock() {
     auto committedOpTime = _topCoord->getLastCommittedOpTime();
-    _externalState->notifyOplogMetadataWaiters(committedOpTime);
 
     // Update the stable timestamp.
     _setStableTimestampForStorage_inlock();
@@ -3060,10 +3059,11 @@ void ReplicationCoordinatorImpl::_updateCommitPoint_inlock() {
 
         // Update committed snapshot and wake up any threads waiting on read concern or
         // write concern.
-        //
-        // This function is only called on secondaries, so only threads waiting for
-        // committed snapshot need to be woken up.
         _updateCommittedSnapshot_inlock(newSnapshot);
+    } else {
+        // Even if we have no new snapshot, we need to notify waiters that the commit point
+        // moved.
+        _externalState->notifyOplogMetadataWaiters(committedOpTime);
     }
 }