diff --git a/src/common/serializer/buffered_file.cpp b/src/common/serializer/buffered_file.cpp index 47a1d7c66c..6c77c92c0f 100644 --- a/src/common/serializer/buffered_file.cpp +++ b/src/common/serializer/buffered_file.cpp @@ -1,6 +1,7 @@ #include "common/serializer/buffered_file.h" #include +#include #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() { diff --git a/src/include/common/serializer/buffered_file.h b/src/include/common/serializer/buffered_file.h index 3279561b44..22025db4c0 100644 --- a/src/include/common/serializer/buffered_file.h +++ b/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; diff --git a/src/include/storage/wal/wal.h b/src/include/storage/wal/wal.h index c0aaebaa42..76490c5b21 100644 --- a/src/include/storage/wal/wal.h +++ b/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& lck); void flushAndSyncNoLock(); void writeHeader(main::ClientContext& context); diff --git a/src/storage/wal/wal.cpp b/src/storage/wal/wal.cpp index a4328e5e2f..fb9f1d09b1 100644 --- a/src/storage/wal/wal.cpp +++ b/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_lockgetWriter()->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_lockgetWriter()->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) { diff --git a/test/transaction/wal_test.cpp b/test/transaction/wal_test.cpp index a7bde256a5..8dd684e459 100644 --- a/test/transaction/wal_test.cpp +++ b/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(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(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(false /* failSync */); + auto* failingFSPtr = failingFS.get(); + vfs.registerFileSystem(std::move(failingFS)); + + auto wal = std::make_unique("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(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(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(true /* failSync */);