fix(transaction): in BEGIN, take the write gate before the public mutex

src/include/storage/checkpointer.h

@@ -118,7 +118,7 @@ class Checkpointer {
     bool walRotated = false;
     // Snapshot timestamp captured at drain time for MVCC catalog serialization.
     common::transaction_t snapshotTS = 0;
-    // Snapshot-horizon floor captured alongside snapshotTS (before the write gate).
+    // Snapshot-horizon floor captured alongside snapshotTS.
     // Version data committed at or after it is retained for open readers, and WAL
     // rotation/retirement is skipped while it pins (the WAL keeps the commit records the
     // pinned snapshot may still need);

src/include/storage/table/node_group.h

@@ -67,7 +67,7 @@ struct NodeGroupCheckpointState {
     // checkpoint wipes version metadata as before). Ignored by CSR groups in this phase.
 
     common::transaction_t retentionHorizon = common::INVALID_TRANSACTION;
-    // Open-reader start timestamps captured alongside the horizon (before the write gate).
+    // Open-reader start timestamps captured alongside the horizon.
     // CSR groups use the set to cover and prune retained pre-checkpoint snapshots;
     // regular groups ignore it.
     std::vector<common::transaction_t> openStartTSs;

src/include/transaction/transaction_manager.h

@@ -56,17 +56,13 @@ class TransactionManager {
 
     // Minimum start timestamp over all currently open transactions (the snapshot horizon
     // that checkpoint retention must preserve). Returns INVALID_TRANSACTION when no
-    // transaction is open. Brief lock acquisition; callers holding the checkpoint write
-    // gate must use a value captured before the gate was acquired.
+    // transaction is open. Brief lock acquisition.
     common::transaction_t snapshotHorizon();
-    // Start timestamps of all currently open transactions, for CSR snapshot coverage and
-    // pruning. Brief lock acquisition; capture before the checkpoint write gate alongside
-    // the horizon (a small skew between the two snapshots is safe in both directions).
+    // Start timestamps of all currently open transactions. Brief lock acquisition.
     std::vector<common::transaction_t> snapshotOpenStartTSs();
     // Pin behind the snapshot horizon: the minimum open start timestamp, the transaction
     // holding it, and the open-transaction count (which transaction pins WAL growth).
-    // INVALID_TRANSACTION/0 when no transaction is open. Brief lock acquisition; capture
-    // before the checkpoint write gate alongside the horizon.
+    // INVALID_TRANSACTION/0 when no transaction is open. Brief lock acquisition.
     struct HorizonPinInfo {
         common::transaction_t horizon = common::INVALID_TRANSACTION;
         common::transaction_t pinningTxnID = common::INVALID_TRANSACTION;
@@ -150,10 +146,8 @@ class TransactionManager {
     // Open-transaction counts are small, so a full rescan on erase is trivial.
     common::transaction_t cachedHorizon = common::INVALID_TRANSACTION;
     common::transaction_t lastTransactionID;
-    // Atomic so checkpointNoLock() can snapshot it without taking
-    // mtxForSerializingPublicFunctionCalls (which would invert the lock order against
-    // beginTransaction()'s public -> start acquisition order and deadlock concurrent
-    // writers during an auto-checkpoint triggered from commit()).
+    // Atomic so checkpointNoLock() can read and bump it before taking
+    // mtxForSerializingPublicFunctionCalls.
     std::atomic<common::transaction_t> lastTimestamp{1};
     uint64_t nextWALCommitSequenceToPublish = 1;
     std::condition_variable cvForPublishingCommit;
@@ -164,6 +158,9 @@ class TransactionManager {
     bool forcedCheckpointPending = false;
     // This mutex serializes begin/commit/rollback calls to protect activeTransactions.
     std::mutex mtxForSerializingPublicFunctionCalls;
+    // Held by a checkpoint for as long as it waits for write transactions to leave and then runs.
+    // Those transactions need mtxForSerializingPublicFunctionCalls to leave, so nothing may block
+    // on this mutex while holding that one: take this one first.
     std::mutex mtxForStartingNewTransactions;
     // Prevents concurrent checkpoints. Separate from mtxForSerializingPublicFunctionCalls so
     // that active writers can commit/rollback while the checkpoint is draining them.

src/transaction/transaction_manager.cpp

@@ -68,18 +68,13 @@ void warnAutoCheckpointFailed(WAL& wal, const std::exception& e) {
 
 Transaction* TransactionManager::beginTransaction(main::ClientContext& clientContext,
     TransactionType type) {
-    std::unique_lock publicFunctionLck{mtxForSerializingPublicFunctionCalls};
     // Only acquire the write gate for write/recovery transactions. Read-only transactions
     // can start freely during checkpoint since they use snapshot isolation.
     std::unique_lock newTransactionLck{mtxForStartingNewTransactions, std::defer_lock};
     if (type != TransactionType::READ_ONLY) {
-        // Wait out a pending forced checkpoint before blocking on the start mutex: blocking there
-        // holds the public mutex, which the writers that the forced commit waits for need in
-        // order to leave.
-        cvForCommittingWriteTransaction.wait(publicFunctionLck,
-            [&]() { return !forcedCheckpointPending; });
         newTransactionLck.lock();
     }
+    std::unique_lock publicFunctionLck{mtxForSerializingPublicFunctionCalls};
     const auto mustWait = [&]() {
         if (type == TransactionType::READ_ONLY) {
             return false;
@@ -95,7 +90,10 @@ Transaction* TransactionManager::beginTransaction(main::ClientContext& clientCon
     while (mustWait()) {
         newTransactionLck.unlock();
         cvForCommittingWriteTransaction.wait(publicFunctionLck, [&]() { return !mustWait(); });
+        // Nothing may block on the start mutex while holding the public one.
+        publicFunctionLck.unlock();
         newTransactionLck.lock();
+        publicFunctionLck.lock();
     }
     switch (type) {
     case TransactionType::READ_ONLY: {
@@ -473,14 +471,6 @@ void TransactionManager::tryCheckpoint(main::ClientContext& clientContext) {
 
 void TransactionManager::checkpointNoLock(main::ClientContext& clientContext) {
     QueryProgressScope progress{clientContext, 0.01};
-    // Capture the retention horizon before acquiring the write gate. Taking the
-    // public-function lock here is safe because the gate is not held yet; the value is
-    // threaded through the checkpointer so no later stage re-acquires that lock under
-    // the gate.
-    const auto preHorizonPin = snapshotHorizonInfo();
-    auto retentionHorizon = preHorizonPin.horizon;
-    auto openStartTSs = snapshotOpenStartTSs();
-    auto pinningTxnID = preHorizonPin.pinningTxnID;
     // We only need to wait for active write transactions to leave the system before
     // checkpointing. Read transactions can continue because they use MVCC snapshot isolation.
     // Column chunk metadata of in-place checkpointed data is published before the shadow pages
@@ -499,56 +489,28 @@ void TransactionManager::checkpointNoLock(main::ClientContext& clientContext) {
     }
     auto checkpointer = initCheckpointerFunc(clientContext);
     try {
-        // lastTimestamp is atomic, so we can snapshot it without taking
-        // mtxForSerializingPublicFunctionCalls. Grabbing that mutex here would invert the
-        // lock order against beginTransaction() (which takes public -> start) while the
-        // checkpoint holds the write gate (start), deadlocking concurrent writers during an
-        // auto-checkpoint triggered from commit().
-        //
-        // A reader that begins between the pre-gate capture above and the write gate (or
-        // during the storage phase) would be missing from the pre-gate horizon and lose its
-        // snapshot. Isolate newcomers by bumping lastTimestamp: the checkpoint snapshot is
-        // the pre-bump value, and every transaction that begins afterwards observes the
-        // collapsed post-checkpoint base with a larger start timestamp, so it needs no
-        // retention. Readers that began before the bump keep startTS <= snapshotTimestamp
-        // and must be covered by retention below.
+        // A reader that begins during the storage phase must not lose its snapshot. Isolate
+        // newcomers by bumping lastTimestamp: the checkpoint snapshot is the pre-bump value, and a
+        // transaction that reads the bumped value observes the collapsed post-checkpoint base, so
+        // it needs no retention. One that read it before the bump keeps startTS <=
+        // snapshotTimestamp and must be covered by retention.
         transaction_t snapshotTimestamp = lastTimestamp.load(std::memory_order_acquire);
         lastTimestamp.fetch_add(1, std::memory_order_acq_rel);
-        // Re-sample the registry to include gap readers without deadlocking: a waiting
-        // writer holds the public lock while blocked on the gate, so only try_lock here.
-        // Keep only start timestamps at or below the checkpoint snapshot; anything larger
-        // began after the bump and observes the post-checkpoint base.
+        transaction_t retentionHorizon = INVALID_TRANSACTION;
+        transaction_t pinningTxnID = INVALID_TRANSACTION;
+        std::vector<transaction_t> openStartTSs;
         {
-            std::unique_lock tryLck{mtxForSerializingPublicFunctionCalls, std::try_to_lock};
-            if (tryLck.owns_lock()) {
-                transaction_t postHorizon = INVALID_TRANSACTION;
-                transaction_t postPin = INVALID_TRANSACTION;
-                std::vector<transaction_t> postOpen;
-                postOpen.reserve(openTxnStartTS.size());
-                for (const auto& entry : openTxnStartTS) {
-                    if (entry.second > snapshotTimestamp) {
-                        continue;
-                    }
-                    postOpen.push_back(entry.second);
-                    if (postHorizon == INVALID_TRANSACTION || entry.second < postHorizon) {
-                        postHorizon = entry.second;
-                        postPin = entry.first;
-                    }
+            std::unique_lock lck{mtxForSerializingPublicFunctionCalls};
+            openStartTSs.reserve(openTxnStartTS.size());
+            for (const auto& entry : openTxnStartTS) {
+                if (entry.second > snapshotTimestamp) {
+                    continue;
+                }
+                openStartTSs.push_back(entry.second);
+                if (retentionHorizon == INVALID_TRANSACTION || entry.second < retentionHorizon) {
+                    retentionHorizon = entry.second;
+                    pinningTxnID = entry.first;
                 }
-                retentionHorizon = postHorizon;
-                openStartTSs = std::move(postOpen);
-                pinningTxnID = postPin;
-            } else if (retentionHorizon == INVALID_TRANSACTION) {
-                // A writer is waiting on the gate and holds the public lock. Fall back to a
-                // conservative horizon: any gap reader holds startTS == snapshotTimestamp.
-                // This over-retains only when a writer races the checkpoint with no actual
-                // gap reader, which is rare and still correct.
-                retentionHorizon = snapshotTimestamp;
-                openStartTSs.push_back(snapshotTimestamp);
-            } else if (std::ranges::find(openStartTSs, snapshotTimestamp) == openStartTSs.end()) {
-                // Pre-gate horizon already covers gap readers (their startTS >= horizon),
-                // but CSR coverage needs their timestamp present to take a snapshot.
-                openStartTSs.push_back(snapshotTimestamp);
             }
         }
         checkpointer->beginCheckpoint(snapshotTimestamp, retentionHorizon, openStartTSs,

test/transaction/checkpoint_test.cpp

@@ -94,6 +94,41 @@ class FlakyCheckpointer {
         return false;
     }
 
+    // Returns whether a checkpoint closed the write gate within 10 s. A BEGIN holds the gate's
+    // mutex only briefly, so it counts as closed once it stays held across a 20 ms gap.
+    static bool waitForWriteGateClosed(main::ClientContext& context) {
+        auto* transactionManager = TransactionManager::Get(context);
+        const auto isHeld = [transactionManager]() {
+            std::unique_lock lck{transactionManager->mtxForStartingNewTransactions,
+                std::try_to_lock};
+            return !lck.owns_lock();
+        };
+        const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(10);
+        while (std::chrono::steady_clock::now() < deadline) {
+            const bool held = isHeld();
+            std::this_thread::sleep_for(std::chrono::milliseconds(20));
+            if (held && isHeld()) {
+                return true;
+            }
+        }
+        return false;
+    }
+
+    // Holds the mutex that every begin, commit and rollback takes for 100 ms on a new thread, and
+    // returns once it is held.
+    static std::thread holdPublicMutexBriefly(main::ClientContext& context) {
+        auto* transactionManager = TransactionManager::Get(context);
+        std::promise<void> held;
+        auto heldFuture = held.get_future();
+        std::thread holder([transactionManager, held = std::move(held)]() mutable {
+            std::unique_lock lck{transactionManager->mtxForSerializingPublicFunctionCalls};
+            held.set_value();
+            std::this_thread::sleep_for(std::chrono::milliseconds(100));
+        });
+        heldFuture.wait();
+        return holder;
+    }
+
 private:
     TransactionManager::init_checkpointer_func_t initFunc;
 };
@@ -961,6 +996,94 @@ TEST_F(PostCommitCheckpointFailureTest, AutoCheckpointBlockedByOpenWriterDoesNot
 
 // A COPY is durable only once its checkpoint runs, so it must not commit while another write
 // transaction could keep that checkpoint from running.
+TEST_F(PostCommitCheckpointFailureTest, WriterWaitingBehindAutoCheckpointLetsOthersThrough) {
+    if (inMemMode) {
+        GTEST_SKIP();
+    }
+    conn.reset();
+    createDB(10000000 /* checkpoint wait timeout in microseconds */);
+    conn = std::make_unique<main::Connection>(database.get());
+    checkpointAfterEveryCommit();
+    ASSERT_TRUE(conn->query("CALL enable_multi_writes=true;")->isSuccess());
+    main::Connection openWriter{database.get()};
+    main::Connection waitingWriter{database.get()};
+    main::Connection reader{database.get()};
+    ASSERT_TRUE(openWriter.query("BEGIN TRANSACTION;")->isSuccess());
+    ASSERT_TRUE(openWriter.query("CREATE (a:test {id: 1, name: 'name_1'});")->isSuccess());
+    // This commit's checkpoint closes the write gate and waits for openWriter to leave.
+    std::thread committing([this]() { insertNodes(0, 1); });
+    const bool gateClosed = FlakyCheckpointer::waitForWriteGateClosed(*getClientContext(*conn));
+    std::unique_ptr<main::QueryResult> waitingRes;
+    std::promise<void> waitingStarted;
+    std::thread waiting([&waitingWriter, &waitingRes, &waitingStarted]() {
+        waitingStarted.set_value();
+        waitingRes = waitingWriter.query("CREATE (a:test {id: 2, name: 'name_2'});");
+    });
+    waitingStarted.get_future().wait();
+    std::this_thread::sleep_for(std::chrono::milliseconds(100));
+    const auto start = std::chrono::steady_clock::now();
+    auto readRes = reader.query("MATCH (a:test) RETURN COUNT(a);");
+    auto commitRes = openWriter.query("COMMIT;");
+    const auto waited = std::chrono::steady_clock::now() - start;
+    committing.join();
+    waiting.join();
+    ASSERT_TRUE(gateClosed);
+    ASSERT_TRUE(readRes->isSuccess()) << readRes->getErrorMessage();
+    ASSERT_TRUE(commitRes->isSuccess()) << commitRes->getErrorMessage();
+    ASSERT_TRUE(waitingRes->isSuccess()) << waitingRes->getErrorMessage();
+    EXPECT_LT(waited, std::chrono::seconds(5));
+    checkNodes(3);
+}
+
+TEST_F(PostCommitCheckpointFailureTest, WriterWaitingAtWriteGateDoesNotPinCheckpoint) {
+    if (inMemMode) {
+        GTEST_SKIP();
+    }
+    auto context = getClientContext(*conn);
+    main::Connection waitingWriter{database.get()};
+    std::thread waiting;
+    // Checkpointers are created with the write gate closed.
+    FlakyCheckpointer writerArrives(
+        [&waiting, &waitingWriter](main::ClientContext& ctx) -> std::unique_ptr<Checkpointer> {
+            std::promise<void> started;
+            auto startedFuture = started.get_future();
+            waiting = std::thread([&waitingWriter, started = std::move(started)]() mutable {
+                started.set_value();
+                waitingWriter.query("BEGIN TRANSACTION;");
+            });
+            startedFuture.wait();
+            std::this_thread::sleep_for(std::chrono::milliseconds(100));
+            return std::make_unique<Checkpointer>(ctx);
+        });
+    writerArrives.setCheckpointer(*context);
+    insertNodes(0, 1);
+    FlakyCheckpointer::resetCheckpointer(*context);
+    waiting.join();
+    EXPECT_EQ(WAL::Get(*context)->getFileSize(), 0u);
+    ASSERT_TRUE(waitingWriter.query("COMMIT;")->isSuccess());
+    checkNodes(1);
+}
+
+TEST_F(PostCommitCheckpointFailureTest, BusyPublicMutexDoesNotPinCheckpoint) {
+    if (inMemMode) {
+        GTEST_SKIP();
+    }
+    auto context = getClientContext(*conn);
+    std::thread holder;
+    // Checkpointers are created with the write gate closed.
+    FlakyCheckpointer publicMutexBusy(
+        [&holder](main::ClientContext& ctx) -> std::unique_ptr<Checkpointer> {
+            holder = FlakyCheckpointer::holdPublicMutexBriefly(ctx);
+            return std::make_unique<Checkpointer>(ctx);
+        });
+    publicMutexBusy.setCheckpointer(*context);
+    insertNodes(0, 1);
+    FlakyCheckpointer::resetCheckpointer(*context);
+    holder.join();
+    EXPECT_EQ(WAL::Get(*context)->getFileSize(), 0u);
+    checkNodes(1);
+}
+
 TEST_F(PostCommitCheckpointFailureTest, CopyBesideOpenWriterCommitsNothing) {
     if (inMemMode) {
         GTEST_SKIP();
@@ -5300,16 +5423,8 @@ class ReviewFixesTest : public PrivateGraphTest {
 
 // Fix #1 – lastTimestamp data race
 // ─────────────────────────────────────────────────────────────────────────────
-// Before the fix, checkpointNoLock() read `lastTimestamp` without holding
-// mtxForSerializingPublicFunctionCalls, which is UB when commit() concurrently
-// increments it.  The fix snapshots the value under the mutex.
-//
-// Observable invariant tested here: a write transaction committed while the
-// checkpoint drain is waiting for it must be included in the checkpoint.
-// Without the fix the snapshot could see a stale (too-low) lastTimestamp,
-// causing the MVCC catalog snapshot to exclude the final committed entry.
-// After the fix the snapshot is taken under the mutex *after* the drain, so
-// it is guaranteed to reflect all commits that happened before the gate.
+// A write transaction committed while the checkpoint drain is waiting for it must be included in
+// the checkpoint: the snapshot timestamp is read only after the drain.
 TEST_F(ReviewFixesTest, CheckpointDrainWaitsForInFlightWrite) {
     if (inMemMode || systemConfig->checkpointThreshold == 0) {
         GTEST_SKIP();
@@ -5357,13 +5472,13 @@ TEST_F(ReviewFixesTest, CheckpointDrainWaitsForInFlightWrite) {
     conn2.reset();
 
     // On reload only checkpointed data is present (WAL has been rotated/cleared).
-    // All N+1 rows must be visible because the final commit occurred before the
-    // write gate was acquired and snapshotTS must capture it.
+    // All N+1 rows must be visible because the final commit landed during the drain, and the
+    // snapshot timestamp is read after the drain sees it.
     createDBAndConn();
     auto res = conn->query("MATCH (n:drain_test) RETURN count(n) AS c;");
     ASSERT_TRUE(res->isSuccess()) << res->getErrorMessage();
     const auto count = res->getNext()->getValue(0)->getValue<int64_t>();
-    ASSERT_EQ(count, N + 1) << "Row committed just before the write gate must survive checkpoint";
+    ASSERT_EQ(count, N + 1) << "Row committed during the checkpoint drain must survive checkpoint";
 }
 
 TEST_F(ReviewFixesTest, CheckpointUsesMatchingCatalogForMainAndDefaultGraph) {