SERVER-71145 Allow observed `all_durable` to move backward

3 files changed, 27 insertions, 87 deletions

diff --git a/src/mongo/db/storage/kv/kv_engine_test_harness.cpp b/src/mongo/db/storage/kv/kv_engine_test_harness.cpp
index 69f23a73803..3dc5d407488 100644
--- a/src/mongo/db/storage/kv/kv_engine_test_harness.cpp
+++ b/src/mongo/db/storage/kv/kv_engine_test_harness.cpp
@@ -351,41 +351,29 @@ TEST_F(KVEngineTestHarness, AllDurableTimestamp) {
         Timestamp t52(5, 2);
         Timestamp t61(6, 1);
 
-        Timestamp allDurable = engine->getAllDurableTimestamp();
-        ASSERT_EQ(allDurable, Timestamp(StorageEngine::kMinimumTimestamp));
+        ASSERT_EQ(engine->getAllDurableTimestamp(), Timestamp(StorageEngine::kMinimumTimestamp));
 
         auto opCtx1 = opCtxs[0].second.get();
         WriteUnitOfWork uow1(opCtx1);
         ASSERT_OK(rs->insertRecord(opCtx1, "abc", 4, t51));
 
-        Timestamp lastAllDurable = allDurable;
-        allDurable = engine->getAllDurableTimestamp();
-        ASSERT_GT(allDurable, lastAllDurable);
-        ASSERT_EQ(allDurable, t51);
+        // TODO (SERVER-71148): The all_durable timestamp should be t51-1 here.
+        ASSERT_EQ(engine->getAllDurableTimestamp(), t51);
 
         auto opCtx2 = opCtxs[1].second.get();
         WriteUnitOfWork uow2(opCtx2);
         ASSERT_OK(rs->insertRecord(opCtx2, "abc", 4, t61));
         uow2.commit();
 
-        lastAllDurable = allDurable;
-        allDurable = engine->getAllDurableTimestamp();
-        ASSERT_EQ(allDurable, lastAllDurable);
-        ASSERT_EQ(allDurable, t51);
+        ASSERT_EQ(engine->getAllDurableTimestamp(), t51 - 1);
 
         ASSERT_OK(rs->insertRecord(opCtx1, "abc", 4, t52));
 
-        lastAllDurable = allDurable;
-        allDurable = engine->getAllDurableTimestamp();
-        ASSERT_EQ(allDurable, lastAllDurable);
-        ASSERT_EQ(allDurable, t51);
+        ASSERT_EQ(engine->getAllDurableTimestamp(), t51 - 1);
 
         uow1.commit();
 
-        lastAllDurable = allDurable;
-        allDurable = engine->getAllDurableTimestamp();
-        ASSERT_GT(allDurable, lastAllDurable);
-        ASSERT_EQ(allDurable, t61);
+        ASSERT_EQ(engine->getAllDurableTimestamp(), t61);
     }
 }
 
@@ -461,11 +449,12 @@ TEST_F(KVEngineTestHarness, PinningOldestWithAnotherSession) {
  * | Session 1            | Session 2            | GlobalActor                      |
  * |----------------------+----------------------+----------------------------------|
  * | Begin                |                      |                                  |
- * | Commit :commit 10    |                      |                                  |
+ * | Timestamp :commit 10 |                      |                                  |
+ * | Commit               |                      |                                  |
  * |                      |                      | QueryTimestamp :all_durable (10) |
  * | Begin                |                      |                                  |
  * | Timestamp :commit 20 |                      |                                  |
- * |                      |                      | QueryTimestamp :all_durable (19) |
+ * |                      |                      | QueryTimestamp :all_durable (10) |
  * |                      | Begin                |                                  |
  * |                      | Timestamp :commit 30 |                                  |
  * |                      | Commit               |                                  |
@@ -474,7 +463,7 @@ TEST_F(KVEngineTestHarness, PinningOldestWithAnotherSession) {
  * |                      |                      | QueryTimestamp :all_durable (30) |
  * | Begin                |                      |                                  |
  * | Timestamp :commit 25 |                      |                                  |
- * |                      |                      | QueryTimestamp :all_durable (30) |
+ * |                      |                      | QueryTimestamp :all_durable (24) |
  */
 TEST_F(KVEngineTestHarness, AllDurable) {
     std::unique_ptr<KVHarnessHelper> helper(KVHarnessHelper::create(getServiceContext()));
@@ -498,27 +487,20 @@ TEST_F(KVEngineTestHarness, AllDurable) {
         const Timestamp kInsertTimestamp3 = Timestamp(30, 30);
         const Timestamp kInsertTimestamp4 = Timestamp(25, 25);
 
-        Timestamp allDurable = engine->getAllDurableTimestamp();
         auto opCtx1 = opCtxs[0].second.get();
         WriteUnitOfWork uow1(opCtx1);
         auto swRid = rs->insertRecord(opCtx1, "abc", 4, kInsertTimestamp1);
         ASSERT_OK(swRid);
         uow1.commit();
 
-        Timestamp lastAllDurable = allDurable;
-        allDurable = engine->getAllDurableTimestamp();
-        ASSERT_GTE(allDurable, lastAllDurable);
-        ASSERT_LTE(allDurable, kInsertTimestamp1);
+        ASSERT_EQ(engine->getAllDurableTimestamp(), kInsertTimestamp1);
 
         auto opCtx2 = opCtxs[1].second.get();
         WriteUnitOfWork uow2(opCtx2);
         swRid = rs->insertRecord(opCtx2, "abc", 4, kInsertTimestamp2);
         ASSERT_OK(swRid);
 
-        lastAllDurable = allDurable;
-        allDurable = engine->getAllDurableTimestamp();
-        ASSERT_GTE(allDurable, lastAllDurable);
-        ASSERT_LT(allDurable, kInsertTimestamp2);
+        ASSERT_EQ(engine->getAllDurableTimestamp(), kInsertTimestamp1);
 
         auto opCtx3 = opCtxs[2].second.get();
         WriteUnitOfWork uow3(opCtx3);
@@ -526,27 +508,19 @@ TEST_F(KVEngineTestHarness, AllDurable) {
         ASSERT_OK(swRid);
         uow3.commit();
 
-        lastAllDurable = allDurable;
-        allDurable = engine->getAllDurableTimestamp();
-        ASSERT_GTE(allDurable, lastAllDurable);
-        ASSERT_LT(allDurable, kInsertTimestamp2);
+        ASSERT_EQ(engine->getAllDurableTimestamp(), kInsertTimestamp2 - 1);
 
         uow2.commit();
 
-        lastAllDurable = allDurable;
-        allDurable = engine->getAllDurableTimestamp();
-        ASSERT_GTE(allDurable, lastAllDurable);
-        ASSERT_LTE(allDurable, kInsertTimestamp3);
+        ASSERT_EQ(engine->getAllDurableTimestamp(), kInsertTimestamp3);
 
         auto opCtx4 = opCtxs[3].second.get();
         WriteUnitOfWork uow4(opCtx4);
         swRid = rs->insertRecord(opCtx4, "abc", 4, kInsertTimestamp4);
         ASSERT_OK(swRid);
 
-        lastAllDurable = allDurable;
-        allDurable = engine->getAllDurableTimestamp();
-        ASSERT_GTE(allDurable, lastAllDurable);
-        ASSERT_LTE(allDurable, kInsertTimestamp3);
+        ASSERT_EQ(engine->getAllDurableTimestamp(), kInsertTimestamp4 - 1);
+
         uow4.commit();
     }
 }
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.cpp
index f98b8c0993a..9aea381762f 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.cpp
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.cpp
@@ -560,10 +560,6 @@ WiredTigerKVEngine::WiredTigerKVEngine(OperationContext* opCtx,
             setStableTimestamp(_recoveryTimestamp, false);
 
             _sessionCache->snapshotManager().setLastApplied(_recoveryTimestamp);
-            {
-                stdx::lock_guard<Latch> lk(_highestDurableTimestampMutex);
-                _highestSeenDurableTimestamp = _recoveryTimestamp.asULL();
-            }
         }
     }
 
@@ -2125,25 +2121,6 @@ void WiredTigerKVEngine::setJournalListener(JournalListener* jl) {
     return _sessionCache->setJournalListener(jl);
 }
 
-namespace {
-uint64_t _fetchAllDurableValue(WT_CONNECTION* conn) {
-    // Fetch the latest all_durable value from the storage engine. This value will be a timestamp
-    // that has no holes (uncommitted transactions with lower timestamps) behind it.
-    char buf[(2 * 8 /*bytes in hex*/) + 1 /*nul terminator*/];
-    invariantWTOK(conn->query_timestamp(conn, buf, "get=all_durable"), nullptr);
-
-    uint64_t tmp;
-    fassert(38002, NumberParser().base(16)(buf, &tmp));
-    if (tmp == 0) {
-        // Treat this as lowest possible timestamp; we need to see all preexisting data but no new
-        // (timestamped) data.
-        return StorageEngine::kMinimumTimestamp;
-    }
-
-    return tmp;
-}
-}  // namespace
-
 void WiredTigerKVEngine::setStableTimestamp(Timestamp stableTimestamp, bool force) {
     if (MONGO_unlikely(WTPauseStableTimestamp.shouldFail())) {
         return;
@@ -2179,8 +2156,6 @@ void WiredTigerKVEngine::setStableTimestamp(Timestamp stableTimestamp, bool forc
         stableTSConfigString =
             "force=true,oldest_timestamp={0:x},durable_timestamp={0:x},stable_timestamp={0:x}"_format(
                 ts);
-        stdx::lock_guard<Latch> lk(_highestDurableTimestampMutex);
-        _highestSeenDurableTimestamp = ts;
     } else {
         stableTSConfigString = "stable_timestamp={:x}"_format(ts);
     }
@@ -2255,8 +2230,7 @@ void WiredTigerKVEngine::setOldestTimestamp(Timestamp newOldestTimestamp, bool f
                 newOldestTimestamp.asULL());
         invariantWTOK(_conn->set_timestamp(_conn, oldestTSConfigString.c_str()), nullptr);
         _oldestTimestamp.store(newOldestTimestamp.asULL());
-        stdx::lock_guard<Latch> lk(_highestDurableTimestampMutex);
-        _highestSeenDurableTimestamp = newOldestTimestamp.asULL();
+
         LOGV2_DEBUG(22342,
                     2,
                     "oldest_timestamp and durable_timestamp force set to {newOldestTimestamp}",
@@ -2400,27 +2374,23 @@ StatusWith<Timestamp> WiredTigerKVEngine::recoverToStableTimestamp(OperationCont
                 str::stream() << "Error rolling back to stable. Err: " << wiredtiger_strerror(ret)};
     }
 
-    {
-        // Rollback the highest seen durable timestamp to the stable timestamp.
-        stdx::lock_guard<Latch> lk(_highestDurableTimestampMutex);
-        _highestSeenDurableTimestamp = stableTimestamp.asULL();
-    }
-
     _sizeStorer = std::make_unique<WiredTigerSizeStorer>(_conn, _sizeStorerUri);
 
     return {stableTimestamp};
 }
 
 Timestamp WiredTigerKVEngine::getAllDurableTimestamp() const {
-    auto ret = _fetchAllDurableValue(_conn);
+    // Fetch the latest all_durable value from the storage engine. This value will be a timestamp
+    // that has no holes (uncommitted transactions with lower timestamps) behind it.
+    char buf[(2 * 8 /* bytes in hex */) + 1 /* null terminator */];
+    invariantWTOK(_conn->query_timestamp(_conn, buf, "get=all_durable"), nullptr);
 
-    stdx::lock_guard<Latch> lk(_highestDurableTimestampMutex);
-    if (ret < _highestSeenDurableTimestamp) {
-        ret = _highestSeenDurableTimestamp;
-    } else {
-        _highestSeenDurableTimestamp = ret;
-    }
-    return Timestamp(ret);
+    uint64_t ts;
+    fassert(38002, NumberParser{}.base(16)(buf, &ts));
+
+    // If all_durable is 0, treat this as lowest possible timestamp; we need to see all pre-existing
+    // data but no new (timestamped) data.
+    return Timestamp{ts == 0 ? StorageEngine::kMinimumTimestamp : ts};
 }
 
 boost::optional<Timestamp> WiredTigerKVEngine::getRecoveryTimestamp() const {
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.h b/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.h
index 73b720774d5..d52e1d8d9e1 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.h
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.h
@@ -547,10 +547,6 @@ private:
 
     std::unique_ptr<WiredTigerEngineRuntimeConfigParameter> _runTimeConfigParam;
 
-    mutable Mutex _highestDurableTimestampMutex =
-        MONGO_MAKE_LATCH("WiredTigerKVEngine::_highestDurableTimestampMutex");
-    mutable unsigned long long _highestSeenDurableTimestamp = StorageEngine::kMinimumTimestamp;
-
     mutable Mutex _oldestTimestampPinRequestsMutex =
         MONGO_MAKE_LATCH("WiredTigerKVEngine::_oldestTimestampPinRequestsMutex");
     std::map<std::string, Timestamp> _oldestTimestampPinRequests;