fix(wal): refuse writes after a WAL write fails instead of hanging

src/common/serializer/buffered_file.cpp

@@ -1,6 +1,7 @@
 #include "common/serializer/buffered_file.h"
 
 #include <cstring>
+#include <utility>
 
 #include "common/assert.h"
 #include "common/exception/runtime.h"
@@ -52,10 +53,11 @@ void BufferedFileWriter::flush() {
     if (bufferOffset == 0) {
         return;
     }
-    fileInfo.writeFile(buffer.get(), bufferOffset, fileOffset);
-    fileOffset += bufferOffset;
-    bufferOffset = 0;
-    memset(buffer.get(), 0, BUFFER_SIZE);
+    // The bytes leave the buffer even when the write fails: neither a later flush nor the
+    // destructor may write them after the caller has seen the failure.
+    const auto size = std::exchange(bufferOffset, 0);
+    fileInfo.writeFile(buffer.get(), size, fileOffset);
+    fileOffset += size;
 }
 
 void BufferedFileWriter::sync() {

src/include/common/serializer/buffered_file.h

@@ -18,6 +18,8 @@ class BufferedFileWriter final : public Writer {
     void write(const uint8_t* data, uint64_t size) override;
 
     void clear() override;
+    // The bytes leave the buffer even when the write fails; fileOffset advances only after it
+    // succeeds.
     void flush() override;
     void sync() override;
 

src/include/storage/wal/wal.h

@@ -73,6 +73,9 @@ class WAL {
     void retireFileNoLock(const std::string& path, const char* walName);
     void throwIfPoisonedNoLock() const;
     void poisonNoLock(const std::string& reason);
+    // A failed write can leave the WAL ending in part of a record, which later appends would
+    // follow, so writes are refused until a restart, whose recovery drops that tail.
+    [[noreturn]] void failWriteNoLock(const std::string& error);
     void waitForDurabilityNoLock(uint64_t commitSequence, std::unique_lock<std::mutex>& lck);
     void flushAndSyncNoLock();
     void writeHeader(main::ClientContext& context);

src/storage/wal/wal.cpp

@@ -48,7 +48,13 @@ void WAL::logCommittedWAL(LocalWAL& localWAL, main::ClientContext* context,
     std::unique_lock lck{mtx};
     throwIfPoisonedNoLock();
     initWriter(context);
-    localWAL.inMemWriter->flush(*serializer->getWriter());
+    try {
+        localWAL.inMemWriter->flush(*serializer->getWriter());
+    } catch (const std::exception& e) {
+        failWriteNoLock(e.what());
+    } catch (...) {
+        failWriteNoLock("unknown exception");
+    }
     commitSequence = ++appendedCommitSequence;
     waitForDurabilityNoLock(commitSequence, lck);
 }
@@ -186,6 +192,14 @@ void WAL::logAndFlushCheckpointToFrozen(main::ClientContext* context) {
     frozenSerializer->getWriter()->onObjectEnd();
     try {
         frozenSerializer->getWriter()->flush();
+    } catch (const std::exception& e) {
+        std::unique_lock lck{mtx};
+        failWriteNoLock(e.what());
+    } catch (...) {
+        std::unique_lock lck{mtx};
+        failWriteNoLock("unknown exception");
+    }
+    try {
         frozenSerializer->getWriter()->sync();
     } catch (const std::exception& e) {
         std::unique_lock lck{mtx};
@@ -281,7 +295,13 @@ void WAL::waitForDurabilityNoLock(uint64_t commitSequence, std::unique_lock<std:
         while (durableCommitSequence < appendedCommitSequence) {
             const auto targetSequence = appendedCommitSequence;
             const auto syncParentDirectory = !activeWALDirectorySynced;
-            serializer->getWriter()->flush();
+            try {
+                serializer->getWriter()->flush();
+            } catch (const std::exception& e) {
+                failWriteNoLock(e.what());
+            } catch (...) {
+                failWriteNoLock("unknown exception");
+            }
             auto* fileToSync = fileInfo.get();
             lck.unlock();
             try {
@@ -317,7 +337,13 @@ void WAL::waitForDurabilityNoLock(uint64_t commitSequence, std::unique_lock<std:
 
 // NOLINTNEXTLINE(readability-make-member-function-const): semantically non-const function.
 void WAL::flushAndSyncNoLock() {
-    serializer->getWriter()->flush();
+    try {
+        serializer->getWriter()->flush();
+    } catch (const std::exception& e) {
+        failWriteNoLock(e.what());
+    } catch (...) {
+        failWriteNoLock("unknown exception");
+    }
     try {
         serializer->getWriter()->sync();
         if (!activeWALDirectorySynced) {
@@ -338,6 +364,13 @@ void WAL::flushAndSyncNoLock() {
     }
 }
 
+void WAL::failWriteNoLock(const std::string& error) {
+    poisonNoLock("WAL write failed: " + error);
+    throw RuntimeException("WAL write failed; database is in a panic state and refuses further "
+                           "writes until restart. Original error: " +
+                           error);
+}
+
 uint64_t WAL::getFileSize() {
     std::unique_lock lck{mtx};
     if (!serializer) {

test/transaction/wal_test.cpp

@@ -61,6 +61,17 @@ class FailingSyncFileSystem final : public FileSystem, public DirectorySyncFileS
         failDirectorySync = fail;
     }
 
+    void setFailWrite(bool fail) {
+        std::unique_lock lck{mtx};
+        failWrite = fail;
+    }
+
+    uint64_t getFileSize(const std::string& path) const {
+        std::unique_lock lck{mtx};
+        const auto file = files.find(path);
+        return file == files.end() ? 0 : file->second.size();
+    }
+
     bool canHandleFile(const std::string_view path) const override {
         return path.starts_with("failing-sync://");
     }
@@ -167,6 +178,9 @@ class FailingSyncFileSystem final : public FileSystem, public DirectorySyncFileS
     void writeFile(FileInfo& fileInfo, const uint8_t* buffer, uint64_t numBytes,
         uint64_t offset) const override {
         std::unique_lock lck{mtx};
+        if (failWrite) {
+            throw IOException{"Injected WAL write failure."};
+        }
         auto& file = files[fileInfo.path];
         if (file.size() < offset + numBytes) {
             file.resize(offset + numBytes);
@@ -196,6 +210,7 @@ class FailingSyncFileSystem final : public FileSystem, public DirectorySyncFileS
     bool failRemove = false;
     bool failAfterRename = false;
     bool failDirectorySync = false;
+    bool failWrite = false;
     mutable uint64_t parentDirectorySyncCount = 0;
     mutable std::string parentDirectorySyncPath;
     mutable std::string truncatedPath;
@@ -236,6 +251,157 @@ TEST_F(WalTest, WALSyncFailurePoisonsWALAndReturnsAllocatedCommitSequence) {
     EXPECT_EQ(walCommitSequence, 0);
 }
 
+// A commit whose WAL write fails must not leave the WAL waiting for that write (#1397).
+TEST_F(WalTest, WALWriteFailurePoisonsWAL) {
+    VirtualFileSystem vfs;
+    auto failingFS = std::make_unique<FailingSyncFileSystem>(false /* failSync */);
+    auto* failingFSPtr = failingFS.get();
+    vfs.registerFileSystem(std::move(failingFS));
+
+    lbug::storage::WAL wal("failing-sync://db", false /* readOnly */, false /* enableChecksums */,
+        &vfs);
+    lbug::storage::LocalWAL localWAL(*lbug::storage::MemoryManager::Get(*conn->getClientContext()),
+        false /* enableChecksums */);
+    localWAL.logLoadExtension("" /* main */, "dummy");
+    localWAL.logCommit();
+    failingFSPtr->setFailWrite(true);
+    uint64_t walCommitSequence = 0;
+    std::string error;
+    try {
+        wal.logCommittedWAL(localWAL, conn->getClientContext(), walCommitSequence);
+    } catch (const Exception& e) {
+        error = e.what();
+    }
+    failingFSPtr->setFailWrite(false);
+    ASSERT_NE(error.find("panic state"), std::string::npos) << error;
+    // The flush failed after the commit took its sequence, as a failed sync does.
+    EXPECT_EQ(walCommitSequence, 1);
+
+    lbug::storage::LocalWAL secondLocalWAL(
+        *lbug::storage::MemoryManager::Get(*conn->getClientContext()), false /* enableChecksums */);
+    secondLocalWAL.logLoadExtension("" /* main */, "dummy2");
+    secondLocalWAL.logCommit();
+    EXPECT_THROW(wal.logCommittedWAL(secondLocalWAL, conn->getClientContext(), walCommitSequence),
+        RuntimeException);
+    EXPECT_EQ(walCommitSequence, 0);
+}
+
+// A commit larger than the writer's buffer reaches the file while it is appended, before the
+// flush that waits for durability.
+TEST_F(WalTest, WALAppendFailurePoisonsWAL) {
+    VirtualFileSystem vfs;
+    auto failingFS = std::make_unique<FailingSyncFileSystem>(false /* failSync */);
+    auto* failingFSPtr = failingFS.get();
+    vfs.registerFileSystem(std::move(failingFS));
+
+    lbug::storage::WAL wal("failing-sync://db", false /* readOnly */, true /* enableChecksums */,
+        &vfs);
+    lbug::storage::LocalWAL localWAL(*lbug::storage::MemoryManager::Get(*conn->getClientContext()),
+        true /* enableChecksums */);
+    localWAL.logLoadExtension("" /* main */, std::string(3 * LBUG_PAGE_SIZE, 'x'));
+    localWAL.logCommit();
+    failingFSPtr->setFailWrite(true);
+    uint64_t walCommitSequence = 0;
+    std::string error;
+    try {
+        wal.logCommittedWAL(localWAL, conn->getClientContext(), walCommitSequence);
+    } catch (const Exception& e) {
+        error = e.what();
+    }
+    failingFSPtr->setFailWrite(false);
+    ASSERT_NE(error.find("panic state"), std::string::npos) << error;
+    EXPECT_EQ(walCommitSequence, 0);
+
+    lbug::storage::LocalWAL secondLocalWAL(
+        *lbug::storage::MemoryManager::Get(*conn->getClientContext()), true /* enableChecksums */);
+    secondLocalWAL.logLoadExtension("" /* main */, "dummy2");
+    secondLocalWAL.logCommit();
+    EXPECT_THROW(wal.logCommittedWAL(secondLocalWAL, conn->getClientContext(), walCommitSequence),
+        RuntimeException);
+}
+
+// The bytes of a commit whose WAL write failed must not reach the file later, when closing.
+TEST_F(WalTest, FailedWALWriteIsNotWrittenOnClose) {
+    VirtualFileSystem vfs;
+    auto failingFS = std::make_unique<FailingSyncFileSystem>(false /* failSync */);
+    auto* failingFSPtr = failingFS.get();
+    vfs.registerFileSystem(std::move(failingFS));
+
+    auto wal = std::make_unique<lbug::storage::WAL>("failing-sync://db", false /* readOnly */,
+        false /* enableChecksums */, &vfs);
+    lbug::storage::LocalWAL localWAL(*lbug::storage::MemoryManager::Get(*conn->getClientContext()),
+        false /* enableChecksums */);
+    localWAL.logLoadExtension("" /* main */, "dummy");
+    localWAL.logCommit();
+    failingFSPtr->setFailWrite(true);
+    uint64_t walCommitSequence = 0;
+    try {
+        wal->logCommittedWAL(localWAL, conn->getClientContext(), walCommitSequence);
+    } catch (const Exception&) {} // NOLINT(bugprone-empty-catch): the failure is expected.
+    failingFSPtr->setFailWrite(false);
+    wal.reset();
+    EXPECT_EQ(
+        failingFSPtr->getFileSize(lbug::storage::StorageUtils::getWALFilePath("failing-sync://db")),
+        0);
+}
+
+TEST_F(WalTest, CheckpointRecordWriteFailurePoisonsWAL) {
+    VirtualFileSystem vfs;
+    auto failingFS = std::make_unique<FailingSyncFileSystem>(false /* failSync */);
+    auto* failingFSPtr = failingFS.get();
+    vfs.registerFileSystem(std::move(failingFS));
+
+    lbug::storage::WAL wal("failing-sync://db", false /* readOnly */, false /* enableChecksums */,
+        &vfs);
+    failingFSPtr->setFailWrite(true);
+    std::string error;
+    try {
+        wal.logAndFlushCheckpoint(conn->getClientContext());
+    } catch (const Exception& e) {
+        error = e.what();
+    }
+    failingFSPtr->setFailWrite(false);
+    ASSERT_NE(error.find("panic state"), std::string::npos) << error;
+    EXPECT_TRUE(wal.mayHaveCheckpointRecord());
+
+    lbug::storage::LocalWAL localWAL(*lbug::storage::MemoryManager::Get(*conn->getClientContext()),
+        false /* enableChecksums */);
+    localWAL.logLoadExtension("" /* main */, "dummy");
+    localWAL.logCommit();
+    uint64_t walCommitSequence = 0;
+    EXPECT_THROW(wal.logCommittedWAL(localWAL, conn->getClientContext(), walCommitSequence),
+        RuntimeException);
+}
+
+// A failed write of the CHECKPOINT record into the frozen WAL is reported as a write failure.
+TEST_F(WalTest, FrozenCheckpointRecordWriteFailurePoisonsWAL) {
+    VirtualFileSystem vfs;
+    auto failingFS = std::make_unique<FailingSyncFileSystem>(false /* failSync */);
+    auto* failingFSPtr = failingFS.get();
+    vfs.registerFileSystem(std::move(failingFS));
+
+    lbug::storage::WAL wal("failing-sync://db", false /* readOnly */, false /* enableChecksums */,
+        &vfs);
+    lbug::storage::LocalWAL localWAL(*lbug::storage::MemoryManager::Get(*conn->getClientContext()),
+        false /* enableChecksums */);
+    localWAL.logLoadExtension("" /* main */, "dummy");
+    localWAL.logCommit();
+    uint64_t walCommitSequence = 0;
+    ASSERT_NO_THROW(wal.logCommittedWAL(localWAL, conn->getClientContext(), walCommitSequence));
+    ASSERT_NO_THROW(wal.rotateForCheckpoint(conn->getClientContext()));
+
+    failingFSPtr->setFailWrite(true);
+    std::string error;
+    try {
+        wal.logAndFlushCheckpointToFrozen(conn->getClientContext());
+    } catch (const Exception& e) {
+        error = e.what();
+    }
+    failingFSPtr->setFailWrite(false);
+    EXPECT_NE(error.find("WAL write failed"), std::string::npos) << error;
+    EXPECT_THROW(wal.throwIfPoisoned(), RuntimeException);
+}
+
 TEST_F(WalTest, CheckpointSyncFailureReportsPossibleCommitRecord) {
     VirtualFileSystem vfs;
     auto failingFS = std::make_unique<FailingSyncFileSystem>(true /* failSync */);