diff --git a/docs/checkpoint_recovery.md b/docs/checkpoint_recovery.md index bed775f34d..49a7e8dd3a 100644 --- a/docs/checkpoint_recovery.md +++ b/docs/checkpoint_recovery.md @@ -30,7 +30,16 @@ directory sync. There the ordering still recovers from a process crash — file reached the OS — but power-loss durability of those directory entries is not guaranteed. If a step fails after the commit point, the database refuses further writes until it is reopened, -and recovery finishes the checkpoint. +and recovery finishes the checkpoint. A failure before it rolls the checkpoint back; if that rollback +fails too, the database refuses further writes the same way. + +A checkpoint that runs automatically after a commit never fails that commit, which is already in +the WAL: its failure is printed as a warning, and a later commit tries again, or fails if the +failure made the database refuse writes. The checkpoint that `COPY` or a large `CREATE INDEX` +requires is different, since their data is not in the WAL: its failure fails the statement. If the +checkpoint was rolled back, the changes are lost on restart unless a later `CHECKPOINT` succeeds; +if it made the database refuse writes, they survive the restart only if the checkpoint's record +reached storage. ## Checkpoint bundle format diff --git a/src/include/storage/checkpointer.h b/src/include/storage/checkpointer.h index fa829e9881..ffceee14fe 100644 --- a/src/include/storage/checkpointer.h +++ b/src/include/storage/checkpointer.h @@ -82,6 +82,8 @@ class Checkpointer { // the horizon pins the WAL. virtual void onWALRetired(bool walRotated); virtual void beforeGraphShadowApply(StorageManager& storageManager); + // Runs first in rollback(), so tests can make the rollback itself fail. + virtual void beforeRollback(); void applyShadowPagesForPartitionChildren(); void flushGraphShadowFiles(); void applyGraphShadowPages(); diff --git a/src/include/storage/wal/wal.h b/src/include/storage/wal/wal.h index c0aaebaa42..8ef1c64c1e 100644 --- a/src/include/storage/wal/wal.h +++ b/src/include/storage/wal/wal.h @@ -62,6 +62,7 @@ class WAL { // source for WAL pin observability alongside getFileSize() (active). uint64_t getFrozenWALFileSize(); bool mayHaveCheckpointRecord(); + bool isPoisoned(); void throwIfPoisoned(); void poison(const std::string& reason); diff --git a/src/storage/checkpointer.cpp b/src/storage/checkpointer.cpp index 7f6e9e1818..e99283c6a8 100644 --- a/src/storage/checkpointer.cpp +++ b/src/storage/checkpointer.cpp @@ -578,6 +578,8 @@ void Checkpointer::onWALRetired(bool) {} void Checkpointer::beforeGraphShadowApply(StorageManager&) {} +void Checkpointer::beforeRollback() {} + void Checkpointer::logCheckpointAndApplyShadowPages(bool walRotated_) { auto& shadowFile = mainStorageManager->getShadowFile(); shadowFile.flushAll(mainStorageManager->getOrInitDatabaseID(clientContext)); @@ -633,10 +635,11 @@ void Checkpointer::applyShadowPagesForPartitionChildren() { } } -void Checkpointer::rollback() { +void Checkpointer::rollback() try { if (isInMemory) { return; } + beforeRollback(); // Any pages freed during the checkpoint are no longer freed for (const auto& target : checkpointTargets) { target.storageManager->rollbackCheckpoint(*target.catalog, &clientContext); @@ -661,6 +664,20 @@ void Checkpointer::rollback() { } } releaseCheckpointLocks(); +} catch (const std::exception& e) { + // A half-undone checkpoint leaves page and WAL state that later writes must not build on. + releaseCheckpointLocks(); + mainStorageManager->getWAL().poison(std::format("Checkpoint rollback failed: {}", e.what())); + throw common::RuntimeException(std::format( + "Checkpoint rollback failed; the database refuses further writes until restart. Original " + "error: {}", + e.what())); +} catch (...) { + releaseCheckpointLocks(); + mainStorageManager->getWAL().poison("Checkpoint rollback failed: unknown exception"); + throw common::RuntimeException( + "Checkpoint rollback failed; the database refuses further writes until restart. Original " + "error: unknown exception"); } bool Checkpointer::canAutoCheckpoint(const main::ClientContext& clientContext, diff --git a/src/storage/wal/wal.cpp b/src/storage/wal/wal.cpp index a4328e5e2f..3a8512f23b 100644 --- a/src/storage/wal/wal.cpp +++ b/src/storage/wal/wal.cpp @@ -362,6 +362,11 @@ bool WAL::mayHaveCheckpointRecord() { return checkpointRecordMayExist; } +bool WAL::isPoisoned() { + std::unique_lock lck{mtx}; + return poisoned; +} + void WAL::throwIfPoisoned() { std::unique_lock lck{mtx}; throwIfPoisonedNoLock(); diff --git a/src/transaction/transaction_manager.cpp b/src/transaction/transaction_manager.cpp index e469a575f0..a1b12eec0f 100644 --- a/src/transaction/transaction_manager.cpp +++ b/src/transaction/transaction_manager.cpp @@ -1,6 +1,7 @@ #include "transaction/transaction_manager.h" #include +#include #include #include "common/exception/checkpoint.h" @@ -52,6 +53,16 @@ class QueryProgressScope { std::optional queryID; }; +void warnAutoCheckpointFailed(WAL& wal, const std::exception& e) { + std::string walSize = "unknown"; + try { + walSize = std::to_string(wal.getFileSize()); + } catch (...) { // NOLINT: The statement has committed, so reporting must not fail it. + } + std::cerr << "Warning: automatic checkpoint after a commit failed, WAL " << walSize + << " bytes: " << e.what() << std::endl; +} + } // namespace Transaction* TransactionManager::beginTransaction(main::ClientContext& clientContext, @@ -175,17 +186,35 @@ void TransactionManager::commit(main::ClientContext& clientContext, Transaction* // (commit/rollback) during the drain phase instead of deadlocking. The transaction has // already been removed from activeTransactions at this point, so any later failure is a // checkpoint failure, not a transaction failure that can be rolled back. - try { - if (shouldForceCheckpoint) { + if (shouldForceCheckpoint) { + try { checkpoint(clientContext); - } else if (shouldAutoCheckpoint) { + } catch (const std::exception& e) { + // COPY FROM and CREATE INDEX on a large ART rely on this checkpoint instead of the WAL. + if (wal.isPoisoned()) { + // No later CHECKPOINT can run before the restart, whose recovery applies this one + // only if its record reached storage. + throw CheckpointException{std::format( + "Transaction committed successfully, but the checkpoint it requires failed. " + "The database refuses further writes until restart, and the changes survive " + "it only if the checkpoint's record reached storage: {}", + e.what())}; + } + throw CheckpointException{std::format( + "Transaction committed successfully, but the checkpoint it requires failed. Its " + "changes are not in the WAL and may be lost on restart unless a later CHECKPOINT " + "succeeds: {}", + e.what())}; + } + } else if (shouldAutoCheckpoint) { + try { tryCheckpoint(clientContext); + } catch (const std::exception& e) { + // The commit is in the WAL whatever the checkpoint did, so failing the statement would + // only invite a retry that applies it twice. A rolled-back checkpoint is retried by a + // later write commit; a poisoned WAL fails the next write instead. + warnAutoCheckpointFailed(wal, e); } - } catch (const std::exception& e) { - throw CheckpointException{std::format( - "Transaction committed successfully, but the post-commit checkpoint failed. " - "The committed data is durable and will be recovered on restart: {}", - e.what())}; } } diff --git a/test/transaction/checkpoint_test.cpp b/test/transaction/checkpoint_test.cpp index 269af81455..c6ea3aa87c 100644 --- a/test/transaction/checkpoint_test.cpp +++ b/test/transaction/checkpoint_test.cpp @@ -366,6 +366,42 @@ TEST_F(CheckpointRetryAfterFailureTest, RetryFailsWithNewWritesThenReopen) { checkNodes(200); } +class FlakyCheckpointerFailsOnRollback final : public Checkpointer { +public: + explicit FlakyCheckpointerFailsOnRollback(main::ClientContext& context) + : Checkpointer(context) {} + +protected: + bool checkpointStorage() override { throw RuntimeException("checkpoint failed."); } + void beforeRollback() override { throw RuntimeException("rollback failed."); } +}; + +TEST_F(CheckpointRetryAfterFailureTest, FailedRollbackPoisonsWAL) { + if (inMemMode || systemConfig->checkpointThreshold == 0) { + GTEST_SKIP(); + } + insertNodes(0, 100); + FlakyCheckpointer flakyCheckpointer([](main::ClientContext& context) { + return std::make_unique(context); + }); + flakyCheckpointer.setCheckpointer(*getClientContext(*conn)); + auto res = conn->query("CHECKPOINT;"); + ASSERT_FALSE(res->isSuccess()); + EXPECT_NE(res->getErrorMessage().find("refuses further writes until restart"), + std::string::npos) + << res->getErrorMessage(); + checkNodes(100); + res = conn->query("CREATE (a:test {id: 100, name: 'name_100'});"); + ASSERT_FALSE(res->isSuccess()); + EXPECT_NE(res->getErrorMessage().find("panic state"), std::string::npos) + << res->getErrorMessage(); + EXPECT_NE(res->getErrorMessage().find("Checkpoint rollback failed"), std::string::npos) + << res->getErrorMessage(); + res.reset(); + createDBAndConn(); + checkNodes(100); +} + class FlakyCheckpointerFailsOnSerialization final : public Checkpointer { public: explicit FlakyCheckpointerFailsOnSerialization(main::ClientContext& context) @@ -814,6 +850,189 @@ TEST_F(CheckpointRetryAfterFailureTest, PartitionShadowsSurviveUntilMainWALRetir ASSERT_EQ(result->getNext()->getValue(0)->getValue(), 1); } +// Checkpoints automatically after every commit. +class PostCommitCheckpointFailureTest : public CheckpointRetryAfterFailureTest { +public: + void SetUp() override { + CheckpointRetryAfterFailureTest::SetUp(); + checkpointAfterEveryCommit(); + } + + void checkpointAfterEveryCommit() const { + ASSERT_TRUE(conn->query("CALL force_checkpoint_on_close=false;")->isSuccess()); + ASSERT_TRUE(conn->query("CALL auto_checkpoint=true;")->isSuccess()); + ASSERT_TRUE(conn->query("CALL checkpoint_threshold=0;")->isSuccess()); + } + + // Counts checkpoint attempts; each one fails while failCheckpoints is set. + template + void countCheckpoints() { + FlakyCheckpointer counting( + [this](main::ClientContext& context) -> std::unique_ptr { + numCheckpoints++; + if (failCheckpoints) { + return std::make_unique(context); + } + return std::make_unique(context); + }); + counting.setCheckpointer(*getClientContext(*conn)); + } + + uint64_t numCheckpoints = 0; + bool failCheckpoints = true; +}; + +TEST_F(PostCommitCheckpointFailureTest, FailedAutoCheckpointDoesNotFailCommit) { + if (inMemMode) { + GTEST_SKIP(); + } + countCheckpoints(); + ::testing::internal::CaptureStderr(); + insertNodes(0, 1); + ASSERT_TRUE(conn->query("BEGIN TRANSACTION;")->isSuccess()); + insertNodes(1, 2); + auto res = conn->query("COMMIT;"); + const auto warnings = ::testing::internal::GetCapturedStderr(); + ASSERT_TRUE(res->isSuccess()) << res->getErrorMessage(); + ASSERT_EQ(numCheckpoints, 2u); + EXPECT_NE(warnings.find("Warning: automatic checkpoint after a commit failed, WAL "), + std::string::npos) + << warnings; + EXPECT_NE(warnings.find("checkpoint failed."), std::string::npos) << warnings; + // Reopen without checkpointing: the WAL alone recovers both commits. + res.reset(); + createDBAndConn(); + checkNodes(2); +} + +TEST_F(PostCommitCheckpointFailureTest, NextCommitRetriesFailedAutoCheckpoint) { + if (inMemMode) { + GTEST_SKIP(); + } + countCheckpoints(); + insertNodes(0, 1); + EXPECT_GT(WAL::Get(*getClientContext(*conn))->getFileSize(), 0u); + failCheckpoints = false; + insertNodes(1, 2); + ASSERT_EQ(numCheckpoints, 2u); + EXPECT_EQ(WAL::Get(*getClientContext(*conn))->getFileSize(), 0u); + createDBAndConn(); + checkNodes(2); +} + +TEST_F(PostCommitCheckpointFailureTest, AutoCheckpointBlockedByOpenWriterDoesNotFailCommit) { + if (inMemMode) { + GTEST_SKIP(); + } + conn.reset(); + createDB(10000 /* checkpoint wait timeout in microseconds */); + conn = std::make_unique(database.get()); + checkpointAfterEveryCommit(); + ASSERT_TRUE(conn->query("CALL debug_enable_multi_writes=true;")->isSuccess()); + auto context = getClientContext(*conn); + main::Connection other{database.get()}; + ASSERT_TRUE(other.query("BEGIN TRANSACTION;")->isSuccess()); + ASSERT_TRUE(other.query("CREATE (a:test {id: 1, name: 'name_1'});")->isSuccess()); + // This commit's checkpoint times out waiting for the other connection's write transaction. + insertNodes(0, 1); + EXPECT_GT(WAL::Get(*context)->getFileSize(), 0u); + auto res = other.query("COMMIT;"); + ASSERT_TRUE(res->isSuccess()) << res->getErrorMessage(); + EXPECT_EQ(WAL::Get(*context)->getFileSize(), 0u); + checkNodes(2); +} + +TEST_F(PostCommitCheckpointFailureTest, AutoCheckpointThatPoisonsWALFailsNextWrite) { + if (inMemMode) { + GTEST_SKIP(); + } + countCheckpoints(); + insertNodes(0, 1); + ASSERT_EQ(numCheckpoints, 1u); + auto res = conn->query("CREATE (a:test {id: 1, name: 'name_1'});"); + ASSERT_FALSE(res->isSuccess()); + EXPECT_NE(res->getErrorMessage().find("refuses further writes until restart"), + std::string::npos) + << res->getErrorMessage(); + res.reset(); + createDBAndConn(); + checkNodes(1); +} + +TEST_F(PostCommitCheckpointFailureTest, FailedForcedCheckpointFailsCopy) { + if (inMemMode) { + GTEST_SKIP(); + } + countCheckpoints(); + auto res = conn->query("COPY test FROM (UNWIND range(0, 9) AS i RETURN i, 'n');"); + ASSERT_FALSE(res->isSuccess()); + EXPECT_NE( + res->getErrorMessage().find("may be lost on restart unless a later CHECKPOINT succeeds"), + std::string::npos) + << res->getErrorMessage(); + ASSERT_EQ(numCheckpoints, 1u); + failCheckpoints = false; + checkpoint(); + createDBAndConn(); + checkNodes(10); +} + +TEST_F(PostCommitCheckpointFailureTest, ForcedCheckpointFailingAfterItsRecordRefusesWrites) { + if (inMemMode) { + GTEST_SKIP(); + } + countCheckpoints(); + auto res = conn->query("COPY test FROM (UNWIND range(0, 9) AS i RETURN i, 'n');"); + ASSERT_FALSE(res->isSuccess()); + const auto error = res->getErrorMessage(); + // A later CHECKPOINT cannot run: the failure after the record poisoned the WAL. + EXPECT_NE(error.find("refuses further writes until restart"), std::string::npos) << error; + EXPECT_EQ(error.find("unless a later CHECKPOINT succeeds"), std::string::npos) << error; + res.reset(); + createDBAndConn(); + checkNodes(10); +} + +TEST_F(PostCommitCheckpointFailureTest, FailedForcedCheckpointFailsCreateIndex) { + if (inMemMode) { + GTEST_SKIP(); + } + insertNodes(0, 2); + countCheckpoints(); + setenv("LBUG_CREATE_INDEX_WAL_THRESHOLD", "1", 1); + auto res = conn->query("CREATE ART INDEX test_name_idx FOR (n:test) ON (n.name);"); + unsetenv("LBUG_CREATE_INDEX_WAL_THRESHOLD"); + ASSERT_FALSE(res->isSuccess()); + EXPECT_NE( + res->getErrorMessage().find("may be lost on restart unless a later CHECKPOINT succeeds"), + std::string::npos) + << res->getErrorMessage(); + ASSERT_EQ(numCheckpoints, 1u); + failCheckpoints = false; + checkpoint(); + createDBAndConn(); + res = conn->query("CALL SHOW_INDEXES() WHERE index_name = 'test_name_idx' RETURN count(*);"); + ASSERT_TRUE(res->isSuccess()) << res->getErrorMessage(); + EXPECT_EQ(res->getNext()->getValue(0)->getValue(), 1); +} + +TEST_F(PostCommitCheckpointFailureTest, FailedCheckpointCleanupFailsNextWrite) { + if (inMemMode) { + GTEST_SKIP(); + } + countCheckpoints(); + insertNodes(0, 1); + ASSERT_EQ(numCheckpoints, 1u); + auto res = conn->query("CREATE (a:test {id: 1, name: 'name_1'});"); + ASSERT_FALSE(res->isSuccess()); + EXPECT_NE(res->getErrorMessage().find("checkpoint interrupted before WAL retirement"), + std::string::npos) + << res->getErrorMessage(); + res.reset(); + createDBAndConn(); + checkNodes(1); +} + class CommittedPartitionMissingBaseTest : public CheckpointRetryAfterFailureTest, public ::testing::WithParamInterface {};