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 3bb397572..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; @@ -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. diff --git a/src/transaction/transaction_manager.cpp b/src/transaction/transaction_manager.cpp index 923089397..8963f1f61 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: { @@ -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 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 b1f4294dc..0a7ae2271 100644 --- a/test/transaction/checkpoint_test.cpp +++ b/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 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(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, 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(); @@ -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(); - 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) {