From 76bb827507fff6b4212917684806bf4ae9f11082 Mon Sep 17 00:00:00 2001 From: Darek Date: Thu, 5 Nov 2026 11:15:14 +0100 Subject: [PATCH 1/2] fix(transaction): don't block at the write gate holding the public mutex A checkpoint closes the write gate, then waits up to the checkpoint wait timeout for open write transactions to leave. A write BEGIN arriving meanwhile took the public mutex and then blocked on the gate while holding it. Every BEGIN, COMMIT and ROLLBACK needs that mutex, so the open writers could not leave, reads could not start, and the checkpoint timed out. Its next attempt, on the next commit past the threshold, met the same state, so under steady load with three or more writers the database stalled for the timeout over and over and no checkpoint completed. A write BEGIN now takes the gate first and the public mutex second. A BEGIN waiting for a checkpoint holds nothing the open writers or readers need. The wait for a pending forced checkpoint at the top of beginTransaction existed for this same reason and is now part of the mustWait loop. A BEGIN waiting at the gate also made the checkpoint's try_lock on the public mutex fail, which pinned the WAL with nothing open. Co-Authored-By: Claude Opus 5.5 --- src/include/transaction/transaction_manager.h | 3 + src/transaction/transaction_manager.cpp | 10 +-- test/transaction/checkpoint_test.cpp | 88 +++++++++++++++++++ 3 files changed, 95 insertions(+), 6 deletions(-) diff --git a/src/include/transaction/transaction_manager.h b/src/include/transaction/transaction_manager.h index 3bb397572..bad0adc87 100644 --- a/src/include/transaction/transaction_manager.h +++ b/src/include/transaction/transaction_manager.h @@ -164,6 +164,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. diff --git a/src/transaction/transaction_manager.cpp b/src/transaction/transaction_manager.cpp index 923089397..088d73c4c 100644 --- a/src/transaction/transaction_manager.cpp +++ b/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: { diff --git a/test/transaction/checkpoint_test.cpp b/test/transaction/checkpoint_test.cpp index b1f4294dc..59024577a 100644 --- a/test/transaction/checkpoint_test.cpp +++ b/test/transaction/checkpoint_test.cpp @@ -94,6 +94,26 @@ 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; + } + private: TransactionManager::init_checkpointer_func_t initFunc; }; @@ -961,6 +981,74 @@ 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(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 waitingRes; + std::promise 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 { + std::promise 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(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, CopyBesideOpenWriterCommitsNothing) { if (inMemMode) { GTEST_SKIP(); From eddff460554e8f936b06bd54eadd990f7a16f2a3 Mon Sep 17 00:00:00 2001 From: Darek Date: Thu, 5 Nov 2026 11:48:37 +0100 Subject: [PATCH 2/2] fix(transaction): always sample open transactions under the public mutex The checkpoint sampled the open transactions after closing the write gate with a try_lock, because a write BEGIN could hold the public mutex while blocked on the gate. When the try_lock failed it fell back to a horizon captured before the gate, or to its own snapshot timestamp if nothing had been open. Either way the checkpoint counted as pinned and kept the WAL, even with no transaction open, so the WAL grew past the threshold and every later commit ran another checkpoint. No holder of the public mutex blocks on the gate any more, so the checkpoint now takes the mutex and always samples what is really open. The capture before the gate and both fallbacks are gone. Co-Authored-By: Claude Opus 5.5 --- src/include/storage/checkpointer.h | 2 +- src/include/storage/table/node_group.h | 2 +- src/include/transaction/transaction_manager.h | 16 ++--- src/transaction/transaction_manager.cpp | 72 +++++-------------- test/transaction/checkpoint_test.cpp | 53 ++++++++++---- 5 files changed, 65 insertions(+), 80 deletions(-) diff --git a/src/include/storage/checkpointer.h b/src/include/storage/checkpointer.h index 487c78ac0..f3a7f4cf8 100644 --- a/src/include/storage/checkpointer.h +++ b/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); diff --git a/src/include/storage/table/node_group.h b/src/include/storage/table/node_group.h index 81bdc1e26..05184162b 100644 --- a/src/include/storage/table/node_group.h +++ b/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 openStartTSs; diff --git a/src/include/transaction/transaction_manager.h b/src/include/transaction/transaction_manager.h index bad0adc87..568030b80 100644 --- a/src/include/transaction/transaction_manager.h +++ b/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 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 lastTimestamp{1}; uint64_t nextWALCommitSequenceToPublish = 1; std::condition_variable cvForPublishingCommit; diff --git a/src/transaction/transaction_manager.cpp b/src/transaction/transaction_manager.cpp index 088d73c4c..8963f1f61 100644 --- a/src/transaction/transaction_manager.cpp +++ b/src/transaction/transaction_manager.cpp @@ -471,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 @@ -497,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 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 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, diff --git a/test/transaction/checkpoint_test.cpp b/test/transaction/checkpoint_test.cpp index 59024577a..0a7ae2271 100644 --- a/test/transaction/checkpoint_test.cpp +++ b/test/transaction/checkpoint_test.cpp @@ -114,6 +114,21 @@ class FlakyCheckpointer { 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 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; }; @@ -1049,6 +1064,26 @@ TEST_F(PostCommitCheckpointFailureTest, WriterWaitingAtWriteGateDoesNotPinCheckp 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 { + holder = FlakyCheckpointer::holdPublicMutexBriefly(ctx); + return std::make_unique(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(); @@ -5388,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(); @@ -5445,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(); - 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) {