feat: frame TP WAL into per-commit records with validated recovery

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 (24B)
+  ├─ magic / format_version / header_size
+  └─ writer_slot_id       # diagnostics only, not part of transaction order
+Frame* (one frame per committed transaction)
+  ├─ FrameHeader (24B): magic, record_kind,
+  │                     commit_timestamp, payload_length
+  ├─ payload              # redo bytes; empty only for compaction frames
+  └─ FrameTrailer (8B): commit marker, frame CRC32C over
+                          header + payload + marker
+```
+
+Key properties:
+
+- **One commit = exactly one complete frame.** The commit marker is written
+  last, so a frame whose trailer is present was fully persisted.
+- **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

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

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;

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_; }
 

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,14 @@ 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;
+  WalWritePhase write_phase() const 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

include/neug/transaction/wal/local_wal_parser.h

@@ -24,31 +24,46 @@
 
 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<IWalParser> Make(const std::string& wal_dir) {
     return std::unique_ptr<IWalParser>(new LocalWalParser(wal_dir));
   }
 
   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<UpdateWalUnit>& get_update_wals() const override;
+  const std::vector<WalReplayUnit>& replay_units() const override;
 
  private:
-  std::vector<int> fds_;
-  std::vector<void*> mmapped_ptrs_;
-  std::vector<size_t> mmapped_size_;
-  std::vector<WalContentUnit> 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<MappedFiles> mapped_files_;
+  std::vector<WalReplayUnit> replay_units_;
   uint32_t last_ts_{0};
 
-  std::vector<UpdateWalUnit> update_wal_list_;
-
   static const bool registered_;
 };
 

include/neug/transaction/wal/local_wal_writer.h

@@ -15,38 +15,67 @@
 #pragma once
 
 #include <stddef.h>
+#include <stdint.h>
 #include <memory>
 #include <string>
 
 #include "neug/transaction/wal/wal.h"
 
 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 and payload, then persists the commit trailer separately so the
+ * marker is never present before the full payload.
+ */
 class LocalWalWriter : public IWalWriter {
  public:
   static std::unique_ptr<IWalWriter> 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) {}
+        append_offset_(0),
+        write_phase_(WalWritePhase::kIdle) {}
   ~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;
+  WalWritePhase write_phase() const override { return write_phase_; }
   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, kTrailer };
+  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_;
+  WalWritePhase write_phase_;
+  FailNextWrite fail_next_write_{FailNextWrite::kNone};
+  bool failed_{false};
 
   static const bool registered_;
 };

include/neug/transaction/wal/wal.h

@@ -18,54 +18,56 @@
 #include <stdint.h>
 #include <memory>
 #include <string>
+#include <string_view>
 #include <unordered_map>
 #include <vector>
 
 #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);
 
+/// Narrow I/O phase information exposed by the writer around the commit
+/// marker. P1-3 consumes this to decide commit durability; P1-2 does not map
+/// it to any externally visible commit result.
+enum class WalWritePhase : uint8_t {
+  /// No frame write in progress (or the last frame is fully persisted).
+  kIdle,
+  /// Frame header and payload are persisted; the commit marker write has not
+  /// been attempted yet.
+  kBeforeMarker,
+  /// The commit marker write has been attempted (success not implied).
+  kMarkerAttempted,
+};
+
 /**
  * The interface of wal writer.
  *
  * 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:
   virtual ~IWalWriter() noexcept = default;
 
   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,36 +79,59 @@ class IWalWriter {
   virtual void close() = 0;
 
   /**
-   * Append data to the wal file.
+   * Append one complete transaction frame:
+   * frame header + payload first, then the commit trailer with its marker.
    */
-  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;
+
+  /// Current marker-phase information (see WalWritePhase).
+  virtual WalWritePhase write_phase() const = 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;
 
   virtual void close() = 0;
 
   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<UpdateWalUnit>& get_update_wals() const = 0;
+  virtual const std::vector<WalReplayUnit>& replay_units() const = 0;
 };
 
 class WalWriterFactory {