fix(txn): log an ABORT record when a committed transaction cannot be applied

docs/checkpoint_recovery.md

@@ -32,6 +32,14 @@ reached the OS — but power-loss durability of those directory entries is not g
 If a step fails after the commit point, the database refuses further writes until it is reopened,
 and recovery finishes the checkpoint.
 
+## Failed commits
+
+A transaction whose changes cannot be applied after its `COMMIT` record is durable, for example
+because the buffer pool is full, is rolled back in memory and gets an `ABORT` record naming the WAL
+offset where its records start. Recovery skips the transactions such records name, so a `COMMIT`
+that reported an error is never applied. If the `ABORT` record cannot be written either, the
+database refuses further writes until it is reopened, and recovery then applies the transaction.
+
 ## Checkpoint bundle format
 
 `CheckpointRecord::bundleFormatVersion` is 2 for checkpoints written by current builds, including
@@ -94,10 +102,11 @@ An ordinary version 2 WAL tail — owner trailers on the records but no pending
 quieter hazard: an older build does not reliably refuse it. A record type the older build knows
 decodes with the trailer skipped as unknown trailing bytes and replays against the main catalog,
 silently misapplying records meant for a graph. A record type the older build does not know, such
-as `UPDATE_SEQUENCE_NAMED`, is no safer: with the default non-throwing replay configuration the
-decoding error is treated like a corrupt WAL tail — the older build replays the committed prefix
-and truncates the WAL there, silently discarding that transaction and every transaction recorded
-after it. Setting `throwOnWalReplayFailure` to true rejects an unknown record before that WAL
+as `UPDATE_SEQUENCE_NAMED` or `ABORT`, is no safer: with the default non-throwing replay
+configuration the decoding error is treated like a corrupt WAL tail — the older build replays the
+committed prefix and truncates the WAL there, silently discarding that transaction and every
+transaction recorded after it. For `ABORT`, which marks a transaction that failed after its
+`COMMIT` record was written, the older build also applies the transaction it undoes. Setting `throwOnWalReplayFailure` to true rejects an unknown record before that WAL
 is applied, but it still does not make downgrading safe: a WAL containing only recognized
 record types passes validation, and its owner trailers are ignored. This is why the successful
 `CHECKPOINT`, not a clean close, is the downgrade gate.

src/include/storage/wal/record/abort_record.h

@@ -0,0 +1,31 @@
+//===----------------------------------------------------------------------===//
+// This file is automatically generated by scripts/generate_wal_typespec.py.
+// Do not edit this file manually.
+//===----------------------------------------------------------------------===//
+
+#pragma once
+
+#include <cstdint>
+#include <memory>
+#include <utility>
+#include <vector>
+
+#include "storage/wal/record/wal_record_base.h"
+
+namespace lbug {
+namespace storage {
+
+struct AbortRecord final : WALRecord {
+    uint64_t abortedBlockStart;
+
+    AbortRecord() : WALRecord{WALRecordType::ABORT_RECORD}, abortedBlockStart{0} {}
+
+    explicit AbortRecord(uint64_t abortedBlockStart)
+        : WALRecord{WALRecordType::ABORT_RECORD}, abortedBlockStart{abortedBlockStart} {}
+
+    void serialize(common::Serializer& serializer) const override;
+    static std::unique_ptr<AbortRecord> deserialize(common::Deserializer& deserializer);
+};
+
+} // namespace storage
+} // namespace lbug

src/include/storage/wal/record/wal_record_base.h

@@ -25,6 +25,7 @@ enum class WALRecordType : uint8_t {
                         // accidentally read from an empty buffer.
     BEGIN_TRANSACTION_RECORD = 1,
     COMMIT_RECORD = 2,
+    ABORT_RECORD = 3,
 
     COPY_TABLE_RECORD = 13,
     CREATE_CATALOG_ENTRY_RECORD = 14,

src/include/storage/wal/wal.h

@@ -37,8 +37,12 @@ class WAL {
         common::VirtualFileSystem* vfs, StorageManager* storageManager = nullptr);
     ~WAL();
 
-    void logCommittedWAL(LocalWAL& localWAL, main::ClientContext* context,
-        uint64_t& commitSequence);
+    // blockStart, if given, receives the WAL offset where the transaction's records begin.
+    void logCommittedWAL(LocalWAL& localWAL, main::ClientContext* context, uint64_t& commitSequence,
+        uint64_t* blockStart = nullptr);
+    // Logs and syncs that the transaction whose records begin at abortedBlockStart was not
+    // applied after its COMMIT record became durable, so recovery skips it.
+    void logAbort(main::ClientContext* context, uint64_t abortedBlockStart);
     void logAndFlushCheckpoint(main::ClientContext* context);
 
     // Renames the active WAL to the frozen checkpoint WAL. Returns false if there is nothing to

src/include/storage/wal/wal_record.h

@@ -1,5 +1,6 @@
 #pragma once
 
+#include "storage/wal/record/abort_record.h"
 #include "storage/wal/record/alter_table_entry_record.h"
 #include "storage/wal/record/begin_transaction_record.h"
 #include "storage/wal/record/checkpoint_record.h"

src/include/storage/wal/wal_replayer.h

@@ -22,10 +22,17 @@ class Checkpointer;
 class StorageManager;
 class WALReplayer {
 public:
+    // A committed transaction's records [start, end) that a later ABORT record undid.
+    struct AbortedBlock {
+        uint64_t start;
+        uint64_t end;
+    };
+
     struct GraphRecoveryState {
         struct WALReplayRange {
             std::string path;
             uint64_t endOffset;
+            std::vector<AbortedBlock> abortedBlocks;
         };
 
         std::vector<WALReplayRange> walReplayRanges;
@@ -62,6 +69,8 @@ class WALReplayer {
         uint64_t offsetBeforeFirstCheckpoint = 0;
         common::uuid walDatabaseID{};
         uint64_t checkpointFormatVersion = 0;
+        // Sorted by start; replay skips these transactions.
+        std::vector<AbortedBlock> abortedBlocks;
     };
 
     void replayWALRecord(WALRecord& walRecord) const;

src/include/transaction/transaction.h

@@ -121,8 +121,7 @@ class LBUG_API Transaction {
 
     bool shouldForceCheckpoint() const;
 
-    void commit(storage::WAL* wal);
-    void writeCommitToWAL(storage::WAL* wal, uint64_t& walCommitSequence);
+    void writeCommitToWAL(storage::WAL* wal, uint64_t& walCommitSequence, uint64_t& walBlockStart);
     void publishCommit();
     void rollback(storage::WAL* wal);
     void pushCommitCallback(std::function<void(Transaction&)> callback);

src/include/transaction/transaction_manager.h

@@ -75,6 +75,8 @@ class TransactionManager {
 private:
     bool hasNoActiveTransactions() const;
     void checkpointNoLock(main::ClientContext& clientContext);
+    void logAbortOfDurableCommit(main::ClientContext& clientContext, uint64_t walBlockStart,
+        const std::string& publishError);
     // Try to checkpoint without blocking. Returns immediately if another checkpoint is in
     // progress. Used by auto-checkpoint after commit.
     void tryCheckpoint(main::ClientContext& clientContext);

src/storage/wal/CMakeLists.txt

@@ -24,6 +24,7 @@ add_library(lbug_storage_wal
         wal.cpp
         wal_record.cpp
         wal_replayer.cpp
+        records/abort_record.cpp
         records/begin_transaction_record.cpp
         records/commit_record.cpp
         records/checkpoint_record.cpp

src/storage/wal/records/abort_record.cpp

@@ -0,0 +1,32 @@
+//===----------------------------------------------------------------------===//
+// This file is automatically generated by scripts/generate_wal_typespec.py.
+// Do not edit this file manually.
+//===----------------------------------------------------------------------===//
+
+#include "common/serializer/deserializer.h"
+#include "common/serializer/serializer.h"
+#include "storage/wal/wal_record.h"
+
+using namespace lbug::common;
+namespace lbug {
+namespace storage {
+
+void AbortRecord::serialize(Serializer& serializer) const {
+    WALRecord::serialize(serializer);
+    serializer.writeDebuggingInfo("abortedBlockStart");
+    serializer.write<uint64_t>(abortedBlockStart);
+}
+
+std::unique_ptr<AbortRecord> AbortRecord::deserialize(Deserializer& deserializer) {
+    std::string key;
+    uint64_t abortedBlockStart = 0;
+    if (deserializer.hasRemainingData()) {
+        deserializer.validateDebuggingInfo(key, "abortedBlockStart");
+        deserializer.deserializeValue<uint64_t>(abortedBlockStart);
+    }
+
+    return std::make_unique<AbortRecord>(std::move(abortedBlockStart));
+}
+
+} // namespace storage
+} // namespace lbug

src/storage/wal/typespec/records/abort_record.tsp

@@ -0,0 +1,7 @@
+import "../common.tsp";
+import <cstdint>;
+
+// @explicit_constructor=true
+model AbortRecord extends WalRecordBase {
+    abortedBlockStart: uint64_t;
+}

src/storage/wal/wal.cpp

@@ -39,7 +39,7 @@ WAL::WAL(const std::string& dbPath, bool readOnly, bool enableChecksums, Virtual
 WAL::~WAL() {}
 
 void WAL::logCommittedWAL(LocalWAL& localWAL, main::ClientContext* context,
-    uint64_t& commitSequence) {
+    uint64_t& commitSequence, uint64_t* blockStart) {
     DASSERT(!readOnly);
     commitSequence = 0;
     if (inMemory || localWAL.getSize() == 0) {
@@ -48,11 +48,24 @@ void WAL::logCommittedWAL(LocalWAL& localWAL, main::ClientContext* context,
     std::unique_lock lck{mtx};
     throwIfPoisonedNoLock();
     initWriter(context);
+    if (blockStart != nullptr) {
+        *blockStart = serializer->getWriter()->getSize();
+    }
     localWAL.inMemWriter->flush(*serializer->getWriter());
     commitSequence = ++appendedCommitSequence;
     waitForDurabilityNoLock(commitSequence, lck);
 }
 
+// Holds the WAL lock across the sync, like the CHECKPOINT record: this runs only after a failure.
+void WAL::logAbort(main::ClientContext* context, uint64_t abortedBlockStart) {
+    DASSERT(!readOnly);
+    std::unique_lock lck{mtx};
+    throwIfPoisonedNoLock();
+    initWriter(context);
+    addNewWALRecordNoLock(AbortRecord{abortedBlockStart});
+    flushAndSyncNoLock();
+}
+
 void WAL::logAndFlushCheckpoint(main::ClientContext* context) {
     std::unique_lock lck{mtx};
     throwIfPoisonedNoLock();

src/storage/wal/wal_record.cpp

@@ -48,6 +48,9 @@ std::unique_ptr<WALRecord> WALRecord::deserialize(Deserializer& deserializer,
     case WALRecordType::COMMIT_RECORD: {
         walRecord = CommitRecord::deserialize(deserializer);
     } break;
+    case WALRecordType::ABORT_RECORD: {
+        walRecord = AbortRecord::deserialize(deserializer);
+    } break;
     case WALRecordType::CREATE_CATALOG_ENTRY_RECORD: {
         walRecord = CreateCatalogEntryRecord::deserialize(deserializer);
     } break;

src/storage/wal/wal_replayer.cpp

@@ -1,7 +1,9 @@
 #include "storage/wal/wal_replayer.h"
 
+#include <algorithm>
 #include <optional>
 #include <string>
+#include <unordered_map>
 
 #include "catalog/catalog.h"
 #include "common/constants.h"
@@ -75,6 +77,13 @@ class ReplayOwnerScope {
 };
 } // namespace
 
+static bool isInAbortedBlock(const std::vector<WALReplayer::AbortedBlock>& abortedBlocks,
+    uint64_t offset) {
+    const auto next = std::upper_bound(abortedBlocks.begin(), abortedBlocks.end(), offset,
+        [](uint64_t value, const WALReplayer::AbortedBlock& block) { return value < block.start; });
+    return next != abortedBlocks.begin() && offset < std::prev(next)->end;
+}
+
 static void removePartitionChildShadowFiles(main::ClientContext& clientContext) {
     auto* dbManager = main::DatabaseManager::Get(clientContext);
     if (dbManager == nullptr) {
@@ -479,7 +488,8 @@ WALReplayer::GraphRecoveryState WALReplayer::prepareGraphCheckpoint(StorageManag
             shadowDatabaseID);
         if (frozenWAL.has_value() && frozenWAL->isLastRecordCheckpoint && activeWAL.has_value() &&
             !activeWAL->isLastRecordCheckpoint) {
-            recoveryState.walReplayRanges.push_back({activeWALPath, activeWAL->offsetDeserialized});
+            recoveryState.walReplayRanges.push_back(
+                {activeWALPath, activeWAL->offsetDeserialized, activeWAL->abortedBlocks});
         } else if (activeWAL.has_value() && activeWAL->isLastRecordCheckpoint &&
                    frozenWAL.has_value() && !frozenWAL->isLastRecordCheckpoint) {
             throw RuntimeException(std::format(
@@ -490,10 +500,12 @@ WALReplayer::GraphRecoveryState WALReplayer::prepareGraphCheckpoint(StorageManag
     } else {
         removeFileAndSyncParentDirectory(graphShadowPath);
         if (frozenWAL.has_value()) {
-            recoveryState.walReplayRanges.push_back({frozenWALPath, frozenWAL->offsetDeserialized});
+            recoveryState.walReplayRanges.push_back(
+                {frozenWALPath, frozenWAL->offsetDeserialized, frozenWAL->abortedBlocks});
         }
         if (activeWAL.has_value()) {
-            recoveryState.walReplayRanges.push_back({activeWALPath, activeWAL->offsetDeserialized});
+            recoveryState.walReplayRanges.push_back(
+                {activeWALPath, activeWAL->offsetDeserialized, activeWAL->abortedBlocks});
         }
     }
     return recoveryState;
@@ -533,7 +545,11 @@ void WALReplayer::replayGraphWAL(StorageManager& storageManager,
                 storageManager.getOrInitDatabaseID(clientContext), walHeader.databaseID);
             deserializer.getReader()->onObjectEnd();
             while (getReadOffset(deserializer, *walEnableChecksums) < range.endOffset) {
+                const auto recordStart = getReadOffset(deserializer, *walEnableChecksums);
                 auto walRecord = WALRecord::deserialize(deserializer, clientContext);
+                if (isInAbortedBlock(range.abortedBlocks, recordStart)) {
+                    continue;
+                }
                 replayWALRecord(*walRecord);
             }
             truncateWALFile(*fileInfo, range.endOffset);
@@ -657,8 +673,10 @@ void WALReplayer::replayFrozenWAL(Checkpointer& checkpointer, bool throwOnWalRep
             const auto skipBefore = discardCheckpointPrefix ? replayInfo.lastCheckpointOffset : 0;
             while (getReadOffset(deserializer, enableChecksums) < replayEnd) {
                 DASSERT(!deserializer.finished());
+                const auto recordStart = getReadOffset(deserializer, enableChecksums);
                 auto walRecord = WALRecord::deserialize(deserializer, clientContext);
-                if (getReadOffset(deserializer, enableChecksums) <= skipBefore) {
+                if (getReadOffset(deserializer, enableChecksums) <= skipBefore ||
+                    isInAbortedBlock(replayInfo.abortedBlocks, recordStart)) {
                     continue;
                 }
                 replayWALRecord(*walRecord);
@@ -787,8 +805,10 @@ void WALReplayer::replayActiveWAL(Checkpointer& checkpointer, bool throwOnWalRep
             const auto skipBefore = discardCheckpointPrefix ? replayInfo.lastCheckpointOffset : 0;
             while (getReadOffset(deserializer, enableChecksums) < replayEnd) {
                 DASSERT(!deserializer.finished());
+                const auto recordStart = getReadOffset(deserializer, enableChecksums);
                 auto walRecord = WALRecord::deserialize(deserializer, clientContext);
-                if (getReadOffset(deserializer, enableChecksums) <= skipBefore) {
+                if (getReadOffset(deserializer, enableChecksums) <= skipBefore ||
+                    isInAbortedBlock(replayInfo.abortedBlocks, recordStart)) {
                     continue;
                 }
                 replayWALRecord(*walRecord);
@@ -929,6 +949,9 @@ void WALReplayer::replayCommittedCheckpoint(Checkpointer& checkpointer,
 WALReplayer::WALReplayInfo WALReplayer::dryReplay(FileInfo& fileInfo, bool throwOnWalReplayFailure,
     bool enableChecksums) const {
     WALReplayInfo replayInfo;
+    std::optional<uint64_t> openBlockStart;
+    std::unordered_map<uint64_t, uint64_t> committedBlockEnds;
+    std::vector<uint64_t> abortedBlockStarts;
     try {
         Deserializer deserializer = initDeserializer(fileInfo, clientContext, enableChecksums);
 
@@ -940,9 +963,20 @@ WALReplayer::WALReplayInfo WALReplayer::dryReplay(FileInfo& fileInfo, bool throw
 
         bool finishedDeserializing = deserializer.finished();
         while (!finishedDeserializing) {
+            const auto recordStart = getReadOffset(deserializer, enableChecksums);
             auto walRecord = WALRecord::deserialize(deserializer, clientContext);
             finishedDeserializing = deserializer.finished();
             switch (walRecord->type) {
+            case WALRecordType::BEGIN_TRANSACTION_RECORD: {
+                replayInfo.isLastRecordCheckpoint = false;
+                openBlockStart = recordStart;
+            } break;
+            case WALRecordType::ABORT_RECORD: {
+                // Ends a durable unit like a COMMIT, so a torn-tail truncation keeps it.
+                replayInfo.isLastRecordCheckpoint = false;
+                replayInfo.offsetDeserialized = getReadOffset(deserializer, enableChecksums);
+                abortedBlockStarts.push_back(walRecord->constCast<AbortRecord>().abortedBlockStart);
+            } break;
             case WALRecordType::CHECKPOINT_RECORD: {
                 // A CHECKPOINT record need not be final: a pinned checkpoint skips WAL
                 // rotation, so its record lands mid-file in the active WAL (and
@@ -967,6 +1001,10 @@ WALReplayer::WALReplayInfo WALReplayer::dryReplay(FileInfo& fileInfo, bool throw
                 // Update the offset to the end of the last commit record.
                 replayInfo.isLastRecordCheckpoint = false;
                 replayInfo.offsetDeserialized = getReadOffset(deserializer, enableChecksums);
+                if (openBlockStart.has_value()) {
+                    committedBlockEnds[*openBlockStart] = replayInfo.offsetDeserialized;
+                    openBlockStart.reset();
+                }
             } break;
             default: {
                 // Any other record after a checkpoint also ends the checkpoint suffix.
@@ -981,6 +1019,15 @@ WALReplayer::WALReplayInfo WALReplayer::dryReplay(FileInfo& fileInfo, bool throw
             throw;
         }
     }
+    // An ABORT record naming no committed transaction in this file is ignored.
+    for (const auto start : abortedBlockStarts) {
+        const auto block = committedBlockEnds.find(start);
+        if (block != committedBlockEnds.end()) {
+            replayInfo.abortedBlocks.push_back({start, block->second});
+        }
+    }
+    std::sort(replayInfo.abortedBlocks.begin(), replayInfo.abortedBlocks.end(),
+        [](const AbortedBlock& a, const AbortedBlock& b) { return a.start < b.start; });
     return replayInfo;
 }
 
@@ -1037,6 +1084,9 @@ void WALReplayer::replayWALRecord(WALRecord& walRecord) const {
     case WALRecordType::BEGIN_TRANSACTION_RECORD: {
         TransactionContext::Get(clientContext)->beginRecoveryTransaction();
     } break;
+    case WALRecordType::ABORT_RECORD: {
+        // The replay loops skip the transaction it names (see dryReplay).
+    } break;
     case WALRecordType::COMMIT_RECORD: {
         TransactionContext::Get(clientContext)->commit();
         // The enclosing recovery transaction just completed, so graph WALs queued by lazy