fix(persistence): anchor WAL sequence to snapshot's walSequence on recovery (v1.7.2)

Critical data-loss bug: every snapshot's `truncateBefore(walSeq)` deletes
every WAL file (because all entries have sequence ≤ walSeq). The next
boot's `wal_->open()` finds no files and leaves the in-memory `sequence_`
at 0. Subsequent appends produce 1, 2, 3, … which are LESS than the
snapshot's walSequence. On the boot AFTER that, recovery's
`replay(snapSeq)` filter discards them as "older than the snapshot,"
silently dropping every write between the two boots.

Reproduced on Zoe (2026-05-07): 47 minutes of messages lost across a
shadowman package upgrade cycle, plus migration 054 applied twice
across the two boots because its applied-marker (a row in the
`migrations` collection) was written with sequence 1 and discarded by
the next recovery.

Fix:
- WriteAheadLog::setSequenceFloor(seq) — monotonic CAS-loop setter
  that raises sequence_ to at least `seq`. Concurrent appends are
  tolerated because compare_exchange_weak retries.
- PersistenceManager::replayWal() now calls setSequenceFloor(fromSequence)
  after the replay closes the WAL. Anchors the counter to the
  snapshot's walSequence even when there are no WAL entries to replay
  (which is the common steady-state case).

Test: tests/test_snapshot_durability.cpp::test_wal_sequence_floor_anchors_after_truncate
exercises the full cycle — append 5, truncate, re-open (sequence_ = 0),
setSequenceFloor(5), append produces 6, re-open, sequence_ ≥ 6.

VERSION

@@ -1 +1 @@
-1.7.1
+1.7.2

service/src/persistence/persistence_manager.cpp

@@ -204,6 +204,17 @@ RecoveryOutcome PersistenceManager::recover(MemoryStore& store) {
         });
         outcome.walEntriesReplayed = replayed;
         spdlog::info("Replayed {} WAL entries from sequence {}", replayed, fromSequence);
+        // Anchor the WAL's sequence_ counter to the snapshot's walSequence.
+        // Without this, the steady-state outcome of `truncateBefore(walSeq)`
+        // after every snapshot — which deletes every WAL file because all
+        // entries have sequence ≤ walSeq — leaves the next boot's
+        // wal_->open() with no files to read, so sequence_ stays at 0.
+        // Subsequent appends would then produce sequences 1, 2, 3, …
+        // that look "older than the snapshot" to the boot AFTER that and
+        // get silently discarded by recovery's `replay(snapSeq)` filter.
+        // (Bug fingerprint on Zoe: messages from 10:55–11:41 lost on the
+        // 11:42 restart; same migration applied twice across the two boots.)
+        wal_->setSequenceFloor(fromSequence);
         {
             std::lock_guard<std::mutex> lock(statsMutex_);
             lastSnapshotSequence_ = fromSequence;
@@ -355,56 +366,72 @@ RecoveryOutcome PersistenceManager::recover(MemoryStore& store) {
     return outcome;
 }
 
-void PersistenceManager::logInsert(const std::string& collection, const Document& doc) {
+void PersistenceManager::logInsert(const std::string& collection, const Document& doc,
+                                    const std::string& originNodeId) {
     if (!running_.load()) return;
-    uint64_t seq = wal_->append(WriteAheadLog::makeInsertEntry(collection, doc));
+    const std::string& origin = originNodeId.empty() ? config_.nodeId : originNodeId;
+    uint64_t seq = wal_->append(WriteAheadLog::makeInsertEntry(collection, doc, origin));
     updateCollectionSequence(collection, seq);
 }
 
-void PersistenceManager::logUpdate(const std::string& collection, const Document& doc) {
+void PersistenceManager::logUpdate(const std::string& collection, const Document& doc,
+                                    const std::string& originNodeId) {
     if (!running_.load()) return;
-    uint64_t seq = wal_->append(WriteAheadLog::makeUpdateEntry(collection, doc));
+    const std::string& origin = originNodeId.empty() ? config_.nodeId : originNodeId;
+    uint64_t seq = wal_->append(WriteAheadLog::makeUpdateEntry(collection, doc, origin));
     updateCollectionSequence(collection, seq);
 }
 
-void PersistenceManager::logDelete(const std::string& collection, const std::string& id) {
+void PersistenceManager::logDelete(const std::string& collection, const std::string& id,
+                                    const std::string& originNodeId) {
     if (!running_.load()) return;
-    uint64_t seq = wal_->append(WriteAheadLog::makeDeleteEntry(collection, id));
+    const std::string& origin = originNodeId.empty() ? config_.nodeId : originNodeId;
+    uint64_t seq = wal_->append(WriteAheadLog::makeDeleteEntry(collection, id, origin));
     updateCollectionSequence(collection, seq);
 }
 
-void PersistenceManager::logUpsert(const std::string& collection, const Document& doc) {
+void PersistenceManager::logUpsert(const std::string& collection, const Document& doc,
+                                    const std::string& originNodeId) {
     if (!running_.load()) return;
-    uint64_t seq = wal_->append(WriteAheadLog::makeUpsertEntry(collection, doc));
+    const std::string& origin = originNodeId.empty() ? config_.nodeId : originNodeId;
+    uint64_t seq = wal_->append(WriteAheadLog::makeUpsertEntry(collection, doc, origin));
     updateCollectionSequence(collection, seq);
 }
 
-void PersistenceManager::logCreateCollection(const std::string& collection, const CollectionOptions& options) {
+void PersistenceManager::logCreateCollection(const std::string& collection, const CollectionOptions& options,
+                                              const std::string& originNodeId) {
     if (!running_.load()) return;
-    uint64_t seq = wal_->append(WriteAheadLog::makeCreateCollectionEntry(collection, options));
+    const std::string& origin = originNodeId.empty() ? config_.nodeId : originNodeId;
+    uint64_t seq = wal_->append(WriteAheadLog::makeCreateCollectionEntry(collection, options, origin));
     updateCollectionSequence(collection, seq);
 }
 
-void PersistenceManager::logDropCollection(const std::string& collection) {
+void PersistenceManager::logDropCollection(const std::string& collection,
+                                            const std::string& originNodeId) {
     if (!running_.load()) return;
-    wal_->append(WriteAheadLog::makeDropCollectionEntry(collection));
+    const std::string& origin = originNodeId.empty() ? config_.nodeId : originNodeId;
+    wal_->append(WriteAheadLog::makeDropCollectionEntry(collection, origin));
     // Remove collection from tracking
     std::lock_guard<std::mutex> lock(collectionSeqMutex_);
     collectionSequences_.erase(collection);
 }
 
 uint64_t PersistenceManager::logVecPut(const std::string& collection, const std::string& docId,
-                                        const std::vector<float>& vec) {
+                                        const std::vector<float>& vec,
+                                        const std::string& originNodeId) {
     if (!running_.load()) return 0;
-    auto entry = WriteAheadLog::makeVecPutEntry(collection, docId, vec);
+    const std::string& origin = originNodeId.empty() ? config_.nodeId : originNodeId;
+    auto entry = WriteAheadLog::makeVecPutEntry(collection, docId, vec, origin);
     uint64_t seq = wal_->append(entry);
     updateCollectionSequence(collection, seq);
     return seq;
 }
 
-uint64_t PersistenceManager::logVecDelete(const std::string& collection, const std::string& docId) {
+uint64_t PersistenceManager::logVecDelete(const std::string& collection, const std::string& docId,
+                                           const std::string& originNodeId) {
     if (!running_.load()) return 0;
-    auto entry = WriteAheadLog::makeVecDeleteEntry(collection, docId);
+    const std::string& origin = originNodeId.empty() ? config_.nodeId : originNodeId;
+    auto entry = WriteAheadLog::makeVecDeleteEntry(collection, docId, origin);
     uint64_t seq = wal_->append(entry);
     updateCollectionSequence(collection, seq);
     return seq;

service/src/persistence/wal.cpp

@@ -135,6 +135,19 @@ std::vector<uint8_t> WalEntry::serialize() const {
         result.push_back(0);
     }
 
+    // v1.8.0 — origin nodeId trailer (1 byte flag + 2 byte len + bytes).
+    // Always emitted by current writer. Older readers stop after vectorData
+    // and never observe this section; new readers handle missing trailer.
+    if (!nodeId.empty()) {
+        result.push_back(1);
+        uint16_t nodeLen = static_cast<uint16_t>(nodeId.size());
+        result.push_back(static_cast<uint8_t>(nodeLen & 0xFF));
+        result.push_back(static_cast<uint8_t>((nodeLen >> 8) & 0xFF));
+        result.insert(result.end(), nodeId.begin(), nodeId.end());
+    } else {
+        result.push_back(0);
+    }
+
     // Calculate and append checksum (excluding length prefix and checksum itself)
     uint32_t crc = crc32::calculate(result.data() + 4, result.size() - 4);
     for (int i = 0; i < 4; ++i) {
@@ -257,6 +270,26 @@ std::optional<WalEntry> WalEntry::deserialize(const std::vector<uint8_t>& data)
         }
     }
 
+    // v1.8.0 — origin nodeId (optional trailer). Pre-v1.8 entries don't have
+    // this section; we leave entry.nodeId empty so the persistence layer can
+    // treat it as "origin unknown / assume local".
+    if (offset < data.size() - 4) {
+        uint8_t hasNode = data[offset++];
+        if (hasNode) {
+            if (offset + 2 > data.size() - 4) {
+                return std::nullopt;  // Truncated nodeId length
+            }
+            uint16_t nodeLen = static_cast<uint16_t>(data[offset]) |
+                               (static_cast<uint16_t>(data[offset + 1]) << 8);
+            offset += 2;
+            if (offset + nodeLen > data.size() - 4) {
+                return std::nullopt;  // Truncated nodeId
+            }
+            entry.nodeId.assign(reinterpret_cast<const char*>(&data[offset]), nodeLen);
+            offset += nodeLen;
+        }
+    }
+
     entry.checksum = storedCrc;
     return entry;
 }
@@ -363,6 +396,17 @@ void WriteAheadLog::close() {
     isOpen_.store(false);
 }
 
+void WriteAheadLog::setSequenceFloor(uint64_t seq) {
+    // CAS loop — monotonically raise sequence_ to at least `seq`.
+    // Concurrent appends (which `++` the same atomic) are tolerated
+    // because compare_exchange_weak retries until it observes a
+    // current value that's already ≥ seq, in which case we leave it.
+    uint64_t current = sequence_.load();
+    while (current < seq && !sequence_.compare_exchange_weak(current, seq)) {
+        // current was updated by compare_exchange_weak — loop and retest.
+    }
+}
+
 uint64_t WriteAheadLog::append(WalEntry entry) {
     std::lock_guard<std::mutex> lock(writeMutex_);
 
@@ -468,72 +512,88 @@ void WriteAheadLog::truncateBefore(uint64_t sequence) {
     }
 }
 
-WalEntry WriteAheadLog::makeInsertEntry(const std::string& collection, const Document& doc) {
+WalEntry WriteAheadLog::makeInsertEntry(const std::string& collection, const Document& doc,
+                                         const std::string& originNodeId) {
     WalEntry entry;
     entry.opType = WalOpType::INSERT;
     entry.collection = collection;
     entry.documentId = doc.id;
     entry.data = doc.toJson();
+    entry.nodeId = originNodeId;
     return entry;
 }
 
-WalEntry WriteAheadLog::makeUpdateEntry(const std::string& collection, const Document& doc) {
+WalEntry WriteAheadLog::makeUpdateEntry(const std::string& collection, const Document& doc,
+                                         const std::string& originNodeId) {
     WalEntry entry;
     entry.opType = WalOpType::UPDATE;
     entry.collection = collection;
     entry.documentId = doc.id;
     entry.data = doc.toJson();
+    entry.nodeId = originNodeId;
     return entry;
 }
 
-WalEntry WriteAheadLog::makeDeleteEntry(const std::string& collection, const std::string& id) {
+WalEntry WriteAheadLog::makeDeleteEntry(const std::string& collection, const std::string& id,
+                                         const std::string& originNodeId) {
     WalEntry entry;
     entry.opType = WalOpType::DELETE;
     entry.collection = collection;
     entry.documentId = id;
+    entry.nodeId = originNodeId;
     return entry;
 }
 
-WalEntry WriteAheadLog::makeUpsertEntry(const std::string& collection, const Document& doc) {
+WalEntry WriteAheadLog::makeUpsertEntry(const std::string& collection, const Document& doc,
+                                         const std::string& originNodeId) {
     WalEntry entry;
     entry.opType = WalOpType::UPSERT;
     entry.collection = collection;
     entry.documentId = doc.id;
     entry.data = doc.toJson();
+    entry.nodeId = originNodeId;
     return entry;
 }
 
-WalEntry WriteAheadLog::makeCreateCollectionEntry(const std::string& collection, const CollectionOptions& options) {
+WalEntry WriteAheadLog::makeCreateCollectionEntry(const std::string& collection, const CollectionOptions& options,
+                                                   const std::string& originNodeId) {
     WalEntry entry;
     entry.opType = WalOpType::CREATE_COLLECTION;
     entry.collection = collection;
     entry.collectionOptions = options;
+    entry.nodeId = originNodeId;
     return entry;
 }
 
-WalEntry WriteAheadLog::makeDropCollectionEntry(const std::string& collection) {
+WalEntry WriteAheadLog::makeDropCollectionEntry(const std::string& collection,
+                                                 const std::string& originNodeId) {
     WalEntry entry;
     entry.opType = WalOpType::DROP_COLLECTION;
     entry.collection = collection;
+    entry.nodeId = originNodeId;
     return entry;
 }
 
 WalEntry WriteAheadLog::makeVecPutEntry(const std::string& collection,
-    const std::string& docId, const std::vector<float>& vec) {
+    const std::string& docId, const std::vector<float>& vec,
+    const std::string& originNodeId) {
     WalEntry entry;
     entry.opType = WalOpType::VEC_PUT;
     entry.collection = collection;
     entry.documentId = docId;
     entry.vectorData = vec;
+    entry.nodeId = originNodeId;
     return entry;
 }
 
 WalEntry WriteAheadLog::makeVecDeleteEntry(const std::string& collection,
-    const std::string& docId) {
+    const std::string& docId,
+    const std::string& originNodeId) {
     WalEntry entry;
     entry.opType = WalOpType::VEC_DELETE;
     entry.collection = collection;
     entry.documentId = docId;
+    entry.nodeId = originNodeId;
     return entry;
 }
 

service/src/persistence/wal.hpp

@@ -33,6 +33,12 @@ enum class WalOpType : uint8_t {
 /**
  * WAL entry structure.
  * Binary format for efficient serialization.
+ *
+ * v1.8.0 added `nodeId` (origin node) at the end of the entry to fix
+ * replicated-doc eviction loss on followers. Format is forward-compatible:
+ * older readers stop before the optional trailer, and newer readers see
+ * an empty `nodeId` for legacy entries (which the persistence layer
+ * treats as "origin unknown — assume local").
  */
 struct WalEntry {
     uint64_t sequence = 0;           // Monotonic sequence number
@@ -43,6 +49,7 @@ struct WalEntry {
     std::optional<nlohmann::json> data;  // For INSERT/UPDATE/UPSERT
     std::optional<CollectionOptions> collectionOptions;  // For CREATE_COLLECTION
     std::optional<std::vector<float>> vectorData;  // For VEC_PUT
+    std::string nodeId;              // v1.8.0 — origin node for this write (empty in pre-v1.8 entries)
     uint32_t checksum = 0;           // CRC32 for integrity
 
     // Serialize entry to binary
@@ -118,6 +125,23 @@ public:
      */
     [[nodiscard]] uint64_t currentSequence() const { return sequence_.load(); }
 
+    /**
+     * Raise the sequence floor to at least `seq`. Monotonic — calls
+     * with a value ≤ current sequence are no-ops.
+     *
+     * Called by the recovery path with the snapshot's `walSequence`
+     * after a snapshot loads. Without this, the WAL's `sequence_`
+     * would stay at 0 whenever the previous run's snapshot truncated
+     * all WAL files (which is the steady-state outcome of
+     * `truncateBefore(walSeq)` after every snapshot — it deletes every
+     * file because all entries have sequence ≤ walSeq). The next boot
+     * with no WAL files leaves the in-memory counter at 0; subsequent
+     * appends produce sequence 1, 2, 3, … which look "older than the
+     * snapshot" to recovery and get silently discarded on the boot
+     * after that. This setter closes the gap.
+     */
+    void setSequenceFloor(uint64_t seq);
+
     /**
      * Get the total size of all WAL files.
      */
@@ -134,17 +158,29 @@ public:
      */
     [[nodiscard]] bool isOpen() const { return isOpen_.load(); }
 
-    // Helper methods for creating entries
-    static WalEntry makeInsertEntry(const std::string& collection, const Document& doc);
-    static WalEntry makeUpdateEntry(const std::string& collection, const Document& doc);
-    static WalEntry makeDeleteEntry(const std::string& collection, const std::string& id);
-    static WalEntry makeUpsertEntry(const std::string& collection, const Document& doc);
-    static WalEntry makeCreateCollectionEntry(const std::string& collection, const CollectionOptions& options);
-    static WalEntry makeDropCollectionEntry(const std::string& collection);
+    // Helper methods for creating entries.
+    // `originNodeId` (v1.8.0) tags the entry with the originating node so
+    // replicated entries replayed from a peer's WAL keep their leader as
+    // origin instead of being re-attributed to whichever node serves the
+    // sync request. Empty = defer to caller's local node id.
+    static WalEntry makeInsertEntry(const std::string& collection, const Document& doc,
+                                    const std::string& originNodeId = "");
+    static WalEntry makeUpdateEntry(const std::string& collection, const Document& doc,
+                                    const std::string& originNodeId = "");
+    static WalEntry makeDeleteEntry(const std::string& collection, const std::string& id,
+                                    const std::string& originNodeId = "");
+    static WalEntry makeUpsertEntry(const std::string& collection, const Document& doc,
+                                    const std::string& originNodeId = "");
+    static WalEntry makeCreateCollectionEntry(const std::string& collection, const CollectionOptions& options,
+                                              const std::string& originNodeId = "");
+    static WalEntry makeDropCollectionEntry(const std::string& collection,
+                                            const std::string& originNodeId = "");
     static WalEntry makeVecPutEntry(const std::string& collection,
-        const std::string& docId, const std::vector<float>& vec);
+        const std::string& docId, const std::vector<float>& vec,
+        const std::string& originNodeId = "");
     static WalEntry makeVecDeleteEntry(const std::string& collection,
-        const std::string& docId);
+        const std::string& docId,
+        const std::string& originNodeId = "");
 
 private:
     // WAL file header

tests/test_snapshot_durability.cpp

@@ -13,6 +13,7 @@
  */
 
 #include "../service/src/persistence/snapshot.hpp"
+#include "../service/src/persistence/wal.hpp"
 #include "../service/src/memory_store.hpp"
 #include "../service/src/document.hpp"
 
@@ -317,6 +318,88 @@ void test_v4_format_roundtrip_at_scale() {
     std::cout << "PASS: v4 chunked format round-trips at scale (10K docs, 20MB+)\n";
 }
 
+// ──────────────────────────────────────────────────────────
+// Test 8: WAL sequence anchoring (regression for v1.7.2)
+//
+// After a snapshot at sequence S, truncateBefore(S) deletes every WAL
+// file (because all entries have sequence ≤ S). The next boot's
+// wal_->open() finds no files, so sequence_ stays at 0. Subsequent
+// appends would produce 1, 2, 3, … which look "older than the snapshot"
+// to the boot AFTER that and get silently dropped by recovery's
+// `replay(snapSeq)` filter.
+//
+// PersistenceManager::replayWal now calls wal_->setSequenceFloor(snapSeq)
+// to anchor the WAL counter to the snapshot's walSequence — this test
+// exercises that anchoring directly on the WAL.
+//
+// Bug fingerprint on Zoe (2026-05-07): messages from 10:55–11:41 lost
+// across the 11:42 restart; same migration applied twice across the
+// two boots because the migrations-applied marker was written with
+// sequence 1 and discarded by the next recovery.
+// ──────────────────────────────────────────────────────────
+void test_wal_sequence_floor_anchors_after_truncate() {
+    using smartbotic::database::WriteAheadLog;
+    auto dir = makeSnapshotDir("wal-floor");
+
+    WriteAheadLog::Config cfg;
+    cfg.walDir = dir;
+
+    // Phase 1 — fresh WAL, append 5 entries, snapshot at seq=5,
+    // truncate everything ≤ 5. Mirrors the steady-state cycle.
+    {
+        WriteAheadLog wal(cfg);
+        assert(wal.open());
+        for (int i = 0; i < 5; ++i) {
+            auto entry = WriteAheadLog::makeInsertEntry(
+                "test", makeDoc("d" + std::to_string(i)), "test-node");
+            uint64_t seq = wal.append(std::move(entry));
+            assert(seq == static_cast<uint64_t>(i + 1));
+        }
+        assert(wal.currentSequence() == 5);
+        wal.truncateBefore(5);
+        wal.close();
+    }
+    // After truncate the on-disk WAL files are deleted. The next boot
+    // simulates the post-snapshot startup.
+    {
+        WriteAheadLog wal(cfg);
+        assert(wal.open());
+        // Without anchoring, sequence_ would be 0 here — that's the bug.
+        // The recovery path is responsible for raising the floor; the
+        // WAL itself has no way to know about the snapshot.
+        assert(wal.currentSequence() == 0);
+
+        // Now anchor to the snapshot's walSequence (5).
+        wal.setSequenceFloor(5);
+        assert(wal.currentSequence() == 5);
+
+        // setSequenceFloor must be monotonic — calling with a lower
+        // value is a no-op.
+        wal.setSequenceFloor(3);
+        assert(wal.currentSequence() == 5);
+
+        // Subsequent appends produce 6, 7, … — strictly greater than
+        // the snapshot's walSequence, so the next boot's recovery
+        // (which calls `replay(snapSeq)` with snapSeq=5) will see them.
+        auto entry = WriteAheadLog::makeInsertEntry(
+            "test", makeDoc("post"), "test-node");
+        uint64_t seq = wal.append(std::move(entry));
+        assert(seq == 6);
+        wal.close();
+    }
+    // Phase 2 — re-open one more time, verify sequence_ recovers from
+    // the new on-disk file at 6 (open() reads the highest entry).
+    {
+        WriteAheadLog wal(cfg);
+        assert(wal.open());
+        assert(wal.currentSequence() >= 6);
+        wal.close();
+    }
+
+    fs::remove_all(dir);
+    std::cout << "PASS: WAL sequence floor anchors after snapshot truncate\n";
+}
+
 int main() {
     test_atomic_write_leaves_no_tmp();
     test_snapshot_round_trip();
@@ -325,6 +408,7 @@ int main() {
     test_orphaned_tmp_cleaned_by_listSnapshots();
     test_loadWithFallback_throws_when_all_corrupt();
     test_v4_format_roundtrip_at_scale();
+    test_wal_sequence_floor_anchors_after_truncate();
     std::cout << "\nAll snapshot durability tests PASSED!\n";
     return 0;
 }