// Set minValid to the last OpTime that needs to be applied, in this batch or from the
        // (last) failed batch, whichever is larger.
        // This will cause this node to go into RECOVERING state
        // if we should crash and restart before updating finishing.
        const auto& start = lastWriteOpTime;


        // Take the max of the first endOptime (if we recovered) and the end of our batch.

        // Setting end to the max of originalEndOpTime and lastOpTime (the end of the batch)
        // ensures that we keep pushing out the point where we can become consistent
        // and allow reads. If we recover and end up doing smaller batches we must pass the
        // originalEndOpTime before we are good.
        //
        // For example:
        // batch apply, 20-40, end = 40
        // batch failure,
        // restart
        // batch apply, 20-25, end = max(25, 40) = 40
        // batch apply, 25-45, end = 45
        const OpTime end(std::max(originalEndOpTime, lastOpTime));

        // This write will not journal/checkpoint.
        StorageInterface::get(&txn)->setMinValid(&txn, {start, end});       <<<<-------------------,
                                                                                                   |
        const size_t opsInBatch = ops.getCount();                                                  |  Race "on the way in"
        lastWriteOpTime = multiApply(&txn, ops.releaseBatch());             <<<<-------------------+
        if (lastWriteOpTime.isNull()) {                                                            |
            // fassert if oplog application failed for any reasons other than shutdown.            |
            error() << "Failed to apply " << opsInBatch << " operations - batch start:" << start   |
                    << " end:" << end;                                                             |
            fassert(34360, inShutdownStrict());                                                    |
            // Return without setting minvalid in the case of shutdown.                            |
            return;                                                                                |  Race "out the way out"
        }                                                                                          |
                                                                                                   |
        setNewTimestamp(lastWriteOpTime.getTimestamp());                                           |
        StorageInterface::get(&txn)->setMinValid(&txn, end, DurableRequirement::None);   <<<<------'
        minValidBoundaries.start = {};
        minValidBoundaries.end = end;
        finalizer->record(lastWriteOpTime);

StatusWith<OpTime> multiApply(OperationContext* txn,
                              OldThreadPool* workerPool,
                              MultiApplier::Operations ops,
                              MultiApplier::ApplyOperationFn applyOperation) {
    if (!txn) {
        return {ErrorCodes::BadValue, "invalid operation context"};
    }

    if (!workerPool) {
        return {ErrorCodes::BadValue, "invalid worker pool"};
    }

    if (ops.empty()) {
        return {ErrorCodes::EmptyArrayOperation, "no operations provided to multiApply"};
    }

    if (!applyOperation) {
        return {ErrorCodes::BadValue, "invalid apply operation function"};
    }

    if (getGlobalServiceContext()->getGlobalStorageEngine()->isMmapV1()) {
        // Use a ThreadPool to prefetch all the operations in a batch.
        prefetchOps(ops, workerPool);
    }

    LOG(2) << "replication batch size is " << ops.size();
    // We must grab this because we're going to grab write locks later.
    // We hold this mutex the entire time we're writing; it doesn't matter
    // because all readers are blocked anyway.
    stdx::lock_guard<SimpleMutex> fsynclk(filesLockedFsync);

    // Stop all readers until we're done. This also prevents doc-locking engines from deleting old
    // entries from the oplog until we finish writing.
    Lock::ParallelBatchWriterMode pbwm(txn->lockState());

    auto replCoord = ReplicationCoordinator::get(txn);
    if (replCoord->getMemberState().primary() && !replCoord->isWaitingForApplierToDrain()) {
        severe() << "attempting to replicate ops while primary";
        return {ErrorCodes::CannotApplyOplogWhilePrimary,
                "attempting to replicate ops while primary"};
    }

    ...