diff --git a/doc/source/transaction/transaction.md b/doc/source/transaction/transaction.md index 6df9b01e6..c996fb8df 100644 --- a/doc/source/transaction/transaction.md +++ b/doc/source/transaction/transaction.md @@ -201,6 +201,40 @@ session.close() An optional checkpoint consolidates the WAL but is not required for statement durability. See [Checkpoints](checkpoint.md). +### WAL File Format (v1) + +Each WAL file lives under the `wal` directory of exactly one checkpoint and +uses a framed, self-describing layout. Checkpoint ownership comes from the +directory the file was written in (guaranteed by checkpoint rotation), not +from a field inside the file: + +```text +FileHeader (16B) + └─ magic / format_version / header_size / reserved +Frame* (one frame per committed transaction) + ├─ FrameHeader (13B): record_kind, payload_length, + │ commit_timestamp, frame_checksum + │ (CRC32C over header prefix + payload) + └─ payload # redo bytes; empty only for compaction frames +``` + +Key properties: + +- **One commit = exactly one complete frame.** The frame checksum lives in + the frame header and is computed over the payload before any byte is + written, so a frame is complete exactly when it is fully present and its + checksum matches; no trailing commit marker is needed. +- **Record kinds:** `kInsert` (append redo), `kCowUpdate` (update/delete redo), + `kCompact` (compaction, empty payload). AP/embedded paths never write WAL. +- **Real logical EOF.** Files are never preallocated or zero-padded; anything + after the last complete frame is torn-write residue from a crash. +- **Recovery validates first, then replays.** All files of the current + checkpoint's `wal` directory are parsed and checksummed up front; frames + from all writers are merged by `commit_timestamp` and replayed in global + order. Duplicate timestamps, unknown record kinds, legacy-format files and + any corruption *before* the final frame are hard errors. Only an incomplete + frame at EOF is silently discarded (the crashed transaction never committed). + ## Error Recovery ### Embedded Mode diff --git a/include/neug/transaction/compact_transaction.h b/include/neug/transaction/compact_transaction.h index 55c6b80ba..0c6d17e80 100644 --- a/include/neug/transaction/compact_transaction.h +++ b/include/neug/transaction/compact_transaction.h @@ -16,7 +16,6 @@ #include "neug/storages/graph_snapshot_store.h" #include "neug/utils/property/types.h" -#include "neug/utils/serialization/in_archive.h" namespace neug { @@ -41,8 +40,6 @@ class CompactTransaction { IWalWriter& wal_writer_; IVersionManager& vm_; timestamp_t timestamp_; - - InArchive arc_; }; } // namespace neug diff --git a/include/neug/transaction/insert_transaction.h b/include/neug/transaction/insert_transaction.h index f66d0380a..24e4efbab 100644 --- a/include/neug/transaction/insert_transaction.h +++ b/include/neug/transaction/insert_transaction.h @@ -192,7 +192,7 @@ class InsertTransaction { * @param length Byte length of @p data. * @param alloc Per-thread allocator for adjacency-list growth in CSR. */ - static void IngestWal(GraphView& view, uint32_t timestamp, char* data, + static void IngestWal(GraphView& view, uint32_t timestamp, const char* data, size_t length, Allocator& alloc); const Schema& schema() const; diff --git a/include/neug/transaction/update_transaction.h b/include/neug/transaction/update_transaction.h index 08eb35e02..da214b051 100644 --- a/include/neug/transaction/update_transaction.h +++ b/include/neug/transaction/update_transaction.h @@ -110,8 +110,8 @@ class UpdateTransaction { void Abort(); - static void IngestWal(PropertyGraph& graph, uint32_t timestamp, char* data, - size_t length, Allocator& alloc); + static void IngestWal(PropertyGraph& graph, uint32_t timestamp, + const char* data, size_t length, Allocator& alloc); const GraphView& view() const { return view_; } diff --git a/include/neug/transaction/wal/dummy_wal_writer.h b/include/neug/transaction/wal/dummy_wal_writer.h index de3848955..1f0d5eb7a 100644 --- a/include/neug/transaction/wal/dummy_wal_writer.h +++ b/include/neug/transaction/wal/dummy_wal_writer.h @@ -22,7 +22,8 @@ namespace neug { /** * @brief DummyWalWriter is a no-op implementation of the IWalWriter interface. - * It is used when write-ahead logging is disabled or not required. + * It is used as a test spy/fake for TP transactions; the AP execution path + * must never use it to simulate a NoWal boundary. */ class DummyWalWriter : public IWalWriter { public: @@ -34,6 +35,13 @@ class DummyWalWriter : public IWalWriter { void open(const std::string& wal_uri) override; void close() override; - bool append(const char* data, size_t length) override; + bool append_frame(uint32_t commit_timestamp, WalRecordKind kind, + const char* payload, size_t length) override; + + /// Number of frames accepted since construction (test observability). + size_t appended_frame_num() const { return appended_frame_num_; } + + private: + size_t appended_frame_num_{0}; }; } // namespace neug diff --git a/include/neug/transaction/wal/local_wal_parser.h b/include/neug/transaction/wal/local_wal_parser.h index 7d5c819ea..815bf07eb 100644 --- a/include/neug/transaction/wal/local_wal_parser.h +++ b/include/neug/transaction/wal/local_wal_parser.h @@ -24,6 +24,22 @@ namespace neug { +/** + * Local WAL parser implementing the P1-2 recovery protocol: + * validate first, then order, then replay. + * + * open() validates every file header, frame header, frame checksum and + * commit trailer of the given wal_dir() against the v1 protocol. Only the + * last candidate frame of a file may be dropped as crash residue when it is + * truncated by EOF; every other inconsistency rejects the whole recovery + * with a typed WalRecoveryException. After all files are validated, frames + * are merged into a strictly timestamp-ascending sequence and duplicate + * timestamps are rejected. No graph mutation happens here. + * + * Validated files stay memory-mapped for the parser's lifetime; replay unit + * payloads are zero-copy views into those mappings. The mappings are + * released by close() or destruction, which invalidates the units. + */ class LocalWalParser : public IWalParser { public: static std::unique_ptr Make(const std::string& wal_dir) { @@ -31,24 +47,23 @@ class LocalWalParser : public IWalParser { } explicit LocalWalParser(const std::string& wal_uri); - ~LocalWalParser() { close(); } + ~LocalWalParser() override; void open(const std::string& wal_uri) override; void close() override; uint32_t last_ts() const override; - const WalContentUnit& get_insert_wal(uint32_t ts) const override; - const std::vector& get_update_wals() const override; + const std::vector& replay_units() const override; private: - std::vector fds_; - std::vector mmapped_ptrs_; - std::vector mmapped_size_; - std::vector insert_wal_list_; + /// Source files kept mapped so replay unit payload views stay valid. + /// WalReplayUnit::file_index indexes this mapping table, which stays + /// opaque here because its element type is an implementation detail. + struct MappedFiles; + std::unique_ptr mapped_files_; + std::vector replay_units_; uint32_t last_ts_{0}; - std::vector update_wal_list_; - static const bool registered_; }; diff --git a/include/neug/transaction/wal/local_wal_writer.h b/include/neug/transaction/wal/local_wal_writer.h index 3dc00d281..19e6c228b 100644 --- a/include/neug/transaction/wal/local_wal_writer.h +++ b/include/neug/transaction/wal/local_wal_writer.h @@ -15,6 +15,7 @@ #pragma once #include +#include #include #include @@ -22,31 +23,53 @@ namespace neug { +/** + * Append-only local WAL writer producing the v1 framed format. + * + * The file holds a real logical EOF: no ftruncate() preallocation and no + * zero-byte terminator. open() persists and syncs the file header first; + * frames are only accepted afterwards. Each append_frame() writes the frame + * header (whose checksum is computed over the payload before any byte is + * written) followed by the payload, then syncs once per frame. + */ class LocalWalWriter : public IWalWriter { public: static std::unique_ptr Make(const std::string& wal_uri, int slot_id); - static constexpr size_t TRUNC_SIZE = 1ul << 30; LocalWalWriter(const std::string& wal_uri, int slot_id) - : wal_uri_(wal_uri), - slot_id_(slot_id), - fd_(-1), - file_size_(0), - file_used_(0) {} + : wal_uri_(wal_uri), slot_id_(slot_id), fd_(-1), append_offset_(0) {} ~LocalWalWriter() noexcept override; void open(const std::string& wal_uri) override; void close() override; - bool append(const char* data, size_t length) override; + bool append_frame(uint32_t commit_timestamp, WalRecordKind kind, + const char* payload, size_t length) override; std::string type() const override { return "file"; } + /// Test seam: fail the next write_all() call at the given phase. The + /// injection is one-shot and cleared after it fires. + enum class FailNextWrite { kNone, kHeader, kPayload }; + void fail_next_write(FailNextWrite phase) { fail_next_write_ = phase; } + private: + /// Writes every byte of @p buffer, handling short writes. Throws + /// IOException on error. Returns false only when the write failure was + /// injected via fail_next_write(). + bool write_all(const char* buffer, size_t length, FailNextWrite phase); + void sync_file(); + /// Restores the clean logical EOF at @p offset, discarding the bytes of a + /// frame whose write failed. Returns false and marks the writer failed if + /// the file cannot be restored; a failed writer rejects further frames. + bool restore_clean_eof(size_t offset); + std::string wal_uri_; + std::string path_; int slot_id_; int fd_; - size_t file_size_; - size_t file_used_; + size_t append_offset_; + FailNextWrite fail_next_write_{FailNextWrite::kNone}; + bool failed_{false}; static const bool registered_; }; diff --git a/include/neug/transaction/wal/wal.h b/include/neug/transaction/wal/wal.h index 6d2714a89..8771b7914 100644 --- a/include/neug/transaction/wal/wal.h +++ b/include/neug/transaction/wal/wal.h @@ -18,35 +18,20 @@ #include #include #include +#include #include #include #include "neug/common/types/value.h" #include "neug/storages/graph/operation_params.h" #include "neug/transaction/transaction_utils.h" +#include "neug/transaction/wal/wal_codec.h" #include "neug/utils/property/types.h" #include "neug/utils/serialization/in_archive.h" #include "neug/utils/serialization/out_archive.h" namespace neug { -struct WalHeader { - uint32_t timestamp; - uint8_t type : 1; - int32_t length : 31; -}; - -struct WalContentUnit { - char* ptr{NULL}; - size_t size{0}; -}; - -struct UpdateWalUnit { - uint32_t timestamp{0}; - char* ptr{NULL}; - size_t size{0}; -}; - std::string get_wal_uri_scheme(const std::string& uri); std::string get_wal_uri_path(const std::string& uri); @@ -56,6 +41,9 @@ std::string get_wal_uri_path(const std::string& uri); * Implementations own their resources and must release them without throwing * from their destructors. close() remains available for callers that need * explicit error reporting. + * + * A writer belongs to the wal_dir() it was opened on; checkpoint ownership + * of WAL files is guaranteed by checkpoint rotation, not by writer state. */ class IWalWriter { public: @@ -63,9 +51,10 @@ class IWalWriter { virtual std::string type() const = 0; /** - * Open a WAL file. In service mode, each logical execution slot owns one - * writer. The slot may move between pthread workers and must retain the same - * writer for the full transaction. + * Open a WAL file under @p wal_uri and persist the v1 file header. + * In service mode, each logical execution slot owns one writer. The slot + * may move between pthread workers and must retain the same writer for the + * full transaction. * The uri could be a file_path or a remote connection string. */ virtual void open(const std::string& wal_uri) = 0; @@ -77,20 +66,42 @@ class IWalWriter { virtual void close() = 0; /** - * Append data to the wal file. + * Append one complete transaction frame: the frame header (including the + * checksum computed over the payload) followed by the payload itself. */ - virtual bool append(const char* data, size_t length) = 0; + virtual bool append_frame(uint32_t commit_timestamp, WalRecordKind kind, + const char* payload, size_t length) = 0; +}; + +/// A validated replay record produced by the parser. +/// +/// The payload is a view into a WAL file that the parser keeps mapped; it is +/// valid until the parser is closed or destroyed. file_index and +/// source_offset locate the frame inside the parser's source files and are +/// used for diagnostics. +struct WalReplayUnit { + uint32_t commit_timestamp{0}; + WalRecordKind kind{WalRecordKind::kInsert}; + std::string_view payload; + uint32_t file_index{0}; + uint64_t source_offset{0}; }; /** * The interface of wal parser. + * + * open() validates every WAL file in the given wal_dir() against the v1 + * protocol, collects all complete frames across writer files and exposes + * them as a strictly timestamp-ordered, duplicate-free replay sequence. Any + * validation failure aborts open() with a typed recovery error before any + * graph mutation. */ class IWalParser { public: virtual ~IWalParser() {} /** - * Open wals from a uri and parse the wal files. + * Open and validate the wal files under @p wal_uri. */ virtual void open(const std::string& wal_uri) = 0; @@ -98,15 +109,13 @@ class IWalParser { virtual uint32_t last_ts() const = 0; - /* - * Get the insert wal unit with the given timestamp. - */ - virtual const WalContentUnit& get_insert_wal(uint32_t ts) const = 0; - /** - * Get all the update wal units. + * All validated frames across every writer file, strictly ascending by + * commit timestamp. Duplicate timestamps are rejected during open(). + * Payloads are views into files the parser keeps mapped; they stay valid + * until close() or destruction. */ - virtual const std::vector& get_update_wals() const = 0; + virtual const std::vector& replay_units() const = 0; }; class WalWriterFactory { diff --git a/include/neug/transaction/wal/wal_builder.h b/include/neug/transaction/wal/wal_builder.h index 486d40ea9..6604a8909 100644 --- a/include/neug/transaction/wal/wal_builder.h +++ b/include/neug/transaction/wal/wal_builder.h @@ -27,15 +27,20 @@ namespace neug { -/// Accumulates WAL operations for a single update transaction. +/// Accumulates WAL redo operations for a single update transaction. /// /// Each LogXxx method serializes the corresponding redo entry into an internal /// buffer and increments the operation count. DDL Log methods additionally set /// schema_changed_ = true. /// +/// WalBuilder only records redo bytes. File headers, frame headers, +/// checksums, commit markers and sync are owned by the frame writer and +/// never leak into the Log* methods. +/// /// UpdateTransaction::Commit() uses: -/// - op_num() == 0 → nothing to do, early return -/// - op_num() > 0 → must publish snapshot +/// - op_num() == 0 and size() == 0 → nothing to do, early return +/// - size() > 0 → append a kCowUpdate frame +/// - op_num() > 0 and size() == 0 → append an empty kCompact frame class WalBuilder { public: WalBuilder(); @@ -86,15 +91,13 @@ class WalBuilder { int op_num() const { return op_num_; } bool schema_changed() const { return schema_changed_; } - /// Size of the WAL content excluding its header. - size_t content_size() const { return arc_.GetSize() - sizeof(WalHeader); } - - /// Finalize the WAL header. Call only when op_num() > 0. - void finalize(timestamp_t timestamp); + /// Size of the redo payload. 0 means the transaction carries no redo + /// bytes (e.g. a checkpoint-only commit serialized as an empty kCompact + /// frame). + size_t size() const { return arc_.GetSize(); } - /// Full buffer (header + content) after finalize(). + /// Redo payload bytes. char* data() { return arc_.GetBuffer(); } - size_t size() const { return arc_.GetSize(); } /// Reset all state for reuse or release. void clear(); diff --git a/include/neug/transaction/wal/wal_codec.h b/include/neug/transaction/wal/wal_codec.h new file mode 100644 index 000000000..1c519ca43 --- /dev/null +++ b/include/neug/transaction/wal/wal_codec.h @@ -0,0 +1,170 @@ +/** Copyright 2020 Alibaba Group Holding Limited. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#pragma once + +#include +#include +#include +#include + +namespace neug { + +// ============================================================================= +// WAL v1 on-disk protocol. +// +// The framing layer is encoded/decoded byte-by-byte through the explicit +// codec functions in this header; it never goes through reinterpret_cast, +// C++ bit-fields or struct dumps. This is a parse-safety property (bounds +// checks, rejection of invalid values), not cross-platform portability: the +// redo payload and the data files are persisted in host layout, so a WAL +// file is only meaningful on the machine that wrote it. +// +// File layout: +// WalFileHeader | (WalFrameHeader | payload)* +// +// The frame checksum lives in the frame header and covers the remaining +// header bytes plus the payload. A frame is therefore complete exactly when +// it is fully present and its checksum matches; recovery never needs a +// trailing commit marker: +// - EOF with fewer bytes than the frame needs -> torn tail, discarded +// - full frame, checksum matches -> committed, replayed +// - full frame, checksum mismatch -> corruption, rejected +// +// A file without a valid WalFileHeader is never silently interpreted; the +// legacy pre-v1 format is rejected with a typed recovery error. +// ============================================================================= + +// "NEUW" as a little-endian u32 constant (codec implementation detail). +constexpr uint32_t kWalFileMagic = 0x5755454Eu; + +constexpr uint32_t kWalFormatVersion = 1; + +constexpr uint32_t kWalFileHeaderSize = 16; +constexpr uint32_t kWalFrameHeaderSize = 13; + +// Protocol upper bound for a frame payload: the on-wire length field is a +// u32. Checked before any conversion to size_t so corrupted values cannot +// trigger integer overflow, out-of-bounds reads, or huge allocations. +constexpr uint64_t kWalMaxPayloadLength = 0xFFFFFFFFull; + +/// Explicit record categories. Unknown values are always rejected; semantics +/// are never inferred from payload length or bit flags. +enum class WalRecordKind : uint32_t { + kInsert = 1, + kCowUpdate = 2, + kCompact = 3, +}; + +/// Typed WAL recovery error categories. Messages produced alongside these +/// must include the WAL path, the byte offset and the timestamp so operators +/// can locate the problem. +enum class WalRecoveryErrorKind { + kUnsupportedFormat, + kCorruptedFrame, + kDuplicateTimestamp, + kUnknownRecordKind, +}; + +std::string WalRecordKindName(WalRecordKind kind); +std::string WalRecoveryErrorKindName(WalRecoveryErrorKind kind); + +/// Per-file fixed header. A WAL file belongs to the checkpoint whose +/// wal_dir() it lives in; that ownership is guaranteed by checkpoint +/// rotation, not by a field inside the file. The writer identity is already +/// carried by the file name (thread__.wal), so the header +/// keeps no diagnostic slot field. +/// +/// All fields are validated by exact match on decode, so no header checksum +/// is needed: only the reserved bytes would be protected. +/// +/// Wire layout (little-endian): +/// magic u32 | format_version u32 | header_size u32 | reserved u32 +struct WalFileHeader { + uint32_t magic{kWalFileMagic}; + uint32_t format_version{kWalFormatVersion}; + uint32_t header_size{kWalFileHeaderSize}; + uint32_t reserved{0}; +}; + +/// Per-transaction frame header, written before the payload. The frame +/// checksum is part of the header, so the writer computes it over the +/// payload before writing anything. +/// +/// The format version is a file-level property carried by WalFileHeader; a +/// single writer file never mixes frame versions, so no per-frame version is +/// stored. There is no per-frame magic either: frames are parsed strictly +/// sequentially from a validated file header, and the checksum rejects any +/// corrupted frame regardless of its position. +/// +/// Wire layout (little-endian, fields packed in order; padding is only ever +/// appended at the end, never between fields): +/// record_kind u8 | payload_length u32 | commit_timestamp u32 | +/// frame_checksum u32 +struct WalFrameHeader { + WalRecordKind record_kind{WalRecordKind::kInsert}; + uint32_t payload_length{0}; + uint32_t commit_timestamp{0}; + /// CRC32C (Castagnani, hardware-accelerated via absl) over the encoded + /// header bytes preceding this field and the payload. Computed by + /// EncodeWalFrameHeader; callers never fill it. + uint32_t frame_checksum{0}; +}; + +/// Status of a decode step. Decode functions never read past @c remaining. +enum class WalDecodeStatus { + kOk = 0, + /// Fewer bytes available than the structure needs. Only meaningful for the + /// last candidate frame of a file (crash residue). + kTruncated, + kBadMagic, + kBadVersion, + kBadHeaderSize, + kBadChecksum, + kUnknownRecordKind, + kPayloadTooLarge, +}; + +std::string WalDecodeStatusName(WalDecodeStatus status); + +/// Encoders. All checksum fields are computed here; callers never fill them. +std::array EncodeWalFileHeader( + const WalFileHeader& header); +/// Computes @c frame_checksum over the encoded header bytes preceding the +/// checksum field plus the payload, protecting the kind, timestamp and +/// length fields, not only the payload. +std::array EncodeWalFrameHeader( + const WalFrameHeader& header, const uint8_t* payload, + size_t payload_length); + +/// Decoders. Each returns the decode status and, on success, fills @p out and +/// reports the consumed byte count. +WalDecodeStatus DecodeWalFileHeader(const uint8_t* data, size_t remaining, + WalFileHeader& out, size_t& consumed); +WalDecodeStatus DecodeWalFrameHeader(const uint8_t* data, size_t remaining, + WalFrameHeader& out, size_t& consumed); + +/// Validates a fully read frame: recomputes the frame checksum over the +/// header bytes preceding the checksum field plus the payload. Returns kOk +/// on success. +WalDecodeStatus ValidateWalFrame(const WalFrameHeader& header, + const uint8_t* frame_header_bytes, + const uint8_t* payload); + +/// Header bytes covered by frame_checksum: everything except the checksum +/// field itself. +constexpr size_t kWalFrameHeaderStableSize = + kWalFrameHeaderSize - sizeof(uint32_t); + +} // namespace neug diff --git a/include/neug/utils/exception/exception.h b/include/neug/utils/exception/exception.h index 16c4be378..f01f87a5c 100644 --- a/include/neug/utils/exception/exception.h +++ b/include/neug/utils/exception/exception.h @@ -240,6 +240,13 @@ class NEUG_API TxStateConflictException : public Exception { const std::string& file_line); }; +class NEUG_API WalRecoveryException : public Exception { + public: + explicit WalRecoveryException(const std::string& msg); + + WalRecoveryException(const std::string& msg, const std::string& file_line); +}; + } // namespace exception } // namespace neug @@ -344,6 +351,9 @@ class NEUG_API TxStateConflictException : public Exception { #define THROW_TX_STATE_CONFLICT(msg) \ THROW_EXCEPTION_WITH_FILE_LINE_AND_TYPE(TxStateConflictException, msg) +#define THROW_WAL_RECOVERY_EXCEPTION(msg) \ + THROW_EXCEPTION_WITH_FILE_LINE_AND_TYPE(WalRecoveryException, msg) + #define THROW_IF_ARROW_NOT_OK(expr) \ do { \ auto status = (expr); \ diff --git a/include/neug/utils/serialization/out_archive.h b/include/neug/utils/serialization/out_archive.h index 2281616aa..2513aef4c 100644 --- a/include/neug/utils/serialization/out_archive.h +++ b/include/neug/utils/serialization/out_archive.h @@ -64,9 +64,9 @@ class OutArchive { inline void Rewind() { begin_ = buffer_.data(); } - inline void SetSlice(char* buffer, size_t size) { + inline void SetSlice(const char* buffer, size_t size) { buffer_.clear(); - begin_ = buffer; + begin_ = const_cast(buffer); end_ = begin_ + size; } diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index 8a825b860..9b5ab00a1 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -35,6 +35,8 @@ if (ENABLE_GCOV) endif() target_link_libraries(neug PUBLIC OpenSSL::SSL OpenSSL::Crypto) target_link_libraries(neug PRIVATE ${GFLAGS_LIBRARIES} ${YAML_CPP_LIBRARIES} ${COMPILER_LIBRARIES}) +# CRC32C for WAL frame checksums (hardware-accelerated on x86/ARM64). +target_link_libraries(neug PRIVATE absl::crc32c) target_link_libraries(neug PUBLIC ${GLOG_LIBRARIES}) # Protobuf is PRIVATE to avoid double-initialization of global statics when # downstream shared libraries (e.g. neug_py_bind.so) also load libneug.dylib. diff --git a/src/main/neug_db.cc b/src/main/neug_db.cc index bd5f99907..57c9a73aa 100644 --- a/src/main/neug_db.cc +++ b/src/main/neug_db.cc @@ -58,26 +58,6 @@ inline std::string allocator_prefix(const std::string& allocator_dir, } class Connection; -static void IngestWalRange(PropertyGraph& graph, - std::vector>& allocators, - const IWalParser& parser, uint32_t from, - uint32_t to) { - if (from >= to) { - return; - } - // Build a single writable GraphView covering the whole replay range. - // read_ts = MAX_TIMESTAMP so vertices inserted earlier in the loop are - // visible to later edge-resolution lookups regardless of the per-unit - // commit timestamp. - GraphView view(graph); - for (size_t j = from; j < to; ++j) { - const auto& unit = parser.get_insert_wal(j); - InsertTransaction::IngestWal(view, j, unit.ptr, unit.size, *allocators[0]); - if (j % 1000000 == 0) { - LOG(INFO) << "Ingested " << j << " WALs"; - } - } -} NeugDB::NeugDB() : closed_(true), is_pure_memory_(false), max_thread_num_(1) {} @@ -452,25 +432,47 @@ timestamp_t NeugDB::openGraphAndIngestWals() { } timestamp_t NeugDB::ingestWals(IWalParser& parser, PropertyGraph& graph) { - uint32_t from_ts = 1; - LOG(INFO) << "Ingesting update wals size: " - << parser.get_update_wals().size(); - - for (auto& update_wal : parser.get_update_wals()) { - uint32_t to_ts = update_wal.timestamp; - if (from_ts < to_ts) { - IngestWalRange(graph, allocators_, parser, from_ts, to_ts); + // The parser already validated every frame and ordered them strictly by + // commit timestamp; replay consumes that unified sequence without any + // re-sorting or gap assumptions here. Unit payloads are views into files + // the parser keeps mapped, so @p parser must outlive this call. + const auto& units = parser.replay_units(); + LOG(INFO) << "Ingesting " << units.size() << " wal frames"; + + size_t i = 0; + size_t ingested = 0; + while (i < units.size()) { + const auto& unit = units[i]; + switch (unit.kind) { + case WalRecordKind::kInsert: { + // Replay a consecutive insert run through one writable GraphView so + // vertices inserted earlier resolve edges inserted later. + GraphView view(graph); + while (i < units.size() && units[i].kind == WalRecordKind::kInsert) { + const auto& insert_unit = units[i]; + InsertTransaction::IngestWal( + view, insert_unit.commit_timestamp, insert_unit.payload.data(), + insert_unit.payload.size(), *allocators_[0]); + ++i; + if (++ingested % 1000000 == 0) { + LOG(INFO) << "Ingested " << ingested << " wal frames"; + } + } + break; + } + case WalRecordKind::kCowUpdate: { + UpdateTransaction::IngestWal(graph, unit.commit_timestamp, + unit.payload.data(), unit.payload.size(), + *allocators_[0]); + ++i; + break; } - if (update_wal.size == 0) { + case WalRecordKind::kCompact: { graph.Compact(); - } else { - UpdateTransaction::IngestWal(graph, to_ts, update_wal.ptr, - update_wal.size, *allocators_[0]); + ++i; + break; + } } - from_ts = to_ts + 1; - } - if (from_ts <= parser.last_ts()) { - IngestWalRange(graph, allocators_, parser, from_ts, parser.last_ts() + 1); } LOG(INFO) << "Finish ingesting wals up to timestamp: " << parser.last_ts(); return parser.last_ts(); diff --git a/src/transaction/compact_transaction.cc b/src/transaction/compact_transaction.cc index 8e36c8f84..5c3c36931 100644 --- a/src/transaction/compact_transaction.cc +++ b/src/transaction/compact_transaction.cc @@ -33,9 +33,7 @@ CompactTransaction::CompactTransaction(GraphSnapshotStore& snapshot_store, : guard_(snapshot_store), wal_writer_(wal_writer), vm_(vm), - timestamp_(timestamp) { - arc_.Resize(sizeof(WalHeader)); -} + timestamp_(timestamp) {} CompactTransaction::~CompactTransaction() { Abort(); } @@ -43,17 +41,14 @@ timestamp_t CompactTransaction::timestamp() const { return timestamp_; } bool CompactTransaction::Commit() { if (timestamp_ != INVALID_TIMESTAMP) { - auto* header = reinterpret_cast(arc_.GetBuffer()); - header->length = 0; - header->timestamp = timestamp_; - header->type = 1; - - if (!wal_writer_.append(arc_.GetBuffer(), arc_.GetSize())) { + // Compact is an explicit record kind with an empty payload; semantics are + // never inferred from length==0 anymore. + if (!wal_writer_.append_frame(timestamp_, WalRecordKind::kCompact, nullptr, + 0)) { LOG(ERROR) << "Failed to append wal log"; Abort(); return false; } - arc_.Clear(); LOG(INFO) << "before compact - " << timestamp_; { @@ -75,7 +70,6 @@ bool CompactTransaction::Commit() { void CompactTransaction::Abort() { if (timestamp_ != INVALID_TIMESTAMP) { - arc_.Clear(); guard_.release(); vm_.revert_compact_timestamp(timestamp_); timestamp_ = INVALID_TIMESTAMP; diff --git a/src/transaction/insert_transaction.cc b/src/transaction/insert_transaction.cc index 669496edc..740ae8990 100644 --- a/src/transaction/insert_transaction.cc +++ b/src/transaction/insert_transaction.cc @@ -41,9 +41,7 @@ InsertTransaction::InsertTransaction(SnapshotGuard guard, Allocator& alloc, alloc_(alloc), wal_writer_(wal_writer), vm_(vm), - timestamp_(timestamp) { - arc_.Resize(sizeof(WalHeader)); -} + timestamp_(timestamp) {} InsertTransaction::~InsertTransaction() { Abort(); } @@ -156,27 +154,25 @@ bool InsertTransaction::Commit() { if (timestamp_ == INVALID_TIMESTAMP) { return true; } - if (arc_.GetSize() == sizeof(WalHeader)) { + if (arc_.GetSize() == 0) { view_ = nullptr; guard_.release(); vm_.release_insert_timestamp(timestamp_); clear(); return true; } - auto* header = reinterpret_cast(arc_.GetBuffer()); - header->length = arc_.GetSize() - sizeof(WalHeader); - header->type = 0; - header->timestamp = timestamp_; - if (!wal_writer_.append(arc_.GetBuffer(), arc_.GetSize())) { + // One commit produces exactly one kInsert frame; redo serialization order + // inside the payload is unchanged. + if (!wal_writer_.append_frame(timestamp_, WalRecordKind::kInsert, + arc_.GetBuffer(), arc_.GetSize())) { LOG(ERROR) << "Failed to append wal log"; Abort(); return false; } // Apply WAL operations through the writable view. Capacity is assumed // to be sufficient; the strict insert path will throw if exhausted. - IngestWal(*view_, timestamp_, arc_.GetBuffer() + sizeof(WalHeader), - header->length, alloc_); + IngestWal(*view_, timestamp_, arc_.GetBuffer(), arc_.GetSize(), alloc_); view_ = nullptr; guard_.release(); @@ -198,7 +194,8 @@ void InsertTransaction::Abort() { timestamp_t InsertTransaction::timestamp() const { return timestamp_; } void InsertTransaction::IngestWal(GraphView& view, uint32_t timestamp, - char* data, size_t length, Allocator& alloc) { + const char* data, size_t length, + Allocator& alloc) { OutArchive arc; arc.SetSlice(data, length); while (!arc.Empty()) { diff --git a/src/transaction/update_transaction.cc b/src/transaction/update_transaction.cc index 9f12248c4..71ea3afa4 100644 --- a/src/transaction/update_transaction.cc +++ b/src/transaction/update_transaction.cc @@ -341,7 +341,7 @@ bool UpdateTransaction::Commit() { if (timestamp() == INVALID_TIMESTAMP) { return true; } - if (wal_builder_.op_num() == 0 && wal_builder_.content_size() == 0) { + if (wal_builder_.op_num() == 0 && wal_builder_.size() == 0) { release(std::nullopt); return true; } @@ -367,8 +367,19 @@ bool UpdateTransaction::Commit() { auto prepared = std::move(prepared_result).value(); - wal_builder_.finalize(timestamp()); - if (!logger_.append(wal_builder_.data(), wal_builder_.size())) { + // One commit produces exactly one frame; the redo payload keeps WalBuilder's + // serialization order and carries no legacy header. A checkpoint-only + // update transaction has no redo payload and migrates to an explicit + // kCompact frame: the legacy format replayed every empty update record as + // a compaction, and replay keeps that semantics. + WalRecordKind kind = WalRecordKind::kCowUpdate; + const char* payload = wal_builder_.data(); + size_t payload_length = wal_builder_.size(); + if (payload_length == 0) { + kind = WalRecordKind::kCompact; + payload = nullptr; + } + if (!logger_.append_frame(timestamp(), kind, payload, payload_length)) { LOG(ERROR) << "Failed to append wal log"; Abort(); return false; @@ -1000,7 +1011,8 @@ Status StorageTPUpdateInterface::UpdateEdgePropertyImpl( } void UpdateTransaction::IngestWal(PropertyGraph& graph, uint32_t timestamp, - char* data, size_t length, Allocator& alloc) { + const char* data, size_t length, + Allocator& alloc) { OutArchive arc; arc.SetSlice(data, length); while (!arc.Empty()) { diff --git a/src/transaction/wal/dummy_wal_writer.cc b/src/transaction/wal/dummy_wal_writer.cc index 6b075f5f7..dc9c7d651 100644 --- a/src/transaction/wal/dummy_wal_writer.cc +++ b/src/transaction/wal/dummy_wal_writer.cc @@ -19,5 +19,9 @@ namespace neug { std::string DummyWalWriter::type() const { return "dummy"; } void DummyWalWriter::open(const std::string&) {} void DummyWalWriter::close() {} -bool DummyWalWriter::append(const char* data, size_t length) { return true; } +bool DummyWalWriter::append_frame(uint32_t, WalRecordKind, const char*, + size_t) { + ++appended_frame_num_; + return true; +} } // namespace neug diff --git a/src/transaction/wal/local_wal_parser.cc b/src/transaction/wal/local_wal_parser.cc index c48f9fb37..06e9c309c 100644 --- a/src/transaction/wal/local_wal_parser.cc +++ b/src/transaction/wal/local_wal_parser.cc @@ -23,16 +23,228 @@ #include #include #include +#include #include "neug/transaction/wal/wal.h" #include "neug/utils/exception/exception.h" namespace neug { -LocalWalParser::LocalWalParser(const std::string& wal_uri) { +namespace { + +[[noreturn]] void ThrowRecovery(WalRecoveryErrorKind kind, + const std::string& message) { + THROW_WAL_RECOVERY_EXCEPTION(std::string("[") + + WalRecoveryErrorKindName(kind) + "] " + message); +} + +/// RAII guard for one mmapped WAL file. Ownership is transferred to the +/// parser once the file is validated, so replay unit payload views stay +/// valid until the parser is closed. +class MappedWalFile { + public: + MappedWalFile(const std::string& path, size_t size) : size_(size) { + fd_ = ::open(path.c_str(), O_RDONLY); + if (fd_ == -1) { + THROW_IO_EXCEPTION("Failed to open wal file: " + path + ": " + + strerror(errno)); + } + mapped_ = ::mmap(nullptr, size, PROT_READ, MAP_PRIVATE, fd_, 0); + if (mapped_ == MAP_FAILED) { + ::close(fd_); + fd_ = -1; + THROW_IO_EXCEPTION("Failed to mmap wal file: " + path + ": " + + strerror(errno)); + } + path_ = path; + } + + MappedWalFile(const MappedWalFile&) = delete; + MappedWalFile& operator=(const MappedWalFile&) = delete; + + MappedWalFile(MappedWalFile&& other) noexcept + : fd_(other.fd_), + mapped_(other.mapped_), + size_(other.size_), + path_(std::move(other.path_)) { + other.fd_ = -1; + other.mapped_ = nullptr; + other.size_ = 0; + } + + ~MappedWalFile() { + if (mapped_ != nullptr) { + ::munmap(mapped_, size_); + } + if (fd_ != -1) { + ::close(fd_); + } + } + + const std::string& path() const { return path_; } + const uint8_t* data() const { return static_cast(mapped_); } + size_t size() const { return size_; } + + private: + int fd_{-1}; + void* mapped_{nullptr}; + size_t size_; + std::string path_; +}; + +/// Validates one WAL file against the v1 protocol and appends every complete +/// frame to @p units as zero-copy views into the mapping, which is kept +/// alive by @p mapped_files. Only the last candidate frame may be skipped as +/// crash residue when EOF truncates it; any other inconsistency throws a +/// typed recovery error. +void ValidateAndCollect( + const std::string& path, std::vector& units, + std::vector>& mapped_files) { + const size_t file_size = std::filesystem::file_size(path); + if (file_size == 0) { + return; + } + MappedWalFile file(path, file_size); + const uint8_t* data = file.data(); + // A committed frame occupies at least a header; this bounds the per-file + // unit growth. + units.reserve(units.size() + file_size / kWalFrameHeaderSize); + + // ---- File header ----------------------------------------------------- + if (file_size < kWalFileHeaderSize) { + ThrowRecovery( + WalRecoveryErrorKind::kUnsupportedFormat, + "wal file " + path + " has " + std::to_string(file_size) + + " bytes, fewer than the " + std::to_string(kWalFileHeaderSize) + + "-byte v1 file header; legacy or torn formats are not parsed"); + } + WalFileHeader file_header; + size_t consumed = 0; + const auto header_status = + DecodeWalFileHeader(data, file_size, file_header, consumed); + switch (header_status) { + case WalDecodeStatus::kOk: + break; + case WalDecodeStatus::kBadMagic: + ThrowRecovery( + WalRecoveryErrorKind::kUnsupportedFormat, + "wal file " + path + + " does not carry the v1 file magic; the legacy pre-v1 format " + "is rejected instead of being silently parsed. Complete a " + "checkpoint on the old binary so the wal directory is empty " + "before upgrading."); + case WalDecodeStatus::kBadVersion: + ThrowRecovery(WalRecoveryErrorKind::kUnsupportedFormat, + "wal file " + path + " has unsupported format version " + + std::to_string(file_header.format_version) + + ", expected " + std::to_string(kWalFormatVersion)); + default: + ThrowRecovery(WalRecoveryErrorKind::kCorruptedFrame, + "wal file " + path + " header is corrupted at offset 0: " + + WalDecodeStatusName(header_status)); + } + + // ---- Frames ------------------------------------------------------------ + size_t offset = kWalFileHeaderSize; + while (offset < file_size) { + const size_t remaining = file_size - offset; + if (remaining < kWalFrameHeaderSize) { + // Torn header write of the last frame: crash residue at EOF. + break; + } + WalFrameHeader frame_header; + size_t header_consumed = 0; + const auto frame_status = DecodeWalFrameHeader( + data + offset, remaining, frame_header, header_consumed); + if (frame_status == WalDecodeStatus::kTruncated) { + break; // unreachable given the remaining check above + } + if (frame_status != WalDecodeStatus::kOk) { + const WalRecoveryErrorKind kind = + frame_status == WalDecodeStatus::kUnknownRecordKind + ? WalRecoveryErrorKind::kUnknownRecordKind + : WalRecoveryErrorKind::kCorruptedFrame; + ThrowRecovery(kind, "wal file " + path + " frame at offset " + + std::to_string(offset) + + " failed header validation: " + + WalDecodeStatusName(frame_status)); + } + + const uint64_t frame_total = + kWalFrameHeaderSize + frame_header.payload_length; + if (frame_total > remaining) { + // The last candidate frame is truncated by EOF: header or payload + // never completed. This is crash residue, not corruption; the frame + // cannot pass its checksum and is not replayed. + break; + } + + const uint8_t* header_bytes = data + offset; + const uint8_t* payload = header_bytes + kWalFrameHeaderSize; + + const auto validate_status = + ValidateWalFrame(frame_header, header_bytes, payload); + if (validate_status != WalDecodeStatus::kOk) { + ThrowRecovery( + WalRecoveryErrorKind::kCorruptedFrame, + "wal file " + path + " frame at offset " + std::to_string(offset) + + " failed checksum validation: " + + WalDecodeStatusName(validate_status) + ", commit_timestamp=" + + std::to_string(frame_header.commit_timestamp)); + } + + // Kind/payload constraints are enforced uniformly here, not inferred + // from length or flags at the call sites. + if (frame_header.record_kind == WalRecordKind::kCompact && + frame_header.payload_length != 0) { + ThrowRecovery(WalRecoveryErrorKind::kCorruptedFrame, + "wal file " + path + " frame at offset " + + std::to_string(offset) + + ": kCompact frame must carry an empty payload, got " + + std::to_string(frame_header.payload_length) + " bytes"); + } + if (frame_header.record_kind != WalRecordKind::kCompact && + frame_header.payload_length == 0) { + ThrowRecovery(WalRecoveryErrorKind::kCorruptedFrame, + "wal file " + path + " frame at offset " + + std::to_string(offset) + + ": empty transactions never write frames, got an " + "empty-payload " + + WalRecordKindName(frame_header.record_kind) + " frame"); + } + + WalReplayUnit unit; + unit.commit_timestamp = frame_header.commit_timestamp; + unit.kind = frame_header.record_kind; + unit.payload = + std::string_view(reinterpret_cast(payload), + static_cast(frame_header.payload_length)); + unit.file_index = static_cast(mapped_files.size()); + unit.source_offset = offset; + units.push_back(std::move(unit)); + + offset += static_cast(frame_total); + } + + // The file validated; keep it mapped so the payload views above outlive + // this scope. + mapped_files.push_back(std::make_unique(std::move(file))); +} + +} // namespace + +/// Mapping table kept by the parser so replay unit payload views stay valid. +struct LocalWalParser::MappedFiles { + std::vector> files; +}; + +LocalWalParser::LocalWalParser(const std::string& wal_uri) + : mapped_files_(std::make_unique()) { LocalWalParser::open(wal_uri); } +LocalWalParser::~LocalWalParser() { close(); } + void LocalWalParser::open(const std::string& wal_uri) { close(); auto wal_dir = get_wal_uri_path(wal_uri); @@ -40,96 +252,58 @@ void LocalWalParser::open(const std::string& wal_uri) { std::filesystem::create_directory(wal_dir); } + // Collect regular files first; replay order is decided by commit + // timestamp, never by directory enumeration order. Sorting only makes + // error reporting deterministic. std::vector paths; for (const auto& entry : std::filesystem::directory_iterator(wal_dir)) { - paths.push_back(entry.path().string()); - } - for (auto path : paths) { - size_t file_size = std::filesystem::file_size(path); - if (file_size == 0) { - continue; - } - int fd = ::open(path.c_str(), O_RDONLY); - if (fd == -1) { - close(); - THROW_IO_EXCEPTION("Failed to open wal file: " + path + ": " + - strerror(errno)); - } - void* mmapped_buffer = - ::mmap(NULL, file_size, PROT_READ, MAP_PRIVATE, fd, 0); - if (mmapped_buffer == MAP_FAILED) { - ::close(fd); - close(); - THROW_IO_EXCEPTION("Failed to mmap wal file: " + path + ": " + - strerror(errno)); + if (entry.is_regular_file()) { + paths.push_back(entry.path().string()); } + } + std::sort(paths.begin(), paths.end()); - fds_.push_back(fd); - mmapped_ptrs_.push_back(mmapped_buffer); - mmapped_size_.push_back(file_size); + std::vector units; + for (const auto& path : paths) { + ValidateAndCollect(path, units, mapped_files_->files); } - insert_wal_list_.resize(4096); - for (size_t i = 0; i < mmapped_ptrs_.size(); ++i) { - char* ptr = static_cast(mmapped_ptrs_[i]); - while (true) { - const WalHeader* header = reinterpret_cast(ptr); - ptr += sizeof(WalHeader); - uint32_t ts = header->timestamp; - if (ts == 0) { - break; - } - int length = header->length; - if (header->type) { - UpdateWalUnit unit; - unit.timestamp = ts; - unit.ptr = ptr; - unit.size = length; - update_wal_list_.push_back(unit); - } else { - if (ts >= insert_wal_list_.size()) { - insert_wal_list_.resize(ts + 1); - } - insert_wal_list_[ts].ptr = ptr; - insert_wal_list_[ts].size = length; - } - ptr += length; - last_ts_ = std::max(ts, last_ts_); + // All files are validated before anything is merged. Order strictly by + // commit timestamp; gaps are allowed, duplicates are always rejected. + std::sort(units.begin(), units.end(), + [](const WalReplayUnit& lhs, const WalReplayUnit& rhs) { + return lhs.commit_timestamp < rhs.commit_timestamp; + }); + for (size_t i = 1; i < units.size(); ++i) { + if (units[i].commit_timestamp == units[i - 1].commit_timestamp) { + ThrowRecovery( + WalRecoveryErrorKind::kDuplicateTimestamp, + "duplicate commit timestamp " + + std::to_string(units[i].commit_timestamp) + " found in " + + mapped_files_->files[units[i - 1].file_index]->path() + + " (offset " + std::to_string(units[i - 1].source_offset) + + ") and " + mapped_files_->files[units[i].file_index]->path() + + " (offset " + std::to_string(units[i].source_offset) + ")"); } } - if (!update_wal_list_.empty()) { - std::sort(update_wal_list_.begin(), update_wal_list_.end(), - [](const UpdateWalUnit& lhs, const UpdateWalUnit& rhs) { - return lhs.timestamp < rhs.timestamp; - }); + if (!units.empty()) { + last_ts_ = units.back().commit_timestamp; } + replay_units_ = std::move(units); } void LocalWalParser::close() { - insert_wal_list_.clear(); - size_t ptr_num = mmapped_ptrs_.size(); - for (size_t i = 0; i < ptr_num; ++i) { - munmap(mmapped_ptrs_[i], mmapped_size_[i]); - } - for (auto fd : fds_) { - ::close(fd); - } - fds_.clear(); - mmapped_ptrs_.clear(); - mmapped_size_.clear(); - update_wal_list_.clear(); + // Release the units first: their payload views reference the mappings. + replay_units_.clear(); + mapped_files_->files.clear(); last_ts_ = 0; } uint32_t LocalWalParser::last_ts() const { return last_ts_; } -const WalContentUnit& LocalWalParser::get_insert_wal(uint32_t ts) const { - return insert_wal_list_[ts]; -} - -const std::vector& LocalWalParser::get_update_wals() const { - return update_wal_list_; +const std::vector& LocalWalParser::replay_units() const { + return replay_units_; } const bool LocalWalParser::registered_ = WalParserFactory::RegisterWalParser( diff --git a/src/transaction/wal/local_wal_writer.cc b/src/transaction/wal/local_wal_writer.cc index 166c551d4..296f6b37e 100644 --- a/src/transaction/wal/local_wal_writer.cc +++ b/src/transaction/wal/local_wal_writer.cc @@ -49,10 +49,11 @@ LocalWalWriter::~LocalWalWriter() noexcept { void LocalWalWriter::open(const std::string& wal_uri) { close(); wal_uri_ = wal_uri; - auto prefix = get_wal_uri_path(wal_uri_); + auto prefix = get_wal_uri_path(wal_uri); if (!std::filesystem::exists(prefix)) { std::filesystem::create_directories(prefix); } + path_.clear(); const int max_version = 65536; for (int version = 0; version != max_version; ++version) { // Keep the historical on-disk prefix for WAL replay compatibility. The @@ -63,6 +64,7 @@ void LocalWalWriter::open(const std::string& wal_uri) { if (std::filesystem::exists(path)) { continue; } + path_ = path; fd_ = ::open(path.c_str(), O_RDWR | O_CREAT | O_TRUNC, 0644); break; } @@ -70,12 +72,20 @@ void LocalWalWriter::open(const std::string& wal_uri) { THROW_IO_EXCEPTION("Failed to open wal file " + std::string(strerror(errno))); } - if (ftruncate(fd_, TRUNC_SIZE) != 0) { - THROW_IO_EXCEPTION("Failed to truncate wal file " + - std::string(strerror(errno))); + + // Persist the v1 file header before any frame is accepted. The file + // belongs to the wal_dir() it is created in; checkpoint ownership is + // guaranteed by checkpoint rotation, and the writer identity is already + // carried by the file name. + const WalFileHeader header; + const auto encoded = EncodeWalFileHeader(header); + if (!write_all(reinterpret_cast(encoded.data()), encoded.size(), + FailNextWrite::kHeader)) { + close(); + THROW_IO_EXCEPTION("Injected write failure while writing wal file header"); } - file_size_ = TRUNC_SIZE; - file_used_ = 0; + sync_file(); + append_offset_ = encoded.size(); } void LocalWalWriter::close() { @@ -85,54 +95,121 @@ void LocalWalWriter::close() { // and reused by another thread. const int fd = fd_; fd_ = -1; - file_size_ = 0; - file_used_ = 0; + path_.clear(); + append_offset_ = 0; + failed_ = false; if (::close(fd) != 0) { THROW_IO_EXCEPTION("Failed to close file" + std::string(strerror(errno))); } } } -bool LocalWalWriter::append(const char* data, size_t length) { - if (NEUG_UNLIKELY(fd_ == -1)) { +bool LocalWalWriter::restore_clean_eof(size_t offset) { + if (fd_ == -1) { + failed_ = true; return false; } - size_t expected_size = file_used_ + length; - if (expected_size > file_size_) { - size_t new_file_size = (expected_size / TRUNC_SIZE + 1) * TRUNC_SIZE; - if (ftruncate(fd_, new_file_size) != 0) { - THROW_IO_EXCEPTION("Failed to truncate wal file " + - std::string(strerror(errno))); - } - file_size_ = new_file_size; + if (::ftruncate(fd_, static_cast(offset)) != 0 || + ::lseek(fd_, static_cast(offset), SEEK_SET) == + static_cast(-1)) { + LOG(ERROR) << "Failed to restore wal file " << path_ << " to offset " + << offset << ": " << strerror(errno); + failed_ = true; + return false; } + append_offset_ = offset; + return true; +} - file_used_ += length; - - if (static_cast(write(fd_, data, length)) != length) { - THROW_IO_EXCEPTION("Failed to write wal file " + - std::string(strerror(errno))); +bool LocalWalWriter::write_all(const char* buffer, size_t length, + FailNextWrite phase) { + if (fail_next_write_ != FailNextWrite::kNone && fail_next_write_ == phase) { + fail_next_write_ = FailNextWrite::kNone; + return false; } + size_t written = 0; + while (written < length) { + const ssize_t ret = ::write(fd_, buffer + written, length - written); + if (ret < 0) { + if (errno == EINTR) { + continue; + } + THROW_IO_EXCEPTION("Failed to write wal file " + path_ + ": " + + std::string(strerror(errno))); + } + written += static_cast(ret); + } + append_offset_ += length; + return true; +} -#if 1 +void LocalWalWriter::sync_file() { + // Keep the current synchronous durability strategy as the transitional + // implementation: one full sync per frame. #ifdef F_FULLFSYNC if (fcntl(fd_, F_FULLFSYNC) != 0) { -#ifdef __APPLE__ - THROW_IO_EXCEPTION("Failed to fcntl sync wal file " + + THROW_IO_EXCEPTION("Failed to fcntl sync wal file " + path_ + ": " + std::string(strerror(errno))); -#else - THROW_IO_EXCEPTION("Failed to fcntl sync wal file " + - std::string(strerrno(errno))); -#endif } #else - // if (fsync(fd_) != 0) { if (fdatasync(fd_) != 0) { - THROW_IO_EXCEPTION("Failed to fsync wal file " + + THROW_IO_EXCEPTION("Failed to fsync wal file " + path_ + ": " + std::string(strerror(errno))); } #endif -#endif +} + +bool LocalWalWriter::append_frame(uint32_t commit_timestamp, WalRecordKind kind, + const char* payload, size_t length) { + if (NEUG_UNLIKELY(fd_ == -1 || failed_)) { + return false; + } + if (length > kWalMaxPayloadLength) { + THROW_INVALID_ARGUMENT_EXCEPTION("WAL frame payload too large: " + + std::to_string(length)); + } + + const auto* payload_bytes = reinterpret_cast(payload); + // Clean EOF before this attempt; a failed frame is rolled back to here so + // its residue can never be buried mid-file by a later successful append. + const size_t frame_start = append_offset_; + const auto rollback_failed_frame = [&](const char* phase) { + LOG(ERROR) << "Injected write failure at " << phase + << ", wal file: " << path_; + restore_clean_eof(frame_start); + return false; + }; + + // 1) Frame header. The checksum is computed over the payload before any + // byte is written, so a persisted header always describes the frame that + // follows it; recovery decides completeness purely from the checksum. + WalFrameHeader header; + header.record_kind = kind; + header.commit_timestamp = commit_timestamp; + header.payload_length = static_cast(length); + const auto encoded_header = + EncodeWalFrameHeader(header, payload_bytes, length); + + try { + // 2) Header, then payload. A crash between or during the two writes + // leaves a short frame at EOF, which recovery discards as torn residue. + if (!write_all(reinterpret_cast(encoded_header.data()), + encoded_header.size(), FailNextWrite::kHeader)) { + return rollback_failed_frame("frame header"); + } + if (length > 0 && !write_all(payload, length, FailNextWrite::kPayload)) { + return rollback_failed_frame("frame payload"); + } + } catch (...) { + // Real I/O error: roll back the partial frame before rethrowing so the + // file keeps a clean logical EOF. + restore_clean_eof(frame_start); + throw; + } + + // The frame is persisted; post-write sync failures belong to the commit + // durability decision and must not roll the frame back. + sync_file(); return true; } diff --git a/src/transaction/wal/wal_builder.cc b/src/transaction/wal/wal_builder.cc index 66573decd..08477d85c 100644 --- a/src/transaction/wal/wal_builder.cc +++ b/src/transaction/wal/wal_builder.cc @@ -19,14 +19,7 @@ namespace neug { -WalBuilder::WalBuilder() { arc_.Resize(sizeof(WalHeader)); } - -void WalBuilder::finalize(timestamp_t timestamp) { - auto* header = reinterpret_cast(arc_.GetBuffer()); - header->length = arc_.GetSize() - sizeof(WalHeader); - header->type = 1; - header->timestamp = timestamp; -} +WalBuilder::WalBuilder() = default; void WalBuilder::clear() { arc_.Clear(); diff --git a/src/transaction/wal/wal_codec.cc b/src/transaction/wal/wal_codec.cc new file mode 100644 index 000000000..6cdcbba7e --- /dev/null +++ b/src/transaction/wal/wal_codec.cc @@ -0,0 +1,204 @@ +/** Copyright 2020 Alibaba Group Holding Limited. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "neug/transaction/wal/wal_codec.h" + +#include + +#include "absl/crc/crc32c.h" + +namespace neug { + +namespace { + +// Little-endian primitives are serialized byte-by-byte; the codec never +// depends on host endianness or struct layout. +void PutU32(uint8_t* dst, uint32_t value) { + for (int i = 0; i < 4; ++i) { + dst[i] = static_cast(value >> (8 * i)); + } +} + +uint32_t GetU32(const uint8_t* src) { + uint32_t value = 0; + for (int i = 0; i < 4; ++i) { + value |= static_cast(src[i]) << (8 * i); + } + return value; +} + +bool IsKnownRecordKind(uint32_t value) { + return value == static_cast(WalRecordKind::kInsert) || + value == static_cast(WalRecordKind::kCowUpdate) || + value == static_cast(WalRecordKind::kCompact); +} + +/// Encodes the frame header bytes preceding the checksum field: kind, +/// payload length and commit timestamp. These are the header bytes covered +/// by frame_checksum. +std::array EncodeWalFrameHeaderPrefix( + const WalFrameHeader& header) { + std::array bytes{}; + bytes[0] = static_cast(header.record_kind); + PutU32(bytes.data() + 1, header.payload_length); + PutU32(bytes.data() + 5, header.commit_timestamp); + return bytes; +} + +/// Single source of truth for what the frame checksum covers: the encoded +/// frame header bytes preceding the checksum field plus the payload. This +/// protects the kind, timestamp and length fields, not only the payload. +/// CRC32C is hardware-accelerated on x86 (SSE4.2) and ARM64 via absl. +uint32_t WalFrameChecksum(const uint8_t* frame_header_bytes, + const uint8_t* payload, size_t payload_length) { + absl::crc32c_t crc = absl::ExtendCrc32c( + absl::crc32c_t{0}, + absl::string_view(reinterpret_cast(frame_header_bytes), + kWalFrameHeaderStableSize)); + if (payload_length > 0) { + crc = absl::ExtendCrc32c( + crc, absl::string_view(reinterpret_cast(payload), + payload_length)); + } + return static_cast(crc); +} + +} // namespace + +std::string WalRecordKindName(WalRecordKind kind) { + switch (kind) { + case WalRecordKind::kInsert: + return "kInsert"; + case WalRecordKind::kCowUpdate: + return "kCowUpdate"; + case WalRecordKind::kCompact: + return "kCompact"; + } + return "kUnknown(" + std::to_string(static_cast(kind)) + ")"; +} + +std::string WalRecoveryErrorKindName(WalRecoveryErrorKind kind) { + switch (kind) { + case WalRecoveryErrorKind::kUnsupportedFormat: + return "unsupported_format"; + case WalRecoveryErrorKind::kCorruptedFrame: + return "corrupted_frame"; + case WalRecoveryErrorKind::kDuplicateTimestamp: + return "duplicate_timestamp"; + case WalRecoveryErrorKind::kUnknownRecordKind: + return "unknown_record_kind"; + } + return "unknown"; +} + +std::string WalDecodeStatusName(WalDecodeStatus status) { + switch (status) { + case WalDecodeStatus::kOk: + return "ok"; + case WalDecodeStatus::kTruncated: + return "truncated"; + case WalDecodeStatus::kBadMagic: + return "bad_magic"; + case WalDecodeStatus::kBadVersion: + return "bad_version"; + case WalDecodeStatus::kBadHeaderSize: + return "bad_header_size"; + case WalDecodeStatus::kBadChecksum: + return "bad_checksum"; + case WalDecodeStatus::kUnknownRecordKind: + return "unknown_record_kind"; + case WalDecodeStatus::kPayloadTooLarge: + return "payload_too_large"; + } + return "unknown"; +} + +std::array EncodeWalFileHeader( + const WalFileHeader& header) { + std::array bytes{}; + PutU32(bytes.data() + 0, header.magic); + PutU32(bytes.data() + 4, header.format_version); + PutU32(bytes.data() + 8, header.header_size); + PutU32(bytes.data() + 12, header.reserved); + return bytes; +} + +std::array EncodeWalFrameHeader( + const WalFrameHeader& header, const uint8_t* payload, + size_t payload_length) { + std::array bytes{}; + const auto prefix = EncodeWalFrameHeaderPrefix(header); + std::memcpy(bytes.data(), prefix.data(), prefix.size()); + PutU32(bytes.data() + prefix.size(), + WalFrameChecksum(prefix.data(), payload, payload_length)); + return bytes; +} + +WalDecodeStatus DecodeWalFileHeader(const uint8_t* data, size_t remaining, + WalFileHeader& out, size_t& consumed) { + consumed = 0; + if (remaining < kWalFileHeaderSize) { + return WalDecodeStatus::kTruncated; + } + out.magic = GetU32(data + 0); + if (out.magic != kWalFileMagic) { + return WalDecodeStatus::kBadMagic; + } + out.format_version = GetU32(data + 4); + if (out.format_version != kWalFormatVersion) { + return WalDecodeStatus::kBadVersion; + } + out.header_size = GetU32(data + 8); + if (out.header_size != kWalFileHeaderSize) { + return WalDecodeStatus::kBadHeaderSize; + } + out.reserved = GetU32(data + 12); + consumed = kWalFileHeaderSize; + return WalDecodeStatus::kOk; +} + +WalDecodeStatus DecodeWalFrameHeader(const uint8_t* data, size_t remaining, + WalFrameHeader& out, size_t& consumed) { + consumed = 0; + if (remaining < kWalFrameHeaderSize) { + return WalDecodeStatus::kTruncated; + } + const uint32_t kind = data[0]; + if (!IsKnownRecordKind(kind)) { + return WalDecodeStatus::kUnknownRecordKind; + } + out.record_kind = static_cast(kind); + out.payload_length = GetU32(data + 1); + if (out.payload_length > kWalMaxPayloadLength) { + return WalDecodeStatus::kPayloadTooLarge; + } + out.commit_timestamp = GetU32(data + 5); + out.frame_checksum = GetU32(data + 9); + consumed = kWalFrameHeaderSize; + return WalDecodeStatus::kOk; +} + +WalDecodeStatus ValidateWalFrame(const WalFrameHeader& header, + const uint8_t* frame_header_bytes, + const uint8_t* payload) { + if (header.frame_checksum != + WalFrameChecksum(frame_header_bytes, payload, + static_cast(header.payload_length))) { + return WalDecodeStatus::kBadChecksum; + } + return WalDecodeStatus::kOk; +} + +} // namespace neug diff --git a/src/utils/exception/exception.cc b/src/utils/exception/exception.cc index 84583da22..697e5de78 100644 --- a/src/utils/exception/exception.cc +++ b/src/utils/exception/exception.cc @@ -252,5 +252,13 @@ TxStateConflictException::TxStateConflictException(const std::string& msg, : Exception("Transaction state conflict: " + msg, file_line, neug::StatusCode::ERR_TX_STATE_CONFLICT) {} +WalRecoveryException::WalRecoveryException(const std::string& msg) + : Exception("WAL recovery error: " + msg, + neug::StatusCode::ERR_CORRUPTION_DETECTED) {} +WalRecoveryException::WalRecoveryException(const std::string& msg, + const std::string& file_line) + : Exception("WAL recovery error: " + msg, file_line, + neug::StatusCode::ERR_CORRUPTION_DETECTED) {} + } // namespace exception } // namespace neug diff --git a/tests/storage/test_tp_index.cc b/tests/storage/test_tp_index.cc index d2748bbe3..72113f5bc 100644 --- a/tests/storage/test_tp_index.cc +++ b/tests/storage/test_tp_index.cc @@ -48,16 +48,26 @@ namespace { class CapturingWalWriter : public IWalWriter { public: + struct Frame { + uint32_t commit_timestamp{0}; + WalRecordKind kind{WalRecordKind::kInsert}; + std::vector payload; + }; + std::string type() const override { return "capturing"; } void open(const std::string&) override {} void close() override {} - bool append(const char* data, size_t length) override { - records.emplace_back(data, data + length); + bool append_frame(uint32_t commit_timestamp, WalRecordKind kind, + const char* payload, size_t length) override { + frames.emplace_back(); + frames.back().commit_timestamp = commit_timestamp; + frames.back().kind = kind; + frames.back().payload.assign(payload, payload + length); return true; } - std::vector> records; + std::vector frames; }; class StubPlanner : public IGraphPlanner { @@ -112,7 +122,7 @@ class TPIndexTest : public ::testing::Test { ap_ = std::make_unique(*graph_, *view_, 0, allocator_); version_manager_.init_ts({0, 0}, 1); - wal_writer_.records.clear(); + wal_writer_.frames.clear(); auto global_cache = std::make_shared( std::make_shared()); } @@ -720,29 +730,26 @@ TEST_F(TPIndexTest, WalReplayRestoresIndexData) { AddPersonTP(tp, 3, "Charlie", 30); Commit(txn); } - ASSERT_EQ(wal_writer_.records.size(), 1); - const auto& wal = wal_writer_.records.back(); - ASSERT_GT(wal.size(), sizeof(WalHeader)); - const auto* header = reinterpret_cast(wal.data()); - ASSERT_EQ(static_cast(header->length), - wal.size() - sizeof(WalHeader)); + ASSERT_EQ(wal_writer_.frames.size(), 1); + const auto& wal = wal_writer_.frames.back(); + ASSERT_EQ(wal.kind, WalRecordKind::kCowUpdate); + ASSERT_FALSE(wal.payload.empty()); { GraphView before_replay_view(*replay_graph); StorageReadInterface before_replay_reader(before_replay_view, - header->timestamp); + wal.commit_timestamp); EXPECT_EQ(SearchPersonNames(before_replay_reader, 30), (std::vector{})); EXPECT_EQ(SearchPersonNames(before_replay_reader, 25), (std::vector{})); } - UpdateTransaction::IngestWal( - *replay_graph, header->timestamp, - const_cast(wal.data() + sizeof(WalHeader)), header->length, - allocator_); + UpdateTransaction::IngestWal(*replay_graph, wal.commit_timestamp, + wal.payload.data(), wal.payload.size(), + allocator_); GraphView replay_view(*replay_graph); - StorageReadInterface replay_reader(replay_view, header->timestamp); + StorageReadInterface replay_reader(replay_view, wal.commit_timestamp); EXPECT_EQ(SearchPersonNames(replay_reader, 30), (std::vector{"Alice", "Charlie"})); diff --git a/tests/transaction/CMakeLists.txt b/tests/transaction/CMakeLists.txt index ee41a776c..871243f53 100644 --- a/tests/transaction/CMakeLists.txt +++ b/tests/transaction/CMakeLists.txt @@ -1,5 +1,6 @@ add_neug_test(runtime_wait_test test_runtime_wait.cc) add_neug_test(read_view_publication_test test_read_view_publication.cc) +add_neug_test(wal_codec_test test_wal_codec.cc) if (BUILD_HTTP_SERVER) add_neug_test( diff --git a/tests/transaction/test_insert_transaction.cc b/tests/transaction/test_insert_transaction.cc index 613be174a..674edd29e 100644 --- a/tests/transaction/test_insert_transaction.cc +++ b/tests/transaction/test_insert_transaction.cc @@ -238,26 +238,25 @@ class LocalWalParserTest : public ::testing::Test { } } - // Write a single WAL entry (header + payload) into a buffer. - void AppendWalEntry(std::vector& buf, uint32_t ts, uint8_t type, - const std::string& payload) { - neug::WalHeader header; - header.timestamp = ts; - header.type = type; - header.length = static_cast(payload.size()); - const char* hdr = reinterpret_cast(&header); - buf.insert(buf.end(), hdr, hdr + sizeof(neug::WalHeader)); + // Write a complete v1 WAL file (file header + one committed frame) using + // the same codec the writer uses. + void WriteWalFrameFile(const std::string& filename, uint32_t ts, + neug::WalRecordKind kind, const std::string& payload) { + neug::WalFileHeader fh; + auto fhb = neug::EncodeWalFileHeader(fh); + + neug::WalFrameHeader fhdr; + fhdr.record_kind = kind; + fhdr.commit_timestamp = ts; + fhdr.payload_length = static_cast(payload.size()); + auto fhdrb = neug::EncodeWalFrameHeader( + fhdr, reinterpret_cast(payload.data()), payload.size()); + + std::vector buf; + buf.insert(buf.end(), fhb.begin(), fhb.end()); + buf.insert(buf.end(), fhdrb.begin(), fhdrb.end()); buf.insert(buf.end(), payload.begin(), payload.end()); - } - - // Append a terminator entry (timestamp=0) to mark end of WAL stream. - void AppendWalTerminator(std::vector& buf) { - neug::WalHeader terminator; - terminator.timestamp = 0; - terminator.type = 0; - terminator.length = 0; - const char* hdr = reinterpret_cast(&terminator); - buf.insert(buf.end(), hdr, hdr + sizeof(neug::WalHeader)); + WriteWalFile(filename, buf); } // Write buffer contents to a .wal file in the WAL directory. @@ -274,19 +273,17 @@ class LocalWalParserTest : public ::testing::Test { // Test: LocalWalParser can correctly parse a valid WAL file. TEST_F(LocalWalParserTest, OpenAndParseValidWalFile) { - std::vector buf; std::string payload = "insert_vertex_data"; - AppendWalEntry(buf, /*ts=*/1, /*type=*/0, payload); // insert WAL - AppendWalTerminator(buf); - WriteWalFile("thread_0_0.wal", buf); + WriteWalFrameFile("thread_0_0.wal", /*ts=*/1, neug::WalRecordKind::kInsert, + payload); neug::LocalWalParser parser(wal_dir_); - EXPECT_EQ(parser.last_ts(), 1); - const auto& unit = parser.get_insert_wal(1); - EXPECT_EQ(unit.size, payload.size()); - // The ptr should point to the payload data within the mmap region. - EXPECT_NE(unit.ptr, nullptr); - EXPECT_EQ(std::string(unit.ptr, unit.size), payload); + EXPECT_EQ(parser.last_ts(), 1u); + ASSERT_EQ(parser.replay_units().size(), 1u); + const auto& unit = parser.replay_units()[0]; + EXPECT_EQ(unit.commit_timestamp, 1u); + EXPECT_EQ(unit.kind, neug::WalRecordKind::kInsert); + EXPECT_EQ(unit.payload, payload); } // Test: LocalWalParser throws IOException when ::open() on a WAL file fails @@ -378,5 +375,5 @@ TEST_F(LocalWalParserTest, MmapFailureThrowsIOException) { // Test: Opening an empty WAL directory does not throw and last_ts is 0. TEST_F(LocalWalParserTest, OpenEmptyWalDirNoThrow) { neug::LocalWalParser parser(wal_dir_); - EXPECT_EQ(parser.last_ts(), 0); + EXPECT_EQ(parser.last_ts(), 0u); } diff --git a/tests/transaction/test_wal_codec.cc b/tests/transaction/test_wal_codec.cc new file mode 100644 index 000000000..7f2114e06 --- /dev/null +++ b/tests/transaction/test_wal_codec.cc @@ -0,0 +1,579 @@ +/** Copyright 2020 Alibaba Group Holding Limited. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "neug/transaction/wal/wal_codec.h" + +#include +#include +#include +#include +#include +#include +#include +#include + +#include "gtest/gtest.h" +#include "neug/transaction/wal/local_wal_parser.h" +#include "neug/transaction/wal/local_wal_writer.h" +#include "neug/transaction/wal/wal.h" +#include "neug/utils/exception/exception.h" + +namespace neug { +namespace { + +// --------------------------------------------------------------------------- +// Codec table-driven tests +// --------------------------------------------------------------------------- + +struct WalFileFixture { + WalFileHeader file_header; + std::array file_header_bytes; +}; + +WalFileFixture MakeFileHeader() { + WalFileFixture fixture; + fixture.file_header_bytes = EncodeWalFileHeader(fixture.file_header); + return fixture; +} + +/// Encodes one committed frame (header + payload) into bytes. +std::vector EncodeFrame(uint32_t ts, WalRecordKind kind, + const std::vector& payload) { + WalFrameHeader header; + header.record_kind = kind; + header.commit_timestamp = ts; + header.payload_length = static_cast(payload.size()); + auto header_bytes = + EncodeWalFrameHeader(header, payload.data(), payload.size()); + + std::vector bytes; + bytes.insert(bytes.end(), header_bytes.begin(), header_bytes.end()); + bytes.insert(bytes.end(), payload.begin(), payload.end()); + return bytes; +} + +TEST(WalCodecTest, FrameRoundTripAllKinds) { + const std::vector payload = {0xDE, 0xAD, 0xBE, 0xEF}; + const std::vector kinds = {WalRecordKind::kInsert, + WalRecordKind::kCowUpdate, + WalRecordKind::kCompact}; + for (const auto kind : kinds) { + // kCompact frames carry no payload by protocol; encode empty for it. + const auto& body = + kind == WalRecordKind::kCompact ? std::vector{} : payload; + const auto frame = EncodeFrame(/*ts=*/42, kind, body); + + WalFrameHeader header; + size_t consumed = 0; + ASSERT_EQ( + DecodeWalFrameHeader(frame.data(), frame.size(), header, consumed), + WalDecodeStatus::kOk) + << "kind=" << WalRecordKindName(kind); + EXPECT_EQ(consumed, kWalFrameHeaderSize); + EXPECT_EQ(header.record_kind, kind); + EXPECT_EQ(header.commit_timestamp, 42u); + EXPECT_EQ(header.payload_length, body.size()); + + EXPECT_EQ(ValidateWalFrame(header, frame.data(), + frame.data() + kWalFrameHeaderSize), + WalDecodeStatus::kOk); + } +} + +TEST(WalCodecTest, FileHeaderRoundTrip) { + WalFileHeader header; + header.reserved = 3; + auto bytes = EncodeWalFileHeader(header); + + WalFileHeader decoded; + size_t consumed = 0; + ASSERT_EQ(DecodeWalFileHeader(bytes.data(), bytes.size(), decoded, consumed), + WalDecodeStatus::kOk); + EXPECT_EQ(consumed, kWalFileHeaderSize); + EXPECT_EQ(decoded.reserved, 3u); +} + +TEST(WalCodecTest, FileHeaderDecodeRejectsBadMagicVersionSizeTruncation) { + WalFileHeader header; + WalFileHeader decoded; + size_t consumed = 0; + + auto bytes = EncodeWalFileHeader(header); + EXPECT_EQ(DecodeWalFileHeader(bytes.data(), kWalFileHeaderSize - 1, decoded, + consumed), + WalDecodeStatus::kTruncated); + + header.magic = 0x12345678; + bytes = EncodeWalFileHeader(header); + EXPECT_EQ(DecodeWalFileHeader(bytes.data(), bytes.size(), decoded, consumed), + WalDecodeStatus::kBadMagic); + + header = WalFileHeader{}; + header.format_version = 2; + bytes = EncodeWalFileHeader(header); + EXPECT_EQ(DecodeWalFileHeader(bytes.data(), bytes.size(), decoded, consumed), + WalDecodeStatus::kBadVersion); + + header = WalFileHeader{}; + header.header_size = kWalFileHeaderSize + 4; + bytes = EncodeWalFileHeader(header); + EXPECT_EQ(DecodeWalFileHeader(bytes.data(), bytes.size(), decoded, consumed), + WalDecodeStatus::kBadHeaderSize); +} + +TEST(WalCodecTest, FrameHeaderDecodeRejectsTruncationAndUnknownKind) { + const std::vector payload = {1, 2, 3}; + auto frame = EncodeFrame(1, WalRecordKind::kInsert, payload); + WalFrameHeader decoded; + size_t consumed = 0; + + EXPECT_EQ(DecodeWalFrameHeader(frame.data(), kWalFrameHeaderSize - 1, decoded, + consumed), + WalDecodeStatus::kTruncated); + + // Unknown record kind: the kind is the first header byte; the kind check + // runs before checksum validation at decode time. + auto corrupted = frame; + corrupted[0] = 9; + EXPECT_EQ(DecodeWalFrameHeader(corrupted.data(), corrupted.size(), decoded, + consumed), + WalDecodeStatus::kUnknownRecordKind); +} + +TEST(WalCodecTest, TamperingDetectedInPayloadAndHeader) { + const std::vector payload = {0x11, 0x22, 0x33, 0x44}; + auto frame = EncodeFrame(7, WalRecordKind::kCowUpdate, payload); + + WalFrameHeader header; + size_t consumed = 0; + ASSERT_EQ(DecodeWalFrameHeader(frame.data(), frame.size(), header, consumed), + WalDecodeStatus::kOk); + + // Payload bit flip: the frame checksum must catch it. + auto flipped = frame; + flipped[kWalFrameHeaderSize + 1] ^= 0x80; + WalFrameHeader flipped_header; + ASSERT_EQ(DecodeWalFrameHeader(flipped.data(), flipped.size(), flipped_header, + consumed), + WalDecodeStatus::kOk); + EXPECT_EQ(ValidateWalFrame(flipped_header, flipped.data(), + flipped.data() + kWalFrameHeaderSize), + WalDecodeStatus::kBadChecksum); + + // Frame header bit flip (commit_timestamp field): the frame checksum + // covers the header bytes preceding it and must catch it too. + auto flipped_header_bytes = frame; + flipped_header_bytes[5] ^= 0x01; + WalFrameHeader ts_header; + ASSERT_EQ( + DecodeWalFrameHeader(flipped_header_bytes.data(), + flipped_header_bytes.size(), ts_header, consumed), + WalDecodeStatus::kOk); + EXPECT_EQ(ValidateWalFrame(ts_header, flipped_header_bytes.data(), + flipped_header_bytes.data() + kWalFrameHeaderSize), + WalDecodeStatus::kBadChecksum); + + // Checksum field tamper: decoding succeeds (the checksum is opaque at + // decode time), but validation fails. + auto bad_checksum = frame; + bad_checksum[9] ^= 0xFF; + WalFrameHeader checksum_header; + ASSERT_EQ(DecodeWalFrameHeader(bad_checksum.data(), bad_checksum.size(), + checksum_header, consumed), + WalDecodeStatus::kOk); + EXPECT_EQ(ValidateWalFrame(checksum_header, bad_checksum.data(), + bad_checksum.data() + kWalFrameHeaderSize), + WalDecodeStatus::kBadChecksum); +} + +// --------------------------------------------------------------------------- +// Parser file-level tests +// --------------------------------------------------------------------------- + +void ExpectWalRecoveryError(const std::string& wal_dir, + const std::string& needle) { + try { + LocalWalParser parser(wal_dir); + FAIL() << "Expected WalRecoveryException containing '" << needle << "'"; + } catch (const neug::exception::WalRecoveryException& e) { + EXPECT_NE(std::string(e.what()).find(needle), std::string::npos) + << "actual: " << e.what(); + } +} + +/// Shared temp-dir lifecycle: each test gets an isolated wal directory +/// under the system temp dir, removed on teardown. +class WalTempDirTest : public ::testing::Test { + protected: + explicit WalTempDirTest(const char* prefix) : prefix_(prefix) {} + + void SetUp() override { + const auto* info = ::testing::UnitTest::GetInstance()->current_test_info(); + wal_dir_ = (std::filesystem::temp_directory_path() / + (prefix_ + std::to_string(::getpid()) + "_" + info->name())) + .string(); + std::filesystem::remove_all(wal_dir_); + } + + void TearDown() override { std::filesystem::remove_all(wal_dir_); } + + void ExpectRecoveryErrorContaining(const std::string& needle) { + ExpectWalRecoveryError(wal_dir_, needle); + } + + std::string wal_dir_; + + private: + std::string prefix_; +}; + +class WalParserFileTest : public WalTempDirTest { + protected: + WalParserFileTest() : WalTempDirTest("neug_wal_codec_test_") {} + + void SetUp() override { + WalTempDirTest::SetUp(); + std::filesystem::create_directories(wal_dir_); + } + + /// Clears the wal dir for multi-case tests that run several scenarios + /// inside one TEST_F. + void ResetDir() { + std::filesystem::remove_all(wal_dir_); + std::filesystem::create_directories(wal_dir_); + } + + void WriteFile(const std::string& name, const std::vector& bytes) { + std::ofstream out(wal_dir_ + "/" + name, std::ios::binary); + out.write(reinterpret_cast(bytes.data()), + static_cast(bytes.size())); + } + + /// Builds a complete v1 file: file header + all encoded frames. + std::vector BuildWalFile( + const std::vector>& frames) { + auto fixture = MakeFileHeader(); + std::vector bytes(fixture.file_header_bytes.begin(), + fixture.file_header_bytes.end()); + for (const auto& frame : frames) { + bytes.insert(bytes.end(), frame.begin(), frame.end()); + } + return bytes; + } +}; + +TEST_F(WalParserFileTest, ParsesCommittedFrames) { + const std::vector payload = {1, 2, 3, 4, 5}; + WriteFile("thread_0_0.wal", + BuildWalFile({EncodeFrame(10, WalRecordKind::kInsert, payload), + EncodeFrame(11, WalRecordKind::kCompact, {})})); + + LocalWalParser parser(wal_dir_); + EXPECT_EQ(parser.last_ts(), 11u); + ASSERT_EQ(parser.replay_units().size(), 2u); + EXPECT_EQ(parser.replay_units()[0].kind, WalRecordKind::kInsert); + EXPECT_EQ(parser.replay_units()[0].commit_timestamp, 10u); + EXPECT_EQ(parser.replay_units()[0].payload, + std::string(payload.begin(), payload.end())); + EXPECT_EQ(parser.replay_units()[1].kind, WalRecordKind::kCompact); + EXPECT_TRUE(parser.replay_units()[1].payload.empty()); +} + +TEST_F(WalParserFileTest, NonV1FileIsRejectedWithoutFallback) { + // Pre-v1 content: a legacy redo record starting with a record kind, and a + // file too short to hold a v1 header. Neither may be "best-effort" parsed. + WriteFile("legacy.wal", + std::vector{0x01, 0x00, 0x00, 0x00, 0xAA, 0xBB}); + ExpectRecoveryErrorContaining("[unsupported_format]"); + + ResetDir(); + WriteFile("short.wal", std::vector(16, 0x00)); + ExpectRecoveryErrorContaining("[unsupported_format]"); +} + +TEST_F(WalParserFileTest, BadVersionIsRejected) { + WalFileHeader header; + header.format_version = 2; + const auto encoded = EncodeWalFileHeader(header); + WriteFile("thread_0_0.wal", + std::vector(encoded.begin(), encoded.end())); + ExpectRecoveryErrorContaining("[unsupported_format]"); +} + +TEST_F(WalParserFileTest, NonTailPayloadCorruptionIsRejected) { + // Two committed frames; flip a payload bit in the FIRST one. Corruption + // before the final frame must never be treated as crash residue. + auto frame1 = + EncodeFrame(1, WalRecordKind::kInsert, std::vector{1, 2, 3, 4}); + auto frame2 = + EncodeFrame(2, WalRecordKind::kInsert, std::vector{5, 6, 7, 8}); + frame1[kWalFrameHeaderSize + 1] ^= 0x40; + WriteFile("thread_0_0.wal", BuildWalFile({frame1, frame2})); + ExpectRecoveryErrorContaining("[corrupted_frame]"); +} + +TEST_F(WalParserFileTest, UnknownRecordKindIsRejected) { + auto frame = + EncodeFrame(1, WalRecordKind::kInsert, std::vector{1, 2, 3}); + frame[0] = 42; // unknown kind (first header byte) + WriteFile("thread_0_0.wal", BuildWalFile({frame})); + ExpectRecoveryErrorContaining("[unknown_record_kind]"); +} + +TEST_F(WalParserFileTest, KindPayloadConstraintsAreRejected) { + // kCompact with a non-empty payload. + WriteFile("thread_0_0.wal", + BuildWalFile({EncodeFrame(1, WalRecordKind::kCompact, + std::vector{9})})); + ExpectRecoveryErrorContaining("[corrupted_frame]"); + + ResetDir(); + + // kInsert with an empty payload. + WriteFile("thread_0_0.wal", + BuildWalFile({EncodeFrame(1, WalRecordKind::kInsert, {})})); + ExpectRecoveryErrorContaining("[corrupted_frame]"); +} + +TEST_F(WalParserFileTest, TailResidueIsIgnored) { + const std::vector payload = {1, 2, 3, 4, 5, 6, 7, 8}; + + // Each file carries one committed frame plus a torn tail of the next one. + const auto build_torn = [&](uint32_t committed_ts, size_t torn_prefix) { + const auto good = + EncodeFrame(committed_ts, WalRecordKind::kInsert, payload); + const auto next = + EncodeFrame(committed_ts + 1, WalRecordKind::kInsert, payload); + auto bytes = BuildWalFile({good}); + bytes.insert(bytes.end(), next.begin(), next.begin() + torn_prefix); + return bytes; + }; + + // Torn frame header at EOF. + WriteFile("a_half_header.wal", build_torn(1, kWalFrameHeaderSize / 2)); + // Torn payload at EOF. + WriteFile("b_half_payload.wal", + build_torn(2, kWalFrameHeaderSize + payload.size() / 2)); + + LocalWalParser parser(wal_dir_); + EXPECT_EQ(parser.last_ts(), 2u); + ASSERT_EQ(parser.replay_units().size(), 2u); + for (size_t i = 0; i < parser.replay_units().size(); ++i) { + EXPECT_EQ(parser.replay_units()[i].commit_timestamp, i + 1); + } +} + +// --------------------------------------------------------------------------- +// Multi-writer merge tests +// --------------------------------------------------------------------------- + +TEST_F(WalParserFileTest, InterleavedWritersMergeByTimestampWithGaps) { + // Writer A: ts 2, 7; writer B: ts 3, 10. Gaps are allowed. + WriteFile( + "thread_0_0.wal", + BuildWalFile( + {EncodeFrame(2, WalRecordKind::kInsert, std::vector{'a'}), + EncodeFrame(7, WalRecordKind::kInsert, std::vector{'c'})})); + WriteFile("thread_0_1.wal", + BuildWalFile({EncodeFrame(3, WalRecordKind::kCowUpdate, + std::vector{'b'}), + EncodeFrame(10, WalRecordKind::kCompact, {})})); + + LocalWalParser parser(wal_dir_); + ASSERT_EQ(parser.replay_units().size(), 4u); + EXPECT_EQ(parser.replay_units()[0].commit_timestamp, 2u); + EXPECT_EQ(parser.replay_units()[1].commit_timestamp, 3u); + EXPECT_EQ(parser.replay_units()[1].kind, WalRecordKind::kCowUpdate); + EXPECT_EQ(parser.replay_units()[2].commit_timestamp, 7u); + EXPECT_EQ(parser.replay_units()[3].commit_timestamp, 10u); + EXPECT_EQ(parser.replay_units()[3].kind, WalRecordKind::kCompact); + EXPECT_EQ(parser.last_ts(), 10u); +} + +TEST_F(WalParserFileTest, DuplicateTimestampsAreAlwaysRejected) { + struct Case { + WalRecordKind first; + WalRecordKind second; + }; + const Case cases[] = { + {WalRecordKind::kInsert, WalRecordKind::kInsert}, + {WalRecordKind::kCowUpdate, WalRecordKind::kCowUpdate}, + {WalRecordKind::kInsert, WalRecordKind::kCowUpdate}, + }; + int case_index = 0; + for (const auto& c : cases) { + ++case_index; + ResetDir(); + WriteFile( + "thread_0_0.wal", + BuildWalFile({EncodeFrame(5, c.first, std::vector{'x'})})); + WriteFile( + "thread_0_1.wal", + BuildWalFile({EncodeFrame(5, c.second, std::vector{'y'})})); + try { + LocalWalParser parser(wal_dir_); + FAIL() << "case " << case_index << ": duplicate timestamp accepted"; + } catch (const neug::exception::WalRecoveryException& e) { + EXPECT_NE(std::string(e.what()).find("[duplicate_timestamp]"), + std::string::npos) + << "case " << case_index << ": " << e.what(); + EXPECT_NE(std::string(e.what()).find("thread_0_0.wal"), std::string::npos) + << "case " << case_index << ": both sources must be reported"; + EXPECT_NE(std::string(e.what()).find("thread_0_1.wal"), std::string::npos) + << "case " << case_index << ": both sources must be reported"; + } + } +} + +// A failed frame attempt must be rolled back to the clean logical EOF; the +// writer stays usable and its next frame lands directly after the last +// committed one, so recovery never sees residue buried mid-file. +TEST_F(WalParserFileTest, FailedFrameAttemptsRollBackToCleanEof) { + const auto phases = {LocalWalWriter::FailNextWrite::kHeader, + LocalWalWriter::FailNextWrite::kPayload}; + int phase_index = 0; + for (const auto phase : phases) { + ++phase_index; + ResetDir(); + + LocalWalWriter writer(wal_dir_, /*slot_id=*/0); + writer.open(wal_dir_); + ASSERT_TRUE(writer.append_frame(1, WalRecordKind::kInsert, "one", 3)) + << "phase_index=" << phase_index; + + writer.fail_next_write(phase); + ASSERT_FALSE(writer.append_frame(2, WalRecordKind::kInsert, "two", 3)) + << "phase_index=" << phase_index; + + ASSERT_TRUE(writer.append_frame(3, WalRecordKind::kInsert, "three", 5)) + << "phase_index=" << phase_index; + writer.close(); + + LocalWalParser parser(wal_dir_); + ASSERT_EQ(parser.replay_units().size(), 2u) + << "phase_index=" << phase_index; + EXPECT_EQ(parser.replay_units()[0].commit_timestamp, 1u); + EXPECT_EQ(parser.replay_units()[0].payload, "one"); + EXPECT_EQ(parser.replay_units()[1].commit_timestamp, 3u); + EXPECT_EQ(parser.replay_units()[1].payload, "three"); + } +} + +// --------------------------------------------------------------------------- +// Subprocess crash tests: the child writes through the real LocalWalWriter +// and dies at a precise phase; the parent recovers. +// --------------------------------------------------------------------------- + +class WalCrashSubprocessTest : public WalTempDirTest { + protected: + WalCrashSubprocessTest() : WalTempDirTest("neug_wal_crash_test_") {} + + /// Runs @p child_body in a forked process that exits via _exit() without + /// any cleanup, simulating a crash. Waits for normal termination. + void RunCrashChild(const std::function& child_body) { + const pid_t pid = ::fork(); + ASSERT_NE(pid, -1); + if (pid == 0) { + child_body(); + ::_exit(0); + } + int status = 0; + ASSERT_EQ(::waitpid(pid, &status, 0), pid); + ASSERT_TRUE(WIFEXITED(status)); + ASSERT_EQ(WEXITSTATUS(status), 0); + } +}; + +TEST_F(WalCrashSubprocessTest, CrashAtEachWritePhaseLeavesOnlyCommittedFrames) { + const std::string payload_a = "committed-payload"; + const std::string payload_b = "crashed-payload-not-visible"; + + const auto phases = {LocalWalWriter::FailNextWrite::kHeader, + LocalWalWriter::FailNextWrite::kPayload}; + int phase_index = 0; + for (const auto phase : phases) { + ++phase_index; + std::filesystem::remove_all(wal_dir_); + + RunCrashChild([&] { + LocalWalWriter writer(wal_dir_, /*slot_id=*/0); + writer.open(wal_dir_); + if (!writer.append_frame(1, WalRecordKind::kInsert, payload_a.data(), + payload_a.size())) { + ::_exit(1); + } + writer.fail_next_write(phase); + // The injected failure makes this commit fail cleanly; the process + // then dies without closing the file, like a real crash. + if (writer.append_frame(2, WalRecordKind::kInsert, payload_b.data(), + payload_b.size())) { + ::_exit(1); + } + }); + + LocalWalParser parser(wal_dir_); + EXPECT_EQ(parser.last_ts(), 1u) << "phase_index=" << phase_index; + ASSERT_EQ(parser.replay_units().size(), 1u) + << "phase_index=" << phase_index; + EXPECT_EQ(parser.replay_units()[0].payload, payload_a); + } +} + +TEST_F(WalCrashSubprocessTest, ReplayAcrossCompactInOrder) { + RunCrashChild([&] { + LocalWalWriter writer(wal_dir_, /*slot_id=*/0); + writer.open(wal_dir_); + if (!writer.append_frame(1, WalRecordKind::kInsert, "insert-a", 8) || + !writer.append_frame(2, WalRecordKind::kCompact, nullptr, 0) || + !writer.append_frame(3, WalRecordKind::kInsert, "insert-b", 8)) { + ::_exit(1); + } + }); + + LocalWalParser parser(wal_dir_); + ASSERT_EQ(parser.replay_units().size(), 3u); + EXPECT_EQ(parser.replay_units()[0].kind, WalRecordKind::kInsert); + EXPECT_EQ(parser.replay_units()[1].kind, WalRecordKind::kCompact); + EXPECT_EQ(parser.replay_units()[2].kind, WalRecordKind::kInsert); + EXPECT_EQ(parser.replay_units()[2].payload, "insert-b"); +} + +TEST_F(WalCrashSubprocessTest, BitFlippedPayloadIsRejected) { + RunCrashChild([&] { + LocalWalWriter writer(wal_dir_, /*slot_id=*/0); + writer.open(wal_dir_); + if (!writer.append_frame(1, WalRecordKind::kInsert, "stable-payload", 14)) { + ::_exit(1); + } + }); + + // Corrupt one payload byte after the "crash". + for (auto& entry : std::filesystem::directory_iterator(wal_dir_)) { + std::fstream file(entry.path(), + std::ios::in | std::ios::out | std::ios::binary); + ASSERT_TRUE(file.is_open()); + char byte = 0; + file.seekp(kWalFileHeaderSize + kWalFrameHeaderSize + 2); + file.read(&byte, 1); + file.seekp(kWalFileHeaderSize + kWalFrameHeaderSize + 2); + byte ^= 0x55; + file.write(&byte, 1); + } + + ExpectRecoveryErrorContaining("[corrupted_frame]"); +} + +} // namespace +} // namespace neug diff --git a/tests/transaction/test_wal_replay.cc b/tests/transaction/test_wal_replay.cc index 8081584f0..b364906dd 100644 --- a/tests/transaction/test_wal_replay.cc +++ b/tests/transaction/test_wal_replay.cc @@ -31,7 +31,6 @@ #include #include #include -#include #include #include #include @@ -40,6 +39,7 @@ #include #include "gtest/gtest.h" +#include "neug/transaction/wal/local_wal_parser.h" #include "neug/transaction/wal/wal.h" #include "unittest/utils.h" @@ -62,40 +62,44 @@ std::string make_test_dir() { return (std::filesystem::temp_directory_path() / dir_name).string(); } -TEST(WalWriterTest, ReopensSameInstanceOnNewTimeline) { +TEST(WalWriterTest, WriterOnlyWritesItsOwnWalDirAndRejectsAppendAfterClose) { const auto test_dir = make_test_dir(); const auto old_wal_dir = (std::filesystem::path(test_dir) / "checkpoint-0" / "wal").string(); const auto new_wal_dir = (std::filesystem::path(test_dir) / "checkpoint-1" / "wal").string(); - constexpr uint32_t old_marker = 17; - constexpr uint32_t new_marker = 29; { auto writer = neug::WalWriterFactory::CreateWalWriter(old_wal_dir, 0); auto* const identity = writer.get(); writer->open(old_wal_dir); - ASSERT_TRUE(writer->append(reinterpret_cast(&old_marker), - sizeof(old_marker))); + ASSERT_TRUE(writer->append_frame(/*commit_timestamp=*/1, + neug::WalRecordKind::kInsert, + "old-payload", 11)); writer->open(new_wal_dir); EXPECT_EQ(writer.get(), identity); - ASSERT_TRUE(writer->append(reinterpret_cast(&new_marker), - sizeof(new_marker))); + ASSERT_TRUE(writer->append_frame(/*commit_timestamp=*/1, + neug::WalRecordKind::kInsert, + "new-payload", 11)); writer->close(); + + // A closed writer must reject further appends. + EXPECT_FALSE(writer->append_frame(/*commit_timestamp=*/2, + neug::WalRecordKind::kInsert, "late", 4)); } - const auto read_marker = [](const std::string& wal_dir) { - const auto begin = std::filesystem::directory_iterator(wal_dir); - const auto end = std::filesystem::directory_iterator(); - EXPECT_NE(begin, end); - std::ifstream wal_file(begin->path(), std::ios::binary); - uint32_t marker = 0; - wal_file.read(reinterpret_cast(&marker), sizeof(marker)); - return marker; + // Each wal dir holds exactly the frames written to its own timeline. + const auto payloads = [](const std::string& wal_dir) { + neug::LocalWalParser parser(wal_dir); + std::vector result; + for (const auto& unit : parser.replay_units()) { + result.emplace_back(unit.payload); + } + return result; }; - EXPECT_EQ(read_marker(old_wal_dir), old_marker); - EXPECT_EQ(read_marker(new_wal_dir), new_marker); + EXPECT_EQ(payloads(old_wal_dir), std::vector{"old-payload"}); + EXPECT_EQ(payloads(new_wal_dir), std::vector{"new-payload"}); std::filesystem::remove_all(test_dir); } @@ -244,6 +248,21 @@ bool read_has_person(neug::NeugDBService& service, int64_t id) { return found; } +std::string read_person_name(neug::NeugDBService& service, int64_t id) { + auto slot = service.AcquireExecutionSlot(); + auto txn = slot->GetReadTransaction(); + neug::StorageReadInterface graph(txn.view(), txn.timestamp()); + const auto person_label = graph.schema().get_vertex_label_id("person"); + neug::vid_t vid = 0; + std::string name; + if (graph.GetVertexIndex(person_label, Value::INT64(id), vid)) { + auto name_col = graph.GetVertexPropColumn(person_label, "name"); + name = name_col->get_any(vid).GetValue(); + } + EXPECT_TRUE(txn.Commit()); + return name; +} + void create_wal_with_insert_compact_insert_collision( const std::string& db_dir) { neug::NeugDB db; @@ -575,8 +594,9 @@ TEST(CheckpointCoordinatorTest, wal_writer->open(old_wal_dir); constexpr uint32_t before_marker = 17; constexpr uint32_t after_marker = 29; - ASSERT_TRUE(wal_writer->append(reinterpret_cast(&before_marker), - sizeof(before_marker))); + ASSERT_TRUE(wal_writer->append_frame( + before_marker, neug::WalRecordKind::kInsert, + reinterpret_cast(&before_marker), sizeof(before_marker))); // Keep the checkpoint manager's only staging slot occupied. // PublishManualCheckpoint must fail before destructive graph maintenance, @@ -598,8 +618,9 @@ TEST(CheckpointCoordinatorTest, EXPECT_FALSE(allocator_reopened); EXPECT_EQ(allocators[0]->allocated_memory(), allocator_marker_size); EXPECT_FALSE(cache_invalidated); - EXPECT_TRUE(wal_writer->append(reinterpret_cast(&after_marker), - sizeof(after_marker))); + EXPECT_TRUE(wal_writer->append_frame( + after_marker, neug::WalRecordKind::kInsert, + reinterpret_cast(&after_marker), sizeof(after_marker))); const auto read_view = version_manager.acquire_read_view(); EXPECT_EQ(read_view.visibility_ts, update_ts); @@ -766,3 +787,79 @@ TEST_F(WalReplayTest, ReopenReplaysInsertWalAcrossCompactionInDependencyOrder) { }, ::testing::ExitedWithCode(0), ".*"); } + +// AP writes (schema + direct data path) must produce zero WAL frames; each +// TP transaction kind produces exactly one committed frame. +TEST_F(WalReplayTest, TpCommitsProduceOneFrameEachAndApWritesNone) { + neug::NeugDB db; + ASSERT_TRUE(db.Open(make_config(db_dir_))); + create_person_schema(db); // AP path: must not write any WAL frame + { + neug::NeugDBService service(db); + insert_person(service, 1, "seed"); + + { + auto slot = service.AcquireExecutionSlot(); + auto update_result = slot->ExecuteTransactionalRequest( + R"({"query":"MATCH (p:person {id: 1}) SET p.name = 'renamed';","access_mode":"update","parameters":{}})"); + ASSERT_TRUE(update_result) << update_result.error().ToString(); + } + compact(service); + } + db.Close(); + + // Locate the checkpoint wal directory. + std::string wal_dir; + for (const auto& entry : std::filesystem::directory_iterator(db_dir_)) { + if (entry.is_directory() && + entry.path().filename().string().rfind("checkpoint-", 0) == 0) { + auto candidate = entry.path() / "wal"; + if (std::filesystem::exists(candidate)) { + wal_dir = candidate.string(); + } + } + } + ASSERT_FALSE(wal_dir.empty()); + + neug::LocalWalParser parser(wal_dir); + ASSERT_EQ(parser.replay_units().size(), 3u) + << "AP statements must not leave frames behind"; + EXPECT_EQ(parser.replay_units()[0].kind, neug::WalRecordKind::kInsert); + EXPECT_EQ(parser.replay_units()[1].kind, neug::WalRecordKind::kCowUpdate); + EXPECT_EQ(parser.replay_units()[2].kind, neug::WalRecordKind::kCompact); +} + +// Reopen must replay the whole insert -> compact -> update sequence on top +// of the checkpoint, including an update committed after the compaction. +TEST_F(WalReplayTest, ReopenReplaysUpdateCommittedAfterCompaction) { + create_checkpointed_base_graph(db_dir_); + + { + neug::NeugDB db; + ASSERT_TRUE(db.Open(make_config(db_dir_))); + { + neug::NeugDBService service(db); + insert_person(service, 2, "pre-compact"); + compact(service); + auto slot = service.AcquireExecutionSlot(); + auto update_result = slot->ExecuteTransactionalRequest( + R"({"query":"MATCH (p:person {id: 1}) SET p.name = 'post-compact';","access_mode":"update","parameters":{}})"); + ASSERT_TRUE(update_result) << update_result.error().ToString(); + } + db.Close(); + } + + { + neug::NeugDB db; + ASSERT_TRUE(db.Open(make_config(db_dir_))); + { + neug::NeugDBService service(db); + EXPECT_EQ(read_person_count(service), 2); + EXPECT_EQ(read_person_name(service, 2), "pre-compact") + << "the pre-compaction insert must survive replay"; + EXPECT_EQ(read_person_name(service, 1), "post-compact") + << "the update committed after compaction must be replayed"; + } + db.Close(); + } +}