fix(txn): don't fail a committed write when its automatic checkpoint fails

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
 

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();

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);
 

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,

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();

src/transaction/transaction_manager.cpp

@@ -1,6 +1,7 @@
 #include "transaction/transaction_manager.h"
 
 #include <algorithm>
+#include <iostream>
 #include <thread>
 
 #include "common/exception/checkpoint.h"
@@ -52,6 +53,16 @@ class QueryProgressScope {
     std::optional<uint64_t> 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())};
     }
 }
 

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<FlakyCheckpointerFailsOnRollback>(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<int64_t>(), 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<typename FAILING_CHECKPOINTER>
+    void countCheckpoints() {
+        FlakyCheckpointer counting(
+            [this](main::ClientContext& context) -> std::unique_ptr<Checkpointer> {
+                numCheckpoints++;
+                if (failCheckpoints) {
+                    return std::make_unique<FAILING_CHECKPOINTER>(context);
+                }
+                return std::make_unique<Checkpointer>(context);
+            });
+        counting.setCheckpointer(*getClientContext(*conn));
+    }
+
+    uint64_t numCheckpoints = 0;
+    bool failCheckpoints = true;
+};
+
+TEST_F(PostCommitCheckpointFailureTest, FailedAutoCheckpointDoesNotFailCommit) {
+    if (inMemMode) {
+        GTEST_SKIP();
+    }
+    countCheckpoints<FlakyCheckpointerFailsOnCheckpointStorage>();
+    ::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<FlakyCheckpointerFailsOnCheckpointStorage>();
+    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<main::Connection>(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<FlakyCheckpointerFailsOnApplyingShadow>();
+    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<FlakyCheckpointerFailsOnCheckpointStorage>();
+    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<FlakyCheckpointerFailsOnApplyingShadow>();
+    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<FlakyCheckpointerFailsOnCheckpointStorage>();
+    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<int64_t>(), 1);
+}
+
+TEST_F(PostCommitCheckpointFailureTest, FailedCheckpointCleanupFailsNextWrite) {
+    if (inMemMode) {
+        GTEST_SKIP();
+    }
+    countCheckpoints<CheckpointerFailsBeforeWALRetirement>();
+    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<bool> {};